coven_protocol/remote_object/
reclaim.rs

1use super::ownership::*;
2use super::*;
3
4impl RemoteObjectRecord {
5    pub fn object(&self) -> &ExactObjectRef {
6        match self {
7            Self::CandidateCommit(record) => &record.identity.object,
8            Self::CandidateExclusive(record) => &record.identity.object,
9            Self::RetainedAuthority(record) => &record.identity.object,
10            Self::SharedLiveSet(record) => &record.identity.object,
11        }
12    }
13
14    pub fn payloads(&self) -> &RemoteObjectPayloads {
15        match self {
16            Self::CandidateCommit(record) => &record.payloads,
17            Self::CandidateExclusive(record) => &record.payloads,
18            Self::RetainedAuthority(record) => &record.payloads,
19            Self::SharedLiveSet(record) => &record.payloads,
20        }
21    }
22
23    /// Where this record's plaintext is. A stored blob carries its locator in
24    /// the row, the image domains have no plaintext of their own, and every
25    /// other domain names its plaintext in the spool by the identity's semantic
26    /// hash.
27    pub fn semantic_payload(&self) -> SemanticPayload<'_> {
28        if let Some(locator_bytes) = self.payloads().carried_locator_bytes() {
29            return SemanticPayload::Carried(locator_bytes);
30        }
31        match self {
32            Self::CandidateCommit(record) => SemanticPayload::Spooled(record.semantic_hash),
33            Self::CandidateExclusive(record) => match &record.identity.domain {
34                CandidateExclusiveObjectDomain::CircleBootstrapImage { .. } => {
35                    SemanticPayload::Absent
36                }
37                _ => SemanticPayload::Spooled(record.identity.semantic_hash),
38            },
39            Self::RetainedAuthority(record) => {
40                SemanticPayload::Spooled(record.identity.semantic_hash)
41            }
42            Self::SharedLiveSet(record) => match &record.identity.domain {
43                // The images and the membership rollup all went to the provider
44                // from the flow that built them, and none is kept here after —
45                // so the record claims no spool file, only the reference.
46                SharedLiveSetObjectDomain::StoreSnapshotImage { .. }
47                | SharedLiveSetObjectDomain::StoreMembershipRollup { .. }
48                | SharedLiveSetObjectDomain::CircleBootstrapImage { .. } => SemanticPayload::Absent,
49                _ => SemanticPayload::Spooled(record.identity.semantic_hash),
50            },
51        }
52    }
53
54    /// The spooled ciphertext this record uploads, when the ciphertext is its
55    /// own to upload.
56    pub fn stored_payload(&self) -> Option<ObjectHash> {
57        match self.payloads() {
58            RemoteObjectPayloads::SpooledInline => Some(self.object().stored_hash()),
59            RemoteObjectPayloads::SpooledExternal | RemoteObjectPayloads::RowBlob { .. } => None,
60        }
61    }
62
63    /// Every spool file this record names. The claim it holds while its row
64    /// exists, and what the row's deletion lets go of.
65    pub fn payload_claims(&self) -> BTreeSet<ObjectHash> {
66        let mut claims = BTreeSet::new();
67        if let SemanticPayload::Spooled(hash) = self.semantic_payload() {
68            claims.insert(hash);
69        }
70        claims.extend(self.stored_payload());
71        claims
72    }
73
74    pub fn object_id(&self) -> ObjectHash {
75        remote_object_id(self.object())
76    }
77
78    pub fn is_activated_stored_blob(&self) -> bool {
79        matches!(
80            self,
81            Self::SharedLiveSet(record)
82                if record.identity.domain == SharedLiveSetObjectDomain::StoredBlob
83                    && matches!(
84                        &record.state,
85                        OwnedObjectState::UploadedVerified { ownership }
86                            if !ownership.activated.is_empty()
87                    )
88        )
89    }
90
91    pub fn validate_reclaimable_store_package(
92        &self,
93        target: &crate::store_commit::StorePackageRef,
94        activation: &StoreBatchCommitRef,
95    ) -> Result<(), RemoteObjectRecordError> {
96        let ownership = self.activated_store_package_ownership(target, activation)?;
97        if !ownership.pending.is_empty() || ownership.activated.len() != 1 {
98            return Err(RemoteObjectRecordError::InvalidReclaim);
99        }
100        Ok(())
101    }
102
103    pub fn validate_reclaimable_circle_package(
104        &self,
105        target: &crate::store_commit::CirclePackageRef,
106        activation: &StoreBatchCommitRef,
107    ) -> Result<(), RemoteObjectRecordError> {
108        let ownership = self.activated_circle_package_ownership(target, activation)?;
109        if !ownership.pending.is_empty() || ownership.activated.len() != 1 {
110            return Err(RemoteObjectRecordError::InvalidReclaim);
111        }
112        Ok(())
113    }
114
115    pub fn store_package_is_retained_for_replay(
116        &self,
117        target: &crate::store_commit::StorePackageRef,
118        activation: &StoreBatchCommitRef,
119    ) -> Result<bool, RemoteObjectRecordError> {
120        let ownership = self.activated_store_package_ownership(target, activation)?;
121        Ok(ownership
122            .activated
123            .iter()
124            .any(|owner| matches!(owner, SharedObjectOwner::RetainedReplay(_))))
125    }
126
127    fn activated_store_package_ownership<'a>(
128        &'a self,
129        target: &crate::store_commit::StorePackageRef,
130        activation: &StoreBatchCommitRef,
131    ) -> Result<&'a SharedObjectOwnership, RemoteObjectRecordError> {
132        self.validate()?;
133        let Self::SharedLiveSet(record) = self else {
134            return Err(RemoteObjectRecordError::InvalidReclaim);
135        };
136        let expected_owner = SharedObjectOwner::StoreCommit(activation.clone());
137        if !matches!(
138            &record.identity.domain,
139            SharedLiveSetObjectDomain::StorePackage { reference } if reference == target
140        ) || record.identity.semantic_hash != target.content_hash
141            || record.identity.object != target.object
142        {
143            return Err(RemoteObjectRecordError::InvalidReclaim);
144        }
145        let OwnedObjectState::UploadedVerified { ownership } = &record.state else {
146            return Err(RemoteObjectRecordError::InvalidReclaim);
147        };
148        if !ownership.activated.contains(&expected_owner) {
149            return Err(RemoteObjectRecordError::InvalidReclaim);
150        }
151        Ok(ownership)
152    }
153
154    pub fn circle_package_is_retained_for_replay(
155        &self,
156        target: &crate::store_commit::CirclePackageRef,
157        activation: &StoreBatchCommitRef,
158    ) -> Result<bool, RemoteObjectRecordError> {
159        let ownership = self.activated_circle_package_ownership(target, activation)?;
160        Ok(ownership
161            .activated
162            .iter()
163            .any(|owner| matches!(owner, SharedObjectOwner::RetainedReplay(_))))
164    }
165
166    fn activated_circle_package_ownership<'a>(
167        &'a self,
168        target: &crate::store_commit::CirclePackageRef,
169        activation: &StoreBatchCommitRef,
170    ) -> Result<&'a SharedObjectOwnership, RemoteObjectRecordError> {
171        self.validate()?;
172        let Self::SharedLiveSet(record) = self else {
173            return Err(RemoteObjectRecordError::InvalidReclaim);
174        };
175        let expected_owner = SharedObjectOwner::StoreCommit(activation.clone());
176        if !matches!(
177            &record.identity.domain,
178            SharedLiveSetObjectDomain::CirclePackage { reference } if reference == target
179        ) || record.identity.semantic_hash != target.package.content_hash
180            || record.identity.object != target.package.object
181        {
182            return Err(RemoteObjectRecordError::InvalidReclaim);
183        }
184        let OwnedObjectState::UploadedVerified { ownership } = &record.state else {
185            return Err(RemoteObjectRecordError::InvalidReclaim);
186        };
187        if !ownership.activated.contains(&expected_owner) {
188            return Err(RemoteObjectRecordError::InvalidReclaim);
189        }
190        Ok(ownership)
191    }
192
193    /// A Circle bootstrap image is reclaimable when its single activating Store
194    /// commit is its only surviving owner: no pending activation and exactly one
195    /// activated owner. A bootstrap image accretes a per-activating-commit owner,
196    /// so more than one means a live successor still references it.
197    pub fn validate_reclaimable_circle_bootstrap_image(
198        &self,
199        image: &crate::store_commit::SnapshotImageRef,
200        activation: &StoreBatchCommitRef,
201    ) -> Result<(), RemoteObjectRecordError> {
202        let ownership = self.activated_circle_bootstrap_image_ownership(image, activation)?;
203        if !ownership.pending.is_empty() || ownership.activated.len() != 1 {
204            return Err(RemoteObjectRecordError::InvalidReclaim);
205        }
206        Ok(())
207    }
208
209    /// A snapshot image is reclaimable when the generation that published it is its
210    /// only surviving owner: no pending activation and exactly one activated owner,
211    /// the `Snapshot` owner naming that stream and generation. A snapshot image
212    /// accretes no further owners, so anything else means the record is not the one
213    /// the claim describes.
214    pub fn validate_reclaimable_snapshot_image(
215        &self,
216        image: &crate::store_commit::SnapshotImageRef,
217        owner: &SnapshotObjectOwner,
218    ) -> Result<(), RemoteObjectRecordError> {
219        let ownership = self.activated_snapshot_image_ownership(image, owner)?;
220        if !ownership.pending.is_empty() || ownership.activated.len() != 1 {
221            return Err(RemoteObjectRecordError::InvalidReclaim);
222        }
223        Ok(())
224    }
225
226    fn activated_snapshot_image_ownership<'a>(
227        &'a self,
228        image: &crate::store_commit::SnapshotImageRef,
229        owner: &SnapshotObjectOwner,
230    ) -> Result<&'a SharedObjectOwnership, RemoteObjectRecordError> {
231        self.validate()?;
232        let Self::SharedLiveSet(record) = self else {
233            return Err(RemoteObjectRecordError::InvalidReclaim);
234        };
235        if !matches!(
236            &record.identity.domain,
237            SharedLiveSetObjectDomain::StoreSnapshotImage { reference } if reference == image
238        ) || record.identity.semantic_hash != image.image_hash
239            || record.identity.object != image.object
240            || !matches!(record.payloads, RemoteObjectPayloads::SpooledExternal)
241        {
242            return Err(RemoteObjectRecordError::InvalidReclaim);
243        }
244        let OwnedObjectState::UploadedVerified { ownership } = &record.state else {
245            return Err(RemoteObjectRecordError::InvalidReclaim);
246        };
247        if !ownership
248            .activated
249            .contains(&SharedObjectOwner::Snapshot(owner.clone()))
250        {
251            return Err(RemoteObjectRecordError::InvalidReclaim);
252        }
253        Ok(ownership)
254    }
255
256    /// A membership rollup is reclaimable when the generation named as its
257    /// owner is its *only* owner. A rollup two generations point at carries
258    /// both, and stays until the other one is reclaimed as well.
259    pub fn validate_reclaimable_membership_rollup(
260        &self,
261        rollup: &crate::store_commit::MembershipRollupRef,
262        owner: &SnapshotObjectOwner,
263    ) -> Result<(), RemoteObjectRecordError> {
264        self.validate()?;
265        let Self::SharedLiveSet(record) = self else {
266            return Err(RemoteObjectRecordError::InvalidReclaim);
267        };
268        if !matches!(
269            &record.identity.domain,
270            SharedLiveSetObjectDomain::StoreMembershipRollup { reference } if reference == rollup
271        ) || record.identity.semantic_hash != rollup.rollup_hash
272            || record.identity.object != rollup.object
273            || !matches!(record.payloads, RemoteObjectPayloads::SpooledExternal)
274        {
275            return Err(RemoteObjectRecordError::InvalidReclaim);
276        }
277        let OwnedObjectState::UploadedVerified { ownership } = &record.state else {
278            return Err(RemoteObjectRecordError::InvalidReclaim);
279        };
280        if !ownership
281            .activated
282            .contains(&SharedObjectOwner::Snapshot(owner.clone()))
283            || !ownership.pending.is_empty()
284            || ownership.activated.len() != 1
285        {
286            return Err(RemoteObjectRecordError::InvalidReclaim);
287        }
288        Ok(())
289    }
290
291    fn activated_circle_bootstrap_image_ownership<'a>(
292        &'a self,
293        image: &crate::store_commit::SnapshotImageRef,
294        activation: &StoreBatchCommitRef,
295    ) -> Result<&'a SharedObjectOwnership, RemoteObjectRecordError> {
296        self.validate()?;
297        let Self::SharedLiveSet(record) = self else {
298            return Err(RemoteObjectRecordError::InvalidReclaim);
299        };
300        let expected_owner = SharedObjectOwner::StoreCommit(activation.clone());
301        if !matches!(
302            &record.identity.domain,
303            SharedLiveSetObjectDomain::CircleBootstrapImage { reference } if reference == image
304        ) || record.identity.semantic_hash != image.image_hash
305            || record.identity.object != image.object
306            || !matches!(record.payloads, RemoteObjectPayloads::SpooledExternal)
307        {
308            return Err(RemoteObjectRecordError::InvalidReclaim);
309        }
310        let OwnedObjectState::UploadedVerified { ownership } = &record.state else {
311            return Err(RemoteObjectRecordError::InvalidReclaim);
312        };
313        if !ownership.activated.contains(&expected_owner) {
314            return Err(RemoteObjectRecordError::InvalidReclaim);
315        }
316        Ok(ownership)
317    }
318
319    /// A stored blob is reclaimable when it is the exact activated blob the target
320    /// names. Unlike a package or an image, a blob legitimately carries several
321    /// activated owners — one per commit that bound it — so ownership count is not
322    /// the eligibility question here; whether any live row or installable image
323    /// still needs it is, and the reclaim verified that before reaching closure.
324    pub fn validate_reclaimable_stored_blob(
325        &self,
326        stored: &crate::blob::locator::StoredBlobRef,
327    ) -> Result<(), RemoteObjectRecordError> {
328        self.validate()?;
329        let Self::SharedLiveSet(record) = self else {
330            return Err(RemoteObjectRecordError::InvalidReclaim);
331        };
332        let locator_bytes = stored.locator().to_bytes();
333        if record.identity.domain != SharedLiveSetObjectDomain::StoredBlob
334            || record.identity.semantic_hash != ObjectHash::digest(&locator_bytes)
335            || record.identity.object != *stored.object()
336            || record.payloads.carried_locator_bytes() != Some(locator_bytes.as_slice())
337        {
338            return Err(RemoteObjectRecordError::InvalidReclaim);
339        }
340        let OwnedObjectState::UploadedVerified { ownership } = &record.state else {
341            return Err(RemoteObjectRecordError::InvalidReclaim);
342        };
343        if !ownership.pending.is_empty() || ownership.activated.is_empty() {
344            return Err(RemoteObjectRecordError::InvalidReclaim);
345        }
346        Ok(())
347    }
348
349    /// The activated Store commits that published this stored blob. A blob accretes
350    /// one per commit whose package bindings named it, so a republished blob carries
351    /// several.
352    pub fn stored_blob_commit_owners(&self) -> Vec<StoreBatchCommitRef> {
353        let Self::SharedLiveSet(record) = self else {
354            return Vec::new();
355        };
356        if record.identity.domain != SharedLiveSetObjectDomain::StoredBlob {
357            return Vec::new();
358        }
359        let OwnedObjectState::UploadedVerified { ownership } = &record.state else {
360            return Vec::new();
361        };
362        ownership
363            .activated
364            .iter()
365            .filter_map(|owner| match owner {
366                SharedObjectOwner::StoreCommit(commit) => Some(commit.clone()),
367                _ => None,
368            })
369            .collect()
370    }
371
372    pub fn snapshot_owners(&self) -> impl Iterator<Item = &SnapshotObjectOwner> {
373        let owners = match self {
374            Self::SharedLiveSet(record)
375                if matches!(
376                    record.identity.domain,
377                    SharedLiveSetObjectDomain::StoredBlob
378                        | SharedLiveSetObjectDomain::StoreSnapshotImage { .. }
379                ) =>
380            {
381                match &record.state {
382                    OwnedObjectState::UploadedVerified { ownership } => Some(&ownership.activated),
383                    OwnedObjectState::Prepared { .. }
384                    | OwnedObjectState::RetirementPending { .. } => None,
385                }
386            }
387            _ => None,
388        };
389        owners.into_iter().flat_map(|owners| {
390            owners.iter().filter_map(|owner| match owner {
391                SharedObjectOwner::Snapshot(owner) => Some(owner),
392                SharedObjectOwner::StoreCommit(_) | SharedObjectOwner::RetainedReplay(_) => None,
393            })
394        })
395    }
396
397    pub fn retained_replay_owners(&self) -> impl Iterator<Item = &RetainedReplayOwner> {
398        let owners = match self {
399            Self::SharedLiveSet(record) => match &record.state {
400                OwnedObjectState::UploadedVerified { ownership } => Some(&ownership.activated),
401                OwnedObjectState::Prepared { .. } | OwnedObjectState::RetirementPending { .. } => {
402                    None
403                }
404            },
405            Self::CandidateCommit(_) | Self::CandidateExclusive(_) | Self::RetainedAuthority(_) => {
406                None
407            }
408        };
409        owners.into_iter().flat_map(|owners| {
410            owners.iter().filter_map(|owner| match owner {
411                SharedObjectOwner::RetainedReplay(owner) => Some(owner),
412                SharedObjectOwner::StoreCommit(_) | SharedObjectOwner::Snapshot(_) => None,
413            })
414        })
415    }
416}
coven_protocol/remote_object/reclaim.rs

coven_protocol/remote_object/
reclaim.rs
