forge_workspace/store/
tasks.rs

1//! Durable task persistence on the redb `tasks` table.
2//!
3//! The whole [`Task`] record is stored as serde-json keyed by its
4//! project and id - no field schema on disk, so a type change needs no
5//! table migration. Machine-local, like the crons beside it.
6
7use anyhow::Context;
8use forge_primitives::tasks::{Estimate, LinkKind, Task, TaskId, TaskLink, TaskStatus, Waiting};
9use redb::{ReadableTable, TableDefinition};
10
11use super::Db;
12
13const TASKS: TableDefinition<&str, &[u8]> = TableDefinition::new("tasks");
14
15/// The stored shape: [`Task`] plus the two spellings earlier versions
16/// wrote - `artifact` as a bare string, and `blocked` as a status. Reads
17/// go through here so a legacy row loads as the record it meant rather
18/// than failing the whole set.
19#[derive(serde::Deserialize)]
20struct TaskRecord {
21    id: forge_primitives::tasks::TaskId,
22    project_name: String,
23    subject: String,
24    #[serde(default)]
25    active_form: Option<String>,
26    #[serde(default)]
27    detail: Option<String>,
28    status: RecordStatus,
29    #[serde(default)]
30    owner: Option<forge_primitives::SessionSlot>,
31    #[serde(default)]
32    parent: Option<forge_primitives::tasks::TaskId>,
33    #[serde(default)]
34    waiting_on: Option<Waiting>,
35    #[serde(default)]
36    estimate: Option<RecordEstimate>,
37    #[serde(default)]
38    rank: Option<i64>,
39    #[serde(default)]
40    verify: Option<forge_primitives::tasks::Verify>,
41    #[serde(default)]
42    attested: Option<forge_primitives::tasks::Attestation>,
43    #[serde(default)]
44    reserved: Option<forge_primitives::tasks::ReservedBy>,
45    #[serde(default)]
46    links: Vec<TaskLink>,
47    #[serde(default)]
48    attempt: u32,
49    #[serde(default)]
50    archived_at: Option<std::time::SystemTime>,
51    created_at: std::time::SystemTime,
52    updated_at: std::time::SystemTime,
53    #[serde(default)]
54    artifact: Option<String>,
55}
56
57#[derive(Clone, Copy, serde::Deserialize)]
58#[serde(rename_all = "snake_case")]
59enum RecordStatus {
60    Pending,
61    InProgress,
62    Waiting,
63    Completed,
64    Failed,
65    Canceled,
66    /// The old spelling: a wait whose reason was never stated.
67    Blocked,
68}
69
70#[derive(serde::Deserialize)]
71#[serde(untagged)]
72enum RecordEstimate {
73    /// The old shape: free words, parsed on read.
74    Words(String),
75    Parsed(Estimate),
76}
77
78impl TaskRecord {
79    fn into_task(self) -> Task {
80        let blocked = matches!(self.status, RecordStatus::Blocked);
81        let status = match self.status {
82            RecordStatus::Pending => TaskStatus::Pending,
83            RecordStatus::InProgress => TaskStatus::InProgress,
84            RecordStatus::Waiting | RecordStatus::Blocked => TaskStatus::Waiting,
85            RecordStatus::Completed => TaskStatus::Completed,
86            RecordStatus::Failed => TaskStatus::Failed,
87            RecordStatus::Canceled => TaskStatus::Canceled,
88        };
89        let mut links = self.links;
90        if let Some(artifact) = &self.artifact
91            && !links.iter().any(|l| l.target == *artifact)
92        {
93            links.push(TaskLink {
94                kind: LinkKind::for_target(artifact),
95                label: None,
96                target: artifact.clone(),
97                state: None,
98                added_at: self.created_at,
99            });
100        }
101        Task {
102            id: self.id,
103            project_name: self.project_name,
104            subject: self.subject,
105            active_form: self.active_form,
106            detail: self.detail,
107            status,
108            owner: self.owner,
109            parent: self.parent,
110            waiting_on: if blocked {
111                // The migration's unstated wait: the board draws it as a
112                // miss, which is what an old blocked row was.
113                Some(Waiting {
114                    kind: None,
115                    detail: None,
116                    on: None,
117                    verification: false,
118                    parked: false,
119                    since: None,
120                })
121            } else {
122                self.waiting_on
123            },
124            estimate: self.estimate.map(|estimate| match estimate {
125                // Words the grammar cannot read keep their text and carry
126                // no seconds, which the board reads as no estimate rather
127                // than inventing one.
128                RecordEstimate::Words(words) => {
129                    Estimate::parse(&words).unwrap_or(Estimate { secs: 0, words })
130                }
131                RecordEstimate::Parsed(parsed) => parsed,
132            }),
133            rank: self.rank,
134            verify: self.verify,
135            attested: self.attested,
136            reserved: self.reserved,
137            links,
138            attempt: self.attempt,
139            archived_at: self.archived_at,
140            created_at: self.created_at,
141            updated_at: self.updated_at,
142        }
143    }
144}
145
146/// The row key for a task: its project as well as its id, separated by a
147/// NUL a project name cannot carry.
148fn row_key(task: &Task) -> String {
149    row_key_for(&task.project_name, &task.id)
150}
151
152/// The row key for a project and id, the pair [`row_key`] joins.
153fn row_key_for(project_name: &str, id: &TaskId) -> String {
154    format!("{project_name}\u{0}{}", id.as_str())
155}
156
157/// Every persisted task.
158pub fn list(db: &Db) -> anyhow::Result<Vec<Task>> {
159    let txn = db.database().begin_read()?;
160    let table = match txn.open_table(TASKS) {
161        Ok(t) => t,
162        // A fresh database has no table until the first write; an absent
163        // table is an empty list, not an error.
164        Err(redb::TableError::TableDoesNotExist(_)) => return Ok(Vec::new()),
165        Err(e) => return Err(e.into()),
166    };
167    let mut out = Vec::new();
168    for entry in table.iter()? {
169        let (key, value) = entry?;
170        match serde_json::from_slice::<TaskRecord>(value.value()) {
171            Ok(record) => out.push(record.into_task()),
172            // One undecodable record (schema drift, a corrupt blob) must not
173            // wipe the rest of the durable set - skip it and warn.
174            Err(err) => tracing::warn!(
175                target: "forge_workspace::store::tasks",
176                row = %key.value(),
177                error = %err,
178                "skipping task record that failed to decode",
179            ),
180        }
181    }
182    Ok(out)
183}
184
185/// Persist one task, replacing any prior record for its project and id.
186/// One write txn per row, so a write can only touch the row it names.
187pub fn upsert(db: &Db, task: &Task) -> anyhow::Result<()> {
188    let value = serde_json::to_vec(task).context("serialize task")?;
189    let txn = db.database().begin_write()?;
190    {
191        let mut table = txn.open_table(TASKS)?;
192        table.insert(row_key(task).as_str(), value.as_slice())?;
193    }
194    txn.commit()?;
195    Ok(())
196}
197
198/// Delete a task by project and id. Returns whether a record existed.
199pub fn delete(db: &Db, project_name: &str, id: &TaskId) -> anyhow::Result<bool> {
200    let key = row_key_for(project_name, id);
201    let txn = db.database().begin_write()?;
202    let existed = {
203        let mut table = txn.open_table(TASKS)?;
204        table.remove(key.as_str())?.is_some()
205    };
206    txn.commit()?;
207    Ok(existed)
208}
209
210#[cfg(test)]
211mod tests {
212    use std::path::Path;
213
214    use super::*;
215    use forge_primitives::tasks::{TaskId, TaskStatus};
216    use tempfile::tempdir;
217
218    fn open_db(dir: &Path) -> Db {
219        Db::open(&dir.join("db.redb")).expect("open db")
220    }
221
222    fn sample_task(id: &str, project: &str) -> Task {
223        Task {
224            id: TaskId::from(id),
225            project_name: project.to_owned(),
226            subject: format!("subject {id}"),
227            active_form: None,
228            detail: None,
229            status: TaskStatus::Pending,
230            owner: None,
231            parent: None,
232            waiting_on: None,
233            estimate: None,
234            rank: None,
235            verify: None,
236            attested: None,
237            reserved: None,
238            links: Vec::new(),
239            attempt: 0,
240            archived_at: None,
241            created_at: std::time::SystemTime::UNIX_EPOCH,
242            updated_at: std::time::SystemTime::UNIX_EPOCH,
243        }
244    }
245
246    #[test]
247    fn tasks_are_scoped_by_project() {
248        let dir = tempdir().expect("tempdir");
249        let db = open_db(dir.path());
250        let forge_task = sample_task("t-1", "forge");
251        let other_task = sample_task("t-2", "other");
252        upsert(&db, &forge_task).expect("write forge's");
253        upsert(&db, &other_task).expect("write the other project's");
254        let read = list(&db).expect("list");
255        assert_eq!(read.len(), 2, "both projects' tasks are stored");
256        assert!(
257            read.iter().any(|t| t.project_name == "forge" && t.id == TaskId::from("t-1")),
258            "the forge task is stored under its own project",
259        );
260    }
261
262    #[test]
263    fn the_same_id_in_two_projects_stores_two_rows() {
264        let dir = tempdir().expect("tempdir");
265        let db = open_db(dir.path());
266        upsert(&db, &sample_task("t-1", "forge")).expect("write forge's");
267        upsert(&db, &sample_task("t-1", "other")).expect("write the other project's");
268        let read = list(&db).expect("list");
269        assert_eq!(read.len(), 2, "one project's row cannot overwrite another's on a shared id");
270    }
271
272    /// A delete removes the row it names and nothing else: the project is
273    /// half the key, so the same id under another project is a different
274    /// row, and the named one is gone from the store rather than only from
275    /// memory - a row left in redb comes back on the next boot.
276    #[test]
277    fn a_delete_drops_only_the_named_row() {
278        let dir = tempdir().expect("tempdir");
279        let db = open_db(dir.path());
280        upsert(&db, &sample_task("t-1", "forge")).expect("seed t-1");
281        upsert(&db, &sample_task("t-2", "forge")).expect("seed t-2");
282        upsert(&db, &sample_task("t-1", "other")).expect("seed the other project's t-1");
283
284        assert!(delete(&db, "forge", &TaskId::from("t-1")).expect("delete"));
285        assert!(
286            !delete(&db, "forge", &TaskId::from("t-1")).expect("delete again"),
287            "a row that is not there deletes to false",
288        );
289        let read = list(&db).expect("list");
290        assert_eq!(read.len(), 2, "one row left per project: {read:?}");
291        assert!(
292            read.iter().any(|t| t.project_name == "forge" && t.id == TaskId::from("t-2")),
293            "the deleted row is gone from the store, not just from memory: {read:?}",
294        );
295        assert!(
296            read.iter().any(|t| t.project_name == "other" && t.id == TaskId::from("t-1")),
297            "and a same-named row under another project is its own row: {read:?}",
298        );
299    }
300
301    #[test]
302    fn an_absent_table_reads_as_empty_not_an_error() {
303        let dir = tempdir().expect("tempdir");
304        let db = open_db(dir.path());
305        assert!(list(&db).expect("list").is_empty(), "a fresh database has no tasks");
306    }
307
308    /// Write a raw JSON row, the way an older forge wrote it.
309    fn write_raw(db: &Db, key: &str, json: &str) {
310        let txn = db.database().begin_write().expect("write txn");
311        {
312            let mut table = txn.open_table(TASKS).expect("open");
313            table.insert(key, json.as_bytes()).expect("insert");
314        }
315        txn.commit().expect("commit");
316    }
317
318    /// A row an earlier forge wrote: `artifact` as a bare string, the
319    /// estimate as free words.
320    #[test]
321    fn a_legacy_artifact_row_reads_as_a_link() {
322        let dir = tempdir().expect("tempdir");
323        let db = open_db(dir.path());
324        write_raw(
325            &db,
326            "forge\u{0}t-1",
327            r#"{
328                "id": "t-1", "project_name": "forge", "subject": "old row",
329                "active_form": null, "detail": null, "status": "in_progress",
330                "owner": null, "parent": null,
331                "artifact": "https://github.com/busytools/forge/pull/1889",
332                "estimate": "1d",
333                "created_at": {"secs_since_epoch": 0, "nanos_since_epoch": 0},
334                "updated_at": {"secs_since_epoch": 0, "nanos_since_epoch": 0}
335            }"#,
336        );
337        let read = list(&db).expect("list");
338        assert_eq!(read.len(), 1, "the legacy row loads: {read:?}");
339        let task = &read[0];
340        assert_eq!(task.links.len(), 1, "the artifact folded into a link");
341        assert_eq!(task.links[0].kind, LinkKind::Pr, "classified by its shape");
342        assert_eq!(task.links[0].target, "https://github.com/busytools/forge/pull/1889",);
343        assert_eq!(
344            task.estimate.as_ref().map(|e| (e.words.as_str(), e.secs)),
345            Some(("1d", 86_400)),
346            "the legacy estimate parsed with its words kept",
347        );
348    }
349
350    /// The old `blocked` spelling is a wait whose reason was never stated:
351    /// the board draws it as a miss rather than a blank.
352    #[test]
353    fn a_legacy_blocked_row_reads_as_an_unstated_wait() {
354        let dir = tempdir().expect("tempdir");
355        let db = open_db(dir.path());
356        write_raw(
357            &db,
358            "forge\u{0}t-2",
359            r#"{
360                "id": "t-2", "project_name": "forge", "subject": "stuck row",
361                "active_form": null, "detail": null, "status": "blocked",
362                "owner": null, "parent": null,
363                "created_at": {"secs_since_epoch": 0, "nanos_since_epoch": 0},
364                "updated_at": {"secs_since_epoch": 0, "nanos_since_epoch": 0}
365            }"#,
366        );
367        let read = list(&db).expect("list");
368        assert_eq!(read.len(), 1, "the blocked row loads: {read:?}");
369        assert_eq!(read[0].status, TaskStatus::Waiting);
370        assert_eq!(
371            read[0].waiting_on,
372            Some(Waiting {
373                kind: None,
374                detail: None,
375                on: None,
376                verification: false,
377                parked: false,
378                since: None,
379            }),
380            "no kind, no words - the miss the board shows",
381        );
382    }
383
384    /// A fresh row round-trips through the record shape unchanged.
385    #[test]
386    fn a_current_row_round_trips_through_the_store() {
387        let dir = tempdir().expect("tempdir");
388        let db = open_db(dir.path());
389        let mut task = sample_task("t-3", "forge");
390        task.estimate = Some(Estimate { words: "2h".to_owned(), secs: 7_200 });
391        task.links.push(TaskLink {
392            kind: LinkKind::Issue,
393            label: Some("#1886".to_owned()),
394            target: "https://example.invalid/issues/1886".to_owned(),
395            state: Some("open".to_owned()),
396            added_at: std::time::SystemTime::UNIX_EPOCH,
397        });
398        upsert(&db, &task).expect("write");
399        assert_eq!(list(&db).expect("list"), vec![task], "no migration touches a current row");
400    }
401}
forge_workspace/store/tasks.rs

forge_workspace/store/
tasks.rs
