coven_database/

database_retry.rs

//! Persisted retry waits through the owning database (§6.7).

use super::Database;
use crate::{
    retry::{self, RetrySubject, RetryWait},
    DbError,
};

impl Database {
    /// Persist a failed pass's connection condition and retry deadline together.
    /// A failure already owned by a slot or operation supplies no connection record.
    pub async fn record_sync_retry<P: crate::PendingValue>(
        &self,
        connection: P::Subject,
        record: Option<P>,
        wait: Option<RetryWait>,
    ) -> Result<(), DbError>
    where
        P::Subject: Send + 'static,
    {
        self.call(move |inner| {
            inner.access.with_writer(|writer| {
                writer.state_transaction(|db| {
                    if let Some(record) = record {
                        crate::pending::save(db, &record)?;
                    } else {
                        crate::pending::clear(db, &connection)?;
                    }
                    retry::set(db, RetrySubject::Sync, wait.as_ref())
                })
            })
        })
        .await
    }

    /// A successful pass settles its connection condition and wait atomically.
    pub async fn clear_sync_retry<S: serde::Serialize + Send + 'static>(
        &self,
        connection: S,
    ) -> Result<(), DbError> {
        self.call(move |inner| {
            let changed = inner.access.readers.acquire_reader().with_reader(|db| {
                Ok::<_, DbError>(
                    retry::find(db, RetrySubject::Sync)?.is_some()
                        || crate::pending::recorded(db, &connection)?,
                )
            })?;
            if !changed {
                return Ok(());
            }
            inner.access.with_writer(|writer| {
                writer.state_transaction(|db| {
                    retry::clear(db, RetrySubject::Sync)?;
                    crate::pending::clear(db, &connection)
                })
            })
        })
        .await
    }

    /// Persist an operation's failure and its automatic retry deadline together.
    pub async fn record_operation_retry<P: crate::PendingValue>(
        &self,
        record: P,
        wait: Option<RetryWait>,
    ) -> Result<(), DbError> {
        self.call(move |inner| {
            let id = record
                .operation()
                .ok_or(DbError::Invariant { what: "an operation retry names its journal row" })?;
            inner.access.with_writer(|writer| {
                writer.state_transaction(|db| {
                    retry::save_failure(db, &record, RetrySubject::Operation(id), wait.as_ref())
                })
            })
        })
        .await
    }

    /// Persist an operation's failure and its automatic retry deadline
    /// together, while the operation is still at `step`, the step the failed
    /// attempt read, and return whether it did. A writer outside the attempt,
    /// such as the app's write deciding an invite or the invite's expiry, may
    /// have moved the operation on since, or ended it: the failure then
    /// belongs to a step the operation has left, and recording it would hold
    /// the operation's next step back behind a wait it never met.
    pub async fn record_operation_retry_at<P: crate::PendingValue>(
        &self,
        record: P,
        wait: Option<RetryWait>,
        step: u32,
    ) -> Result<bool, DbError> {
        self.call(move |inner| {
            let id = record
                .operation()
                .ok_or(DbError::Invariant { what: "an operation retry names its journal row" })?;
            inner.access.with_writer(|writer| {
                writer.state_transaction(|db| {
                    if crate::operation::find(db, id)?.is_none_or(|row| row.last_step != step) {
                        return Ok(false);
                    }
                    retry::save_failure(db, &record, RetrySubject::Operation(id), wait.as_ref())
                        .map(|()| true)
                })
            })
        })
        .await
    }

    /// Every persisted wait, for arming on reopen (§6.7).
    pub async fn retry_waits(&self) -> Result<Vec<(RetrySubject, RetryWait)>, DbError> {
        self.call(move |inner| inner.access.readers.acquire_reader().with_reader(retry::read_all))
            .await
    }

    /// The subject's persisted wait, if one is armed.
    pub async fn retry_wait(&self, subject: RetrySubject) -> Result<Option<RetryWait>, DbError> {
        self.call(move |inner| {
            inner.access.readers.acquire_reader().with_reader(|db| retry::find(db, subject))
        })
        .await
    }

    /// The same wait, live (E4), for a test waiting for a failure to arm it.
    #[cfg(any(test, feature = "test-utils"))]
    pub fn subscribe_retry_wait(
        &self,
        subject: RetrySubject,
    ) -> crate::LiveQuery<Option<RetryWait>, crate::DbError> {
        self.subscribe(move |sql| sql.retry_wait(subject))
    }

    /// Persist `T` and `W` before the subject waits. A failure here fails the
    /// initiating work; it is never permission to retry immediately.
    pub async fn record_retry(
        &self,
        subject: RetrySubject,
        wait: RetryWait,
    ) -> Result<(), DbError> {
        self.call(move |inner| {
            inner
                .access
                .with_writer(|writer| writer.transaction(|db| retry::save(db, subject, &wait)))
        })
        .await
    }

    /// End, in one transaction, the persisted waits `select` picks after
    /// reading every wait, every pending record, the upload queue and the pin
    /// requests as one committed state. Each ended wait keeps its failure count; its deadline
    /// becomes `now` and its full wait zero, so reopening arms none (§6.7).
    /// Returns the ended waits.
    pub async fn end_retry_waits<P: crate::PendingValue>(
        &self,
        now: std::time::SystemTime,
        select: impl FnOnce(
                &[(RetrySubject, RetryWait)],
                &[P],
                &[crate::FileUpload],
                &[crate::PinRequest],
            ) -> Vec<RetrySubject>
            + Send
            + 'static,
    ) -> Result<Vec<(RetrySubject, RetryWait)>, DbError> {
        self.call(move |inner| {
            #[cfg(any(test, feature = "test-utils"))]
            let injected = inner
                .retry_end_failures
                .fetch_update(
                    std::sync::atomic::Ordering::SeqCst,
                    std::sync::atomic::Ordering::SeqCst,
                    |left| left.checked_sub(1),
                )
                .is_ok();
            inner.access.with_writer(|writer| {
                writer.transaction(|db| {
                    #[cfg(any(test, feature = "test-utils"))]
                    if injected {
                        return Err(DbError::Sqlite(rusqlite::Error::SqliteFailure(
                            rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_IOERR),
                            Some("injected retry end failure".into()),
                        )));
                    }
                    let waits = retry::read_all(db)?;
                    let pending = crate::pending::read::<P>(db)?;
                    let uploads = crate::file_queue::read(db)?;
                    let pins = crate::file_pins::read(db)?;
                    let mut ended = Vec::new();
                    for subject in select(&waits, &pending, &uploads, &pins) {
                        let Some((_, wait)) = waits.iter().find(|(held, _)| *held == subject)
                        else {
                            return Err(DbError::Invariant {
                                what: "an ended retry wait is persisted",
                            });
                        };
                        retry::end(db, subject, now)?;
                        ended.push((
                            subject,
                            RetryWait {
                                failures: wait.failures,
                                not_before: now,
                                wait: std::time::Duration::ZERO,
                            },
                        ));
                    }
                    Ok(ended)
                })
            })
        })
        .await
    }

    /// Fail the next `count` `end_retry_waits` transactions with a disk error.
    #[cfg(any(test, feature = "test-utils"))]
    pub async fn fail_retry_ends(&self, count: usize) {
        self.call(move |inner| {
            inner.retry_end_failures.store(count, std::sync::atomic::Ordering::SeqCst);
            Ok::<_, DbError>(())
        })
        .await
        .expect("an open store arms its retry-end failures");
    }

    /// Forget a wait once its work succeeds or no longer retries automatically.
    pub async fn clear_retry(&self, subject: RetrySubject) -> Result<(), DbError> {
        self.call(move |inner| {
            inner.access.with_writer(|writer| writer.transaction(|db| retry::clear(db, subject)))
        })
        .await
    }
}