coven_protocol/store_commit/
pending_device_join.rs

1use super::*;
2
3impl StoreBatchCommit {
4    /// Opening an Attempt creates an image consumer. Separate registration or
5    /// abandonment ends that consumer; combined registration keeps the image
6    /// until the registered device has published against accepted history.
7    pub fn has_pending_device_join_bootstrap<'a>(
8        &self,
9        state: &ResolvedStoreDeviceState,
10        frontier: &CommitFrontier,
11        accepted: impl IntoIterator<Item = &'a StoreBatchCommit>,
12    ) -> Result<bool, StoreProtocolError> {
13        let accepted = accepted.into_iter().collect::<Vec<_>>();
14        for decision in self.device_join_attempt_decisions() {
15            let DeviceJoinAttemptDecisionRef::Attempt(attempt_id) = decision else {
16                continue;
17            };
18            if let Some(activation) = self.device_registrations().iter().find(|activation| {
19                matches!(&activation.authority, StoreDeviceRegistrationActivationRef::Join { attempt_id: registered } if registered == attempt_id)
20            }) {
21                if self.registration_has_unconsumed_bootstrap(&activation.registration, state, frontier)? {
22                    return Ok(true);
23                }
24                continue;
25            }
26            let terminated = accepted.iter().any(|commit| {
27                commit.device_registrations().iter().any(|activation| {
28                    matches!(&activation.authority, StoreDeviceRegistrationActivationRef::Join { attempt_id: registered } if registered == attempt_id)
29                }) || commit.device_join_attempt_decisions().iter().any(|decision| {
30                    matches!(decision, DeviceJoinAttemptDecisionRef::Abandoned(abandonment) if &abandonment.attempt_id == attempt_id)
31                })
32            });
33            if !terminated {
34                return Ok(true);
35            }
36        }
37        Ok(false)
38    }
39
40    /// A same-principal activation opens its attempt and registers the device
41    /// together. Until that exact device publishes, its original handoff is a
42    /// live consumer of the selected snapshot and following accepted interval.
43    pub fn pending_bootstrap_registration(
44        &self,
45        state: &ResolvedStoreDeviceState,
46        frontier: &CommitFrontier,
47    ) -> Result<Option<&StoreDeviceRegistrationRef>, StoreProtocolError> {
48        let [registration] = self.device_registrations() else {
49            return Ok(None);
50        };
51        let StoreDeviceRegistrationActivationRef::Join { attempt_id } = &registration.authority
52        else {
53            return Ok(None);
54        };
55        if !self.device_join_attempt_decisions().iter().any(|decision| {
56            matches!(decision, DeviceJoinAttemptDecisionRef::Attempt(opened) if opened == attempt_id)
57        }) {
58            return Ok(None);
59        }
60        let target = &registration.registration;
61        Ok(self
62            .registration_has_unconsumed_bootstrap(target, state, frontier)?
63            .then_some(target))
64    }
65
66    fn registration_has_unconsumed_bootstrap(
67        &self,
68        target: &StoreDeviceRegistrationRef,
69        state: &ResolvedStoreDeviceState,
70        frontier: &CommitFrontier,
71    ) -> Result<bool, StoreProtocolError> {
72        let device = state
73            .devices
74            .get(&target.device_id)
75            .filter(|device| &device.registration == target)
76            .ok_or(StoreProtocolError::DeviceStateMismatch)?;
77        let stream = StreamActivation::device_authorized_stream_id(
78            self.store_root_hash,
79            target,
80            StreamAnchorDomain::StoreAnnouncements,
81        );
82        Ok(matches!(device.status, StoreDeviceStatus::Active)
83            && !frontier.commits().contains_key(&stream))
84    }
85}
86
87impl RetainedVerifiedMergeHistorySummary {
88    pub fn pending_device_join_accepted_commit(
89        &self,
90        reference: &StoreBatchCommitRef,
91    ) -> Result<Option<AcceptedStoreCommitPublication>, StoreProtocolError> {
92        let mut accepted = None;
93        for closure in self.pending_device_joins.values() {
94            if !closure
95                .commits
96                .iter()
97                .any(|commit| &commit.reference == reference)
98            {
99                continue;
100            }
101            let exact = closure.accepted_commit(reference)?;
102            if accepted.as_ref().is_some_and(|previous| previous != &exact) {
103                return Err(StoreProtocolError::Malformed(
104                    "pending Join closures disagree on an accepted commit".into(),
105                ));
106            }
107            accepted = Some(exact);
108        }
109        Ok(accepted)
110    }
111
112    /// Physical deletion stays unfinished while an accepted Join still consumes
113    /// the original snapshot and the exact interval offered with it.
114    pub fn pending_device_join_artifacts(
115        &self,
116    ) -> Result<BTreeSet<crate::objects::ExactObjectRef>, StoreProtocolError> {
117        let mut objects = BTreeSet::new();
118        for closure in self.pending_device_joins.values() {
119            let snapshot = closure
120                .publication
121                .current
122                .latest_snapshot()
123                .ok_or_else(|| {
124                    StoreProtocolError::Malformed("pending Join has no snapshot boundary".into())
125                })?;
126            let retained = self
127                .reclaim
128                .snapshots
129                .get(&snapshot.snapshot.snapshot_hash)
130                .filter(|retained| &retained.accepted == snapshot)
131                .ok_or_else(|| {
132                    StoreProtocolError::Malformed(
133                        "pending Join snapshot has no exact retained artifact ownership".into(),
134                    )
135                })?;
136            objects.extend(retained.objects().into_iter().cloned());
137            objects.extend(
138                closure
139                    .publication
140                    .verify()?
141                    .entries()
142                    .iter()
143                    .map(|entry| entry.reference().object.clone()),
144            );
145        }
146        Ok(objects)
147    }
148
149    pub fn pending_device_join_snapshot_slots(&self) -> BTreeSet<crate::objects::ObjectSlot> {
150        self.pending_device_joins
151            .values()
152            .filter_map(|closure| {
153                closure
154                    .publication
155                    .current
156                    .latest_snapshot()
157                    .map(|snapshot| snapshot.snapshot.object.slot().clone())
158            })
159            .collect()
160    }
161}
coven_protocol/store_commit/pending_device_join.rs

coven_protocol/store_commit/
pending_device_join.rs
