diff --git a/Cargo.lock b/Cargo.lock index d12daa43..24beb9ea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3300,7 +3300,7 @@ dependencies = [ [[package]] name = "git-remote-gitlawb" -version = "0.5.0" +version = "0.5.1" dependencies = [ "anyhow", "gitlawb-core", @@ -3312,7 +3312,7 @@ dependencies = [ [[package]] name = "gitlawb-attest" -version = "0.5.0" +version = "0.5.1" dependencies = [ "base64", "ed25519-dalek", @@ -3329,7 +3329,7 @@ dependencies = [ [[package]] name = "gitlawb-core" -version = "0.5.0" +version = "0.5.1" dependencies = [ "anyhow", "base64", @@ -3356,7 +3356,7 @@ dependencies = [ [[package]] name = "gitlawb-node" -version = "0.5.0" +version = "0.5.1" dependencies = [ "alloy", "anyhow", @@ -3412,7 +3412,7 @@ dependencies = [ [[package]] name = "gl" -version = "0.5.0" +version = "0.5.1" dependencies = [ "alloy", "anyhow", @@ -3824,6 +3824,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "sha2", "thiserror 2.0.18", "tracing", ] diff --git a/crates/gitlawb-node/src/db/mod.rs b/crates/gitlawb-node/src/db/mod.rs index 5324a4b9..93e9ee8b 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; @@ -2158,19 +2159,16 @@ 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), this updates it with + /// the real IPFS CID so that `has_ipfs_cid` correctly reflects IPFS state. 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 = COALESCE(pinned_cids.cid, EXCLUDED.cid), + pinned_at = EXCLUDED.pinned_at", ) .bind(sha256_hex) .bind(cid) @@ -2267,6 +2265,17 @@ impl Db { .collect()) } + /// Returns true if this object already has a local IPFS CID recorded. + 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", + ) + .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( @@ -2278,9 +2287,42 @@ 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 already have + /// a local IPFS CID recorded. Used by the reconciliation sweep to skip objects + /// that local IPFS has already handled. + 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", + ) + .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) @@ -2288,7 +2330,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)) +}