coven/
cloud_outbox_live_query.rs

1use std::sync::Arc;
2
3use coven_database::{CloudOutboxSnapshot, CommittedChanges, StoreDatabase};
4
5const OUTBOX_TABLES: &[&str] = &["cloud_outbox", "blob_make_remote_intents"];
6
7/// A committed view of coven's durable cloud work.
8///
9/// The initial snapshot is returned immediately. Later calls wait for a
10/// transaction that changes the upload queue or a make-remote intent, then read
11/// both tables in one database operation. Transfer byte callbacks do not write
12/// here; hosts combine their in-memory progress with this durable lower bound.
13pub struct CloudOutboxLiveQuery {
14    database: StoreDatabase,
15    changes: tokio::sync::broadcast::Receiver<Arc<CommittedChanges>>,
16    initial: bool,
17}
18
19impl CloudOutboxLiveQuery {
20    pub(crate) fn new(database: StoreDatabase) -> Self {
21        let changes = database.subscribe_committed_changes();
22        Self {
23            database,
24            changes,
25            initial: true,
26        }
27    }
28
29    /// Return the initial snapshot, or wait for the next relevant committed
30    /// change and return the resulting snapshot.
31    pub async fn next(&mut self) -> Result<CloudOutboxSnapshot, crate::DbError> {
32        if self.initial {
33            self.initial = false;
34        } else {
35            loop {
36                match self.changes.recv().await {
37                    Ok(changes) if changes.affects_any_table(OUTBOX_TABLES) => break,
38                    Ok(_) => {}
39                    Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => break,
40                    Err(tokio::sync::broadcast::error::RecvError::Closed) => {
41                        panic!("the cloud outbox live query retains its database")
42                    }
43                }
44            }
45        }
46        self.database.cloud_outbox_snapshot().await
47    }
48}
coven/cloud_outbox_live_query.rs

coven/
cloud_outbox_live_query.rs
