test(node): pin the per-object marker's third-repo residual

record_pin_source's doc already names this residual and accepts it: the
pin_sources_incomplete marker is one boolean per object rather than per
(object, repo), so a genuine record from a third repo clears a marker that a
different repo's failed record set. Its two neighbouring cases are tested,
the same repo re-recording and a no-op insert that must not clear, and this
one sat between them described in prose and guarded by nothing.

The test drives the real path rather than writing rows. Repo A first-pins the
object. Repo B genuinely holds it and pushes, its skip-branch record fails on
the definite-error arm, and the code marks the set incomplete. An anonymous
read is then SERVED from B, which is the assertion that matters: it proves the
unrecorded holder is reachable while the marker stands. Repo C pushes, its
record succeeds, and the clear fires inside C's transaction. The identical
read now 404s, with the preload counter proving the fallback scan never ran.

Nothing here changes behavior. It converts an accepted tradeoff from a comment
into a guard, so widening it takes a deliberate edit rather than going
unnoticed. The doc comment says the residual fails closed, and that holds:
the provenance loop and the fallback scan both serve through the same
gate_and_serve, so clearing the marker only ever removes a gated search and
can never serve an object the caller could not already read. The window is
exactly one to fifteen sources, because at cap the count-guarded insert is a
no-op and the marker survives.

If someone implements the per-(oid, repo) marker the doc proposes, this test
is expected to go red. It should be updated to the new behavior, not deleted.

Proven load-bearing: removing the clear from record_pin_source leaves the
second read at 200 and turns the test red on the marker assertion.

crates/gitlawb-node/src/ipfs_pin.rs

@@ -11,6 +11,22 @@ use anyhow::Result;
 use gitlawb_core::cid::Cid;
 use std::time::{Duration, Instant};
 
+/// How much READ work one source-less legacy row may cost a boot-time sweep.
+///
+/// The contract this number encodes: discovery for one pre-provenance row is allowed
+/// at most this many bounded object reads from warm local repos, whatever the node's
+/// repo count. The unit counted is the expensive one, a `git cat-file` pair against a
+/// candidate repo; a candidate rejected at filter time (quarantined, cold, an unsafe
+/// path) costs nothing against it. Without the cap the sweep is the O(repos x objects)
+/// fan-out this subsystem's cost rule exists to forbid, paid at boot on the node with
+/// the most history.
+///
+/// It happens to equal the resolver's serve-time per-request source cap
+/// (`db::MAX_PIN_SOURCES`), but it is a different bound with a different owner: that
+/// one bounds how many sources ONE `/ipfs` request may gate, this one bounds how many
+/// repos ONE background row may read. If either moves, the other does not follow.
+pub(crate) const MAX_LEGACY_DISCOVERY_PROBES: usize = 16;
+
 /// Attempts (including the first) for a transient DB-record retry.
 const PIN_RECORD_ATTEMPTS: u32 = 3;
 /// Backoff between DB-record retry attempts.
@@ -274,6 +290,195 @@ pub(crate) struct SweepStats {
     pub retryable_skips: usize,
 }
 
+/// The warm, non-quarantined repos one pass may probe for a source-less legacy row,
+/// plus the one absolute deadline every probe in the pass shares.
+///
+/// Loaded LAZILY, once per pass, on the first source-less row, mirroring the resolver's
+/// own legacy-scan context: a pass with no such row pays nothing. The `is_dir` warm
+/// filter runs ONCE here rather than per row, on the blocking pool, because O(repos)
+/// stat calls per row would park a tokio worker for the whole boot sweep.
+struct DiscoveryCtx {
+    /// Warm candidates with their validated disk paths, oldest-first by
+    /// `(created_at, id)`.
+    candidates: Vec<(crate::db::RepoRecord, std::path::PathBuf)>,
+    /// Shared by every discovery read in the pass, so one pass's discovery costs at
+    /// most one `git_timeout` in total on top of the per-row probe cap.
+    deadline: Instant,
+}
+
+/// Build one pass's discovery candidate list.
+///
+/// Three filters, all applied before any probe so a rejected candidate costs nothing
+/// against [`MAX_LEGACY_DISCOVERY_PROBES`]:
+///
+/// - QUARANTINE. `list_all_repos` is a bare SELECT with no visibility or quarantine
+///   filter, and a quarantined repo is hidden from every reader, so it must not become
+///   a discovery source either. The resolver's legacy scan loads the same set and gates
+///   on it. Private, non-quarantined repos DO stay in the list: an additive source
+///   record binds nothing to one repo's ACL, because the resolver gates every source
+///   independently at serve time, so probing a private repo leaks nothing.
+/// - WARM ONLY. The path is resolved through the repo store's validated resolver and
+///   kept only if it is on local disk. Nothing here goes through `repo_store.acquire`:
+///   the sweep is opportunistic background maintenance over every pinned row on the
+///   node, and pulling cold repos back from remote storage would turn a repair pass
+///   into a bulk restore.
+/// - UNSAFE PATH. A name that fails the validated resolver is dropped with a warn and
+///   is terminal; nothing a later run changes.
+///
+/// The survivors are ordered oldest-first by `(created_at, id)` rather than by id
+/// alone. `repo_id` derives from the owner DID, which anyone can grind, so an id sort
+/// would let an attacker register low-sorting repos and push the true holder past the
+/// probe cap. Source-less rows predate provenance and their holders are old repos,
+/// while freshly registered repos sort last and cannot be backdated.
+async fn load_discovery_ctx(
+    repos_dir: &std::path::Path,
+    git_timeout: Duration,
+    db: &crate::db::Db,
+) -> Result<DiscoveryCtx> {
+    let repos = db.list_all_repos().await?;
+    let quarantined: std::collections::HashSet<String> = db
+        .list_quarantined_repos()
+        .await?
+        .into_iter()
+        .map(|r| r.id)
+        .collect();
+    let mut candidates: Vec<crate::db::RepoRecord> = repos
+        .into_iter()
+        .filter(|r| !quarantined.contains(&r.id))
+        .collect();
+    candidates.sort_by(|a, b| {
+        a.created_at
+            .cmp(&b.created_at)
+            .then_with(|| a.id.cmp(&b.id))
+    });
+
+    let repos_dir = repos_dir.to_path_buf();
+    let warm = tokio::task::spawn_blocking(move || {
+        candidates
+            .into_iter()
+            .filter_map(|repo| {
+                match crate::git::repo_store::validated_repo_disk_path(
+                    &repos_dir,
+                    &repo.owner_did,
+                    &repo.name,
+                ) {
+                    Ok(p) if p.is_dir() => Some((repo, p)),
+                    // Cold: not on this node's disk right now. It is not evidence about
+                    // any row (see `discover_legacy_row`), so it is simply absent here.
+                    Ok(_) => None,
+                    Err(e) => {
+                        tracing::warn!(repo_id = %repo.id, err = %e, "sweep discovery: rejected unsafe repo path");
+                        None
+                    }
+                }
+            })
+            .collect::<Vec<_>>()
+    })
+    .await?;
+
+    Ok(DiscoveryCtx {
+        candidates: warm,
+        deadline: Instant::now() + git_timeout,
+    })
+}
+
+/// What discovery did with one source-less legacy row, in the same three-way shape
+/// [`RepairOutcome`] uses so the row accounting is unchanged.
+enum DiscoveryOutcome {
+    /// Nothing here a later run would find either.
+    Settled,
+    /// Worth walking again: a warm candidate's read failed, the candidate list could
+    /// not be loaded, or the probe cap was reached with candidates still unprobed.
+    Retryable,
+    /// The row's key was rewritten from bytes verified in a warm local repo.
+    Repaired,
+}
+
+/// Probe a bounded set of warm local repos for a source-less legacy row's object.
+///
+/// On a hit, record ONLY what discovery actually knows. Reading identical bytes proves
+/// the repo HOLDS the object, not that it is the first pinner: forks, a shared LICENSE
+/// blob and the empty tree all collide, and `backfill_pin_provenance`'s
+/// `AND repo_id IS NULL` guard would make a guessed exclusive claim permanent. Worse,
+/// the resolver's `needs_scan` is `sources.is_empty() || at_cap || incomplete`, so an
+/// exclusive claim would permanently disable the fallback scan for that object. So
+/// `pinned_cids.repo_id` stays NULL, the discovered repo goes in ADDITIVELY, and the
+/// incomplete marker goes with it because one discovered holder never proves the set
+/// complete.
+///
+/// Both writes are best-effort and warn-only, and the degradation is stated rather
+/// than deferred to a healing pass that does not exist: if the source record fails the
+/// row is raw-CIDv1 with an empty or incomplete source set, which is exactly the state
+/// `needs_scan` routes to the bounded legacy scan, so the object stays servable. The
+/// sweep itself never revisits it (the cost gate skips a raw row free from then on),
+/// so the resolver's fallback is the healing path, not a retry.
+async fn discover_legacy_row(
+    sha: &str,
+    ctx: &mut Option<Option<DiscoveryCtx>>,
+    repos_dir: &std::path::Path,
+    git_bin: &str,
+    git_timeout: Duration,
+    db: &crate::db::Db,
+) -> DiscoveryOutcome {
+    if ctx.is_none() {
+        *ctx = Some(match load_discovery_ctx(repos_dir, git_timeout, db).await {
+            Ok(c) => Some(c),
+            Err(e) => {
+                tracing::warn!(err = %e, "sweep discovery: failed to load the candidate list");
+                None
+            }
+        });
+    }
+    let ctx = match ctx.as_ref().expect("the candidate list was just loaded") {
+        Some(c) => c,
+        // A failed load says nothing about the row, so a later run retries it.
+        None => return DiscoveryOutcome::Retryable,
+    };
+
+    let mut retryable = false;
+    // Every candidate that gets this far is READ, so taking the first
+    // MAX_LEGACY_DISCOVERY_PROBES bounds the expensive work exactly. Candidates the
+    // filters already rejected never reach here and so cost nothing against the cap.
+    for (repo, repo_path) in ctx.candidates.iter().take(MAX_LEGACY_DISCOVERY_PROBES) {
+        match repair_legacy_provider_cid(repo_path, git_bin, ctx.deadline, sha, db).await {
+            Ok(RepairOutcome::Repaired) => {
+                if let Err(e) = db.record_pin_source(sha, &repo.id).await {
+                    tracing::warn!(sha = %sha, repo_id = %repo.id, err = %e, "sweep discovery: failed to record the discovered pin source");
+                }
+                // After the record, never before: a successful `record_pin_source`
+                // clears this marker.
+                if let Err(e) = db.mark_pin_sources_incomplete(sha).await {
+                    tracing::warn!(sha = %sha, err = %e, "sweep discovery: failed to mark the pin-source set incomplete");
+                }
+                return DiscoveryOutcome::Repaired;
+            }
+            // The bytes could not be read from this WARM candidate right now, which IS
+            // evidence about the row: try the next one and walk the row again later.
+            Ok(RepairOutcome::Retryable) => retryable = true,
+            // Absent here, or the row was repaired concurrently. Next candidate.
+            Ok(RepairOutcome::Settled) => {}
+            Err(e) => {
+                tracing::warn!(sha = %sha, repo_id = %repo.id, err = %e, "sweep discovery: probe failed");
+                retryable = true;
+            }
+        }
+    }
+    if ctx.candidates.len() > MAX_LEGACY_DISCOVERY_PROBES {
+        // Cap exhausted with candidates left unprobed: RETRYABLE, never terminal. The
+        // probe order is deterministic, but "a re-walk finds the same nothing" only
+        // holds if the candidate set cannot be steered, and it can: repo ids derive
+        // from grindable owner DIDs, so a terminal verdict would let an attacker bury
+        // the true holder past the cap permanently. The oldest-first order makes that
+        // expensive, and this arm makes it non-permanent.
+        return DiscoveryOutcome::Retryable;
+    }
+    if retryable {
+        DiscoveryOutcome::Retryable
+    } else {
+        DiscoveryOutcome::Settled
+    }
+}
+
 /// One bounded pass of the U4 sweep: read at most `batch` `pinned_cids` rows after
 /// the persisted cursor, repair the legacy ones, and persist the new cursor.
 ///
@@ -298,6 +503,10 @@ async fn sweep_pass(
     let mut repaired = 0usize;
     let mut retryable_skips = 0usize;
     let mut last = cursor;
+    // Loaded on the first source-less row and reused by every later one. The outer
+    // `None` is "not loaded yet"; `Some(None)` is "the load failed this pass", which is
+    // remembered so a broken DB is not re-queried once per row.
+    let mut discovery: Option<Option<DiscoveryCtx>> = None;
 
     for (sha, stored) in rows {
         // Advance FIRST: every path below this line may skip the row, and none of them
@@ -310,7 +519,10 @@ async fn sweep_pass(
         }
         // Resolve the row's repo from its recorded provenance (first-pinner plus the
         // bounded additional source set). An empty set is a pin recorded before
-        // provenance existed: nothing to read the bytes from, so skip it.
+        // provenance existed, which is the pre-provenance-est shape of row and exactly
+        // what this sweep is for, so it is not skipped: discovery below probes a
+        // bounded, quarantine-filtered set of warm local repos for the object and
+        // records what it finds ADDITIVELY.
         let sources = match db.pin_sources_for_oid(&sha).await {
             Ok(s) => s,
             Err(e) => {
@@ -324,6 +536,18 @@ async fn sweep_pass(
         // along the way was a transient obstacle rather than a permanent one.
         let mut row_repaired = false;
         let mut row_retryable = false;
+        if sources.is_empty() {
+            match discover_legacy_row(&sha, &mut discovery, repos_dir, git_bin, git_timeout, db)
+                .await
+            {
+                DiscoveryOutcome::Repaired => {
+                    repaired += 1;
+                    row_repaired = true;
+                }
+                DiscoveryOutcome::Retryable => row_retryable = true,
+                DiscoveryOutcome::Settled => {}
+            }
+        }
         for repo_id in sources {
             let repo = match db.get_repo_by_id(&repo_id).await {
                 Ok(Some(r)) => r,
@@ -428,6 +652,14 @@ pub(crate) async fn sweep_legacy_provider_cids_once(
 /// the node already has, so on an upgraded node that push generally never comes. This
 /// walks the table instead.
 ///
+/// A row with NO recorded source is the pre-provenance case this exists for, so it is
+/// not skipped: the pass probes a bounded, quarantine-filtered set of WARM local repos
+/// for the object (at most [`MAX_LEGACY_DISCOVERY_PROBES`] reads per row, sharing one
+/// per-pass deadline) and, on a hit, rewrites the key from the verified bytes and
+/// records the discovered repo ADDITIVELY alongside the incomplete marker. It never
+/// writes an exclusive first-pinner claim and never pulls a cold repo back from remote
+/// storage. See `discover_legacy_row` for why both of those matter.
+///
 /// Runs until a pass comes back short of a full batch, which is the end of the table.
 /// Sleeps `delay` between full batches so it cannot monopolize the DB, and persists
 /// its cursor every pass so a restart continues instead of rewinding. Errors reading