coven_replication/sync/store/device_join/
transport.rs

1//! Storage-mediated delivery for the device-join exchange.
2//!
3//! The join protocol owned by [`crate::sync::store::Store`] produces signed
4//! artifacts plus the unwind artifacts, and hands each to the host as a
5//! [`DeviceJoinAction`] to deliver however it likes. This module is the delivery
6//! coven ships by default: each artifact travels as one create-once object in
7//! the store's cloud home, under a per-attempt namespace, sealed with a key
8//! minted for that attempt alone.
9//!
10//! The layer carries bytes and nothing else. It never inspects an artifact
11//! beyond naming which slot it belongs in, and unsealing is not part of the
12//! trust story — the artifact's own signature and hash chaining, checked by the
13//! protocol when it accepts the artifact, are.
14//!
15//! The offer does not travel here. It is the out-of-band kickoff: the host
16//! encodes a [`DeviceJoinOfferBundle`] (the offer plus the slots and seal key
17//! this module needs) as a QR, a link, or a typed code, and the joiner's copy of
18//! that bundle is what bootstraps everything below.
19
20use std::collections::BTreeMap;
21use std::time::Duration;
22
23use serde::{Deserialize, Serialize};
24
25use crate::sync::store::{
26    DeviceJoinAbandonment, DeviceJoinAction, DeviceJoinActivation, DeviceJoinError,
27    DeviceJoinOffer, DeviceJoinReadiness, DeviceJoinRole, DeviceJoinStatus,
28    DeviceProviderAccessAdministrator, DeviceProviderAccessRequest,
29    DeviceProviderAdmissionApproval, DeviceRegistrationRequest, SamePrincipalDeviceJoin, Store,
30};
31use coven_keys::encryption::{EncryptionService, MasterKeyring, SealError};
32use coven_protocol::objects::ObjectSlot;
33use coven_protocol::objects::{ProtocolObjectContext, ProtocolObjectDomain, StorageError};
34use coven_protocol::store_commit::device_join_exchange::DeviceProviderAdmission;
35use coven_protocol::store_commit::device_join_exchange::DeviceProviderChallengePublication;
36use coven_protocol::store_commit::{DeviceJoinAttemptId, ObjectHash, STORE_PROTOCOL_VERSION};
37use coven_storage::CloudSyncObjectStorage;
38
39/// The prefix every transport slot's logical key starts with.
40const TRANSPORT_ROOT: &str = "store-v1/device-join-transport";
41
42/// Domain separation for the per-attempt seal, so a sealed artifact cannot be
43/// opened as anything but the kind and attempt it was written for.
44const SEAL_AAD_LABEL: &[u8] = b"coven.device-join-transport.v1";
45
46/// One artifact kind in transit. Every kind has exactly one producing role in
47/// the protocol and exactly one slot per attempt.
48#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
49#[serde(rename_all = "kebab-case", deny_unknown_fields)]
50pub enum DeviceJoinTransportKind {
51    ProviderAccessRequest,
52    ProviderAdmissionApproval,
53    RegistrationRequest,
54    ProviderReadyBootstrap,
55    Readiness,
56    SamePrincipalJoin,
57    Activation,
58    Abandonment,
59}
60
61impl DeviceJoinTransportKind {
62    /// Every kind, in protocol order. An attempt's namespace holds one slot per
63    /// entry — allocated together, deleted together.
64    pub const ALL: [Self; 8] = [
65        Self::ProviderAccessRequest,
66        Self::ProviderAdmissionApproval,
67        Self::RegistrationRequest,
68        Self::ProviderReadyBootstrap,
69        Self::Readiness,
70        Self::SamePrincipalJoin,
71        Self::Activation,
72        Self::Abandonment,
73    ];
74
75    /// The last path component of this kind's slot.
76    fn slug(self) -> &'static str {
77        match self {
78            Self::ProviderAccessRequest => "provider-access-request",
79            Self::ProviderAdmissionApproval => "provider-admission-approval",
80            Self::RegistrationRequest => "registration-request",
81            Self::ProviderReadyBootstrap => "provider-ready-bootstrap",
82            Self::Readiness => "readiness",
83            Self::SamePrincipalJoin => "same-principal-join",
84            Self::Activation => "activation",
85            Self::Abandonment => "abandonment",
86        }
87    }
88
89    /// The one role the protocol lets produce this kind. A publish from any
90    /// other role is refused before it reaches storage.
91    fn producer(self) -> DeviceJoinRole {
92        match self {
93            Self::ProviderAccessRequest | Self::RegistrationRequest | Self::Readiness => {
94                DeviceJoinRole::Joiner
95            }
96            Self::ProviderAdmissionApproval
97            | Self::ProviderReadyBootstrap
98            | Self::SamePrincipalJoin
99            | Self::Activation
100            | Self::Abandonment => DeviceJoinRole::Owner,
101        }
102    }
103
104    /// The kind an action's artifact belongs in, or `None` for the actions that
105    /// name local work rather than a transfer (`CompleteJoin`,
106    /// `ResumeOperation`) and for the offer, which travels out of band.
107    fn of(action: &DeviceJoinAction) -> Option<Self> {
108        match action {
109            DeviceJoinAction::TransferProviderAccessRequest(_) => Some(Self::ProviderAccessRequest),
110            DeviceJoinAction::TransferProviderAdmissionApproval(_) => {
111                Some(Self::ProviderAdmissionApproval)
112            }
113            DeviceJoinAction::TransferRegistrationRequest(_) => Some(Self::RegistrationRequest),
114            DeviceJoinAction::TransferProviderReadyBootstrap(_) => {
115                Some(Self::ProviderReadyBootstrap)
116            }
117            DeviceJoinAction::TransferReadiness(_) => Some(Self::Readiness),
118            DeviceJoinAction::TransferSamePrincipalJoin(_) => Some(Self::SamePrincipalJoin),
119            DeviceJoinAction::TransferActivation(_) => Some(Self::Activation),
120            DeviceJoinAction::TransferAbandonment(_) => Some(Self::Abandonment),
121            DeviceJoinAction::TransferOffer(_)
122            | DeviceJoinAction::CompleteJoin(_)
123            | DeviceJoinAction::ResumeOperation { .. } => None,
124        }
125    }
126}
127
128/// The artifact type a kind carries. Awaiting a kind yields exactly this type,
129/// so a caller never re-matches the action enum it just asked for by kind.
130pub trait DeviceJoinArtifact: Sized {
131    const KIND: DeviceJoinTransportKind;
132
133    fn from_action(action: DeviceJoinAction) -> Option<Self>;
134}
135
136macro_rules! device_join_artifact {
137    ($type:ty, $kind:ident, $variant:ident) => {
138        impl DeviceJoinArtifact for $type {
139            const KIND: DeviceJoinTransportKind = DeviceJoinTransportKind::$kind;
140
141            fn from_action(action: DeviceJoinAction) -> Option<Self> {
142                match action {
143                    DeviceJoinAction::$variant(value) => Some(value),
144                    _ => None,
145                }
146            }
147        }
148    };
149}
150
151device_join_artifact!(
152    DeviceProviderAccessRequest,
153    ProviderAccessRequest,
154    TransferProviderAccessRequest
155);
156device_join_artifact!(
157    DeviceProviderAdmissionApproval,
158    ProviderAdmissionApproval,
159    TransferProviderAdmissionApproval
160);
161device_join_artifact!(
162    DeviceRegistrationRequest,
163    RegistrationRequest,
164    TransferRegistrationRequest
165);
166device_join_artifact!(
167    coven_protocol::store_commit::device_join_exchange::ProviderReadyDeviceBootstrap,
168    ProviderReadyBootstrap,
169    TransferProviderReadyBootstrap
170);
171device_join_artifact!(DeviceJoinReadiness, Readiness, TransferReadiness);
172device_join_artifact!(
173    SamePrincipalDeviceJoin,
174    SamePrincipalJoin,
175    TransferSamePrincipalJoin
176);
177device_join_artifact!(DeviceJoinActivation, Activation, TransferActivation);
178device_join_artifact!(DeviceJoinAbandonment, Abandonment, TransferAbandonment);
179
180/// The slots and seal key one attempt's artifacts travel through.
181///
182/// The owner allocates the slots when it begins the join, because on providers
183/// whose exact slots carry an opaque provider locator (Google Drive) a reader
184/// cannot derive a slot from its logical key — the same reason the protocol's
185/// own attempt, outcome, and registration slots are reserved up front and named
186/// in the signed artifact that precedes them.
187#[derive(Clone, Debug, Serialize, Deserialize)]
188#[serde(deny_unknown_fields)]
189pub struct DeviceJoinTransportParams {
190    pub version: u32,
191    pub attempt_namespace: String,
192    pub slots: BTreeMap<DeviceJoinTransportKind, ObjectSlot>,
193    #[serde(with = "seal_key")]
194    seal_key: MasterKeyring,
195}
196
197/// `MasterKeyring` is the codebase's symmetric-key carrier and travels as its
198/// own serialized form; the transport adds no second key encoding.
199mod seal_key {
200    use super::MasterKeyring;
201    use serde::{Deserialize, Deserializer, Serializer};
202
203    pub(super) fn serialize<S: Serializer>(
204        keyring: &MasterKeyring,
205        serializer: S,
206    ) -> Result<S::Ok, S::Error> {
207        serializer.serialize_str(&keyring.to_serialized())
208    }
209
210    pub(super) fn deserialize<'de, D: Deserializer<'de>>(
211        deserializer: D,
212    ) -> Result<MasterKeyring, D::Error> {
213        let encoded = String::deserialize(deserializer)?;
214        MasterKeyring::from_serialized(&encoded).map_err(serde::de::Error::custom)
215    }
216}
217
218impl DeviceJoinTransportParams {
219    pub(crate) fn new(
220        attempt_namespace: String,
221        slots: BTreeMap<DeviceJoinTransportKind, ObjectSlot>,
222        seal_key: MasterKeyring,
223    ) -> Self {
224        Self {
225            version: STORE_PROTOCOL_VERSION,
226            attempt_namespace,
227            slots,
228            seal_key,
229        }
230    }
231
232    fn slot(&self, kind: DeviceJoinTransportKind) -> Result<&ObjectSlot, DeviceJoinTransportError> {
233        self.slots
234            .get(&kind)
235            .ok_or(DeviceJoinTransportError::MissingSlot { kind })
236    }
237
238    fn validate_for(&self, offer: &DeviceJoinOffer) -> Result<(), DeviceJoinTransportError> {
239        if self.version != STORE_PROTOCOL_VERSION
240            || self.attempt_namespace != attempt_namespace(offer.attempt_id)
241        {
242            return Err(DeviceJoinTransportError::BundleMismatch);
243        }
244        let context = slot_context(offer.store_root.store_root_hash);
245        for kind in DeviceJoinTransportKind::ALL {
246            context.validate_slot(
247                self.slot(kind)?,
248                &semantic_prefix(&self.attempt_namespace, kind),
249            )?;
250        }
251        Ok(())
252    }
253}
254
255/// The out-of-band kickoff: the offer plus everything the transport needs to
256/// carry the rest of the exchange. The host encodes this however it delivers a
257/// join code; coven does not choose that encoding.
258#[derive(Clone, Debug, Serialize, Deserialize)]
259#[serde(deny_unknown_fields)]
260pub struct DeviceJoinOfferBundle {
261    pub version: u32,
262    pub offer: DeviceJoinOffer,
263    pub transport: DeviceJoinTransportParams,
264}
265
266impl DeviceJoinOfferBundle {
267    pub fn to_bytes(&self) -> Vec<u8> {
268        serde_json::to_vec(self).expect("device join offer bundle serialization cannot fail")
269    }
270
271    pub fn from_bytes(bytes: &[u8]) -> Result<Self, DeviceJoinTransportError> {
272        let bundle: Self = serde_json::from_slice(bytes)?;
273        if bundle.version != STORE_PROTOCOL_VERSION {
274            return Err(DeviceJoinTransportError::BundleMismatch);
275        }
276        bundle.transport.validate_for(&bundle.offer)?;
277        Ok(bundle)
278    }
279}
280
281/// What a joining device found while waiting for its next artifact: the
282/// artifact, or the owner's abandonment of the attempt.
283#[derive(Clone, Debug, PartialEq, Eq)]
284pub enum DeviceJoinStep<T> {
285    Continue(T),
286    Abandoned(DeviceJoinAbandonment),
287}
288
289/// How a driven join ended for the admitting side.
290#[derive(Clone, Debug, PartialEq, Eq)]
291pub enum DeviceJoinDriveOutcome {
292    Activated(DeviceJoinActivation),
293    Abandoned(DeviceJoinAbandonment),
294}
295
296/// The joining device's current user-visible operation. These values describe
297/// the work actually executing or the exact counterpart artifact being
298/// awaited; hosts render them directly instead of collapsing the whole join
299/// into one indeterminate state.
300#[derive(Clone, Debug, PartialEq, Eq)]
301pub enum JoiningDeviceJoinProgress {
302    WaitingForApproval,
303    RequestingProviderAccess,
304    WaitingForProviderAccess,
305    RegisteringDevice,
306    WaitingForLibrary,
307    DownloadingSnapshot { bytes_done: u64, bytes_total: u64 },
308    InstallingSnapshot,
309    WaitingForActivation,
310    CatchingUp,
311    SavingLibrary,
312}
313
314/// A joining device's retained progress sink. Provider reads keep a clone while
315/// their response stream is active, so every received buffer reaches the host.
316pub type JoiningDeviceJoinProgressObserver =
317    std::sync::Arc<dyn Fn(JoiningDeviceJoinProgress) + Send + Sync>;
318
319/// The existing device's current user-visible operation while admitting the
320/// joining device.
321#[derive(Clone, Copy, Debug, PartialEq, Eq)]
322pub enum AdmittingDeviceJoinProgress {
323    PreparingInvitation,
324    WaitingForProviderAccessRequest,
325    GrantingProviderAccess,
326    WaitingForRegistrationRequest,
327    RegisteringDevice,
328    PreparingLibrary,
329    WaitingForJoiningDevice,
330    ActivatingDevice,
331}
332
333/// How often to look for a counterpart's artifact, and how long to keep
334/// looking before giving up on it.
335#[derive(Clone, Copy, Debug, PartialEq, Eq)]
336pub struct DeviceJoinTransportTiming {
337    pub poll: Duration,
338    pub deadline: Duration,
339}
340
341impl DeviceJoinTransportTiming {
342    /// Pairing's product timing. Hosts render the states; coven decides how
343    /// frequently storage and the local pairing endpoint are observed and when
344    /// an absent counterpart becomes a failure.
345    pub const fn interactive() -> Self {
346        Self {
347            poll: Duration::from_millis(100),
348            deadline: Duration::from_secs(180),
349        }
350    }
351
352    /// The cadence a wait on this timing uses.
353    fn polls(self) -> JoinPollBackoff {
354        JoinPollBackoff {
355            next: self.poll,
356            ceiling: JOIN_POLL_CEILING.max(self.poll),
357        }
358    }
359}
360
361/// The longest a wait ever sleeps between looks.
362///
363/// A wait on a counterpart is a wait on a person — an owner reading an approval
364/// prompt — or on that device's next sync cycle, which is tens of seconds away.
365/// Looking every hundred milliseconds for all of it is hundreds of provider
366/// reads that answer "not yet", and a provider that rate-limits them makes the
367/// join slower, not faster. The first look is immediate and the cadence backs
368/// off to this, so a counterpart that answers at once is still seen at once.
369const JOIN_POLL_CEILING: Duration = Duration::from_secs(2);
370
371struct JoinPollBackoff {
372    next: Duration,
373    ceiling: Duration,
374}
375
376impl JoinPollBackoff {
377    fn next(&mut self) -> Duration {
378        let current = self.next;
379        self.next = (current * 2).min(self.ceiling);
380        current
381    }
382}
383
384/// Time one owner-side device-join step and report it the way every other
385/// staged run reports.
386///
387/// Each of these is one transition in the Add-a-device flow — approve the
388/// provider access, accept the registration, activate — and each is one or more
389/// provider round trips, which is what `requests` counts. Two flows reach them:
390/// the discrete command API a host drives itself, and the pairing driver's
391/// `drive_once`. They share this function so a run through either one reads the
392/// same in the log, and each passes the counter of the home it drives.
393pub async fn timed_owner_join_step<T>(
394    step: &'static str,
395    requests: Option<std::sync::Arc<dyn coven_foundation::stage_timing::ProviderRequests>>,
396    work: impl std::future::Future<Output = T>,
397) -> T {
398    let mut timings =
399        coven_foundation::stage_timing::StageTimings::counting("Device join owner step", requests);
400    let outcome = timings.stage(step, work).await;
401    timings.report();
402    outcome
403}
404
405/// One wait on the counterpart, reported when it ends.
406///
407/// A join that took four minutes is either waiting on the other device or
408/// fetching, and until these lines existed the logs could not say which. The
409/// poll count separates a wait that sat through the owner's next sync cycle
410/// from one that answered immediately.
411struct JoinWait {
412    kind: DeviceJoinTransportKind,
413    started: coven_foundation::clock::Stopwatch,
414    polls: std::sync::atomic::AtomicU64,
415}
416
417impl JoinWait {
418    fn begin(kind: DeviceJoinTransportKind) -> Self {
419        Self {
420            kind,
421            started: coven_foundation::clock::Stopwatch::start(),
422            polls: std::sync::atomic::AtomicU64::new(0),
423        }
424    }
425
426    fn polled(&self) {
427        self.polls
428            .fetch_add(1, std::sync::atomic::Ordering::Relaxed);
429    }
430
431    fn report(self) {
432        tracing::info!(
433            kind = ?self.kind,
434            produced_by = ?self.kind.producer(),
435            waited_ms = self.started.elapsed().as_millis() as u64,
436            looks = self.polls.load(std::sync::atomic::Ordering::Relaxed),
437            "Device join waited for its counterpart"
438        );
439    }
440}
441
442/// Why a transfer through the transport failed.
443#[derive(Debug, thiserror::Error)]
444pub enum DeviceJoinTransportError {
445    #[error("storage: {0}")]
446    Storage(#[from] StorageError),
447    #[error("device join: {0}")]
448    DeviceJoin(#[from] DeviceJoinError),
449    #[error("transport artifact is not valid JSON: {0}")]
450    Malformed(#[from] serde_json::Error),
451    #[error("transport artifact could not be unsealed: {0}")]
452    Unsealable(#[from] SealError),
453    #[error("the offer bundle does not describe this attempt's transport")]
454    BundleMismatch,
455    #[error("this attempt's transport has no {kind:?} slot")]
456    MissingSlot { kind: DeviceJoinTransportKind },
457    /// The action carries no transferable artifact — the offer travels out of
458    /// band, and `CompleteJoin`/`CompleteCleanup`/`ResumeOperation` name local
459    /// work rather than a transfer.
460    #[error("{0:?} carries nothing for the transport to deliver")]
461    NotTransferable(Box<DeviceJoinAction>),
462    /// Only one role produces each kind, and this device does not hold it.
463    #[error("a {kind:?} artifact is the {role:?}'s to publish, not this device's")]
464    WrongProducer {
465        kind: DeviceJoinTransportKind,
466        role: DeviceJoinRole,
467    },
468    /// The slot already holds a different artifact of this kind. Republishing
469    /// the same artifact after a crash succeeds; a different one never
470    /// overwrites what a counterpart may already have read.
471    #[error("the {kind:?} slot already holds a different artifact")]
472    ArtifactConflict { kind: DeviceJoinTransportKind },
473    /// The slot's stored bytes are not the ones this write produced — a
474    /// concurrent writer reached it first.
475    #[error("the {kind:?} slot was written concurrently with different bytes")]
476    SlotConflict { kind: DeviceJoinTransportKind },
477    /// The unsealed bytes decode as a different kind than the slot they sat in.
478    #[error("the {kind:?} slot holds an artifact of another kind")]
479    KindMismatch { kind: DeviceJoinTransportKind },
480    #[error("the {producer:?} never published its {kind:?} artifact")]
481    Timeout {
482        kind: DeviceJoinTransportKind,
483        producer: DeviceJoinRole,
484    },
485}
486
487/// One attempt's slot namespace, bound to the side of the exchange this device
488/// is on.
489pub struct DeviceJoinTransport<'a> {
490    storage: &'a dyn CloudSyncObjectStorage,
491    params: &'a DeviceJoinTransportParams,
492    store_root_hash: ObjectHash,
493    seal: EncryptionService,
494    role: DeviceJoinRole,
495}
496
497impl<'a> DeviceJoinTransport<'a> {
498    /// Open the transport described by `bundle` against `storage`, for the role
499    /// this device plays. It may publish only the kinds that role produces; it
500    /// may read every kind.
501    pub fn open(
502        storage: &'a dyn CloudSyncObjectStorage,
503        bundle: &'a DeviceJoinOfferBundle,
504        role: DeviceJoinRole,
505    ) -> Result<Self, DeviceJoinTransportError> {
506        bundle.transport.validate_for(&bundle.offer)?;
507        Ok(Self {
508            storage,
509            params: &bundle.transport,
510            store_root_hash: bundle.offer.store_root.store_root_hash,
511            seal: EncryptionService::from(bundle.transport.seal_key.clone()),
512            role,
513        })
514    }
515
516    /// Seal an artifact and create it at its slot.
517    ///
518    /// Republishing an artifact already at its slot succeeds — that is what a
519    /// crash between the durable journal advance and the create resumes into.
520    /// The seal draws a fresh nonce per call, so sameness is decided on the
521    /// artifact, not on the stored ciphertext; the first write's bytes stay.
522    /// A *different* artifact at an occupied slot is refused: a counterpart may
523    /// already have read what is there.
524    pub async fn publish(&self, action: &DeviceJoinAction) -> Result<(), DeviceJoinTransportError> {
525        let kind = DeviceJoinTransportKind::of(action)
526            .ok_or_else(|| DeviceJoinTransportError::NotTransferable(Box::new(action.clone())))?;
527        let producer = kind.producer();
528        if self.role != producer {
529            return Err(DeviceJoinTransportError::WrongProducer {
530                kind,
531                role: producer,
532            });
533        }
534        let sealed = self
535            .seal
536            .seal_app_data(&serde_json::to_vec(action)?, &self.seal_aad(kind));
537        let prepared = self.storage.prepare_protocol_object(
538            &slot_context(self.store_root_hash),
539            self.params.slot(kind)?.clone(),
540            &self.semantic_prefix(kind),
541            sealed,
542        )?;
543        match self.storage.create_protocol_object(&prepared).await {
544            Ok(()) => Ok(()),
545            Err(StorageError::SlotCollision(_)) => match self.read(kind).await? {
546                Some(existing) if existing == *action => Ok(()),
547                Some(_) => Err(DeviceJoinTransportError::ArtifactConflict { kind }),
548                None => Err(DeviceJoinTransportError::SlotConflict { kind }),
549            },
550            Err(error) => Err(error.into()),
551        }
552    }
553
554    /// Read one kind's artifact, or `None` while its slot is still empty.
555    pub async fn read(
556        &self,
557        kind: DeviceJoinTransportKind,
558    ) -> Result<Option<DeviceJoinAction>, DeviceJoinTransportError> {
559        let sealed = match self
560            .storage
561            .read_protocol_slot(
562                &slot_context(self.store_root_hash),
563                self.params.slot(kind)?,
564                &self.semantic_prefix(kind),
565            )
566            .await
567        {
568            Ok((sealed, _)) => sealed,
569            Err(StorageError::NotFound(_)) => return Ok(None),
570            Err(error) => return Err(error.into()),
571        };
572        let opened = self.seal.open_app_data(&sealed, &self.seal_aad(kind))?;
573        let action: DeviceJoinAction = serde_json::from_slice(&opened)?;
574        if DeviceJoinTransportKind::of(&action) != Some(kind) {
575            return Err(DeviceJoinTransportError::KindMismatch { kind });
576        }
577        Ok(Some(action))
578    }
579
580    /// Poll for the counterpart's artifact of type `T` until the deadline. The
581    /// timeout names the role that never published, so a host can tell the user
582    /// which device it is waiting on.
583    pub async fn await_artifact<T: DeviceJoinArtifact>(
584        &self,
585        timing: DeviceJoinTransportTiming,
586    ) -> Result<T, DeviceJoinTransportError> {
587        let kind = T::KIND;
588        let wait = JoinWait::begin(kind);
589        let polled = tokio::time::timeout(timing.deadline, async {
590            let mut poll = timing.polls();
591            loop {
592                wait.polled();
593                if let Some(action) = self.read(kind).await? {
594                    return T::from_action(action)
595                        .ok_or(DeviceJoinTransportError::KindMismatch { kind });
596                }
597                tokio::time::sleep(poll.next()).await;
598            }
599        })
600        .await;
601        wait.report();
602        match polled {
603            Ok(artifact) => artifact,
604            Err(_) => Err(DeviceJoinTransportError::Timeout {
605                kind,
606                producer: kind.producer(),
607            }),
608        }
609    }
610
611    /// Observe one artifact without imposing a phase deadline. A concurrent
612    /// operation owns the deadline; this observation exists to interrupt that
613    /// operation when a terminal artifact appears.
614    ///
615    /// This is the longest-running wait in a join and the one least likely to
616    /// find anything: it watches for the owner cancelling, for the whole join,
617    /// alongside the snapshot download and the install. At the asked-for
618    /// cadence that is a provider read every hundred milliseconds for minutes
619    /// to answer "not yet" — which is exactly what the poll backoff was
620    /// introduced to stop for the phase waits, and this one was left behind
621    /// because it takes a bare interval rather than a timing. It takes the
622    /// timing now and backs off like the others: the first look is immediate,
623    /// so a cancellation still interrupts promptly, and the cadence settles at
624    /// the same ceiling instead of running flat out under a several-second
625    /// download.
626    pub async fn observe_artifact<T: DeviceJoinArtifact>(
627        &self,
628        timing: DeviceJoinTransportTiming,
629    ) -> Result<T, DeviceJoinTransportError> {
630        let kind = T::KIND;
631        let mut poll = timing.polls();
632        loop {
633            if let Some(action) = self.read(kind).await? {
634                return T::from_action(action)
635                    .ok_or(DeviceJoinTransportError::KindMismatch { kind });
636            }
637            tokio::time::sleep(poll.next()).await;
638        }
639    }
640
641    /// Poll for the next artifact of type `T`, or for the owner's abandonment
642    /// of the whole attempt, whichever appears first.
643    ///
644    /// The owner may give up on an attempt while the joining device is waiting
645    /// for the next step, so every joiner wait watches both slots. A wait that
646    /// watched only its own kind would sit until its deadline against an
647    /// abandonment already published.
648    pub async fn await_step<T: DeviceJoinArtifact>(
649        &self,
650        timing: DeviceJoinTransportTiming,
651    ) -> Result<DeviceJoinStep<T>, DeviceJoinTransportError> {
652        let kind = T::KIND;
653        let wait = JoinWait::begin(kind);
654        let polled = tokio::time::timeout(timing.deadline, async {
655            let mut poll = timing.polls();
656            loop {
657                wait.polled();
658                if let Some(action) = self.read(DeviceJoinTransportKind::Abandonment).await? {
659                    return DeviceJoinAbandonment::from_action(action)
660                        .map(DeviceJoinStep::Abandoned)
661                        .ok_or(DeviceJoinTransportError::KindMismatch {
662                            kind: DeviceJoinTransportKind::Abandonment,
663                        });
664                }
665                if let Some(action) = self.read(kind).await? {
666                    return T::from_action(action)
667                        .map(DeviceJoinStep::Continue)
668                        .ok_or(DeviceJoinTransportError::KindMismatch { kind });
669                }
670                tokio::time::sleep(poll.next()).await;
671            }
672        })
673        .await;
674        wait.report();
675        match polled {
676            Ok(step) => step,
677            Err(_) => Err(DeviceJoinTransportError::Timeout {
678                kind,
679                producer: kind.producer(),
680            }),
681        }
682    }
683
684    /// Remove everything under this attempt's namespace.
685    ///
686    /// Called once the exchange has reached an end the joining device has
687    /// consumed — its completed join or its accepted abandonment. The joining
688    /// device is the last reader on both, which is why the deletion is its to
689    /// make: the admitting device has no artifact by which it could learn that
690    /// the joiner read the last thing it published. There is no sweep behind
691    /// this.
692    ///
693    /// The namespace is listed rather than probed kind by kind. Probing asks
694    /// for every name this build knows and so leaves behind anything written
695    /// under a name it does not — an artifact from a different version, or from
696    /// anyone else who can write to the provider. A listing names what is
697    /// actually there, which is what "remove the namespace" has to mean.
698    ///
699    /// Each object is still deleted by the exact reference its own stored bytes
700    /// produce, so a delete cannot race a concurrent write: the reference
701    /// carries the size and hash observed, and the delete refuses if what sits
702    /// there no longer matches. Nothing is opened — this is removing a
703    /// namespace, not reading it, and an object this device cannot decrypt is
704    /// exactly as much garbage as one it can.
705    pub async fn delete_attempt_slots(&self) -> Result<(), DeviceJoinTransportError> {
706        let context = slot_context(self.store_root_hash);
707        let listed = self
708            .storage
709            .list_protocol_slots(&context, &format!("{}/", self.params.attempt_namespace))
710            .await?;
711        let deletions = futures_util::future::join_all(listed.iter().map(|slot| async move {
712            let Some(object) = self.storage.observe_exact_slot(slot).await? else {
713                return Ok(());
714            };
715            self.storage
716                .delete_protocol_object(&object)
717                .await
718                .map_err(DeviceJoinTransportError::from)
719        }))
720        .await;
721        for result in deletions {
722            result?;
723        }
724        Ok(())
725    }
726
727    fn semantic_prefix(&self, kind: DeviceJoinTransportKind) -> String {
728        semantic_prefix(&self.params.attempt_namespace, kind)
729    }
730
731    /// Bind a sealed artifact to its store, its attempt, and its kind, so bytes
732    /// lifted from one slot cannot be opened as another.
733    fn seal_aad(&self, kind: DeviceJoinTransportKind) -> Vec<u8> {
734        let prefix = self.semantic_prefix(kind);
735        let mut aad = SEAL_AAD_LABEL.to_vec();
736        aad.extend_from_slice(self.store_root_hash.as_bytes());
737        aad.extend_from_slice(&(prefix.len() as u64).to_le_bytes());
738        aad.extend_from_slice(prefix.as_bytes());
739        aad
740    }
741}
742
743pub(crate) fn attempt_namespace(attempt_id: DeviceJoinAttemptId) -> String {
744    format!("{TRANSPORT_ROOT}/{attempt_id}")
745}
746
747pub(crate) fn semantic_prefix(attempt_namespace: &str, kind: DeviceJoinTransportKind) -> String {
748    format!("{attempt_namespace}/{}", kind.slug())
749}
750
751pub(crate) fn slot_context(store_root_hash: ObjectHash) -> ProtocolObjectContext {
752    ProtocolObjectContext::recipient_sealed(
753        store_root_hash,
754        ProtocolObjectDomain::DeviceJoinTransport,
755    )
756}
757
758/// Whether the driver approves an access request, and on whose say-so.
759pub enum DeviceJoinApprovalPolicy<'a> {
760    /// Approve requests against an attempt this device itself issued: its own
761    /// owner journal holds the attempt, and the request carries the offer this
762    /// bundle names. Anything else is refused. The host opts into this; it is
763    /// never the implicit behavior.
764    AutoApproveSelfIssued,
765    /// Ask the host, which prompts whoever is at the device.
766    Ask(&'a (dyn Fn(&DeviceProviderAccessRequest) -> DeviceJoinApproval + Send + Sync)),
767}
768
769#[derive(Clone, Copy, Debug, PartialEq, Eq)]
770pub enum DeviceJoinApproval {
771    Approve,
772    Refuse,
773}
774
775pub struct StoreDeviceJoinTransport<'store> {
776    store: &'store Store,
777}
778
779impl<'store> StoreDeviceJoinTransport<'store> {
780    pub(crate) fn new(store: &'store Store) -> Self {
781        Self { store }
782    }
783
784    pub async fn allocate_bundle(
785        &self,
786        offer: DeviceJoinOffer,
787    ) -> Result<DeviceJoinOfferBundle, DeviceJoinTransportError> {
788        self.store
789            .allocate_device_join_transport_bundle(offer)
790            .await
791    }
792
793    pub async fn drive(
794        &self,
795        bundle: &DeviceJoinOfferBundle,
796        policy: DeviceJoinApprovalPolicy<'_>,
797        access_administrator: Option<&dyn DeviceProviderAccessAdministrator>,
798        on_progress: &(dyn Fn(AdmittingDeviceJoinProgress) + Send + Sync),
799        timing: DeviceJoinTransportTiming,
800    ) -> Result<DeviceJoinDriveOutcome, DeviceJoinTransportError> {
801        retrying_activation_conflicts(|| async {
802            AttemptTransport::open(self.store, bundle)
803                .await?
804                .drive_once(&policy, access_administrator, on_progress, timing)
805                .await
806        })
807        .await
808    }
809
810    pub async fn abandon(
811        &self,
812        bundle: &DeviceJoinOfferBundle,
813    ) -> Result<DeviceJoinAbandonment, DeviceJoinTransportError> {
814        let attempt = AttemptTransport::open(self.store, bundle).await?;
815        let abandonment = self.store.abandon_device_join(bundle.offer.clone()).await?;
816        attempt.finish_abandonment(&abandonment).await?;
817        Ok(abandonment)
818    }
819
820    /// Give up on an attempt this device offered.
821    ///
822    /// Only an attempt that has not reached its Store commit can be given up on:
823    /// up to that point nothing is published about the joining device, so an
824    /// abandonment is the whole story. Past it the device has been approved and
825    /// holds storage access, and taking that back is member removal with a key
826    /// rotation — not something a pairing window can do.
827    pub async fn abort(
828        &self,
829        bundle: &DeviceJoinOfferBundle,
830    ) -> Result<(), DeviceJoinTransportError> {
831        let attempt = AttemptTransport::open(self.store, bundle).await?;
832        match attempt.owner_status().await? {
833            // No row means the attempt already finished and its terminal step
834            // deleted it. There is nothing left to give up on, and minting a
835            // second abandonment for a finished attempt would only republish
836            // what was already delivered.
837            None => Ok(()),
838            Some(
839                DeviceJoinStatus::AwaitingAccessRequest { .. }
840                | DeviceJoinStatus::AwaitingProviderAdmission { .. }
841                | DeviceJoinStatus::ProviderAccessGrantCreatePending { .. }
842                | DeviceJoinStatus::AwaitingRegistrationRequest { .. }
843                | DeviceJoinStatus::AwaitingBootstrap { .. }
844                | DeviceJoinStatus::AbandonmentCreatePending { .. }
845                | DeviceJoinStatus::Abandoned { .. },
846            ) => {
847                self.abandon(bundle).await?;
848                Ok(())
849            }
850            status => Err(DeviceJoinError::Store(format!(
851                "device join {} is past the point it could be given up on: {status:?}",
852                bundle.offer.attempt_id
853            ))
854            .into()),
855        }
856    }
857}
858
859/// One attempt in flight: the bundle naming its transport slots and the attempt
860/// every status read addresses. Every step of a drive shares both.
861struct AttemptTransport<'attempt> {
862    store: &'attempt Store,
863    bundle: &'attempt DeviceJoinOfferBundle,
864    attempt_id: DeviceJoinAttemptId,
865}
866
867impl<'attempt> AttemptTransport<'attempt> {
868    async fn open(
869        store: &'attempt Store,
870        bundle: &'attempt DeviceJoinOfferBundle,
871    ) -> Result<Self, DeviceJoinTransportError> {
872        store.require_device_join_admitter(&bundle.offer).await?;
873        Ok(Self {
874            store,
875            bundle,
876            attempt_id: bundle.offer.attempt_id,
877        })
878    }
879
880    /// Put an artifact at its transport slot. An artifact already at its slot is
881    /// the same transfer, so a step that produced its artifact and died before
882    /// publishing it republishes here for nothing.
883    async fn publish(&self, action: DeviceJoinAction) -> Result<(), DeviceJoinTransportError> {
884        self.step(
885            "publish artifact",
886            self.store
887                .publish_device_join_transport_artifact(self.bundle, &action),
888        )
889        .await
890    }
891
892    /// Put the abandonment at its slot and drop the row that anchored getting
893    /// it there.
894    ///
895    /// These are one step. Until the artifact is published the row is what a
896    /// resumed drive reads to know it still owes it; once published there is
897    /// nothing further to say about the attempt, and a row left behind would
898    /// keep offering the same transfer on every pass forever.
899    async fn finish_abandonment(
900        &self,
901        abandonment: &DeviceJoinAbandonment,
902    ) -> Result<(), DeviceJoinTransportError> {
903        self.publish(DeviceJoinAction::TransferAbandonment(abandonment.clone()))
904            .await?;
905        self.store
906            .retire_device_join_row(self.attempt_id, DeviceJoinRole::Owner)
907            .await
908    }
909
910    /// Time one step of the driven exchange under the shared owner-step line.
911    ///
912    /// The work is boxed: `drive_once` holds a dozen of these, and leaving each
913    /// one inline grows its already-large state machine past the stack a test
914    /// runner gives it.
915    async fn step<T>(&self, step: &'static str, work: impl std::future::Future<Output = T>) -> T {
916        timed_owner_join_step(step, self.store.provider_requests(), Box::pin(work)).await
917    }
918
919    /// Read the artifact the other side owes this step, waiting for it to appear.
920    async fn await_artifact<T: DeviceJoinArtifact>(
921        &self,
922        timing: DeviceJoinTransportTiming,
923    ) -> Result<T, DeviceJoinTransportError> {
924        self.store
925            .await_device_join_transport_artifact::<T>(self.bundle, timing)
926            .await
927    }
928
929    async fn owner_status(&self) -> Result<Option<DeviceJoinStatus>, DeviceJoinTransportError> {
930        self.store
931            .device_join_transport_status(self.attempt_id, DeviceJoinRole::Owner)
932            .await
933    }
934
935    /// Carry the admitting side of one attempt as far as it will go.
936    ///
937    /// One device admits, so there is one journal and one status to read: every
938    /// pass takes the durable state and performs the step that follows it. A
939    /// step that produced its artifact and died before publishing republishes
940    /// here for nothing, and a step already past does nothing.
941    async fn drive_once(
942        &self,
943        policy: &DeviceJoinApprovalPolicy<'_>,
944        access_administrator: Option<&dyn DeviceProviderAccessAdministrator>,
945        on_progress: &(dyn Fn(AdmittingDeviceJoinProgress) + Send + Sync),
946        timing: DeviceJoinTransportTiming,
947    ) -> Result<DeviceJoinDriveOutcome, DeviceJoinTransportError> {
948        loop {
949            match self.owner_status().await? {
950                // A row still at Abandoned is one whose terminal step did not
951                // finish. Delivering the artifact is what a driver started
952                // after the abandonment owes a joining device that has not seen
953                // it yet, and retiring the row behind it is the rest of that
954                // same step.
955                Some(DeviceJoinStatus::Abandoned { abandonment }) => {
956                    self.finish_abandonment(&abandonment).await?;
957                    return Ok(DeviceJoinDriveOutcome::Abandoned(abandonment));
958                }
959                Some(DeviceJoinStatus::SamePrincipalCompleted { join }) => {
960                    self.publish(DeviceJoinAction::TransferSamePrincipalJoin(join.clone()))
961                        .await?;
962                    return Ok(DeviceJoinDriveOutcome::Activated(join.activation));
963                }
964                Some(DeviceJoinStatus::AwaitingCompletion { activation }) => {
965                    self.publish(DeviceJoinAction::TransferActivation(activation.clone()))
966                        .await?;
967                    return Ok(DeviceJoinDriveOutcome::Activated(activation));
968                }
969                None | Some(DeviceJoinStatus::AwaitingAccessRequest { .. }) => {
970                    on_progress(AdmittingDeviceJoinProgress::WaitingForProviderAccessRequest);
971                    let request = self
972                        .await_artifact::<DeviceProviderAccessRequest>(timing)
973                        .await?;
974                    self.step(
975                        "approve access request",
976                        self.approve_access_request(&request, policy),
977                    )
978                    .await?;
979                    if request.offer.provider_admin.provider == request.peer_provider {
980                        on_progress(AdmittingDeviceJoinProgress::RegisteringDevice);
981                        let join = self
982                            .step(
983                                "activate same-provider device",
984                                self.activate_same_principal(request, access_administrator),
985                            )
986                            .await?;
987                        self.publish(DeviceJoinAction::TransferSamePrincipalJoin(join.clone()))
988                            .await?;
989                        return Ok(DeviceJoinDriveOutcome::Activated(join.activation));
990                    }
991                    on_progress(AdmittingDeviceJoinProgress::GrantingProviderAccess);
992                    let approval = self
993                        .step(
994                            "authorize provider access",
995                            self.store
996                                .authorize_device_provider_access(request, access_administrator),
997                        )
998                        .await?;
999                    self.publish(DeviceJoinAction::TransferProviderAdmissionApproval(
1000                        approval,
1001                    ))
1002                    .await?;
1003                }
1004                Some(
1005                    DeviceJoinStatus::AwaitingProviderAdmission { request }
1006                    | DeviceJoinStatus::ProviderAccessGrantCreatePending { request, .. },
1007                ) => {
1008                    on_progress(AdmittingDeviceJoinProgress::GrantingProviderAccess);
1009                    let approval = self
1010                        .step(
1011                            "authorize provider access",
1012                            self.store
1013                                .authorize_device_provider_access(request, access_administrator),
1014                        )
1015                        .await?;
1016                    self.publish(DeviceJoinAction::TransferProviderAdmissionApproval(
1017                        approval,
1018                    ))
1019                    .await?;
1020                }
1021                Some(DeviceJoinStatus::AwaitingRegistrationRequest { approval }) => {
1022                    self.publish(DeviceJoinAction::TransferProviderAdmissionApproval(
1023                        approval.clone(),
1024                    ))
1025                    .await?;
1026                    if matches!(approval.admission, DeviceProviderAdmission::SamePrincipal) {
1027                        let request = DeviceRegistrationRequest::same_principal(approval)
1028                            .map_err(DeviceJoinError::from)?;
1029                        on_progress(AdmittingDeviceJoinProgress::RegisteringDevice);
1030                        let join = self
1031                            .step(
1032                                "activate same-provider device",
1033                                self.store.resume_same_principal_device_join(request),
1034                            )
1035                            .await?;
1036                        self.publish(DeviceJoinAction::TransferSamePrincipalJoin(join.clone()))
1037                            .await?;
1038                        return Ok(DeviceJoinDriveOutcome::Activated(join.activation));
1039                    }
1040                    on_progress(AdmittingDeviceJoinProgress::WaitingForRegistrationRequest);
1041                    let request = self
1042                        .await_artifact::<DeviceRegistrationRequest>(timing)
1043                        .await?;
1044                    on_progress(AdmittingDeviceJoinProgress::RegisteringDevice);
1045                    self.accept_registration(request).await?;
1046                }
1047                Some(DeviceJoinStatus::AwaitingBootstrap { request }) => {
1048                    on_progress(AdmittingDeviceJoinProgress::RegisteringDevice);
1049                    if matches!(request, DeviceRegistrationRequest::SamePrincipal { .. }) {
1050                        let join = self
1051                            .step(
1052                                "activate same-provider device",
1053                                self.store.resume_same_principal_device_join(request),
1054                            )
1055                            .await?;
1056                        self.publish(DeviceJoinAction::TransferSamePrincipalJoin(join.clone()))
1057                            .await?;
1058                        return Ok(DeviceJoinDriveOutcome::Activated(join.activation));
1059                    }
1060                    self.accept_registration(request).await?;
1061                }
1062                Some(DeviceJoinStatus::SamePrincipalActivationCreatePending { request }) => {
1063                    on_progress(AdmittingDeviceJoinProgress::RegisteringDevice);
1064                    let join = self
1065                        .step(
1066                            "activate same-provider device",
1067                            self.store.resume_same_principal_device_join(request),
1068                        )
1069                        .await?;
1070                    self.publish(DeviceJoinAction::TransferSamePrincipalJoin(join.clone()))
1071                        .await?;
1072                    return Ok(DeviceJoinDriveOutcome::Activated(join.activation));
1073                }
1074                Some(DeviceJoinStatus::AwaitingChallengePublication { bootstrap }) => {
1075                    on_progress(AdmittingDeviceJoinProgress::PreparingLibrary);
1076                    let ready = self
1077                        .step(
1078                            "publish provider challenge",
1079                            self.store.publish_device_provider_challenge(bootstrap),
1080                        )
1081                        .await?;
1082                    self.publish(DeviceJoinAction::TransferProviderReadyBootstrap(ready))
1083                        .await?;
1084                }
1085                Some(DeviceJoinStatus::AwaitingReadiness { bootstrap }) => {
1086                    self.publish(DeviceJoinAction::TransferProviderReadyBootstrap(
1087                        bootstrap.clone(),
1088                    ))
1089                    .await?;
1090                    if matches!(
1091                        bootstrap.challenge_publication,
1092                        DeviceProviderChallengePublication::SamePrincipal
1093                    ) {
1094                        self.step(
1095                            "complete same-provider admission",
1096                            self.store
1097                                .complete_same_principal_device_admission(bootstrap),
1098                        )
1099                        .await?;
1100                        continue;
1101                    }
1102                    on_progress(AdmittingDeviceJoinProgress::WaitingForJoiningDevice);
1103                    let readiness = self.await_artifact::<DeviceJoinReadiness>(timing).await?;
1104                    on_progress(AdmittingDeviceJoinProgress::ActivatingDevice);
1105                    self.step(
1106                        "complete provider admission",
1107                        self.store.complete_device_provider_admission(readiness),
1108                    )
1109                    .await?;
1110                }
1111                Some(DeviceJoinStatus::AwaitingProviderCompletion { readiness }) => {
1112                    on_progress(AdmittingDeviceJoinProgress::ActivatingDevice);
1113                    self.step(
1114                        "complete provider admission",
1115                        self.store.complete_device_provider_admission(readiness),
1116                    )
1117                    .await?;
1118                }
1119                Some(DeviceJoinStatus::AwaitingActivation { completion }) => {
1120                    on_progress(AdmittingDeviceJoinProgress::ActivatingDevice);
1121                    let activation = self
1122                        .step(
1123                            "publish activation",
1124                            self.store.finalize_device_join(completion),
1125                        )
1126                        .await?;
1127                    self.publish(DeviceJoinAction::TransferActivation(activation.clone()))
1128                        .await?;
1129                    return Ok(DeviceJoinDriveOutcome::Activated(activation));
1130                }
1131                status => {
1132                    return Err(DeviceJoinError::Store(format!(
1133                        "device join {} has no admitting step from {status:?}",
1134                        self.attempt_id
1135                    ))
1136                    .into());
1137                }
1138            }
1139        }
1140    }
1141
1142    async fn accept_registration(
1143        &self,
1144        request: DeviceRegistrationRequest,
1145    ) -> Result<(), DeviceJoinTransportError> {
1146        self.step(
1147            "accept registration",
1148            self.store.accept_device_registration_request(request),
1149        )
1150        .await?;
1151        Ok(())
1152    }
1153
1154    /// Admit a device that uses this Store's provider account through one
1155    /// authorized writer. Each protocol transition is still journaled before
1156    /// the next begins, so a failure resumes through `drive_once`; keeping the
1157    /// writer open avoids reconstructing and re-verifying the same Store
1158    /// authority between consecutive transitions.
1159    async fn activate_same_principal(
1160        &self,
1161        request: DeviceProviderAccessRequest,
1162        access_administrator: Option<&dyn DeviceProviderAccessAdministrator>,
1163    ) -> Result<SamePrincipalDeviceJoin, DeviceJoinTransportError> {
1164        // Sixty-eight seconds hid behind this one step in a live run. It is
1165        // three provider-facing pieces, and they report as three.
1166        let mut timings = coven_foundation::stage_timing::StageTimings::counting(
1167            "Device join same-provider activation",
1168            self.store.provider_requests(),
1169        );
1170        let outcome = async {
1171            let mut writer = timings
1172                .stage("authorize writer", self.store.authorize_writer())
1173                .await
1174                .map_err(DeviceJoinError::from)?;
1175            let approval = timings
1176                .stage(
1177                    "authorize provider access",
1178                    writer
1179                        .join_operation()
1180                        .authorize_access(request, access_administrator),
1181                )
1182                .await?;
1183            let registration = DeviceRegistrationRequest::same_principal(approval)
1184                .map_err(DeviceJoinError::from)?;
1185            timings
1186                .stage(
1187                    "activate the join",
1188                    writer
1189                        .join_operation()
1190                        .activate_same_principal_join(registration),
1191                )
1192                .await
1193                .map_err(DeviceJoinTransportError::from)
1194        }
1195        .await;
1196        timings.report();
1197        outcome
1198    }
1199
1200    async fn approve_access_request(
1201        &self,
1202        request: &DeviceProviderAccessRequest,
1203        policy: &DeviceJoinApprovalPolicy<'_>,
1204    ) -> Result<(), DeviceJoinTransportError> {
1205        let offer = &self.bundle.offer;
1206        let approval = match policy {
1207            DeviceJoinApprovalPolicy::AutoApproveSelfIssued => {
1208                if self.self_issued().await? && request.offer.as_ref() == offer {
1209                    DeviceJoinApproval::Approve
1210                } else {
1211                    DeviceJoinApproval::Refuse
1212                }
1213            }
1214            DeviceJoinApprovalPolicy::Ask(ask) => ask(request),
1215        };
1216        match approval {
1217            DeviceJoinApproval::Approve => Ok(()),
1218            DeviceJoinApproval::Refuse => Err(DeviceJoinError::OfferMismatch.into()),
1219        }
1220    }
1221
1222    /// Whether this device issued the offer being admitted — the bound
1223    /// `AutoApproveSelfIssued` keeps to.
1224    ///
1225    /// Two facts decide it, both authoritative: this device is the offer's owner,
1226    /// and its own owner journal holds a record for this attempt. That record
1227    /// exists only because this device ran `begin_device_join` for it. A provider
1228    /// administrator that is a *different* device never satisfies this, so it
1229    /// prompts rather than admitting an offer it did not make.
1230    async fn self_issued(&self) -> Result<bool, DeviceJoinTransportError> {
1231        Ok(self.owner_status().await?.is_some())
1232    }
1233}
1234
1235/// How many times a driver re-derives after losing an activation slot, and how
1236/// long it waits before each retry.
1237///
1238/// A device holding the join also runs its sync loop, so the two publish Store
1239/// operations against the same positions. Losing that race persists nothing, so
1240/// the answer is to re-derive and go again — but only so many times: a store
1241/// that keeps refusing is not a race, and has to surface.
1242const ACTIVATION_CONFLICT_RETRIES: usize = 8;
1243const ACTIVATION_CONFLICT_BACKOFF: Duration = Duration::from_millis(25);
1244
1245/// Whether this failure is another writer having taken the activation slot
1246/// first — which persisted nothing, so the operation can simply be re-derived.
1247fn is_activation_conflict(error: &DeviceJoinTransportError) -> bool {
1248    matches!(
1249        error,
1250        DeviceJoinTransportError::DeviceJoin(DeviceJoinError::Outbound(
1251            crate::sync::store::StoreError::ActivationConflict
1252        ))
1253    )
1254}
1255
1256/// Run a driver pass, re-entering it when it loses an activation slot.
1257///
1258/// Every pass starts from the role journals, so a re-entry resumes rather than
1259/// repeating: the phases already settled are skipped and the one that lost the
1260/// race is re-derived against whatever the winner just committed. The backoff
1261/// grows so a busy store is not hammered, and the last failure propagates
1262/// unchanged once the budget is spent — this retries a lost race, it does not
1263/// paper over a wedged store.
1264async fn retrying_activation_conflicts<Pass, Fut, T>(
1265    mut pass: Pass,
1266) -> Result<T, DeviceJoinTransportError>
1267where
1268    Pass: FnMut() -> Fut,
1269    Fut: std::future::Future<Output = Result<T, DeviceJoinTransportError>>,
1270{
1271    // Each pass is boxed: a driver pass composes many large generators, and
1272    // holding one inline here would add its whole frame to this loop's own.
1273    for attempt in 0..ACTIVATION_CONFLICT_RETRIES {
1274        match Box::pin(pass()).await {
1275            Err(error) if is_activation_conflict(&error) => {
1276                tokio::time::sleep(ACTIVATION_CONFLICT_BACKOFF * (attempt as u32 + 1)).await;
1277            }
1278            settled => return settled,
1279        }
1280    }
1281    Box::pin(pass()).await
1282}
1283
1284impl From<coven_database::DeviceJoinJournalError> for DeviceJoinTransportError {
1285    fn from(error: coven_database::DeviceJoinJournalError) -> Self {
1286        DeviceJoinTransportError::from(super::DeviceJoinError::from(error))
1287    }
1288}
1289
1290#[cfg(test)]
1291#[path = "transport_tests.rs"]
1292mod tests;
coven_replication/sync/store/device_join/transport.rs

coven_replication/sync/store/device_join/
transport.rs
