perspt_store/
repair.rs

1//! Recoverable DuckDB WAL quarantine.
2
3use std::ffi::OsString;
4use std::fs::{File, OpenOptions};
5use std::io::{Read, Write};
6use std::path::{Path, PathBuf};
7use std::time::{SystemTime, UNIX_EPOCH};
8
9use anyhow::{Context, Result};
10use serde::{Deserialize, Serialize};
11use sha2::{Digest, Sha256};
12
13#[derive(Debug, Clone, Serialize, Deserialize)]
14pub struct RepairReport {
15    pub database_path: PathBuf,
16    pub wal_path: PathBuf,
17    pub wal_size: u64,
18    pub wal_sha256: String,
19    pub database_backup: PathBuf,
20    pub wal_backup: PathBuf,
21    pub wal_quarantine: PathBuf,
22    pub recovered_table_count: u64,
23}
24
25/// Quarantine a WAL only after durable backups exist, then verify the database
26/// through a read-only connection. Historical bytes are never deleted.
27pub fn repair_database(path: &Path, discard_wal: bool) -> Result<RepairReport> {
28    anyhow::ensure!(discard_wal, "repair requires explicit --discard-wal");
29    validate_regular(path, "database")?;
30    let wal = wal_path(path);
31    validate_regular(&wal, "WAL")?;
32    let parent = path.parent().context("database path has no parent")?;
33    let stamp = timestamp()?;
34    let database_backup = sibling(path, &format!("backup.{stamp}"))?;
35    let wal_backup = sibling(&wal, &format!("backup.{stamp}"))?;
36    let wal_quarantine = sibling(&wal, &format!("quarantine.{stamp}"))?;
37    ensure_targets_absent(&[&database_backup, &wal_backup, &wal_quarantine])?;
38
39    let wal_size = std::fs::metadata(&wal)?.len();
40    let wal_sha256 = sha256_file(&wal)?;
41    durable_copy(path, &database_backup)?;
42    durable_copy(&wal, &wal_backup)?;
43    std::fs::rename(&wal, &wal_quarantine).context("quarantining DuckDB WAL")?;
44    sync_directory(parent)?;
45
46    let recovered = verify_recovery(path);
47    let recovered_table_count = match recovered {
48        Ok(count) => count,
49        Err(error) => {
50            restore_wal(&wal, &wal_quarantine, parent)?;
51            return Err(error.context("database recovery failed; original WAL restored"));
52        }
53    };
54    Ok(RepairReport {
55        database_path: path.to_path_buf(),
56        wal_path: wal,
57        wal_size,
58        wal_sha256,
59        database_backup,
60        wal_backup,
61        wal_quarantine,
62        recovered_table_count,
63    })
64}
65
66fn validate_regular(path: &Path, label: &str) -> Result<()> {
67    let metadata = std::fs::symlink_metadata(path)
68        .with_context(|| format!("reading {label} metadata at {}", path.display()))?;
69    anyhow::ensure!(
70        metadata.file_type().is_file(),
71        "{label} is not a regular file"
72    );
73    Ok(())
74}
75
76fn wal_path(database: &Path) -> PathBuf {
77    let mut name = database.as_os_str().to_os_string();
78    name.push(".wal");
79    PathBuf::from(name)
80}
81
82fn sibling(path: &Path, suffix: &str) -> Result<PathBuf> {
83    let parent = path.parent().context("repair target has no parent")?;
84    let mut name: OsString = path
85        .file_name()
86        .context("repair target has no filename")?
87        .into();
88    name.push(".");
89    name.push(suffix);
90    Ok(parent.join(name))
91}
92
93fn timestamp() -> Result<u128> {
94    Ok(SystemTime::now()
95        .duration_since(UNIX_EPOCH)
96        .context("system clock precedes Unix epoch")?
97        .as_millis())
98}
99
100fn ensure_targets_absent(paths: &[&Path]) -> Result<()> {
101    for path in paths {
102        anyhow::ensure!(
103            !path.exists(),
104            "repair output already exists: {}",
105            path.display()
106        );
107    }
108    Ok(())
109}
110
111fn durable_copy(source: &Path, destination: &Path) -> Result<()> {
112    let mut input = File::open(source)?;
113    let mut output = OpenOptions::new()
114        .write(true)
115        .create_new(true)
116        .open(destination)?;
117    std::io::copy(&mut input, &mut output)?;
118    output.flush()?;
119    output.sync_all()?;
120    sync_directory(destination.parent().context("backup has no parent")?)
121}
122
123fn sha256_file(path: &Path) -> Result<String> {
124    let mut file = File::open(path)?;
125    let mut digest = Sha256::new();
126    let mut buffer = [0u8; 64 * 1024];
127    loop {
128        let count = file.read(&mut buffer)?;
129        if count == 0 {
130            break;
131        }
132        digest.update(&buffer[..count]);
133    }
134    Ok(format!("{:x}", digest.finalize()))
135}
136
137fn verify_recovery(path: &Path) -> Result<u64> {
138    let config = duckdb::Config::default()
139        .access_mode(duckdb::AccessMode::ReadOnly)
140        .context("configuring read-only recovery verification")?;
141    let connection = duckdb::Connection::open_with_flags(path, config)
142        .context("opening recovered database read-only")?;
143    connection
144        .query_row(
145            "SELECT count(*) FROM information_schema.tables WHERE table_schema = 'main'",
146            [],
147            |row| row.get(0),
148        )
149        .context("querying recovered database catalog")
150}
151
152fn restore_wal(wal: &Path, quarantine: &Path, parent: &Path) -> Result<()> {
153    if wal.exists() {
154        let failed = sibling(wal, &format!("failed-recovery.{}", timestamp()?))?;
155        std::fs::rename(wal, failed).context("preserving WAL created during failed recovery")?;
156    }
157    std::fs::rename(quarantine, wal).context("restoring original DuckDB WAL")?;
158    sync_directory(parent)
159}
160
161fn sync_directory(path: &Path) -> Result<()> {
162    File::open(path)?.sync_all()?;
163    Ok(())
164}
165
166#[cfg(test)]
167mod tests {
168    use super::*;
169
170    #[test]
171    fn wal_path_appends_instead_of_replacing_extension() {
172        assert_eq!(
173            wal_path(Path::new("state.db")),
174            PathBuf::from("state.db.wal")
175        );
176    }
177
178    #[test]
179    fn repair_requires_explicit_discard_flag() {
180        let temporary = tempfile::tempdir().unwrap();
181        let database = temporary.path().join("state.db");
182        std::fs::write(&database, b"not opened").unwrap();
183        assert!(repair_database(&database, false).is_err());
184    }
185}
perspt_store/repair.rs

perspt_store/
repair.rs
