sofie_db/
read_versions.rs

1use std::cell::RefCell;
2use std::hash::{DefaultHasher, Hash, Hasher};
3
4use sofie_core::{ColumnReader, Date, Result, SqlType, SqlValue, Timestamp, Uuid};
5
6use crate::RowSink;
7
8// Task-local rather than carried by `Read`, so the rows a lookup reads
9// through its own `Db` count too.
10tokio::task_local! {
11    static READS: RefCell<Reads>;
12}
13
14struct Reads {
15    hasher: DefaultHasher,
16    untracked: bool,
17}
18
19/// `seed` keys the fingerprint by the answer's other inputs; `None` when one has no version.
20pub async fn track_reads<F: Future>(seed: u64, future: F) -> (F::Output, Option<u64>) {
21    let mut hasher = DefaultHasher::new();
22    seed.hash(&mut hasher);
23    let reads = Reads {
24        hasher,
25        untracked: false,
26    };
27    READS
28        .scope(RefCell::new(reads), async {
29            let output = future.await;
30            (output, reads_so_far())
31        })
32        .await
33}
34
35/// Every row read so far in this task; `None` outside `track_reads` or after `read_untracked`.
36#[must_use]
37pub fn reads_so_far() -> Option<u64> {
38    READS
39        .try_with(|reads| {
40            let reads = reads.borrow();
41            (!reads.untracked).then(|| reads.hasher.clone().finish())
42        })
43        .ok()
44        .flatten()
45}
46
47/// The answer depends on another app's or a vendor's reply, which no fingerprint here follows.
48pub fn read_untracked() {
49    let _ = READS.try_with(|reads| reads.borrow_mut().untracked = true);
50}
51
52/// The answer depends on a cached value, which changes the fingerprint as a read row does.
53pub fn read_cached(value: &str) {
54    let _ = READS.try_with(|reads| value.hash(&mut reads.borrow_mut().hasher));
55}
56
57pub(crate) fn tracking() -> bool {
58    READS.try_with(|_| ()).is_ok()
59}
60
61/// Hashes every value its rows' targets read, so a changed row changes the fingerprint.
62pub(crate) struct Fingerprinted<'s> {
63    rows: &'s mut dyn RowSink,
64    hasher: DefaultHasher,
65    count: u64,
66}
67
68impl<'s> Fingerprinted<'s> {
69    pub(crate) fn new(rows: &'s mut dyn RowSink) -> Self {
70        Self {
71            rows,
72            hasher: DefaultHasher::new(),
73            count: 0,
74        }
75    }
76
77    pub(crate) fn record(self) {
78        let _ = READS.try_with(|reads| {
79            let hasher = &mut reads.borrow_mut().hasher;
80            self.count.hash(hasher);
81            self.hasher.finish().hash(hasher);
82        });
83    }
84}
85
86impl RowSink for Fingerprinted<'_> {
87    fn expects(&self) -> &[Option<SqlType>] {
88        self.rows.expects()
89    }
90
91    fn row(&mut self, row: &mut dyn ColumnReader) -> Result<()> {
92        self.count += 1;
93        self.rows.row(&mut Hashing {
94            row,
95            hasher: &mut self.hasher,
96        })
97    }
98}
99
100struct Hashing<'r> {
101    row: &'r mut dyn ColumnReader,
102    hasher: &'r mut DefaultHasher,
103}
104
105macro_rules! hashed {
106    ($($method:ident -> $type:ty),*) => {
107        $(fn $method(&mut self) -> Result<$type> {
108            let value = self.row.$method()?;
109            value.hash(self.hasher);
110            Ok(value)
111        })*
112    };
113}
114
115impl ColumnReader for Hashing<'_> {
116    fn remaining(&self) -> usize {
117        self.row.remaining()
118    }
119
120    fn is_null_at(&self, ahead: usize) -> Result<bool> {
121        self.row.is_null_at(ahead)
122    }
123
124    fn skip(&mut self) -> Result<()> {
125        self.value().map(drop)
126    }
127
128    hashed!(bool -> bool, int -> i64, uuid -> Uuid, timestamp -> Timestamp, date -> Date,
129        date_time -> sofie_core::DateTime, text -> &str, bytes -> &[u8]);
130
131    fn float(&mut self) -> Result<f64> {
132        let value = self.row.float()?;
133        value.to_bits().hash(self.hasher);
134        Ok(value)
135    }
136
137    fn json(&mut self) -> Result<serde_json::Value> {
138        let value = self.row.json()?;
139        value.to_string().hash(self.hasher);
140        Ok(value)
141    }
142
143    fn value(&mut self) -> Result<SqlValue> {
144        let value = self.row.value()?;
145        hash_value(&value, self.hasher);
146        Ok(value)
147    }
148}
149
150fn hash_value(value: &SqlValue, hasher: &mut DefaultHasher) {
151    std::mem::discriminant(value).hash(hasher);
152    match value {
153        SqlValue::Null => {}
154        SqlValue::Bool(value) => value.hash(hasher),
155        SqlValue::Int(value) => value.hash(hasher),
156        SqlValue::Float(value) => value.to_bits().hash(hasher),
157        SqlValue::Text(value) => value.hash(hasher),
158        SqlValue::Bytes(value) => value.hash(hasher),
159        SqlValue::Uuid(value) => value.hash(hasher),
160        SqlValue::Timestamp(value) => value.hash(hasher),
161        SqlValue::Date(value) => value.hash(hasher),
162        SqlValue::DateTime(value) => value.hash(hasher),
163        SqlValue::Json(value) => value.to_string().hash(hasher),
164        SqlValue::Array(values) => {
165            values.len().hash(hasher);
166            for value in values {
167                hash_value(value, hasher);
168            }
169        }
170    }
171}
172
173#[cfg(test)]
174mod tests {
175    use sofie_core::RowReader;
176
177    use super::*;
178    use crate::Decoding;
179
180    fn row(version: i64) -> Vec<SqlValue> {
181        vec![SqlValue::Int(version), SqlValue::Text("open".to_owned())]
182    }
183
184    fn read(rows: &[Vec<SqlValue>], skip_version: bool) {
185        let mut decoding = Decoding::new(Vec::new(), |row: &mut dyn ColumnReader| {
186            if skip_version {
187                row.skip()?;
188            } else {
189                row.int()?;
190            }
191            row.text().map(str::to_owned)
192        });
193        let mut fingerprinted = Fingerprinted::new(&mut decoding);
194        for values in rows {
195            fingerprinted.row(&mut RowReader::new(values)).unwrap();
196        }
197        fingerprinted.record();
198    }
199
200    fn fingerprint(seed: u64, rows: &[Vec<SqlValue>], skip_version: bool) -> u64 {
201        let runtime = tokio::runtime::Builder::new_current_thread()
202            .build()
203            .unwrap();
204        let reading = async { read(rows, skip_version) };
205        runtime.block_on(track_reads(seed, reading)).1.unwrap()
206    }
207
208    fn tracked(reading: impl FnOnce()) -> Option<u64> {
209        let runtime = tokio::runtime::Builder::new_current_thread()
210            .build()
211            .unwrap();
212        runtime.block_on(track_reads(1, async { reading() })).1
213    }
214
215    #[test]
216    fn the_fingerprint_follows_the_values_read_and_the_seed() {
217        let first = fingerprint(1, &[row(1)], false);
218
219        assert_eq!(first, fingerprint(1, &[row(1)], false));
220        assert_ne!(first, fingerprint(1, &[row(2)], false));
221        assert_ne!(first, fingerprint(2, &[row(1)], false));
222        assert_ne!(first, fingerprint(1, &[row(1), row(1)], false));
223    }
224
225    #[test]
226    fn a_skipped_column_still_counts() {
227        assert_ne!(
228            fingerprint(1, &[row(1)], true),
229            fingerprint(1, &[row(2)], true)
230        );
231    }
232
233    #[test]
234    fn an_answer_from_outside_is_not_fingerprinted() {
235        assert_eq!(tracked(read_untracked), None);
236    }
237
238    #[test]
239    fn a_cached_value_read_changes_the_fingerprint() {
240        let first = tracked(|| read_cached("{\"total\":1}"));
241
242        assert!(first.is_some());
243        assert_eq!(first, tracked(|| read_cached("{\"total\":1}")));
244        assert_ne!(first, tracked(|| read_cached("{\"total\":2}")));
245    }
246
247    #[test]
248    fn nothing_is_tracked_outside_a_scope() {
249        read(&[row(1)], false);
250
251        assert_eq!(reads_so_far(), None);
252    }
253}
sofie_db/read_versions.rs

sofie_db/
read_versions.rs
