forge_workspace/store/
task_chases.rs

1//! Durable chase log: the newest firing of each rung per subject.
2//!
3//! This is what makes "each rung fires once per row per threshold" a fact
4//! rather than a hope: the sweep checks here before it speaks. **Only the
5//! newest firing under a subject's key is kept**, because every reader asks
6//! "did it fire at or after T?", which the newest row answers alone - and a
7//! held state re-appends once per cadence window, so without that the table
8//! would grow with time rather than with what the board is tracking (one
9//! permanently-over row is 288 rows a day at the 5-minute cadence).
10
11use anyhow::Context;
12use forge_primitives::tasks::{ChaseRung, TaskChase, TaskId};
13use redb::{ReadableTable, TableDefinition};
14
15use super::Db;
16
17const TASK_CHASES: TableDefinition<&str, &[u8]> = TableDefinition::new("task_chases");
18
19/// The key prefix every firing of one (project, row, rung, subject) shares:
20/// keys sort by it first, so its rows are one contiguous run and its newest
21/// firing is its last.
22fn bucket_key(
23    project: &str,
24    task_id: Option<&TaskId>,
25    subject: Option<&str>,
26    rung: ChaseRung,
27) -> String {
28    let task = task_id.map_or("-", TaskId::as_str);
29    let subject = subject.unwrap_or("-");
30    format!("{project}\u{0}{task}\u{0}{subject}\u{0}{rung:?}\u{0}")
31}
32
33/// The row key for one fired chase: its bucket, then the instant (padded so
34/// a lexicographic order is the time order).
35fn row_key(chase: &TaskChase) -> String {
36    let nanos =
37        chase.at.duration_since(std::time::SystemTime::UNIX_EPOCH).map_or(0, |d| d.as_nanos());
38    let bucket = bucket_key(
39        &chase.project_name,
40        chase.task_id.as_ref(),
41        chase.subject.as_deref(),
42        chase.rung,
43    );
44    format!("{bucket}{nanos:040}")
45}
46
47/// Append `chases`, each replacing any older firing of its own subject.
48///
49/// The newest firing is the one every reader asks for, so the older ones are
50/// dead weight the moment it exists - and a held state appends one every
51/// cadence window, so keeping them is what would make this table grow with
52/// time on the board rather than with the board's own rows.
53pub fn append(db: &Db, chases: &[TaskChase]) -> anyhow::Result<()> {
54    let txn = db.database().begin_write()?;
55    {
56        let mut table = txn.open_table(TASK_CHASES)?;
57        for chase in chases {
58            let key = row_key(chase);
59            let bucket = bucket_key(
60                &chase.project_name,
61                chase.task_id.as_ref(),
62                chase.subject.as_deref(),
63                chase.rung,
64            );
65            // Collected before anything is removed: the range borrows the
66            // table, and a key cannot be removed under a live iterator.
67            let mut stale = Vec::new();
68            for entry in table.range(bucket.as_str()..key.as_str())? {
69                let (held, _) = entry?;
70                stale.push(held.value().to_owned());
71            }
72            for old in stale {
73                table.remove(old.as_str())?;
74            }
75            let value = serde_json::to_vec(chase).context("serialize chase")?;
76            table.insert(key.as_str(), value.as_slice())?;
77        }
78    }
79    txn.commit()?;
80    Ok(())
81}
82
83/// Forget every firing of one subject: the rung's state cleared, so the
84/// slate resets durably rather than in a process's memory alone.
85pub fn forget(
86    db: &Db,
87    project: &str,
88    rung: ChaseRung,
89    task_id: Option<&TaskId>,
90    subject: Option<&str>,
91) -> anyhow::Result<()> {
92    let bucket = bucket_key(project, task_id, subject, rung);
93    let txn = db.database().begin_write()?;
94    {
95        let mut table = txn.open_table(TASK_CHASES)?;
96        let mut doomed = Vec::new();
97        for entry in table.range(bucket.as_str()..)? {
98            let (key, _) = entry?;
99            if !key.value().starts_with(bucket.as_str()) {
100                break;
101            }
102            doomed.push(key.value().to_owned());
103        }
104        for key in doomed {
105            table.remove(key.as_str())?;
106        }
107    }
108    txn.commit()?;
109    Ok(())
110}
111
112/// `project`'s fired chases, ordered by row and then by time.
113pub fn list_for_project(db: &Db, project: &str) -> anyhow::Result<Vec<TaskChase>> {
114    let txn = db.database().begin_read()?;
115    let table = match txn.open_table(TASK_CHASES) {
116        Ok(t) => t,
117        Err(redb::TableError::TableDoesNotExist(_)) => return Ok(Vec::new()),
118        Err(e) => return Err(e.into()),
119    };
120    let prefix = format!("{project}\u{0}");
121    let mut out = Vec::new();
122    // **One range, not the whole table.** Keys sort by project first, so a
123    // project's rows are one contiguous run: the walk stops at its end
124    // instead of reading every other project's rows once per project, per
125    // sweep.
126    for entry in table.range(prefix.as_str()..)? {
127        let (key, value) = entry?;
128        if !key.value().starts_with(&prefix) {
129            break;
130        }
131        match serde_json::from_slice::<TaskChase>(value.value()) {
132            Ok(chase) => out.push(chase),
133            Err(err) => tracing::warn!(
134                target: "forge_workspace::store::task_chases",
135                row = %key.value(),
136                error = %err,
137                "skipping chase row that failed to decode",
138            ),
139        }
140    }
141    Ok(out)
142}
143
144#[cfg(test)]
145mod tests {
146    use std::path::Path;
147
148    use super::*;
149    use forge_primitives::tasks::{ChaseRung, TaskId};
150    use tempfile::tempdir;
151
152    fn open_db(dir: &Path) -> Db {
153        Db::open(&dir.join("db.redb")).expect("open db")
154    }
155
156    fn chase(task: Option<&str>, project: &str, rung: ChaseRung, at_secs: u64) -> TaskChase {
157        TaskChase {
158            task_id: task.map(TaskId::from),
159            subject: None,
160            project_name: project.to_owned(),
161            rung,
162            at: std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(at_secs),
163        }
164    }
165
166    /// A subject's older firings are dead weight once a newer one exists:
167    /// a held state appends one every cadence window, and the newest alone
168    /// answers a "fired since" question.
169    #[test]
170    fn only_the_newest_firing_of_a_subject_is_kept() {
171        let dir = tempdir().expect("tempdir");
172        let db = open_db(dir.path());
173        let mut older = chase(None, "forge", ChaseRung::Unaccounted, 10);
174        older.subject = Some("w-1".to_owned());
175        let mut newer = chase(None, "forge", ChaseRung::Unaccounted, 20);
176        newer.subject = Some("w-1".to_owned());
177        append(&db, &[older, newer]).expect("append");
178
179        let forge = list_for_project(&db, "forge").expect("list");
180        assert_eq!(forge.len(), 1, "one firing stands per subject: {forge:?}");
181        assert_eq!(
182            forge[0].at,
183            std::time::SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(20),
184            "and it is the newest",
185        );
186    }
187
188    /// A project whose name extends another's is still read apart: the walk
189    /// is scoped by the whole name and its separator, not by a prefix of it.
190    #[test]
191    fn a_projects_rows_do_not_leak_into_an_extension_of_its_name() {
192        let dir = tempdir().expect("tempdir");
193        let db = open_db(dir.path());
194        append(
195            &db,
196            &[
197                chase(Some("t-1"), "forge", ChaseRung::EstimateNudge, 10),
198                chase(Some("t-2"), "forge2", ChaseRung::EstimateNudge, 11),
199            ],
200        )
201        .expect("append");
202
203        let forge = list_for_project(&db, "forge").expect("list");
204        assert_eq!(forge.len(), 1, "only forge's own row: {forge:?}");
205        assert_eq!(forge[0].task_id.as_ref().map(TaskId::as_str), Some("t-1"));
206    }
207
208    /// Clearing a state forgets its firings, durably, and only its own: a
209    /// recurrence after a restart reads no record and is named at once.
210    #[test]
211    fn forgetting_a_subject_drops_only_its_firings() {
212        let dir = tempdir().expect("tempdir");
213        let db = open_db(dir.path());
214        let mut cleared = chase(None, "forge", ChaseRung::Unaccounted, 10);
215        cleared.subject = Some("w-1".to_owned());
216        let mut held = chase(None, "forge", ChaseRung::Unaccounted, 10);
217        held.subject = Some("w-2".to_owned());
218        append(&db, &[cleared, held]).expect("append");
219
220        forget(&db, "forge", ChaseRung::Unaccounted, None, Some("w-1")).expect("forget");
221        let forge = list_for_project(&db, "forge").expect("list");
222        assert_eq!(forge.len(), 1, "the cleared worker's firing is gone: {forge:?}");
223        assert_eq!(forge[0].subject.as_deref(), Some("w-2"), "the held one stays");
224    }
225
226    #[test]
227    fn the_chase_log_answers_whether_a_rung_already_fired() {
228        let dir = tempdir().expect("tempdir");
229        let db = open_db(dir.path());
230        append(
231            &db,
232            &[
233                chase(Some("t-1"), "forge", ChaseRung::EstimateNudge, 10),
234                chase(None, "forge", ChaseRung::Unaccounted, 11),
235                chase(Some("t-9"), "other", ChaseRung::EstimateNudge, 12),
236            ],
237        )
238        .expect("append");
239
240        let forge = list_for_project(&db, "forge").expect("list");
241        assert_eq!(forge.len(), 2, "only forge's chases: {forge:?}");
242        assert!(
243            forge.iter().any(|c| {
244                c.task_id.as_ref() == Some(&TaskId::from("t-1"))
245                    && c.rung == ChaseRung::EstimateNudge
246            }),
247            "the fired rung is found for its row",
248        );
249        assert!(
250            forge.iter().any(|c| c.task_id.is_none() && c.rung == ChaseRung::Unaccounted),
251            "a seat-level rung records with no task",
252        );
253    }
254
255    /// Two workers named by one sweep share an instant, so the label is
256    /// part of the key: without it the second write would overwrite the
257    /// first and one worker's naming would vanish from the log.
258    #[test]
259    fn two_seat_level_chases_at_one_instant_both_survive() {
260        let dir = tempdir().expect("tempdir");
261        let db = open_db(dir.path());
262        let mut w1 = chase(None, "forge", ChaseRung::Unaccounted, 10);
263        w1.subject = Some("w-1".to_owned());
264        let mut w2 = chase(None, "forge", ChaseRung::Unaccounted, 10);
265        w2.subject = Some("w-2".to_owned());
266        append(&db, &[w1, w2]).expect("append");
267
268        let forge = list_for_project(&db, "forge").expect("list");
269        assert_eq!(forge.len(), 2, "both namings stand: {forge:?}");
270    }
271
272    #[test]
273    fn an_absent_chase_table_reads_as_empty() {
274        let dir = tempdir().expect("tempdir");
275        let db = open_db(dir.path());
276        assert!(list_for_project(&db, "forge").expect("list").is_empty());
277    }
278}
forge_workspace/store/task_chases.rs

forge_workspace/store/
task_chases.rs
