dpm_store/sqlite/runs/
recovery.rs

1//! Backup, restore and read-only verification of a run store.
2//!
3//! The run store is not part of the project store's backup file, so the project store's backup,
4//! restore and verification each carry it explicitly: a backup writes `<backup>.runs` beside the
5//! project backup, a restore copies it into `<store>.runs` bound to the restored lineage, and
6//! verification checks it. Runs recorded under an earlier lineage keep their attribution; they are
7//! reported as foreign, never reattached.
8
9use super::{
10    RunBinding, RunStore, check_workspace,
11    codec::{read_activity, read_event, read_link, read_run},
12    pairing::{PairFacts, read_forks},
13    read::{head, pruned_through, tally_row},
14    read_binding, schema,
15};
16use crate::{
17    LineageError, RunStoreError, StoreError,
18    error::database_error,
19    schema::Layout,
20    sqlite::{
21        integrity::integrity_check_blocking,
22        recovery::{
23            copy_blocking,
24            files::{
25                ReadMode, canonical_blocking, create_target_blocking, discard_blocking,
26                fingerprint_blocking, open_read_only_blocking, read_mode_blocking,
27            },
28        },
29        snapshot::revision_to_sql,
30    },
31};
32use chrono::{DateTime, Utc};
33use dpm_engine::{check_run_transition, run_start_event};
34use dpm_model::{LineageId, RunId, RunLink, RunState, WorkspaceId};
35use rusqlite::Connection;
36use serde::Serialize;
37use std::{
38    collections::BTreeMap,
39    path::{Path, PathBuf},
40};
41
42/// What a verified run store holds.
43#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
44pub struct RunStoreReport {
45    /// Verified file as a canonical absolute path.
46    pub path: PathBuf,
47    /// Layout version in the header.
48    pub schema_version: i64,
49    /// Workspace the run store is bound to.
50    pub workspace_id: WorkspaceId,
51    /// History the run store was created or last restored for; runs recorded earlier are foreign.
52    pub lineage_id: LineageId,
53    /// Whether the file is a backup that only a restore makes writable.
54    pub archived: bool,
55    /// Runs recorded.
56    pub runs: u64,
57    /// Lifecycle facts recorded; they are never pruned.
58    pub lifecycle_events: u64,
59    /// Activity records still held.
60    pub activity_retained: u64,
61    /// Highest activity sequence retention has removed.
62    pub activity_pruned_through: u64,
63}
64
65pub(super) fn corrupt(path: &Path, detail: impl Into<String>) -> RunStoreError {
66    RunStoreError::Corrupt {
67        path: path.to_path_buf(),
68        detail: detail.into(),
69    }
70}
71
72/// Verify a run store without writing to it or creating any file next to it.
73pub(in crate::sqlite) fn verify_blocking(path: &Path) -> Result<RunStoreReport, RunStoreError> {
74    verify_with_facts_blocking(path).map(|(report, _)| report)
75}
76
77/// Verify a run store and keep what it claims about its project store.
78pub(in crate::sqlite) fn verify_with_facts_blocking(
79    path: &Path,
80) -> Result<(RunStoreReport, PairFacts), RunStoreError> {
81    let path = canonical_blocking(path)?;
82    if read_mode_blocking(&path)? == ReadMode::Immutable {
83        let before = fingerprint_blocking(&path)?;
84        let result = verify_with_blocking(&path, ReadMode::Immutable);
85        // A writer that started meanwhile may have changed pages under the unlocked read.
86        if fingerprint_blocking(&path)? == before {
87            return result;
88        }
89    }
90    verify_with_blocking(&path, ReadMode::Shared)
91}
92
93fn verify_with_blocking(
94    path: &Path,
95    mode: ReadMode,
96) -> Result<(RunStoreReport, PairFacts), RunStoreError> {
97    let mut connection = open_read_only_blocking(path, mode)?;
98    let transaction = connection
99        .transaction()
100        .map_err(database_error(path, "begin run store verification"))?;
101    if schema::check_blocking(&transaction, path)? == Layout::Empty {
102        return Err(StoreError::UnrecognizedSchema {
103            path: path.to_path_buf(),
104            detail: "no run store tables".into(),
105        }
106        .into());
107    }
108    integrity_check_blocking(&transaction, path)?;
109    let binding = read_binding(&transaction, path)?;
110    let runs = check_runs(&transaction, path)?;
111    let (lifecycle_events, last) = check_lifecycle(&transaction, path, &runs)?;
112    let (activity_retained, pruned) = check_activity(&transaction, path, &runs)?;
113    let links = check_links(&transaction, path, &runs)?;
114    let report = RunStoreReport {
115        path: path.to_path_buf(),
116        schema_version: schema::RUN_STORE_VERSION,
117        workspace_id: binding.workspace_id,
118        lineage_id: binding.lineage_id,
119        archived: binding.archived,
120        runs: u64::try_from(runs.len()).unwrap_or(u64::MAX),
121        lifecycle_events,
122        activity_retained,
123        activity_pruned_through: pruned,
124    };
125    let runs = runs
126        .into_values()
127        .map(|record| {
128            let ended = last
129                .get(&record.id)
130                .filter(|(state, _)| state.is_terminal())
131                .map(|(_, at)| *at);
132            (record, ended)
133        })
134        .collect();
135    let facts = PairFacts {
136        path: path.to_path_buf(),
137        binding,
138        runs,
139        links,
140        forks: read_forks(&transaction, path)?,
141    };
142    Ok((report, facts))
143}
144
145/// Every run decodes and sits under its own identity; returns each one's task.
146fn check_runs(
147    connection: &Connection,
148    path: &Path,
149) -> Result<BTreeMap<RunId, dpm_model::RunRecord>, RunStoreError> {
150    let mut statement = connection
151        .prepare("SELECT record_json, run_id, work_id FROM runs")
152        .map_err(database_error(path, "prepare run verification"))?;
153    let mut rows = statement
154        .query([])
155        .map_err(database_error(path, "read runs"))?;
156    let mut runs = BTreeMap::new();
157    while let Some(row) = rows.next().map_err(database_error(path, "read runs"))? {
158        let record = read_run(row, path)?;
159        let (id, work): (String, String) = (
160            row.get(1).map_err(database_error(path, "read run id"))?,
161            row.get(2).map_err(database_error(path, "read run task"))?,
162        );
163        if record.id.to_string() != id || record.work.to_string() != work {
164            return Err(corrupt(
165                path,
166                format!("run row {id} disagrees with its record"),
167            ));
168        }
169        runs.insert(record.id, record);
170    }
171    Ok(runs)
172}
173
174/// Each run's last lifecycle state and when it was recorded.
175type LastStates = BTreeMap<RunId, (RunState, DateTime<Utc>)>;
176
177/// Lifecycle is contiguous, starts every run at its own identity and obeys the state machine.
178fn check_lifecycle(
179    connection: &Connection,
180    path: &Path,
181    runs: &BTreeMap<RunId, dpm_model::RunRecord>,
182) -> Result<(u64, LastStates), RunStoreError> {
183    let mut statement = connection
184        .prepare(&format!(
185            "{} ORDER BY sequence",
186            super::codec::LIFECYCLE_COLUMNS
187        ))
188        .map_err(database_error(path, "prepare lifecycle verification"))?;
189    let mut rows = statement
190        .query([])
191        .map_err(database_error(path, "read lifecycle"))?;
192    let mut states = LastStates::new();
193    let mut previous = 0_u64;
194    while let Some(row) = rows
195        .next()
196        .map_err(database_error(path, "read lifecycle"))?
197    {
198        let entry = read_event(row, path)?;
199        if entry.sequence != previous.wrapping_add(1) {
200            return Err(corrupt(
201                path,
202                format!(
203                    "lifecycle skips from sequence {previous} to {}",
204                    entry.sequence
205                ),
206            ));
207        }
208        previous = entry.sequence;
209        let event = entry.event;
210        let Some(record) = runs.get(&event.run) else {
211            return Err(corrupt(
212                path,
213                format!("lifecycle names unknown run {}", event.run),
214            ));
215        };
216        match states.get(&event.run) {
217            None => {
218                let begins = event.id == run_start_event(event.run)
219                    && event.state == RunState::Working
220                    && event.recorded_at == record.started_at;
221                if !begins {
222                    return Err(corrupt(
223                        path,
224                        format!("run {} does not begin with its start", event.run),
225                    ));
226                }
227            }
228            Some((from, _)) => {
229                check_run_transition(event.run, *from, event.state).map_err(|error| {
230                    corrupt(path, format!("illegal transition recorded: {error}"))
231                })?;
232            }
233        }
234        states.insert(event.run, (event.state, event.recorded_at));
235    }
236    if let Some(unstarted) = runs.keys().find(|id| !states.contains_key(id)) {
237        return Err(corrupt(
238            path,
239            format!("run {unstarted} has no lifecycle record"),
240        ));
241    }
242    if head(connection, path, "run_lifecycle")? != previous {
243        return Err(corrupt(path, "lifecycle records were removed from the end"));
244    }
245    Ok((previous, states))
246}
247
248/// Retained activity is contiguous above what retention removed, and tallies cover what is held.
249fn check_activity(
250    connection: &Connection,
251    path: &Path,
252    runs: &BTreeMap<RunId, dpm_model::RunRecord>,
253) -> Result<(u64, u64), RunStoreError> {
254    let pruned = pruned_through(connection, path)?;
255    let mut statement = connection
256        .prepare(&format!(
257            "{} ORDER BY sequence",
258            super::codec::ACTIVITY_COLUMNS
259        ))
260        .map_err(database_error(path, "prepare activity verification"))?;
261    let mut rows = statement
262        .query([])
263        .map_err(database_error(path, "read activity"))?;
264    let (mut previous, mut retained) = (None::<u64>, 0_u64);
265    let mut per_run: BTreeMap<RunId, u64> = BTreeMap::new();
266    let mut newest_source: BTreeMap<RunId, u64> = BTreeMap::new();
267    while let Some(row) = rows.next().map_err(database_error(path, "read activity"))? {
268        let entry = read_activity(row, path)?;
269        let expected = previous.map_or(pruned.wrapping_add(1), |last| last.wrapping_add(1));
270        if entry.sequence != expected {
271            return Err(corrupt(
272                path,
273                format!(
274                    "activity holds sequence {} where {expected} should follow",
275                    entry.sequence
276                ),
277            ));
278        }
279        if !runs.contains_key(&entry.record.run) {
280            return Err(corrupt(
281                path,
282                format!("activity names unknown run {}", entry.record.run),
283            ));
284        }
285        previous = Some(entry.sequence);
286        retained = retained.wrapping_add(1);
287        let held = per_run.entry(entry.record.run).or_default();
288        *held = held.wrapping_add(1);
289        let source = entry.record.source_sequence;
290        if newest_source
291            .insert(entry.record.run, source)
292            .is_some_and(|earlier| earlier >= source)
293        {
294            return Err(corrupt(
295                path,
296                format!(
297                    "run {} holds source sequence {source} after a later or equal one",
298                    entry.record.run
299                ),
300            ));
301        }
302    }
303    if head(connection, path, "run_activity")? != pruned.wrapping_add(retained) {
304        return Err(corrupt(
305            path,
306            "activity records were removed outside retention",
307        ));
308    }
309    // The receipt count is a wrapping counter, so only the high-water mark is checked: it must
310    // exist for every run that holds activity and may not sit below a sequence that was accepted.
311    for (run, newest) in newest_source {
312        let high_water = tally_row(connection, path, run)?.map(|counts| counts.high_water);
313        if high_water.is_none_or(|high_water| high_water < newest) {
314            return Err(corrupt(
315                path,
316                format!("run {run} holds source sequence {newest} above its high-water mark"),
317            ));
318        }
319    }
320    Ok((retained, pruned))
321}
322
323fn check_links(
324    connection: &Connection,
325    path: &Path,
326    runs: &BTreeMap<RunId, dpm_model::RunRecord>,
327) -> Result<Vec<RunLink>, RunStoreError> {
328    let mut statement = connection
329        .prepare(&format!(
330            "{} ORDER BY operation_id",
331            super::codec::LINK_COLUMNS
332        ))
333        .map_err(database_error(path, "prepare link verification"))?;
334    let mut rows = statement
335        .query([])
336        .map_err(database_error(path, "read links"))?;
337    let mut links = Vec::new();
338    while let Some(row) = rows.next().map_err(database_error(path, "read links"))? {
339        let link = read_link(row, path)?;
340        match runs.get(&link.run) {
341            Some(record) if record.work == link.work => {}
342            _ => {
343                return Err(corrupt(
344                    path,
345                    format!(
346                        "operation {} is linked to a run that does not match it",
347                        link.operation
348                    ),
349                ));
350            }
351        }
352        links.push(link);
353    }
354    Ok(links)
355}
356
357/// What a copied run store becomes before it is closed.
358#[derive(Clone, Copy)]
359enum Seal {
360    Archive,
361    /// A restore: the copy continues as `lineage`, starting at the project's `revision`.
362    Fork {
363        lineage: LineageId,
364        revision: u64,
365    },
366}
367
368fn seal_blocking(target: &Connection, path: &Path, seal: Seal) -> Result<(), StoreError> {
369    let changed = match seal {
370        Seal::Archive => target.execute("UPDATE run_binding SET archived = 1", []),
371        Seal::Fork { lineage, revision } => {
372            // The fork is recorded as explicit evidence, so the runs of the lineage it continued
373            // can still be traced to the history that holds their observed revisions.
374            let parent: String = target
375                .query_row("SELECT lineage_id FROM run_binding", [], |row| row.get(0))
376                .map_err(database_error(path, "read the run store's lineage"))?;
377            target
378                .execute(
379                    "INSERT INTO run_lineage_forks(lineage_id, parent_lineage_id, revision) \
380                     VALUES(?1, ?2, ?3)",
381                    rusqlite::params![lineage.to_string(), parent, revision_to_sql(revision)],
382                )
383                .map_err(database_error(path, "record the lineage fork"))?;
384            target.execute(
385                "UPDATE run_binding SET lineage_id = ?1, archived = 0",
386                [lineage.to_string()],
387            )
388        }
389    }
390    .map_err(database_error(path, "seal copied run store"))?;
391    if changed == 1 {
392        Ok(())
393    } else {
394        Err(StoreError::Lineage(LineageError::Missing {
395            path: path.to_path_buf(),
396        }))
397    }
398}
399
400impl RunStore {
401    /// Copy this run store into a new file, sealed as an archive of its binding, and verify the
402    /// copy. The target is created by the caller and removed again if this fails.
403    pub(in crate::sqlite) fn backup_to_blocking(
404        &self,
405        to: &Path,
406    ) -> Result<RunStoreReport, RunStoreError> {
407        copy_blocking(&self.connection, to, "delete", |target| {
408            seal_blocking(target, to, Seal::Archive)
409        })?;
410        verify_blocking(to)
411    }
412}
413
414/// Copy a verified run archive into a new run store bound to `lineage`, the lineage of the
415/// restored project store, which starts at project `revision`, and verify the copy. The target is created by the caller.
416///
417/// Runs in the archive keep the lineage they were recorded under; they read as foreign in the
418/// restored store, so nothing is silently reattached to the new history.
419pub(in crate::sqlite) fn restore_blocking(
420    from: &Path,
421    to: &Path,
422    workspace: WorkspaceId,
423    lineage: LineageId,
424    revision: u64,
425) -> Result<RunStoreReport, RunStoreError> {
426    let from = canonical_blocking(from)?;
427    let report = verify_blocking(&from)?;
428    if !report.archived {
429        return Err(StoreError::from(LineageError::NotAnArchive { path: from }).into());
430    }
431    check_workspace(
432        &RunBinding {
433            workspace_id: report.workspace_id,
434            lineage_id: report.lineage_id,
435            archived: report.archived,
436        },
437        workspace,
438        &from,
439    )?;
440    let mode = read_mode_blocking(&from)?;
441    let before = fingerprint_blocking(&from)?;
442    let source = open_read_only_blocking(&from, mode)?;
443    copy_blocking(&source, to, "wal", |target| {
444        seal_blocking(target, to, Seal::Fork { lineage, revision })
445    })?;
446    // An immutable read cannot see a writer that started meanwhile, so the copy is only trusted
447    // if the source provably did not change.
448    if mode == ReadMode::Immutable && fingerprint_blocking(&from)? != before {
449        return Err(StoreError::BackupIncomplete {
450            path: to.to_path_buf(),
451            state: format!("{} changed during the restore", from.display()),
452        }
453        .into());
454    }
455    verify_blocking(to)
456}
457
458/// Create a run store backup target exclusively, as the project backup creates its own.
459pub(in crate::sqlite) fn create_target(to: &Path) -> Result<(), StoreError> {
460    create_target_blocking(to)
461}
462
463/// Remove a run store target this command created after a failure.
464pub(in crate::sqlite) fn discard(to: &Path) {
465    discard_blocking(to);
466}
dpm_store/sqlite/runs/recovery.rs

dpm_store/sqlite/runs/
recovery.rs
