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 exact snapshot named by the
210    /// claim is its only surviving owner, with no pending activation.
211    pub fn validate_reclaimable_snapshot_image(
212        &self,
213        image: &crate::store_commit::SnapshotImageRef,
214        owner: &SnapshotObjectOwner,
215    ) -> Result<(), RemoteObjectRecordError> {
216        let ownership = self.activated_snapshot_image_ownership(image, owner)?;
217        if !ownership.pending.is_empty() || ownership.activated.len() != 1 {
218            return Err(RemoteObjectRecordError::InvalidReclaim);
219        }
220        Ok(())
221    }
222
223    fn activated_snapshot_image_ownership<'a>(
224        &'a self,
225        image: &crate::store_commit::SnapshotImageRef,
226        owner: &SnapshotObjectOwner,
227    ) -> Result<&'a SharedObjectOwnership, RemoteObjectRecordError> {
228        self.validate()?;
229        let Self::SharedLiveSet(record) = self else {
230            return Err(RemoteObjectRecordError::InvalidReclaim);
231        };
232        if !matches!(
233            &record.identity.domain,
234            SharedLiveSetObjectDomain::StoreSnapshotImage { reference } if reference == image
235        ) || record.identity.semantic_hash != image.image_hash
236            || record.identity.object != image.object
237            || !matches!(record.payloads, RemoteObjectPayloads::SpooledExternal)
238        {
239            return Err(RemoteObjectRecordError::InvalidReclaim);
240        }
241        let OwnedObjectState::UploadedVerified { ownership } = &record.state else {
242            return Err(RemoteObjectRecordError::InvalidReclaim);
243        };
244        if !ownership
245            .activated
246            .contains(&SharedObjectOwner::Snapshot(owner.clone()))
247        {
248            return Err(RemoteObjectRecordError::InvalidReclaim);
249        }
250        Ok(ownership)
251    }
252
253    /// A membership rollup is reclaimable when the exact snapshot named as its
254    /// owner is its only owner. Another snapshot owner keeps the shared object live.
255    pub fn validate_reclaimable_membership_rollup(
256        &self,
257        rollup: &crate::store_commit::MembershipRollupRef,
258        owner: &SnapshotObjectOwner,
259    ) -> Result<(), RemoteObjectRecordError> {
260        self.validate()?;
261        let Self::SharedLiveSet(record) = self else {
262            return Err(RemoteObjectRecordError::InvalidReclaim);
263        };
264        if !matches!(
265            &record.identity.domain,
266            SharedLiveSetObjectDomain::StoreMembershipRollup { reference } if reference == rollup
267        ) || record.identity.semantic_hash != rollup.rollup_hash
268            || record.identity.object != rollup.object
269            || !matches!(record.payloads, RemoteObjectPayloads::SpooledExternal)
270        {
271            return Err(RemoteObjectRecordError::InvalidReclaim);
272        }
273        let OwnedObjectState::UploadedVerified { ownership } = &record.state else {
274            return Err(RemoteObjectRecordError::InvalidReclaim);
275        };
276        if !ownership
277            .activated
278            .contains(&SharedObjectOwner::Snapshot(owner.clone()))
279            || !ownership.pending.is_empty()
280            || ownership.activated.len() != 1
281        {
282            return Err(RemoteObjectRecordError::InvalidReclaim);
283        }
284        Ok(())
285    }
286
287    fn activated_circle_bootstrap_image_ownership<'a>(
288        &'a self,
289        image: &crate::store_commit::SnapshotImageRef,
290        activation: &StoreBatchCommitRef,
291    ) -> Result<&'a SharedObjectOwnership, RemoteObjectRecordError> {
292        self.validate()?;
293        let Self::SharedLiveSet(record) = self else {
294            return Err(RemoteObjectRecordError::InvalidReclaim);
295        };
296        let expected_owner = SharedObjectOwner::StoreCommit(activation.clone());
297        if !matches!(
298            &record.identity.domain,
299            SharedLiveSetObjectDomain::CircleBootstrapImage { reference } if reference == image
300        ) || record.identity.semantic_hash != image.image_hash
301            || record.identity.object != image.object
302            || !matches!(record.payloads, RemoteObjectPayloads::SpooledExternal)
303        {
304            return Err(RemoteObjectRecordError::InvalidReclaim);
305        }
306        let OwnedObjectState::UploadedVerified { ownership } = &record.state else {
307            return Err(RemoteObjectRecordError::InvalidReclaim);
308        };
309        if !ownership.activated.contains(&expected_owner) {
310            return Err(RemoteObjectRecordError::InvalidReclaim);
311        }
312        Ok(ownership)
313    }
314
315    /// A stored blob is reclaimable when it is the exact activated blob the target
316    /// names. Unlike a package or an image, a blob legitimately carries several
317    /// activated owners — one per commit that bound it — so ownership count is not
318    /// the eligibility question here; whether any live row or installable image
319    /// still needs it is, and the reclaim verified that before reaching closure.
320    pub fn validate_reclaimable_stored_blob(
321        &self,
322        stored: &crate::blob::locator::StoredBlobRef,
323    ) -> Result<(), RemoteObjectRecordError> {
324        self.validate()?;
325        let Self::SharedLiveSet(record) = self else {
326            return Err(RemoteObjectRecordError::InvalidReclaim);
327        };
328        let locator_bytes = stored.locator().to_bytes();
329        if record.identity.domain != SharedLiveSetObjectDomain::StoredBlob
330            || record.identity.semantic_hash != ObjectHash::digest(&locator_bytes)
331            || record.identity.object != *stored.object()
332            || record.payloads.carried_locator_bytes() != Some(locator_bytes.as_slice())
333        {
334            return Err(RemoteObjectRecordError::InvalidReclaim);
335        }
336        let OwnedObjectState::UploadedVerified { ownership } = &record.state else {
337            return Err(RemoteObjectRecordError::InvalidReclaim);
338        };
339        if !ownership.pending.is_empty() || ownership.activated.is_empty() {
340            return Err(RemoteObjectRecordError::InvalidReclaim);
341        }
342        Ok(())
343    }
344
345    /// The activated Store commits that published this stored blob. A blob accretes
346    /// one per commit whose package bindings named it, so a republished blob carries
347    /// several.
348    pub fn stored_blob_commit_owners(&self) -> Vec<StoreBatchCommitRef> {
349        let Self::SharedLiveSet(record) = self else {
350            return Vec::new();
351        };
352        if record.identity.domain != SharedLiveSetObjectDomain::StoredBlob {
353            return Vec::new();
354        }
355        let OwnedObjectState::UploadedVerified { ownership } = &record.state else {
356            return Vec::new();
357        };
358        ownership
359            .activated
360            .iter()
361            .filter_map(|owner| match owner {
362                SharedObjectOwner::StoreCommit(commit) => Some(commit.clone()),
363                _ => None,
364            })
365            .collect()
366    }
367
368    pub fn snapshot_owners(&self) -> impl Iterator<Item = &SnapshotObjectOwner> {
369        let owners = match self {
370            Self::SharedLiveSet(record) => match &record.state {
371                OwnedObjectState::UploadedVerified { ownership } => Some(&ownership.activated),
372                OwnedObjectState::Prepared { .. } | OwnedObjectState::RetirementPending { .. } => {
373                    None
374                }
375            },
376            _ => None,
377        };
378        owners.into_iter().flat_map(|owners| {
379            owners.iter().filter_map(|owner| match owner {
380                SharedObjectOwner::Snapshot(owner) => Some(owner),
381                SharedObjectOwner::StoreCommit(_) | SharedObjectOwner::RetainedReplay(_) => None,
382            })
383        })
384    }
385
386    pub fn retained_replay_owners(&self) -> impl Iterator<Item = &RetainedReplayOwner> {
387        let owners = match self {
388            Self::SharedLiveSet(record) => match &record.state {
389                OwnedObjectState::UploadedVerified { ownership } => Some(&ownership.activated),
390                OwnedObjectState::Prepared { .. } | OwnedObjectState::RetirementPending { .. } => {
391                    None
392                }
393            },
394            Self::CandidateCommit(_) | Self::CandidateExclusive(_) | Self::RetainedAuthority(_) => {
395                None
396            }
397        };
398        owners.into_iter().flat_map(|owners| {
399            owners.iter().filter_map(|owner| match owner {
400                SharedObjectOwner::RetainedReplay(owner) => Some(owner),
401                SharedObjectOwner::StoreCommit(_) | SharedObjectOwner::Snapshot(_) => None,
402            })
403        })
404    }
405}
coven_protocol/remote_object/reclaim.rs

coven_protocol/remote_object/
reclaim.rs
