sequel_mcp/backup/
journal.rs

1//! Operation journal (D3): durable, crash-recoverable records linking
2//! backups, mutations, and audit across the local/remote database
3//! boundary. Because the audit DB and the target server cannot share a
4//! distributed transaction, every gated mutation writes journal
5//! transitions; a crash between remote commit and local finalization
6//! leaves a visibly ambiguous record rather than silent success/failure.
7
8use crate::audit::AuditDb;
9use rusqlite::params;
10use std::sync::Arc;
11use thiserror::Error;
12
13#[derive(Debug, Clone, Copy, PartialEq, Eq)]
14pub enum JournalState {
15    Planned,
16    BackupCapturing,
17    BackupDurable,
18    MutationExecuting,
19    MutationCommitted,
20    AuditFinalized,
21    Failed,
22    Uncertain,
23}
24
25impl JournalState {
26    fn as_str(&self) -> &'static str {
27        match self {
28            JournalState::Planned => "planned",
29            JournalState::BackupCapturing => "backup_capturing",
30            JournalState::BackupDurable => "backup_durable",
31            JournalState::MutationExecuting => "mutation_executing",
32            JournalState::MutationCommitted => "mutation_committed",
33            JournalState::AuditFinalized => "audit_finalized",
34            JournalState::Failed => "failed",
35            JournalState::Uncertain => "uncertain",
36        }
37    }
38}
39
40#[derive(Debug, Error)]
41pub enum JournalError {
42    #[error("journal write failed: {0}")]
43    Db(#[from] rusqlite::Error),
44    #[error("transition {from} -> {to} is not allowed")]
45    IllegalTransition { from: String, to: String },
46}
47
48/// Allowed forward transitions; `failed`/`uncertain` are terminal for
49/// recovery purposes, `uncertain` additionally reachable from any
50/// executing-or-later state (crash/timeout ambiguity).
51const TRANSITIONS: &[(&str, &[&str])] = &[
52    ("planned", &["backup_capturing", "failed"]),
53    (
54        "backup_capturing",
55        &["backup_durable", "mutation_executing", "failed"],
56    ),
57    ("backup_durable", &["mutation_executing", "failed"]),
58    (
59        "mutation_executing",
60        &["mutation_committed", "failed", "uncertain"],
61    ),
62    ("mutation_committed", &["audit_finalized", "uncertain"]),
63    ("audit_finalized", &[]),
64    ("failed", &[]),
65    ("uncertain", &[]),
66];
67
68pub struct Journal<'a> {
69    db: &'a Arc<AuditDb>,
70    id: i64,
71}
72
73pub fn ensure_table(db: &Arc<AuditDb>) -> rusqlite::Result<()> {
74    db.with(|c| {
75        c.execute_batch(
76            "CREATE TABLE IF NOT EXISTS operation_journal (
77               id           INTEGER PRIMARY KEY AUTOINCREMENT,
78               ts           TEXT    NOT NULL,
79               request_id   TEXT    NOT NULL,
80               connection   TEXT    NOT NULL,
81               databases    TEXT    NOT NULL,
82               category     TEXT    NOT NULL,
83               state        TEXT    NOT NULL,
84               detail       TEXT,
85               backup_id    INTEGER,
86               audit_id     INTEGER
87             );
88             CREATE INDEX IF NOT EXISTS idx_journal_state ON operation_journal(state, ts);",
89        )
90    })
91}
92
93impl<'a> Journal<'a> {
94    pub fn create(
95        db: &'a Arc<AuditDb>,
96        request_id: &str,
97        connection: &str,
98        databases: &[String],
99        category: &str,
100    ) -> Result<Journal<'a>, JournalError> {
101        let ts = now_iso();
102        let databases = serde_json::to_string(databases).unwrap_or_else(|_| "[]".into());
103        let id = db.with_tx(|c| {
104            c.execute(
105                "INSERT INTO operation_journal
106                   (ts, request_id, connection, databases, category, state)
107                 VALUES (?1, ?2, ?3, ?4, ?5, 'planned')",
108                params![ts, request_id, connection, databases, category],
109            )?;
110            Ok(c.last_insert_rowid())
111        })?;
112        Ok(Journal { db, id })
113    }
114
115    pub fn id(&self) -> i64 {
116        self.id
117    }
118
119    /// Re-open an existing journal row by id (for finalization by the
120    /// caller that performed the audit write).
121    pub fn from_id(db: &'a Arc<AuditDb>, id: i64) -> Journal<'a> {
122        Journal { db, id }
123    }
124
125    pub fn transition(&self, to: JournalState, detail: Option<&str>) -> Result<(), JournalError> {
126        self.db.with_tx(|c| {
127            let from: String = c.query_row(
128                "SELECT state FROM operation_journal WHERE id = ?1",
129                params![self.id],
130                |r| r.get(0),
131            )?;
132            let allowed = TRANSITIONS
133                .iter()
134                .find(|(f, _)| *f == from.as_str())
135                .map(|(_, ts)| ts.contains(&to.as_str()))
136                .unwrap_or(false);
137            if !allowed {
138                return Err(rusqlite::Error::ToSqlConversionFailure(Box::new(
139                    std::io::Error::other(format!("illegal transition {from} -> {}", to.as_str())),
140                )));
141            }
142            c.execute(
143                "UPDATE operation_journal
144                    SET state = ?2, detail = COALESCE(?3, detail), ts = ?4
145                  WHERE id = ?1",
146                params![self.id, to.as_str(), detail, now_iso()],
147            )?;
148            Ok(())
149        })?;
150        Ok(())
151    }
152
153    pub fn link_backup(&self, backup_id: i64) -> rusqlite::Result<()> {
154        self.db.with(|c| {
155            c.execute(
156                "UPDATE operation_journal SET backup_id = ?2 WHERE id = ?1",
157                params![self.id, backup_id],
158            )
159        })?;
160        Ok(())
161    }
162
163    pub fn link_audit(&self, audit_id: i64) -> rusqlite::Result<()> {
164        self.db.with(|c| {
165            c.execute(
166                "UPDATE operation_journal SET audit_id = ?2 WHERE id = ?1",
167                params![self.id, audit_id],
168            )
169        })?;
170        Ok(())
171    }
172
173    /// Rows left in a non-terminal or ambiguous state — the crash-recovery
174    /// surface. `mutation_committed` without `audit_finalized` is the
175    /// visibly-ambiguous case: the remote commit may have landed.
176    pub fn recoverable(db: &Arc<AuditDb>) -> rusqlite::Result<Vec<(i64, String, String)>> {
177        db.with(|c| {
178            let mut stmt = c.prepare(
179                "SELECT id, state, COALESCE(request_id, '') FROM operation_journal
180                  WHERE state NOT IN ('audit_finalized', 'failed', 'uncertain')
181                     OR (state = 'mutation_committed' AND audit_id IS NULL)",
182            )?;
183            let rows = stmt.query_map([], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)))?;
184            rows.collect()
185        })
186    }
187}
188
189fn now_iso() -> String {
190    time::OffsetDateTime::now_utc()
191        .format(&time::format_description::well_known::Rfc3339)
192        .unwrap_or_else(|_| "1970-01-01T00:00:00Z".into())
193}
194
195#[cfg(test)]
196mod tests {
197    use super::*;
198
199    fn db() -> (tempfile::TempDir, Arc<AuditDb>) {
200        let dir = tempfile::TempDir::new().unwrap();
201        let d = Arc::new(AuditDb::at_path(&dir.path().join("a.sqlite")).unwrap());
202        ensure_table(&d).unwrap();
203        (dir, d)
204    }
205
206    #[test]
207    fn happy_path_transitions() {
208        let (_dir, db) = db();
209        let j = Journal::create(&db, "req-1", "c1", &["app".into()], "write").unwrap();
210        j.transition(JournalState::BackupCapturing, None).unwrap();
211        j.transition(JournalState::BackupDurable, None).unwrap();
212        j.transition(JournalState::MutationExecuting, None).unwrap();
213        j.transition(JournalState::MutationCommitted, None).unwrap();
214        j.transition(JournalState::AuditFinalized, None).unwrap();
215        assert!(Journal::recoverable(&db).unwrap().is_empty());
216    }
217
218    #[test]
219    fn illegal_transitions_rejected() {
220        let (_dir, db) = db();
221        let j = Journal::create(&db, "req-2", "c1", &[], "ddl").unwrap();
222        assert!(j.transition(JournalState::MutationCommitted, None).is_err());
223        j.transition(JournalState::BackupCapturing, None).unwrap();
224        // backup_capturing -> mutation_executing is legal for backup-less
225        // operations; the illegal jump is straight to committed.
226        assert!(j.transition(JournalState::MutationCommitted, None).is_err());
227    }
228
229    #[test]
230    fn crash_before_audit_finalization_is_recoverable() {
231        let (_dir, db) = db();
232        let j = Journal::create(&db, "req-3", "c1", &[], "write").unwrap();
233        for s in [
234            JournalState::BackupCapturing,
235            JournalState::BackupDurable,
236            JournalState::MutationExecuting,
237            JournalState::MutationCommitted,
238        ] {
239            j.transition(s, None).unwrap();
240        }
241        // Crash here: committed remotely, audit not finalized.
242        let r = Journal::recoverable(&db).unwrap();
243        assert_eq!(r.len(), 1, "{r:?}");
244        assert_eq!(r[0].1, "mutation_committed");
245    }
246
247    #[test]
248    fn uncertain_is_terminal_and_not_recoverable() {
249        let (_dir, db) = db();
250        let j = Journal::create(&db, "req-4", "c1", &[], "write").unwrap();
251        j.transition(JournalState::BackupCapturing, None).unwrap();
252        j.transition(JournalState::BackupDurable, None).unwrap();
253        j.transition(JournalState::MutationExecuting, None).unwrap();
254        j.transition(JournalState::Uncertain, Some("cancellation inconclusive"))
255            .unwrap();
256        assert!(Journal::recoverable(&db).unwrap().is_empty());
257    }
258
259    #[test]
260    fn backup_failure_leads_to_failed_not_mutation() {
261        let (_dir, db) = db();
262        let j = Journal::create(&db, "req-5", "c1", &[], "write").unwrap();
263        j.transition(JournalState::BackupCapturing, None).unwrap();
264        j.transition(JournalState::Failed, Some("backup overflow"))
265            .unwrap();
266        assert!(Journal::recoverable(&db).unwrap().is_empty());
267    }
268}
sequel_mcp/backup/journal.rs

sequel_mcp/backup/
journal.rs
