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

sofie_http/
screen_cache.rs
