coven_storage/
token_custody.rs

1//! Where a connection's OAuth sign-in is held, and its one refresh after the
2//! provider rejects a token (§20.2, E10). No device clock decides expiry.
3
4use crate::{OAuthTokens, S3Credentials, StorageCredentials, StorageError};
5use std::sync::Arc;
6
7/// The custody of a connection's sign-in: coven's session custody before
8/// setup or bootstrap publishes the store, device custody afterwards (§20.2).
9#[async_trait::async_trait]
10pub trait TokenCustody: Send + Sync {
11    /// Replace tokens the provider rejected with HTTP 401. When the held
12    /// sign-in no longer is `rejected` (another connection refreshed it, or a
13    /// new sign-in was committed), return the held one; otherwise refresh once
14    /// and commit the replacement here before returning it. A failed refresh
15    /// or commit returns its error and installs nothing.
16    async fn replace(&self, rejected: &OAuthTokens) -> Result<OAuthTokens, StorageError>;
17}
18
19/// What a connection authenticates with (§20.2).
20#[derive(Clone)]
21pub enum ProviderAuth {
22    /// This member's S3 key.
23    S3(S3Credentials),
24    /// A provider sign-in, refreshed in its custody when the provider rejects it.
25    OAuth {
26        /// The tokens the connection starts with.
27        tokens: OAuthTokens,
28        /// Where a rejected token's replacement is committed.
29        custody: Arc<dyn TokenCustody>,
30    },
31    /// The device's Apple account.
32    CloudKit,
33}
34
35impl ProviderAuth {
36    /// Committed credentials, with the custody that refreshes an OAuth sign-in.
37    pub fn new(credentials: StorageCredentials, custody: Arc<dyn TokenCustody>) -> Self {
38        match credentials {
39            StorageCredentials::S3(keys) => Self::S3(keys),
40            StorageCredentials::OAuth(tokens) => Self::OAuth { tokens, custody },
41            StorageCredentials::CloudKit => Self::CloudKit,
42        }
43    }
44
45    /// The credentials this connection starts with.
46    pub fn credentials(&self) -> StorageCredentials {
47        match self {
48            Self::S3(keys) => StorageCredentials::S3(keys.clone()),
49            Self::OAuth { tokens, .. } => StorageCredentials::OAuth(tokens.clone()),
50            Self::CloudKit => StorageCredentials::CloudKit,
51        }
52    }
53}
54
55/// The tokens one connection sends, and its single path to a replacement.
56/// Requests rejected concurrently with the same token share one replacement.
57pub(crate) struct HeldTokens {
58    current: tokio::sync::RwLock<OAuthTokens>,
59    custody: Arc<dyn TokenCustody>,
60    replacing: tokio::sync::Mutex<()>,
61    /// How many times an install or rejection has found a replacement in
62    /// flight and waited for it, for a test.
63    #[cfg(test)]
64    waits: tokio::sync::watch::Sender<usize>,
65}
66
67impl HeldTokens {
68    pub(crate) fn new(tokens: OAuthTokens, custody: Arc<dyn TokenCustody>) -> Self {
69        Self {
70            current: tokio::sync::RwLock::new(tokens),
71            custody,
72            replacing: tokio::sync::Mutex::new(()),
73            #[cfg(test)]
74            waits: tokio::sync::watch::channel(0).0,
75        }
76    }
77
78    /// One replacement at a time: an install or rejection waits for the
79    /// one in flight, and a test hears that it waits.
80    async fn one_replacement(&self) -> tokio::sync::MutexGuard<'_, ()> {
81        #[cfg(test)]
82        match self.replacing.try_lock() {
83            Ok(one) => return one,
84            Err(_) => self.waits.send_modify(|waits| *waits += 1),
85        }
86        self.replacing.lock().await
87    }
88
89    /// Wait until an install or rejection finds a replacement in flight and
90    /// waits for it, after this call.
91    #[cfg(test)]
92    pub(crate) fn waiting_from_now(&self) -> impl std::future::Future<Output = ()> {
93        let mut waits = self.waits.subscribe();
94        let now = *waits.borrow_and_update();
95        async move {
96            coven_foundation::clock::within(waits.wait_for(|waits| *waits > now))
97                .await
98                .expect("a call waits for the replacement")
99                .expect("the waits are kept");
100        }
101    }
102
103    /// Hold the tokens so requests wait for them.
104    #[cfg(test)]
105    pub(crate) async fn hold(&self) -> tokio::sync::RwLockWriteGuard<'_, OAuthTokens> {
106        self.current.write().await
107    }
108
109    /// The tokens the next request sends.
110    pub(crate) async fn current(&self) -> OAuthTokens {
111        self.current.read().await.clone()
112    }
113
114    /// Install tokens its owner has already committed to custody. A
115    /// replacement in flight finishes first: it was committed earlier, so
116    /// these later tokens must not be overwritten by it.
117    pub(crate) async fn install(&self, tokens: OAuthTokens) {
118        let _one = self.one_replacement().await;
119        *self.current.write().await = tokens;
120    }
121
122    /// The tokens to retry with after the provider rejected `used`.
123    pub(crate) async fn rejected(&self, used: &OAuthTokens) -> Result<OAuthTokens, StorageError> {
124        let _one = self.one_replacement().await;
125        let current = self.current().await;
126        if current.access_token.as_str() != used.access_token.as_str() {
127            return Ok(current);
128        }
129        let replacement = self.custody.replace(used).await?;
130        *self.current.write().await = replacement.clone();
131        Ok(replacement)
132    }
133}
134
135#[cfg(test)]
136#[path = "token_custody_tests.rs"]
137pub(crate) mod tests;
coven_storage/token_custody.rs

coven_storage/
token_custody.rs
