coven_database/
retry.rs

1//! Durable retry deadlines (§6.7).
2//!
3//! Before an automatic retry waits, its owner persists the not-before
4//! wall-clock deadline `T` and the full wait `W` with the work: one row per
5//! operation, queued upload, pin coven keeps, the joiner's approval poll, or
6//! the provider cooldown. A running session enforces the wait monotonically;
7//! reopening arms `max(0, min(T - now, W))` from these rows. An operation's,
8//! upload's or pin's row is deleted with that work, an operation's also when
9//! its pending condition clears (it advanced, or the app retried it). The
10//! app's own reads and pins keep none: only its next call tries again
11//! (§17.7).
12
13use crate::{sqlite::DatabaseConnection, DbError, OperationId};
14use std::time::{Duration, SystemTime};
15
16/// The work a persisted wait delays.
17#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
18pub enum RetrySubject {
19    /// A journal operation.
20    Operation(OperationId),
21    /// A queued file upload, by its queue id, which is never another
22    /// upload's.
23    Upload(i64),
24    /// A file a switch to the cloud asked this device to keep, which coven
25    /// downloads whole itself (§17.8), by its request's id: a later request
26    /// for the same file is another subject.
27    Pin(i64),
28    /// The joining device's polling for its request's outcome (§13.2).
29    Join,
30    /// A failed sequence pass.
31    Sync,
32    /// The provider/account cooldown a `Retry-After` requested; every
33    /// consumer of the store's connection waits for it.
34    Provider,
35}
36
37/// A persisted wait: `T` and `W`, and the consecutive unsuccessful attempts
38/// that chose `W`, so reopening does not reset the backoff exponent.
39#[derive(Clone, Copy, Debug, PartialEq, Eq)]
40pub struct RetryWait {
41    /// Consecutive unsuccessful attempts; zero for a provider cooldown.
42    pub failures: u32,
43    /// `T`: the wall-clock time before which no automatic attempt starts.
44    pub not_before: SystemTime,
45    /// `W`: the full wait chosen when `T` was persisted.
46    pub wait: Duration,
47}
48impl RetryWait {
49    /// A wait as a retry column set stores it: failures, `T` and `W` in
50    /// milliseconds.
51    pub(crate) fn stored(failures: i64, not_before: &[u8], wait: i64) -> Result<Self, DbError> {
52        Ok(Self {
53            failures: u32::try_from(failures).map_err(|_overflow| DbError::DamagedDatabase)?,
54            not_before: crate::user_file::decode_time(not_before)?,
55            wait: crate::stored_millis::stored_wait(wait)?,
56        })
57    }
58}
59
60/// The row's one subject column and its value.
61enum Key {
62    Scope(&'static str),
63    Operation(i64),
64    Upload(i64),
65    Pin(i64),
66}
67
68fn key(subject: RetrySubject) -> Key {
69    match subject {
70        RetrySubject::Operation(id) => Key::Operation(id.0),
71        RetrySubject::Upload(id) => Key::Upload(id),
72        RetrySubject::Pin(id) => Key::Pin(id),
73        RetrySubject::Join => Key::Scope("join"),
74        RetrySubject::Sync => Key::Scope("sync"),
75        RetrySubject::Provider => Key::Scope("provider"),
76    }
77}
78
79/// The statements on `_coven_retries` for one subject column: read its
80/// row, insert one, end its wait and delete it, each with the subject as
81/// `?1`.
82struct KeyStatements {
83    find: &'static str,
84    insert: &'static str,
85    end: &'static str,
86    delete: &'static str,
87}
88
89const SCOPE: KeyStatements = KeyStatements {
90    find: "SELECT scope,operation,upload,pin,failures,not_before,wait FROM _coven_retries WHERE scope=?1",
91    insert: "INSERT INTO _coven_retries(scope,failures,not_before,wait) VALUES(?1,?2,?3,?4)",
92    end: "UPDATE _coven_retries SET not_before=?2,wait=0 WHERE scope=?1",
93    delete: "DELETE FROM _coven_retries WHERE scope=?1",
94};
95
96const OPERATION: KeyStatements = KeyStatements {
97    find: "SELECT scope,operation,upload,pin,failures,not_before,wait FROM _coven_retries WHERE operation=?1",
98    insert: "INSERT INTO _coven_retries(operation,failures,not_before,wait) VALUES(?1,?2,?3,?4)",
99    end: "UPDATE _coven_retries SET not_before=?2,wait=0 WHERE operation=?1",
100    delete: "DELETE FROM _coven_retries WHERE operation=?1",
101};
102
103const UPLOAD: KeyStatements = KeyStatements {
104    find: "SELECT scope,operation,upload,pin,failures,not_before,wait FROM _coven_retries WHERE upload=?1",
105    insert: "INSERT INTO _coven_retries(upload,failures,not_before,wait) VALUES(?1,?2,?3,?4)",
106    end: "UPDATE _coven_retries SET not_before=?2,wait=0 WHERE upload=?1",
107    delete: "DELETE FROM _coven_retries WHERE upload=?1",
108};
109
110const PIN: KeyStatements = KeyStatements {
111    find: "SELECT scope,operation,upload,pin,failures,not_before,wait FROM _coven_retries WHERE pin=?1",
112    insert: "INSERT INTO _coven_retries(pin,failures,not_before,wait) VALUES(?1,?2,?3,?4)",
113    end: "UPDATE _coven_retries SET not_before=?2,wait=0 WHERE pin=?1",
114    delete: "DELETE FROM _coven_retries WHERE pin=?1",
115};
116
117impl Key {
118    fn value(&self) -> rusqlite::types::Value {
119        match self {
120            Self::Scope(scope) => rusqlite::types::Value::Text((*scope).into()),
121            Self::Operation(id) | Self::Upload(id) | Self::Pin(id) => {
122                rusqlite::types::Value::Integer(*id)
123            }
124        }
125    }
126
127    fn statements(&self) -> KeyStatements {
128        match self {
129            Self::Scope(_) => SCOPE,
130            Self::Operation(_) => OPERATION,
131            Self::Upload(_) => UPLOAD,
132            Self::Pin(_) => PIN,
133        }
134    }
135}
136
137fn read_row(row: &rusqlite::Row<'_>) -> rusqlite::Result<RawRow> {
138    Ok(RawRow {
139        scope: row.get(0)?,
140        operation: row.get(1)?,
141        upload: row.get(2)?,
142        pin: row.get(3)?,
143        failures: row.get(4)?,
144        not_before: row.get(5)?,
145        wait: row.get(6)?,
146    })
147}
148
149struct RawRow {
150    scope: Option<String>,
151    operation: Option<i64>,
152    upload: Option<i64>,
153    pin: Option<i64>,
154    failures: i64,
155    not_before: Vec<u8>,
156    wait: i64,
157}
158
159impl RawRow {
160    fn decode(self) -> Result<(RetrySubject, RetryWait), DbError> {
161        let subject = match (self.scope.as_deref(), self.operation, self.upload, self.pin) {
162            (None, Some(id), None, None) => RetrySubject::Operation(OperationId(id)),
163            (None, None, Some(id), None) => RetrySubject::Upload(id),
164            (None, None, None, Some(id)) => RetrySubject::Pin(id),
165            (Some("join"), None, None, None) => RetrySubject::Join,
166            (Some("sync"), None, None, None) => RetrySubject::Sync,
167            (Some("provider"), None, None, None) => RetrySubject::Provider,
168            _ => return Err(DbError::DamagedDatabase),
169        };
170        Ok((subject, RetryWait::stored(self.failures, &self.not_before, self.wait)?))
171    }
172}
173
174pub(crate) fn read_all(db: &DatabaseConnection) -> Result<Vec<(RetrySubject, RetryWait)>, DbError> {
175    db.query(
176        "SELECT scope,operation,upload,pin,failures,not_before,wait FROM _coven_retries",
177        [],
178        read_row,
179    )?
180    .into_iter()
181    .map(RawRow::decode)
182    .collect()
183}
184
185pub(crate) fn find(
186    db: &DatabaseConnection,
187    subject: RetrySubject,
188) -> Result<Option<RetryWait>, DbError> {
189    let key = key(subject);
190    db.query(key.statements().find, [key.value()], read_row)?
191        .pop()
192        .map(|row| row.decode().map(|(_, wait)| wait))
193        .transpose()
194}
195
196/// `W` is stored in whole milliseconds, rounded up so a persisted wait never
197/// shortens the one the session armed.
198fn millis(wait: &RetryWait) -> Result<i64, DbError> {
199    crate::stored_millis::millis(wait.wait, crate::stored_millis::Round::Up)
200        .ok_or(DbError::Invariant { what: "a retry wait fits SQLite's integer" })
201}
202
203/// Replace the subject's wait.
204pub(crate) fn save(
205    db: &DatabaseConnection,
206    subject: RetrySubject,
207    wait: &RetryWait,
208) -> Result<(), DbError> {
209    clear(db, subject)?;
210    let key = key(subject);
211    db.internal_execute(
212        key.statements().insert,
213        rusqlite::params![
214            key.value(),
215            i64::from(wait.failures),
216            crate::user_file::encode_time(wait.not_before),
217            millis(wait)?,
218        ],
219    )?;
220    Ok(())
221}
222
223/// Record a failed attempt's pending `record` with the subject's next wait,
224/// or with none, in the caller's transaction.
225pub(crate) fn save_failure<P: crate::PendingValue>(
226    db: &DatabaseConnection,
227    record: &P,
228    subject: RetrySubject,
229    wait: Option<&RetryWait>,
230) -> Result<(), DbError> {
231    crate::pending::save(db, record)?;
232    set(db, subject, wait)
233}
234
235/// Replace the subject's wait, or end it when there is none.
236pub(crate) fn set(
237    db: &DatabaseConnection,
238    subject: RetrySubject,
239    wait: Option<&RetryWait>,
240) -> Result<(), DbError> {
241    match wait {
242        Some(wait) => save(db, subject, wait),
243        None => clear(db, subject),
244    }
245}
246
247/// End the subject's wait at `now` in place: `T = now`, `W = 0`, keeping its
248/// failure count.
249pub(crate) fn end(
250    db: &DatabaseConnection,
251    subject: RetrySubject,
252    now: SystemTime,
253) -> Result<(), DbError> {
254    let key = key(subject);
255    db.internal_execute(
256        key.statements().end,
257        rusqlite::params![key.value(), crate::user_file::encode_time(now)],
258    )?;
259    Ok(())
260}
261
262pub(crate) fn clear(db: &DatabaseConnection, subject: RetrySubject) -> Result<(), DbError> {
263    let key = key(subject);
264    db.internal_execute(key.statements().delete, [key.value()])?;
265    Ok(())
266}
267
268#[cfg(test)]
269#[path = "retry_tests.rs"]
270pub(crate) mod tests;
coven_database/retry.rs

coven_database/
retry.rs
