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/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 33f5bd67..ffb8f238 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; @@ -157,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, } @@ -883,6 +886,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 ───────────────────────────────────────────────────────────────────── @@ -1057,6 +1072,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( @@ -1227,6 +1258,36 @@ 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 + /// keyset cursor deterministically covers every repo regardless of push + /// activity. Used by the reconciliation sweep to avoid starving idle repos. + /// 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 + 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?; + + 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 @@ -2169,19 +2230,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) @@ -2271,13 +2331,28 @@ 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"), }) .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 +2364,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 +2410,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) @@ -3654,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 59e34c84..dd6a7283 100644 --- a/crates/gitlawb-node/src/git/mod.rs +++ b/crates/gitlawb-node/src/git/mod.rs @@ -5,3 +5,232 @@ 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_scan_context(ctx.clone()); +// // ... then call list_all_objects / replicable_blob_set / etc. ... +// // 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. + +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! { + /// 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 scan context on drop. +pub struct ScanGuard; + +impl Drop for ScanGuard { + fn drop(&mut self) { + SCAN_CTX.with(|ctx| { + *ctx.borrow_mut() = None; + }); + } +} + +/// 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); + }); + ScanGuard +} + +// ── 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_scan_context`. +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 } + } + + #[allow(dead_code)] + 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. 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(mut self) -> io::Result { + self.inner.stdout(Stdio::piped()); + self.inner.stderr(Stdio::piped()); + 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 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 ctx.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:: + } + }; + + // 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_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. + 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 }; + Ok((child, guard)) + } +} + +/// Deregisters a pgid from the active scan context when dropped. +struct PgidGuard { + pgid: Option, + ctx: Option>, +} + +impl Drop for PgidGuard { + fn drop(&mut self) { + if let (Some(pgid), Some(ref ctx)) = (self.pgid, &self.ctx) { + let _ = ctx + .registry + .lock() + .unwrap_or_else(|e| e.into_inner()) + .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/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/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/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..849b7aa1 --- /dev/null +++ b/crates/gitlawb-node/src/reconciliation.rs @@ -0,0 +1,706 @@ +use std::collections::HashSet; +use std::path::PathBuf; +use std::sync::atomic::Ordering; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::watch; + +#[cfg(unix)] +#[allow(clippy::single_component_path_imports)] +use libc; + +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; + +/// 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); + +/// 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, or when +/// `reconciliation_sweep` is disabled. +pub fn spawn( + db: Arc, + config: Arc, + http_client: Arc, + node_keypair: Arc, + node_did: gitlawb_core::did::Did, + mut shutdown_rx: watch::Receiver, +) { + if !should_spawn(&config) { + tracing::info!( + "reconciliation sweep: disabled or neither IPFS nor Pinata configured, skipping spawn" + ); + return; + } + + tokio::spawn(async move { + let node_seed = *node_keypair.to_seed(); + let mut cursor: Option = None; + + 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"); + } + } + + 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 Option, + 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. 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)); + } + + *cursor = Some(batch.last().unwrap().id.clone()); + + let mut total_gaps_found = 0usize; + let mut total_gaps_filled = 0usize; + + for repo in &batch { + 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; + } + + // 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 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_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, + )) + }), + ) + .await; + + 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; + } + Ok(Err(e)) => { + tracing::warn!(repo = %repo_slug, err = %e, "full-scan 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, "full-scan deadline exceeded, killed active git subprocesses, skipping"); + continue; + } + }; + + if object_list.is_empty() { + continue; + } + + // ── Phase 1: Public-object pinning (IPFS + Pinata) ──────────────── + // 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 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"); + continue; + } + Ok(false) => {} + Err(e) => { + tracing::warn!(repo = %repo_slug, err = %e, "quarantine check failed, skipping"); + 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; + } + + // Visibility may have narrowed mid-scan with a path-scoped deny. + // 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 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(); + + // 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 + }; + + 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.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 + } else { + Vec::new() + }; + + 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 = 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 = match tokio::time::timeout( + PIN_PHASE_DEADLINE, + crate::pinata::pin_new_objects( + http_client, + &config.pinata_upload_url, + &config.pinata_jwt, + &disk, + pinata_missing, + db, + ), + ) + .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(); + if repo_filled > 0 { + total_gaps_filled += repo_filled; + 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) ── + + // 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"); + continue; + } + Ok(false) => {} + Err(e) => { + tracing::warn!(repo = %repo_slug, err = %e, "quarantine recheck failed, skipping encrypted pin"); + 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() { + 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::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; + + 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(Err(e)) => { + tracing::warn!( + repo = %repo_slug, err = %e, + "withheld_blob_recipients task panicked, skipping encrypted pin" + ); + continue; + } + Err(_) => { + ctx2.canceled.store(true, Ordering::SeqCst); + #[cfg(unix)] + { + 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); + } + } + } + tracing::warn!( + repo = %repo_slug, + "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 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 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((batch.len(), total_gaps_found, total_gaps_filled)) +} + +#[cfg(test)] +mod tests { + 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) + } + + /// 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. + #[tokio::test] + async fn test_spawn_gate_skips_when_no_pin_backends_configured() { + let config = empty_pin_config(); + assert!(config.ipfs_api.is_empty(), "ipfs_api should be empty"); + assert!(config.pinata_jwt.is_empty(), "pinata_jwt should be empty"); + + // Use a dummy Db built from a disconnected pool; spawn() must not + // reach any code that would touch it. + 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. + #[test] + fn sweep_interval_constant_is_nonzero() { + assert_ne!(super::SWEEP_INTERVAL_SECS, 0); + } +}