coven/
storage_setup.rs

1//! Storage setup as an operation (§11.6, §20.8). Setup journals itself
2//! before its first write to storage, keeping the credentials it connects
3//! with apart from the committed ones. Until its point of no return a
4//! cancel or a failure undoes it, and opening the store undoes one a crash
5//! cut. Past that point it finishes: in the call, or after a crash in the
6//! first connection attempt, with the credentials it kept. Only finishing
7//! commits the connection.
8
9use super::*;
10use coven_crypto::custody::StoreKeychain;
11use coven_database::{OperationOutcome, OperationRejection, UndoCause};
12use coven_sync::SetupJournal;
13
14/// Where this session's storage setup stands, which a cancel reads (E5).
15pub(crate) enum SetupRun {
16    /// No setup runs.
17    Idle,
18    /// A setup call has started and journaled nothing yet; `cancelled` once
19    /// the app asked to cancel it, which it keeps until it journals.
20    Called { cancelled: bool },
21    /// A setup checks the provider; `cancelled` once the app asked to
22    /// cancel it.
23    Checking { cancelled: bool },
24    /// A setup is past its point of no return: this session's, or one a
25    /// crash cut, which the first connection finishes.
26    Finishing,
27}
28
29/// A setup call from its start until it journals itself, when a cancel
30/// already reaches it (§11.6). A call that ends before its journal ends it.
31pub(super) struct SetupCall(Arc<StorageConnections>);
32
33impl SetupCall {
34    /// The call has journaled itself: it checks the provider now, keeping
35    /// a cancel that came before. Answers whether one did.
36    fn journaled(self) -> Result<bool, StorageSetupError> {
37        let mut run = lock_unpoisoned(&self.0.setup);
38        match *run {
39            SetupRun::Called { cancelled } => {
40                *run = SetupRun::Checking { cancelled };
41                Ok(cancelled)
42            }
43            SetupRun::Idle | SetupRun::Checking { .. } | SetupRun::Finishing => {
44                Err(setup_error(DbError::Invariant { what: "a setup call runs alone" }.into()))
45            }
46        }
47    }
48}
49
50impl Drop for SetupCall {
51    fn drop(&mut self) {
52        let mut run = lock_unpoisoned(&self.0.setup);
53        if matches!(*run, SetupRun::Called { .. }) {
54            *run = SetupRun::Idle;
55        }
56    }
57}
58
59/// What cancelling storage setup did (E5).
60#[derive(Clone, Copy, Debug, PartialEq, Eq)]
61pub enum SetupCancellation {
62    /// Setup hadn't reached its point of no return: it undoes what it
63    /// wrote and fails with `StorageSetupError::Cancelled`.
64    Undoing,
65    /// Setup is past its point of no return, and finishes.
66    Finishing,
67    /// No setup is running.
68    NoSetup,
69}
70
71/// A place between two of setup's steps, where a test stops it as a crash
72/// would, or holds it.
73#[derive(Clone, Copy, Debug, PartialEq, Eq)]
74pub(crate) enum SetupStop {
75    /// Its call has started and journaled nothing.
76    Called,
77    /// Its row and credentials are journaled.
78    Journaled,
79    /// S3 detection is done.
80    Detected,
81    /// The provider check is done.
82    Checked,
83    /// A creation's store key is kept in custody.
84    KeyKept,
85    /// Its row is past the point of no return.
86    Claiming,
87    /// What it claims has landed.
88    Created,
89    /// The connection is committed.
90    Committed,
91    /// Its row has ended.
92    Finished,
93}
94
95/// What a test does to the next setup at one of its stops.
96#[cfg(test)]
97pub(crate) enum SetupHook {
98    /// Stop there as a crash would: the call returns at once, undoing and
99    /// finishing nothing.
100    Crash(SetupStop),
101    /// Say it got there on `entered`, then wait for `resume`.
102    Pause {
103        at: SetupStop,
104        entered: tokio::sync::oneshot::Sender<()>,
105        resume: tokio::sync::oneshot::Receiver<()>,
106    },
107}
108
109/// The error a crash hook stops setup with.
110#[cfg(test)]
111#[derive(Debug, thiserror::Error)]
112#[error("setup stopped as a crash would")]
113struct Crashed;
114
115/// Whether `error` is a crash hook's.
116#[cfg(test)]
117fn crashed(error: &(dyn std::error::Error + 'static)) -> bool {
118    let mut cause = Some(error);
119    while let Some(error) = cause {
120        if error.is::<Crashed>() {
121            return true;
122        }
123        cause = error.source();
124    }
125    false
126}
127
128/// The connections a setup past its point of no return finishes on: the
129/// store's, which it commits, and the one its own requests go through.
130struct FinishingConnections {
131    storage: Arc<StorageConnection>,
132    setup: Arc<StorageConnection>,
133}
134
135/// A setup failure, and the undo or start that failed after it.
136#[derive(Debug, thiserror::Error)]
137#[error("{operation}; cleanup failed: {cleanup}")]
138struct SetupCleanup {
139    operation: StorageSetupError,
140    #[source]
141    cleanup: SyncError,
142}
143
144pub(super) fn with_cleanup(operation: StorageSetupError, cleanup: SyncError) -> StorageSetupError {
145    StorageSetupError::Internal(Box::new(SetupCleanup { operation, cleanup }))
146}
147
148/// Settle a setup a crash cut, as the store opens (§11.6). One before its
149/// point of no return is undone: a creation's store key and its credentials
150/// go, and `coven_operations` lists it undone. One past it stays for the first connection to finish.
151/// Credentials a finished setup left behind go too. Answers whether a setup
152/// is finishing.
153pub(crate) async fn settle_opened(
154    database: &coven_database::Database,
155    keychain: &Arc<StoreKeychain>,
156    keys: &Arc<KeySession<StoreKeyring>>,
157) -> Result<bool, SyncError> {
158    let running = SetupJournal::running(database).await?;
159    if running.as_ref().is_some_and(|journal| journal.location().is_some()) {
160        return Ok(true);
161    }
162    let creating = running.as_ref().is_some_and(SetupJournal::creating);
163    let (kept, held) = (keychain.clone(), keys.clone());
164    let bugs = database.bugs().clone();
165    crate::task::blocking_completing("undo storage setup's local state", bugs, move || {
166        // Before the point of no return nothing claimed the store key a
167        // creation kept.
168        if creating {
169            held.forget()?;
170        }
171        if kept.setup_credentials()?.is_some() {
172            kept.delete_setup_credentials()?;
173        }
174        Ok::<_, KeyError>(())
175    })
176    .await??;
177    if let Some(journal) = running {
178        journal.end(database, OperationOutcome::Undone(UndoCause::Interrupted)).await?;
179    }
180    Ok(false)
181}
182
183impl StorageConnections {
184    /// Cancel this session's setup (E5): before its point of no return it
185    /// undoes what it wrote; past it, it finishes.
186    pub(crate) fn cancel_setup(&self) -> SetupCancellation {
187        let answer = match &mut *lock_unpoisoned(&self.setup) {
188            SetupRun::Idle => SetupCancellation::NoSetup,
189            SetupRun::Called { cancelled } | SetupRun::Checking { cancelled } => {
190                *cancelled = true;
191                SetupCancellation::Undoing
192            }
193            SetupRun::Finishing => SetupCancellation::Finishing,
194        };
195        self.setup_cancelled.notify_waiters();
196        answer
197    }
198
199    /// Whether a setup is past its point of no return and not yet finished.
200    pub(super) fn setup_finishing(&self) -> bool {
201        matches!(*lock_unpoisoned(&self.setup), SetupRun::Finishing)
202    }
203
204    pub(super) fn check_cancel(&self) -> Result<(), StorageSetupError> {
205        match *lock_unpoisoned(&self.setup) {
206            SetupRun::Checking { cancelled: true } => Err(StorageSetupError::Cancelled),
207            _ => Ok(()),
208        }
209    }
210
211    /// Resolves once the app has cancelled the setup checking now.
212    async fn cancelled(&self) {
213        loop {
214            let notified = self.setup_cancelled.notified();
215            tokio::pin!(notified);
216            notified.as_mut().enable();
217            if self.check_cancel().is_err() {
218                return;
219            }
220            notified.await;
221        }
222    }
223
224    /// Where a test stops or holds setup.
225    pub(super) async fn reached(
226        &self,
227        #[cfg_attr(not(test), expect(unused_variables, reason = "only tests stop setup"))]
228        at: SetupStop,
229    ) -> Result<(), StorageError> {
230        #[cfg(test)]
231        {
232            let hook = {
233                let mut held = lock_unpoisoned(&self.setup_hook);
234                match &*held {
235                    Some(SetupHook::Crash(stop) | SetupHook::Pause { at: stop, .. })
236                        if *stop == at =>
237                    {
238                        held.take()
239                    }
240                    _ => None,
241                }
242            };
243            match hook {
244                Some(SetupHook::Crash(_)) => {
245                    return Err(StorageFailure::Network.with_source(Crashed));
246                }
247                Some(SetupHook::Pause { entered, resume, .. }) => {
248                    entered.send(()).expect("the test waits for the pause");
249                    resume.await.expect("the test resumes the pause");
250                }
251                None => {}
252            }
253        }
254        Ok(())
255    }
256
257    /// Have the next setup stop or wait at one of its stops.
258    #[cfg(test)]
259    pub(crate) fn test_setup_hook(&self, hook: SetupHook) {
260        *lock_unpoisoned(&self.setup_hook) = Some(hook);
261    }
262
263    /// Stop the next S3 key replacement after `at`, as a crash would.
264    #[cfg(test)]
265    pub(crate) fn test_replace_crash(&self, at: coven_sync::ReplaceStop) {
266        self.codes.test_replace_crash(at);
267    }
268
269    /// Start a setup call, which a cancel reaches from now on (§11.6).
270    /// Refused while another runs, or one past its point of no return is
271    /// still finishing.
272    pub(super) fn start_setup(self: &Arc<Self>) -> Result<SetupCall, StorageSetupError> {
273        let mut run = lock_unpoisoned(&self.setup);
274        if !matches!(*run, SetupRun::Idle) {
275            return Err(StorageSetupError::SetupRunning);
276        }
277        *run = SetupRun::Called { cancelled: false };
278        Ok(SetupCall(self.clone()))
279    }
280
281    /// Journal `call`'s setup before its first write to storage: its row,
282    /// then `credentials`, kept apart from the committed ones (§11.6). A
283    /// cancel that came before the row undoes it at once.
284    pub(super) async fn begin_setup(
285        &self,
286        call: SetupCall,
287        device_name: &str,
288        credentials: &StorageCredentials,
289    ) -> Result<SetupJournal, StorageSetupError> {
290        self.reached(SetupStop::Called).await?;
291        let creating = self.codes.location().await.map_err(setup_error)?.is_none();
292        let journal = SetupJournal::start(&self.database, creating, device_name)
293            .await
294            .map_err(setup_error)?;
295        match call.journaled() {
296            Ok(false) => {}
297            Ok(true) => return Err(self.undo_setup(&journal, StorageSetupError::Cancelled).await),
298            Err(error) => return Err(self.undo_setup(&journal, error).await),
299        }
300        let kept = credentials
301            .encode()
302            .map_err(StorageSetupError::from)
303            .and_then(|bytes| Ok(self.keychain.set_setup_credentials(&bytes)?));
304        if let Err(error) = kept {
305            return Err(self.undo_setup(&journal, error).await);
306        }
307        self.reached(SetupStop::Journaled).await?;
308        Ok(journal)
309    }
310
311    /// Run setup's `steps` under `journal`. A failure before the point of no
312    /// return undoes setup. One after it leaves setup finishing, unless a
313    /// refusal undid it, and starts the loop, whose connection finishes it
314    /// (§11.6).
315    pub(super) async fn run_setup<T>(
316        self: &Arc<Self>,
317        mut journal: SetupJournal,
318        steps: impl AsyncFnOnce(&mut SetupJournal) -> Result<T, StorageSetupError>,
319    ) -> Result<T, StorageSetupError> {
320        let error = match steps(&mut journal).await {
321            Ok(done) => return Ok(done),
322            Err(error) => error,
323        };
324        #[cfg(test)]
325        if crashed(&error) {
326            return Err(error);
327        }
328        if journal.location().is_none() {
329            return Err(self.undo_setup(&journal, error).await);
330        }
331        if self.setup_finishing() {
332            if let Err(cleanup) = self.start_locked().await {
333                return Err(with_cleanup(error, cleanup));
334            }
335        }
336        Err(error)
337    }
338
339    /// Undo a setup before its point of no return: its credentials go, then
340    /// its row ends cancelled, or undone with `error` (§11.6). Nothing it
341    /// wrote to storage is state: a test object or a probe is space.
342    async fn undo_setup(
343        &self,
344        journal: &SetupJournal,
345        error: StorageSetupError,
346    ) -> StorageSetupError {
347        let outcome = match &error {
348            StorageSetupError::Cancelled => OperationOutcome::Cancelled,
349            error => OperationOutcome::Undone(UndoCause::Failed { error: error.to_string() }),
350        };
351        let undone = async {
352            // Before the point of no return nothing claimed the store key a
353            // creation kept.
354            if journal.creating() {
355                self.forget_keys()?;
356            }
357            self.keychain.delete_setup_credentials()?;
358            journal.end(&self.database, outcome).await?;
359            *lock_unpoisoned(&self.setup) = SetupRun::Idle;
360            Ok::<_, SyncError>(())
361        }
362        .await;
363        match undone {
364            Ok(()) => error,
365            Err(cleanup) => with_cleanup(error, cleanup),
366        }
367    }
368
369    /// Pass setup's point of no return, claiming at `location`, unless the
370    /// app has cancelled it: from here on a cancel is answered `Finishing`.
371    async fn pass_point(
372        &self,
373        journal: &mut SetupJournal,
374        location: StorageConfig,
375    ) -> Result<(), StorageSetupError> {
376        {
377            let mut run = lock_unpoisoned(&self.setup);
378            if matches!(*run, SetupRun::Checking { cancelled: true }) {
379                return Err(StorageSetupError::Cancelled);
380            }
381            *run = SetupRun::Finishing;
382        }
383        if let Err(error) = journal.claim(&self.database, location).await {
384            *lock_unpoisoned(&self.setup) = SetupRun::Checking { cancelled: false };
385            return Err(setup_error(error));
386        }
387        Ok(())
388    }
389
390    /// Check and connect `config` reached through `auth`, under `journal`.
391    /// A new OAuth sign-in is `sign_in`, the session custody `auth`
392    /// refreshes in until setup commits it to device custody. Reconnecting
393    /// then loads what other devices published; a cancel still undoes setup
394    /// meanwhile. Then setup passes its point of no return and finishes.
395    /// The setup call's own requests, its checks and its claims, go through
396    /// a connection made beside the store's that takes the call's `waits`
397    /// and sends rate-limited requests again while they allow; the store's
398    /// connection, which the sequence, the files and the sync loop keep,
399    /// never does (§20.8). Its steps go to `progress`, from `Connecting` on.
400    pub(super) async fn setup_connection(
401        &self,
402        journal: &mut SetupJournal,
403        config: StorageConfig,
404        auth: ProviderAuth,
405        sign_in: Option<Arc<SessionSignIn>>,
406        waits: SetupWaits,
407        progress: &mut ProgressReport<StorageSetupStep>,
408    ) -> Result<StorageSetup, StorageSetupError> {
409        progress.begin(StorageSetupStep::Connecting, 1);
410        let credentials = auth.credentials();
411        let storage = self
412            .connector
413            .connect(
414                config.clone(),
415                auth,
416                self.device,
417                self.state.clone(),
418                self.waits.clone(),
419                self.transfers.clone(),
420            )
421            .await?;
422        let setup = Arc::new(storage.during_setup(waits));
423        if !journal.creating() {
424            setup.check_location_claim(self.store()).await?;
425        }
426        // The folder's label as this sign-in reads it, never one the app
427        // passed (§20.2).
428        let config = setup.labelled_config().await?;
429        let path = ObjectPath::file(self.store(), self.device, ObjectId(self.ids.new_id()));
430        let plaintext = b"coven storage check";
431        let header = coven_format::file::FileHeader::new(plaintext.len() as u64);
432        let key = coven_crypto::FileKey::generate()
433            .map_err(|error| StorageSetupError::Internal(Box::new(error)))?;
434        let mut bytes = header.encode().to_vec();
435        bytes.extend(
436            header
437                .seal_chunk(&key, path.as_str(), 0, plaintext)
438                .map_err(|error| StorageSetupError::Internal(Box::new(error)))?,
439        );
440        {
441            let _permit = self.state.permit_provider_check(&path);
442            coven_storage::check_provider(setup.as_ref(), &path, &bytes).await?;
443        }
444        progress.done(1);
445        self.reached(SetupStop::Checked).await?;
446        self.check_cancel()?;
447        if journal.creating() {
448            // A custody failure here still undoes setup with nothing claimed.
449            self.keep_creation_key().map_err(setup_error)?;
450            self.reached(SetupStop::KeyKept).await?;
451        }
452        let mut live = self.live.lock().await;
453        let sync = self.sequence(storage.clone(), &live).await.map_err(setup_error)?;
454        // The claims, and the catch-up an existing store needs, are the
455        // setup call's own requests, sent on its connection.
456        let claiming = Arc::new(sync.reconnect(setup.clone()));
457        if !journal.creating() {
458            let relayed = std::sync::Mutex::new(&mut *progress);
459            let loading = claiming.subscribe_progress().forward(claiming.unlock(), |load| {
460                lock_unpoisoned(&relayed).relay(setup_step::loading(load))
461            });
462            // Loading claims nothing, so a cancel still undoes setup.
463            tokio::select! {
464                biased;
465                () = self.cancelled() => return Err(StorageSetupError::Cancelled),
466                loaded = loading => loaded.map_err(setup_error)?,
467            }
468        }
469        self.pass_point(journal, config.clone()).await?;
470        self.reached(SetupStop::Claiming).await?;
471        let restore_code = self
472            .finish_setup(
473                &mut live,
474                journal,
475                sync,
476                claiming,
477                FinishingConnections { storage, setup },
478                credentials,
479                sign_in,
480                false,
481                true,
482                progress,
483            )
484            .await
485            .map_err(setup_error)?;
486        Ok(StorageSetup {
487            connection: ConnectedStorage { storage: config, key_state: StoreKeyState::Available },
488            restore_code,
489        })
490    }
491
492    /// Finish a setup a crash cut past its point of no return, connecting
493    /// with the credentials it kept, as the first connection after opening
494    /// does (§11.6). An OAuth sign-in refreshes in this session's custody
495    /// until finishing commits it.
496    pub(super) async fn resume_setup(
497        &self,
498        live: &mut ConnectionState,
499        journal: SetupJournal,
500    ) -> Result<(), SyncError> {
501        let location = journal
502            .location()
503            .ok_or(DbError::Invariant { what: "a finishing setup is past its point of no return" })?
504            .clone();
505        let bytes = self
506            .keychain
507            .setup_credentials()?
508            .ok_or(DbError::Invariant { what: "a finishing setup keeps its credentials" })?;
509        let credentials = StorageCredentials::decode(bytes.as_bytes())?;
510        let (auth, sign_in) = match &credentials {
511            StorageCredentials::S3(keys) => (ProviderAuth::S3(keys.clone()), None),
512            StorageCredentials::CloudKit => (ProviderAuth::CloudKit, None),
513            StorageCredentials::OAuth(tokens) => {
514                let sign_in = SessionSignIn::new(tokens.clone(), self.clients.clone());
515                let auth = ProviderAuth::OAuth { tokens: tokens.clone(), custody: sign_in.clone() };
516                (auth, Some(sign_in))
517            }
518        };
519        let storage = self
520            .connector
521            .connect(
522                location,
523                auth,
524                self.device,
525                self.state.clone(),
526                self.waits.clone(),
527                self.transfers.clone(),
528            )
529            .await?;
530        let sync = self.sequence(storage.clone(), live).await?;
531        let claiming = Arc::new(sync.reconnect(storage.clone()));
532        let mut progress =
533            ProgressReport::new(tokio::sync::watch::channel(None).0, self.clock.clone());
534        self.finish_setup(
535            live,
536            &journal,
537            sync,
538            claiming,
539            FinishingConnections { storage: storage.clone(), setup: storage },
540            credentials,
541            sign_in,
542            true,
543            false,
544            &mut progress,
545        )
546        .await
547        .map(drop)
548    }
549
550    /// Past the point of no return: claim what creating the store or a
551    /// changed access needs on the candidate's own connection, commit its
552    /// store connection, end the
553    /// journal row with its outcome and install `sync`, starting its loop
554    /// when `start` (§11.6). A reconnect resumed after a crash catches up
555    /// first, when `catch_up`. A claim refused before it landed undoes
556    /// setup.
557    #[expect(
558        clippy::too_many_arguments,
559        reason = "the call and a crash's resumption differ in each"
560    )]
561    async fn finish_setup(
562        &self,
563        live: &mut ConnectionState,
564        journal: &SetupJournal,
565        sync: Arc<SequenceSync>,
566        claiming: Arc<SequenceSync>,
567        candidate: FinishingConnections,
568        credentials: StorageCredentials,
569        sign_in: Option<Arc<SessionSignIn>>,
570        catch_up: bool,
571        start: bool,
572        progress: &mut ProgressReport<StorageSetupStep>,
573    ) -> Result<SecretText, SyncError> {
574        let location = journal
575            .location()
576            .ok_or(DbError::Invariant { what: "a finishing setup is past its point of no return" })?
577            .clone();
578        let provider = location.provider();
579        let FinishingConnections { storage, setup } = candidate;
580        let access = member_access(&credentials, &setup).await?;
581        // The old provider remains installed until every fallible step
582        // succeeds. Its stopped or started state survives a refused claim.
583        let prepared = Self::with_passes_held(live, async {
584            let claimed = if journal.creating() {
585                progress.begin(StorageSetupStep::CreatingStore, 1);
586                match setup.claim_location(self.store()).await {
587                    Ok(()) => {
588                        self.claim_creation(&claiming, access, journal.device_name(), progress)
589                            .await
590                    }
591                    Err(error) => Err(error.into()),
592                }
593            } else {
594                self.claim_access(&claiming, access, catch_up, progress).await
595            };
596            let slot = match claimed {
597                Ok(slot) => slot,
598                Err(error) if refused(&error) => {
599                    return Err(self.refuse_setup(journal, error).await)
600                }
601                Err(error) => return Err(error),
602            };
603            self.reached(SetupStop::Created).await?;
604            let (commit, restore_code) = self
605                .codes
606                .prepare_setup(ConnectionCredentials { location, credentials }, sign_in)
607                .await?;
608            // Claims may use file transfers; only the custody commit runs
609            // inside the transfer exclusion needed to replace their provider.
610            self.files.set_storage(Some(storage), async { commit().map(drop) }).await?;
611            self.reached(SetupStop::Committed).await?;
612            Ok::<_, SyncError>((slot, restore_code))
613        })
614        .await;
615        let (slot, restore_code) = prepared?;
616        let outcome = match slot {
617            Some(slot) => OperationOutcome::Landed(slot),
618            None => OperationOutcome::Satisfied,
619        };
620        journal.end(&self.database, outcome).await?;
621        *lock_unpoisoned(&self.setup) = SetupRun::Idle;
622        self.reached(SetupStop::Finished).await?;
623        // The row has ended, so these are credentials nothing uses; opening
624        // deletes them if this fails.
625        if let Err(error) = self.keychain.delete_setup_credentials() {
626            tracing::warn!("a finished setup's own credentials can't be deleted yet: {error}");
627        }
628        let released = self.close_live(live).await;
629        let started = self.activate(live, sync, start).await;
630        // Custody and provider are committed, and setup read the store to its
631        // end. Worker failures now belong to the sync state; setup must
632        // return the connection it published. A row that can't be written
633        // has failed the handle, which every later call reports; what it
634        // would have recorded goes to the log.
635        let recorded = coven_database::RecordedLocation::Connected(provider);
636        if let Err(written) = self.record_location(recorded, Some(self.clock.now())).await {
637            tracing::warn!("storage set up after setup can't be recorded: {written}");
638        }
639        let result = match (released, started) {
640            (Ok(()), result) | (result, Ok(())) => result,
641            (Err(operation), Err(cleanup)) => Err(SyncError::cleanup(operation, cleanup)),
642        };
643        if let Err(error) = result {
644            let failure = LastPass::failed(error);
645            if let Err(written) = self.show(failure.clone()).await {
646                tracing::warn!(
647                    ?failure,
648                    "how sync failed after setup can't be recorded: {written}"
649                );
650            }
651        }
652        Ok(restore_code)
653    }
654
655    /// Create the store, once its location is claimed where it holds one
656    /// store (§20.3): keep its first store key, then claim slot 1 and the
657    /// creator's key copy (§10.3). A crash may have left the creation journaled or
658    /// applied: it is settled first, and an applied one claims nothing more.
659    async fn claim_creation(
660        &self,
661        sync: &Arc<SequenceSync>,
662        access: coven_format::MemberAccess,
663        device_name: &str,
664        progress: &mut ProgressReport<StorageSetupStep>,
665    ) -> Result<Option<coven_merge::Slot>, SyncError> {
666        let member = self.identity.read()?.ok_or(StorageFailure::MemberKeysMissing)?;
667        let ring = self.keep_creation_key()?;
668        let key = ring.store_key_ids().next().ok_or(StorageFailure::MemberKeysMissing)?;
669        let key_hash = ring.store_key(key)?.hash();
670        sync.settle_operations().await?;
671        if self.database.store_state().await?.store.is_none() {
672            sync.administer_at(coven_format::admin::StoreChange::CreateStore {
673                store: self.store(),
674                name: self.database.store_name().await.map_err(SyncError::from)?,
675                admin: member.keys().public_keys(),
676                access,
677                device_name: device_name.into(),
678                key,
679                key_hash,
680            })
681            .await?;
682        }
683        progress.done(1);
684        Ok(Some(coven_merge::Slot::FIRST))
685    }
686
687    /// The store key a creation's slot 1 introduces, kept in custody before
688    /// setup's point of no return: the creator's sequence owner seals slot 1
689    /// and its key copy with the store key custody holds. One kept by this
690    /// setup, or by one a crash cut, is the same key.
691    fn keep_creation_key(&self) -> Result<StoreKeyring, SyncError> {
692        let ring = match self.with_keys(|keys| keys.read())? {
693            Some(ring) => ring,
694            None => StoreKeyring::new(coven_crypto::StoreKey::generate(KeyId(self.ids.new_id()))?),
695        };
696        self.persist_keys(&ring)?;
697        Ok(ring)
698    }
699
700    /// Record changed access for an existing device (§10.1, §14.2), once a
701    /// reconnect resumed after a crash has caught up, when `catch_up`.
702    async fn claim_access(
703        &self,
704        sync: &Arc<SequenceSync>,
705        access: coven_format::MemberAccess,
706        catch_up: bool,
707        progress: &mut ProgressReport<StorageSetupStep>,
708    ) -> Result<Option<coven_merge::Slot>, SyncError> {
709        if catch_up {
710            sync.unlock().await?;
711        }
712        let member = self.identity.read()?.ok_or(StorageFailure::MemberKeysMissing)?;
713        let state = self.database.store_state().await?;
714        match access_change(&state, member.member(), access)? {
715            Some(change) => {
716                let recording = sync.administer_at(change);
717                setup_step::single(progress, StorageSetupStep::RecordingAccess, recording).await
718            }
719            None => Ok(None),
720        }
721    }
722
723    /// Undo a setup whose claim was refused before anything it claimed
724    /// landed: the store key it kept for a creation no claim can still use
725    /// goes, then its credentials, and its row ends with the refusal
726    /// (§11.2, §11.6).
727    async fn refuse_setup(&self, journal: &SetupJournal, error: SyncError) -> SyncError {
728        let outcome = match &error {
729            SyncError::Rejected(reason) => OperationOutcome::Rejected(reason.clone()),
730            SyncError::StoreOwner => OperationOutcome::Rejected(OperationRejection::StoreOwner),
731            error => OperationOutcome::Undone(UndoCause::Failed { error: error.to_string() }),
732        };
733        let undone = async {
734            if journal.creating()
735                && self.database.current_claim().await?.is_none()
736                && self.database.store_state().await?.store.is_none()
737            {
738                self.forget_keys()?;
739            }
740            self.keychain.delete_setup_credentials()?;
741            journal.end(&self.database, outcome).await?;
742            *lock_unpoisoned(&self.setup) = SetupRun::Idle;
743            Ok::<_, SyncError>(())
744        }
745        .await;
746        match undone {
747            Ok(()) => error,
748            Err(cleanup) => SyncError::cleanup(error, cleanup),
749        }
750    }
751}
752
753/// Whether a claim past the point of no return was refused before it
754/// landed: another store holds the location or slot 1, or the caught-up
755/// history refuses the access change. Any other failure is retried.
756fn refused(error: &SyncError) -> bool {
757    match error {
758        SyncError::Rejected(_) | SyncError::StoreOwner => true,
759        SyncError::Storage(error) => matches!(
760            error.failure(),
761            StorageFailure::StoreExists | StorageFailure::OwnLocationRequired
762        ),
763        _ => false,
764    }
765}
766
767#[cfg(test)]
768#[path = "storage_setup_tests.rs"]
769mod tests;
coven/storage_setup.rs

coven/
storage_setup.rs
