stackless_core/state/
journal.rs

1//! The per-step checkpoint journal (§2): every step that creates a
2//! resource records it before moving on.
3
4use super::error::StateError;
5use super::row::Row;
6use super::store::Store;
7
8#[derive(Debug, Clone, PartialEq, Eq)]
9pub struct Checkpoint {
10    pub instance: String,
11    pub step_id: String,
12    pub resource_kind: String,
13    pub resource_id: String,
14    /// Step-specific JSON the substrate needs to re-find the resource.
15    pub payload: String,
16    pub recorded_at: i64,
17}
18
19impl TryFrom<&Row> for Checkpoint {
20    type Error = StateError;
21
22    fn try_from(row: &Row) -> Result<Self, Self::Error> {
23        Ok(Self {
24            instance: row.get_string(0)?,
25            step_id: row.get_string(1)?,
26            resource_kind: row.get_string(2)?,
27            resource_id: row.get_string(3)?,
28            payload: row.get_string(4)?,
29            recorded_at: row.get_i64(5)?,
30        })
31    }
32}
33
34impl Store {
35    pub fn record_checkpoint(
36        &self,
37        instance: &str,
38        step_id: &str,
39        resource_kind: &str,
40        resource_id: &str,
41        payload: &str,
42    ) -> Result<(), StateError> {
43        self.execute(
44            "INSERT INTO checkpoints (instance, step_id, resource_kind, resource_id, payload, recorded_at)
45             VALUES (?1, ?2, ?3, ?4, ?5, ?6)
46             ON CONFLICT(instance, step_id) DO UPDATE SET
47               resource_kind = excluded.resource_kind,
48               resource_id = excluded.resource_id,
49               payload = excluded.payload,
50               recorded_at = excluded.recorded_at",
51            &[
52                instance.into(),
53                step_id.into(),
54                resource_kind.into(),
55                resource_id.into(),
56                payload.into(),
57                Self::now().into(),
58            ],
59        )?;
60        Ok(())
61    }
62
63    pub fn checkpoint(
64        &self,
65        instance: &str,
66        step_id: &str,
67    ) -> Result<Option<Checkpoint>, StateError> {
68        self.query_row(
69            "SELECT instance, step_id, resource_kind, resource_id, payload, recorded_at
70             FROM checkpoints WHERE instance = ?1 AND step_id = ?2",
71            &[instance.into(), step_id.into()],
72            |row| Checkpoint::try_from(row),
73        )
74    }
75
76    /// All checkpoints for an instance in recording order — what an
77    /// interrupted `down` must hunt down.
78    pub fn checkpoints(&self, instance: &str) -> Result<Vec<Checkpoint>, StateError> {
79        self.query_map(
80            "SELECT instance, step_id, resource_kind, resource_id, payload, recorded_at
81             FROM checkpoints WHERE instance = ?1 ORDER BY rowid",
82            &[instance.into()],
83            |row| Checkpoint::try_from(row),
84        )
85    }
86
87    /// Remove one checkpoint after its resource is verifiably gone.
88    pub fn remove_checkpoint(&self, instance: &str, step_id: &str) -> Result<(), StateError> {
89        self.execute(
90            "DELETE FROM checkpoints WHERE instance = ?1 AND step_id = ?2",
91            &[instance.into(), step_id.into()],
92        )?;
93        Ok(())
94    }
95}
stackless_core/state/journal.rs

stackless_core/state/
journal.rs
