test(pds): e2e & durability coverage for MST self-heal

Lewis: May this revision serve well! <lu5a@proton.me>

crates/tranquil-pds/Cargo.toml

@@ -94,6 +94,7 @@ frontend = []
 native-tls-roots = ["tranquil-oauth/native-tls-roots"]
 
 [dev-dependencies]
+tempfile = "3"
 ciborium = { workspace = true }
 ctor = { workspace = true }
 testcontainers = { workspace = true }

crates/tranquil-pds/tests/car_export_detects_corruption.rs

@@ -0,0 +1,76 @@
+use std::sync::Arc;
+
+use cid::Cid;
+use jacquard_repo::mst::Mst;
+use jacquard_repo::storage::BlockStore;
+use tranquil_pds::api::error::ApiError;
+use tranquil_pds::repo::AnyBlockStore;
+use tranquil_pds::scheduled::generate_repo_car;
+use tranquil_store::blockstore::{BlockStoreConfig, GroupCommitConfig, TranquilBlockStore};
+
+const RECORD_COUNT: usize = 200;
+
+fn open_store(dir: &std::path::Path) -> AnyBlockStore {
+    let cfg = BlockStoreConfig {
+        data_dir: dir.join("data"),
+        index_dir: dir.join("index"),
+        max_file_size: 64 * 1024,
+        group_commit: GroupCommitConfig::default(),
+        shard_count: 1,
+    };
+    AnyBlockStore::TranquilStore(TranquilBlockStore::open(cfg).expect("open block store"))
+}
+
+async fn build_tree(any: &AnyBlockStore) -> Cid {
+    let mut mst = Mst::new(Arc::new(any.clone()));
+    for i in 0..RECORD_COUNT {
+        let key = format!("app.bsky.feed.post/{i:0>6}");
+        let cid = any
+            .put(format!("record body {i}").as_bytes())
+            .await
+            .expect("put record");
+        mst.add_mut(&key, cid).await.expect("mst add");
+    }
+    mst.persist().await.expect("persist mst")
+}
+
+fn shred_data_files(data_dir: &std::path::Path) {
+    let mut shredded = false;
+    for entry in std::fs::read_dir(data_dir).expect("read data dir") {
+        let path = entry.expect("dir entry").path();
+        if path.extension().and_then(|e| e.to_str()) != Some("tqb") {
+            continue;
+        }
+        let mut bytes = std::fs::read(&path).expect("read data file");
+        let mut off = 5usize;
+        while off + 48 < bytes.len() {
+            bytes[off..off + 48].iter_mut().for_each(|b| *b = 0xFF);
+            off += 192;
+            shredded = true;
+        }
+        std::fs::write(&path, &bytes).expect("write corrupted data file");
+    }
+    assert!(shredded, "no .tqb data file was corrupted");
+}
+
+#[tokio::test]
+async fn car_export_error_is_classified_as_repo_corruption() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let any = open_store(dir.path());
+    let root = build_tree(&any).await;
+
+    generate_repo_car(&any, &root)
+        .await
+        .expect("pristine CAR must generate");
+
+    shred_data_files(&dir.path().join("data"));
+
+    let err = generate_repo_car(&any, &root)
+        .await
+        .expect_err("corrupt CAR export must error");
+    let chain = format!("{err:#}");
+    assert!(
+        ApiError::detail_is_repo_corruption(&chain),
+        "CAR export error must carry the corruption marker so the sync path can schedule self-heal; got: {chain}"
+    );
+}

crates/tranquil-pds/tests/common/mod.rs

@@ -130,6 +130,13 @@ pub fn pds_endpoint() -> String {
     format!("https://{}", pds_hostname())
 }
 
+#[allow(dead_code)]
+pub fn store_data_dir() -> Option<PathBuf> {
+    std::env::var("TRANQUIL_STORE_DATA_DIR")
+        .ok()
+        .map(PathBuf::from)
+}
+
 pub async fn base_url() -> &'static str {
     SERVER_URL.get_or_init(|| {
         let (tx, rx) = std::sync::mpsc::channel();
@@ -950,8 +957,7 @@ pub async fn sequenced_event_for_did(
         .await
         .expect("get_events_since_seq")
         .into_iter()
-        .filter(|event| &event.did == did)
-        .last()
+        .rfind(|event| &event.did == did)
         .unwrap_or_else(|| panic!("event for did {did} not found after flush"))
 }
 

crates/tranquil-pds/tests/mst_repair_audit.rs

@@ -0,0 +1,185 @@
+use std::path::Path;
+use std::sync::Arc;
+
+use cid::Cid;
+use jacquard_repo::mst::Mst;
+use jacquard_repo::storage::BlockStore;
+use tranquil_pds::api::error::ApiError;
+use tranquil_pds::repo::AnyBlockStore;
+use tranquil_store::blockstore::{
+    BLOCK_HEADER_SIZE, BlockStoreConfig, CID_SIZE, GroupCommitConfig, TranquilBlockStore,
+};
+
+const RECORD_COUNT: usize = 300;
+
+fn open_store(dir: &Path) -> AnyBlockStore {
+    let cfg = BlockStoreConfig {
+        data_dir: dir.join("data"),
+        index_dir: dir.join("index"),
+        max_file_size: 64 * 1024,
+        group_commit: GroupCommitConfig::default(),
+        shard_count: 1,
+    };
+    AnyBlockStore::TranquilStore(TranquilBlockStore::open(cfg).expect("open block store"))
+}
+
+async fn build_repo(any: &AnyBlockStore) -> (Cid, Vec<(String, Cid)>) {
+    let mut mst = Mst::new(Arc::new(any.clone()));
+    let mut entries: Vec<(String, Cid)> = Vec::with_capacity(RECORD_COUNT);
+    for i in 0..RECORD_COUNT {
+        let key = format!("app.bsky.feed.post/{i:0>6}");
+        let body = format!("record body number {i}").into_bytes();
+        let cid = any.put(&body).await.expect("put record");
+        mst.add_mut(&key, cid).await.expect("mst add");
+        entries.push((key, cid));
+    }
+    let data_root = mst.persist().await.expect("persist mst");
+    (data_root, entries)
+}
+
+fn shred_data_files(data_dir: &Path) {
+    let mut shredded = false;
+    for entry in std::fs::read_dir(data_dir).expect("read data dir") {
+        let path = entry.expect("dir entry").path();
+        if path.extension().and_then(|e| e.to_str()) != Some("tqb") {
+            continue;
+        }
+        let mut bytes = std::fs::read(&path).expect("read data file");
+        let mut off = 5usize;
+        while off + 48 < bytes.len() {
+            bytes[off..off + 48].iter_mut().for_each(|b| *b = 0xFF);
+            off += 192;
+            shredded = true;
+        }
+        std::fs::write(&path, &bytes).expect("write corrupted data file");
+    }
+    assert!(shredded, "no .tqb data file was corrupted");
+}
+
+fn corrupt_block_with_cid(data_dir: &Path, target: &[u8]) -> bool {
+    for entry in std::fs::read_dir(data_dir).expect("read data dir") {
+        let path = entry.expect("dir entry").path();
+        if path.extension().and_then(|e| e.to_str()) != Some("tqb") {
+            continue;
+        }
+        let mut bytes = std::fs::read(&path).expect("read data file");
+        let mut pos = BLOCK_HEADER_SIZE;
+        while pos + CID_SIZE + 4 <= bytes.len() {
+            let cid = bytes[pos..pos + CID_SIZE].to_vec();
+            let len = u32::from_le_bytes(
+                bytes[pos + CID_SIZE..pos + CID_SIZE + 4]
+                    .try_into()
+                    .unwrap(),
+            ) as usize;
+            let data_start = pos + CID_SIZE + 4;
+            let rec_end = data_start + len + 4;
+            if rec_end > bytes.len() {
+                break;
+            }
+            if cid.as_slice() == target && len > 0 {
+                bytes[data_start] ^= 0xFF;
+                std::fs::write(&path, &bytes).expect("write corrupted data file");
+                return true;
+            }
+            pos = rec_end;
+        }
+    }
+    false
+}
+
+#[tokio::test]
+async fn corrupt_mst_node_classifies_as_repo_corruption() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let any = open_store(dir.path());
+    let (root, entries) = build_repo(&any).await;
+
+    shred_data_files(&dir.path().join("data"));
+
+    let mst = Mst::load(Arc::new(any.clone()), root, None);
+    let mut classified_corruption = false;
+    for (key, _) in &entries {
+        if let Err(e) = mst.get(key).await {
+            assert!(
+                ApiError::from_mst_error("audit", &e).is_repo_corruption(),
+                "corrupt MST node must classify as RepoCorruption via from_mst_error; raw error: {e}"
+            );
+            assert!(
+                ApiError::detail_is_repo_corruption(&e.to_string()),
+                "to_string of corrupt-node error must carry the marker; raw error: {e}"
+            );
+            classified_corruption = true;
+            break;
+        }
+    }
+    assert!(
+        classified_corruption,
+        "shredded tree must produce at least one corrupt-node read error"
+    );
+}
+
+#[tokio::test]
+async fn missing_mst_node_classifies_as_repo_corruption() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let any = open_store(dir.path());
+    let (root, entries) = build_repo(&any).await;
+
+    let empty_dir = tempfile::tempdir().expect("empty tempdir");
+    let empty = open_store(empty_dir.path());
+    let mst = Mst::load(Arc::new(empty.clone()), root, None);
+
+    let err = mst
+        .get(&entries[0].0)
+        .await
+        .expect_err("loading a root absent from the store must error");
+
+    assert!(
+        ApiError::from_mst_error("audit", &err).is_repo_corruption(),
+        "a missing MST node must classify as repairable so self-heal triggers; raw error: {err}"
+    );
+    assert!(
+        ApiError::detail_is_repo_corruption(&format!("{err:#}")),
+        "missing-node error must carry a repairable marker; raw error: {err}"
+    );
+}
+
+#[tokio::test]
+async fn leaf_block_corruption_is_not_repaired_by_structural_repair() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let any = open_store(dir.path());
+    let (root, entries) = build_repo(&any).await;
+
+    let (_, rec_cid) = &entries[0];
+    assert!(
+        any.get(rec_cid).await.expect("read leaf").is_some(),
+        "leaf must be readable before corruption"
+    );
+
+    let target = rec_cid.to_bytes();
+    assert!(
+        corrupt_block_with_cid(&dir.path().join("data"), &target),
+        "must locate the record leaf block to corrupt"
+    );
+
+    let read_err = any
+        .get(rec_cid)
+        .await
+        .expect_err("corrupt leaf must fail to read");
+    assert!(
+        ApiError::detail_is_repo_corruption(&read_err.to_string()),
+        "corrupt leaf read error must carry the marker; raw error: {read_err}"
+    );
+
+    let outcome = any
+        .repair_structure(&entries, root)
+        .await
+        .expect("structural repair must succeed");
+    assert_eq!(
+        outcome.nodes_repaired, 0,
+        "structural repair only touches MST nodes, so a leaf-only corruption yields zero repairs"
+    );
+
+    assert!(
+        any.get(rec_cid).await.is_err(),
+        "leaf corruption is NOT healed by structural repair"
+    );
+}

crates/tranquil-pds/tests/mst_repair_durability.rs

@@ -0,0 +1,152 @@
+use std::sync::Arc;
+
+use cid::Cid;
+use jacquard_repo::mst::Mst;
+use jacquard_repo::storage::BlockStore;
+use tranquil_pds::repo::AnyBlockStore;
+use tranquil_store::blockstore::{BlockStoreConfig, GroupCommitConfig, TranquilBlockStore};
+
+const RECORD_COUNT: usize = 300;
+
+fn open_store(dir: &std::path::Path) -> AnyBlockStore {
+    let cfg = BlockStoreConfig {
+        data_dir: dir.join("data"),
+        index_dir: dir.join("index"),
+        max_file_size: 64 * 1024,
+        group_commit: GroupCommitConfig::default(),
+        shard_count: 1,
+    };
+    AnyBlockStore::TranquilStore(TranquilBlockStore::open(cfg).expect("open block store"))
+}
+
+async fn build_repo(any: &AnyBlockStore) -> (Cid, Vec<(String, Cid)>) {
+    let mut mst = Mst::new(Arc::new(any.clone()));
+    let mut entries: Vec<(String, Cid)> = Vec::with_capacity(RECORD_COUNT);
+    for i in 0..RECORD_COUNT {
+        let key = format!("app.bsky.feed.post/{i:0>6}");
+        let body = format!("record body number {i}").into_bytes();
+        let cid = any.put(&body).await.expect("put record");
+        mst.add_mut(&key, cid).await.expect("mst add");
+        entries.push((key, cid));
+    }
+    let data_root = mst.persist().await.expect("persist mst");
+    (data_root, entries)
+}
+
+fn shred_data_files(data_dir: &std::path::Path) {
+    let mut shredded = false;
+    for entry in std::fs::read_dir(data_dir).expect("read data dir") {
+        let path = entry.expect("dir entry").path();
+        if path.extension().and_then(|e| e.to_str()) != Some("tqb") {
+            continue;
+        }
+        let mut bytes = std::fs::read(&path).expect("read data file");
+        let mut off = 5usize;
+        while off + 48 < bytes.len() {
+            bytes[off..off + 48].iter_mut().for_each(|b| *b = 0xFF);
+            off += 192;
+            shredded = true;
+        }
+        std::fs::write(&path, &bytes).expect("write corrupted data file");
+    }
+    assert!(shredded, "no .tqb data file was corrupted");
+}
+
+async fn walk_all(
+    any: &AnyBlockStore,
+    root: Cid,
+    entries: &[(String, Cid)],
+) -> Result<usize, String> {
+    let mst = Mst::load(Arc::new(any.clone()), root, None);
+    let mut resolved = 0usize;
+    for (key, expected) in entries {
+        match mst.get(key).await {
+            Ok(Some(cid)) if cid == *expected => resolved += 1,
+            Ok(Some(cid)) => {
+                return Err(format!("{key}: resolved to {cid} != expected {expected}"));
+            }
+            Ok(None) => return Err(format!("{key}: missing")),
+            Err(e) => return Err(format!("{key}: read error {e}")),
+        }
+    }
+    Ok(resolved)
+}
+
+fn copy_dir_recursive(src: &std::path::Path, dst: &std::path::Path) {
+    std::fs::create_dir_all(dst).expect("create dst dir");
+    for entry in std::fs::read_dir(src).expect("read src dir") {
+        let entry = entry.expect("dir entry");
+        let from = entry.path();
+        let to = dst.join(entry.file_name());
+        if from.is_dir() {
+            copy_dir_recursive(&from, &to);
+        } else {
+            std::fs::copy(&from, &to).expect("copy file");
+        }
+    }
+}
+
+#[tokio::test]
+async fn pristine_repo_survives_copy_and_reopen() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let any = open_store(dir.path());
+    let (data_root, entries) = build_repo(&any).await;
+    assert_eq!(
+        walk_all(&any, data_root, &entries).await.expect("pristine"),
+        RECORD_COUNT
+    );
+
+    let snap = tempfile::tempdir().expect("snap tempdir");
+    copy_dir_recursive(dir.path(), snap.path());
+
+    let reopened = open_store(snap.path());
+    assert_eq!(
+        walk_all(&reopened, data_root, &entries)
+            .await
+            .expect("pristine repo must survive copy+reopen"),
+        RECORD_COUNT
+    );
+}
+
+#[tokio::test]
+async fn repair_survives_crash_and_reopen() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let any = open_store(dir.path());
+
+    let (data_root, entries) = build_repo(&any).await;
+    assert_eq!(
+        walk_all(&any, data_root, &entries).await.expect("pristine"),
+        RECORD_COUNT
+    );
+
+    shred_data_files(&dir.path().join("data"));
+    assert!(
+        walk_all(&any, data_root, &entries).await.is_err(),
+        "corruption must break the walk"
+    );
+
+    let outcome = any
+        .repair_structure(&entries, data_root)
+        .await
+        .expect("repair_structure");
+    assert!(outcome.nodes_repaired > 0, "repair must rewrite a node");
+
+    assert_eq!(
+        walk_all(&any, data_root, &entries)
+            .await
+            .expect("in-process walk after repair"),
+        RECORD_COUNT
+    );
+
+    let snap = tempfile::tempdir().expect("snap tempdir");
+    copy_dir_recursive(dir.path(), snap.path());
+
+    let reopened = open_store(snap.path());
+    assert_eq!(
+        walk_all(&reopened, data_root, &entries)
+            .await
+            .expect("every key must resolve after crash-recovery of a repair"),
+        RECORD_COUNT,
+        "repair did not survive reopen"
+    );
+}

crates/tranquil-pds/tests/mst_self_heal_e2e.rs

@@ -0,0 +1,151 @@
+mod common;
+
+use cid::Cid;
+use common::{base_url, client, create_account_and_login, store_data_dir};
+use reqwest::StatusCode;
+use serde_json::{Value, json};
+use std::str::FromStr;
+use tranquil_store::blockstore::{BLOCK_HEADER_SIZE, CID_SIZE};
+
+#[ctor::ctor]
+fn force_store_backend() {
+    unsafe {
+        std::env::set_var("TRANQUIL_TEST_BACKEND", "store");
+    }
+}
+
+const COLLECTION: &str = "app.bsky.feed.post";
+
+fn post_record(i: usize) -> Value {
+    json!({
+        "$type": COLLECTION,
+        "text": format!("self-heal record {i}"),
+        "createdAt": "2024-01-01T00:00:00.000Z"
+    })
+}
+
+async fn apply_creates(token: &str, did: &str, start: usize, count: usize) {
+    let writes: Vec<Value> = (start..start + count)
+        .map(|i| {
+            json!({
+                "$type": "com.atproto.repo.applyWrites#create",
+                "collection": COLLECTION,
+                "rkey": format!("selfheal{i:05}"),
+                "value": post_record(i)
+            })
+        })
+        .collect();
+    let res = client()
+        .post(format!(
+            "{}/xrpc/com.atproto.repo.applyWrites",
+            base_url().await
+        ))
+        .bearer_auth(token)
+        .json(&json!({ "repo": did, "validate": false, "writes": writes }))
+        .send()
+        .await
+        .expect("applyWrites send");
+    assert_eq!(
+        res.status(),
+        StatusCode::OK,
+        "applyWrites failed: {:?}",
+        res.text().await
+    );
+}
+
+async fn latest_commit_cid(did: &str) -> Cid {
+    let res = client()
+        .get(format!(
+            "{}/xrpc/com.atproto.sync.getLatestCommit?did={did}",
+            base_url().await
+        ))
+        .send()
+        .await
+        .expect("getLatestCommit send");
+    assert_eq!(res.status(), StatusCode::OK, "getLatestCommit failed");
+    let body: Value = res.json().await.expect("getLatestCommit json");
+    Cid::from_str(body["cid"].as_str().expect("commit cid")).expect("parse commit cid")
+}
+
+fn collect_tqb(dir: &std::path::Path, out: &mut Vec<std::path::PathBuf>) {
+    let Ok(entries) = std::fs::read_dir(dir) else {
+        return;
+    };
+    for entry in entries.flatten() {
+        let path = entry.path();
+        if path.is_dir() {
+            collect_tqb(&path, out);
+        } else if path.extension().and_then(|e| e.to_str()) == Some("tqb") {
+            out.push(path);
+        }
+    }
+}
+
+fn corrupt_every_block_except(data_dir: &std::path::Path, keep: &[u8]) -> usize {
+    let mut corrupted = 0usize;
+    let mut files = Vec::new();
+    collect_tqb(data_dir, &mut files);
+    for path in files {
+        let mut bytes = std::fs::read(&path).expect("read tqb");
+        let mut pos = BLOCK_HEADER_SIZE;
+        while pos + CID_SIZE + 4 <= bytes.len() {
+            let cid = &bytes[pos..pos + CID_SIZE];
+            let len = u32::from_le_bytes(
+                bytes[pos + CID_SIZE..pos + CID_SIZE + 4]
+                    .try_into()
+                    .unwrap(),
+            ) as usize;
+            let data_start = pos + CID_SIZE + 4;
+            let rec_end = data_start + len + 4;
+            if rec_end > bytes.len() {
+                break;
+            }
+            if cid != keep && len > 0 {
+                bytes[data_start] ^= 0xFF;
+                corrupted += 1;
+            }
+            pos = rec_end;
+        }
+        std::fs::write(&path, &bytes).expect("write corrupted tqb");
+    }
+    corrupted
+}
+
+#[tokio::test]
+async fn write_self_heals_after_mst_node_corruption() {
+    let client = client();
+    let (token, did) = create_account_and_login(&client).await;
+
+    apply_creates(&token, &did, 0, 150).await;
+    apply_creates(&token, &did, 150, 150).await;
+
+    let commit_cid = latest_commit_cid(&did).await;
+    let commit_bytes = commit_cid.to_bytes();
+
+    let data_dir = store_data_dir().expect("store backend data dir");
+    let corrupted = corrupt_every_block_except(&data_dir, &commit_bytes);
+    assert!(corrupted > 0, "expected to corrupt committed blocks");
+
+    let res = client
+        .post(format!(
+            "{}/xrpc/com.atproto.repo.createRecord",
+            base_url().await
+        ))
+        .bearer_auth(&token)
+        .json(&json!({
+            "repo": did,
+            "collection": COLLECTION,
+            "validate": false,
+            "record": post_record(9999)
+        }))
+        .send()
+        .await
+        .expect("createRecord send");
+
+    assert_eq!(
+        res.status(),
+        StatusCode::OK,
+        "write should self-heal corrupted MST and succeed: {:?}",
+        res.text().await
+    );
+}

crates/tranquil-pds/tests/mst_structural_repair.rs

@@ -0,0 +1,108 @@
+use std::sync::Arc;
+
+use cid::Cid;
+use jacquard_repo::mst::Mst;
+use jacquard_repo::storage::BlockStore;
+use tranquil_pds::repo::AnyBlockStore;
+use tranquil_store::blockstore::{BlockStoreConfig, GroupCommitConfig, TranquilBlockStore};
+
+const RECORD_COUNT: usize = 300;
+
+fn open_store(dir: &std::path::Path) -> AnyBlockStore {
+    let cfg = BlockStoreConfig {
+        data_dir: dir.join("data"),
+        index_dir: dir.join("index"),
+        max_file_size: 64 * 1024,
+        group_commit: GroupCommitConfig::default(),
+        shard_count: 1,
+    };
+    AnyBlockStore::TranquilStore(TranquilBlockStore::open(cfg).expect("open block store"))
+}
+
+async fn build_repo(any: &AnyBlockStore) -> (Cid, Vec<(String, Cid)>) {
+    let mut mst = Mst::new(Arc::new(any.clone()));
+    let mut entries: Vec<(String, Cid)> = Vec::with_capacity(RECORD_COUNT);
+    for i in 0..RECORD_COUNT {
+        let key = format!("app.bsky.feed.post/{i:0>6}");
+        let body = format!("record body number {i}").into_bytes();
+        let cid = any.put(&body).await.expect("put record");
+        mst.add_mut(&key, cid).await.expect("mst add");
+        entries.push((key, cid));
+    }
+    let data_root = mst.persist().await.expect("persist mst");
+    (data_root, entries)
+}
+
+fn shred_data_files(data_dir: &std::path::Path) {
+    let mut shredded = false;
+    for entry in std::fs::read_dir(data_dir).expect("read data dir") {
+        let path = entry.expect("dir entry").path();
+        if path.extension().and_then(|e| e.to_str()) != Some("tqb") {
+            continue;
+        }
+        let mut bytes = std::fs::read(&path).expect("read data file");
+        let mut off = 5usize;
+        while off + 48 < bytes.len() {
+            bytes[off..off + 48].iter_mut().for_each(|b| *b = 0xFF);
+            off += 192;
+            shredded = true;
+        }
+        std::fs::write(&path, &bytes).expect("write corrupted data file");
+    }
+    assert!(shredded, "no .tqb data file was corrupted");
+}
+
+async fn walk_all(
+    any: &AnyBlockStore,
+    root: Cid,
+    entries: &[(String, Cid)],
+) -> Result<usize, String> {
+    let mst = Mst::load(Arc::new(any.clone()), root, None);
+    let mut resolved = 0usize;
+    for (key, expected) in entries {
+        match mst.get(key).await {
+            Ok(Some(cid)) if cid == *expected => resolved += 1,
+            Ok(Some(cid)) => {
+                return Err(format!("{key}: resolved to {cid} != expected {expected}"));
+            }
+            Ok(None) => return Err(format!("{key}: missing")),
+            Err(e) => return Err(format!("{key}: read error {e}")),
+        }
+    }
+    Ok(resolved)
+}
+
+#[tokio::test]
+async fn repair_restores_mst_after_node_corruption() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let any = open_store(dir.path());
+
+    let (data_root, entries) = build_repo(&any).await;
+
+    let resolved = walk_all(&any, data_root, &entries)
+        .await
+        .expect("pristine tree must resolve every key");
+    assert_eq!(resolved, RECORD_COUNT);
+
+    shred_data_files(&dir.path().join("data"));
+
+    let broken = walk_all(&any, data_root, &entries).await;
+    assert!(
+        broken.is_err(),
+        "corruption must break the MST walk, got {broken:?}"
+    );
+
+    let outcome = any
+        .repair_structure(&entries, data_root)
+        .await
+        .expect("repair_structure");
+    assert!(
+        outcome.nodes_repaired > 0,
+        "repair must rewrite at least one node, got {outcome:?}"
+    );
+
+    let resolved = walk_all(&any, data_root, &entries)
+        .await
+        .expect("every key must resolve after repair");
+    assert_eq!(resolved, RECORD_COUNT);
+}