fix(reconciliation): address coderabbit P1/P2 review findings

- Remove  (conflated IPFS/Pinata); add  + batch
  filter methods for per-backend missing-set computation
- Apply per-repo cap AFTER filtering already-pinned objects so the cap
  reflects actual work, not raw object list size
- Recheck quarantine before Phase 1 and Phase 2 pinning
- Anchor ALL existing encrypted blobs (not just newly-sealed) so failed
  manifest anchors retry on subsequent passes
- Pass shutdown_rx through run_pass for cooperative mid-pass shutdown

Author: Gravirei

crates/gitlawb-node/src/db/mod.rs

@@ -2159,14 +2159,6 @@ impl Db {
 // ── Pinned CIDs ───────────────────────────────────────────────────────────────
 
 impl Db {
-    pub async fn is_pinned(&self, sha256_hex: &str) -> Result<bool> {
-        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::<i64, _>("cnt") > 0)
-    }
-
     pub async fn record_pinned_cid(&self, sha256_hex: &str, cid: &str) -> Result<()> {
         sqlx::query(
             "INSERT INTO pinned_cids (sha256_hex, cid, pinned_at)
@@ -2268,6 +2260,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<bool> {
+        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::<i64, _>("cnt") > 0)
+    }
+
     /// Returns true if this object already has a Pinata CID recorded.
     pub async fn has_pinata_cid(&self, sha256_hex: &str) -> Result<bool> {
         let row = sqlx::query(
@@ -2279,6 +2282,38 @@ impl Db {
         Ok(row.get::<i64, _>("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<Vec<String>> {
+        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<Vec<String>> {
+        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).

crates/gitlawb-node/src/ipfs_pin.rs

@@ -107,12 +107,14 @@ 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;
+        // Pinata-only rows have cid = pinata_cid but we still skip them
+        // here since the object is already stored off-node).
+        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;
             }
         }

crates/gitlawb-node/src/reconciliation.rs

@@ -1,4 +1,4 @@
-use std::collections::HashSet;
+use std::collections::{HashMap, HashSet};
 use std::path::PathBuf;
 use std::sync::Arc;
 use tokio::sync::watch;
@@ -13,8 +13,9 @@ const SWEEP_INTERVAL_SECS: u64 = 3600;
 /// the O(repos) amplification the admission-control work exists to prevent.
 const REPOS_PER_PASS: usize = 100;
 
-/// Maximum objects to pin per repo in a single pass — prevents one large
-/// repo from monopolizing the blocking pool or the hourly budget.
+/// 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.
@@ -30,7 +31,6 @@ pub fn spawn(
         let node_seed = *node_keypair.to_seed();
         let mut cursor = 0usize;
 
-        // Run the first pass immediately on startup, then periodically.
         loop {
             let start = std::time::Instant::now();
             match run_pass(
@@ -40,6 +40,7 @@ pub fn spawn(
                 &node_seed,
                 &node_did,
                 &mut cursor,
+                &mut shutdown_rx,
             )
             .await
             {
@@ -57,6 +58,12 @@ pub fn spawn(
                 }
             }
 
+            // 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() => {
@@ -78,9 +85,8 @@ async fn run_pass(
     node_seed: &[u8; 32],
     node_did: &gitlawb_core::did::Did,
     cursor: &mut usize,
+    shutdown_rx: &mut watch::Receiver<bool>,
 ) -> anyhow::Result<(usize, usize, usize)> {
-    // Use the canonical/deduplicated listing so mirror rows never bypass
-    // visibility rules. The dedup CTE also excludes quarantined repos.
     let all = db.list_all_repos_deduped().await?;
 
     if all.is_empty() {
@@ -98,6 +104,12 @@ 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;
+        }
+
         let repo_slug = format!(
             "{}/{}",
             crate::db::normalize_owner_key(&repo.owner_did),
@@ -128,10 +140,6 @@ async fn run_pass(
         let is_public = repo.is_public;
         let object_list = tokio::task::spawn_blocking(move || -> anyhow::Result<Vec<String>> {
             let all_objs = crate::git::push_delta::list_all_objects(&disk_clone)?;
-            // Always compute the reachable, visibility-allowed blob set so
-            // ordinary public repos recover their blobs too (not just commits
-            // and trees). replicable_blob_set handles the no-rule case
-            // correctly — all reachable blobs are allowed.
             let allowed = crate::git::visibility_pack::replicable_blob_set(
                 &disk_clone,
                 &rules_clone,
@@ -145,7 +153,7 @@ async fn run_pass(
         })
         .await;
 
-        let mut object_list = match object_list {
+        let object_list = match object_list {
             Ok(Ok(list)) => list,
             Ok(Err(e)) => {
                 tracing::warn!(repo = %repo_slug, err = %e, "full-scan failed, skipping");
@@ -161,45 +169,94 @@ async fn run_pass(
             continue;
         }
 
-        // Enforce per-repo object cap so a huge repo cannot monopolize the
-        // blocking pool or run past the hourly interval.
-        let truncated = object_list.len() > MAX_OBJECTS_PER_REPO;
-        if truncated {
-            object_list.truncate(MAX_OBJECTS_PER_REPO);
+        // Pre-cap the object list before batch-filtering to keep queries bounded.
+        let candidates: Vec<String> = if object_list.len() > MAX_OBJECTS_PER_REPO {
             tracing::warn!(
                 repo = %repo_slug,
                 cap = MAX_OBJECTS_PER_REPO,
-                "reconciliation per-repo object cap reached, truncating"
+                total = object_list.len(),
+                "reconciliation per-repo candidate list truncated to cap"
             );
-        }
-
-        let has_path_scoped = crate::git::visibility_pack::has_path_scoped_rule(&rules);
+            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
-        // pass the full replicable set to both.  A row in pinned_cids from
-        // IPFS does not imply a Pinata upload succeeded, and vice versa.
+        // 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::<HashSet<_>>()
+            .difference(&already_ipfs.iter().map(|s| s.as_str()).collect())
+            .copied()
+            .collect();
+        let mut ipfs_candidates: Vec<String> =
+            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::<HashSet<_>>()
+            .difference(&already_pinata.iter().map(|s| s.as_str()).collect())
+            .copied()
+            .collect();
+        let mut pinata_candidates: Vec<String> =
+            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, object_list.clone(), db)
-                .await;
+            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,
-            object_list,
+            pinata_candidates,
             db,
         )
         .await;
 
         let repo_filled = pinned_ipfs.len() + pinned_pinata.len();
         if repo_filled > 0 {
             total_gaps_filled += repo_filled;
-            // Approximate gaps count for observability — objects that were
-            // missing from at least one backend.
-            let missing_ipfs = pinned_ipfs.len();
-            let missing_pinata = pinned_pinata.len();
             let deduped = pinned_ipfs
                 .iter()
                 .chain(&pinned_pinata)
@@ -211,8 +268,8 @@ async fn run_pass(
 
             tracing::info!(
                 repo = %repo_slug,
-                ipfs = missing_ipfs,
-                pinata = missing_pinata,
+                ipfs = pinned_ipfs.len(),
+                pinata = pinned_pinata.len(),
                 total = repo_filled,
                 "reconciliation sweep filled public-object gaps"
             );
@@ -221,6 +278,21 @@ 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,
+            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();
@@ -242,18 +314,35 @@ async fn run_pass(
                         &rec,
                     )
                     .await;
-                    if !sealed.is_empty() && !config.irys_url.is_empty() {
+
+                    // 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();
-                        let blobs: Vec<(String, String)> = sealed;
+
+                        // Merge existing blobs with freshly-sealed ones,
+                        // preferring later entries (newly-sealed) on conflict.
+                        let mut blob_map: HashMap<String, String> = 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: &blobs,
+                            blobs: &merged,
                         };
                         if let Err(e) = crate::arweave::anchor_encrypted_manifest(
                             http_client,
@@ -265,7 +354,7 @@ async fn run_pass(
                             tracing::warn!(
                                 repo = %slug,
                                 err = %e,
-                                "encrypted manifest anchor failed (best-effort)"
+                                "encrypted manifest anchor failed (will retry next pass)"
                             );
                         }
                     }