salvor_runtime/
wake.rs

1//! Which sleeping runs are due: the one question both wakers ask, answered
2//! once.
3//!
4//! A run parked on a durable timer is passive data. Nothing in the process
5//! holds it, and nothing fires when its instant arrives; the only way it moves
6//! again is for something to re-drive it, at which point
7//! [`RunCtx::await_wake`](crate::RunCtx::await_wake) reads the injected clock
8//! and either records the wake or reports the run still asleep. So a waker
9//! needs exactly two things: a list of runs whose deadline has passed, and the
10//! ordinary resume path. This module is the first; the second already exists.
11//!
12//! # Why this lives in the runtime crate
13//!
14//! Both wakers are elsewhere (`salvor wake` in the CLI, the `serve` sweeper in
15//! the server), and neither may own the answer, or the two would drift on what
16//! "due" means. The question is about the store and the fold, not about a
17//! terminal or an HTTP route, and this is the lowest crate that sees both a
18//! [`EventStore`] and the sleep primitives that put a run into this state. The
19//! pure-renderer crate (`salvor-cli-core`) cannot hold it: it compiles for
20//! `wasm32-unknown-unknown` and names no store.
21//!
22//! # It drives nothing
23//!
24//! [`due_runs`] reads. It appends no event, builds no agent, and calls no
25//! clock of its own: the caller passes the instant to measure against, so a
26//! test asks "what is due at this moment" without moving a real clock, and
27//! both wakers measure against the same clock their drive will use.
28
29use salvor_core::{RunId, RunStatus, derive_state};
30use salvor_store::{EventStore, StoreError};
31use time::OffsetDateTime;
32
33/// A run whose durable timer has come due.
34///
35/// The run id is what a caller re-drives; `wake_at` is the deadline the log
36/// recorded, carried so a report can say how overdue the run is without
37/// folding it again.
38#[derive(Debug, Clone, Copy, PartialEq, Eq)]
39pub struct DueRun {
40    /// The sleeping run.
41    pub run_id: RunId,
42    /// The instant its recorded `SleepStarted` said it may continue at.
43    pub wake_at: OffsetDateTime,
44}
45
46/// Every run in `store` whose status folds to
47/// [`RunStatus::Sleeping`](salvor_core::RunStatus::Sleeping) with a `wake_at`
48/// at or before `now`, oldest deadline first.
49///
50/// The comparison is `wake_at <= now`, the same inclusive edge
51/// [`RunCtx::await_wake`](crate::RunCtx::await_wake) applies, so a run this
52/// function reports as due is one that drive will actually wake rather than
53/// send straight back to sleep.
54///
55/// # Why every log is read
56///
57/// Status is a replay-time projection, not a stored column, so there is
58/// nothing to select on until each log has been folded; this walks
59/// [`EventStore::list_runs`] and folds each one, exactly as `salvor list`
60/// does. That makes the cost linear in the store, which is the honest cost of
61/// keeping status out of the schema, and it is why a caller sweeps on an
62/// interval rather than in a tight loop.
63///
64/// A run whose log fails to read (a broken chain) is skipped rather than
65/// failing the whole selection: one damaged run must not stop every other due
66/// run from waking, and the damage surfaces the moment anything asks for that
67/// log by name.
68///
69/// # Errors
70///
71/// [`StoreError`] when the run listing itself cannot be read.
72pub async fn due_runs(
73    store: &dyn EventStore,
74    now: OffsetDateTime,
75) -> Result<Vec<DueRun>, StoreError> {
76    let mut due = Vec::new();
77    for summary in store.list_runs().await? {
78        let log = match store.read_log(summary.run_id).await {
79            Ok(log) => log,
80            Err(err) => {
81                tracing::warn!(
82                    run_id = %summary.run_id.as_uuid(),
83                    error = %err,
84                    "skipping a run whose log will not read while selecting due timers"
85                );
86                continue;
87            }
88        };
89        if let RunStatus::Sleeping { wake_at } = derive_state(&log).status
90            && wake_at <= now
91        {
92            due.push(DueRun {
93                run_id: summary.run_id,
94                wake_at,
95            });
96        }
97    }
98    // Oldest deadline first, so the most overdue run is driven first and a
99    // sweep that runs out of time leaves the least-overdue behind. The run id
100    // breaks ties, so the order is total and a report is reproducible.
101    due.sort_by(|a, b| {
102        a.wake_at
103            .cmp(&b.wake_at)
104            .then_with(|| a.run_id.as_uuid().cmp(&b.run_id.as_uuid()))
105    });
106    Ok(due)
107}
salvor_runtime/wake.rs

salvor_runtime/
wake.rs
