cronoq_server/
info.rs

1//! Bounded cache of task details for `Get` and for result messages.
2//!
3//! The scheduler engine only answers "what is the status of task X"; this cache adds the
4//! pool, key, kind, attempt counts, timestamps and the last error message. It is filled when
5//! tasks are submitted through this server and from the scheduler's event stream, and evicts
6//! the oldest records beyond its capacity.
7
8use std::collections::VecDeque;
9use std::sync::{Arc, Mutex};
10
11use cronoq::{Priority, StatusEvent, TaskStatus};
12use rustc_hash::FxHashMap;
13
14/// What is known about one task.
15#[derive(Clone, Debug, Default)]
16pub struct TaskMeta {
17    /// Pool name.
18    pub pool: Arc<str>,
19    /// Unique key, if any.
20    pub key: Option<Arc<str>>,
21    /// Kind, if submitted through this server.
22    pub kind: Option<Arc<str>>,
23    /// Attempts allowed (0 if unknown).
24    pub max_attempts: u32,
25    /// Last observed status.
26    pub status: Option<TaskStatus>,
27    /// Last observed attempt number.
28    pub attempt: u32,
29    /// Submission time (Unix ns).
30    pub created_ns: Option<i64>,
31    /// Start of the latest attempt (Unix ns).
32    pub started_ns: Option<i64>,
33    /// Terminal time (Unix ns).
34    pub finished_ns: Option<i64>,
35    /// Last error or status message.
36    pub message: Option<Arc<str>>,
37    /// Tenant, if submitted through this server.
38    pub tenant: Option<Arc<str>>,
39    /// Priority, if submitted through this server.
40    pub priority: Option<Priority>,
41}
42
43#[derive(Default)]
44struct Inner {
45    map: FxHashMap<u64, TaskMeta>,
46    // Keys come from callers: std's seeded SipHash, not FxHash, so they cannot be made to collide.
47    by_key: std::collections::HashMap<Arc<str>, u64>,
48    order: VecDeque<u64>,
49}
50
51/// Bounded map of task id to [`TaskMeta`].
52pub struct TaskInfoCache {
53    inner: Mutex<Inner>,
54    cap: usize,
55}
56
57impl TaskInfoCache {
58    /// Cache holding at most `cap` tasks.
59    pub fn new(cap: usize) -> Arc<Self> {
60        Arc::new(TaskInfoCache {
61            inner: Mutex::new(Inner::default()),
62            cap,
63        })
64    }
65
66    fn evict(st: &mut Inner, cap: usize) {
67        while st.map.len() > cap {
68            let Some(id) = st.order.pop_front() else {
69                break;
70            };
71            if let Some(m) = st.map.remove(&id)
72                && let Some(k) = m.key
73                && st.by_key.get(&k) == Some(&id)
74            {
75                st.by_key.remove(&k);
76            }
77        }
78    }
79
80    fn entry<'a>(st: &'a mut Inner, id: u64, pool: &Arc<str>) -> &'a mut TaskMeta {
81        if !st.map.contains_key(&id) {
82            st.order.push_back(id);
83        }
84        st.map.entry(id).or_insert_with(|| TaskMeta {
85            pool: pool.clone(),
86            ..Default::default()
87        })
88    }
89
90    /// Record a submission (not for duplicates: the original details stay).
91    pub fn on_submit(&self, id: u64, meta: TaskMeta) {
92        if self.cap == 0 {
93            return;
94        }
95        let mut st = self.inner.lock().unwrap();
96        if let Some(k) = &meta.key {
97            st.by_key.insert(k.clone(), id);
98        }
99        let pool = meta.pool.clone();
100        let e = Self::entry(&mut st, id, &pool);
101        let (status, attempt, started, finished, message) = (
102            e.status,
103            e.attempt,
104            e.started_ns,
105            e.finished_ns,
106            e.message.clone(),
107        );
108        *e = TaskMeta {
109            status,
110            attempt,
111            started_ns: started,
112            finished_ns: finished,
113            message,
114            ..meta
115        };
116        Self::evict(&mut st, self.cap);
117    }
118
119    /// Record a status event of a task in `pool`, observed at `at_ns` (Unix ns).
120    pub fn on_event(&self, ev: &StatusEvent, pool: &Arc<str>, at_ns: i64) {
121        if self.cap == 0 {
122            return;
123        }
124        let id = ev.id.0;
125        let mut st = self.inner.lock().unwrap();
126        if let Some(k) = &ev.key {
127            st.by_key.entry(k.clone()).or_insert(id);
128        }
129        let e = Self::entry(&mut st, id, pool);
130        if e.key.is_none() {
131            e.key = ev.key.clone();
132        }
133        e.status = Some(ev.status);
134        e.attempt = e.attempt.max(ev.attempt);
135        if ev.status == TaskStatus::Running {
136            e.started_ns = Some(at_ns);
137        }
138        if ev.status.is_terminal() {
139            e.finished_ns = Some(at_ns);
140        }
141        if let Some(m) = &ev.message {
142            e.message = Some(m.clone());
143        }
144        Self::evict(&mut st, self.cap);
145    }
146
147    /// Details of a task.
148    pub fn get(&self, id: u64) -> Option<TaskMeta> {
149        self.inner.lock().unwrap().map.get(&id).cloned()
150    }
151
152    /// Finished tasks in id order: those in one of `pools` (any if empty) with one of
153    /// `statuses` (any finished status if empty) and ids above `after`, at most `limit`.
154    pub fn list_finished(
155        &self,
156        pools: &[&str],
157        statuses: &[TaskStatus],
158        after: u64,
159        limit: usize,
160    ) -> Vec<(u64, TaskMeta)> {
161        let st = self.inner.lock().unwrap();
162        let mut ids: Vec<u64> = st
163            .map
164            .iter()
165            .filter(|(id, m)| {
166                **id > after
167                    && m.status.is_some_and(|s| {
168                        s.is_terminal() && (statuses.is_empty() || statuses.contains(&s))
169                    })
170                    && (pools.is_empty() || pools.contains(&&*m.pool))
171            })
172            .map(|(id, _)| *id)
173            .collect();
174        ids.sort_unstable();
175        ids.truncate(limit);
176        ids.into_iter()
177            .map(|id| (id, st.map[&id].clone()))
178            .collect()
179    }
180
181    /// Latest task id seen for `key`.
182    pub fn id_for_key(&self, key: &str) -> Option<u64> {
183        self.inner.lock().unwrap().by_key.get(key).copied()
184    }
185}
186
187#[cfg(test)]
188mod tests {
189    use super::*;
190
191    #[test]
192    fn records_and_evicts() {
193        let c = TaskInfoCache::new(2);
194        let p: Arc<str> = "p".into();
195        c.on_submit(
196            1,
197            TaskMeta {
198                pool: p.clone(),
199                key: Some("k1".into()),
200                max_attempts: 3,
201                ..Default::default()
202            },
203        );
204        let ev = |status, message: Option<&str>| StatusEvent {
205            id: cronoq::TaskId(1),
206            pool: cronoq::PoolId(0),
207            key: None,
208            status,
209            attempt: 1,
210            at: cronoq::Time(0),
211            message: message.map(Arc::from),
212        };
213        c.on_event(&ev(TaskStatus::Running, None), &p, 10);
214        c.on_event(&ev(TaskStatus::Failed, Some("boom")), &p, 20);
215        let m = c.get(1).unwrap();
216        assert_eq!(
217            (m.status, m.attempt, m.started_ns, m.finished_ns),
218            (Some(TaskStatus::Failed), 1, Some(10), Some(20))
219        );
220        assert_eq!(m.message.as_deref(), Some("boom"));
221        assert_eq!(c.id_for_key("k1"), Some(1));
222        c.on_submit(
223            2,
224            TaskMeta {
225                pool: p.clone(),
226                ..Default::default()
227            },
228        );
229        c.on_submit(
230            3,
231            TaskMeta {
232                pool: p.clone(),
233                ..Default::default()
234            },
235        );
236        assert!(
237            c.get(1).is_none() && c.id_for_key("k1").is_none(),
238            "oldest evicted"
239        );
240    }
241
242    #[test]
243    fn lists_finished_tasks_by_pool_status_and_page() {
244        let c = TaskInfoCache::new(10);
245        let ev = |id, status| StatusEvent {
246            id: cronoq::TaskId(id),
247            pool: cronoq::PoolId(0),
248            key: None,
249            status,
250            attempt: 1,
251            at: cronoq::Time(0),
252            message: None,
253        };
254        let (a, b): (Arc<str>, Arc<str>) = ("a".into(), "b".into());
255        c.on_event(&ev(4, TaskStatus::Succeeded), &a, 1);
256        c.on_event(&ev(1, TaskStatus::Failed), &a, 1);
257        c.on_event(&ev(2, TaskStatus::Running), &a, 1);
258        c.on_event(&ev(3, TaskStatus::Succeeded), &b, 1);
259        let ids = |v: Vec<(u64, TaskMeta)>| v.into_iter().map(|(id, _)| id).collect::<Vec<_>>();
260        assert_eq!(ids(c.list_finished(&[], &[], 0, 10)), [1, 3, 4]);
261        assert_eq!(ids(c.list_finished(&["a"], &[], 0, 10)), [1, 4]);
262        assert_eq!(
263            ids(c.list_finished(&[], &[TaskStatus::Succeeded], 0, 10)),
264            [3, 4]
265        );
266        assert_eq!(ids(c.list_finished(&[], &[], 1, 1)), [3]);
267    }
268}
cronoq_server/info.rs

cronoq_server/
info.rs
