From cee483e358c49a77efd0daba9dbb5c7f39f797af Mon Sep 17 00:00:00 2001 From: Lewis Date: Sun, 31 May 2026 13:24:03 +0300 Subject: [PATCH] feat(pds): selfhealing repo writing by detecting corruption & retrying Lewis: May this revision serve well! --- crates/tranquil-api/src/repo/record/batch.rs | 100 +++--- crates/tranquil-api/src/repo/record/delete.rs | 61 ++-- crates/tranquil-api/src/repo/record/write.rs | 111 ++++--- crates/tranquil-pds/src/api/error.rs | 32 +- crates/tranquil-pds/src/repo/mod.rs | 18 +- crates/tranquil-pds/src/repo_ops.rs | 303 +++++++++++++++++- crates/tranquil-sync/src/repo.rs | 16 +- 7 files changed, 523 insertions(+), 118 deletions(-) diff --git a/crates/tranquil-api/src/repo/record/batch.rs b/crates/tranquil-api/src/repo/record/batch.rs index 68a2b82..1af2dee 100644 --- a/crates/tranquil-api/src/repo/record/batch.rs +++ b/crates/tranquil-api/src/repo/record/batch.rs @@ -14,8 +14,8 @@ use tranquil_pds::auth::{ }; use tranquil_pds::repo::TrackingBlockStore; use tranquil_pds::repo_ops::{ - FinalizeParams, RecordOp, begin_repo_write, extract_backlinks, extract_blob_cids, - finalize_repo_write, + CommitResult, FinalizeParams, RecordOp, begin_repo_write, extract_backlinks, extract_blob_cids, + finalize_repo_write, with_repair_retry, }; use tranquil_pds::state::AppState; use tranquil_pds::types::{AtIdentifier, AtUri, Did, Nsid, Rkey}; @@ -74,7 +74,7 @@ async fn process_single_write( if mst .get(&key) .await - .map_err(|e| ApiError::InternalError(Some(format!("Failed to read MST: {e}"))))? + .map_err(|e| ApiError::from_mst_error("read MST for applyWrites create", &e))? .is_some() { return Err(ApiError::InvalidRequest(format!( @@ -92,7 +92,7 @@ async fn process_single_write( let new_mst = mst .add(&key, record_cid) .await - .map_err(|_| ApiError::InternalError(Some("Failed to add to MST".into())))?; + .map_err(|e| ApiError::from_mst_error("add record to MST", &e))?; let uri = AtUri::from_parts(did, collection, &rkey); backlinks_to_add.extend(extract_backlinks(&uri, value)); results.push(WriteResult::CreateResult { @@ -138,9 +138,7 @@ async fn process_single_write( let prev_record_cid = mst .get(&key) .await - .map_err(|e| { - ApiError::InternalError(Some(format!("Failed to read prev record: {}", e))) - })? + .map_err(|e| ApiError::from_mst_error("read update target from MST", &e))? .ok_or_else(|| { ApiError::InvalidRequest("Update target record does not exist".into()) })?; @@ -155,7 +153,7 @@ async fn process_single_write( let new_mst = mst .update(&key, record_cid) .await - .map_err(|_| ApiError::InternalError(Some("Failed to update MST".into())))?; + .map_err(|e| ApiError::from_mst_error("update record in MST", &e))?; let uri = AtUri::from_parts(did, collection, rkey); backlinks_to_remove.push(uri.clone()); backlinks_to_add.extend(extract_backlinks(&uri, value)); @@ -184,16 +182,14 @@ async fn process_single_write( let prev_record_cid = mst .get(&key) .await - .map_err(|e| { - ApiError::InternalError(Some(format!("Failed to read prev record: {}", e))) - })? + .map_err(|e| ApiError::from_mst_error("read delete target from MST", &e))? .ok_or_else(|| { ApiError::InvalidRequest("Delete target record does not exist".into()) })?; let new_mst = mst .delete(&key) .await - .map_err(|_| ApiError::InternalError(Some("Failed to delete from MST".into())))?; + .map_err(|e| ApiError::from_mst_error("delete record from MST", &e))?; backlinks_to_remove.push(AtUri::from_parts(did, collection, rkey)); results.push(WriteResult::DeleteResult {}); ops.push(RecordOp::Delete { @@ -236,6 +232,45 @@ async fn process_writes( .await } +async fn execute_apply_writes( + state: &AppState, + user_id: uuid::Uuid, + did: &Did, + input: &ApplyWritesInput, + controller_did: Option<&Did>, + write_summary: Option, +) -> Result<(CommitResult, Vec), ApiError> { + let (ctx, mst) = begin_repo_write(state, user_id, input.swap_commit.as_deref()).await?; + + let WriteAccumulator { + mst: final_mst, + results, + ops, + all_blob_cids, + backlinks_to_add, + backlinks_to_remove, + } = process_writes(&input.writes, mst, did, input.validate, &ctx.tracking_store).await?; + + let commit_result = finalize_repo_write( + state, + ctx, + final_mst, + FinalizeParams { + did, + user_id, + controller_did, + delegation_detail: write_summary, + ops, + blob_cids: &all_blob_cids, + backlinks_to_add, + backlinks_to_remove, + }, + ) + .await?; + + Ok((commit_result, results)) +} + #[derive(Deserialize)] #[serde(tag = "$type")] pub enum WriteOp { @@ -350,24 +385,6 @@ pub async fn apply_writes( .log_db_err("fetching user for batch write")? .ok_or(ApiError::InternalError(Some("User not found".into())))?; - let (ctx, mst) = begin_repo_write(&state, user_id, input.swap_commit.as_deref()).await?; - - let WriteAccumulator { - mst: final_mst, - results, - ops, - all_blob_cids, - backlinks_to_add, - backlinks_to_remove, - } = process_writes( - &input.writes, - mst, - &did, - input.validate, - &ctx.tracking_store, - ) - .await?; - let write_summary: Option = controller_did.as_ref().map(|_| { let writes: Vec = input .writes @@ -401,21 +418,16 @@ pub async fn apply_writes( }) }); - let commit_result = finalize_repo_write( - &state, - ctx, - final_mst, - FinalizeParams { - did: &did, + let (commit_result, results) = with_repair_retry(&state, user_id, || { + execute_apply_writes( + &state, user_id, - controller_did: controller_did.as_ref(), - delegation_detail: write_summary, - ops, - blob_cids: &all_blob_cids, - backlinks_to_add, - backlinks_to_remove, - }, - ) + &did, + &input, + controller_did.as_ref(), + write_summary.clone(), + ) + }) .await?; Ok(Json(ApplyWritesOutput { diff --git a/crates/tranquil-api/src/repo/record/delete.rs b/crates/tranquil-api/src/repo/record/delete.rs index 00b39e4..3134bff 100644 --- a/crates/tranquil-api/src/repo/record/delete.rs +++ b/crates/tranquil-api/src/repo/record/delete.rs @@ -4,13 +4,15 @@ use cid::Cid; use serde::{Deserialize, Serialize}; use serde_json::json; use std::str::FromStr; -use tracing::error; use tranquil_pds::api::error::ApiError; use tranquil_pds::auth::{Active, Auth, VerifyScope}; use tranquil_pds::cid_types::RecordCid; -use tranquil_pds::repo_ops::{FinalizeParams, RecordOp, begin_repo_write, finalize_repo_write}; +use tranquil_pds::repo_ops::{ + FinalizeParams, RecordOp, begin_repo_write, finalize_repo_write, with_repair_retry, +}; use tranquil_pds::state::AppState; -use tranquil_pds::types::{AtIdentifier, AtUri, Nsid, Rkey}; +use tranquil_pds::types::{AtIdentifier, AtUri, Did, Nsid, Rkey}; +use uuid::Uuid; #[derive(Deserialize)] pub struct DeleteRecordInput { @@ -41,13 +43,30 @@ pub async fn delete_record( let user_id = repo_auth.user_id; let controller_did = repo_auth.controller_did; - let (ctx, mst) = begin_repo_write(&state, user_id, input.swap_commit.as_deref()).await?; + let out = with_repair_retry(&state, user_id, || { + delete_record_inner(&state, &did, user_id, controller_did.as_ref(), &input) + }) + .await?; + Ok(Json(out)) +} + +async fn delete_record_inner( + state: &AppState, + did: &Did, + user_id: Uuid, + controller_did: Option<&Did>, + input: &DeleteRecordInput, +) -> Result { + let (ctx, mst) = begin_repo_write(state, user_id, input.swap_commit.as_deref()).await?; let key = format!("{}/{}", input.collection, input.rkey); if let Some(swap_record_str) = &input.swap_record { let expected_cid = Cid::from_str(swap_record_str).ok(); - let actual_cid = mst.get(&key).await.ok().flatten(); + let actual_cid = mst + .get(&key) + .await + .map_err(|e| ApiError::from_mst_error("read swap target from MST", &e))?; if expected_cid != actual_cid { return Err(ApiError::InvalidSwap(Some( "Record has been modified or does not exist".into(), @@ -55,18 +74,18 @@ pub async fn delete_record( } } - let prev_record_cid = mst.get(&key).await.map_err(|e| { - error!("Failed to read prev record from MST: {}", e); - ApiError::InternalError(Some("Failed to read MST".into())) - })?; + let prev_record_cid = mst + .get(&key) + .await + .map_err(|e| ApiError::from_mst_error("read prev record from MST", &e))?; let Some(prev_record_cid) = prev_record_cid else { - return Ok(Json(DeleteRecordOutput { commit: None })); + return Ok(DeleteRecordOutput { commit: None }); }; - let new_mst = mst.delete(&key).await.map_err(|e| { - error!("Failed to delete from MST: {}", e); - ApiError::InternalError(Some("Failed to delete from MST".into())) - })?; + let new_mst = mst + .delete(&key) + .await + .map_err(|e| ApiError::from_mst_error("delete record from MST", &e))?; let op = RecordOp::Delete { collection: input.collection.clone(), @@ -74,17 +93,17 @@ pub async fn delete_record( prev: RecordCid::from(prev_record_cid), }; - let deleted_uri = AtUri::from_parts(&did, &input.collection, &input.rkey); + let deleted_uri = AtUri::from_parts(did, &input.collection, &input.rkey); let commit_result = finalize_repo_write( - &state, + state, ctx, new_mst, FinalizeParams { - did: &did, + did, user_id, - controller_did: controller_did.as_ref(), - delegation_detail: controller_did.as_ref().map(|_| { + controller_did, + delegation_detail: controller_did.map(|_| { json!({ "action": "delete", "collection": input.collection, @@ -99,10 +118,10 @@ pub async fn delete_record( ) .await?; - Ok(Json(DeleteRecordOutput { + Ok(DeleteRecordOutput { commit: Some(CommitInfo { cid: commit_result.commit_cid.to_string(), rev: commit_result.rev, }), - })) + }) } diff --git a/crates/tranquil-api/src/repo/record/write.rs b/crates/tranquil-api/src/repo/record/write.rs index 702f188..b275fb4 100644 --- a/crates/tranquil-api/src/repo/record/write.rs +++ b/crates/tranquil-api/src/repo/record/write.rs @@ -7,7 +7,6 @@ use serde::{Deserialize, Serialize}; use serde_json::json; use std::borrow::Cow; use std::str::FromStr; -use tracing::error; use tranquil_pds::api::error::{ApiError, DbResultExt}; use tranquil_pds::auth::{ Active, Auth, AuthSource, RepoScopeAction, ScopeVerified, VerifyScope, require_not_migrated, @@ -15,7 +14,7 @@ use tranquil_pds::auth::{ }; use tranquil_pds::repo_ops::{ FinalizeParams, RecordOp, begin_repo_write, extract_backlinks, extract_blob_cids, - finalize_repo_write, + finalize_repo_write, with_repair_retry, }; use tranquil_pds::state::AppState; use tranquil_pds::types::{AtIdentifier, AtUri, Did, Nsid, Rkey}; @@ -130,7 +129,21 @@ pub async fn create_record( let user_id = repo_auth.user_id; let controller_did = repo_auth.controller_did; - let (ctx, mut mst) = begin_repo_write(&state, user_id, input.swap_commit.as_deref()).await?; + let out = with_repair_retry(&state, user_id, || { + create_record_inner(&state, &did, user_id, controller_did.as_ref(), &input) + }) + .await?; + Ok(Json(out)) +} + +async fn create_record_inner( + state: &AppState, + did: &Did, + user_id: Uuid, + controller_did: Option<&Did>, + input: &CreateRecordInput, +) -> Result { + let (ctx, mut mst) = begin_repo_write(state, user_id, input.swap_commit.as_deref()).await?; let validation_status = if input.validate.should_skip() { None @@ -146,12 +159,12 @@ pub async fn create_record( ) }; - let rkey = input.rkey.unwrap_or_else(Rkey::generate); + let rkey = input.rkey.clone().unwrap_or_else(Rkey::generate); let mut ops: Vec = Vec::new(); let mut conflict_uris_to_cleanup: Vec = Vec::new(); if !input.validate.should_skip() { - let record_uri = AtUri::from_parts(&did, &input.collection, &rkey); + let record_uri = AtUri::from_parts(did, &input.collection, &rkey); let backlinks = extract_backlinks(&record_uri, &input.record); if !backlinks.is_empty() { @@ -176,24 +189,18 @@ pub async fn create_record( Ok(Some(cid)) => cid, Ok(None) => continue, Err(e) => { - error!( - "Failed to read conflict record from MST {}: {:?}", - conflict_uri, e - ); - return Err(ApiError::InternalError(Some( - "Failed to read conflicting record from MST".into(), - ))); + return Err(ApiError::from_mst_error( + &format!("read conflict record from MST {conflict_uri}"), + &e, + )); } }; mst = mst.delete(&conflict_key).await.map_err(|e| { - error!( - "Failed to delete conflict from MST {}: {:?}", - conflict_uri, e - ); - ApiError::InternalError(Some( - "Failed to delete conflicting record from MST".into(), - )) + ApiError::from_mst_error( + &format!("delete conflict from MST {conflict_uri}"), + &e, + ) })?; ops.push(RecordOp::Delete { @@ -210,7 +217,7 @@ pub async fn create_record( if mst .get(&key) .await - .map_err(|e| ApiError::InternalError(Some(format!("Failed to read MST: {e}"))))? + .map_err(|e| ApiError::from_mst_error("read MST for create existence check", &e))? .is_some() { return Err(ApiError::InvalidRequest(format!( @@ -229,7 +236,7 @@ pub async fn create_record( mst = mst .add(&key, record_cid) .await - .map_err(|_| ApiError::InternalError(Some("Failed to add to MST".into())))?; + .map_err(|e| ApiError::from_mst_error("add record to MST", &e))?; ops.push(RecordOp::Create { collection: input.collection.clone(), @@ -239,18 +246,18 @@ pub async fn create_record( let blob_cids = extract_blob_cids(&input.record); - let created_uri = AtUri::from_parts(&did, &input.collection, &rkey); + let created_uri = AtUri::from_parts(did, &input.collection, &rkey); let backlinks_to_add = extract_backlinks(&created_uri, &input.record); let commit_result = finalize_repo_write( - &state, + state, ctx, mst, FinalizeParams { - did: &did, + did, user_id, - controller_did: controller_did.as_ref(), - delegation_detail: controller_did.as_ref().map(|_| { + controller_did, + delegation_detail: controller_did.map(|_| { json!({ "action": "create", "collection": input.collection, @@ -265,7 +272,7 @@ pub async fn create_record( ) .await?; - Ok(Json(CreateRecordOutput { + Ok(CreateRecordOutput { uri: created_uri, cid: record_cid.to_string(), commit: CommitInfo { @@ -273,7 +280,7 @@ pub async fn create_record( rev: commit_result.rev, }, validation_status, - })) + }) } #[derive(Deserialize)] @@ -316,7 +323,21 @@ pub async fn put_record( let user_id = repo_auth.user_id; let controller_did = repo_auth.controller_did; - let (ctx, mst) = begin_repo_write(&state, user_id, input.swap_commit.as_deref()).await?; + let out = with_repair_retry(&state, user_id, || { + put_record_inner(&state, &did, user_id, controller_did.as_ref(), &input) + }) + .await?; + Ok(Json(out)) +} + +async fn put_record_inner( + state: &AppState, + did: &Did, + user_id: Uuid, + controller_did: Option<&Did>, + input: &PutRecordInput, +) -> Result { + let (ctx, mst) = begin_repo_write(state, user_id, input.swap_commit.as_deref()).await?; let validation_status = if input.validate.should_skip() { None @@ -334,9 +355,13 @@ pub async fn put_record( let key = format!("{}/{}", input.collection, input.rkey); + let read_cid = |r: Result, jacquard_repo::error::RepoError>| { + r.map_err(|e| ApiError::from_mst_error("read MST for put", &e)) + }; + if let Some(swap_record_str) = &input.swap_record { let expected_cid = Cid::from_str(swap_record_str).ok(); - let actual_cid = mst.get(&key).await.ok().flatten(); + let actual_cid = read_cid(mst.get(&key).await)?; if expected_cid != actual_cid { return Err(ApiError::InvalidSwap(Some( "Record has been modified or does not exist".into(), @@ -344,7 +369,7 @@ pub async fn put_record( } } - let existing_cid = mst.get(&key).await.ok().flatten(); + let existing_cid = read_cid(mst.get(&key).await)?; 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()))?; @@ -355,21 +380,21 @@ pub async fn put_record( .map_err(|_| ApiError::InternalError(Some("Failed to save record block".into())))?; if existing_cid == Some(record_cid) { - return Ok(Json(PutRecordOutput { - uri: AtUri::from_parts(&did, &input.collection, &input.rkey), + return Ok(PutRecordOutput { + uri: AtUri::from_parts(did, &input.collection, &input.rkey), cid: record_cid.to_string(), commit: None, validation_status, - })); + }); } - let record_uri = AtUri::from_parts(&did, &input.collection, &input.rkey); + 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) => { let new_mst = mst .update(&key, record_cid) .await - .map_err(|_| ApiError::InternalError(Some("Failed to update MST".into())))?; + .map_err(|e| ApiError::from_mst_error("update record in MST", &e))?; let op = RecordOp::Update { collection: input.collection.clone(), rkey: input.rkey.clone(), @@ -382,7 +407,7 @@ pub async fn put_record( let new_mst = mst .add(&key, record_cid) .await - .map_err(|_| ApiError::InternalError(Some("Failed to add to MST".into())))?; + .map_err(|e| ApiError::from_mst_error("add record to MST", &e))?; let op = RecordOp::Create { collection: input.collection.clone(), rkey: input.rkey.clone(), @@ -396,14 +421,14 @@ pub async fn put_record( let backlinks_to_add = extract_backlinks(&record_uri, &input.record); let commit_result = finalize_repo_write( - &state, + state, ctx, new_mst, FinalizeParams { - did: &did, + did, user_id, - controller_did: controller_did.as_ref(), - delegation_detail: controller_did.as_ref().map(|_| { + controller_did, + delegation_detail: controller_did.map(|_| { json!({ "action": if is_update { "update" } else { "create" }, "collection": input.collection, @@ -418,7 +443,7 @@ pub async fn put_record( ) .await?; - Ok(Json(PutRecordOutput { + Ok(PutRecordOutput { uri: record_uri, cid: record_cid.to_string(), commit: Some(CommitInfo { @@ -426,5 +451,5 @@ pub async fn put_record( rev: commit_result.rev, }), validation_status, - })) + }) } diff --git a/crates/tranquil-pds/src/api/error.rs b/crates/tranquil-pds/src/api/error.rs index dc1f284..be9fbde 100644 --- a/crates/tranquil-pds/src/api/error.rs +++ b/crates/tranquil-pds/src/api/error.rs @@ -15,6 +15,7 @@ struct ErrorBody<'a> { #[derive(Debug)] pub enum ApiError { InternalError(Option), + RepoCorruption, AuthenticationRequired, AuthenticationFailed(Option), InvalidRequest(String), @@ -121,10 +122,34 @@ pub enum ApiError { }, } +const MST_NODE_MISSING_MARKER: &str = "MST node not found"; + impl ApiError { + pub fn is_repo_corruption(&self) -> bool { + matches!(self, Self::RepoCorruption) + } + + pub fn detail_is_repo_corruption(detail: &str) -> bool { + detail.contains(tranquil_store::blockstore::BLOCK_CORRUPTION_MARKER) + || detail.contains(MST_NODE_MISSING_MARKER) + } + + pub fn from_mst_error(context: &str, e: &jacquard_repo::error::RepoError) -> Self { + let detail = format!("{e:#}"); + if Self::detail_is_repo_corruption(&detail) { + tracing::warn!("{context}: repairable MST damage: {detail}"); + Self::RepoCorruption + } else { + tracing::error!("{context}: {detail}"); + Self::InternalError(None) + } + } + fn status_code(&self) -> StatusCode { match self { - Self::InternalError(_) | Self::DatabaseError => StatusCode::INTERNAL_SERVER_ERROR, + Self::InternalError(_) | Self::RepoCorruption | Self::DatabaseError => { + StatusCode::INTERNAL_SERVER_ERROR + } Self::UpstreamFailure | Self::UpstreamUnavailable(_) | Self::UpstreamErrorMsg(_) => { StatusCode::BAD_GATEWAY } @@ -223,7 +248,9 @@ impl ApiError { } fn error_name(&self) -> Cow<'static, str> { match self { - Self::InternalError(_) | Self::DatabaseError => Cow::Borrowed("InternalServerError"), + Self::InternalError(_) | Self::RepoCorruption | Self::DatabaseError => { + Cow::Borrowed("InternalServerError") + } Self::UpstreamFailure | Self::UpstreamUnavailable(_) | Self::UpstreamErrorMsg(_) => { Cow::Borrowed("UpstreamError") } @@ -332,6 +359,7 @@ impl ApiError { Self::InternalError(msg) => msg .clone() .unwrap_or_else(|| "Internal Server Error".into()), + Self::RepoCorruption => "Internal Server Error".into(), Self::AuthenticationFailed(msg) => msg .clone() .unwrap_or_else(|| "Authentication failed".into()), diff --git a/crates/tranquil-pds/src/repo/mod.rs b/crates/tranquil-pds/src/repo/mod.rs index 110f081..c259a57 100644 --- a/crates/tranquil-pds/src/repo/mod.rs +++ b/crates/tranquil-pds/src/repo/mod.rs @@ -7,7 +7,7 @@ use cid::Cid; use jacquard_repo::error::RepoError; use jacquard_repo::repo::CommitData; use jacquard_repo::storage::BlockStore; -use tranquil_store::blockstore::TranquilBlockStore; +use tranquil_store::blockstore::{RepairOutcome, TranquilBlockStore}; use tranquil_store::{RealIO, SystemClock}; #[derive(Clone)] @@ -37,6 +37,22 @@ impl AnyBlockStore { Self::TranquilStore(s) => s.decrement_refs(cids).await, } } + + pub async fn repair_structure( + &self, + entries: &[(String, Cid)], + expected_root: Cid, + ) -> Result { + match self { + Self::Postgres(_) => Ok(RepairOutcome { + nodes_total: 0, + nodes_repaired: 0, + }), + Self::TranquilStore(s) => { + tranquil_store::blockstore::rebuild_and_repair_mst(s, entries, expected_root).await + } + } + } } impl BlockStore for AnyBlockStore { diff --git a/crates/tranquil-pds/src/repo_ops.rs b/crates/tranquil-pds/src/repo_ops.rs index b682e85..6bc0d90 100644 --- a/crates/tranquil-pds/src/repo_ops.rs +++ b/crates/tranquil-pds/src/repo_ops.rs @@ -16,7 +16,8 @@ use k256::ecdsa::SigningKey; use serde_json::{Value, json}; use std::collections::{BTreeMap, HashMap, HashSet}; use std::str::FromStr; -use std::sync::Arc; +use std::sync::{Arc, LazyLock}; +use std::time::{Duration, Instant}; use tokio::sync::OwnedMutexGuard; use tracing::{error, warn}; use uuid::Uuid; @@ -230,16 +231,194 @@ pub async fn begin_repo_write( Ok((ctx, mst)) } +pub async fn repair_repo_structure( + state: &AppState, + user_id: Uuid, +) -> Result { + let _write_lock = state.repo_write_locks.lock(user_id).await; + + let root_cid_str = state + .repos + .repo + .get_repo_root_cid_by_user_id(user_id) + .await + .map_err(|e| { + error!("repair: DB error fetching repo root: {}", e); + ApiError::InternalError(None) + })? + .ok_or_else(|| ApiError::InternalError(Some("Repo root not found".into())))?; + let current_root_cid = Cid::from_str(root_cid_str.as_str()) + .map_err(|_| ApiError::InternalError(Some("Invalid repo root CID".into())))?; + + let commit_bytes = state + .block_store + .get(¤t_root_cid) + .await + .map_err(|e| { + error!("repair: failed to load commit block: {}", e); + 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 records = state + .repos + .repo + .get_all_records(user_id) + .await + .map_err(|e| { + error!("repair: get_all_records failed: {}", e); + ApiError::InternalError(None) + })?; + let entries: Vec<(String, Cid)> = records + .into_iter() + .filter_map(|r| { + Cid::from_str(r.record_cid.as_str()) + .ok() + .map(|cid| (format!("{}/{}", r.collection, r.rkey), cid)) + }) + .collect(); + + warn!( + user_id = %user_id, + records = entries.len(), + "repair: rebuilding full MST from record set" + ); + + 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())) + }) +} + +pub async fn with_repair_retry( + state: &AppState, + user_id: Uuid, + mut attempt: F, +) -> Result +where + F: FnMut() -> Fut, + Fut: std::future::Future>, +{ + match attempt().await { + Err(e) if e.is_repo_corruption() => { + warn!( + "structural MST damage during repo write for user {user_id}, repairing and retrying" + ); + match repair_repo_structure(state, user_id).await { + Ok(outcome) if outcome.nodes_repaired > 0 => attempt().await, + Ok(_) => { + warn!( + user_id = %user_id, + "structural repair rewrote no nodes; damage is not in the MST structure, returning original error without retry" + ); + Err(e) + } + Err(repair_err) => { + error!(user_id = %user_id, "structural repair failed: {repair_err:?}"); + Err(e) + } + } + } + other => other, + } +} + +const REPAIR_COOLDOWN: Duration = Duration::from_secs(60); +const REPAIR_NOOP_COOLDOWN: Duration = Duration::from_secs(600); + +struct RepairSlot { + in_flight: bool, + next_allowed: Instant, +} + +struct RepairGuard { + slots: parking_lot::Mutex>, +} + +impl RepairGuard { + fn try_claim(&self, user_id: Uuid, now: Instant) -> bool { + let mut slots = self.slots.lock(); + let slot = slots.entry(user_id).or_insert(RepairSlot { + in_flight: false, + next_allowed: now, + }); + if slot.in_flight || now < slot.next_allowed { + return false; + } + slot.in_flight = true; + true + } + + fn release(&self, user_id: Uuid, now: Instant, cooldown: Duration) { + if let Some(slot) = self.slots.lock().get_mut(&user_id) { + slot.in_flight = false; + slot.next_allowed = now + cooldown; + } + } +} + +static REPAIR_GUARD: LazyLock = LazyLock::new(|| RepairGuard { + slots: parking_lot::Mutex::new(HashMap::new()), +}); + +struct RepairLease { + user_id: Uuid, + cooldown: Duration, +} + +impl Drop for RepairLease { + fn drop(&mut self) { + REPAIR_GUARD.release(self.user_id, Instant::now(), self.cooldown); + } +} + +pub fn schedule_repo_repair(state: &AppState, user_id: Uuid) { + if !REPAIR_GUARD.try_claim(user_id, Instant::now()) { + return; + } + let state = state.clone(); + tokio::spawn(async move { + let mut lease = RepairLease { + user_id, + cooldown: REPAIR_COOLDOWN, + }; + match repair_repo_structure(&state, user_id).await { + Ok(outcome) => { + if outcome.nodes_repaired == 0 { + lease.cooldown = REPAIR_NOOP_COOLDOWN; + } + warn!( + user_id = %user_id, + nodes_repaired = outcome.nodes_repaired, + nodes_total = outcome.nodes_total, + "background MST repair complete" + ); + } + Err(e) => error!(user_id = %user_id, "background MST repair failed: {e:?}"), + } + }); +} + pub async fn finalize_repo_write( state: &AppState, ctx: RepoWriteContext, mst: Mst, params: FinalizeParams<'_>, ) -> Result { - let new_mst_root = mst.persist().await.map_err(|e| { - error!("MST persist failed: {}", e); - ApiError::InternalError(None) - })?; + let new_mst_root = mst + .persist() + .await + .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(); @@ -861,3 +1040,117 @@ pub async fn sequence_genesis_commit( .await .map_err(|e| CommitError::DatabaseError(format!("genesis commit event: {}", e))) } + +#[cfg(test)] +mod repair_guard_tests { + use super::*; + + fn guard() -> RepairGuard { + RepairGuard { + slots: parking_lot::Mutex::new(HashMap::new()), + } + } + + #[test] + fn claim_dedups_in_flight_then_respects_cooldown() { + let g = guard(); + let user = Uuid::from_u128(1); + let t0 = Instant::now(); + + assert!(g.try_claim(user, t0), "first claim must succeed"); + assert!( + !g.try_claim(user, t0), + "second claim while a repair is in flight must be rejected" + ); + + g.release(user, t0, REPAIR_COOLDOWN); + assert!( + !g.try_claim(user, t0 + Duration::from_secs(1)), + "claim within the cooldown window must be rejected" + ); + assert!( + g.try_claim(user, t0 + REPAIR_COOLDOWN + Duration::from_millis(1)), + "claim after the cooldown window must succeed" + ); + } + + #[test] + fn distinct_users_do_not_block_each_other() { + let g = guard(); + let t0 = Instant::now(); + assert!(g.try_claim(Uuid::from_u128(1), t0)); + assert!(g.try_claim(Uuid::from_u128(2), t0)); + } + + #[test] + fn concurrent_claims_for_one_user_admit_exactly_one() { + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::{Arc, Barrier}; + + let g = Arc::new(guard()); + let user = Uuid::from_u128(42); + let now = Instant::now(); + let winners = Arc::new(AtomicUsize::new(0)); + let gate = Arc::new(Barrier::new(32)); + + let handles: Vec<_> = (0..32) + .map(|_| { + let g = Arc::clone(&g); + let winners = Arc::clone(&winners); + let gate = Arc::clone(&gate); + std::thread::spawn(move || { + gate.wait(); + if g.try_claim(user, now) { + winners.fetch_add(1, Ordering::Relaxed); + } + }) + }) + .collect(); + handles + .into_iter() + .for_each(|h| h.join().expect("worker thread panicked")); + + assert_eq!( + winners.load(Ordering::Relaxed), + 1, + "exactly one concurrent claim must win the dedup race" + ); + } + + #[test] + fn release_re_enables_claim_after_cooldown_for_recurring_corruption() { + let g = guard(); + let user = Uuid::from_u128(7); + let t0 = Instant::now(); + + assert!(g.try_claim(user, t0)); + g.release(user, t0, REPAIR_COOLDOWN); + let after_cooldown = t0 + REPAIR_COOLDOWN + Duration::from_millis(1); + assert!( + g.try_claim(user, after_cooldown), + "a fresh corruption after the cooldown must be repairable again" + ); + assert!( + !g.try_claim(user, after_cooldown), + "the re-claimed repair must again dedup while in flight" + ); + } + + #[test] + fn noop_repair_uses_longer_cooldown() { + let g = guard(); + let user = Uuid::from_u128(9); + let t0 = Instant::now(); + + assert!(g.try_claim(user, t0)); + g.release(user, t0, REPAIR_NOOP_COOLDOWN); + assert!( + !g.try_claim(user, t0 + REPAIR_COOLDOWN + Duration::from_millis(1)), + "after a no-op repair the standard cooldown must not re-admit a claim" + ); + assert!( + g.try_claim(user, t0 + REPAIR_NOOP_COOLDOWN + Duration::from_millis(1)), + "after the longer no-op cooldown a fresh claim must be admitted" + ); + } +} diff --git a/crates/tranquil-sync/src/repo.rs b/crates/tranquil-sync/src/repo.rs index f961ae6..99e4af6 100644 --- a/crates/tranquil-sync/src/repo.rs +++ b/crates/tranquil-sync/src/repo.rs @@ -163,6 +163,9 @@ pub async fn get_repo( { Ok(bytes) => bytes, Err(e) => { + if ApiError::detail_is_repo_corruption(&format!("{e:#}")) { + tranquil_pds::repo_ops::schedule_repo_repair(&state, account.user_id); + } error!("Failed to generate repo CAR: {}", e); return ApiError::InternalError(None).into_response(); } @@ -230,6 +233,9 @@ async fn get_repo_since(state: &AppState, did: &Did, head_cid: &Cid, since: &str let blocks = match state.block_store.get_many(chunk).await { Ok(b) => b, Err(e) => { + if ApiError::detail_is_repo_corruption(&format!("{e:#}")) { + tranquil_pds::repo_ops::schedule_repo_repair(state, user_id); + } error!("Block store error in get_repo_since: {:?}", e); return ApiError::InternalError(Some("Failed to get blocks".into())) .into_response(); @@ -301,7 +307,10 @@ pub async fn get_record( Ok(None) => { return ApiError::RecordNotFound.into_response(); } - Err(_) => { + Err(e) => { + if ApiError::detail_is_repo_corruption(&format!("{e:#}")) { + tranquil_pds::repo_ops::schedule_repo_repair(&state, account.user_id); + } return ApiError::InternalError(Some("Failed to lookup record".into())).into_response(); } }; @@ -312,7 +321,10 @@ pub async fn get_record( } }; let mut proof_blocks: BTreeMap = BTreeMap::new(); - if mst.blocks_for_path(&key, &mut proof_blocks).await.is_err() { + if let Err(e) = mst.blocks_for_path(&key, &mut proof_blocks).await { + if ApiError::detail_is_repo_corruption(&format!("{e:#}")) { + tranquil_pds::repo_ops::schedule_repo_repair(&state, account.user_id); + } return ApiError::InternalError(Some("Failed to build proof path".into())).into_response(); } let header = match encode_car_header(&commit_cid) {