repo: the pg side of MST structural repair

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

crates/tranquil-pds/src/repo/mod.rs

@@ -44,10 +44,26 @@ impl AnyBlockStore {
         expected_root: Cid,
     ) -> Result<RepairOutcome, RepoError> {
         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<Cid> = 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
             }

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, &current_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::<Result<Vec<Cid>, _>>()
+            .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<Cid> = 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::<Vec<_>>(),
+                "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<T, F, Fut>(

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<Cid>),
+    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::<Vec<_>>()
+                    .join(", ")
+            ),
+            Self::Source(e) => write!(f, "{e:#}"),
+        }
+    }
+}
+
+impl std::error::Error for RepoCarError {}
+
+impl From<anyhow::Error> 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<Vec<u8>> {
+) -> Result<Vec<u8>, RepoCarError> {
     let block_cids_bytes = collect_current_repo_blocks(block_store, head_cid).await?;
     let block_cids: Vec<Cid> = 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::<Vec<_>>()
-                    .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<Vec<u8>> {
+) -> Result<Vec<u8>, RepoCarError> {
     use std::str::FromStr;
 
     let repo_root_cid_str: String = repo_repo

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());
+}

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

@@ -30,6 +30,7 @@ static TEST_TEMP_DIR: OnceLock<PathBuf> = OnceLock::new();
 static CLUSTER: OnceLock<Vec<ServerInstance>> = OnceLock::new();
 static TEST_REPOS: OnceLock<Arc<tranquil_db::PostgresRepositories>> = OnceLock::new();
 static TEST_BLOCK_STORE: OnceLock<tranquil_pds::repo::AnyBlockStore> = OnceLock::new();
+static TEST_APP_STATE: OnceLock<AppState> = 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,

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<serde_json::Value> = (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"
+    );
+}

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::{

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<Vec<(CidBytes, Vec<u8>)>, RepoError> {
+) -> Result<Vec<(Cid, Bytes)>, 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<Vec<(CidBytes, Vec<u8>)>, RepoError> {
+    rebuild_mst_nodes(&entries, expected_root)
+        .await?
         .into_iter()
         .map(|(cid, bytes)| Ok((cid_to_bytes(&cid)?, bytes.to_vec())))
         .collect()

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);