coven/
storage.rs

1//! Application storage setup, credential custody and connection control.
2
3use crate::{
4    error::{setup_error, unlock_error},
5    *,
6};
7use coven_crypto::custody::{KeySession, PersonEntries};
8use coven_foundation::locks::lock_unpoisoned;
9use coven_foundation::progress::ProgressReport;
10use coven_storage::{
11    providers::{OAuthExchange, OAuthFlow, StorageConnector},
12    ConnectionCredentials, ProviderAuth, PublicationGate, SetupWaits, Storage, StorageConnection,
13    StorageCredentials,
14};
15use coven_sync::{
16    DeviceCredentials, LastPass, RestoreCodes, SequenceState, SequenceSync, SessionSignIn, SyncLoop,
17};
18use std::{
19    future::Future,
20    sync::{
21        atomic::{AtomicBool, AtomicU64, Ordering},
22        Arc,
23    },
24};
25
26/// Whether this device's custody holds opened store keys.
27#[derive(Clone, Copy, Debug, PartialEq, Eq)]
28pub enum StoreKeyState {
29    /// Store keys are held by custody.
30    Available,
31    /// Store keys must be opened from this member's sealed copy.
32    Locked,
33}
34
35/// The configured location and the result of opening this device's keys.
36#[derive(Clone, Debug)]
37pub struct ConnectedStorage {
38    /// The connected provider location.
39    pub storage: StorageConfig,
40    /// This device's key availability.
41    pub key_state: StoreKeyState,
42}
43
44/// Storage setup's committed connection and this member's restore code.
45#[derive(Clone, Debug)]
46pub struct StorageSetup {
47    /// The location and availability of its store keys.
48    pub connection: ConnectedStorage,
49    /// The code for restoring this member on another installation.
50    pub restore_code: SecretText,
51}
52
53/// Where a setup call reports its steps, for the handle to forward to the app.
54pub(crate) type SetupProgress = tokio::sync::watch::Sender<Option<Progress<StorageSetupStep>>>;
55
56/// Write what the sync state starts from as a store opens: whether a
57/// storage location is recorded here and its credentials held, that no pass
58/// moves, and, for a store joining or restoring read to its end, when sync
59/// finished (E5); then end each invite whose lifetime ended while it was
60/// closed (§13.2). Opening writes them before the handle exists, so no app
61/// reads a move a process that closed left behind, or an invite that has
62/// expired.
63pub(crate) async fn record_opened(
64    database: &coven_database::Database,
65    clock: &ClockRef,
66    location: coven_database::RecordedLocation<CloudProvider>,
67    finished: Option<std::time::SystemTime>,
68) -> Result<(), SyncError> {
69    coven_sync::record_opened(database, location, finished).await.map_err(SyncError::from)?;
70    coven_sync::InviteExpiry::opened(database, clock.now()).await.map_err(SyncError::from)
71}
72
73/// A read-only handle publishes nothing.
74pub(crate) struct NoPublications;
75
76impl PublicationGate for NoPublications {
77    fn check(&self, _: &ObjectPath) -> Result<(), StorageError> {
78        Err(StorageFailure::PermissionDenied.with_source("a read-only handle publishes nothing"))
79    }
80}
81
82/// One provider connection's sequence owner and loop.
83struct Live {
84    sync: Arc<SequenceSync>,
85    sync_loop: SyncLoop,
86}
87
88/// Bootstrap retains its loaded provider before its sequence worker is needed.
89enum ConnectionState {
90    Released,
91    Loaded(Arc<StorageConnection>),
92    Active(Live),
93}
94
95struct ConnectionAttempt {
96    epoch: u64,
97    task: crate::task::OwnedTask,
98    /// Each time the attempt has taken everything that can wake its
99    /// backoff and waits.
100    #[cfg(test)]
101    parks: coven_sync::Parks,
102}
103
104pub(crate) struct StorageConnections {
105    /// The store's credential owner, connected or not; a change that needs
106    /// the connection is given its sequence owner.
107    codes: RestoreCodes,
108    database: coven_database::Database,
109    files: coven_sync::Files,
110    identity: Arc<KeySession<MemberIdentity>>,
111    /// The person's entries, for each connection's sequence owner.
112    person: Arc<PersonEntries>,
113    credentials: Arc<DeviceCredentials>,
114    /// The sequence state whose gate every connection publishes through.
115    state: Arc<SequenceState>,
116    directory: StoreDir,
117    waits: Arc<coven_sync::RetryWaits>,
118    transfers: Arc<coven_storage::TransferBudget>,
119    keys: std::sync::Mutex<Option<Arc<KeySession<StoreKeyring>>>>,
120    connector: Arc<dyn StorageConnector>,
121    oauth: Option<OAuthFlow>,
122    /// The app's clients, which refresh a session sign-in.
123    clients: Option<OAuthExchange>,
124    /// The latest sign-in, held for this session until setup publishes it.
125    authentication: tokio::sync::Mutex<Option<Arc<SessionSignIn>>>,
126    /// The connection, if any; held while it changes or works.
127    live: tokio::sync::Mutex<ConnectionState>,
128    /// Ends invites as their lifetimes end, connected or not (§13.2).
129    expiry: coven_sync::InviteExpiry,
130    /// The connection's loop, for requests that return at once.
131    control: std::sync::Mutex<Option<SyncLoop>>,
132    /// Whether custody held storage credentials when last recorded: start
133    /// connects only then, without reading custody before the first attempt.
134    configured: AtomicBool,
135    /// Changed by a new start, stop or close: a connection attempt or a
136    /// queued stop acts only while no later request has superseded it.
137    epoch: AtomicU64,
138    /// Wakes a connection attempt's backoff when stopping is requested.
139    stopped: tokio::sync::Notify,
140    /// The started loop's connection attempts, awaited when closing.
141    attempts: std::sync::Mutex<Option<ConnectionAttempt>>,
142    device: DeviceId,
143    ids: IdSourceRef,
144    clock: ClockRef,
145    calls: tokio::sync::Mutex<()>,
146    /// How many times a call has found the call lock held and waited.
147    #[cfg(test)]
148    call_waits: tokio::sync::watch::Sender<usize>,
149    /// A failure the next close returns after closing, for a test of what
150    /// follows a failed close.
151    #[cfg(test)]
152    close_failure: std::sync::Mutex<Option<SyncError>>,
153    /// Run where the next deletion is about to write its record.
154    #[cfg(test)]
155    deletion_gate: std::sync::Mutex<Option<Box<dyn FnOnce() + Send>>>,
156    /// This store's device-only entries, where a setup keeps its own
157    /// credentials until it finishes (§11.6).
158    keychain: Arc<coven_crypto::custody::StoreKeychain>,
159    /// Where this session's storage setup stands (E5).
160    setup: std::sync::Mutex<setup::SetupRun>,
161    /// Wakes a setup checking the provider when the app cancels it.
162    setup_cancelled: tokio::sync::Notify,
163    /// What the next setup does at one of its stops, for a test.
164    #[cfg(test)]
165    setup_hook: std::sync::Mutex<Option<setup::SetupHook>>,
166}
167
168impl StorageConnections {
169    pub(crate) fn new(
170        database: coven_database::Database,
171        files: coven_sync::Files,
172        identity: Arc<KeySession<MemberIdentity>>,
173        person: Arc<PersonEntries>,
174        credentials: Arc<DeviceCredentials>,
175        codes: RestoreCodes,
176        state: Arc<SequenceState>,
177        directory: StoreDir,
178        waits: Arc<coven_sync::RetryWaits>,
179        transfers: Arc<coven_storage::TransferBudget>,
180        keys: Arc<KeySession<StoreKeyring>>,
181        connector: Arc<dyn StorageConnector>,
182        oauth: Option<OAuthFlow>,
183        clients: Option<OAuthExchange>,
184        authentication: Option<Arc<SessionSignIn>>,
185        storage: Option<Arc<StorageConnection>>,
186        configured: bool,
187        keychain: Arc<coven_crypto::custody::StoreKeychain>,
188        setup_finishing: bool,
189        device: DeviceId,
190        ids: IdSourceRef,
191        clock: ClockRef,
192    ) -> Self {
193        let expiry = coven_sync::InviteExpiry::start(
194            database.clone(),
195            clock.clone(),
196            waits.clone(),
197            state.clone(),
198        );
199        Self {
200            codes,
201            expiry,
202            database,
203            files,
204            identity,
205            person,
206            credentials,
207            state,
208            directory,
209            waits,
210            transfers,
211            keys: std::sync::Mutex::new(Some(keys)),
212            connector,
213            oauth,
214            clients,
215            authentication: tokio::sync::Mutex::new(authentication),
216            live: tokio::sync::Mutex::new(match storage {
217                Some(storage) => ConnectionState::Loaded(storage),
218                None => ConnectionState::Released,
219            }),
220            control: std::sync::Mutex::new(None),
221            configured: AtomicBool::new(configured),
222            epoch: AtomicU64::new(0),
223            stopped: tokio::sync::Notify::new(),
224            attempts: std::sync::Mutex::new(None),
225            device,
226            ids,
227            clock,
228            calls: tokio::sync::Mutex::new(()),
229            #[cfg(test)]
230            call_waits: tokio::sync::watch::channel(0).0,
231            #[cfg(test)]
232            close_failure: std::sync::Mutex::new(None),
233            #[cfg(test)]
234            deletion_gate: std::sync::Mutex::new(None),
235            keychain,
236            setup: std::sync::Mutex::new(match setup_finishing {
237                true => setup::SetupRun::Finishing,
238                false => setup::SetupRun::Idle,
239            }),
240            setup_cancelled: tokio::sync::Notify::new(),
241            #[cfg(test)]
242            setup_hook: std::sync::Mutex::new(None),
243        }
244    }
245
246    /// Initialize a loaded connection when a caller first needs its sequence owner.
247    async fn ready(&self, live: &mut ConnectionState) -> Result<(), SyncError> {
248        if let ConnectionState::Loaded(storage) = &*live {
249            let storage = storage.clone();
250            self.attach(live, storage).await?;
251        }
252        Ok(())
253    }
254
255    fn store(&self) -> StoreId {
256        self.directory.id()
257    }
258
259    /// Write how sync failed outside a pass's end, starting or stopping, as
260    /// the last pass's failure, in the step that fails (E5). A failed write
261    /// has failed the store's handle with its cause, which the caller
262    /// returns.
263    pub(crate) async fn show(&self, failure: LastPass) -> Result<(), SyncError> {
264        failure.record(&self.database).await.map_err(SyncError::from)
265    }
266
267    /// Write whether a storage location is recorded here and custody holds
268    /// its credentials, and keep whether starting connects (E5). `finished`,
269    /// for setup, is when it read the store to its end. A failed write has
270    /// failed the store's handle with its cause, which the caller returns.
271    pub(crate) async fn record_location(
272        &self,
273        location: coven_database::RecordedLocation<CloudProvider>,
274        finished: Option<std::time::SystemTime>,
275    ) -> Result<(), SyncError> {
276        self.configured.store(location.is_connected(), Ordering::SeqCst);
277        coven_sync::record_location(&self.database, location, finished)
278            .await
279            .map_err(SyncError::from)
280    }
281
282    /// Record the location as custody holds it now, as `record_location`
283    /// does: after credentials change, or a connection is released.
284    pub(crate) async fn record_custody(&self) -> Result<(), SyncError> {
285        let location = self.credentials.location()?;
286        self.record_location(location, None).await
287    }
288
289    pub(crate) async fn storage_config(&self) -> Result<Option<StorageConfig>, SyncError> {
290        let _call = self.lock_open().await?;
291        self.codes.location().await
292    }
293
294    /// Run `use_keys` on this session's store keys, holding the keys lock
295    /// throughout so closing cannot take them away mid-use. `StoreClosed`
296    /// once they are gone.
297    fn with_keys<T>(
298        &self,
299        use_keys: impl FnOnce(&Arc<KeySession<StoreKeyring>>) -> Result<T, KeyError>,
300    ) -> Result<T, KeyError> {
301        let keys = lock_unpoisoned(&self.keys);
302        use_keys(keys.as_ref().ok_or(KeyError::StoreClosed)?)
303    }
304
305    /// The call lock, which one call holds at a time; a test hears each
306    /// call that finds it held and waits.
307    pub(super) async fn lock_calls(&self) -> tokio::sync::MutexGuard<'_, ()> {
308        #[cfg(test)]
309        match self.calls.try_lock() {
310            Ok(call) => return call,
311            Err(_) => self.call_waits.send_modify(|waits| *waits += 1),
312        }
313        self.calls.lock().await
314    }
315
316    async fn lock_open(&self) -> Result<tokio::sync::MutexGuard<'_, ()>, KeyError> {
317        let call = self.lock_calls().await;
318        self.with_keys(|_| Ok(()))?;
319        Ok(call)
320    }
321
322    // The task owns both the call and its lock until it finishes, even if the
323    // awaiting app future is dropped. Sign-in deliberately does not spawn.
324    async fn call<T, E, F>(
325        self: &Arc<Self>,
326        run: impl FnOnce(Arc<Self>) -> F + Send + 'static,
327    ) -> Result<T, E>
328    where
329        T: Send + 'static,
330        E: From<KeyError>
331            + From<crate::task::TaskFailed>
332            + crate::task::CarriesBug
333            + Send
334            + 'static,
335        F: Future<Output = Result<T, E>> + Send,
336    {
337        let owner = self.clone();
338        let call = async move {
339            let _call = owner.lock_open().await?;
340            run(owner.clone()).await
341        };
342        crate::task::spawn_completing("storage call", self.database.bugs().clone(), call).await?
343    }
344
345    /// Compose a sequence owner and its loop on `storage` as this store's
346    /// connection.
347    async fn attach(
348        &self,
349        live: &mut ConnectionState,
350        storage: Arc<StorageConnection>,
351    ) -> Result<(), SyncError> {
352        let sync = self.sequence(storage.clone(), live).await?;
353        self.files.set_storage(Some(storage), std::future::ready(Ok(()))).await?;
354        self.activate(live, sync, false).await
355    }
356
357    /// Compose a candidate without replacing the committed connection.
358    async fn sequence(
359        &self,
360        storage: Arc<StorageConnection>,
361        previous: &ConnectionState,
362    ) -> Result<Arc<SequenceSync>, SyncError> {
363        if let ConnectionState::Active(previous) = previous {
364            return Ok(Arc::new(previous.sync.reconnect(storage)));
365        }
366        let keys = self.with_keys(|keys| Ok(keys.clone()))?;
367        Ok(Arc::new(
368            SequenceSync::open(
369                self.database.clone(),
370                storage,
371                self.directory.clone(),
372                self.device,
373                keys,
374                self.identity.clone(),
375                self.person.clone(),
376                self.state.clone(),
377                self.clock.clone(),
378                self.ids.clone(),
379                self.waits.clone(),
380                self.files.clone(),
381            )
382            .await?,
383        ))
384    }
385
386    /// Install the sequence owner only after connection custody commits.
387    async fn activate(
388        &self,
389        live: &mut ConnectionState,
390        sync: Arc<SequenceSync>,
391        start: bool,
392    ) -> Result<(), SyncError> {
393        let sync_loop = SyncLoop::new(sync.clone(), self.codes.clone());
394        let started = if start { sync_loop.start().await } else { Ok(()) };
395        *lock_unpoisoned(&self.control) = Some(sync_loop.clone());
396        *live = ConnectionState::Active(Live { sync, sync_loop });
397        started
398    }
399
400    /// Run `work` with the active connection's passes held: uploads run
401    /// beside them, as during a connection change (§17.4). Work that fails
402    /// or panics lets the passes run again, keeping a failure to resume them
403    /// as cleanup; work that succeeds leaves them held for the caller to
404    /// end. Unwound, a panic would leave them held for as long as the store
405    /// stays open (E15).
406    async fn with_passes_held<T>(
407        live: &ConnectionState,
408        work: impl std::future::Future<Output = Result<T, SyncError>>,
409    ) -> Result<T, SyncError> {
410        let ConnectionState::Active(previous) = live else {
411            return work.await;
412        };
413        previous.sync_loop.pause().await?;
414        let done = crate::task::caught("work with the passes held", work)
415            .await
416            .unwrap_or_else(|bug| Err(SyncError::Bug(bug)));
417        done.map_err(|error| match previous.sync_loop.resume() {
418            Ok(()) => error,
419            Err(cleanup) => SyncError::cleanup(error, cleanup),
420        })
421    }
422
423    /// Leave the connection released. An active one closes after its pass
424    /// and transfers, and after the credential change running on it, even
425    /// one whose caller stopped waiting.
426    async fn close_live(&self, live: &mut ConnectionState) -> Result<(), SyncError> {
427        match std::mem::replace(live, ConnectionState::Released) {
428            ConnectionState::Active(previous) => {
429                let result = previous.sync_loop.close().await;
430                self.codes.settled().await;
431                result
432            }
433            ConnectionState::Released | ConnectionState::Loaded(_) => Ok(()),
434        }
435    }
436
437    /// End the sync loop: retries waiting for a pass are answered
438    /// `NotSyncing`, as no connection plans afterwards, and the connection's
439    /// owners close after its active pass. Answers whether the retries heard
440    /// so, then whether the owners closed.
441    async fn end_loop(
442        &self,
443        live: &mut ConnectionState,
444    ) -> (Result<(), DbError>, Result<(), SyncError>) {
445        let stopped = self.waits.sync_stopped().await;
446        lock_unpoisoned(&self.control).take();
447        (stopped, self.close_live(live).await)
448    }
449
450    /// Release the current connection after its active pass and transfers,
451    /// with the files' provider client.
452    async fn release(&self, live: &mut ConnectionState) -> Result<(), SyncError> {
453        let (stopped, closed) = self.end_loop(live).await;
454        let released = self.files.set_storage(None, std::future::ready(Ok(()))).await;
455        closed.and(released).and(stopped.map_err(SyncError::from))
456    }
457
458    /// Connect with custody's credentials unless this session already has a
459    /// connection, without starting the loop.
460    pub(crate) async fn connected(&self) -> Result<(), SyncError> {
461        let mut live = self.live.lock().await;
462        if let ConnectionState::Active(_) = &*live {
463            return Ok(());
464        }
465        self.connect(&mut live).await
466    }
467
468    /// Reach the recorded location with custody's credentials, removing
469    /// what recorded detection runs left there (§20.7), or finish the setup
470    /// a crash cut past its point of no return (§11.6).
471    async fn connect(&self, live: &mut ConnectionState) -> Result<(), SyncError> {
472        if self.setup_finishing() {
473            if let Some(journal) = coven_sync::SetupJournal::running(&self.database).await? {
474                return self.resume_setup(live, journal).await;
475            }
476        }
477        if let ConnectionState::Loaded(storage) = live {
478            let storage = storage.clone();
479            return self.attach(live, storage).await;
480        }
481        let data = self.codes.connection().await?.ok_or(SyncError::NoStorage)?;
482        let storage = self
483            .connector
484            .connect(
485                data.location.clone(),
486                self.codes.provider_auth(data.credentials),
487                self.device,
488                self.state.clone(),
489                self.waits.clone(),
490                self.transfers.clone(),
491            )
492            .await?;
493        storage.check_location_claim(self.store()).await?;
494        // The folder's label as its provider names it now, so a rename since
495        // the last connection shows (§20.2).
496        self.codes.record_labels(&storage.labelled_config().await?).await?;
497        s3::end_detection_runs(&self.database, self.store(), &data.location, &storage).await?;
498        self.attach(live, storage).await
499    }
500
501    pub(crate) fn key_state(&self) -> Result<StoreKeyState, KeyError> {
502        let held = self.with_keys(|keys| keys.read())?;
503        Ok(match held {
504            Some(_) => StoreKeyState::Available,
505            None => StoreKeyState::Locked,
506        })
507    }
508
509    pub(crate) async fn authenticate(&self, access: OAuthAccess) -> Result<(), StorageSetupError> {
510        let _call = self.lock_open().await?;
511        let flow = self.oauth.as_ref().ok_or(OAuthError::Unavailable(access.provider()))?;
512        let tokens = flow.authenticate(access).await?;
513        *self.authentication.lock().await = Some(SessionSignIn::new(tokens, self.clients.clone()));
514        Ok(())
515    }
516
517    pub(crate) async fn setup_oauth(
518        self: &Arc<Self>,
519        storage: StorageConfig,
520        device_name: &str,
521        progress: SetupProgress,
522    ) -> Result<StorageSetup, StorageSetupError> {
523        let device_name = device_name.to_owned();
524        let call = self.start_setup()?;
525        self.call(move |owner| async move {
526            let mut progress = ProgressReport::new(progress, owner.clock.clone());
527            let waits = SetupWaits::default();
528            let provider = storage.provider();
529            if !provider.signs_in_with_oauth() {
530                return Err(OAuthError::Unavailable(provider).into());
531            }
532            owner.check_location(&storage).await?;
533            let mut held = owner.authentication.lock().await;
534            let (auth, sign_in) = match held.as_ref() {
535                Some(sign_in) => {
536                    let auth = crate::authentication::held_auth(sign_in, provider)?;
537                    (auth, Some(sign_in.clone()))
538                }
539                None => {
540                    let data = owner
541                        .codes
542                        .connection()
543                        .await
544                        .map_err(setup_error)?
545                        .ok_or(OAuthError::Reauthorize(provider))?;
546                    if data.location.provider() != provider {
547                        return Err(OAuthError::Reauthorize(provider).into());
548                    }
549                    (owner.codes.provider_auth(data.credentials), None)
550                }
551            };
552            let journal = owner.begin_setup(call, &device_name, &auth.credentials()).await?;
553            let setup = owner.run_setup(journal, async |journal| {
554                owner
555                    .check_new_location(journal, &storage, auth.clone(), &waits, &mut progress)
556                    .await?;
557                let (connector, cooldown) = (owner.connector.as_ref(), &owner.waits);
558                let (location, signed_in) = (&storage, &auth);
559                let reading = waits.resend(owner.clock.as_ref(), move || {
560                    crate::authentication::label_location(
561                        connector,
562                        location.clone(),
563                        signed_in.clone(),
564                        cooldown.clone(),
565                    )
566                });
567                let storage =
568                    setup_step::single(&mut progress, StorageSetupStep::ReadingAccount, reading)
569                        .await?;
570                owner.setup_connection(journal, storage, auth, sign_in, waits, &mut progress).await
571            });
572            let result = setup.await?;
573            held.take();
574            Ok(result)
575        })
576        .await
577    }
578
579    pub(crate) async fn setup_cloudkit(
580        self: &Arc<Self>,
581        container: &str,
582        device_name: &str,
583        progress: SetupProgress,
584    ) -> Result<StorageSetup, StorageSetupError> {
585        let container = container.to_owned();
586        let device_name = device_name.to_owned();
587        let call = self.start_setup()?;
588        self.call(move |owner| async move {
589            let mut progress = ProgressReport::new(progress, owner.clock.clone());
590            let waits = SetupWaits::default();
591            let recorded = owner.codes.location().await.map_err(setup_error)?;
592            if recorded.as_ref().is_some_and(|recorded| {
593                !matches!(recorded, StorageConfig::CloudKit { container: kept, .. } if *kept == container)
594            }) {
595                return Err(StorageSetupError::LocationChangeUnsupported);
596            }
597            let journal =
598                owner.begin_setup(call, &device_name, &StorageCredentials::CloudKit).await?;
599            owner
600                .run_setup(journal, async |journal| {
601                    let storage = match recorded {
602                        Some(storage) => storage,
603                        None => {
604                            // Preparing reads the account and saves the zone,
605                            // as it does for a zone that already exists: a
606                            // refusal can be sent again.
607                            let (connector, container, store) =
608                                (owner.connector.as_ref(), container.as_str(), owner.store());
609                            let preparing = waits.resend(owner.clock.as_ref(), move || {
610                                connector.prepare_cloudkit(container, store)
611                            });
612                            setup_step::single(
613                                &mut progress,
614                                StorageSetupStep::PreparingZone,
615                                preparing,
616                            )
617                            .await?
618                        }
619                    };
620                    owner
621                        .setup_connection(
622                            journal,
623                            storage,
624                            ProviderAuth::CloudKit,
625                            None,
626                            waits,
627                            &mut progress,
628                        )
629                        .await
630                })
631                .await
632        })
633        .await
634    }
635
636    /// Before a store's first setup at a location every member reaches
637    /// whole, check that it holds nothing but this store (§20.3). Reads only,
638    /// so it runs before setup's first write; a location that holds several
639    /// stores, and a reconnect, check nothing here.
640    async fn check_new_location(
641        &self,
642        journal: &coven_sync::SetupJournal,
643        config: &StorageConfig,
644        auth: ProviderAuth,
645        waits: &SetupWaits,
646        progress: &mut ProgressReport<StorageSetupStep>,
647    ) -> Result<(), StorageSetupError> {
648        if !journal.creating() || !config.holds_one_store() {
649            return Ok(());
650        }
651        let checking = self.connector.check_new_store(config.clone(), auth, self.store(), waits);
652        setup_step::single(progress, StorageSetupStep::CheckingLocation, checking).await?;
653        Ok(())
654    }
655
656    async fn check_location(&self, config: &StorageConfig) -> Result<(), StorageSetupError> {
657        if self
658            .codes
659            .location()
660            .await
661            .map_err(setup_error)?
662            .is_some_and(|recorded| !recorded.same_location(config))
663        {
664            return Err(StorageSetupError::LocationChangeUnsupported);
665        }
666        Ok(())
667    }
668
669    fn persist_keys(&self, ring: &StoreKeyring) -> Result<(), KeyError> {
670        self.with_keys(|keys| keys.persist(ring))
671    }
672
673    fn forget_keys(&self) -> Result<(), KeyError> {
674        self.with_keys(|keys| keys.forget())
675    }
676
677    pub(crate) async fn unlock(self: &Arc<Self>) -> Result<ConnectedStorage, StoreKeyUnlockError> {
678        self.call(|owner| async move {
679            let data = owner
680                .codes
681                .connection()
682                .await
683                .map_err(unlock_error)?
684                .ok_or(StoreKeyUnlockError::NoStorage)?;
685            let config = data.location.clone();
686            let mut live = owner.live.lock().await;
687            owner.release(&mut live).await.map_err(unlock_error)?;
688            owner.connect(&mut live).await.map_err(unlock_error)?;
689            let unlocked = match &*live {
690                ConnectionState::Active(candidate) => candidate.sync.unlock().await,
691                _ => Err(SyncError::NoStorage),
692            };
693            if let Err(error) = unlocked {
694                let error = match owner.release(&mut live).await {
695                    Ok(()) => error,
696                    Err(cleanup) => SyncError::cleanup(error, cleanup),
697                };
698                let error = match owner.publish_released(Ok(())).await {
699                    Ok(()) => error,
700                    Err(cleanup) => SyncError::cleanup(error, cleanup),
701                };
702                return Err(unlock_error(error));
703            }
704            // Connected without syncing.
705            owner.publish_released(Ok(())).await.map_err(unlock_error)?;
706            Ok(ConnectedStorage { storage: config, key_state: StoreKeyState::Available })
707        })
708        .await
709    }
710
711    /// Load a store a join published once it found its admission, before
712    /// anything of it was loaded: connect, through the connection the join
713    /// made, and read forward with the keys its copies opened, sending each
714    /// of loading's steps to `progress` (§13.2, §13.4). The loop doesn't
715    /// start; a failure leaves the store as a failed pass would, for the
716    /// loop's passes to go on from.
717    pub(crate) async fn load_joined(
718        self: &Arc<Self>,
719        progress: tokio::sync::watch::Sender<Option<Progress<coven_database::LoadStep>>>,
720    ) -> Result<(), SyncError> {
721        self.call(|owner| async move {
722            let mut live = owner.live.lock().await;
723            if !matches!(&*live, ConnectionState::Active(_)) {
724                owner.connect(&mut live).await?;
725            }
726            let ConnectionState::Active(candidate) = &*live else {
727                return Err(SyncError::NoStorage);
728            };
729            candidate.sync.load_joined(progress).await
730        })
731        .await
732    }
733
734    pub(crate) async fn forget_store_keys(self: &Arc<Self>) -> Result<(), KeyError> {
735        self.call(|owner| async move {
736            let mut live = owner.live.lock().await;
737            let result = async {
738                Self::with_passes_held(
739                    &live,
740                    owner
741                        .files
742                        .set_storage(None, async { owner.forget_keys().map_err(SyncError::from) }),
743                )
744                .await?;
745                let (stopped, released) = owner.end_loop(&mut live).await;
746                // Forgetting succeeded. A worker release failure is recorded
747                // as the last pass's failure; no later custody read can turn
748                // removal into an error. Stopping writes nothing else the
749                // sync state reads. A failed write has failed the handle.
750                if let Err(error) = released {
751                    owner.show(LastPass::failed(error)).await?;
752                }
753                Ok(stopped?)
754            }
755            .await;
756            result.map_err(|error| match error {
757                SyncError::SecureStorage(error) => error,
758                SyncError::Database(DbError::StoreClosed) => KeyError::StoreClosed,
759                // A bug is the call's own `Bug`, never custody that is
760                // unavailable (E15).
761                error => match crate::task::CarriesBug::bug(&error) {
762                    Some(bug) => KeyError::Bug(bug.clone()),
763                    None => KeyError::Unavailable(Box::new(error)),
764                },
765            })
766        })
767        .await
768    }
769}
770
771#[path = "storage_control.rs"]
772mod control;
773
774#[path = "storage_folders.rs"]
775mod folders;
776#[cfg(test)]
777pub(crate) use folders::with_folder_label;
778
779#[path = "storage_s3.rs"]
780mod s3;
781
782#[path = "storage_setup_step.rs"]
783mod setup_step;
784pub use setup_step::StorageSetupStep;
785
786#[path = "storage_setup.rs"]
787mod setup;
788pub(crate) use setup::settle_opened;
789pub use setup::SetupCancellation;
790#[cfg(test)]
791pub(crate) use setup::SetupHook;
792
793#[path = "storage_admin.rs"]
794mod admin;
795pub use admin::GuidedAccessKey;
796
797#[path = "storage_operations.rs"]
798mod operations;
799
800/// How this member reaches storage over `storage` signed in with
801/// `credentials`: the S3 key, or the signed-in account (§20.2).
802pub(crate) async fn member_access(
803    credentials: &StorageCredentials,
804    storage: &StorageConnection,
805) -> Result<coven_format::MemberAccess, coven_storage::StorageError> {
806    Ok(match credentials {
807        StorageCredentials::S3(keys) => {
808            coven_format::MemberAccess::S3AccessKey { access_key_id: keys.access_key_id.clone() }
809        }
810        _ => coven_format::MemberAccess::ProviderAccount(storage.account().await?),
811    })
812}
813
814/// The entry recording `access` as `member`'s own, when history records
815/// another: setup and restore keep the recorded access the one the member
816/// uses (§14.2).
817pub(crate) fn access_change(
818    state: &coven_database::StoreState,
819    member: MemberId,
820    access: coven_format::MemberAccess,
821) -> Result<Option<coven_format::admin::StoreChange>, SyncError> {
822    let recorded = state.active_member(&member).ok_or(SyncError::Removed)?;
823    Ok((!recorded.access.same_credential(&access))
824        .then_some(coven_format::admin::StoreChange::SetAccess { member, access }))
825}
826
827#[cfg(test)]
828#[path = "storage_tests.rs"]
829mod tests;
coven/storage.rs

coven/
storage.rs
