diff --git a/crates/tranquil-pds/src/repo/mod.rs b/crates/tranquil-pds/src/repo/mod.rs index c259a57..3001e68 100644 --- a/crates/tranquil-pds/src/repo/mod.rs +++ b/crates/tranquil-pds/src/repo/mod.rs @@ -44,10 +44,26 @@ impl AnyBlockStore { expected_root: Cid, ) -> Result { match self { - Self::Postgres(_) => Ok(RepairOutcome { - nodes_total: 0, - nodes_repaired: 0, - }), + Self::Postgres(s) => { + let nodes = + tranquil_store::blockstore::rebuild_mst_nodes(entries, expected_root).await?; + let nodes_total = nodes.len(); + let cids: Vec = nodes.iter().map(|(cid, _)| *cid).collect(); + let present = s.get_many(&cids).await?; + let missing: Vec<(Cid, Bytes)> = nodes + .into_iter() + .zip(present) + .filter_map(|((cid, bytes), found)| found.is_none().then_some((cid, bytes))) + .collect(); + let nodes_repaired = missing.len() as u64; + if !missing.is_empty() { + s.put_many(missing).await?; + } + Ok(RepairOutcome { + nodes_total, + nodes_repaired, + }) + } Self::TranquilStore(s) => { tranquil_store::blockstore::rebuild_and_repair_mst(s, entries, expected_root).await } diff --git a/crates/tranquil-pds/src/repo_ops.rs b/crates/tranquil-pds/src/repo_ops.rs index 6bc0d90..8b3d913 100644 --- a/crates/tranquil-pds/src/repo_ops.rs +++ b/crates/tranquil-pds/src/repo_ops.rs @@ -259,12 +259,12 @@ pub async fn repair_repo_structure( ApiError::InternalError(None) })? .ok_or_else(|| ApiError::InternalError(Some("Commit block not found".into())))?; - let data_root = Commit::from_cbor(&commit_bytes) - .map_err(|e| { - error!("repair: failed to parse commit: {}", e); - ApiError::InternalError(None) - })? - .data; + let commit = Commit::from_cbor(&commit_bytes).map_err(|e| { + error!("repair: failed to parse commit: {}", e); + ApiError::InternalError(None) + })?; + let data_root = commit.data; + let repo_rev = commit.rev().to_string(); let records = state .repos @@ -290,14 +290,71 @@ pub async fn repair_repo_structure( "repair: rebuilding full MST from record set" ); - state + let outcome = state .block_store .repair_structure(&entries, data_root) .await .map_err(|e| { error!("repair: structural repair failed: {}", e); ApiError::InternalError(Some("Structural repair failed".into())) - }) + })?; + + if outcome.nodes_repaired > 0 { + let block_cids = + crate::scheduled::collect_current_repo_blocks(&state.block_store, ¤t_root_cid) + .await + .map_err(|e| { + error!("repair: re-walk for user_blocks backfill failed: {}", e); + ApiError::InternalError(None) + })?; + + let cids = block_cids + .iter() + .map(|bytes| Cid::try_from(bytes.as_slice())) + .collect::, _>>() + .map_err(|e| { + error!("repair: unparseable CID in repaired DAG walk: {e}"); + ApiError::InternalError(None) + })?; + let present = state.block_store.get_many(&cids).await.map_err(|e| { + error!("repair: presence check during user_blocks backfill failed: {e}"); + ApiError::InternalError(None) + })?; + let missing: Vec = cids + .iter() + .zip(present) + .filter_map(|(cid, found)| found.is_none().then_some(*cid)) + .collect(); + if !missing.is_empty() { + error!( + user_id = %user_id, + missing = missing.len(), + sample = ?missing.iter().take(5).map(|c| c.to_string()).collect::>(), + "repair: unrecoverable leaf data loss after structural repair" + ); + return Err(ApiError::InternalError(Some(format!( + "unrecoverable leaf data loss: {} record block(s) missing after structural repair", + missing.len() + )))); + } + + state + .repos + .repo + .insert_user_blocks(user_id, &block_cids, &repo_rev) + .await + .map_err(|e| { + error!("repair: user_blocks backfill failed: {}", e); + ApiError::InternalError(None) + })?; + warn!( + user_id = %user_id, + blocks = block_cids.len(), + "repair: backfilled user_blocks from repaired DAG" + ); + } + + Ok(outcome) } pub async fn with_repair_retry( diff --git a/crates/tranquil-pds/src/scheduled.rs b/crates/tranquil-pds/src/scheduled.rs index e32ccea..9e5240a 100644 --- a/crates/tranquil-pds/src/scheduled.rs +++ b/crates/tranquil-pds/src/scheduled.rs @@ -730,10 +730,53 @@ async fn delete_account_data( const CAR_BLOCK_BATCH_SIZE: usize = 500; +#[derive(Debug)] +pub enum RepoCarError { + MissingBlocks(Vec), + Source(anyhow::Error), +} + +impl RepoCarError { + pub fn is_repairable(&self) -> bool { + match self { + Self::MissingBlocks(_) => true, + Self::Source(e) => { + crate::api::error::ApiError::detail_is_repo_corruption(&format!("{e:#}")) + } + } + } +} + +impl std::fmt::Display for RepoCarError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::MissingBlocks(cids) => write!( + f, + "repo CAR is incomplete: {} block(s) referenced by the MST are missing from storage. First 5: {}", + cids.len(), + cids.iter() + .take(5) + .map(|c| c.to_string()) + .collect::>() + .join(", ") + ), + Self::Source(e) => write!(f, "{e:#}"), + } + } +} + +impl std::error::Error for RepoCarError {} + +impl From for RepoCarError { + fn from(e: anyhow::Error) -> Self { + Self::Source(e) + } +} + pub async fn generate_repo_car( block_store: &AnyBlockStore, head_cid: &Cid, -) -> anyhow::Result> { +) -> Result, RepoCarError> { let block_cids_bytes = collect_current_repo_blocks(block_store, head_cid).await?; let block_cids: Vec = block_cids_bytes .iter() @@ -760,16 +803,7 @@ pub async fn generate_repo_car( .filter_map(|(cid, block_opt)| block_opt.is_none().then_some(*cid)) .collect(); if !missing.is_empty() { - anyhow::bail!( - "repo CAR is incomplete: {} block(s) referenced by the MST are missing from storage. First 5: {}", - missing.len(), - missing - .iter() - .take(5) - .map(|c| c.to_string()) - .collect::>() - .join(", ") - ); + return Err(RepoCarError::MissingBlocks(missing)); } chunk @@ -803,7 +837,7 @@ pub async fn generate_repo_car_from_user_blocks( block_store: &AnyBlockStore, user_id: uuid::Uuid, _head_cid: &Cid, -) -> anyhow::Result> { +) -> Result, RepoCarError> { use std::str::FromStr; let repo_root_cid_str: String = repo_repo diff --git a/crates/tranquil-pds/tests/car_export_detects_corruption.rs b/crates/tranquil-pds/tests/car_export_detects_corruption.rs index 34edf3f..0625f3f 100644 --- a/crates/tranquil-pds/tests/car_export_detects_corruption.rs +++ b/crates/tranquil-pds/tests/car_export_detects_corruption.rs @@ -3,9 +3,8 @@ 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_pds::scheduled::{RepoCarError, generate_repo_car}; use tranquil_store::blockstore::{BlockStoreConfig, GroupCommitConfig, TranquilBlockStore}; const RECORD_COUNT: usize = 200; @@ -68,9 +67,28 @@ async fn car_export_error_is_classified_as_repo_corruption() { 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}" + err.is_repairable(), + "{err} should classify as repairable so the sync path can self-heal" ); } + +#[tokio::test] +async fn car_export_missing_block_is_repairable() { + let dir = tempfile::tempdir().expect("tempdir"); + let source = open_store(dir.path()); + let root = build_tree(&source).await; + + let pristine = open_store(&dir.path().join("pristine")); + let head_block = source.get(&root).await.expect("read root").expect("root present"); + pristine.put(&head_block).await.expect("seed root only"); + + let err = generate_repo_car(&pristine, &root) + .await + .expect_err("CAR export over a store missing MST children must error"); + assert!( + matches!(err, RepoCarError::MissingBlocks(ref cids) if !cids.is_empty()), + "{err} should surface as MissingBlocks when referenced blocks are absent" + ); + assert!(err.is_repairable()); +} diff --git a/crates/tranquil-pds/tests/common/mod.rs b/crates/tranquil-pds/tests/common/mod.rs index fe114fb..6754726 100644 --- a/crates/tranquil-pds/tests/common/mod.rs +++ b/crates/tranquil-pds/tests/common/mod.rs @@ -30,6 +30,7 @@ static TEST_TEMP_DIR: OnceLock = OnceLock::new(); static CLUSTER: OnceLock> = OnceLock::new(); static TEST_REPOS: OnceLock> = OnceLock::new(); static TEST_BLOCK_STORE: OnceLock = OnceLock::new(); +static TEST_APP_STATE: OnceLock = OnceLock::new(); #[allow(dead_code)] pub fn is_store_backend() -> bool { @@ -586,6 +587,7 @@ async fn spawn_server(config: ServerConfig) -> ServerInstance { if let Some((cache, distributed_rate_limiter)) = config.cache { state = state.with_cache(cache, distributed_rate_limiter); } + TEST_APP_STATE.set(state.clone()).ok(); tranquil_sync::listener::start_sequencer_listener(state.clone()).await; let app = tranquil_pds::app_with_routes( state, @@ -928,6 +930,12 @@ pub async fn get_test_block_store() -> &'static tranquil_pds::repo::AnyBlockStor .expect("TEST_BLOCK_STORE not initialized") } +#[allow(dead_code)] +pub async fn get_test_app_state() -> &'static AppState { + base_url().await; + TEST_APP_STATE.get().expect("TEST_APP_STATE not initialized") +} + #[allow(dead_code)] pub async fn flushed_max_seq( repos: &tranquil_db::PostgresRepositories, diff --git a/crates/tranquil-pds/tests/repair_leaf_loss.rs b/crates/tranquil-pds/tests/repair_leaf_loss.rs new file mode 100644 index 0000000..3bd4ff1 --- /dev/null +++ b/crates/tranquil-pds/tests/repair_leaf_loss.rs @@ -0,0 +1,142 @@ +mod common; +mod helpers; + +use cid::Cid; +use common::*; +use helpers::*; +use jacquard_repo::commit::Commit; +use jacquard_repo::storage::BlockStore; +use serde_json::json; +use std::str::FromStr; +use tranquil_types::Did; + +#[tokio::test] +async fn repair_fails_loud_on_missing_leaf_block() { + let client = client(); + let repos = get_test_repos().await; + let block_store = get_test_block_store().await; + let state = get_test_app_state().await; + + let Some(pg) = block_store.as_postgres() else { + eprintln!( + "repair_fails_loud_on_missing_leaf_block: requires postgres backend, skipping under store backend" + ); + return; + }; + let pool = pg.pool(); + + let (did, jwt) = setup_new_user("repair-leaf-loss").await; + let writes: Vec = (0..6) + .map(|i| { + json!({ + "$type": "com.atproto.repo.applyWrites#create", + "collection": "app.bsky.feed.post", + "rkey": format!("leafloss{i:05}"), + "value": { + "$type": "app.bsky.feed.post", + "text": format!("repair leaf loss {i}"), + "createdAt": "2026-01-01T00:00:00.000Z" + } + }) + }) + .collect(); + let res = client + .post(format!( + "{}/xrpc/com.atproto.repo.applyWrites", + base_url().await + )) + .bearer_auth(&jwt) + .json(&json!({ "repo": did, "validate": false, "writes": writes })) + .send() + .await + .expect("applyWrites send"); + assert_eq!( + res.status(), + reqwest::StatusCode::OK, + "applyWrites failed: {:?}", + res.text().await + ); + + let user_id = repos + .user + .get_id_by_did(&Did::new(did.clone()).unwrap()) + .await + .expect("DB error") + .expect("user not found"); + + let root_str = repos + .repo + .get_repo_root_cid_by_user_id(user_id) + .await + .expect("DB error") + .expect("repo root not found"); + let commit_cid = Cid::from_str(&root_str).expect("parse commit cid"); + let commit_bytes = block_store + .get(&commit_cid) + .await + .expect("read commit") + .expect("commit present"); + let mst_root_cid = Commit::from_cbor(&commit_bytes).expect("parse commit").data; + + let records = repos + .repo + .get_all_records(user_id) + .await + .expect("get_all_records"); + assert!(!records.is_empty(), "repo must contain records"); + let leaf_cid = Cid::from_str(records[0].record_cid.as_str()).expect("parse leaf cid"); + assert!( + block_store.get(&leaf_cid).await.expect("read leaf").is_some(), + "leaf must be present before corruption" + ); + + repos + .repo + .delete_user_blocks(user_id, &[leaf_cid.to_bytes()]) + .await + .expect("clear leaf user_blocks row"); + + sqlx::query("DELETE FROM blocks WHERE cid = $1") + .bind(mst_root_cid.to_bytes()) + .execute(pool) + .await + .expect("delete mst root node block"); + sqlx::query("DELETE FROM blocks WHERE cid = $1") + .bind(leaf_cid.to_bytes()) + .execute(pool) + .await + .expect("delete leaf record block"); + + assert!( + block_store.get(&mst_root_cid).await.expect("read").is_none(), + "mst root node must be gone to force a structural repair" + ); + assert!( + block_store.get(&leaf_cid).await.expect("read").is_none(), + "leaf block must be gone to simulate data loss" + ); + + let err = tranquil_pds::repo_ops::repair_repo_structure(state, user_id) + .await + .expect_err("repair must fail loud when a leaf block is unrecoverable"); + let detail = format!("{err:?}"); + assert!( + detail.contains("leaf data loss"), + "expected an unrecoverable-leaf-loss error, got: {detail}" + ); + + assert!( + block_store.get(&mst_root_cid).await.expect("read").is_some(), + "structural repair must still re-insert the regenerable MST node" + ); + + let recorded = repos + .repo + .get_user_block_cids_since_rev(user_id, "") + .await + .expect("read user_blocks"); + assert!( + !recorded.contains(&leaf_cid.to_bytes()), + "missing leaf must not be phantom-inserted into user_blocks" + ); +} diff --git a/crates/tranquil-store/src/blockstore/mod.rs b/crates/tranquil-store/src/blockstore/mod.rs index f86cc81..a612285 100644 --- a/crates/tranquil-store/src/blockstore/mod.rs +++ b/crates/tranquil-store/src/blockstore/mod.rs @@ -27,7 +27,7 @@ pub use hint::{ }; pub use manager::{CachedHandle, DEFAULT_MAX_FILE_SIZE, DataFileManager}; pub use reader::{BLOCK_CORRUPTION_MARKER, BlockStoreReader, ReadError}; -pub use repair::{RepairOutcome, rebuild_and_repair_mst}; +pub use repair::{RepairOutcome, rebuild_and_repair_mst, rebuild_mst_nodes}; pub use store::QuiesceGuard; pub use store::{BlockStoreConfig, DEFAULT_SHARD_COUNT, OpenRetryPolicy, TranquilBlockStore}; pub use types::{ diff --git a/crates/tranquil-store/src/blockstore/repair.rs b/crates/tranquil-store/src/blockstore/repair.rs index d05351e..5f589f5 100644 --- a/crates/tranquil-store/src/blockstore/repair.rs +++ b/crates/tranquil-store/src/blockstore/repair.rs @@ -1,5 +1,6 @@ use std::sync::Arc; +use bytes::Bytes; use cid::Cid; use jacquard_repo::error::RepoError; use jacquard_repo::mst::Mst; @@ -23,13 +24,13 @@ fn rebuild_err(context: &str, e: impl std::fmt::Display) -> RepoError { RepoError::storage(std::io::Error::other(format!("{context}: {e}"))) } -async fn rebuild_node_blocks( - entries: Vec<(String, Cid)>, +pub async fn rebuild_mst_nodes( + entries: &[(String, Cid)], expected_root: Cid, -) -> Result)>, RepoError> { +) -> Result, RepoError> { let scratch = Arc::new(MemoryBlockStore::new()); let mut mst = Mst::new(scratch); - for (key, cid) in &entries { + for (key, cid) in entries { mst.add_mut(key.as_str(), *cid) .await .map_err(|e| rebuild_err("mst rebuild add", e))?; @@ -49,7 +50,15 @@ async fn rebuild_node_blocks( ))); } - blocks + Ok(blocks.into_iter().collect()) +} + +async fn rebuild_node_blocks( + entries: Vec<(String, Cid)>, + expected_root: Cid, +) -> Result)>, RepoError> { + rebuild_mst_nodes(&entries, expected_root) + .await? .into_iter() .map(|(cid, bytes)| Ok((cid_to_bytes(&cid)?, bytes.to_vec()))) .collect() diff --git a/crates/tranquil-sync/src/repo.rs b/crates/tranquil-sync/src/repo.rs index 99e4af6..fa1231f 100644 --- a/crates/tranquil-sync/src/repo.rs +++ b/crates/tranquil-sync/src/repo.rs @@ -163,7 +163,7 @@ pub async fn get_repo( { Ok(bytes) => bytes, Err(e) => { - if ApiError::detail_is_repo_corruption(&format!("{e:#}")) { + if e.is_repairable() { tranquil_pds::repo_ops::schedule_repo_repair(&state, account.user_id); } error!("Failed to generate repo CAR: {}", e);