sofie_http/
screen_cache.rs

1//! A screen's loaded reply kept for its path, query, reach and principal until a record it read
2//! changes, here or on another instance, or its time runs out.
3
4use std::collections::{HashMap, HashSet, VecDeque};
5use std::sync::{Arc, Mutex, OnceLock, Weak};
6use std::time::{Duration, Instant};
7
8use serde_json::Value;
9use sofie_core::{Result, Shown, Uuid, unpoisoned};
10use sofie_db::{Announced, Db, Reach, Reply, ReplyData, Topic, Topics};
11use tokio::sync::broadcast::Receiver;
12use tokio::sync::broadcast::error::{RecvError, TryRecvError};
13
14use crate::context::Matched;
15use crate::target::Target;
16use crate::{Area, PageRoute};
17
18/// Bounds the memory a crawl over many paths can take.
19const MAX_ENTRIES: usize = 10_000;
20const MAX_TIME: Duration = Duration::from_hours(1);
21
22impl<A: Area> PageRoute<'_, A> {
23    /// Reuses the loader's reply for the same path, query, reach and principal for at most `time`
24    /// (an hour at most), dropping it once a record it read changes. Only for a loader
25    /// whose reply depends on nothing else: not on flags, permissions or the clock.
26    pub fn cached(&mut self, time: Duration) -> &mut Self {
27        self.rule(|rules| rules.cached = Some(time.min(MAX_TIME)))
28    }
29}
30
31/// The principal too: a permitted area reaches everything for every principal, yet a loader
32/// may still answer by who asks.
33#[derive(PartialEq, Eq, Hash)]
34struct Key {
35    target: String,
36    reach: Reach,
37    principal: Option<Uuid>,
38}
39
40/// Its topics are never `Every`, which nothing keeps.
41struct Entry {
42    reply: Arc<Reply>,
43    topics: Topics,
44    until: Instant,
45}
46
47/// The changes heard last, so a load that spanned some keeps its reply unless one concerns it.
48const RECENT_CHANGES: usize = 256;
49
50#[derive(Default)]
51struct Kept {
52    entries: HashMap<Arc<Key>, Entry>,
53    by_topic: HashMap<Topic, HashSet<Arc<Key>>>,
54    recent: VecDeque<Announced>,
55    heard: u64,
56    changes: Option<Receiver<Announced>>,
57}
58
59#[derive(Default)]
60pub(crate) struct ScreenCache {
61    kept: Arc<Mutex<Kept>>,
62    listening: OnceLock<()>,
63}
64
65impl ScreenCache {
66    pub(crate) async fn read(
67        &self,
68        db: &Db,
69        target: &Target,
70        time: Duration,
71        matched: Matched,
72        asked: &http::Uri,
73    ) -> (Result<Reply>, Topics) {
74        self.listen(db);
75        let key = Key {
76            target: asked.to_string(),
77            reach: matched.reach,
78            principal: matched.session.principal(),
79        };
80        let hearing = sofie_db::hears_other_instances(db);
81        let before = match unpoisoned::lock(&self.kept).hit(&key, hearing) {
82            Ok((reply, topics)) => return (Ok(shared(reply)), topics),
83            Err(heard) => heard,
84        };
85        let load = (target.endpoint)(db.clone(), matched);
86        let (loaded, topics) = sofie_db::track_topics(load).await;
87        let Ok(reply) = loaded else {
88            return (loaded, topics);
89        };
90        let reply = Arc::new(reply);
91        if topics != Topics::Every {
92            let entry = Entry {
93                reply: Arc::clone(&reply),
94                topics: topics.clone(),
95                until: Instant::now() + time,
96            };
97            let hearing = sofie_db::hears_other_instances(db);
98            unpoisoned::lock(&self.kept).keep(key, entry, before, hearing);
99        }
100        (Ok(shared(reply)), topics)
101    }
102
103    fn listen(&self, db: &Db) {
104        self.listening.get_or_init(|| {
105            unpoisoned::lock(&self.kept).changes = Some(sofie_db::committed_changes(db));
106            let woken = sofie_db::committed_changes(db);
107            tokio::spawn(drain_on_change(Arc::downgrade(&self.kept), woken));
108        });
109    }
110}
111
112impl Kept {
113    /// `hearing` is read before draining: the listener queues `MaybeMissed` before it says so.
114    /// While other instances go unheard, nothing kept can be trusted.
115    fn catch_up(&mut self, hearing: bool) {
116        self.drain();
117        if !hearing {
118            self.hear(Announced::MaybeMissed);
119        }
120    }
121
122    fn drain(&mut self) {
123        while let Some(changes) = &mut self.changes {
124            let announced = match changes.try_recv() {
125                Ok(announced) => announced,
126                Err(TryRecvError::Lagged(_)) => Announced::MaybeMissed,
127                Err(TryRecvError::Empty | TryRecvError::Closed) => return,
128            };
129            self.hear(announced);
130        }
131    }
132
133    /// A miss tells how many changes were heard by then.
134    fn hit(&mut self, key: &Key, hearing: bool) -> Result<(Arc<Reply>, Topics), u64> {
135        self.catch_up(hearing);
136        self.entries
137            .get(key)
138            .filter(|entry| entry.until > Instant::now())
139            .map(|entry| (Arc::clone(&entry.reply), entry.topics.clone()))
140            .ok_or(self.heard)
141    }
142
143    /// Under the lock the listener forgets under, so a change heard later finds the entry.
144    fn keep(&mut self, key: Key, entry: Entry, before: u64, hearing: bool) {
145        self.catch_up(hearing);
146        if self.changed_since(before, &entry.topics, key.reach) {
147            return;
148        }
149        if self.entries.len() >= MAX_ENTRIES {
150            self.make_room(Instant::now());
151        }
152        let key = Arc::new(key);
153        self.remove(&key);
154        for topic in read(&entry.topics) {
155            self.by_topic
156                .entry(*topic)
157                .or_default()
158                .insert(Arc::clone(&key));
159        }
160        self.entries.insert(key, entry);
161    }
162
163    fn changed_since(&self, before: u64, topics: &Topics, reach: Reach) -> bool {
164        let Ok(missed) = usize::try_from(self.heard - before) else {
165            return true;
166        };
167        if missed > self.recent.len() {
168            return true;
169        }
170        self.recent
171            .iter()
172            .rev()
173            .take(missed)
174            .any(|announced| announced.concerns(topics, reach))
175    }
176
177    fn hear(&mut self, announced: Announced) {
178        self.heard += 1;
179        match &announced {
180            Announced::Changed(changes) => {
181                for change in changes.iter() {
182                    let concerned: Vec<Arc<Key>> = self
183                        .by_topic
184                        .get(&change.topic)
185                        .into_iter()
186                        .flatten()
187                        .filter(|key| change.reaches(key.reach))
188                        .cloned()
189                        .collect();
190                    for key in concerned {
191                        self.remove(&key);
192                    }
193                }
194            }
195            Announced::MaybeMissed => {
196                self.entries.clear();
197                self.by_topic.clear();
198            }
199        }
200        if self.recent.len() == RECENT_CHANGES {
201            self.recent.pop_front();
202        }
203        self.recent.push_back(announced);
204    }
205
206    fn remove(&mut self, key: &Key) {
207        let Some(entry) = self.entries.remove(key) else {
208            return;
209        };
210        for topic in read(&entry.topics) {
211            if let Some(keys) = self.by_topic.get_mut(topic) {
212                keys.remove(key);
213                if keys.is_empty() {
214                    self.by_topic.remove(topic);
215                }
216            }
217        }
218    }
219
220    /// Expired entries first, then the sixteenth closest to expiring, so a crawl over many
221    /// queries cannot keep anything new from being cached.
222    fn make_room(&mut self, now: Instant) {
223        self.drop_until(now);
224        if self.entries.len() < MAX_ENTRIES {
225            return;
226        }
227        let mut untils: Vec<Instant> = self.entries.values().map(|entry| entry.until).collect();
228        let (_, cutoff, _) = untils.select_nth_unstable(MAX_ENTRIES / 16);
229        self.drop_until(*cutoff);
230    }
231
232    fn drop_until(&mut self, cutoff: Instant) {
233        let dropped: Vec<Arc<Key>> = self
234            .entries
235            .iter()
236            .filter(|(_, entry)| entry.until <= cutoff)
237            .map(|(key, _)| Arc::clone(key))
238            .collect();
239        for key in dropped {
240            self.remove(&key);
241        }
242    }
243}
244
245fn read(topics: &Topics) -> impl Iterator<Item = &Topic> {
246    match topics {
247        Topics::Of(read) => Some(read),
248        Topics::Every => None,
249    }
250    .into_iter()
251    .flatten()
252}
253
254/// Its own receiver only wakes it, so entries are forgotten between requests too; what it heard
255/// is drained from the cache's receiver, which a request drains as well.
256async fn drain_on_change(kept: Weak<Mutex<Kept>>, mut woken: Receiver<Announced>) {
257    loop {
258        if let Err(RecvError::Closed) = woken.recv().await {
259            return;
260        }
261        while !matches!(
262            woken.try_recv(),
263            Err(TryRecvError::Empty | TryRecvError::Closed)
264        ) {}
265        let Some(kept) = kept.upgrade() else {
266            return;
267        };
268        unpoisoned::lock(&kept).drain();
269    }
270}
271
272fn shared(reply: Arc<Reply>) -> Reply {
273    Reply::Data(Box::new(Shared(reply)))
274}
275
276/// Many requests answer from one kept reply.
277struct Shared(Arc<Reply>);
278
279impl ReplyData for Shared {
280    fn write_json(&self, out: &mut Vec<u8>) -> serde_json::Result<()> {
281        self.0.write_json(out).map_err(json_error)
282    }
283
284    fn to_json(&self) -> serde_json::Result<Value> {
285        self.0.to_json().map_err(json_error)
286    }
287
288    fn type_name(&self) -> &'static str {
289        match self.0.as_ref() {
290            Reply::Data(data) => data.type_name(),
291            _ => std::any::type_name::<Reply>(),
292        }
293    }
294
295    fn shown(&self) -> Option<Shown<'_>> {
296        self.0.shown()
297    }
298}
299
300fn json_error(error: sofie_core::Error) -> serde_json::Error {
301    serde_json::Error::io(std::io::Error::other(error))
302}
303
304#[cfg(test)]
305mod tests {
306    use std::collections::BTreeSet;
307
308    use sofie_db::{Readers, TableChange};
309    use tokio::sync::broadcast;
310
311    use super::*;
312
313    fn key(target: &str) -> Key {
314        Key {
315            target: target.to_owned(),
316            reach: Reach::All,
317            principal: None,
318        }
319    }
320
321    fn entry(tables: &[&str], time: Duration) -> Entry {
322        let topics = tables
323            .iter()
324            .map(|table| Topic::named("shop", table))
325            .collect::<BTreeSet<_>>();
326        Entry {
327            reply: Arc::new(Reply::Empty),
328            topics: Topics::Of(topics),
329            until: Instant::now() + time,
330        }
331    }
332
333    fn changed(table: &str) -> Announced {
334        Announced::Changed(Arc::from([TableChange {
335            topic: Topic::named("shop", table),
336            readers: Readers::Anyone,
337        }]))
338    }
339
340    #[test]
341    fn a_hit_hears_a_change_already_sent_and_nothing_is_kept_while_others_go_unheard() {
342        let (sender, receiver) = broadcast::channel(16);
343        let mut kept = Kept {
344            changes: Some(receiver),
345            ..Kept::default()
346        };
347        kept.keep(key("/orders"), entry(&["order"], MAX_TIME), 0, true);
348        sender.send(changed("order")).unwrap();
349
350        assert!(kept.hit(&key("/orders"), true).is_err());
351
352        let before = kept.heard;
353        kept.keep(key("/orders"), entry(&["order"], MAX_TIME), before, true);
354        assert!(kept.hit(&key("/orders"), false).is_err());
355        let before = kept.heard;
356        kept.keep(key("/orders"), entry(&["order"], MAX_TIME), before, false);
357        assert!(kept.hit(&key("/orders"), true).is_err());
358    }
359
360    #[test]
361    fn a_reply_loaded_across_a_change_is_kept_unless_the_change_concerns_it() {
362        let mut kept = Kept::default();
363        let before = kept.heard;
364        kept.hear(changed("customer"));
365
366        kept.keep(key("/orders"), entry(&["order"], MAX_TIME), before, true);
367        assert!(kept.hit(&key("/orders"), true).is_ok());
368
369        kept.hear(changed("order"));
370        kept.keep(
371            key("/orders?page=2"),
372            entry(&["order"], MAX_TIME),
373            before,
374            true,
375        );
376        assert!(kept.hit(&key("/orders?page=2"), true).is_err());
377
378        kept.hear(Announced::MaybeMissed);
379        kept.keep(
380            key("/customers"),
381            entry(&["customer"], MAX_TIME),
382            before + 2,
383            true,
384        );
385        assert!(kept.hit(&key("/customers"), true).is_err());
386    }
387
388    #[test]
389    fn a_change_forgets_only_the_replies_that_read_its_table() {
390        let mut kept = Kept::default();
391        kept.keep(
392            key("/orders"),
393            entry(&["order", "customer"], MAX_TIME),
394            0,
395            true,
396        );
397        kept.keep(key("/products"), entry(&["product"], MAX_TIME), 0, true);
398
399        kept.hear(changed("customer"));
400
401        assert!(kept.hit(&key("/orders"), true).is_err());
402        assert!(kept.hit(&key("/products"), true).is_ok());
403        assert!(!kept.by_topic.contains_key(&Topic::named("shop", "order")));
404    }
405
406    #[test]
407    fn a_full_cache_forgets_the_entries_closest_to_expiring() {
408        let mut kept = Kept::default();
409        let page = |index: usize| key(&format!("/orders?page={index}"));
410        for index in 0..MAX_ENTRIES {
411            let time = Duration::from_millis(u64::try_from(index).unwrap() + 1);
412            kept.keep(page(index), entry(&["order"], time), 0, true);
413        }
414
415        kept.keep(page(MAX_ENTRIES), entry(&["order"], MAX_TIME), 0, true);
416
417        assert!(kept.hit(&page(MAX_ENTRIES), true).is_ok());
418        assert!(kept.hit(&page(0), true).is_err());
419        assert!(kept.hit(&page(MAX_ENTRIES - 1), true).is_ok());
420    }
421}
sofie_http/screen_cache.rs

sofie_http/
screen_cache.rs
