From 1e2311f8fc6bf0ffbd5c1c6d32cc129b6e89d7ec Mon Sep 17 00:00:00 2001 From: Lewis Date: Sun, 19 Jul 2026 17:19:22 +0300 Subject: [PATCH] repo: keep user_blocks equal to the reachable block set Lewis: May this revision serve well! --- crates/tranquil-api/src/repo/import.rs | 47 ++- crates/tranquil-api/src/repo/record/write.rs | 24 +- crates/tranquil-pds/src/repo_ops.rs | 284 +++++++++++++++--- crates/tranquil-pds/src/scheduled.rs | 54 +++- .../tests/gc_compaction_restart.rs | 4 +- 5 files changed, 348 insertions(+), 65 deletions(-) diff --git a/crates/tranquil-api/src/repo/import.rs b/crates/tranquil-api/src/repo/import.rs index f5f281a..c939704 100644 --- a/crates/tranquil-api/src/repo/import.rs +++ b/crates/tranquil-api/src/repo/import.rs @@ -233,7 +233,7 @@ pub async fn import_repo( ApiError::InternalError(None) })?; let new_root_cid_link = CidLink::from(&new_root_cid); - let new_rev_tid = tranquil_pds::types::Tid::from(new_rev_str.clone()); + let new_rev_tid = tranquil_pds::types::Tid::from(new_rev.clone()); state .repos .repo @@ -243,17 +243,40 @@ pub async fn import_repo( error!("Failed to update repo root: {:?}", e); ApiError::InternalError(None) })?; - let mut all_block_cids: Vec> = blocks.keys().map(|c| c.to_bytes()).collect(); - all_block_cids.push(new_root_cid.to_bytes()); - state - .repos - .repo - .insert_user_blocks(user_id, &all_block_cids, &new_rev_tid) - .await - .map_err(|e| { - error!("Failed to insert user_blocks: {:?}", e); - ApiError::InternalError(None) - })?; + match tranquil_pds::scheduled::collect_current_repo_blocks( + &state.block_store, + &new_root_cid, + ) + .await + { + Ok(reachable) => { + if !reachable.is_complete() { + error!( + unreadable = reachable.unreadable, + "scheduling a structural repair because the imported repo walk could \ + not read every block" + ); + tranquil_pds::repo_ops::schedule_repo_repair(&state, user_id); + } + state + .repos + .repo + .insert_user_blocks(user_id, &reachable.block_cids, &new_rev_tid) + .await + .map_err(|e| { + error!("Failed to insert user_blocks: {:?}", e); + ApiError::InternalError(None) + })?; + } + Err(e) => { + error!( + "Failed to walk the imported repo: {:?}. The root is already updated and \ + a scheduled structural repair will rebuild user_blocks", + e + ); + tranquil_pds::repo_ops::schedule_repo_repair(&state, user_id); + } + } let new_root_str = new_root_cid.to_string(); info!( "Created new commit for imported repo: cid={}, rev={}", diff --git a/crates/tranquil-api/src/repo/record/write.rs b/crates/tranquil-api/src/repo/record/write.rs index 65ef14d..e5966f7 100644 --- a/crates/tranquil-api/src/repo/record/write.rs +++ b/crates/tranquil-api/src/repo/record/write.rs @@ -7,6 +7,7 @@ use serde::{Deserialize, Serialize}; use serde_json::json; use std::borrow::Cow; use std::str::FromStr; +use tracing::warn; use tranquil_pds::api::error::{ApiError, DbResultExt}; use tranquil_pds::auth::{ Active, Auth, AuthSource, RepoScopeAction, ScopeVerified, VerifyScope, require_not_migrated, @@ -181,8 +182,15 @@ async fn create_record_inner( else { continue; }; - let conflict_rkey = Rkey::from(conflict_rkey_str.to_string()); - let conflict_collection = Nsid::from(conflict_col_str.to_string()); + let (Ok(conflict_rkey), Ok(conflict_collection)) = + (Rkey::new(conflict_rkey_str), Nsid::new(conflict_col_str)) + else { + warn!( + uri = %conflict_uri, + "skipping a backlink conflict whose stored URI doesn't parse" + ); + continue; + }; let conflict_key = format!("{}/{}", conflict_collection, conflict_rkey); let prev_cid = match mst.get(&conflict_key).await { @@ -373,11 +381,8 @@ async fn put_record_inner( let record_ipld = tranquil_pds::util::json_to_ipld(&input.record); let record_bytes = serde_ipld_dagcbor::to_vec(&record_ipld) .map_err(|_| ApiError::InvalidRecord("Failed to serialize record".into()))?; - let record_cid = ctx - .tracking_store - .put(&record_bytes) - .await - .map_err(|_| ApiError::InternalError(Some("Failed to save record block".into())))?; + let record_cid = jacquard_repo::mst::util::compute_cid(&record_bytes) + .map_err(|_| ApiError::InvalidRecord("Failed to compute record CID".into()))?; if existing_cid == Some(record_cid) { return Ok(PutRecordOutput { @@ -388,6 +393,11 @@ async fn put_record_inner( }); } + ctx.tracking_store + .put(&record_bytes) + .await + .map_err(|_| ApiError::InternalError(Some("Failed to save record block".into())))?; + let record_uri = AtUri::from_parts(did, &input.collection, &input.rkey); let (new_mst, op, is_update, backlinks_to_remove) = match existing_cid { Some(prev_cid) => { diff --git a/crates/tranquil-pds/src/repo_ops.rs b/crates/tranquil-pds/src/repo_ops.rs index 0186a37..5ce9487 100644 --- a/crates/tranquil-pds/src/repo_ops.rs +++ b/crates/tranquil-pds/src/repo_ops.rs @@ -14,7 +14,7 @@ use jacquard_repo::mst::{Mst, VerifiedWriteOp}; use jacquard_repo::storage::BlockStore; use k256::ecdsa::SigningKey; use serde_json::{Value, json}; -use std::collections::{BTreeMap, HashMap, HashSet}; +use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet}; use std::str::FromStr; use std::sync::{Arc, LazyLock}; use std::time::{Duration, Instant}; @@ -262,7 +262,7 @@ pub async fn repair_repo_structure( ApiError::InternalError(None) })?; let data_root = commit.data; - let repo_rev = crate::types::Tid::from(commit.rev().to_string()); + let repo_rev = crate::types::Tid::from(commit.rev().clone()); let records = state .repos @@ -323,30 +323,67 @@ pub async fn repair_repo_structure( missing.len() )))); } + } - 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 walk = 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 recorded = state + .repos + .repo + .get_user_block_cids_since_rev(user_id, None) + .await + .map_err(|e| { + error!("repair: reading recorded user_blocks set failed: {}", e); + ApiError::InternalError(None) + })?; + let reachable: HashSet<&[u8]> = walk.block_cids.iter().map(Vec::as_slice).collect(); + let stale: Vec> = match walk.is_complete() { + true => recorded + .into_iter() + .filter(|cid| !reachable.contains(cid.as_slice())) + .collect(), + false => { + error!( + user_id = %user_id, + unreadable = walk.unreadable, + "repair: repo walk couldn't read every block, so its complement isn't the \ + unreachable set. Backfilling without pruning." + ); + Vec::new() + } + }; + + state + .repos + .repo + .insert_user_blocks(user_id, &walk.block_cids, &repo_rev) + .await + .map_err(|e| { + error!("repair: user_blocks backfill failed: {}", e); + ApiError::InternalError(None) + })?; + if !stale.is_empty() { state .repos .repo - .insert_user_blocks(user_id, &block_cids, &repo_rev) + .delete_user_blocks(user_id, &stale) .await .map_err(|e| { - error!("repair: user_blocks backfill failed: {}", e); + error!("repair: pruning unreachable user_blocks rows failed: {}", e); ApiError::InternalError(None) })?; - warn!( - user_id = %user_id, - blocks = block_cids.len(), - "repair: backfilled user_blocks from repaired DAG" - ); } + warn!( + user_id = %user_id, + blocks = walk.block_cids.len(), + pruned = stale.len(), + "repair: rewrote user_blocks from current DAG" + ); Ok(outcome) } @@ -460,6 +497,69 @@ pub fn schedule_repo_repair(state: &AppState, user_id: Uuid) { }); } +async fn leaves_referenced_elsewhere( + state: &AppState, + user_id: Uuid, + dropped_leaves: &BTreeSet, + ops: &[RecordOp], +) -> BTreeSet { + if dropped_leaves.is_empty() { + return BTreeSet::new(); + } + let by_link: BTreeMap = dropped_leaves + .iter() + .map(|cid| (crate::types::CidLink::from_cid(cid), *cid)) + .collect(); + let cids: Vec = by_link.keys().cloned().collect(); + // Because the commit only runs later on, records table still shows + // this write's pre-write state at this point + // so excluding the keys that this write touches is the thing that + // keeps a deleted record from counting as a reference to its own block. + let touched: Vec<(&Nsid, &Rkey)> = ops.iter().map(RecordOp::collection_rkey).collect(); + // The two failure directions aren't symmetric at all + // because a user_blocks row for a block that's already gone is only noise + // that the scheduled repair clears... + // While a *missing* row for a block that's still reachable + // makes `getRepo` with a since-cursor give an incomplete tree + // and thus any doubt at all about the lookup ought to mean keeping every row. + let keep_everything = |reason: &str| { + error!( + user_id = %user_id, + "{reason}. Keeping the user_blocks row for every record block this write removed. \ + Their refcounts still drop, so a row may remain after its block is gone until the \ + scheduled structural repair rewrites the set." + ); + schedule_repo_repair(state, user_id); + dropped_leaves.clone() + }; + match state + .repos + .repo + .referenced_record_cids(user_id, &cids, &touched) + .await + { + Ok(referenced) => referenced + .iter() + .map(|link| by_link.get(link).copied()) + .collect::>>() + .unwrap_or_else(|| { + keep_everything("record cid reference lookup returned a cid that wasn't queried") + }), + Err(e) => keep_everything(&format!("record cid reference lookup failed: {e:?}")), + } +} + +pub async fn reachable_tree_cids( + mst: &Mst, +) -> Result, jacquard_repo::error::RepoError> { + let nodes = mst.collect_node_cids().await?; + let leaves = mst.leaves().await?; + Ok(nodes + .into_iter() + .chain(leaves.into_iter().map(|(_, cid)| cid)) + .collect()) +} + pub async fn finalize_repo_write( state: &AppState, ctx: RepoWriteContext, @@ -472,7 +572,6 @@ pub async fn finalize_repo_write( .map_err(|e| ApiError::from_mst_error("MST persist", &e))?; let written_bytes = ctx.tracking_store.take_written_blocks(); - let new_tree_cids: Vec = written_bytes.keys().copied().collect(); let storage_for_proof = Arc::new(ctx.tracking_store.clone()); let original_settled = Mst::load(storage_for_proof.clone(), ctx.prev_data_cid, None); @@ -545,27 +644,124 @@ pub async fn finalize_repo_write( } } - let obsolete_cids = match original_settled.diff(&new_settled).await { + let replaced_refs: Vec = params + .ops + .iter() + .filter_map(RecordOp::replaced_cid) + .collect(); + + let (new_tree_cids, new_mst_bytes, dropped) = match original_settled.diff(&new_settled).await { Ok(diff) => { - let mut obsolete: Vec = - Vec::with_capacity(1 + diff.removed_mst_blocks.len() + diff.removed_cids.len()); - obsolete.push(ctx.current_root_cid); - obsolete.extend(diff.removed_mst_blocks); - obsolete.extend(diff.removed_cids); - obsolete + // The moment that a leaf CID leaves its old position (leaves, get it) + // the diff will report it as removed + // even where the new tree goes on using that same block somewhere else + // so filtering the removed leaves against `live` will stop this write + // from dropping a block the new tree still needs. + // The diffing minuses the unchanged nodes from `removed_mst_blocks` itself + // and the same filter only covers a node which matches the new root. + let live: BTreeSet = diff + .new_mst_blocks + .keys() + .copied() + .chain(diff.new_leaf_cids.iter().copied()) + .chain(std::iter::once(new_mst_root)) + .collect(); + let dropped_leaves: BTreeSet = diff + .removed_cids + .iter() + .copied() + .filter(|cid| !live.contains(cid)) + .collect(); + let still_referenced = + leaves_referenced_elsewhere(state, params.user_id, &dropped_leaves, ¶ms.ops) + .await; + let removed_mst_blocks: Vec = diff + .removed_mst_blocks + .iter() + .copied() + .filter(|cid| !live.contains(cid)) + .collect(); + + // These sets have distinct purposes: + // - `unreachable` deletes user_blocks rows + // and so has to leave out any block another record still references + // - `decrements` drops one refcount for every ref this write released + // including the replaced record CIDs that other records + // continue on referencing + // + // Neither set includes the previous data root + // since the diff reports subtree nodes and never a root of its own. + // So that block keeps its refcount and its user_blocks row + // until the leaked-refcount sweep reclaims it. + let dropped = DroppedBlocks { + unreachable: std::iter::once(ctx.current_root_cid) + .chain(removed_mst_blocks.iter().copied()) + .chain( + dropped_leaves + .iter() + .copied() + .filter(|cid| !still_referenced.contains(cid)), + ) + .collect(), + decrements: std::iter::once(ctx.current_root_cid) + .chain(removed_mst_blocks) + .chain(replaced_refs) + .collect(), + }; + (live.into_iter().collect(), diff.new_mst_blocks, dropped) } Err(e) => { error!( - "MST diff failed during finalize_repo_write: {e}. \ - Proceeding with commit CID only; leaked blocks \ - will be reclaimed by reachability GC." + user_id = %params.user_id, + "MST diff failed during finalize_repo_write: {e}. Walking the new MST insetad \ + and scheduling a structural repair." ); - vec![ctx.current_root_cid] + schedule_repo_repair(state, params.user_id); + let dropped = DroppedBlocks { + unreachable: vec![ctx.current_root_cid], + decrements: std::iter::once(ctx.current_root_cid) + .chain(replaced_refs) + .collect(), + }; + match reachable_tree_cids(&new_settled).await { + Ok(cids) => (cids.into_iter().collect(), BTreeMap::new(), dropped), + Err(walk_err) => { + error!( + "MST walk after diff failure also failed: {walk_err}. \ + Recording only blocks this write observed and the commit CID. \ + user_blocks stays incomplete until a structural repair rewrites the tree." + ); + ( + written_bytes.keys().copied().collect(), + BTreeMap::new(), + dropped, + ) + } + } } }; + // Under jacquard-repo 0.9 that tranquil pins for now, + // `Mst::persist` skips any subtree that's already in store + // so `written_bytes` under-reports the new node set + // and `new_mst_blocks` is full + // where each of these blocks still needs a `put` to take a refcount + // even though its bytes are already on disk. + let recovered: Vec<(Cid, Bytes)> = new_mst_bytes + .iter() + .filter(|(cid, _)| !written_bytes.contains_key(*cid)) + .map(|(cid, bytes)| (*cid, bytes.clone())) + .collect(); + if !recovered.is_empty() { + state.block_store.put_many(recovered).await.map_err(|e| { + error!("failed to reference MST blocks recovered from the diff: {e}"); + ApiError::InternalError(None) + })?; + } + let mut block_bytes = written_bytes; block_bytes.extend(relevant); + block_bytes.extend(new_mst_bytes); let result = commit_and_log( state, @@ -580,7 +776,7 @@ pub async fn finalize_repo_write( block_bytes, new_tree_cids, blobs: params.blob_cids, - obsolete_cids, + dropped, backlinks_to_add: params.backlinks_to_add, backlinks_to_remove: params.backlinks_to_remove, }, @@ -649,6 +845,13 @@ pub enum RecordOp { } impl RecordOp { + pub fn replaced_cid(&self) -> Option { + match self { + Self::Create { .. } => None, + Self::Update { prev, .. } | Self::Delete { prev, .. } => Some(*prev.as_cid()), + } + } + pub fn collection_rkey(&self) -> (&Nsid, &Rkey) { match self { Self::Create { @@ -669,6 +872,11 @@ pub struct CommitResult { pub rev: crate::types::Tid, } +pub struct DroppedBlocks { + pub unreachable: Vec, + pub decrements: Vec, +} + pub struct CommitParams<'a> { pub did: &'a Did, pub user_id: Uuid, @@ -680,7 +888,7 @@ pub struct CommitParams<'a> { pub block_bytes: std::collections::HashMap, pub new_tree_cids: Vec, pub blobs: &'a [crate::types::CidLink], - pub obsolete_cids: Vec, + pub dropped: DroppedBlocks, pub backlinks_to_add: Vec, pub backlinks_to_remove: Vec, } @@ -705,7 +913,7 @@ pub async fn commit_and_log( mut block_bytes, new_tree_cids, blobs, - obsolete_cids, + dropped, backlinks_to_add, backlinks_to_remove, } = params; @@ -726,7 +934,7 @@ pub async fn commit_and_log( let signing_key = SigningKey::from_slice(&key_bytes).map_err(|e| CommitError::InvalidKey(e.to_string()))?; let rev = Tid::now(LimitedU32::MIN); - let rev_str = rev.to_string(); + let repo_rev = crate::types::Tid::from(rev.clone()); let (new_commit_bytes, _sig) = create_signed_commit(did, new_mst_root, &rev, current_root_cid, &signing_key)?; let new_root_cid = @@ -751,7 +959,7 @@ pub async fn commit_and_log( .map(|c| c.to_bytes()) .collect(); - let obsolete_bytes: Vec> = obsolete_cids.iter().map(|c| c.to_bytes()).collect(); + let obsolete_bytes: Vec> = dropped.unreachable.iter().map(|c| c.to_bytes()).collect(); let final_ops: HashMap<(&Nsid, &Rkey), &RecordOp> = ops.iter().map(|op| (op.collection_rkey(), op)).collect(); @@ -864,7 +1072,7 @@ pub async fn commit_and_log( blobs: Some(blobs.to_vec()), blocks: Some(inline_blocks), prev_data_cid: prev_data_cid.map(crate::types::CidLink::from), - rev: Some(crate::types::Tid::from(rev_str.clone())), + rev: Some(repo_rev.clone()), }; let input = ApplyCommitInput { @@ -872,7 +1080,7 @@ pub async fn commit_and_log( did: did.clone(), expected_root_cid: current_root_cid.map(crate::types::CidLink::from), new_root_cid: crate::types::CidLink::from(new_root_cid), - new_rev: crate::types::Tid::from(rev_str.clone()), + new_rev: repo_rev.clone(), new_block_cids: all_block_cids, obsolete_block_cids: obsolete_bytes, record_upserts, @@ -895,7 +1103,7 @@ pub async fn commit_and_log( let apply_result = (|| { let bs = state.block_store.clone(); - let decrements = obsolete_cids.clone(); + let decrements = dropped.decrements.clone(); async move { bs.decrement_refs(&decrements).await } }) .retry( @@ -907,7 +1115,7 @@ pub async fn commit_and_log( .await; if let Err(e) = apply_result { - let leaked: Vec = obsolete_cids.iter().map(Cid::to_string).collect(); + let leaked: Vec = dropped.decrements.iter().map(Cid::to_string).collect(); tracing::error!( error = %e, user_id = %user_id, @@ -920,7 +1128,7 @@ pub async fn commit_and_log( Ok(CommitResult { commit_cid: new_root_cid, - rev: crate::types::Tid::from(rev_str), + rev: repo_rev, }) } pub async fn create_record_internal( diff --git a/crates/tranquil-pds/src/scheduled.rs b/crates/tranquil-pds/src/scheduled.rs index cd05fc1..438e3c6 100644 --- a/crates/tranquil-pds/src/scheduled.rs +++ b/crates/tranquil-pds/src/scheduled.rs @@ -37,7 +37,7 @@ async fn process_repo_rev( } }; let commit = Commit::from_cbor(&block).map_err(|_| user_id)?; - let rev = crate::types::Tid::from(commit.rev().to_string()); + let rev = crate::types::Tid::from(commit.rev().clone()); repo_repo .update_repo_rev(user_id, &rev) .await @@ -98,14 +98,22 @@ async fn process_user_blocks( repo_rev: Option, ) -> Result<(uuid::Uuid, usize), uuid::Uuid> { let root_cid = Cid::from_str(&repo_root_cid).map_err(|_| user_id)?; - let block_cids = collect_current_repo_blocks(block_store, &root_cid) + let walk = collect_current_repo_blocks(block_store, &root_cid) .await .map_err(|_| user_id)?; + if !walk.is_complete() { + error!( + user_id = %user_id, + unreadable = walk.unreadable, + "user_blocks backfill recorded an incomplete set becuase the walk couldn't read every block" + ); + } + let block_cids = walk.block_cids; if block_cids.is_empty() { return Err(user_id); } let count = block_cids.len(); - let rev = repo_rev.unwrap_or_else(|| crate::types::Tid::from("0".to_string())); + let rev = repo_rev.unwrap_or_else(crate::types::Tid::earliest); repo_repo .insert_user_blocks(user_id, &block_cids, &rev) .await @@ -162,13 +170,35 @@ pub async fn backfill_user_blocks(repo_repo: Arc, block_stor info!(success, failed, "Completed user_blocks backfill"); } +// When a walk couldn't read every block, +// what comes back is a subset of the reachable set +// ...and the complement of a subset is *not* the unreachable set. +// So callers check `is_complete` before they prune +// or one unreadable block is enough to delete live rows. +// +// The count only covers a block that the store reports as missing or corrupt, +// whereas a block whose bytes decode as neither commit or MST node ends +// traversal without counting +// so `is_complete` is necessary for a prune and not sufficient. +pub struct RepoWalk { + pub block_cids: Vec>, + pub unreadable: usize, +} + +impl RepoWalk { + pub fn is_complete(&self) -> bool { + self.unreadable == 0 + } +} + pub async fn collect_current_repo_blocks( block_store: &AnyBlockStore, head_cid: &Cid, -) -> anyhow::Result>> { +) -> anyhow::Result { let mut block_cids: Vec> = Vec::new(); let mut to_visit = vec![*head_cid]; let mut visited = std::collections::HashSet::new(); + let mut unreadable = 0usize; while let Some(cid) = to_visit.pop() { if visited.contains(&cid) { @@ -179,9 +209,14 @@ pub async fn collect_current_repo_blocks( let block = match block_store.get(&cid).await { Ok(Some(b)) => b, - Ok(None) => continue, + Ok(None) => { + warn!(cid = %cid, "block missing during repo walk, so its subtree stays unvisited"); + unreadable += 1; + continue; + } Err(e) if crate::api::error::ApiError::detail_is_repo_corruption(&format!("{e:#}")) => { warn!(cid = %cid, error = %format!("{e:#}"), "skipping corrupt block during repo walk"); + unreadable += 1; continue; } Err(e) => anyhow::bail!("Failed to get block {}: {:?}", cid, e), @@ -215,7 +250,10 @@ pub async fn collect_current_repo_blocks( } } - Ok(block_cids) + Ok(RepoWalk { + block_cids, + unreadable, + }) } async fn process_record_blobs( @@ -771,7 +809,9 @@ pub async fn generate_repo_car( block_store: &AnyBlockStore, head_cid: &Cid, ) -> Result, RepoCarError> { - let block_cids_bytes = collect_current_repo_blocks(block_store, head_cid).await?; + let block_cids_bytes = collect_current_repo_blocks(block_store, head_cid) + .await? + .block_cids; let block_cids: Vec = block_cids_bytes .iter() .filter_map(|b| match Cid::try_from(b.as_slice()) { diff --git a/crates/tranquil-pds/tests/gc_compaction_restart.rs b/crates/tranquil-pds/tests/gc_compaction_restart.rs index 1c65228..5199535 100644 --- a/crates/tranquil-pds/tests/gc_compaction_restart.rs +++ b/crates/tranquil-pds/tests/gc_compaction_restart.rs @@ -107,9 +107,11 @@ async fn mst_blocks_survive_full_store_reopen() { let head_cid = cid::Cid::try_from(repo_root_str.as_str()).expect("invalid cid"); - let car_blocks = tranquil_pds::scheduled::collect_current_repo_blocks(block_store, &head_cid) + let walk = tranquil_pds::scheduled::collect_current_repo_blocks(block_store, &head_cid) .await .expect("collect blocks"); + assert!(walk.is_complete(), "repo walk read every block it reached"); + let car_blocks = walk.block_cids; let block_count_before = car_blocks.len();