coven_database/
transfer_watch.rs

1//! How far each file in flight one way has got, for a reader that draws each
2//! one: all of them first, then each file whose progress moved, started or
3//! ended, and how it ended (E7, E8). Progress lives in this process's memory
4//! and changes with each step of bytes, so no table holds it and no live
5//! query reads it.
6
7use crate::observation::CommitSubscription;
8use crate::transfers_in_flight::InFlight;
9use crate::TransferProgress;
10use coven_foundation::id_source::FileId;
11use std::collections::BTreeMap;
12use tokio::sync::watch;
13
14/// A change to one file's progress, or every file's.
15#[derive(Clone, Debug, PartialEq, Eq)]
16pub enum TransferChange {
17    /// Every file in flight, as a reader first hears them; a file not
18    /// listed has nothing in flight.
19    Reset(BTreeMap<FileId, TransferProgress>),
20    /// The file started, or its progress moved.
21    Updated {
22        /// The file, as `coven_sending` or `coven_downloading` names it.
23        file: FileId,
24        /// How far it has got now.
25        progress: TransferProgress,
26    },
27    /// The file's transfer ended.
28    Removed {
29        /// The file whose transfer ended.
30        file: FileId,
31        /// How it ended.
32        end: TransferEnd,
33    },
34}
35
36/// How a file's transfer ended, decided in the step that takes it out of
37/// the files in flight. When two attempts move one file at once, the last
38/// to end says it.
39#[derive(Clone, Copy, Debug, PartialEq, Eq)]
40pub enum TransferEnd {
41    /// Its bytes all moved: an upload stored them, or a download assembled
42    /// and checked the file's whole copy, which the write committing its
43    /// pin keeps.
44    Finished,
45    /// It failed, and its failure went where the transfer's failures go:
46    /// an upload's pending record, a pin's call.
47    Failed,
48    /// A hold stopped it before it finished, and nothing failed: a metered
49    /// network, and for an upload also the pause, or a self-replacement
50    /// pausing writes. It starts again once the hold ends (§12.3, §17.8,
51    /// §17.9).
52    Held,
53    /// Whatever moved its bytes dropped it before it ended, as a pin's
54    /// caller does that drops the call.
55    Dropped,
56}
57
58/// What a reader last heard of one file: how far it had got, or, once it
59/// ended since, how.
60#[derive(Clone, Copy, PartialEq, Eq)]
61pub(crate) enum Heard {
62    Moved(TransferProgress),
63    Ended(TransferEnd),
64}
65
66/// What a reader that heard `heard` must hear to hold `now`, the files in
67/// flight: each file that ended, then each file new, moved, or started
68/// again since it ended. `heard` becomes `now`.
69pub(crate) fn changes_since(
70    heard: &mut BTreeMap<FileId, Heard>,
71    now: BTreeMap<FileId, TransferProgress>,
72) -> Vec<TransferChange> {
73    let mut changes: Vec<TransferChange> = heard
74        .iter()
75        .filter_map(|(file, heard)| match heard {
76            Heard::Ended(end) => Some(TransferChange::Removed { file: *file, end: *end }),
77            Heard::Moved(_) => None,
78        })
79        .collect();
80    changes.extend(
81        now.iter()
82            .filter(|(file, progress)| heard.get(*file) != Some(&Heard::Moved(**progress)))
83            .map(|(file, progress)| TransferChange::Updated { file: *file, progress: *progress }),
84    );
85    *heard = now.into_iter().map(|(file, progress)| (file, Heard::Moved(progress))).collect();
86    changes
87}
88
89/// Hears how far each file in flight one way has got. Changes that arrive
90/// faster than the reader reads are heard together: the next read names
91/// each file that differs from what the reader last heard, and how each
92/// one it heard in flight ended.
93pub struct TransferWatch {
94    owner: InFlight,
95    /// This watch's number among its owner's.
96    number: u64,
97    /// Told each time a file starts, moves or ends.
98    changed: watch::Receiver<()>,
99    /// Ends the watch once the store closes.
100    store: CommitSubscription,
101}
102
103impl TransferWatch {
104    pub(crate) fn new(
105        owner: InFlight,
106        number: u64,
107        changed: watch::Receiver<()>,
108        store: CommitSubscription,
109    ) -> Self {
110        Self { owner, number, changed, store }
111    }
112
113    /// Every file in flight as one `Reset` on the first read; after that,
114    /// wait for a file's progress to change and name each file it changed
115    /// for. `None` once the store has closed.
116    pub async fn next(&mut self) -> Option<Vec<TransferChange>> {
117        loop {
118            if self.store.state().closed() {
119                return None;
120            }
121            // Seen before the read, so a change after the read wakes the
122            // wait below.
123            self.changed.borrow_and_update();
124            let changes = self.owner.read(self.number);
125            if !changes.is_empty() {
126                return Some(changes);
127            }
128            tokio::select! {
129                changed = self.changed.changed() => {
130                    changed.expect("the owner this watch holds keeps the sender");
131                }
132                () = self.store.changed() => {}
133            }
134        }
135    }
136}
137
138impl Drop for TransferWatch {
139    fn drop(&mut self) {
140        self.owner.forget(self.number);
141    }
142}
143
144#[cfg(test)]
145#[path = "transfer_watch_tests.rs"]
146mod tests;
coven_database/transfer_watch.rs

coven_database/
transfer_watch.rs
