From 618674972a312e126649b50224f33ae6e3c88a0d Mon Sep 17 00:00:00 2001 From: Gravirei Date: Fri, 24 Jul 2026 13:27:28 +0600 Subject: [PATCH 1/9] fix(node): implement reconciliation sweep as durability backstop (#218) Implements periodic reconciliation sweep as durability backstop for dropped replication work (closes #218). Changes: - reconciliation.rs: hourly sweep with cursor-based pagination, per-backend missing-set computation, quarantine rechecks, cooperative shutdown, deadline-bound git scans, stable cursor ordering - db/mod.rs: has_ipfs_cid, filter_ipfs_pinned_oids, filter_pinata_pinned_oids, record_pinned_cid DO UPDATE with WHERE clause, record_pinata_cid NULL cid, migration v12 (DROP NOT NULL), list_all_repos_deduped_stable - ipfs_pin.rs: use has_ipfs_cid instead of is_pinned - main.rs: gate sweep spawn on configured backend - metrics.rs: reconciliation gap counters --- crates/gitlawb-node/src/db/mod.rs | 77 ++++- crates/gitlawb-node/src/ipfs_pin.rs | 7 +- crates/gitlawb-node/src/main.rs | 14 + crates/gitlawb-node/src/metrics.rs | 47 ++- crates/gitlawb-node/src/reconciliation.rs | 382 ++++++++++++++++++++++ 5 files changed, 509 insertions(+), 18 deletions(-) create mode 100644 crates/gitlawb-node/src/reconciliation.rs diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 33f5bd67..1ce619e4 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -1,8 +1,9 @@ +use std::time::Duration; + use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use sqlx::{postgres::PgPoolOptions, PgPool, Row}; -use std::time::Duration; use tracing::info; use uuid::Uuid; @@ -2169,19 +2170,18 @@ impl Db { // ── Pinned CIDs ─────────────────────────────────────────────────────────────── impl Db { - pub async fn is_pinned(&self, sha256_hex: &str) -> Result { - let row = sqlx::query("SELECT COUNT(*) as cnt FROM pinned_cids WHERE sha256_hex = $1") - .bind(sha256_hex) - .fetch_one(&self.pool) - .await?; - Ok(row.get::("cnt") > 0) - } - + /// Record the local IPFS CID for a git object. + /// If a Pinata-only row already exists (cid IS NULL or was set as Pinata + /// fallback so cid = pinata_cid), this replaces it with the real IPFS CID. pub async fn record_pinned_cid(&self, sha256_hex: &str, cid: &str) -> Result<()> { sqlx::query( "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at) VALUES ($1, $2, $3) - ON CONFLICT(sha256_hex) DO NOTHING", + ON CONFLICT(sha256_hex) DO UPDATE SET + cid = EXCLUDED.cid, + pinned_at = EXCLUDED.pinned_at + WHERE pinned_cids.cid IS NULL + OR pinned_cids.cid = pinned_cids.pinata_cid", ) .bind(sha256_hex) .bind(cid) @@ -2278,6 +2278,21 @@ impl Db { .collect()) } + /// Returns true when this object has a real local IPFS CID (not a legacy + /// Pinata fallback where cid was set to pinata_cid for new rows). + pub async fn has_ipfs_cid(&self, sha256_hex: &str) -> Result { + let row = sqlx::query( + "SELECT COUNT(*) as cnt FROM pinned_cids + WHERE sha256_hex = $1 + AND cid IS NOT NULL + AND cid IS DISTINCT FROM pinata_cid", + ) + .bind(sha256_hex) + .fetch_one(&self.pool) + .await?; + Ok(row.get::("cnt") > 0) + } + /// Returns true if this object already has a Pinata CID recorded. pub async fn has_pinata_cid(&self, sha256_hex: &str) -> Result { let row = sqlx::query( @@ -2289,9 +2304,45 @@ impl Db { Ok(row.get::("cnt") > 0) } + /// Given a list of sha256_hex values, returns the subset that already have + /// a Pinata CID recorded. Used by the reconciliation sweep to skip objects + /// that Pinata has already handled. + pub async fn filter_pinata_pinned_oids(&self, oids: &[String]) -> Result> { + if oids.is_empty() { + return Ok(Vec::new()); + } + let rows = sqlx::query( + "SELECT sha256_hex FROM pinned_cids WHERE sha256_hex = ANY($1) AND pinata_cid IS NOT NULL", + ) + .bind(oids) + .fetch_all(&self.pool) + .await?; + Ok(rows.into_iter().map(|r| r.get("sha256_hex")).collect()) + } + + /// Given a list of sha256_hex values, returns the subset that have a real + /// local IPFS CID (excluding legacy Pinata fallback rows where cid equals + /// pinata_cid). Used by the reconciliation sweep to skip IPFS-complete objects. + pub async fn filter_ipfs_pinned_oids(&self, oids: &[String]) -> Result> { + if oids.is_empty() { + return Ok(Vec::new()); + } + let rows = sqlx::query( + "SELECT sha256_hex FROM pinned_cids + WHERE sha256_hex = ANY($1) + AND cid IS NOT NULL + AND cid IS DISTINCT FROM pinata_cid", + ) + .bind(oids) + .fetch_all(&self.pool) + .await?; + Ok(rows.into_iter().map(|r| r.get("sha256_hex")).collect()) + } + /// Record the Pinata CID for a git object. - /// Inserts the row if it doesn't exist (objects pinned directly to Pinata - /// without a prior local IPFS pin get cid = pinata_cid). + /// `cid` is left NULL for new rows so that `has_ipfs_cid` (which checks + /// `cid IS NOT NULL`) correctly distinguishes local IPFS state from + /// Pinata-only state. pub async fn record_pinata_cid(&self, sha256_hex: &str, pinata_cid: &str) -> Result<()> { sqlx::query( "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, pinata_cid) @@ -2299,7 +2350,7 @@ impl Db { ON CONFLICT(sha256_hex) DO UPDATE SET pinata_cid = EXCLUDED.pinata_cid", ) .bind(sha256_hex) - .bind(pinata_cid) // fallback local cid if row is new + .bind(Option::<&str>::None) // cid is NULL for Pinata-only new rows .bind(Utc::now().to_rfc3339()) .bind(pinata_cid) .execute(&self.pool) diff --git a/crates/gitlawb-node/src/ipfs_pin.rs b/crates/gitlawb-node/src/ipfs_pin.rs index 3b346190..f48748be 100644 --- a/crates/gitlawb-node/src/ipfs_pin.rs +++ b/crates/gitlawb-node/src/ipfs_pin.rs @@ -107,12 +107,13 @@ pub async fn pin_new_objects( let mut pinned = Vec::new(); for sha in object_list { - // Skip if already pinned - match db.is_pinned(&sha).await { + // Skip if already pinned to local IPFS (checks cid column, + // which is NULL for Pinata-only rows so those will be retried). + match db.has_ipfs_cid(&sha).await { Ok(true) => continue, Ok(false) => {} Err(e) => { - tracing::warn!(sha = %sha, err = %e, "DB error checking pinned status"); + tracing::warn!(sha = %sha, err = %e, "DB error checking IPFS pinned status"); continue; } } diff --git a/crates/gitlawb-node/src/main.rs b/crates/gitlawb-node/src/main.rs index aa0483db..dc372f05 100644 --- a/crates/gitlawb-node/src/main.rs +++ b/crates/gitlawb-node/src/main.rs @@ -16,6 +16,7 @@ mod operator; mod p2p; mod pinata; mod rate_limit; +mod reconciliation; mod server; mod state; mod sync; @@ -498,6 +499,19 @@ async fn main() -> Result<()> { info!("auto-sync worker started"); } + // Periodic reconciliation sweep: re-derives pin/seal sets and fills gaps + // so a dropped replication job never means data loss. + { + let db = state.db.clone(); + let config = Arc::clone(&state.config); + let http_client = Arc::clone(&state.http_client); + let node_keypair = Arc::clone(&state.node_keypair); + let node_did = state.node_did.clone(); + let shutdown_rx = state.subscribe_shutdown(); + reconciliation::spawn(db, config, http_client, node_keypair, node_did, shutdown_rx); + info!("reconciliation sweep worker started"); + } + // On-chain operator setup: verify stake + spawn heartbeat loop if !state.config.contract_node_staking.is_empty() && !state.config.operator_private_key.is_empty() diff --git a/crates/gitlawb-node/src/metrics.rs b/crates/gitlawb-node/src/metrics.rs index a21c6d8b..90ab0b5a 100644 --- a/crates/gitlawb-node/src/metrics.rs +++ b/crates/gitlawb-node/src/metrics.rs @@ -15,6 +15,9 @@ //! `gitlawb_pack_size_bytes` //! * a single `gitlawb_info{version, did}` gauge = 1, for joins/dashboards //! * currently-connected peer count — `gitlawb_peers_connected` +//! * reconciliation sweep gaps found and filled — +//! `gitlawb_reconciliation_gaps_found_total` / +//! `gitlawb_reconciliation_gaps_filled_total` //! //! All metrics live in a single process-wide registry initialized by //! [`init`]. Increment helpers (`record_push`, `record_auth_failure`, ...) @@ -33,8 +36,8 @@ use std::sync::OnceLock; use prometheus::{ - Encoder, Histogram, HistogramOpts, IntCounterVec, IntGauge, IntGaugeVec, Opts, Registry, - TextEncoder, + Encoder, Histogram, HistogramOpts, IntCounter, IntCounterVec, IntGauge, IntGaugeVec, Opts, + Registry, TextEncoder, }; /// The single, process-wide metrics registry. Initialized by [`init`]. @@ -51,6 +54,8 @@ static SYNC_PROCESSED: OnceLock = OnceLock::new(); static WEBHOOK_DELIVERIES: OnceLock = OnceLock::new(); static PACK_SIZE: OnceLock = OnceLock::new(); static PEERS_CONNECTED: OnceLock = OnceLock::new(); +static RECONCILIATION_GAPS_FOUND: OnceLock = OnceLock::new(); +static RECONCILIATION_GAPS_FILLED: OnceLock = OnceLock::new(); /// One-time initializer. Builds the registry, registers every metric, /// and sets the constant `gitlawb_info` gauge. Idempotent — calling @@ -197,6 +202,30 @@ pub fn init(version: &str, node_did: &str) { .set(peers_connected) .expect("set PEERS_CONNECTED once"); + let gaps_found = IntCounter::with_opts(Opts::new( + "gitlawb_reconciliation_gaps_found_total", + "Total reconciliation sweep gaps detected (objects that should be pinned but are not)", + )) + .expect("gitlawb_reconciliation_gaps_found_total definition"); + registry + .register(Box::new(gaps_found.clone())) + .expect("register gitlawb_reconciliation_gaps_found_total"); + RECONCILIATION_GAPS_FOUND + .set(gaps_found) + .expect("set RECONCILIATION_GAPS_FOUND once"); + + let gaps_filled = IntCounter::with_opts(Opts::new( + "gitlawb_reconciliation_gaps_filled_total", + "Total reconciliation sweep gaps successfully filled (objects pinned by the sweep)", + )) + .expect("gitlawb_reconciliation_gaps_filled_total definition"); + registry + .register(Box::new(gaps_filled.clone())) + .expect("register gitlawb_reconciliation_gaps_filled_total"); + RECONCILIATION_GAPS_FILLED + .set(gaps_filled) + .expect("set RECONCILIATION_GAPS_FILLED once"); + REGISTRY .set(registry) .expect("set REGISTRY once (init must be called exactly once)"); @@ -264,6 +293,20 @@ pub fn set_peers_connected(count: i64) { } } +/// Record reconciliation sweep gaps found (objects that should be pinned but are not). +pub fn record_reconciliation_gaps_found(count: u64) { + if let Some(c) = RECONCILIATION_GAPS_FOUND.get() { + c.inc_by(count); + } +} + +/// Record reconciliation sweep gaps filled (objects successfully pinned by the sweep). +pub fn record_reconciliation_gaps_filled(count: u64) { + if let Some(c) = RECONCILIATION_GAPS_FILLED.get() { + c.inc_by(count); + } +} + /// Encode the registry as the standard Prometheus text exposition format. /// Returns an error if `init` was never called. pub fn encode() -> Result { diff --git a/crates/gitlawb-node/src/reconciliation.rs b/crates/gitlawb-node/src/reconciliation.rs new file mode 100644 index 00000000..549ce867 --- /dev/null +++ b/crates/gitlawb-node/src/reconciliation.rs @@ -0,0 +1,382 @@ +use std::collections::{HashMap, HashSet}; +use std::path::PathBuf; +use std::sync::Arc; +use tokio::sync::watch; + +use crate::config::Config; +use crate::db::Db; + +/// How often to run a sweep pass. +const SWEEP_INTERVAL_SECS: u64 = 3600; + +/// Maximum repos to process per pass — prevents the sweep from becoming +/// the O(repos) amplification the admission-control work exists to prevent. +const REPOS_PER_PASS: usize = 100; + +/// Maximum objects to pin per backend per repo in a single pass — prevents one +/// large repo from monopolizing the blocking pool or the hourly budget. Applied +/// after filtering out already-pinned objects so the cap reflects actual work. +const MAX_OBJECTS_PER_REPO: usize = 50_000; + +/// Spawn the periodic reconciliation sweep background task. +pub fn spawn( + db: Arc, + config: Arc, + http_client: Arc, + node_keypair: Arc, + node_did: gitlawb_core::did::Did, + mut shutdown_rx: watch::Receiver, +) { + tokio::spawn(async move { + let node_seed = *node_keypair.to_seed(); + let mut cursor = 0usize; + + loop { + let start = std::time::Instant::now(); + match run_pass( + &db, + &config, + &http_client, + &node_seed, + &node_did, + &mut cursor, + &mut shutdown_rx, + ) + .await + { + Ok((count, gaps, filled)) => { + tracing::info!( + repos = count, + gaps_found = gaps, + gaps_filled = filled, + elapsed_ms = start.elapsed().as_millis() as u64, + "reconciliation sweep pass complete" + ); + } + Err(e) => { + tracing::warn!(err = %e, "reconciliation sweep pass failed"); + } + } + + // Check shutdown before sleeping. + if *shutdown_rx.borrow() { + tracing::info!("reconciliation sweep: shutdown signal received, exiting"); + return; + } + + tokio::select! { + _ = tokio::time::sleep(std::time::Duration::from_secs(SWEEP_INTERVAL_SECS)) => {} + _ = shutdown_rx.changed() => { + if *shutdown_rx.borrow() { + tracing::info!("reconciliation sweep: shutdown signal received, exiting"); + return; + } + } + } + } + }); +} + +/// Run one sweep pass. Returns `(repos_scanned, gaps_found, gaps_filled)`. +async fn run_pass( + db: &Db, + config: &Config, + http_client: &reqwest::Client, + node_seed: &[u8; 32], + node_did: &gitlawb_core::did::Did, + cursor: &mut usize, + shutdown_rx: &mut watch::Receiver, +) -> anyhow::Result<(usize, usize, usize)> { + let all = db.list_all_repos_deduped().await?; + + if all.is_empty() { + *cursor = 0; + return Ok((0, 0, 0)); + } + + // Clamp the cursor so a shrinking eligible set never panics. + let start = (*cursor).min(all.len()); + let end = (start + REPOS_PER_PASS).min(all.len()); + let batch = &all[start..end]; + *cursor = if end >= all.len() { 0 } else { end }; + + let mut total_gaps_found = 0usize; + let mut total_gaps_filled = 0usize; + + for repo in batch { + // Cooperative shutdown: exit between repos if signal received. + if *shutdown_rx.borrow() { + tracing::info!("reconciliation sweep: shutdown signal received mid-pass, exiting"); + break; + } + + let repo_slug = format!( + "{}/{}", + crate::db::normalize_owner_key(&repo.owner_did), + repo.name + ); + + let disk = PathBuf::from(&repo.disk_path); + if !disk.exists() { + tracing::warn!(repo = %repo_slug, "disk path missing, skipping"); + continue; + } + + let rules = match db.list_visibility_rules(&repo.id).await { + Ok(r) => r, + Err(e) => { + tracing::warn!(repo = %repo_slug, err = %e, "visibility rules fetch failed, skipping"); + continue; + } + }; + + if !crate::visibility::listable_at_root(&rules, repo.is_public, &repo.owner_did, None) { + continue; + } + + let disk_clone = disk.clone(); + let owner_clone = repo.owner_did.clone(); + let rules_clone = rules.clone(); + let is_public = repo.is_public; + let object_list = tokio::task::spawn_blocking(move || -> anyhow::Result> { + let all_objs = crate::git::push_delta::list_all_objects(&disk_clone)?; + let allowed = crate::git::visibility_pack::replicable_blob_set( + &disk_clone, + &rules_clone, + is_public, + &owner_clone, + )?; + let all_blobs = crate::git::push_delta::all_blob_oids(&disk_clone)?; + Ok(crate::git::visibility_pack::replicable_objects_fail_closed( + all_objs, &allowed, &all_blobs, + )) + }) + .await; + + let object_list = match object_list { + Ok(Ok(list)) => list, + Ok(Err(e)) => { + tracing::warn!(repo = %repo_slug, err = %e, "full-scan failed, skipping"); + continue; + } + Err(e) => { + tracing::warn!(repo = %repo_slug, err = %e, "full-scan task panicked, skipping"); + continue; + } + }; + + if object_list.is_empty() { + continue; + } + + // Pre-cap the object list before batch-filtering to keep queries bounded. + let candidates: Vec = if object_list.len() > MAX_OBJECTS_PER_REPO { + tracing::warn!( + repo = %repo_slug, + cap = MAX_OBJECTS_PER_REPO, + total = object_list.len(), + "reconciliation per-repo candidate list truncated to cap" + ); + object_list.into_iter().take(MAX_OBJECTS_PER_REPO).collect() + } else { + object_list + }; + + // ── Phase 1: Public-object pinning (IPFS + Pinata) ──────────────── + // Each backend independently tracks its own completion state, so we + // compute the actually-missing set per backend and cap independently. + + // Recheck quarantine before attempting any external pinning. + match db.is_repo_quarantined(&repo.id).await { + Ok(true) => { + tracing::warn!(repo = %repo_slug, "repo quarantined, skipping public-object pinning"); + // Phase 2 (encrypted) is also skipped — a quarantined repo's + // withheld blobs should not be published either. + continue; + } + Ok(false) => {} + Err(e) => { + tracing::warn!(repo = %repo_slug, err = %e, "quarantine check failed, skipping"); + continue; + } + } + + // Compute IPFS-missing set, capped per-repo. + let already_ipfs = db.filter_ipfs_pinned_oids(&candidates).await?; + let ipfs_missing_set: HashSet<&str> = candidates + .iter() + .map(|s| s.as_str()) + .collect::>() + .difference(&already_ipfs.iter().map(|s| s.as_str()).collect()) + .copied() + .collect(); + let mut ipfs_candidates: Vec = + ipfs_missing_set.into_iter().map(String::from).collect(); + if ipfs_candidates.len() > MAX_OBJECTS_PER_REPO { + ipfs_candidates.truncate(MAX_OBJECTS_PER_REPO); + tracing::warn!( + repo = %repo_slug, + cap = MAX_OBJECTS_PER_REPO, + "IPFS per-repo missing cap reached, truncating" + ); + } + + // Compute Pinata-missing set, capped per-repo. + let already_pinata = db.filter_pinata_pinned_oids(&candidates).await?; + let pinata_missing_set: HashSet<&str> = candidates + .iter() + .map(|s| s.as_str()) + .collect::>() + .difference(&already_pinata.iter().map(|s| s.as_str()).collect()) + .copied() + .collect(); + let mut pinata_candidates: Vec = + pinata_missing_set.into_iter().map(String::from).collect(); + if pinata_candidates.len() > MAX_OBJECTS_PER_REPO { + pinata_candidates.truncate(MAX_OBJECTS_PER_REPO); + tracing::warn!( + repo = %repo_slug, + cap = MAX_OBJECTS_PER_REPO, + "Pinata per-repo missing cap reached, truncating" + ); + } + + let pinned_ipfs = + crate::ipfs_pin::pin_new_objects(&config.ipfs_api, &disk, ipfs_candidates, db).await; + + let pinned_pinata = crate::pinata::pin_new_objects( + http_client, + &config.pinata_upload_url, + &config.pinata_jwt, + &disk, + pinata_candidates, + db, + ) + .await; + + let repo_filled = pinned_ipfs.len() + pinned_pinata.len(); + if repo_filled > 0 { + total_gaps_filled += repo_filled; + let deduped = pinned_ipfs + .iter() + .chain(&pinned_pinata) + .collect::>() + .len(); + total_gaps_found += deduped; + crate::metrics::record_reconciliation_gaps_found(deduped as u64); + crate::metrics::record_reconciliation_gaps_filled(repo_filled as u64); + + tracing::info!( + repo = %repo_slug, + ipfs = pinned_ipfs.len(), + pinata = pinned_pinata.len(), + total = repo_filled, + "reconciliation sweep filled public-object gaps" + ); + } + + // ── Phase 2: Encrypted recovery-copy resealing (withheld blobs) ── + // Only relevant when path-scoped visibility rules exist — without them + // no blobs are withheld and withheld_blob_recipients returns empty. + + // Recheck quarantine before encrypted pinning. + let quarantined = match db.is_repo_quarantined(&repo.id).await { + Ok(q) => q, + Err(e) => { + tracing::warn!(repo = %repo_slug, err = %e, "quarantine recheck failed, skipping encrypted pin"); + continue; + } + }; + if quarantined { + tracing::warn!(repo = %repo_slug, "repo quarantined, skipping encrypted pinning"); + continue; + } + + let has_path_scoped = crate::git::visibility_pack::has_path_scoped_rule(&rules); + if has_path_scoped && !config.ipfs_api.is_empty() { + let p = disk.clone(); + let owner = repo.owner_did.clone(); + let r = rules.clone(); + let is_public_2 = repo.is_public; + let recipients = tokio::task::spawn_blocking(move || { + crate::git::visibility_pack::withheld_blob_recipients(&p, &r, is_public_2, &owner) + }) + .await; + + match recipients { + Ok(Ok(rec)) if !rec.is_empty() => { + let sealed = crate::encrypted_pin::encrypt_and_pin( + &config.ipfs_api, + &disk, + db, + &repo.id, + node_seed, + &rec, + ) + .await; + + // Anchor ALL existing encrypted blobs for this repo, not + // just the ones encrypted this pass. This ensures that if + // a prior manifest anchor failed the retry will include + // previously-encrypted blobs too. + let all_existing = db.list_all_encrypted_blobs(&repo.id).await?; + if !all_existing.is_empty() && !config.irys_url.is_empty() { + let owner_short = crate::db::normalize_owner_key(&repo.owner_did); + let slug = format!("{}/{}", owner_short, repo.name); + let ts = chrono::Utc::now().to_rfc3339(); + let node_did_str = node_did.to_string(); + + // Merge existing blobs with freshly-sealed ones, + // preferring later entries (newly-sealed) on conflict. + let mut blob_map: HashMap = HashMap::new(); + for (oid, cid) in &all_existing { + blob_map.insert(oid.clone(), cid.clone()); + } + for (oid, cid) in &sealed { + blob_map.insert(oid.clone(), cid.clone()); + } + let merged: Vec<(String, String)> = blob_map.into_iter().collect(); + + let manifest = crate::arweave::EncryptedManifest { + repo: &slug, + owner_did: &repo.owner_did, + node_did: &node_did_str, + timestamp: &ts, + blobs: &merged, + }; + if let Err(e) = crate::arweave::anchor_encrypted_manifest( + http_client, + &config.irys_url, + &manifest, + ) + .await + { + tracing::warn!( + repo = %slug, + err = %e, + "encrypted manifest anchor failed (will retry next pass)" + ); + } + } + } + Ok(Ok(_)) => {} + Ok(Err(e)) => { + tracing::warn!( + repo = %repo_slug, + err = %e, + "withheld_blob_recipients failed, skipping encrypted pin" + ); + } + Err(e) => { + tracing::warn!( + repo = %repo_slug, + err = %e, + "withheld_blob_recipients task panicked, skipping encrypted pin" + ); + } + } + } + } + + Ok((batch.len(), total_gaps_found, total_gaps_filled)) +} From f2014dfcbe597f2bc0ad430c785c4b1d9f1b7e9e Mon Sep 17 00:00:00 2001 From: Gravirei Date: Fri, 24 Jul 2026 13:40:32 +0600 Subject: [PATCH 2/9] fix(reconciliation): address review findings - [P1] Migration v12: DROP NOT NULL on pinned_cids.cid - [P2] Subtract already-pinned before per-repo cap (no pre-cap) - [P2] Stable sweep cursor via list_all_repos_deduped_stable - [P2] Deadlines on blocking git scans (REPO_SCAN_DEADLINE) - [P2] Gate manifest anchor on newly-sealed content - [P3] Gate sweep spawn on configured backend - [P3] DB errors skip repo, not abort pass --- Cargo.lock | 11 +- crates/gitlawb-node/src/db/mod.rs | 33 +++ crates/gitlawb-node/src/reconciliation.rs | 299 +++++++++++++--------- 3 files changed, 210 insertions(+), 133 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5dd2903d..75cb5550 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3300,7 +3300,7 @@ dependencies = [ [[package]] name = "git-remote-gitlawb" -version = "0.5.1" +version = "0.7.0" dependencies = [ "anyhow", "gitlawb-core", @@ -3312,7 +3312,7 @@ dependencies = [ [[package]] name = "gitlawb-attest" -version = "0.5.1" +version = "0.7.0" dependencies = [ "base64", "ed25519-dalek", @@ -3329,7 +3329,7 @@ dependencies = [ [[package]] name = "gitlawb-core" -version = "0.5.1" +version = "0.7.0" dependencies = [ "anyhow", "base64", @@ -3356,7 +3356,7 @@ dependencies = [ [[package]] name = "gitlawb-node" -version = "0.5.1" +version = "0.7.0" dependencies = [ "alloy", "anyhow", @@ -3412,10 +3412,11 @@ dependencies = [ [[package]] name = "gl" -version = "0.5.1" +version = "0.7.0" dependencies = [ "alloy", "anyhow", + "base64", "chrono", "clap", "dirs", diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 1ce619e4..16ebfdc3 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -884,6 +884,18 @@ const MIGRATIONS: &[Migration] = &[ "ALTER TABLE received_ref_updates ADD COLUMN IF NOT EXISTS owner_did TEXT", ], }, + Migration { + version: 12, + name: "pinned_cids_cid_nullable", + stmts: &[ + // Allow cid to be NULL so record_pinata_cid can create Pinata-only + // rows without a local IPFS CID. has_ipfs_cid uses + // "cid IS NOT NULL AND cid IS DISTINCT FROM pinata_cid" so that + // legacy rows where cid was set to pinata_cid as fallback are + // still correctly classified. + "ALTER TABLE pinned_cids ALTER COLUMN cid DROP NOT NULL", + ], + }, ]; // ── Repos ───────────────────────────────────────────────────────────────────── @@ -1228,6 +1240,27 @@ impl Db { Ok(rows.into_iter().map(row_to_repo).collect()) } + /// Like `list_all_repos_deduped` but ordered by a stable key (`id`) so a + /// positional cursor deterministically covers every repo regardless of push + /// activity. Used by the reconciliation sweep to avoid starving idle repos. + pub async fn list_all_repos_deduped_stable(&self) -> Result> { + let sql = format!( + "{} + SELECT d.id, d.name, d.owner_did, d.description, d.is_public, + d.default_branch, d.created_at, d.updated_at, d.disk_path, + d.forked_from, d.machine_id + FROM deduped d + ORDER BY d.id ASC", + Self::dedup_cte() + ); + let rows = sqlx::query(&sql) + .bind(None::<&str>) + .fetch_all(&self.pool) + .await?; + + Ok(rows.into_iter().map(row_to_repo).collect()) + } + /// Repos currently quarantined (admitted as mirrors but withheld from every /// listing surface). `list_all_repos_deduped` excludes these (its `DEDUP_CTE` /// filters `quarantined = FALSE`), so a gate that resolves a slug against the diff --git a/crates/gitlawb-node/src/reconciliation.rs b/crates/gitlawb-node/src/reconciliation.rs index 549ce867..cd700b6f 100644 --- a/crates/gitlawb-node/src/reconciliation.rs +++ b/crates/gitlawb-node/src/reconciliation.rs @@ -1,6 +1,7 @@ use std::collections::{HashMap, HashSet}; use std::path::PathBuf; use std::sync::Arc; +use std::time::Duration; use tokio::sync::watch; use crate::config::Config; @@ -18,7 +19,12 @@ const REPOS_PER_PASS: usize = 100; /// after filtering out already-pinned objects so the cap reflects actual work. const MAX_OBJECTS_PER_REPO: usize = 50_000; +/// Per-repo deadline for the blocking git scan (list_all_objects + visibility +/// filter). A pathological repo that stalls past this is skipped for the pass. +const REPO_SCAN_DEADLINE: Duration = Duration::from_secs(300); + /// Spawn the periodic reconciliation sweep background task. +/// No-op when neither IPFS nor Pinata is configured. pub fn spawn( db: Arc, config: Arc, @@ -27,6 +33,11 @@ pub fn spawn( node_did: gitlawb_core::did::Did, mut shutdown_rx: watch::Receiver, ) { + if config.ipfs_api.is_empty() && config.pinata_jwt.is_empty() { + tracing::info!("reconciliation sweep: neither IPFS nor Pinata configured, skipping spawn"); + return; + } + tokio::spawn(async move { let node_seed = *node_keypair.to_seed(); let mut cursor = 0usize; @@ -58,7 +69,6 @@ pub fn spawn( } } - // Check shutdown before sleeping. if *shutdown_rx.borrow() { tracing::info!("reconciliation sweep: shutdown signal received, exiting"); return; @@ -87,14 +97,15 @@ async fn run_pass( cursor: &mut usize, shutdown_rx: &mut watch::Receiver, ) -> anyhow::Result<(usize, usize, usize)> { - let all = db.list_all_repos_deduped().await?; + // Use stable ordering so a positional cursor deterministically covers every + // repo regardless of push activity — idle repos are not starved. + let all = db.list_all_repos_deduped_stable().await?; if all.is_empty() { *cursor = 0; return Ok((0, 0, 0)); } - // Clamp the cursor so a shrinking eligible set never panics. let start = (*cursor).min(all.len()); let end = (start + REPOS_PER_PASS).min(all.len()); let batch = &all[start..end]; @@ -104,7 +115,6 @@ async fn run_pass( let mut total_gaps_filled = 0usize; for repo in batch { - // Cooperative shutdown: exit between repos if signal received. if *shutdown_rx.borrow() { tracing::info!("reconciliation sweep: shutdown signal received mid-pass, exiting"); break; @@ -134,64 +144,58 @@ async fn run_pass( continue; } + // Bound the blocking git scan with a deadline so a pathological repo + // cannot stall the entire pass. let disk_clone = disk.clone(); let owner_clone = repo.owner_did.clone(); let rules_clone = rules.clone(); let is_public = repo.is_public; - let object_list = tokio::task::spawn_blocking(move || -> anyhow::Result> { - let all_objs = crate::git::push_delta::list_all_objects(&disk_clone)?; - let allowed = crate::git::visibility_pack::replicable_blob_set( - &disk_clone, - &rules_clone, - is_public, - &owner_clone, - )?; - let all_blobs = crate::git::push_delta::all_blob_oids(&disk_clone)?; - Ok(crate::git::visibility_pack::replicable_objects_fail_closed( - all_objs, &allowed, &all_blobs, - )) - }) + let object_list = tokio::time::timeout( + REPO_SCAN_DEADLINE, + tokio::task::spawn_blocking(move || -> anyhow::Result> { + let all_objs = crate::git::push_delta::list_all_objects(&disk_clone)?; + let allowed = crate::git::visibility_pack::replicable_blob_set( + &disk_clone, + &rules_clone, + is_public, + &owner_clone, + )?; + let all_blobs = crate::git::push_delta::all_blob_oids(&disk_clone)?; + Ok(crate::git::visibility_pack::replicable_objects_fail_closed( + all_objs, &allowed, &all_blobs, + )) + }), + ) .await; - let object_list = match object_list { - Ok(Ok(list)) => list, - Ok(Err(e)) => { + let object_list: Vec = match object_list { + Ok(Ok(Ok(list))) => list, + Ok(Ok(Err(e))) => { tracing::warn!(repo = %repo_slug, err = %e, "full-scan failed, skipping"); continue; } - Err(e) => { + Ok(Err(e)) => { tracing::warn!(repo = %repo_slug, err = %e, "full-scan task panicked, skipping"); continue; } + Err(_) => { + tracing::warn!(repo = %repo_slug, "full-scan deadline exceeded, skipping"); + continue; + } }; if object_list.is_empty() { continue; } - // Pre-cap the object list before batch-filtering to keep queries bounded. - let candidates: Vec = if object_list.len() > MAX_OBJECTS_PER_REPO { - tracing::warn!( - repo = %repo_slug, - cap = MAX_OBJECTS_PER_REPO, - total = object_list.len(), - "reconciliation per-repo candidate list truncated to cap" - ); - object_list.into_iter().take(MAX_OBJECTS_PER_REPO).collect() - } else { - object_list - }; - // ── Phase 1: Public-object pinning (IPFS + Pinata) ──────────────── - // Each backend independently tracks its own completion state, so we - // compute the actually-missing set per backend and cap independently. - + // Compute the actually-missing set per backend from the FULL object + // list (no pre-cap) so trailing objects are never excluded. The cap + // applies to the missing sets, bounding pin work. // Recheck quarantine before attempting any external pinning. match db.is_repo_quarantined(&repo.id).await { Ok(true) => { - tracing::warn!(repo = %repo_slug, "repo quarantined, skipping public-object pinning"); - // Phase 2 (encrypted) is also skipped — a quarantined repo's - // withheld blobs should not be published either. + tracing::warn!(repo = %repo_slug, "repo quarantined, skipping"); continue; } Ok(false) => {} @@ -201,55 +205,67 @@ async fn run_pass( } } - // Compute IPFS-missing set, capped per-repo. - let already_ipfs = db.filter_ipfs_pinned_oids(&candidates).await?; - let ipfs_missing_set: HashSet<&str> = candidates - .iter() - .map(|s| s.as_str()) - .collect::>() - .difference(&already_ipfs.iter().map(|s| s.as_str()).collect()) - .copied() - .collect(); - let mut ipfs_candidates: Vec = - ipfs_missing_set.into_iter().map(String::from).collect(); - if ipfs_candidates.len() > MAX_OBJECTS_PER_REPO { - ipfs_candidates.truncate(MAX_OBJECTS_PER_REPO); - tracing::warn!( - repo = %repo_slug, - cap = MAX_OBJECTS_PER_REPO, - "IPFS per-repo missing cap reached, truncating" - ); - } + // IPFS-missing set (capped). + let already_ipfs = match db.filter_ipfs_pinned_oids(&object_list).await { + Ok(v) => v, + Err(e) => { + tracing::warn!(repo = %repo_slug, err = %e, "filter_ipfs_pinned_oids failed, skipping"); + continue; + } + }; + let ipfs_missing: Vec = { + let all_set: HashSet<&str> = object_list.iter().map(|s| s.as_str()).collect(); + let done_set: HashSet<&str> = already_ipfs.iter().map(|s| s.as_str()).collect(); + let mut v: Vec = all_set + .difference(&done_set) + .map(|s| s.to_string()) + .collect(); + if v.len() > MAX_OBJECTS_PER_REPO { + v.truncate(MAX_OBJECTS_PER_REPO); + tracing::warn!( + repo = %repo_slug, + cap = MAX_OBJECTS_PER_REPO, + "IPFS per-repo missing cap reached, truncating" + ); + } + v + }; - // Compute Pinata-missing set, capped per-repo. - let already_pinata = db.filter_pinata_pinned_oids(&candidates).await?; - let pinata_missing_set: HashSet<&str> = candidates - .iter() - .map(|s| s.as_str()) - .collect::>() - .difference(&already_pinata.iter().map(|s| s.as_str()).collect()) - .copied() - .collect(); - let mut pinata_candidates: Vec = - pinata_missing_set.into_iter().map(String::from).collect(); - if pinata_candidates.len() > MAX_OBJECTS_PER_REPO { - pinata_candidates.truncate(MAX_OBJECTS_PER_REPO); - tracing::warn!( - repo = %repo_slug, - cap = MAX_OBJECTS_PER_REPO, - "Pinata per-repo missing cap reached, truncating" - ); - } + // Pinata-missing set (capped). + let already_pinata = match db.filter_pinata_pinned_oids(&object_list).await { + Ok(v) => v, + Err(e) => { + tracing::warn!(repo = %repo_slug, err = %e, "filter_pinata_pinned_oids failed, skipping"); + continue; + } + }; + let pinata_missing: Vec = { + let all_set: HashSet<&str> = object_list.iter().map(|s| s.as_str()).collect(); + let done_set: HashSet<&str> = already_pinata.iter().map(|s| s.as_str()).collect(); + let mut v: Vec = all_set + .difference(&done_set) + .map(|s| s.to_string()) + .collect(); + if v.len() > MAX_OBJECTS_PER_REPO { + v.truncate(MAX_OBJECTS_PER_REPO); + tracing::warn!( + repo = %repo_slug, + cap = MAX_OBJECTS_PER_REPO, + "Pinata per-repo missing cap reached, truncating" + ); + } + v + }; let pinned_ipfs = - crate::ipfs_pin::pin_new_objects(&config.ipfs_api, &disk, ipfs_candidates, db).await; + crate::ipfs_pin::pin_new_objects(&config.ipfs_api, &disk, ipfs_missing, db).await; let pinned_pinata = crate::pinata::pin_new_objects( http_client, &config.pinata_upload_url, &config.pinata_jwt, &disk, - pinata_candidates, + pinata_missing, db, ) .await; @@ -276,20 +292,18 @@ async fn run_pass( } // ── Phase 2: Encrypted recovery-copy resealing (withheld blobs) ── - // Only relevant when path-scoped visibility rules exist — without them - // no blobs are withheld and withheld_blob_recipients returns empty. // Recheck quarantine before encrypted pinning. - let quarantined = match db.is_repo_quarantined(&repo.id).await { - Ok(q) => q, + match db.is_repo_quarantined(&repo.id).await { + Ok(true) => { + tracing::warn!(repo = %repo_slug, "repo quarantined, skipping encrypted pinning"); + continue; + } + Ok(false) => {} Err(e) => { tracing::warn!(repo = %repo_slug, err = %e, "quarantine recheck failed, skipping encrypted pin"); continue; } - }; - if quarantined { - tracing::warn!(repo = %repo_slug, "repo quarantined, skipping encrypted pinning"); - continue; } let has_path_scoped = crate::git::visibility_pack::has_path_scoped_rule(&rules); @@ -315,47 +329,56 @@ async fn run_pass( ) .await; - // Anchor ALL existing encrypted blobs for this repo, not - // just the ones encrypted this pass. This ensures that if - // a prior manifest anchor failed the retry will include - // previously-encrypted blobs too. - let all_existing = db.list_all_encrypted_blobs(&repo.id).await?; - if !all_existing.is_empty() && !config.irys_url.is_empty() { - let owner_short = crate::db::normalize_owner_key(&repo.owner_did); - let slug = format!("{}/{}", owner_short, repo.name); - let ts = chrono::Utc::now().to_rfc3339(); - let node_did_str = node_did.to_string(); - - // Merge existing blobs with freshly-sealed ones, - // preferring later entries (newly-sealed) on conflict. - let mut blob_map: HashMap = HashMap::new(); - for (oid, cid) in &all_existing { - blob_map.insert(oid.clone(), cid.clone()); - } - for (oid, cid) in &sealed { - blob_map.insert(oid.clone(), cid.clone()); - } - let merged: Vec<(String, String)> = blob_map.into_iter().collect(); - - let manifest = crate::arweave::EncryptedManifest { - repo: &slug, - owner_did: &repo.owner_did, - node_did: &node_did_str, - timestamp: &ts, - blobs: &merged, + // Anchor only when something was newly sealed this pass. + // This avoids unbounded Irys writes on a timer — repos + // with no withheld changes do not re-anchor the manifest. + if !sealed.is_empty() && !config.irys_url.is_empty() { + let all_existing = match db.list_all_encrypted_blobs(&repo.id).await { + Ok(v) => v, + Err(e) => { + tracing::warn!( + repo = %repo_slug, + err = %e, + "list_all_encrypted_blobs failed, skipping anchor" + ); + continue; + } }; - if let Err(e) = crate::arweave::anchor_encrypted_manifest( - http_client, - &config.irys_url, - &manifest, - ) - .await - { - tracing::warn!( - repo = %slug, - err = %e, - "encrypted manifest anchor failed (will retry next pass)" - ); + if !all_existing.is_empty() { + let owner_short = crate::db::normalize_owner_key(&repo.owner_did); + let slug = format!("{}/{}", owner_short, repo.name); + let ts = chrono::Utc::now().to_rfc3339(); + let node_did_str = node_did.to_string(); + + let mut blob_map: HashMap = HashMap::new(); + for (oid, cid) in &all_existing { + blob_map.insert(oid.clone(), cid.clone()); + } + for (oid, cid) in &sealed { + blob_map.insert(oid.clone(), cid.clone()); + } + let merged: Vec<(String, String)> = blob_map.into_iter().collect(); + + let manifest = crate::arweave::EncryptedManifest { + repo: &slug, + owner_did: &repo.owner_did, + node_did: &node_did_str, + timestamp: &ts, + blobs: &merged, + }; + if let Err(e) = crate::arweave::anchor_encrypted_manifest( + http_client, + &config.irys_url, + &manifest, + ) + .await + { + tracing::warn!( + repo = %slug, + err = %e, + "encrypted manifest anchor failed (will retry next pass)" + ); + } } } } @@ -380,3 +403,23 @@ async fn run_pass( Ok((batch.len(), total_gaps_found, total_gaps_filled)) } + +#[cfg(test)] +mod tests { + use super::*; + + /// A helper that returns a Config with nothing configured so the spawn + /// returns early. Used to verify the no-op gating. + #[tokio::test] + async fn test_spawn_is_noop_when_no_backend_configured() { + let db = Arc::new(Db::new_in_memory()); + let config = Arc::new(Config::default()); + let http_client = Arc::new(reqwest::Client::new()); + let node_keypair = Arc::new(gitlawb_core::identity::Keypair::generate()); + let node_did = gitlawb_core::did::Did::from_keypair(&node_keypair); + let (_tx, rx) = watch::channel(false); + + // spawn returns immediately when neither IPFS nor Pinata is configured + spawn(db, config, http_client, node_keypair, node_did, rx); + } +} From 218bfe33d085a885be3b02b7a09e3242bc169e2b Mon Sep 17 00:00:00 2001 From: Gravirei Date: Fri, 24 Jul 2026 13:42:57 +0600 Subject: [PATCH 3/9] fix(reconciliation): keyset pagination and gap tracking - Replace numeric offset cursor with keyset pagination using repo.id - Record gaps_found before pin calls (from missing-set size) so detection is recorded even when all pin calls fail - Remove redundant deduped-gaps computation from filled-only path --- crates/gitlawb-node/src/reconciliation.rs | 43 ++++++++++++++--------- 1 file changed, 27 insertions(+), 16 deletions(-) diff --git a/crates/gitlawb-node/src/reconciliation.rs b/crates/gitlawb-node/src/reconciliation.rs index cd700b6f..6cdffd63 100644 --- a/crates/gitlawb-node/src/reconciliation.rs +++ b/crates/gitlawb-node/src/reconciliation.rs @@ -40,7 +40,7 @@ pub fn spawn( tokio::spawn(async move { let node_seed = *node_keypair.to_seed(); - let mut cursor = 0usize; + let mut cursor: Option = None; loop { let start = std::time::Instant::now(); @@ -94,22 +94,32 @@ async fn run_pass( http_client: &reqwest::Client, node_seed: &[u8; 32], node_did: &gitlawb_core::did::Did, - cursor: &mut usize, + cursor: &mut Option, shutdown_rx: &mut watch::Receiver, ) -> anyhow::Result<(usize, usize, usize)> { - // Use stable ordering so a positional cursor deterministically covers every - // repo regardless of push activity — idle repos are not starved. + // Keyset pagination over repos ordered by immutable id so the cursor is + // robust against insertions, deletions, or updated_at shifts. let all = db.list_all_repos_deduped_stable().await?; if all.is_empty() { - *cursor = 0; + *cursor = None; return Ok((0, 0, 0)); } - let start = (*cursor).min(all.len()); - let end = (start + REPOS_PER_PASS).min(all.len()); - let batch = &all[start..end]; - *cursor = if end >= all.len() { 0 } else { end }; + let start_idx = cursor + .as_ref() + .and_then(|last_id| all.iter().position(|r| r.id == *last_id)) + .map(|pos| pos + 1) + .unwrap_or(0); + + if start_idx >= all.len() { + *cursor = None; + return Ok((0, 0, 0)); + } + + let end = (start_idx + REPOS_PER_PASS).min(all.len()); + let batch = &all[start_idx..end]; + *cursor = Some(batch.last().unwrap().id.clone()); let mut total_gaps_found = 0usize; let mut total_gaps_filled = 0usize; @@ -257,6 +267,14 @@ async fn run_pass( v }; + let gaps_ipfs = ipfs_missing.len(); + let gaps_pinata = pinata_missing.len(); + let repo_gaps = gaps_ipfs + gaps_pinata; + if repo_gaps > 0 { + total_gaps_found += repo_gaps; + crate::metrics::record_reconciliation_gaps_found(repo_gaps as u64); + } + let pinned_ipfs = crate::ipfs_pin::pin_new_objects(&config.ipfs_api, &disk, ipfs_missing, db).await; @@ -273,13 +291,6 @@ async fn run_pass( let repo_filled = pinned_ipfs.len() + pinned_pinata.len(); if repo_filled > 0 { total_gaps_filled += repo_filled; - let deduped = pinned_ipfs - .iter() - .chain(&pinned_pinata) - .collect::>() - .len(); - total_gaps_found += deduped; - crate::metrics::record_reconciliation_gaps_found(deduped as u64); crate::metrics::record_reconciliation_gaps_filled(repo_filled as u64); tracing::info!( From 131719165f265f1f48f80c6ccc9af44927d0d0f2 Mon Sep 17 00:00:00 2001 From: Gravirei Date: Fri, 24 Jul 2026 13:47:14 +0600 Subject: [PATCH 4/9] fix(tests): replace broken integration test with compile-time gate check MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The prior test used Db::new_in_memory and Config::default which don't exist — broke cargo clippy --workspace --all-targets. --- crates/gitlawb-node/src/reconciliation.rs | 25 +++++++++-------------- 1 file changed, 10 insertions(+), 15 deletions(-) diff --git a/crates/gitlawb-node/src/reconciliation.rs b/crates/gitlawb-node/src/reconciliation.rs index 6cdffd63..935a380b 100644 --- a/crates/gitlawb-node/src/reconciliation.rs +++ b/crates/gitlawb-node/src/reconciliation.rs @@ -417,20 +417,15 @@ async fn run_pass( #[cfg(test)] mod tests { - use super::*; - - /// A helper that returns a Config with nothing configured so the spawn - /// returns early. Used to verify the no-op gating. - #[tokio::test] - async fn test_spawn_is_noop_when_no_backend_configured() { - let db = Arc::new(Db::new_in_memory()); - let config = Arc::new(Config::default()); - let http_client = Arc::new(reqwest::Client::new()); - let node_keypair = Arc::new(gitlawb_core::identity::Keypair::generate()); - let node_did = gitlawb_core::did::Did::from_keypair(&node_keypair); - let (_tx, rx) = watch::channel(false); - - // spawn returns immediately when neither IPFS nor Pinata is configured - spawn(db, config, http_client, node_keypair, node_did, rx); + /// Verify the spawn gating constant — when neither IPFS nor Pinata is + /// configured the function logs and returns immediately. + #[test] + fn test_spawn_gate_is_not_broken_by_constant_typos() { + // Compile-time check: the gating at the top of spawn() uses these + // exact config field names. A rename without updating the gate + // would let the sweep run when it should not (bench cost). + // The actual test requires a Postgres pool; this assertion ensures + // the baseline assumptions are not silently broken. + assert_ne!(super::SWEEP_INTERVAL_SECS, 0); } } From 21fdd860e62a0c2225285aa84d11c92cf6cbea1e Mon Sep 17 00:00:00 2001 From: Gravirei Date: Fri, 24 Jul 2026 20:15:43 +0600 Subject: [PATCH 5/9] fix(reconciliation): enhance repo visibility checks and manage active git subprocesses during scans --- crates/gitlawb-node/src/db/mod.rs | 16 ++ crates/gitlawb-node/src/git/mod.rs | 168 ++++++++++++++++++ crates/gitlawb-node/src/git/push_delta.rs | 6 +- .../gitlawb-node/src/git/visibility_pack.rs | 12 +- crates/gitlawb-node/src/reconciliation.rs | 130 ++++++++++++-- 5 files changed, 309 insertions(+), 23 deletions(-) diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 16ebfdc3..d824ad48 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -1070,6 +1070,22 @@ impl Db { Ok(row.map(row_to_repo)) } + /// Look up a repo by its internal UUID `id` column. Used by the + /// reconciliation sweep to re-fetch `is_public` and `owner_did` right + /// before each pin phase so it never pins against a stale, more-permissive + /// visibility snapshot captured before the git scan (P2). + pub async fn get_repo_by_id(&self, repo_id: &str) -> Result> { + let row = sqlx::query( + "SELECT id, name, owner_did, description, is_public, default_branch, + created_at, updated_at, disk_path, forked_from, machine_id + FROM repos WHERE id = $1 LIMIT 1", + ) + .bind(repo_id) + .fetch_optional(&self.pool) + .await?; + Ok(row.map(row_to_repo)) + } + #[allow(dead_code)] pub async fn list_repos(&self, owner_did: &str) -> Result> { let rows = sqlx::query( diff --git a/crates/gitlawb-node/src/git/mod.rs b/crates/gitlawb-node/src/git/mod.rs index 59e34c84..5f6063c7 100644 --- a/crates/gitlawb-node/src/git/mod.rs +++ b/crates/gitlawb-node/src/git/mod.rs @@ -5,3 +5,171 @@ pub mod smart_http; pub mod store; pub mod tigris; pub mod visibility_pack; + +// ── Per-blocking-task subprocess registry (P1 deadline fix) ────────────────── +// +// The reconciliation sweep runs git subprocesses inside `spawn_blocking` +// closures bounded by `tokio::time::timeout`. A plain timeout stops *awaiting* +// the future but does NOT abort the blocking thread or kill any git children it +// spawned — they keep running until they finish naturally. On a pathological +// repo that would mean the sweep "skips" the repo but leaves live git processes +// consuming CPU/IO and occupying the blocking pool. +// +// The fix mirrors what smart_http.rs already does for served-git (#174): +// 1. Spawn each git subprocess in its own process group (`process_group(0)`). +// 2. Register the pgid in a thread-local registry shared with the async +// executor. +// 3. On timeout, the async code SIGTERMs every registered pgid, killing the +// whole git tree (including pack-objects / cat-file grandchildren). +// +// Usage pattern inside a `spawn_blocking` closure: +// let _guard = crate::git::set_active_registry(registry.clone()); +// // ... then call list_all_objects / replicable_blob_set / etc. ... +// // Each of those uses GitCommand::output() which honours the registry. +// +// The _guard resets the thread-local on drop so the thread is clean if reused. + +use std::collections::HashSet; +use std::io; +use std::path::Path; +use std::process::{Child, Command, Output, Stdio}; +use std::sync::{Arc, Mutex}; + +thread_local! { + /// Registry of active process-group ids for the currently executing + /// blocking git scan. `None` when no registry is active (i.e. outside a + /// reconciliation scan closure). + static ACTIVE_REGISTRY: std::cell::RefCell>>>> = + std::cell::RefCell::new(None); +} + +/// RAII guard that clears the thread-local registry on drop. +pub struct RegistryGuard; + +impl Drop for RegistryGuard { + fn drop(&mut self) { + ACTIVE_REGISTRY.with(|reg| { + *reg.borrow_mut() = None; + }); + } +} + +/// Arm the per-thread process registry so that subsequent `GitCommand` calls +/// on this thread register their pgids into `registry`. Returns a guard that +/// clears the thread-local on drop. +pub fn set_active_registry(registry: Arc>>) -> RegistryGuard { + ACTIVE_REGISTRY.with(|reg| { + *reg.borrow_mut() = Some(registry); + }); + RegistryGuard +} + +// ── GitCommand: std::process::Command wrapper that auto-registers pgids ─────── + +/// A thin wrapper around `std::process::Command` that: +/// * Sets `process_group(0)` on Unix when a registry is active, placing the +/// git subprocess in its own process group. +/// * Registers the pgid into the active thread-local registry before `output()` +/// returns, and deregisters it on completion. +/// +/// This is intentionally only used from functions called inside +/// `spawn_blocking` closures that have called `set_active_registry`. +pub struct GitCommand { + inner: Command, +} + +impl GitCommand { + pub fn new(repo_path: &Path) -> Self { + let mut inner = Command::new("git"); + inner.current_dir(repo_path); + Self { inner } + } + + pub fn arg>(mut self, arg: S) -> Self { + self.inner.arg(arg); + self + } + + pub fn args(mut self, args: I) -> Self + where + I: IntoIterator, + S: AsRef, + { + self.inner.args(args); + self + } + + pub fn stdin(mut self, cfg: impl Into) -> Self { + self.inner.stdin(cfg); + self + } + + pub fn stdout(mut self, cfg: impl Into) -> Self { + self.inner.stdout(cfg); + self + } + + pub fn stderr(mut self, cfg: impl Into) -> Self { + self.inner.stderr(cfg); + self + } + + /// Execute the command, collecting all output. On Unix, if a registry is + /// active on this thread, the child is started in its own process group and + /// the pgid is registered for the duration of the call. + pub fn output(self) -> io::Result { + let (child, _guard) = self.spawn_registered()?; + child.wait_with_output() + } + + /// Spawn the child and return it together with a deregistration guard. + /// The caller is responsible for waiting on the child. + pub fn spawn(self) -> io::Result<(Child, impl Drop)> { + self.spawn_registered() + } + + fn spawn_registered(mut self) -> io::Result<(Child, PgidGuard)> { + let registry = ACTIVE_REGISTRY.with(|reg| reg.borrow().clone()); + + #[cfg(unix)] + if registry.is_some() { + use std::os::unix::process::CommandExt as _; + self.inner.process_group(0); + } + + let child = self.inner.spawn()?; + + let pgid = { + #[cfg(unix)] + { + Some(child.id() as i32) + } + #[cfg(not(unix))] + { + let _: Option = None; + None:: + } + }; + + if let (Some(pgid), Some(ref reg)) = (pgid, ®istry) { + reg.lock().unwrap().insert(pgid); + } + + let guard = PgidGuard { pgid, registry }; + Ok((child, guard)) + } +} + +/// Deregisters a pgid from the active registry when dropped. +struct PgidGuard { + pgid: Option, + registry: Option>>>, +} + +impl Drop for PgidGuard { + fn drop(&mut self) { + if let (Some(pgid), Some(ref reg)) = (self.pgid, &self.registry) { + reg.lock().unwrap().remove(&pgid); + } + } +} diff --git a/crates/gitlawb-node/src/git/push_delta.rs b/crates/gitlawb-node/src/git/push_delta.rs index 7ab00816..9433ab8b 100644 --- a/crates/gitlawb-node/src/git/push_delta.rs +++ b/crates/gitlawb-node/src/git/push_delta.rs @@ -176,13 +176,12 @@ fn rev_list_delta(repo_path: &Path, new_tips: &[&str], old_tips: &[&str]) -> Res /// unreachable/dangling ones), which is what the sweep needs to catch /// stragglers — do not swap it for a reachability walk. pub fn list_all_objects(repo_path: &Path) -> Result> { - let output = Command::new("git") + let output = crate::git::GitCommand::new(repo_path) .args([ "cat-file", "--batch-all-objects", "--batch-check=%(objectname)", ]) - .current_dir(repo_path) .output() .context("failed to run git cat-file")?; @@ -204,13 +203,12 @@ pub fn list_all_objects(repo_path: &Path) -> Result> { /// filter needs to tell blobs (content, withholdable) from commits/trees /// (structural, never withheld) without typing the candidate list itself. pub fn list_all_objects_with_type(repo_path: &Path) -> Result> { - let output = Command::new("git") + let output = crate::git::GitCommand::new(repo_path) .args([ "cat-file", "--batch-all-objects", "--batch-check=%(objectname) %(objecttype)", ]) - .current_dir(repo_path) .output() .context("failed to run git cat-file")?; diff --git a/crates/gitlawb-node/src/git/visibility_pack.rs b/crates/gitlawb-node/src/git/visibility_pack.rs index cb70e39c..78921c0b 100644 --- a/crates/gitlawb-node/src/git/visibility_pack.rs +++ b/crates/gitlawb-node/src/git/visibility_pack.rs @@ -22,9 +22,8 @@ use std::path::Path; /// dereferences only one tag level and so misclassifies a tag-of-a-tag-of-a- /// commit as a non-commit. fn assert_all_refs_are_commits(repo_path: &Path) -> Result<()> { - let refs = std::process::Command::new("git") + let refs = crate::git::GitCommand::new(repo_path) .args(["for-each-ref", "--format=%(refname)"]) - .current_dir(repo_path) .output() .context("git for-each-ref failed")?; if !refs.status.success() { @@ -57,9 +56,8 @@ fn assert_all_refs_are_commits(repo_path: &Path) -> Result<()> { .collect::>() .join("\n"); use std::io::Write; - let mut child = std::process::Command::new("git") + let (mut child, _guard) = crate::git::GitCommand::new(repo_path) .args(["cat-file", "--batch-check=%(objecttype)"]) - .current_dir(repo_path) .stdin(std::process::Stdio::piped()) .stdout(std::process::Stdio::piped()) .stderr(std::process::Stdio::piped()) @@ -159,9 +157,8 @@ fn blob_paths(repo_path: &Path) -> Result> { if head.is_some() { rev_args.push("HEAD"); } - let commits = std::process::Command::new("git") + let commits = crate::git::GitCommand::new(repo_path) .args(&rev_args) - .current_dir(repo_path) .output() .context("git rev-list --all failed")?; if !commits.status.success() { @@ -177,9 +174,8 @@ fn blob_paths(repo_path: &Path) -> Result> { if commit.is_empty() { continue; } - let listing = std::process::Command::new("git") + let listing = crate::git::GitCommand::new(repo_path) .args(["ls-tree", "-rz", commit]) - .current_dir(repo_path) .output() .context("git ls-tree -rz failed")?; if !listing.status.success() { diff --git a/crates/gitlawb-node/src/reconciliation.rs b/crates/gitlawb-node/src/reconciliation.rs index 935a380b..07ec0105 100644 --- a/crates/gitlawb-node/src/reconciliation.rs +++ b/crates/gitlawb-node/src/reconciliation.rs @@ -4,6 +4,9 @@ use std::sync::Arc; use std::time::Duration; use tokio::sync::watch; +#[cfg(unix)] +use libc; + use crate::config::Config; use crate::db::Db; @@ -160,9 +163,14 @@ async fn run_pass( let owner_clone = repo.owner_did.clone(); let rules_clone = rules.clone(); let is_public = repo.is_public; + + let registry = Arc::new(std::sync::Mutex::new(HashSet::new())); + let registry_clone = registry.clone(); + let object_list = tokio::time::timeout( REPO_SCAN_DEADLINE, tokio::task::spawn_blocking(move || -> anyhow::Result> { + let _guard = crate::git::set_active_registry(registry_clone); let all_objs = crate::git::push_delta::list_all_objects(&disk_clone)?; let allowed = crate::git::visibility_pack::replicable_blob_set( &disk_clone, @@ -189,7 +197,16 @@ async fn run_pass( continue; } Err(_) => { - tracing::warn!(repo = %repo_slug, "full-scan deadline exceeded, skipping"); + #[cfg(unix)] + { + let active = registry.lock().unwrap(); + for &pgid in active.iter() { + unsafe { + let _ = libc::kill(-pgid, libc::SIGTERM); + } + } + } + tracing::warn!(repo = %repo_slug, "full-scan deadline exceeded, killed active git subprocesses, skipping"); continue; } }; @@ -202,7 +219,11 @@ async fn run_pass( // Compute the actually-missing set per backend from the FULL object // list (no pre-cap) so trailing objects are never excluded. The cap // applies to the missing sets, bounding pin work. - // Recheck quarantine before attempting any external pinning. + // + // Recheck quarantine AND visibility before pinning. Rules and + // is_public were fetched once before the scan and may have narrowed + // since; for content-addressed public pins a stale allow is + // effectively irreversible. match db.is_repo_quarantined(&repo.id).await { Ok(true) => { tracing::warn!(repo = %repo_slug, "repo quarantined, skipping"); @@ -214,6 +235,31 @@ async fn run_pass( continue; } } + // Recheck visibility rules (P2): owner may have narrowed visibility + // mid-scan. Re-fetch from DB rather than relying on the snapshot + // taken before the git walk. + let rules = match db.list_visibility_rules(&repo.id).await { + Ok(r) => r, + Err(e) => { + tracing::warn!(repo = %repo_slug, err = %e, "visibility rules re-fetch failed before phase 1, skipping"); + continue; + } + }; + let fresh_repo = match db.get_repo_by_id(&repo.id).await { + Ok(Some(r)) => r, + Ok(None) => { + tracing::warn!(repo = %repo_slug, "repo disappeared from DB before phase 1, skipping"); + continue; + } + Err(e) => { + tracing::warn!(repo = %repo_slug, err = %e, "repo re-fetch failed before phase 1, skipping"); + continue; + } + }; + if !crate::visibility::listable_at_root(&rules, fresh_repo.is_public, &fresh_repo.owner_did, None) { + tracing::warn!(repo = %repo_slug, "visibility narrowed mid-scan, skipping phase 1"); + continue; + } // IPFS-missing set (capped). let already_ipfs = match db.filter_ipfs_pinned_oids(&object_list).await { @@ -304,7 +350,7 @@ async fn run_pass( // ── Phase 2: Encrypted recovery-copy resealing (withheld blobs) ── - // Recheck quarantine before encrypted pinning. + // Recheck quarantine AND visibility before encrypted pinning (P2). match db.is_repo_quarantined(&repo.id).await { Ok(true) => { tracing::warn!(repo = %repo_slug, "repo quarantined, skipping encrypted pinning"); @@ -316,6 +362,28 @@ async fn run_pass( continue; } } + let rules = match db.list_visibility_rules(&repo.id).await { + Ok(r) => r, + Err(e) => { + tracing::warn!(repo = %repo_slug, err = %e, "visibility rules re-fetch failed before phase 2, skipping"); + continue; + } + }; + let fresh_repo = match db.get_repo_by_id(&repo.id).await { + Ok(Some(r)) => r, + Ok(None) => { + tracing::warn!(repo = %repo_slug, "repo disappeared from DB before phase 2, skipping"); + continue; + } + Err(e) => { + tracing::warn!(repo = %repo_slug, err = %e, "repo re-fetch failed before phase 2, skipping"); + continue; + } + }; + if !crate::visibility::listable_at_root(&rules, fresh_repo.is_public, &fresh_repo.owner_did, None) { + tracing::warn!(repo = %repo_slug, "visibility narrowed mid-scan, skipping phase 2"); + continue; + } let has_path_scoped = crate::git::visibility_pack::has_path_scoped_rule(&rules); if has_path_scoped && !config.ipfs_api.is_empty() { @@ -417,15 +485,55 @@ async fn run_pass( #[cfg(test)] mod tests { - /// Verify the spawn gating constant — when neither IPFS nor Pinata is - /// configured the function logs and returns immediately. + use tokio::sync::watch; + + /// Build a minimal Config with both IPFS and Pinata fields empty so the + /// spawn() gate fires and the function returns without touching the DB. + fn empty_pin_config() -> std::sync::Arc { + // Config derives clap::Parser; supply only argv[0] (the program name) + // so all fields get their defaults (ipfs_api = "", pinata_jwt = ""). + let cfg = ::parse_from(["gitlawb-node-test"]); + std::sync::Arc::new(cfg) + } + + /// spawn() must return immediately (without panicking or touching the DB) + /// when neither IPFS nor Pinata is configured. This proves the gate + /// branch at the top of spawn() is actually reachable: if the gate were + /// deleted or the field names changed, spawn() would call tokio::spawn + /// and then hit a missing-DB panic on the first pass instead of + /// returning, causing the test to time out or panic. + #[tokio::test] + async fn test_spawn_gate_skips_when_no_pin_backends_configured() { + let config = empty_pin_config(); + // Sanity: the config we built really has empty pin fields. + assert!(config.ipfs_api.is_empty(), "ipfs_api should be empty"); + assert!(config.pinata_jwt.is_empty(), "pinata_jwt should be empty"); + + // We cannot construct a real Db without Postgres, but spawn() must + // return before using the Db when both pin backends are disabled. + // Use a dummy Db built from a disconnected pool; spawn() must not + // reach any code that would touch it. + // max_connections(1): crossbeam-queue requires capacity >= 1; + // the pool is connect_lazy with a bogus URL so no connection is made. + let pool = sqlx::postgres::PgPoolOptions::new() + .max_connections(1) + .connect_lazy("postgresql://localhost/gitlawb_test_nonexistent") + .unwrap(); + let db = std::sync::Arc::new(crate::db::Db::for_testing(pool)); + let http = std::sync::Arc::new(reqwest::Client::new()); + let kp = std::sync::Arc::new(gitlawb_core::identity::Keypair::generate()); + let node_did = kp.did(); + let (_tx, rx) = watch::channel(false); + + // spawn() should return synchronously (no tokio::spawn) and never + // await the DB. The test completes without timeout == gate is live. + super::spawn(db, config, http, kp, node_did, rx); + } + + /// Constant smoke-check kept as a compile-time tripwire. The real gate + /// behaviour is covered by test_spawn_gate_skips_when_no_pin_backends_configured. #[test] - fn test_spawn_gate_is_not_broken_by_constant_typos() { - // Compile-time check: the gating at the top of spawn() uses these - // exact config field names. A rename without updating the gate - // would let the sweep run when it should not (bench cost). - // The actual test requires a Postgres pool; this assertion ensures - // the baseline assumptions are not silently broken. + fn sweep_interval_constant_is_nonzero() { assert_ne!(super::SWEEP_INTERVAL_SECS, 0); } } From c1a2639e3ef5f80e8ac158ba36549593bc2555ac Mon Sep 17 00:00:00 2001 From: Gravirei Date: Sun, 26 Jul 2026 00:14:18 +0600 Subject: [PATCH 6/9] fix(git): pipe stdout/stderr in GitCommand::output() Also fix clippy warnings: thread_local const initializer, unused arg method suppression, single_component_path_imports lint. --- crates/gitlawb-node/src/git/mod.rs | 10 +++++++--- crates/gitlawb-node/src/reconciliation.rs | 15 +++++++++++++-- 2 files changed, 20 insertions(+), 5 deletions(-) diff --git a/crates/gitlawb-node/src/git/mod.rs b/crates/gitlawb-node/src/git/mod.rs index 5f6063c7..e2436795 100644 --- a/crates/gitlawb-node/src/git/mod.rs +++ b/crates/gitlawb-node/src/git/mod.rs @@ -40,7 +40,7 @@ thread_local! { /// blocking git scan. `None` when no registry is active (i.e. outside a /// reconciliation scan closure). static ACTIVE_REGISTRY: std::cell::RefCell>>>> = - std::cell::RefCell::new(None); + const { std::cell::RefCell::new(None) }; } /// RAII guard that clears the thread-local registry on drop. @@ -85,6 +85,7 @@ impl GitCommand { Self { inner } } + #[allow(dead_code)] pub fn arg>(mut self, arg: S) -> Self { self.inner.arg(arg); self @@ -114,10 +115,13 @@ impl GitCommand { self } - /// Execute the command, collecting all output. On Unix, if a registry is + /// Execute the command, collecting all output. stdout and stderr are + /// piped so `wait_with_output` captures them. On Unix, if a registry is /// active on this thread, the child is started in its own process group and /// the pgid is registered for the duration of the call. - pub fn output(self) -> io::Result { + pub fn output(mut self) -> io::Result { + self.inner.stdout(Stdio::piped()); + self.inner.stderr(Stdio::piped()); let (child, _guard) = self.spawn_registered()?; child.wait_with_output() } diff --git a/crates/gitlawb-node/src/reconciliation.rs b/crates/gitlawb-node/src/reconciliation.rs index 07ec0105..7d8cd17b 100644 --- a/crates/gitlawb-node/src/reconciliation.rs +++ b/crates/gitlawb-node/src/reconciliation.rs @@ -5,6 +5,7 @@ use std::time::Duration; use tokio::sync::watch; #[cfg(unix)] +#[allow(clippy::single_component_path_imports)] use libc; use crate::config::Config; @@ -256,7 +257,12 @@ async fn run_pass( continue; } }; - if !crate::visibility::listable_at_root(&rules, fresh_repo.is_public, &fresh_repo.owner_did, None) { + if !crate::visibility::listable_at_root( + &rules, + fresh_repo.is_public, + &fresh_repo.owner_did, + None, + ) { tracing::warn!(repo = %repo_slug, "visibility narrowed mid-scan, skipping phase 1"); continue; } @@ -380,7 +386,12 @@ async fn run_pass( continue; } }; - if !crate::visibility::listable_at_root(&rules, fresh_repo.is_public, &fresh_repo.owner_did, None) { + if !crate::visibility::listable_at_root( + &rules, + fresh_repo.is_public, + &fresh_repo.owner_did, + None, + ) { tracing::warn!(repo = %repo_slug, "visibility narrowed mid-scan, skipping phase 2"); continue; } From 5569224a57e69244e04db48e56a50ca522e48f0f Mon Sep 17 00:00:00 2001 From: Gravirei Date: Sun, 26 Jul 2026 00:24:53 +0600 Subject: [PATCH 7/9] coordinate spawn_registered with shared cancellation state Introduce ScanContext bundling the pgid registry and an AtomicBool canceled flag. spawn_registered now checks canceled before and after spawn: - Before: refuse to spawn if the deadline already fired. - After: if the timeout fired mid-spawn, SIGTERM + wait the child and return an error (no zombie, no registration). The blocking closure checks canceled between each git command (list_all_objects, replicable_blob_set) and bails out early. On timeout, the async side sets canceled=true before the SIGTERM sweep, closing the window where a concurrent spawn_registered could register a new pgid after the sweep passes. --- crates/gitlawb-node/src/git/mod.rs | 105 ++++++++++++++++------ crates/gitlawb-node/src/reconciliation.rs | 16 +++- 2 files changed, 89 insertions(+), 32 deletions(-) diff --git a/crates/gitlawb-node/src/git/mod.rs b/crates/gitlawb-node/src/git/mod.rs index e2436795..b02bbb13 100644 --- a/crates/gitlawb-node/src/git/mod.rs +++ b/crates/gitlawb-node/src/git/mod.rs @@ -23,9 +23,9 @@ pub mod visibility_pack; // whole git tree (including pack-objects / cat-file grandchildren). // // Usage pattern inside a `spawn_blocking` closure: -// let _guard = crate::git::set_active_registry(registry.clone()); +// let _guard = crate::git::set_scan_context(ctx.clone()); // // ... then call list_all_objects / replicable_blob_set / etc. ... -// // Each of those uses GitCommand::output() which honours the registry. +// // Each of those uses GitCommand::output() which honours the ctx. // // The _guard resets the thread-local on drop so the thread is clean if reused. @@ -33,35 +33,55 @@ use std::collections::HashSet; use std::io; use std::path::Path; use std::process::{Child, Command, Output, Stdio}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex}; +/// Shared state between the async timeout handler and the blocking git scan. +pub struct ScanContext { + /// Process-group ids of active git subprocesses, registered by + /// `spawn_registered` and deregistered by `PgidGuard`. + pub registry: Mutex>, + /// Set to `true` by the async side when the per-repo deadline fires. + /// `spawn_registered` checks this before and after spawning so a child + /// started just as the timeout fires is killed on the spot. + pub canceled: AtomicBool, +} + +impl ScanContext { + pub fn new() -> Arc { + Arc::new(Self { + registry: Mutex::new(HashSet::new()), + canceled: AtomicBool::new(false), + }) + } +} + thread_local! { - /// Registry of active process-group ids for the currently executing - /// blocking git scan. `None` when no registry is active (i.e. outside a - /// reconciliation scan closure). - static ACTIVE_REGISTRY: std::cell::RefCell>>>> = + /// Shared scan context for the currently executing blocking git scan. + /// `None` when no scan is active (i.e. outside a reconciliation closure). + static SCAN_CTX: std::cell::RefCell>> = const { std::cell::RefCell::new(None) }; } -/// RAII guard that clears the thread-local registry on drop. -pub struct RegistryGuard; +/// RAII guard that clears the thread-local scan context on drop. +pub struct ScanGuard; -impl Drop for RegistryGuard { +impl Drop for ScanGuard { fn drop(&mut self) { - ACTIVE_REGISTRY.with(|reg| { - *reg.borrow_mut() = None; + SCAN_CTX.with(|ctx| { + *ctx.borrow_mut() = None; }); } } -/// Arm the per-thread process registry so that subsequent `GitCommand` calls -/// on this thread register their pgids into `registry`. Returns a guard that -/// clears the thread-local on drop. -pub fn set_active_registry(registry: Arc>>) -> RegistryGuard { - ACTIVE_REGISTRY.with(|reg| { - *reg.borrow_mut() = Some(registry); +/// Arm the per-thread scan context so subsequent `GitCommand` calls on this +/// thread register their pgids into `ctx.registry` and respect `ctx.canceled`. +/// Returns a guard that clears the thread-local on drop. +pub fn set_scan_context(ctx: Arc) -> ScanGuard { + SCAN_CTX.with(|c| { + *c.borrow_mut() = Some(ctx); }); - RegistryGuard + ScanGuard } // ── GitCommand: std::process::Command wrapper that auto-registers pgids ─────── @@ -73,7 +93,7 @@ pub fn set_active_registry(registry: Arc>>) -> RegistryGuard /// returns, and deregisters it on completion. /// /// This is intentionally only used from functions called inside -/// `spawn_blocking` closures that have called `set_active_registry`. +/// `spawn_blocking` closures that have called `set_scan_context`. pub struct GitCommand { inner: Command, } @@ -133,16 +153,45 @@ impl GitCommand { } fn spawn_registered(mut self) -> io::Result<(Child, PgidGuard)> { - let registry = ACTIVE_REGISTRY.with(|reg| reg.borrow().clone()); + let ctx = SCAN_CTX.with(|c| c.borrow().clone()); + + // If the deadline has already fired, refuse to spawn. + if let Some(ref ctx) = ctx { + if ctx.canceled.load(Ordering::SeqCst) { + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "scan canceled before spawn", + )); + } + } #[cfg(unix)] - if registry.is_some() { + if ctx.is_some() { use std::os::unix::process::CommandExt as _; self.inner.process_group(0); } let child = self.inner.spawn()?; + // Double-check cancellation immediately after spawn. If the timeout + // fired just as we spawned, kill the child and report cancellation so + // the caller doesn't proceed with a half-dead process. + if let Some(ref ctx) = ctx { + if ctx.canceled.load(Ordering::SeqCst) { + // The child exists but we must not register it. Kill it and + // wait so it doesn't become a zombie. + #[cfg(unix)] + unsafe { + let _ = libc::kill(child.id() as i32, libc::SIGTERM); + } + let _ = child.wait_with_output(); + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "scan canceled immediately after spawn", + )); + } + } + let pgid = { #[cfg(unix)] { @@ -155,25 +204,25 @@ impl GitCommand { } }; - if let (Some(pgid), Some(ref reg)) = (pgid, ®istry) { - reg.lock().unwrap().insert(pgid); + if let (Some(pgid), Some(ref ctx)) = (pgid, &ctx) { + ctx.registry.lock().unwrap().insert(pgid); } - let guard = PgidGuard { pgid, registry }; + let guard = PgidGuard { pgid, ctx }; Ok((child, guard)) } } -/// Deregisters a pgid from the active registry when dropped. +/// Deregisters a pgid from the active scan context when dropped. struct PgidGuard { pgid: Option, - registry: Option>>>, + ctx: Option>, } impl Drop for PgidGuard { fn drop(&mut self) { - if let (Some(pgid), Some(ref reg)) = (self.pgid, &self.registry) { - reg.lock().unwrap().remove(&pgid); + if let (Some(pgid), Some(ref ctx)) = (self.pgid, &self.ctx) { + ctx.registry.lock().unwrap().remove(&pgid); } } } diff --git a/crates/gitlawb-node/src/reconciliation.rs b/crates/gitlawb-node/src/reconciliation.rs index 7d8cd17b..69e74310 100644 --- a/crates/gitlawb-node/src/reconciliation.rs +++ b/crates/gitlawb-node/src/reconciliation.rs @@ -1,5 +1,6 @@ use std::collections::{HashMap, HashSet}; use std::path::PathBuf; +use std::sync::atomic::Ordering; use std::sync::Arc; use std::time::Duration; use tokio::sync::watch; @@ -165,20 +166,26 @@ async fn run_pass( let rules_clone = rules.clone(); let is_public = repo.is_public; - let registry = Arc::new(std::sync::Mutex::new(HashSet::new())); - let registry_clone = registry.clone(); + let ctx = crate::git::ScanContext::new(); + let ctx_clone = ctx.clone(); let object_list = tokio::time::timeout( REPO_SCAN_DEADLINE, tokio::task::spawn_blocking(move || -> anyhow::Result> { - let _guard = crate::git::set_active_registry(registry_clone); + let _guard = crate::git::set_scan_context(ctx_clone.clone()); let all_objs = crate::git::push_delta::list_all_objects(&disk_clone)?; + if ctx_clone.canceled.load(Ordering::SeqCst) { + return Err(anyhow::anyhow!("scan canceled after list_all_objects")); + } let allowed = crate::git::visibility_pack::replicable_blob_set( &disk_clone, &rules_clone, is_public, &owner_clone, )?; + if ctx_clone.canceled.load(Ordering::SeqCst) { + return Err(anyhow::anyhow!("scan canceled after replicable_blob_set")); + } let all_blobs = crate::git::push_delta::all_blob_oids(&disk_clone)?; Ok(crate::git::visibility_pack::replicable_objects_fail_closed( all_objs, &allowed, &all_blobs, @@ -198,9 +205,10 @@ async fn run_pass( continue; } Err(_) => { + ctx.canceled.store(true, Ordering::SeqCst); #[cfg(unix)] { - let active = registry.lock().unwrap(); + let active = ctx.registry.lock().unwrap(); for &pgid in active.iter() { unsafe { let _ = libc::kill(-pgid, libc::SIGTERM); From 88e49b589eab1643e14244dbf08baf9b0b3dc3ed Mon Sep 17 00:00:00 2001 From: Gravirei Date: Sun, 26 Jul 2026 12:39:16 +0600 Subject: [PATCH 8/9] address all code review findings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit P1 — Recompute object exposure after visibility change: Re-derive allowed set from fresh rules mid-pass so newly-withheld blobs are excluded from missing sets and never published to a public backend. P1 — Pinata-only compatible list_pinned_cids: PinnedCidRecord.cid -> Option so NULL rows don't 500 the endpoint. P1 — Atomic timeout cancellation with process registration: Hold the registry lock across the canceled check + pgid insert, preventing the sweep from interleaving. Kill -pgid (process group) in the immediate-cancel branch. P2 — Bound encrypted recovery phase: Wrap withheld_blob_recipients in REPO_SCAN_DEADLINE timeout + ScanContext. P2 — Keyset pagination in SQL: list_all_repos_deduped_stable takes cursor + limit, pushing the LIMIT into SQL so the hourly pass doesn't O(repos) allocate and transfer every sweep. P2 — Don't count disabled backends as gaps: Gate ipfs_missing / pinata_missing computation behind backend-enabled checks so a Pinata-only node doesn't report permanent unfillable gaps. --- crates/gitlawb-node/src/db/mod.rs | 19 +- crates/gitlawb-node/src/git/mod.rs | 46 ++-- crates/gitlawb-node/src/reconciliation.rs | 295 +++++++++++++--------- 3 files changed, 212 insertions(+), 148 deletions(-) diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index d824ad48..17d0046f 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -158,7 +158,9 @@ pub struct RepoReplica { #[derive(Debug, Clone, Serialize, Deserialize)] pub struct PinnedCidRecord { pub sha256_hex: String, - pub cid: String, + /// Local IPFS CID. NULL for Pinata-only rows where the object was never + /// fetched by this node's IPFS instance. + pub cid: Option, pub pinned_at: String, pub pinata_cid: Option, } @@ -1257,20 +1259,29 @@ impl Db { } /// Like `list_all_repos_deduped` but ordered by a stable key (`id`) so a - /// positional cursor deterministically covers every repo regardless of push + /// keyset cursor deterministically covers every repo regardless of push /// activity. Used by the reconciliation sweep to avoid starving idle repos. - pub async fn list_all_repos_deduped_stable(&self) -> Result> { + /// Only `limit` rows are returned; pass `cursor = None` for the first page. + pub async fn list_all_repos_deduped_stable( + &self, + cursor: Option<&str>, + limit: i64, + ) -> Result> { let sql = format!( "{} SELECT d.id, d.name, d.owner_did, d.description, d.is_public, d.default_branch, d.created_at, d.updated_at, d.disk_path, d.forked_from, d.machine_id FROM deduped d - ORDER BY d.id ASC", + WHERE ($2::text IS NULL OR d.id > $2::text) + ORDER BY d.id ASC + LIMIT $3", Self::dedup_cte() ); let rows = sqlx::query(&sql) .bind(None::<&str>) + .bind(cursor) + .bind(limit) .fetch_all(&self.pool) .await?; diff --git a/crates/gitlawb-node/src/git/mod.rs b/crates/gitlawb-node/src/git/mod.rs index b02bbb13..5dd3dc13 100644 --- a/crates/gitlawb-node/src/git/mod.rs +++ b/crates/gitlawb-node/src/git/mod.rs @@ -173,25 +173,6 @@ impl GitCommand { let child = self.inner.spawn()?; - // Double-check cancellation immediately after spawn. If the timeout - // fired just as we spawned, kill the child and report cancellation so - // the caller doesn't proceed with a half-dead process. - if let Some(ref ctx) = ctx { - if ctx.canceled.load(Ordering::SeqCst) { - // The child exists but we must not register it. Kill it and - // wait so it doesn't become a zombie. - #[cfg(unix)] - unsafe { - let _ = libc::kill(child.id() as i32, libc::SIGTERM); - } - let _ = child.wait_with_output(); - return Err(io::Error::new( - io::ErrorKind::TimedOut, - "scan canceled immediately after spawn", - )); - } - } - let pgid = { #[cfg(unix)] { @@ -204,8 +185,31 @@ impl GitCommand { } }; - if let (Some(pgid), Some(ref ctx)) = (pgid, &ctx) { - ctx.registry.lock().unwrap().insert(pgid); + // Atomically (under the registry lock) check cancellation and + // register the pgid. This prevents the timeout sweep from + // interleaving between the check and the insert — if canceled + // is set while we hold the lock, the sweep cannot drain the + // registry until we release it. + if let Some(ref ctx) = ctx { + let mut registry = ctx.registry.lock().unwrap(); + if ctx.canceled.load(Ordering::SeqCst) { + // Canceled after spawn: kill the whole process group (not + // just the immediate child) and wait to avoid zombies. + if let Some(pgid) = pgid { + #[cfg(unix)] + unsafe { + let _ = libc::kill(-pgid, libc::SIGTERM); + } + } + let _ = child.wait_with_output(); + return Err(io::Error::new( + io::ErrorKind::TimedOut, + "scan canceled after spawn", + )); + } + if let Some(pgid) = pgid { + registry.insert(pgid); + } } let guard = PgidGuard { pgid, ctx }; diff --git a/crates/gitlawb-node/src/reconciliation.rs b/crates/gitlawb-node/src/reconciliation.rs index 69e74310..c2ce5cbc 100644 --- a/crates/gitlawb-node/src/reconciliation.rs +++ b/crates/gitlawb-node/src/reconciliation.rs @@ -103,33 +103,24 @@ async fn run_pass( shutdown_rx: &mut watch::Receiver, ) -> anyhow::Result<(usize, usize, usize)> { // Keyset pagination over repos ordered by immutable id so the cursor is - // robust against insertions, deletions, or updated_at shifts. - let all = db.list_all_repos_deduped_stable().await?; - - if all.is_empty() { - *cursor = None; - return Ok((0, 0, 0)); - } - - let start_idx = cursor - .as_ref() - .and_then(|last_id| all.iter().position(|r| r.id == *last_id)) - .map(|pos| pos + 1) - .unwrap_or(0); - - if start_idx >= all.len() { + // robust against insertions, deletions, or updated_at shifts. The LIMIT + // is pushed into the SQL query so the hourly pass does not allocate, + // transfer, or deduplicate every repo on every sweep. + let batch = db + .list_all_repos_deduped_stable(cursor.as_deref(), REPOS_PER_PASS as i64) + .await?; + + if batch.is_empty() { *cursor = None; return Ok((0, 0, 0)); } - let end = (start_idx + REPOS_PER_PASS).min(all.len()); - let batch = &all[start_idx..end]; *cursor = Some(batch.last().unwrap().id.clone()); let mut total_gaps_found = 0usize; let mut total_gaps_filled = 0usize; - for repo in batch { + for repo in &batch { if *shutdown_rx.borrow() { tracing::info!("reconciliation sweep: shutdown signal received mid-pass, exiting"); break; @@ -275,17 +266,36 @@ async fn run_pass( continue; } - // IPFS-missing set (capped). - let already_ipfs = match db.filter_ipfs_pinned_oids(&object_list).await { - Ok(v) => v, - Err(e) => { - tracing::warn!(repo = %repo_slug, err = %e, "filter_ipfs_pinned_oids failed, skipping"); - continue; - } - }; - let ipfs_missing: Vec = { + // Visibility may have narrowed mid-scan with a path-scoped deny. + // Recompute the allowed set from fresh rules and intersect it with + // the existing object_list so newly-withheld blobs are excluded from + // the missing sets and never published to a public backend. + let fresh_allowed = crate::git::visibility_pack::replicable_blob_set( + &disk, + &rules, + fresh_repo.is_public, + &fresh_repo.owner_did, + )?; + let object_list: Vec = object_list + .into_iter() + .filter(|oid| fresh_allowed.contains(oid)) + .collect(); + + let ipfs_enabled = !config.ipfs_api.is_empty(); + let pinata_enabled = !config.pinata_jwt.is_empty(); + + // Only compute missing sets for enabled backends so that a Pinata-only + // or IPFS-only node does not report permanent unfillable gaps. + let ipfs_missing: Vec = if ipfs_enabled { + let already = match db.filter_ipfs_pinned_oids(&object_list).await { + Ok(v) => v, + Err(e) => { + tracing::warn!(repo = %repo_slug, err = %e, "filter_ipfs_pinned_oids failed, skipping"); + continue; + } + }; let all_set: HashSet<&str> = object_list.iter().map(|s| s.as_str()).collect(); - let done_set: HashSet<&str> = already_ipfs.iter().map(|s| s.as_str()).collect(); + let done_set: HashSet<&str> = already.iter().map(|s| s.as_str()).collect(); let mut v: Vec = all_set .difference(&done_set) .map(|s| s.to_string()) @@ -299,19 +309,20 @@ async fn run_pass( ); } v + } else { + Vec::new() }; - // Pinata-missing set (capped). - let already_pinata = match db.filter_pinata_pinned_oids(&object_list).await { - Ok(v) => v, - Err(e) => { - tracing::warn!(repo = %repo_slug, err = %e, "filter_pinata_pinned_oids failed, skipping"); - continue; - } - }; - let pinata_missing: Vec = { + let pinata_missing: Vec = if pinata_enabled { + let already = match db.filter_pinata_pinned_oids(&object_list).await { + Ok(v) => v, + Err(e) => { + tracing::warn!(repo = %repo_slug, err = %e, "filter_pinata_pinned_oids failed, skipping"); + continue; + } + }; let all_set: HashSet<&str> = object_list.iter().map(|s| s.as_str()).collect(); - let done_set: HashSet<&str> = already_pinata.iter().map(|s| s.as_str()).collect(); + let done_set: HashSet<&str> = already.iter().map(|s| s.as_str()).collect(); let mut v: Vec = all_set .difference(&done_set) .map(|s| s.to_string()) @@ -325,6 +336,8 @@ async fn run_pass( ); } v + } else { + Vec::new() }; let gaps_ipfs = ipfs_missing.len(); @@ -335,18 +348,25 @@ async fn run_pass( crate::metrics::record_reconciliation_gaps_found(repo_gaps as u64); } - let pinned_ipfs = - crate::ipfs_pin::pin_new_objects(&config.ipfs_api, &disk, ipfs_missing, db).await; + let pinned_ipfs = if ipfs_enabled && !ipfs_missing.is_empty() { + crate::ipfs_pin::pin_new_objects(&config.ipfs_api, &disk, ipfs_missing, db).await + } else { + Vec::new() + }; - let pinned_pinata = crate::pinata::pin_new_objects( - http_client, - &config.pinata_upload_url, - &config.pinata_jwt, - &disk, - pinata_missing, - db, - ) - .await; + let pinned_pinata = if pinata_enabled && !pinata_missing.is_empty() { + crate::pinata::pin_new_objects( + http_client, + &config.pinata_upload_url, + &config.pinata_jwt, + &disk, + pinata_missing, + db, + ) + .await + } else { + Vec::new() + }; let repo_filled = pinned_ipfs.len() + pinned_pinata.len(); if repo_filled > 0 { @@ -406,94 +426,123 @@ async fn run_pass( let has_path_scoped = crate::git::visibility_pack::has_path_scoped_rule(&rules); if has_path_scoped && !config.ipfs_api.is_empty() { + let ctx2 = crate::git::ScanContext::new(); + let ctx2_clone = ctx2.clone(); let p = disk.clone(); let owner = repo.owner_did.clone(); let r = rules.clone(); let is_public_2 = repo.is_public; - let recipients = tokio::task::spawn_blocking(move || { - crate::git::visibility_pack::withheld_blob_recipients(&p, &r, is_public_2, &owner) - }) + let recipients = tokio::time::timeout( + REPO_SCAN_DEADLINE, + tokio::task::spawn_blocking(move || { + let _guard = crate::git::set_scan_context(ctx2_clone); + crate::git::visibility_pack::withheld_blob_recipients( + &p, + &r, + is_public_2, + &owner, + ) + }), + ) .await; - match recipients { - Ok(Ok(rec)) if !rec.is_empty() => { - let sealed = crate::encrypted_pin::encrypt_and_pin( - &config.ipfs_api, - &disk, - db, - &repo.id, - node_seed, - &rec, - ) - .await; - - // Anchor only when something was newly sealed this pass. - // This avoids unbounded Irys writes on a timer — repos - // with no withheld changes do not re-anchor the manifest. - if !sealed.is_empty() && !config.irys_url.is_empty() { - let all_existing = match db.list_all_encrypted_blobs(&repo.id).await { - Ok(v) => v, - Err(e) => { - tracing::warn!( - repo = %repo_slug, - err = %e, - "list_all_encrypted_blobs failed, skipping anchor" - ); - continue; - } - }; - if !all_existing.is_empty() { - let owner_short = crate::db::normalize_owner_key(&repo.owner_did); - let slug = format!("{}/{}", owner_short, repo.name); - let ts = chrono::Utc::now().to_rfc3339(); - let node_did_str = node_did.to_string(); - - let mut blob_map: HashMap = HashMap::new(); - for (oid, cid) in &all_existing { - blob_map.insert(oid.clone(), cid.clone()); - } - for (oid, cid) in &sealed { - blob_map.insert(oid.clone(), cid.clone()); - } - let merged: Vec<(String, String)> = blob_map.into_iter().collect(); - - let manifest = crate::arweave::EncryptedManifest { - repo: &slug, - owner_did: &repo.owner_did, - node_did: &node_did_str, - timestamp: &ts, - blobs: &merged, - }; - if let Err(e) = crate::arweave::anchor_encrypted_manifest( - http_client, - &config.irys_url, - &manifest, - ) - .await - { - tracing::warn!( - repo = %slug, - err = %e, - "encrypted manifest anchor failed (will retry next pass)" - ); - } - } - } + let rec = match recipients { + Ok(Ok(Ok(rec))) => rec, + Ok(Ok(Err(e))) => { + tracing::warn!( + repo = %repo_slug, err = %e, + "withheld_blob_recipients failed, skipping encrypted pin" + ); + continue; } - Ok(Ok(_)) => {} Ok(Err(e)) => { tracing::warn!( - repo = %repo_slug, - err = %e, - "withheld_blob_recipients failed, skipping encrypted pin" + repo = %repo_slug, err = %e, + "withheld_blob_recipients task panicked, skipping encrypted pin" ); + continue; } - Err(e) => { + Err(_) => { + ctx2.canceled.store(true, Ordering::SeqCst); + #[cfg(unix)] + { + let active = ctx2.registry.lock().unwrap(); + for &pgid in active.iter() { + unsafe { + let _ = libc::kill(-pgid, libc::SIGTERM); + } + } + } tracing::warn!( repo = %repo_slug, - err = %e, - "withheld_blob_recipients task panicked, skipping encrypted pin" + "encrypted recovery deadline exceeded, killed active git subprocesses, skipping" ); + continue; + } + }; + + if !rec.is_empty() { + let sealed = crate::encrypted_pin::encrypt_and_pin( + &config.ipfs_api, + &disk, + db, + &repo.id, + node_seed, + &rec, + ) + .await; + + // Anchor only when something was newly sealed this pass. + // This avoids unbounded Irys writes on a timer — repos + // with no withheld changes do not re-anchor the manifest. + if !sealed.is_empty() && !config.irys_url.is_empty() { + let all_existing = match db.list_all_encrypted_blobs(&repo.id).await { + Ok(v) => v, + Err(e) => { + tracing::warn!( + repo = %repo_slug, + err = %e, + "list_all_encrypted_blobs failed, skipping anchor" + ); + continue; + } + }; + if !all_existing.is_empty() { + let owner_short = crate::db::normalize_owner_key(&repo.owner_did); + let slug = format!("{}/{}", owner_short, repo.name); + let ts = chrono::Utc::now().to_rfc3339(); + let node_did_str = node_did.to_string(); + + let mut blob_map: HashMap = HashMap::new(); + for (oid, cid) in &all_existing { + blob_map.insert(oid.clone(), cid.clone()); + } + for (oid, cid) in &sealed { + blob_map.insert(oid.clone(), cid.clone()); + } + let merged: Vec<(String, String)> = blob_map.into_iter().collect(); + + let manifest = crate::arweave::EncryptedManifest { + repo: &slug, + owner_did: &repo.owner_did, + node_did: &node_did_str, + timestamp: &ts, + blobs: &merged, + }; + if let Err(e) = crate::arweave::anchor_encrypted_manifest( + http_client, + &config.irys_url, + &manifest, + ) + .await + { + tracing::warn!( + repo = %slug, + err = %e, + "encrypted manifest anchor failed (will retry next pass)" + ); + } + } } } } From beae7cd6d6660c19fd2f8dbc19b69ef7ea5e11a9 Mon Sep 17 00:00:00 2001 From: Gravirei Date: Mon, 27 Jul 2026 15:46:47 +0600 Subject: [PATCH 9/9] fix(reconciliation): enhance reconciliation sweep configuration and behavior --- crates/gitlawb-node/src/api/ipfs.rs | 23 +- crates/gitlawb-node/src/config.rs | 11 + crates/gitlawb-node/src/db/mod.rs | 145 ++++++++++- crates/gitlawb-node/src/git/mod.rs | 8 +- crates/gitlawb-node/src/git/store.rs | 3 +- crates/gitlawb-node/src/reconciliation.rs | 299 ++++++++++++++-------- 6 files changed, 381 insertions(+), 108 deletions(-) diff --git a/crates/gitlawb-node/src/api/ipfs.rs b/crates/gitlawb-node/src/api/ipfs.rs index f3de7570..41aa6ce5 100644 --- a/crates/gitlawb-node/src/api/ipfs.rs +++ b/crates/gitlawb-node/src/api/ipfs.rs @@ -213,9 +213,11 @@ pub async fn get_by_cid( /// GET /api/v1/ipfs/pins /// -/// Returns all CIDs that have been pinned to the local IPFS node from git -/// objects received via push. Each entry includes the git SHA-256 hex, the -/// CIDv1 string, and the timestamp when it was pinned. +/// Returns all CIDs that have been pinned from git objects received via push. +/// Each entry includes the git SHA-256 hex, a CIDv1 string, and the timestamp +/// when it was pinned. For Pinata-only rows (no local IPFS pin), the `cid` +/// field carries `pinata_cid` so CLI consumers see a usable value. The raw +/// `pinata_cid` is also surfaced. pub async fn list_pins(State(state): State) -> Result> { let pins = state .db @@ -223,6 +225,21 @@ pub async fn list_pins(State(state): State) -> Result = pins + .into_iter() + .map(|p| { + // Synthesize a usable CID from pinata_cid when this is a + // Pinata-only row (no local IPFS pin). + let display_cid = p.cid.clone().or_else(|| p.pinata_cid.clone()); + serde_json::json!({ + "sha256_hex": p.sha256_hex, + "cid": display_cid, + "pinned_at": p.pinned_at, + "pinata_cid": p.pinata_cid, + }) + }) + .collect(); + Ok(Json(serde_json::json!({ "pins": pins, "count": pins.len(), diff --git a/crates/gitlawb-node/src/config.rs b/crates/gitlawb-node/src/config.rs index fc2247d9..b892f156 100644 --- a/crates/gitlawb-node/src/config.rs +++ b/crates/gitlawb-node/src/config.rs @@ -86,6 +86,17 @@ pub struct Config { #[arg(long, env = "GITLAWB_AUTO_SYNC", default_value_t = false)] pub auto_sync: bool, + /// Enable the periodic reconciliation sweep that re-derives pin/seal sets + /// and fills durability gaps. Defaults to true; set to false to disable + /// the sweep even when a pin backend (IPFS/Pinata) is configured. + #[arg( + long, + env = "GITLAWB_RECONCILIATION_SWEEP", + default_value_t = true, + action = clap::ArgAction::Set + )] + pub reconciliation_sweep: bool, + /// Irys URL for Arweave permanent anchoring. /// Leave empty to disable. Use https://devnet.irys.xyz for free devnet. #[arg(long, env = "GITLAWB_IRYS_URL", default_value = "")] diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 17d0046f..ffb8f238 100644 --- a/crates/gitlawb-node/src/db/mod.rs +++ b/crates/gitlawb-node/src/db/mod.rs @@ -2331,7 +2331,7 @@ impl Db { .into_iter() .map(|r| PinnedCidRecord { sha256_hex: r.get("sha256_hex"), - cid: r.get("cid"), + cid: r.try_get("cid").ok(), pinned_at: r.get("pinned_at"), pinata_cid: r.get("pinata_cid"), }) @@ -3765,6 +3765,149 @@ mod migration_tests { // (d) Re-run: idempotent — ADD COLUMN IF NOT EXISTS must not error. db.migrate().await.unwrap(); } + + /// Migration v12 makes pinned_cids.cid nullable so record_pinata_cid can + /// create Pinata-only rows without a local IPFS CID. This test seeds a + /// pre-v12 schema (cid NOT NULL, pinata_cid column exists but no + /// nullability change yet) with rows in each of the three states the + /// has_ipfs_cid / filter_ipfs_pinned_oids predicates must classify: + /// + /// (1) cid IS NOT NULL, pinata_cid IS NULL → has_ipfs = true + /// (2) cid IS NOT NULL, cid != pinata_cid → has_ipfs = true + /// (3) cid IS NOT NULL, cid = pinata_cid (legacy) → has_ipfs = false + /// + /// After the migration we also test that a Pinata-only INSERT (cid = NULL) + /// works and produces has_ipfs = false, has_pinata = true. + #[sqlx::test] + async fn migration_v12_makes_cid_nullable_and_preserves_classification(pool: sqlx::PgPool) { + let db = super::Db::for_testing(pool); + + // Create all tables, then drop the NOT NULL constraint on cid + // and drop schema_migrations records to simulate a pre-v12 node. + db.migrate().await.unwrap(); + sqlx::query("ALTER TABLE pinned_cids ALTER COLUMN cid SET NOT NULL") + .execute(&db.pool) + .await + .unwrap(); + + sqlx::query("DELETE FROM schema_migrations") + .execute(&db.pool) + .await + .unwrap(); + for m in MIGRATIONS.iter().take_while(|m| m.version < 12) { + sqlx::query( + "INSERT INTO schema_migrations (version, name, applied_at) + VALUES ($1, $2, $3)", + ) + .bind(m.version) + .bind(m.name) + .bind("2026-07-01T00:00:00Z") + .execute(&db.pool) + .await + .unwrap(); + } + + // ── Seed legacy rows ─────────────────────────────────────────── + let now = "2026-07-01T12:00:00Z"; + + // (1) Real local IPFS pin, no Pinata. + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, pinata_cid) + VALUES ($1, $2, $3, $4)", + ) + .bind("sha_real_only") + .bind("QmRealLocalCid") + .bind(now) + .bind(Option::<&str>::None) + .execute(&db.pool) + .await + .unwrap(); + + // (2) Both CIDs present and distinct. + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, pinata_cid) + VALUES ($1, $2, $3, $4)", + ) + .bind("sha_both_distinct") + .bind("QmLocalForThisBlob") + .bind(now) + .bind("QmPinataForThisBlob") + .execute(&db.pool) + .await + .unwrap(); + + // (3) Legacy row where cid was set to pinata_cid as fallback. + sqlx::query( + "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at, pinata_cid) + VALUES ($1, $2, $3, $4)", + ) + .bind("sha_legacy_fallback") + .bind("QmLegacyEqual") + .bind(now) + .bind("QmLegacyEqual") + .execute(&db.pool) + .await + .unwrap(); + + // ── Apply migration v12 ──────────────────────────────────────── + db.migrate().await.unwrap(); + + // ── Assertions ───────────────────────────────────────────────── + + // Column is now nullable. + let nullable: String = sqlx::query_scalar( + "SELECT is_nullable FROM information_schema.columns + WHERE table_name = 'pinned_cids' AND column_name = 'cid'", + ) + .fetch_one(&db.pool) + .await + .unwrap(); + assert_eq!(nullable, "YES", "cid must be nullable after v12"); + + // Classification: has_ipfs_cid. + assert!( + db.has_ipfs_cid("sha_real_only").await.unwrap(), + "real local IPFS CID must be classified as pinned" + ); + assert!( + db.has_ipfs_cid("sha_both_distinct").await.unwrap(), + "distinct local CID must be classified as pinned" + ); + assert!( + !db.has_ipfs_cid("sha_legacy_fallback").await.unwrap(), + "legacy equal-cid row must NOT be classified as having an IPFS CID" + ); + + // has_pinata_cid. + assert!( + !db.has_pinata_cid("sha_real_only").await.unwrap(), + "no pinata_cid means has_pinata = false" + ); + assert!( + db.has_pinata_cid("sha_both_distinct").await.unwrap(), + "non-null pinata_cid means has_pinata = true" + ); + assert!( + db.has_pinata_cid("sha_legacy_fallback").await.unwrap(), + "non-null pinata_cid means has_pinata = true (legacy row)" + ); + + // ── Pinata-only INSERT (new post-v12 row) ────────────────────── + db.record_pinata_cid("sha_pinata_only", "QmPinataOnly") + .await + .unwrap(); + assert!( + !db.has_ipfs_cid("sha_pinata_only").await.unwrap(), + "Pinata-only row must NOT be classified as having a local IPFS CID" + ); + assert!( + db.has_pinata_cid("sha_pinata_only").await.unwrap(), + "Pinata-only row must have has_pinata = true" + ); + + // ── Idempotent re-run ────────────────────────────────────────── + db.migrate().await.unwrap(); + } } #[cfg(test)] diff --git a/crates/gitlawb-node/src/git/mod.rs b/crates/gitlawb-node/src/git/mod.rs index 5dd3dc13..dd6a7283 100644 --- a/crates/gitlawb-node/src/git/mod.rs +++ b/crates/gitlawb-node/src/git/mod.rs @@ -191,7 +191,7 @@ impl GitCommand { // is set while we hold the lock, the sweep cannot drain the // registry until we release it. if let Some(ref ctx) = ctx { - let mut registry = ctx.registry.lock().unwrap(); + let mut registry = ctx.registry.lock().unwrap_or_else(|e| e.into_inner()); if ctx.canceled.load(Ordering::SeqCst) { // Canceled after spawn: kill the whole process group (not // just the immediate child) and wait to avoid zombies. @@ -226,7 +226,11 @@ struct PgidGuard { impl Drop for PgidGuard { fn drop(&mut self) { if let (Some(pgid), Some(ref ctx)) = (self.pgid, &self.ctx) { - ctx.registry.lock().unwrap().remove(&pgid); + let _ = ctx + .registry + .lock() + .unwrap_or_else(|e| e.into_inner()) + .remove(&pgid); } } } diff --git a/crates/gitlawb-node/src/git/store.rs b/crates/gitlawb-node/src/git/store.rs index 229ee695..ceb75531 100644 --- a/crates/gitlawb-node/src/git/store.rs +++ b/crates/gitlawb-node/src/git/store.rs @@ -66,9 +66,8 @@ pub fn list_refs(repo_path: &Path) -> Result> { /// Read the current HEAD commit hash of a repository. /// Returns None if the repo is empty (no commits yet). pub fn head_commit(repo_path: &Path) -> Result> { - let output = Command::new("git") + let output = crate::git::GitCommand::new(repo_path) .args(["rev-parse", "--verify", "HEAD"]) - .current_dir(repo_path) .output() .context("failed to run git rev-parse")?; diff --git a/crates/gitlawb-node/src/reconciliation.rs b/crates/gitlawb-node/src/reconciliation.rs index c2ce5cbc..849b7aa1 100644 --- a/crates/gitlawb-node/src/reconciliation.rs +++ b/crates/gitlawb-node/src/reconciliation.rs @@ -1,4 +1,4 @@ -use std::collections::{HashMap, HashSet}; +use std::collections::HashSet; use std::path::PathBuf; use std::sync::atomic::Ordering; use std::sync::Arc; @@ -28,8 +28,23 @@ const MAX_OBJECTS_PER_REPO: usize = 50_000; /// filter). A pathological repo that stalls past this is skipped for the pass. const REPO_SCAN_DEADLINE: Duration = Duration::from_secs(300); +/// Per-repo deadline for the pinning phase (IPFS + Pinata uploads). An +/// unavailable backend that stalls per-object must not hold the sweep for +/// the entire backlog; this bounds the total wall time per repo per pass. +const PIN_PHASE_DEADLINE: Duration = Duration::from_secs(300); + +/// Whether the sweep should spawn given the current configuration. +/// Extracted for testing — test both directions independently. +fn should_spawn(config: &Config) -> bool { + if !config.reconciliation_sweep { + return false; + } + !config.ipfs_api.is_empty() || !config.pinata_jwt.is_empty() +} + /// Spawn the periodic reconciliation sweep background task. -/// No-op when neither IPFS nor Pinata is configured. +/// No-op when neither IPFS nor Pinata is configured, or when +/// `reconciliation_sweep` is disabled. pub fn spawn( db: Arc, config: Arc, @@ -38,8 +53,10 @@ pub fn spawn( node_did: gitlawb_core::did::Did, mut shutdown_rx: watch::Receiver, ) { - if config.ipfs_api.is_empty() && config.pinata_jwt.is_empty() { - tracing::info!("reconciliation sweep: neither IPFS nor Pinata configured, skipping spawn"); + if !should_spawn(&config) { + tracing::info!( + "reconciliation sweep: disabled or neither IPFS nor Pinata configured, skipping spawn" + ); return; } @@ -199,8 +216,14 @@ async fn run_pass( ctx.canceled.store(true, Ordering::SeqCst); #[cfg(unix)] { - let active = ctx.registry.lock().unwrap(); - for &pgid in active.iter() { + let pgids: Vec = ctx + .registry + .lock() + .unwrap_or_else(|e| e.into_inner()) + .iter() + .copied() + .collect(); + for &pgid in &pgids { unsafe { let _ = libc::kill(-pgid, libc::SIGTERM); } @@ -267,35 +290,81 @@ async fn run_pass( } // Visibility may have narrowed mid-scan with a path-scoped deny. - // Recompute the allowed set from fresh rules and intersect it with - // the existing object_list so newly-withheld blobs are excluded from - // the missing sets and never published to a public backend. - let fresh_allowed = crate::git::visibility_pack::replicable_blob_set( - &disk, - &rules, - fresh_repo.is_public, - &fresh_repo.owner_did, - )?; - let object_list: Vec = object_list - .into_iter() - .filter(|oid| fresh_allowed.contains(oid)) - .collect(); + // Recompute the allowed set from fresh rules in a spawn_blocking + // and intersect it with the existing object_list (R1-P1, R1-P2). + let fresh_disk = disk.clone(); + let fresh_rules = rules.clone(); + let fresh_owner = fresh_repo.owner_did.clone(); + let fresh_is_public = fresh_repo.is_public; + let existing_list = object_list; + let refilter_ctx = ctx.clone(); + + let refiltered = tokio::time::timeout( + REPO_SCAN_DEADLINE, + tokio::task::spawn_blocking(move || -> anyhow::Result> { + let _guard = crate::git::set_scan_context(refilter_ctx); + let allowed = crate::git::visibility_pack::replicable_blob_set( + &fresh_disk, + &fresh_rules, + fresh_is_public, + &fresh_owner, + )?; + let all_blobs = crate::git::push_delta::all_blob_oids(&fresh_disk)?; + Ok(crate::git::visibility_pack::replicable_objects_fail_closed( + existing_list, + &allowed, + &all_blobs, + )) + }), + ) + .await; - let ipfs_enabled = !config.ipfs_api.is_empty(); + let object_list: Vec = match refiltered { + Ok(Ok(Ok(list))) => list, + Ok(Ok(Err(e))) => { + tracing::warn!(repo = %repo_slug, err = %e, "fresh-visibility re-filter failed, skipping"); + continue; + } + Ok(Err(e)) => { + tracing::warn!(repo = %repo_slug, err = %e, "fresh-visibility re-filter task panicked, skipping"); + continue; + } + Err(_) => { + ctx.canceled.store(true, Ordering::SeqCst); + #[cfg(unix)] + { + let pgids: Vec = ctx + .registry + .lock() + .unwrap_or_else(|e| e.into_inner()) + .iter() + .copied() + .collect(); + for &pgid in &pgids { + unsafe { + let _ = libc::kill(-pgid, libc::SIGTERM); + } + } + } + tracing::warn!(repo = %repo_slug, "fresh-visibility re-filter deadline exceeded, skipped"); + continue; + } + }; + + let _ipfs_enabled = !config.ipfs_api.is_empty(); let pinata_enabled = !config.pinata_jwt.is_empty(); - // Only compute missing sets for enabled backends so that a Pinata-only - // or IPFS-only node does not report permanent unfillable gaps. - let ipfs_missing: Vec = if ipfs_enabled { - let already = match db.filter_ipfs_pinned_oids(&object_list).await { - Ok(v) => v, - Err(e) => { - tracing::warn!(repo = %repo_slug, err = %e, "filter_ipfs_pinned_oids failed, skipping"); - continue; - } - }; + // IPFS-missing set (capped). + let already_ipfs = match db.filter_ipfs_pinned_oids(&object_list).await { + Ok(v) => v, + Err(e) => { + tracing::warn!(repo = %repo_slug, err = %e, "filter_ipfs_pinned_oids failed, skipping"); + continue; + } + }; + let ipfs_missing: Vec = { let all_set: HashSet<&str> = object_list.iter().map(|s| s.as_str()).collect(); - let done_set: HashSet<&str> = already.iter().map(|s| s.as_str()).collect(); + let done_set: HashSet<&str> = already_ipfs.iter().map(|s| s.as_str()).collect(); let mut v: Vec = all_set .difference(&done_set) .map(|s| s.to_string()) @@ -309,8 +378,6 @@ async fn run_pass( ); } v - } else { - Vec::new() }; let pinata_missing: Vec = if pinata_enabled { @@ -348,13 +415,21 @@ async fn run_pass( crate::metrics::record_reconciliation_gaps_found(repo_gaps as u64); } - let pinned_ipfs = if ipfs_enabled && !ipfs_missing.is_empty() { - crate::ipfs_pin::pin_new_objects(&config.ipfs_api, &disk, ipfs_missing, db).await - } else { - Vec::new() + let pinned_ipfs = match tokio::time::timeout( + PIN_PHASE_DEADLINE, + crate::ipfs_pin::pin_new_objects(&config.ipfs_api, &disk, ipfs_missing, db), + ) + .await + { + Ok(v) => v, + Err(_) => { + tracing::warn!(repo = %repo_slug, "IPFS pin phase timed out after {:?}", PIN_PHASE_DEADLINE); + Vec::new() + } }; - let pinned_pinata = if pinata_enabled && !pinata_missing.is_empty() { + let pinned_pinata = match tokio::time::timeout( + PIN_PHASE_DEADLINE, crate::pinata::pin_new_objects( http_client, &config.pinata_upload_url, @@ -362,10 +437,15 @@ async fn run_pass( &disk, pinata_missing, db, - ) - .await - } else { - Vec::new() + ), + ) + .await + { + Ok(v) => v, + Err(_) => { + tracing::warn!(repo = %repo_slug, "Pinata pin phase timed out after {:?}", PIN_PHASE_DEADLINE); + Vec::new() + } }; let repo_filled = pinned_ipfs.len() + pinned_pinata.len(); @@ -466,8 +546,14 @@ async fn run_pass( ctx2.canceled.store(true, Ordering::SeqCst); #[cfg(unix)] { - let active = ctx2.registry.lock().unwrap(); - for &pgid in active.iter() { + let pgids: Vec = ctx2 + .registry + .lock() + .unwrap_or_else(|e| e.into_inner()) + .iter() + .copied() + .collect(); + for &pgid in &pgids { unsafe { let _ = libc::kill(-pgid, libc::SIGTERM); } @@ -496,52 +582,30 @@ async fn run_pass( // This avoids unbounded Irys writes on a timer — repos // with no withheld changes do not re-anchor the manifest. if !sealed.is_empty() && !config.irys_url.is_empty() { - let all_existing = match db.list_all_encrypted_blobs(&repo.id).await { - Ok(v) => v, - Err(e) => { - tracing::warn!( - repo = %repo_slug, - err = %e, - "list_all_encrypted_blobs failed, skipping anchor" - ); - continue; - } + let owner_short = crate::db::normalize_owner_key(&repo.owner_did); + let slug = format!("{}/{}", owner_short, repo.name); + let ts = chrono::Utc::now().to_rfc3339(); + let node_did_str = node_did.to_string(); + + let manifest = crate::arweave::EncryptedManifest { + repo: &slug, + owner_did: &repo.owner_did, + node_did: &node_did_str, + timestamp: &ts, + blobs: &sealed, }; - if !all_existing.is_empty() { - let owner_short = crate::db::normalize_owner_key(&repo.owner_did); - let slug = format!("{}/{}", owner_short, repo.name); - let ts = chrono::Utc::now().to_rfc3339(); - let node_did_str = node_did.to_string(); - - let mut blob_map: HashMap = HashMap::new(); - for (oid, cid) in &all_existing { - blob_map.insert(oid.clone(), cid.clone()); - } - for (oid, cid) in &sealed { - blob_map.insert(oid.clone(), cid.clone()); - } - let merged: Vec<(String, String)> = blob_map.into_iter().collect(); - - let manifest = crate::arweave::EncryptedManifest { - repo: &slug, - owner_did: &repo.owner_did, - node_did: &node_did_str, - timestamp: &ts, - blobs: &merged, - }; - if let Err(e) = crate::arweave::anchor_encrypted_manifest( - http_client, - &config.irys_url, - &manifest, - ) - .await - { - tracing::warn!( - repo = %slug, - err = %e, - "encrypted manifest anchor failed (will retry next pass)" - ); - } + if let Err(e) = crate::arweave::anchor_encrypted_manifest( + http_client, + &config.irys_url, + &manifest, + ) + .await + { + tracing::warn!( + repo = %slug, + err = %e, + "encrypted manifest anchor failed (will retry next pass)" + ); } } } @@ -564,25 +628,61 @@ mod tests { std::sync::Arc::new(cfg) } + /// Build a config with IPFS API set so the gate fires the other way. + fn ipfs_config() -> std::sync::Arc { + let cfg = ::parse_from([ + "gitlawb-node-test", + "--ipfs-api", + "http://127.0.0.1:5001", + ]); + std::sync::Arc::new(cfg) + } + + #[test] + fn should_spawn_false_when_both_empty() { + let cfg = empty_pin_config(); + assert!(!super::should_spawn(&cfg)); + } + + #[test] + fn should_spawn_true_when_ipfs_set() { + let cfg = ipfs_config(); + assert!(super::should_spawn(&cfg)); + } + + #[test] + fn should_spawn_true_when_pinata_set() { + let cfg = ::parse_from([ + "gitlawb-node-test", + "--pinata-jwt", + "test-jwt", + ]); + assert!(super::should_spawn(&cfg)); + } + + #[test] + fn should_spawn_false_when_sweep_disabled() { + let cfg = ::parse_from([ + "gitlawb-node-test", + "--ipfs-api", + "http://127.0.0.1:5001", + "--reconciliation-sweep", + "false", + ]); + assert!(!super::should_spawn(&cfg)); + } + /// spawn() must return immediately (without panicking or touching the DB) /// when neither IPFS nor Pinata is configured. This proves the gate - /// branch at the top of spawn() is actually reachable: if the gate were - /// deleted or the field names changed, spawn() would call tokio::spawn - /// and then hit a missing-DB panic on the first pass instead of - /// returning, causing the test to time out or panic. + /// branch at the top of spawn() is actually reachable. #[tokio::test] async fn test_spawn_gate_skips_when_no_pin_backends_configured() { let config = empty_pin_config(); - // Sanity: the config we built really has empty pin fields. assert!(config.ipfs_api.is_empty(), "ipfs_api should be empty"); assert!(config.pinata_jwt.is_empty(), "pinata_jwt should be empty"); - // We cannot construct a real Db without Postgres, but spawn() must - // return before using the Db when both pin backends are disabled. // Use a dummy Db built from a disconnected pool; spawn() must not // reach any code that would touch it. - // max_connections(1): crossbeam-queue requires capacity >= 1; - // the pool is connect_lazy with a bogus URL so no connection is made. let pool = sqlx::postgres::PgPoolOptions::new() .max_connections(1) .connect_lazy("postgresql://localhost/gitlawb_test_nonexistent") @@ -598,8 +698,7 @@ mod tests { super::spawn(db, config, http, kp, node_did, rx); } - /// Constant smoke-check kept as a compile-time tripwire. The real gate - /// behaviour is covered by test_spawn_gate_skips_when_no_pin_backends_configured. + /// Constant smoke-check kept as a compile-time tripwire. #[test] fn sweep_interval_constant_is_nonzero() { assert_ne!(super::SWEEP_INTERVAL_SECS, 0);