coven_database/
sync_facts.rs

1//! What the sync state is computed from (§19.1, api.md E5), read in one
2//! committed snapshot: `coven_sync_status`'s row, `coven_retry_effect`'s,
3//! the send queue, the upload queue with its pause (the uploads the state
4//! can show, each with its retry wait and record), the network and the
5//! setting to move files only on unmetered networks, the files being sent or
6//! downloaded, and the pending records. Sync computes the state; this reads
7//! what it needs.
8//!
9//! A live query over this read runs again whenever one of them changes: the
10//! files in flight are read through `coven_sending` and `coven_downloading`,
11//! the views their starts and ends change (`transfers_in_flight.rs`), and
12//! the network through `coven_network`, which its changes change
13//! (`network_facts.rs`).
14
15use crate::sqlite::DatabaseConnection;
16use crate::DbError;
17use std::collections::{BTreeMap, BTreeSet};
18use std::time::SystemTime;
19
20/// What the sync state is computed from, in one committed snapshot.
21#[derive(Clone, Debug, PartialEq)]
22pub struct SyncFacts<P> {
23    /// What the sync loop knows: `coven_sync_status`'s row.
24    pub row: crate::SyncStatusRow,
25    /// What the person's retry would do: whether sync runs, and the
26    /// cooldown holding every request.
27    pub retry: crate::RetryState,
28    /// Whether the send queue holds a write.
29    pub unsent: bool,
30    /// The uploads a file waits for that the state can show, in queue
31    /// order: the first one sending, the first queued one (neither sending
32    /// nor stored) with no record, and every one with a record. The state
33    /// shows the first candidate of the highest rank, so no other upload
34    /// changes it: later sending or waiting ones tie with the first, and a
35    /// stored one names none.
36    pub uploads: Vec<UploadFacts<P>>,
37    /// Whether the upload queue holds a file, whether `uploads` lists it or
38    /// not.
39    pub queued: bool,
40    /// Whether this process is downloading a file.
41    pub downloading: bool,
42    /// Every pending record, in the order the records arose.
43    pub records: Vec<P>,
44    /// Whether the app paused trailing uploads.
45    pub paused: bool,
46    /// Whether the app moves files only on unmetered networks.
47    pub unmetered_only: bool,
48    /// Whether the device's network is metered, as the store's watch last
49    /// said.
50    pub metered: bool,
51}
52
53/// One upload in the queue, as the sync state reads it.
54#[derive(Clone, Debug, PartialEq)]
55pub struct UploadFacts<P> {
56    /// This process is sending it now.
57    pub sending: bool,
58    /// Storage holds its bytes; it waits for the rest of its move.
59    pub stored: bool,
60    /// A bound reference's, which the pause doesn't hold (§17.4).
61    pub bound: bool,
62    /// Its last attempt failed, and it retries then (§6.7).
63    pub retry_at: Option<SystemTime>,
64    /// Its own pending record, if it has one.
65    pub record: Option<P>,
66}
67
68/// The uploads the sync state can show (`SyncFacts::uploads`), each read from
69/// where its kind is small: the uploads with a record from the pending
70/// records, the first one sending from the files in flight, and the first
71/// queued one with no record by walking the queue in id order to the first
72/// match. The outer query then reads only those rows.
73pub(crate) const UPLOADS: &str = "SELECT u.id,u.stored,u.holds IS NOT NULL,r.not_before,e.id
74     FROM _coven_file_uploads AS u
75     LEFT JOIN _coven_retries AS r ON r.upload=u.id
76     LEFT JOIN _coven_pending AS e
77         ON e.work='upload' AND e.device=u.device AND e.object=u.object
78     WHERE u.id IN (
79         SELECT f.id FROM _coven_pending AS p
80         JOIN _coven_file_uploads AS f ON f.device=p.device AND f.object=p.object
81         WHERE p.work='upload'
82         UNION
83         SELECT MIN(position) FROM coven_sending
84         UNION
85         SELECT * FROM (
86             SELECT f.id FROM _coven_file_uploads AS f
87             WHERE NOT f.stored
88               AND f.id NOT IN (SELECT position FROM coven_sending)
89               AND EXISTS(SELECT 1 FROM _coven_upload_files AS l WHERE l.upload=f.id)
90               AND NOT EXISTS(SELECT 1 FROM _coven_pending AS p
91                              WHERE p.work='upload' AND p.device=f.device
92                                AND p.object=f.object)
93             ORDER BY f.id LIMIT 1))
94       AND EXISTS(SELECT 1 FROM _coven_upload_files AS l WHERE l.upload=u.id)
95     ORDER BY u.id";
96
97/// Read the facts in the caller's snapshot.
98pub(crate) fn read<P: crate::PendingValue + Clone>(
99    db: &DatabaseConnection,
100) -> Result<SyncFacts<P>, DbError> {
101    let records = crate::pending::positioned::<P>(db)?;
102    let by_position: BTreeMap<i64, &P> =
103        records.iter().map(|(position, record)| (*position, record)).collect();
104    let sending: BTreeSet<i64> =
105        db.query("SELECT position FROM coven_sending", [], |row| row.get(0))?.into_iter().collect();
106    let uploads = db
107        .query(UPLOADS, [], |row| {
108            Ok((
109                row.get::<_, i64>(0)?,
110                row.get::<_, bool>(1)?,
111                row.get::<_, bool>(2)?,
112                row.get::<_, Option<Vec<u8>>>(3)?,
113                row.get::<_, Option<i64>>(4)?,
114            ))
115        })?
116        .into_iter()
117        .map(|(id, stored, bound, retry_at, position)| {
118            let record = position
119                .map(|position| {
120                    by_position.get(&position).map(|record| (*record).clone()).ok_or(
121                        DbError::Invariant {
122                            what: "an upload's record is among the pending records",
123                        },
124                    )
125                })
126                .transpose()?;
127            Ok(UploadFacts {
128                sending: sending.contains(&id),
129                stored,
130                bound,
131                retry_at: retry_at.as_deref().map(crate::user_file::decode_time).transpose()?,
132                record,
133            })
134        })
135        .collect::<Result<_, DbError>>()?;
136    let (metered, unmetered_only) =
137        db.query_row("SELECT metered,unmetered_only FROM coven_network", [], |row| {
138            Ok((row.get::<_, bool>(0)?, row.get::<_, bool>(1)?))
139        })?;
140    Ok(SyncFacts {
141        row: crate::internal_schema::sync_status::read(db)?,
142        retry: crate::internal_schema::retry_effect::read(db)?,
143        unsent: db.query_row("SELECT EXISTS(SELECT 1 FROM _coven_uploads)", [], |r| r.get(0))?,
144        uploads,
145        queued: db.query_row(
146            "SELECT EXISTS(SELECT 1 FROM _coven_file_uploads AS u
147                 WHERE EXISTS(SELECT 1 FROM _coven_upload_files AS l WHERE l.upload=u.id))",
148            [],
149            |r| r.get(0),
150        )?,
151        downloading: db
152            .query_row("SELECT EXISTS(SELECT 1 FROM coven_downloading)", [], |r| r.get(0))?,
153        records: records.into_iter().map(|(_, record)| record).collect(),
154        paused: crate::file_queue::paused(db)?,
155        unmetered_only,
156        metered,
157    })
158}
159
160#[cfg(test)]
161#[path = "sync_facts_tests.rs"]
162mod tests;
coven_database/sync_facts.rs

coven_database/
sync_facts.rs
