coven_database/
transfers_in_flight.rs

1//! The stored objects this process is transferring now (§4, §17.8, §17.9):
2//! the ones it is sending, and the ones it is downloading to keep whole.
3//! Nothing about them is stored: an attempt ends when its task does, a crash
4//! included, so a reopened store starts with none. App SQL reads them through
5//! coven's SQL functions, as `coven_uploads.sending` and the `coven_sending`
6//! and `coven_downloading` views, which name each file over an object, and a
7//! live query that reads one runs again as a transfer it shows starts or
8//! ends. How far each has got is progress no table holds: a `TransferWatch`
9//! hears it, by file, and how each ended (E7, E8).
10
11use crate::observation::{CommitObserver, RowChange};
12use crate::transfer_watch::{changes_since, Heard, TransferChange, TransferEnd};
13use coven_foundation::id_source::{DeviceId, FileId, ObjectId};
14use coven_foundation::locks::lock_unpoisoned;
15use rusqlite::functions::FunctionFlags;
16use std::collections::BTreeMap;
17use std::sync::{Arc, Mutex};
18use tokio::sync::watch;
19
20/// How far one transfer has got: the bytes of its encrypted object
21/// handed to the connection or received, in whatever order, out of the
22/// object's whole size. Known from the transfer's start; never stored.
23#[derive(Clone, Copy, Debug, PartialEq, Eq)]
24pub struct TransferProgress {
25    /// The object's bytes moved so far.
26    pub bytes_transferred: u64,
27    /// The object's whole encrypted size.
28    pub bytes_total: u64,
29}
30
31/// One way files go, and the SQL that reads the files going that way.
32struct Direction {
33    /// What a start or end changes, as live queries read it: a view's
34    /// column, or a view whose rows it changes.
35    read_as: &'static [(&'static str, &'static str)],
36    /// `name(object)`, whether the stored object with this id is in flight,
37    /// for a view that asks of each of its rows.
38    contains: Option<&'static str>,
39    /// `name()`, the objects in flight as a JSON array of hex text, each
40    /// one's uploader's eight id bytes and then its own sixteen: what a view
41    /// joins by index.
42    objects: &'static str,
43}
44
45const SENDING: Direction = Direction {
46    read_as: &[("coven_uploads", "sending"), ("coven_sending", "")],
47    contains: Some("_coven_sending"),
48    objects: "_coven_sending_objects",
49};
50
51const DOWNLOADING: Direction = Direction {
52    read_as: &[("coven_downloading", "")],
53    contains: None,
54    objects: "_coven_downloading_objects",
55};
56
57/// Coven's functions that read the files in flight, which app SQL reaches
58/// only through coven's views.
59pub(crate) const FUNCTIONS: [&str; 3] =
60    ["_coven_sending", "_coven_sending_objects", "_coven_downloading_objects"];
61
62/// Both ways files go: one owner per open store, whose clones share it.
63#[derive(Clone)]
64pub(crate) struct TransfersInFlight {
65    uploads: InFlight,
66    downloads: InFlight,
67}
68
69impl TransfersInFlight {
70    pub(crate) fn new(observer: CommitObserver) -> Self {
71        Self {
72            uploads: InFlight::new(&SENDING, observer.clone()),
73            downloads: InFlight::new(&DOWNLOADING, observer),
74        }
75    }
76
77    /// The files this process is sending, as the upload worker marks them.
78    pub(crate) fn uploads(&self) -> InFlight {
79        self.uploads.clone()
80    }
81
82    /// The files this process is downloading to keep whole, as each pin's
83    /// download marks them.
84    pub(crate) fn downloads(&self) -> InFlight {
85        self.downloads.clone()
86    }
87
88    /// Register both ways' functions on a connection to the store's
89    /// database. None is deterministic; all are safe in coven's views.
90    pub(crate) fn register(&self, connection: &rusqlite::Connection) -> rusqlite::Result<()> {
91        self.uploads.register(connection)?;
92        self.downloads.register(connection)
93    }
94}
95
96/// The stored objects this process is transferring one way. The worker
97/// transferring them adds each attempt as it starts and removes it as it
98/// ends, saying how, cancelled included. Clones share it.
99#[derive(Clone)]
100pub struct InFlight {
101    shared: Arc<InFlightShared>,
102}
103
104struct InFlightShared {
105    direction: &'static Direction,
106    state: Mutex<InFlightState>,
107    /// Told of each start and end, as of a commit that changed what the
108    /// live queries reading the direction's views read.
109    observer: CommitObserver,
110    /// Tells each watch that a file started, moved or ended, after the state
111    /// changed; a watch reads the change from the state.
112    changed: watch::Sender<()>,
113}
114
115#[derive(Default)]
116struct InFlightState {
117    /// Each object in flight, with its uploader and its attempts.
118    objects: BTreeMap<ObjectId, Entry>,
119    /// The next attempt's number, so an ended attempt is never taken for a
120    /// later one of the same object.
121    next: u64,
122    /// Each open watch, by number, with what its reader last heard of each
123    /// file: how far it had got, or how it ended since. `None` before the
124    /// watch's first read. An attempt that ends marks its files ended here
125    /// in the same step that takes them out of flight, so no reader can
126    /// hear a file gone before it hears how it ended.
127    watches: BTreeMap<u64, Option<BTreeMap<FileId, Heard>>>,
128    /// The next watch's number.
129    next_watch: u64,
130}
131
132struct Entry {
133    device: DeviceId,
134    /// Its attempts in flight, oldest first, each with the files it moves
135    /// the object's bytes for and how far it has got. One pin can keep the
136    /// same object twice at once: it stays in flight until both end, and
137    /// shows the newer one's progress.
138    attempts: Vec<(u64, Vec<FileId>, TransferProgress)>,
139}
140
141impl InFlightState {
142    /// An attempt for `files` ended as `end` says: each of them no attempt
143    /// still moves is ended for every watch that heard it.
144    fn ended(&mut self, files: &[FileId], end: TransferEnd) {
145        let now = self.progress();
146        for file in files.iter().filter(|file| !now.contains_key(*file)) {
147            for heard in self.watches.values_mut().flatten() {
148                if let Some(heard) = heard.get_mut(file) {
149                    *heard = Heard::Ended(end);
150                }
151            }
152        }
153    }
154
155    /// Each file in flight with the progress its object's newest attempt
156    /// shows.
157    fn progress(&self) -> BTreeMap<FileId, TransferProgress> {
158        let mut progress = BTreeMap::new();
159        for entry in self.objects.values() {
160            for (_, files, moved) in &entry.attempts {
161                for file in files {
162                    progress.insert(*file, *moved);
163                }
164            }
165        }
166        progress
167    }
168}
169
170impl InFlight {
171    fn new(direction: &'static Direction, observer: CommitObserver) -> Self {
172        Self {
173            shared: Arc::new(InFlightShared {
174                direction,
175                state: Mutex::new(InFlightState::default()),
176                observer,
177                changed: watch::channel(()).0,
178            }),
179        }
180    }
181
182    fn state(&self) -> std::sync::MutexGuard<'_, InFlightState> {
183        lock_unpoisoned(&self.shared.state)
184    }
185
186    /// The stored objects in flight now.
187    pub fn current(&self) -> std::collections::BTreeSet<ObjectId> {
188        self.state().objects.keys().copied().collect()
189    }
190
191    /// Mark the stored object, which `device` uploaded, as in flight for
192    /// `files` until the returned guard drops, as far as `progress` says.
193    pub fn start(
194        &self,
195        device: DeviceId,
196        object: ObjectId,
197        files: Vec<FileId>,
198        progress: TransferProgress,
199    ) -> Transferring {
200        let (attempt, started) = {
201            let mut state = self.state();
202            let attempt = state.next;
203            state.next += 1;
204            let entry = state
205                .objects
206                .entry(object)
207                .or_insert_with(|| Entry { device, attempts: Vec::new() });
208            entry.attempts.push((attempt, files, progress));
209            (attempt, entry.attempts.len() == 1)
210        };
211        self.shared.changed.send_replace(());
212        if started {
213            self.changed();
214        }
215        Transferring { owner: self.clone(), object, attempt, end: TransferEnd::Dropped }
216    }
217
218    /// Hear how far each file in flight has got: all of them first, then
219    /// each change, and how each file ended, until the store closes.
220    pub fn subscribe(&self) -> crate::TransferWatch {
221        let number = {
222            let mut state = self.state();
223            let number = state.next_watch;
224            state.next_watch += 1;
225            state.watches.insert(number, None);
226            number
227        };
228        crate::TransferWatch::new(
229            self.clone(),
230            number,
231            self.shared.changed.subscribe(),
232            self.shared.observer.subscribe(),
233        )
234    }
235
236    /// What watch `number` has yet to hear: every file in flight on its
237    /// first read, and after that each change since its last.
238    pub(crate) fn read(&self, number: u64) -> Vec<TransferChange> {
239        let mut state = self.state();
240        let now = state.progress();
241        let heard = state.watches.get_mut(&number).expect("a watch is registered until it drops");
242        match heard {
243            Some(heard) => changes_since(heard, now),
244            None => {
245                *heard = Some(
246                    now.iter().map(|(file, progress)| (*file, Heard::Moved(*progress))).collect(),
247                );
248                vec![TransferChange::Reset(now)]
249            }
250        }
251    }
252
253    /// Watch `number` dropped: nothing hears for it any more.
254    pub(crate) fn forget(&self, number: u64) {
255        self.state().watches.remove(&number);
256    }
257
258    /// The attempt has got `bytes` of the way; an attempt that ended moves
259    /// nothing.
260    fn reached(&self, object: ObjectId, attempt: u64, bytes: u64) {
261        let mut state = self.state();
262        let Some(progress) = state.objects.get_mut(&object).and_then(|entry| {
263            entry
264                .attempts
265                .iter_mut()
266                .find(|(number, _, _)| *number == attempt)
267                .map(|(_, _, progress)| progress)
268        }) else {
269            return;
270        };
271        progress.bytes_transferred = bytes;
272        drop(state);
273        self.shared.changed.send_replace(());
274    }
275
276    /// Tell live queries after the files changed, so a query that ran
277    /// before the change runs again, and one that runs after it reads them.
278    fn changed(&self) {
279        self.shared.observer.commit(
280            crate::observation::Judged::unjudged(
281                self.shared
282                    .direction
283                    .read_as
284                    .iter()
285                    .map(|(table, column)| RowChange {
286                        table: (*table).into(),
287                        column: (*column).into(),
288                        rows: None,
289                    })
290                    .collect(),
291            ),
292            None,
293        );
294    }
295
296    fn register(&self, connection: &rusqlite::Connection) -> rusqlite::Result<()> {
297        let flags = FunctionFlags::SQLITE_UTF8 | FunctionFlags::SQLITE_INNOCUOUS;
298        if let Some(name) = self.shared.direction.contains {
299            let owner = self.clone();
300            connection.create_scalar_function(name, 1, flags, move |ctx| {
301                // Anything but an object id fails the statement.
302                Ok(owner.state().objects.contains_key(&ctx.get::<ObjectId>(0)?))
303            })?;
304        }
305        let owner = self.clone();
306        connection.create_scalar_function(self.shared.direction.objects, 0, flags, move |_| {
307            let objects: Vec<String> = owner
308                .state()
309                .objects
310                .iter()
311                .map(|(object, entry)| format!("{:016x}{}", entry.device.0, object.0.simple()))
312                .collect();
313            serde_json::to_string(&objects)
314                .map_err(|error| rusqlite::Error::UserFunctionError(Box::new(error)))
315        })
316    }
317}
318
319/// One attempt to transfer a stored object. Its worker ends it as its
320/// transfer ended; dropping it otherwise, as a cancelled attempt does, ends
321/// it as dropped.
322pub struct Transferring {
323    owner: InFlight,
324    object: ObjectId,
325    attempt: u64,
326    /// How the attempt ends when this drops.
327    end: TransferEnd,
328}
329
330impl Transferring {
331    /// What tells this attempt's progress, for whatever moves its bytes.
332    pub fn meter(&self) -> TransferMeter {
333        TransferMeter { owner: self.owner.clone(), object: self.object, attempt: self.attempt }
334    }
335
336    /// End the attempt as `end` says.
337    pub fn end(mut self, end: TransferEnd) {
338        self.end = end;
339    }
340}
341
342/// Tells one attempt how far it has got. Once the attempt ends, it tells
343/// nothing: a request still finishing can't move a later attempt.
344#[derive(Clone)]
345pub struct TransferMeter {
346    owner: InFlight,
347    object: ObjectId,
348    attempt: u64,
349}
350
351impl TransferMeter {
352    /// The attempt has moved `bytes` of the file's object.
353    pub fn reached(&self, bytes: u64) {
354        self.owner.reached(self.object, self.attempt, bytes);
355    }
356}
357
358impl Drop for Transferring {
359    fn drop(&mut self) {
360        let ended = {
361            let mut state = self.owner.state();
362            let attempt = state.objects.get_mut(&self.object).and_then(|entry| {
363                let at =
364                    entry.attempts.iter().position(|(attempt, _, _)| *attempt == self.attempt)?;
365                let (_, files, _) = entry.attempts.remove(at);
366                Some((files, entry.attempts.is_empty()))
367            });
368            attempt.map(|(files, ended)| {
369                if ended {
370                    state.objects.remove(&self.object);
371                }
372                state.ended(&files, self.end);
373                ended
374            })
375        };
376        match ended {
377            Some(ended) => {
378                self.owner.shared.changed.send_replace(());
379                if ended {
380                    self.owner.changed();
381                }
382            }
383            // An attempt in flight keeps its place in its object's entry;
384            // without it the progress watches read is wrong. This drop may
385            // run during an unwind, so it fails the store's handle instead of
386            // panicking, and the bug goes to the call it runs in or the bug
387            // sink (E15).
388            None => self.owner.shared.observer.fail_with_bug(coven_foundation::task::Bug::new(
389                "end a transfer in flight",
390                "an attempt in flight lost its place in its object's entry",
391            )),
392        }
393    }
394}
395
396#[cfg(test)]
397#[path = "transfers_in_flight_tests.rs"]
398mod tests;
coven_database/transfers_in_flight.rs

coven_database/
transfers_in_flight.rs
