From 22f82489d5bca90547f2bb8bc47f1eee02bd90e3 Mon Sep 17 00:00:00 2001 From: Lewis Date: Sun, 31 May 2026 13:24:03 +0300 Subject: [PATCH] test(pds): e2e & durability coverage for MST self-heal Lewis: May this revision serve well! --- crates/tranquil-pds/Cargo.toml | 1 + .../tests/car_export_detects_corruption.rs | 76 +++++++ crates/tranquil-pds/tests/common/mod.rs | 10 +- crates/tranquil-pds/tests/mst_repair_audit.rs | 185 ++++++++++++++++++ .../tests/mst_repair_durability.rs | 152 ++++++++++++++ .../tranquil-pds/tests/mst_self_heal_e2e.rs | 151 ++++++++++++++ .../tests/mst_structural_repair.rs | 108 ++++++++++ 7 files changed, 681 insertions(+), 2 deletions(-) create mode 100644 crates/tranquil-pds/tests/car_export_detects_corruption.rs create mode 100644 crates/tranquil-pds/tests/mst_repair_audit.rs create mode 100644 crates/tranquil-pds/tests/mst_repair_durability.rs create mode 100644 crates/tranquil-pds/tests/mst_self_heal_e2e.rs create mode 100644 crates/tranquil-pds/tests/mst_structural_repair.rs diff --git a/crates/tranquil-pds/Cargo.toml b/crates/tranquil-pds/Cargo.toml index f9dea76..12a7550 100644 --- a/crates/tranquil-pds/Cargo.toml +++ b/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 } diff --git a/crates/tranquil-pds/tests/car_export_detects_corruption.rs b/crates/tranquil-pds/tests/car_export_detects_corruption.rs new file mode 100644 index 0000000..34edf3f --- /dev/null +++ b/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}" + ); +} diff --git a/crates/tranquil-pds/tests/common/mod.rs b/crates/tranquil-pds/tests/common/mod.rs index 92ef5ef..fe114fb 100644 --- a/crates/tranquil-pds/tests/common/mod.rs +++ b/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 { + 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")) } diff --git a/crates/tranquil-pds/tests/mst_repair_audit.rs b/crates/tranquil-pds/tests/mst_repair_audit.rs new file mode 100644 index 0000000..2828423 --- /dev/null +++ b/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" + ); +} diff --git a/crates/tranquil-pds/tests/mst_repair_durability.rs b/crates/tranquil-pds/tests/mst_repair_durability.rs new file mode 100644 index 0000000..7c35bdd --- /dev/null +++ b/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 { + 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" + ); +} diff --git a/crates/tranquil-pds/tests/mst_self_heal_e2e.rs b/crates/tranquil-pds/tests/mst_self_heal_e2e.rs new file mode 100644 index 0000000..8704a63 --- /dev/null +++ b/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 = (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) { + 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 + ); +} diff --git a/crates/tranquil-pds/tests/mst_structural_repair.rs b/crates/tranquil-pds/tests/mst_structural_repair.rs new file mode 100644 index 0000000..8d92b8a --- /dev/null +++ b/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 { + 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); +}