Merge #6caaa160: Bound duplicate repository fetches and hedge slow sour…

Bound duplicate repository fetches and hedge slow sources

nostr:nevent1qgsx2lyl2e4zvfadwcvkd9fkrcwczj7mf858hy85mwqclwgut8wpg2spz3mhxue69uhhyetvv9ujumn8d96zuer9wcq3yamnwvaz7tm8d96xummnw3ezucm0d5q3kamnwvaz7tmwva5hgtnyv9hxxmmwwashjer9wchxxmmdqqsxe24pvp4pqxuzvpvdrdpgx3j2ug9yqtl6ny8z8mjw5k7djmludxsxzmeud

PR-Author: DanConwayDev's Agent
nostr:npub1v47f74n2ycn66asev62nv8sas99akj0g0wg0fkup37u3ckwuzs4q7cwtp0

CoverNote:

# Bound duplicate repository fetches and hedge slow Git sources

## Production problem

The archive burn-in observed several simultaneous `git fetch`/`index-pack`
pipelines writing into the same large bare repository. Existing controls were
keyed by remote domain, protecting each upstream server from our request rate,
and by aggregate cgroup pressure, protecting the host after pressure appears.
Neither prevented independent identifier/domain workers duplicating expensive
work against one local object database.

## Approach

- Reserve per-domain fetch-pass capacity atomically and release it through an
  owned cancellation-safe permit. `ls-remote` remains inside this protection.
- Coordinate work by resolved local repository path. Duplicate callers wait
  for the active wave and then re-check missing OIDs.
- Give one preferred source a 30-second head start. If it remains incomplete,
  permit exactly one source on another domain to hedge; never start a third.
- Make concurrent object ingestion free of `FETCH_HEAD` and automatic
  maintenance side effects, and serialize promotion of fetched OIDs.
- Stream bounded Git stdout/stderr and expose aggregate activity, duration,
  outcome and byte metrics. Five minutes without either stream producing a
  byte terminates that command's private process group; there is no total
  duration limit, so an active large transfer may run indefinitely.
- On inactivity, send the whole process group SIGTERM, retain its unreaped
  leader for a ten-second grace period, then SIGKILL before reaping. A guard
  applies the same group cleanup if the owning Rust future is cancelled.
- After an unproductive primary/hedge pair has fully ended, try the remaining
  sources sequentially. A third fetch never overlaps the bounded pair.

The two attempts deliberately do not produce a learned source ranking. They
write into the same object database, so either command can observe an object
installed by the other and completion order cannot attribute the object to a
source. Truthful ranking would require isolated per-attempt object stores and
explicit winner promotion, which is excluded here in favour of the smaller,
correct hedge.

The 30-second hedge delay is not a timeout. The separate five-minute inactivity
limit measures transport silence rather than elapsed runtime; every output byte
resets it. Timing remains operational evidence only and never ranks sources.

## Commits

- `b9ed6e8` atomically reserves remote-domain capacity.
- `3903ffe` serializes work per object database and rechecks waiting demand.
- `665af5b` removes shared fetch-head and maintenance side effects.
- `e1eee18` adds exactly one delayed distinct-domain hedge without scoring.
- `0d7c77d` introduces bounded concurrent stream draining and operational
  metrics without changing completion policy.
- `25c535f` adds inactivity recovery, cancellation-safe process-group cleanup,
  and remaining-source fallback.

## Validation

- Full library suite: 709 passed.
- Deterministic scenarios cover duplicate-demand single-flight, independent
  repositories, delayed hedge victory, two slow sources with no third attempt,
  and same-domain alternatives excluded from hedging.
- Domain admission contention and cancellation tests pass.
- Stream capture remains bounded while readers continue draining/counting.
- Two real concurrent object-only Git fetches into one bare repository leave
  `FETCH_HEAD` absent, all requested objects readable, and `git fsck` clean.
- Tests cover a productive delayed hedge after an empty primary and prove a
  shared-object observation cannot become source preference.
- An active-output fixture survives a short test inactivity threshold. Silent
  TERM-resistant process groups are forcibly reaped; cancellation also kills
  descendants before follow-up work proceeds; an unproductive pair reaches a
  productive third source without three-way concurrency.

## Archive validation

The earlier `dfa9afc` burn-in demonstrated the hedge and foreground health but
also exposed that its persisted rankings were not attributable in a shared
object database. That recommendation was withdrawn.

Exact corrected tip `890321b867e8cf4765b1470b65c27b0ea4682526` ran on the
constrained disposable archive from 2026-08-10 21:57:38 UTC. In the first
9 minutes 47 seconds it:

- completed 325 repository Git passes and promoted 146 queued repositories;
- exercised 97 successful hedge advertisements and 87 successful hedge batch
  fetches, so the delayed alternative path was active under real load;
- recorded zero restarts, panics/OOMs, Git corruption signatures, or failed
  service health;
- reached 3.50 GiB peak memory and settled to 19 tasks while the existing
  resource-pressure gate paused new fan-out ten times rather than disrupting
  foreground work; and
- emitted one 60-second activity warning for a primary `relay.ngit.dev` batch
  that was still producing progress (`quiet_secs=0`, 20,835 stderr bytes).
  It was observed rather than killed, as intended for a potentially large
  legitimate fetch.

That run remains useful threshold evidence: the only 60-second warning had
`quiet_secs=0`, so the legitimate large fetch would not have approached the new
five-minute inactivity boundary. It predates process-group recovery, however.

Exact corrected tip `25c535f96aa9969543116a5c96a77aa5971130a6` then ran as
the control in the paired constrained archive from 2026-08-12 01:53:19 UTC. By
03:11 UTC it had completed 1,785 Git passes and fetched 2,985 requested OIDs.
The real workload exercised both sides of the hedge: 50 hedge batch fetches and
6 hedge residual fetches completed successfully, alongside 289 primary batch
and 168 primary residual successes. All subprocess gauges returned to zero.

Across that run the exact tip had zero service restarts, surviving zombie Git
processes, panic/OOM, or repository-corruption signatures. Memory peaked at
4.76 GB and returned to 1.42 GB while the service remained responsive; its
`/metrics` endpoint answered in 15.6 ms at the settled comparison. No
five-minute inactivity termination was needed in this workload. Earlier
threshold evidence had already shown a legitimate 60-second fetch continuing
to produce output, and the final implementation correctly leaves such active
work unbounded by total duration.

The burn also exposed one pre-existing synchronous local-OID-copy path outside
this proposal's observed network-fetch/process-group wrapper. A single zombie
was temporarily pinned behind that caller during an earlier refinement run,
did not accumulate, and cleared on the deliberate restart. It is not caused or
masked here and remains excluded from this bounded network-fetch change.

This exact-tip constrained production evidence, together with the deterministic
process-group and fallback tests, supports merging this proposal.

CHANGELOG.md

@@ -27,6 +27,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   filter-count and serialized-byte limits. Repository filter chunks are
   deterministic, and failed tail replacement restores the previous tail.
 
+### Fixed
+
+- Bound duplicate repository Git acquisition to one primary and one delayed
+  distinct-domain hedge, and recover from silent transports without imposing a
+  total fetch deadline. Continuously active large transfers remain unbounded;
+  a process group silent for five minutes is terminated gracefully then
+  forcibly, after which remaining sources are tried sequentially. Concurrent
+  object acquisition does not write `FETCH_HEAD`, run automatic maintenance,
+  or infer source speed from a shared object database.
+
 ## [2.1.2] - 2026-08-08
 
 ngit-grasp 2.1.2 is a patch release improving repository-event sync under

Cargo.lock

@@ -1334,6 +1334,7 @@ dependencies = [
  "hyper-util",
  "indexmap",
  "lazy_static",
+ "libc",
  "nostr",
  "nostr-lmdb",
  "nostr-memory",

Cargo.toml

@@ -37,6 +37,7 @@ base64 = "0.22"
 flate2 = "1.0"
 tar = "0.4"
 fs2 = "0.4"
+libc = "0.2"
 
 # Metrics
 prometheus = { version = "0.14", features = ["process"] }

docs/explanation/grasp-02-proactive-sync-purgatory-git-data.md

@@ -330,12 +330,13 @@ Attempt 4: Try repo-A (index=0) → ...
 
 ### Trigger-Based Processing (Not Polling)
 
-Domain queues **don't poll** for capacity. Instead, processing is triggered by two events:
+Domain queues **don't poll** for concurrent capacity. Instead, processing is triggered by two events:
 
-1. **`complete_request()`** - A request finishes, slot frees
+1. **Fetch-pass permit drop** - A request finishes or is cancelled, so its
+   cancellation-safe RAII permit releases the slot
 2. **`enqueue_identifier()`** - New identifier added to queue
 
-Both methods check `has_capacity()` and trigger `try_process_next()` if true.
+Both paths check `has_capacity()` and trigger `try_process_next()` if true.
 
 **Why trigger-based?**
 
@@ -345,6 +346,14 @@ Both methods check `has_capacity()` and trigger `try_process_next()` if true.
 
 **Implementation**: [`src/purgatory/sync/throttle.rs:ThrottleManager`](../../src/purgatory/sync/throttle.rs)
 
+The capacity check and reservation happen atomically under the domain's
+throttle lock. URL selection's earlier capacity observation is only a routing
+hint: the fetch pass must still acquire the permit before starting, so racing
+identifiers cannot all consume the final slot. One permit accounts for the
+whole current pass (`ls-remote` plus its fetch commands), preserving the
+existing per-pass rate semantics. It also owns process-wide pressure admission
+and drops that admission while waiting for a busy domain.
+
 ---
 
 ## Purgatory Expiry
@@ -425,6 +434,69 @@ advertisement is unchanged; any ref-tip change clears that URL's memo and
 allows them to be retried. Entries expire lazily after 30 minutes and the
 memo is capped at 1,024 URLs, evicting the oldest entry when full.
 
+All three phases run beneath the pass's per-domain permit. Separating the
+comparison into its own helper in future must retain that permit; `ls-remote`
+is outbound work against the same Git server, not an unaccounted preflight.
+
+### Repository single-flight and delayed hedging
+
+Remote politeness and local mutation safety have different keys. Domain
+permits protect each Git server; a repository coordinator keyed by the
+resolved local bare-repository path prevents independent identifier/domain
+queues from starting uncoordinated work in the same object database. Demand
+arriving during an active wave waits, then re-checks the still-missing OIDs, so
+already-satisfied work is coalesced while later OIDs are not lost. Promotion
+of newly available data is serialized by the same coordinator and duplicate
+OIDs from two attempts are processed once.
+
+Each wave starts the first source in the event's deterministic source order. If it is still running after a
+conservative 30-second head start, one source on a different domain may start
+as a hedge. The hedge delay is not a timeout: active transfers may run for any
+duration. A third concurrent attempt is never started, and another URL on the
+primary's domain does not qualify as a hedge. Both attempts independently
+retain process-pressure and per-domain admission. If neither attempt supplies
+an object—whether because it fails or is terminated after sustained
+inactivity—the remaining URLs are attempted sequentially.
+
+The wave does not infer a faster source from which command completes first.
+Both contenders write the same object database, so either command can observe
+objects installed by the other and completion order cannot truthfully
+attribute those objects to a source. Persistent source scoring therefore needs
+isolated per-attempt object stores (and explicit winner promotion) and remains
+out of scope.
+
+### Streamed subprocess observability
+
+Outbound `ls-remote` and `fetch` commands use Tokio child processes with
+stdout and stderr drained concurrently, rather than blocking until a complete
+`Output` is buffered. Fetch commands request Git's progress stream and every
+child runs with `LC_ALL=C`, making phase text stable enough for observation
+without treating it as a protocol. Capture is operation-bounded while readers
+continue draining and counting bytes beyond the bound: advertisements retain
+at most 16 MiB, ordinary stdout 1 MiB, and stderr/progress 4 MiB. A noisy child
+therefore cannot deadlock on a full pipe or grow memory without limit. A
+truncated advertisement is detected explicitly and falls back to the existing
+residual-OID path.
+
+Prometheus exposes active children, completion outcomes, durations, and
+drained bytes using only bounded `operation`, `role`, and `stream` labels.
+After 60 seconds an active child emits a structured warning containing its
+operation, primary/hedge role, elapsed time, time since its last output, and
+stream byte counts; it repeats no more than every five minutes. Five minutes
+without output is treated as a stalled transport, not as a total-duration
+limit. The direct Git child and its helpers run in a private process group;
+recovery sends that group SIGTERM, waits up to ten seconds, then sends SIGKILL
+so a helper cannot retain a pipe, socket, or repository lock. A transfer which
+continues producing output is never terminated, and completion timing is not
+used to rank sources. Cancellation keeps a synchronous process-group guard
+armed while the direct child still anchors the group identity, so dropping a
+fetch future kills its helpers before repository and domain permits return.
+
+Fetches request objects without writing `FETCH_HEAD` and suppress automatic
+maintenance. This keeps the object-only operation free of shared ref-state and
+maintenance side effects before repository-level delayed hedging permits a
+second source to write objects concurrently.
+
 **Why not just list every OID as a want?** An earlier implementation did
 exactly that and dropped one OID from the batch on each `not our ref`
 error. Against a state event declaring hundreds of tips that exist on no

src/metrics/mod.rs

@@ -249,6 +249,54 @@ lazy_static! {
             .expect("register purgatory git fetch oids metric");
         metric
     };
+    static ref PURGATORY_GIT_SUBPROCESS_ACTIVE: GaugeVec = {
+        let metric = GaugeVec::new(
+            Opts::new(
+                "ngit_purgatory_git_subprocess_active",
+                "Active outbound purgatory Git subprocesses by operation and hedge role",
+            ),
+            &["operation", "role"],
+        )
+        .expect("register purgatory Git subprocess active gauge");
+        REGISTRY.register(Box::new(metric.clone())).expect("register metric");
+        metric
+    };
+    static ref PURGATORY_GIT_SUBPROCESS_TOTAL: CounterVec = {
+        let metric = CounterVec::new(
+            Opts::new(
+                "ngit_purgatory_git_subprocess_total",
+                "Completed outbound purgatory Git subprocesses by operation, hedge role, and outcome",
+            ),
+            &["operation", "role", "outcome"],
+        )
+        .expect("register purgatory Git subprocess counter");
+        REGISTRY.register(Box::new(metric.clone())).expect("register metric");
+        metric
+    };
+    static ref PURGATORY_GIT_SUBPROCESS_DURATION_SECONDS: HistogramVec = {
+        let metric = HistogramVec::new(
+            HistogramOpts::new(
+                "ngit_purgatory_git_subprocess_duration_seconds",
+                "Outbound purgatory Git subprocess duration by operation and hedge role",
+            ),
+            &["operation", "role"],
+        )
+        .expect("register purgatory Git subprocess duration histogram");
+        REGISTRY.register(Box::new(metric.clone())).expect("register metric");
+        metric
+    };
+    static ref PURGATORY_GIT_SUBPROCESS_OUTPUT_BYTES_TOTAL: CounterVec = {
+        let metric = CounterVec::new(
+            Opts::new(
+                "ngit_purgatory_git_subprocess_output_bytes_total",
+                "Bytes drained from outbound purgatory Git subprocess output by operation, hedge role, and stream",
+            ),
+            &["operation", "role", "stream"],
+        )
+        .expect("register purgatory Git subprocess output counter");
+        REGISTRY.register(Box::new(metric.clone())).expect("register metric");
+        metric
+    };
     static ref TRANSIENT_REQ_WATCHDOG_TOTAL: CounterVec = {
         let metric = CounterVec::new(
             Opts::new(
@@ -303,6 +351,63 @@ pub fn record_purgatory_git_fetch_pass(
         .inc_by(fetched as f64);
 }
 
+pub struct PurgatoryGitSubprocessGuard {
+    operation: &'static str,
+    role: &'static str,
+    started: Instant,
+    outcome: &'static str,
+}
+
+impl PurgatoryGitSubprocessGuard {
+    pub fn finish(&mut self, success: bool) {
+        self.outcome = if success { "success" } else { "failure" };
+    }
+
+    pub fn finish_with_outcome(&mut self, outcome: &'static str) {
+        self.outcome = outcome;
+    }
+}
+
+impl Drop for PurgatoryGitSubprocessGuard {
+    fn drop(&mut self) {
+        PURGATORY_GIT_SUBPROCESS_ACTIVE
+            .with_label_values(&[self.operation, self.role])
+            .dec();
+        PURGATORY_GIT_SUBPROCESS_TOTAL
+            .with_label_values(&[self.operation, self.role, self.outcome])
+            .inc();
+        PURGATORY_GIT_SUBPROCESS_DURATION_SECONDS
+            .with_label_values(&[self.operation, self.role])
+            .observe(self.started.elapsed().as_secs_f64());
+    }
+}
+
+pub fn start_purgatory_git_subprocess(
+    operation: &'static str,
+    role: &'static str,
+) -> PurgatoryGitSubprocessGuard {
+    PURGATORY_GIT_SUBPROCESS_ACTIVE
+        .with_label_values(&[operation, role])
+        .inc();
+    PurgatoryGitSubprocessGuard {
+        operation,
+        role,
+        started: Instant::now(),
+        outcome: "cancelled",
+    }
+}
+
+pub fn record_purgatory_git_subprocess_output(
+    operation: &str,
+    role: &str,
+    stream: &str,
+    bytes: u64,
+) {
+    PURGATORY_GIT_SUBPROCESS_OUTPUT_BYTES_TOTAL
+        .with_label_values(&[operation, role, stream])
+        .inc_by(bytes as f64);
+}
+
 pub fn record_blacklist_deletion_attempt(phase: &str) {
     BLACKLIST_DELETIONS_TOTAL
         .with_label_values(&[phase, "attempted"])

src/purgatory/sync/functions.rs

@@ -15,10 +15,21 @@ use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
 use tracing::debug;
 
-use super::context::SyncContext;
+use super::context::{GitFetchRole, SyncContext};
 use super::throttle::ThrottleManager;
 use crate::sync::naughty_list::NaughtyListTracker;
 
+/// A primary gets a substantial head start before one alternative server is
+/// allowed to duplicate the same object demand. This is a hedge trigger, not
+/// a timeout: neither legitimate long-running fetch is killed.
+const REPOSITORY_FETCH_HEDGE_DELAY: std::time::Duration = std::time::Duration::from_secs(30);
+
+#[derive(Default)]
+struct RepositoryFetchOutcome {
+    fetched: usize,
+    attempted_urls: Vec<String>,
+}
+
 /// Extract domain from a URL.
 ///
 /// Supports HTTP(S) URLs. SSH URLs (git@...) are not supported.
@@ -324,18 +335,63 @@ pub async fn sync_identifier_from_url<C: SyncContext + ?Sized>(
     url: &str,
     throttle_manager: &Arc<ThrottleManager>,
 ) -> usize {
+    sync_identifier_from_urls(ctx, identifier, vec![url.to_string()], throttle_manager)
+        .await
+        .fetched
+}
+
+async fn fetch_from_url<C: SyncContext + ?Sized>(
+    ctx: &C,
+    repo: &std::path::Path,
+    url: &str,
+    needed_oids: &[String],
+    throttle_manager: &Arc<ThrottleManager>,
+    role: GitFetchRole,
+) -> Vec<String> {
     let domain = match extract_domain(url) {
         Some(d) => d,
-        None => {
-            debug!(
-                identifier = %identifier,
-                url = %url,
-                "Could not extract domain from URL"
-            );
-            return 0;
-        }
+        None => return Vec::new(),
     };
 
+    let _fetch_permit = throttle_manager.acquire_fetch_pass(&domain).await;
+    match ctx.fetch_oids_with_role(repo, url, needed_oids, role).await {
+        Ok(fetched) => fetched,
+        Err(error) => {
+            debug!(url = %url, %error, "Fetch failed");
+            Vec::new()
+        }
+    }
+}
+
+async fn process_unique_fetched<C: SyncContext + ?Sized>(
+    ctx: &C,
+    repo: &std::path::Path,
+    fetched: Vec<String>,
+    processed: &mut HashSet<String>,
+) -> usize {
+    let new_oids: HashSet<String> = fetched
+        .into_iter()
+        .filter(|oid| !processed.contains(oid))
+        .collect();
+    if new_oids.is_empty() {
+        return 0;
+    }
+    if let Err(error) = ctx.process_newly_available_git_data(repo, &new_oids).await {
+        debug!(%error, "Failed to process newly available git data");
+        return 0;
+    }
+    processed.extend(new_oids.iter().cloned());
+    new_oids.len()
+}
+
+async fn sync_identifier_from_urls<C: SyncContext + ?Sized>(
+    ctx: &C,
+    identifier: &str,
+    mut urls: Vec<String>,
+    throttle_manager: &Arc<ThrottleManager>,
+) -> RepositoryFetchOutcome {
+    let mut outcome = RepositoryFetchOutcome::default();
+
     // Get repository data for target repo path
     let repo_data = match ctx.fetch_repository_data_with_purgatory(identifier).await {
         Ok(data) => data,
@@ -345,7 +401,7 @@ pub async fn sync_identifier_from_url<C: SyncContext + ?Sized>(
                 error = %e,
                 "Failed to fetch repo data"
             );
-            return 0;
+            return outcome;
         }
     };
 
@@ -353,10 +409,16 @@ pub async fn sync_identifier_from_url<C: SyncContext + ?Sized>(
         Some(path) => path,
         None => {
             debug!(identifier = %identifier, "No target repo found");
-            return 0;
+            return outcome;
         }
     };
 
+    // The path, rather than the identifier or remote domain, is the local
+    // mutation boundary. A waiting caller re-checks demand after the active
+    // wave, coalescing duplicate notifications without losing newly arrived
+    // OIDs.
+    let _repository = throttle_manager.coordinate_repository(&target_repo).await;
+
     // Collect needed OIDs
     let needed_oids: Vec<String> = ctx.collect_needed_oids(identifier).into_iter().collect();
     if needed_oids.is_empty() {
@@ -364,65 +426,101 @@ pub async fn sync_identifier_from_url<C: SyncContext + ?Sized>(
             identifier = %identifier,
             "No OIDs needed - nothing to fetch"
         );
-        return 0;
+        return outcome;
     }
 
-    // Per-domain limits alone do not bound aggregate work when a cold sync
-    // discovers many domains simultaneously. Wait for the process-wide permit
-    // before occupying a domain slot, then retain it for the complete remote
-    // Git pass so subprocess fan-out stays bounded.
-    let _background_permit = throttle_manager.acquire_global_request().await;
-
-    // Perform the fetch with per-domain throttle tracking.
-    throttle_manager.start_request(&domain);
-    let fetch_result = ctx.fetch_oids(&target_repo, url, &needed_oids).await;
-    throttle_manager.complete_request(&domain);
-
-    let fetched_oids = match fetch_result {
-        Ok(fetched) if !fetched.is_empty() => {
-            debug!(
-                identifier = %identifier,
-                url = %url,
-                oids_fetched = fetched.len(),
-                "Fetch succeeded"
-            );
-            fetched
-        }
-        Ok(_) => {
-            debug!(
-                identifier = %identifier,
-                url = %url,
-                "Fetch returned no OIDs (not available on remote)"
-            );
-            vec![]
-        }
-        Err(e) => {
-            debug!(
-                identifier = %identifier,
-                url = %url,
-                error = %e,
-                "Fetch failed"
-            );
-            vec![]
-        }
+    urls.retain(|url| extract_domain(url).is_some());
+    let Some(primary_url) = urls.first().cloned() else {
+        return outcome;
     };
+    let primary_domain = extract_domain(&primary_url).unwrap();
+    let hedge_url = urls
+        .iter()
+        .skip(1)
+        .find(|url| extract_domain(url).as_deref() != Some(primary_domain.as_str()))
+        .cloned();
 
-    // Try to process any events that can now be satisfied
-    if !fetched_oids.is_empty() {
-        let new_oids: HashSet<String> = fetched_oids.iter().cloned().collect();
-        if let Err(e) = ctx
-            .process_newly_available_git_data(&target_repo, &new_oids)
-            .await
-        {
-            debug!(
-                identifier = %identifier,
-                error = %e,
-                "Failed to process newly available git data"
-            );
+    outcome.attempted_urls.push(primary_url.clone());
+    let primary = fetch_from_url(
+        ctx,
+        &target_repo,
+        &primary_url,
+        &needed_oids,
+        throttle_manager,
+        GitFetchRole::Primary,
+    );
+    tokio::pin!(primary);
+    let mut processed = HashSet::new();
+
+    let delay = tokio::time::sleep(REPOSITORY_FETCH_HEDGE_DELAY);
+    tokio::pin!(delay);
+    tokio::select! {
+        biased;
+        fetched = &mut primary => {
+            outcome.fetched += process_unique_fetched(ctx, &target_repo, fetched, &mut processed).await;
+            return outcome;
+        }
+        _ = &mut delay => {}
+    }
+
+    let Some(hedge_url) = hedge_url else {
+        let fetched = primary.await;
+        outcome.fetched += process_unique_fetched(ctx, &target_repo, fetched, &mut processed).await;
+        return outcome;
+    };
+    outcome.attempted_urls.push(hedge_url.clone());
+    let hedge = fetch_from_url(
+        ctx,
+        &target_repo,
+        &hedge_url,
+        &needed_oids,
+        throttle_manager,
+        GitFetchRole::Hedge,
+    );
+    tokio::pin!(hedge);
+
+    tokio::select! {
+        biased;
+        fetched = &mut primary => {
+            outcome.fetched += process_unique_fetched(ctx, &target_repo, fetched, &mut processed).await;
+            let fetched = hedge.await;
+            outcome.fetched += process_unique_fetched(ctx, &target_repo, fetched, &mut processed).await;
+        }
+        fetched = &mut hedge => {
+            outcome.fetched += process_unique_fetched(ctx, &target_repo, fetched, &mut processed).await;
+            let fetched = primary.await;
+            outcome.fetched += process_unique_fetched(ctx, &target_repo, fetched, &mut processed).await;
         }
     }
 
-    fetched_oids.len()
+    // A stalled subprocess is returned as an ordinary failed attempt after
+    // its process group has been reaped. If neither member of the bounded
+    // two-source wave supplied an object, keep walking the remaining sources
+    // sequentially rather than leaving this repository pinned to the pair.
+    // No third fetch overlaps the primary and hedge.
+    if outcome.fetched == 0 {
+        for fallback_url in urls.into_iter().skip(1) {
+            if fallback_url == hedge_url {
+                continue;
+            }
+            outcome.attempted_urls.push(fallback_url.clone());
+            let fetched = fetch_from_url(
+                ctx,
+                &target_repo,
+                &fallback_url,
+                &needed_oids,
+                throttle_manager,
+                GitFetchRole::Primary,
+            )
+            .await;
+            outcome.fetched +=
+                process_unique_fetched(ctx, &target_repo, fetched, &mut processed).await;
+            if outcome.fetched > 0 {
+                break;
+            }
+        }
+    }
+    outcome
 }
 
 /// Sync git data for an identifier.
@@ -477,9 +575,28 @@ pub async fn sync_identifier<C: SyncContext + ?Sized>(
                     "Found non-throttled URL to try"
                 );
 
-                // Fetch from this URL
-                sync_identifier_from_url(ctx, identifier, &url, throttle_manager).await;
-                tried_urls.insert(url);
+                // Build one wave from currently admissible sources. The
+                // coordinator starts the preferred primary immediately and
+                // at most one distinct-domain hedge after its head start.
+                let mut wave_urls = vec![url.clone()];
+                let mut discovered = tried_urls.clone();
+                discovered.insert(url);
+                while let Some(candidate) = sync_identifier_next_url(
+                    ctx,
+                    identifier,
+                    None,
+                    &discovered,
+                    throttle_manager,
+                    git_naughty_list,
+                )
+                .await
+                {
+                    discovered.insert(candidate.clone());
+                    wave_urls.push(candidate);
+                }
+                let outcome =
+                    sync_identifier_from_urls(ctx, identifier, wave_urls, throttle_manager).await;
+                tried_urls.extend(outcome.attempted_urls);
 
                 // Check if sync is now complete
                 if !ctx.has_pending_events(identifier) {
@@ -838,6 +955,276 @@ mod tests {
         assert!(!throttle_manager.is_throttled("github.com"));
     }
 
+    #[tokio::test]
+    async fn duplicate_repo_demand_joins_one_fetch_wave() {
+        let url = "https://primary.example/repo.git";
+        let mock = Arc::new(
+            MockSyncContext::new()
+                .with_needed_oids(&["abc123"])
+                .url_provides(url, &["abc123"])
+                .url_waits_for_release(url),
+        );
+        let manager = Arc::new(ThrottleManager::new(5, 100));
+
+        let first_mock = mock.clone();
+        let first_manager = manager.clone();
+        let first = tokio::spawn(async move {
+            sync_identifier_from_url(&*first_mock, "repo", url, &first_manager).await
+        });
+        mock.wait_for_fetches(1).await;
+
+        let second_mock = mock.clone();
+        let second_manager = manager.clone();
+        let second = tokio::spawn(async move {
+            sync_identifier_from_url(&*second_mock, "repo", url, &second_manager).await
+        });
+        tokio::task::yield_now().await;
+        assert_eq!(mock.fetch_log().len(), 1);
+
+        mock.release_fetch(url);
+        assert_eq!(first.await.unwrap(), 1);
+        assert_eq!(second.await.unwrap(), 0);
+        assert_eq!(mock.fetch_log().len(), 1);
+        assert_eq!(mock.process_call_count(), 1);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn delayed_distinct_domain_hedge_can_supply_the_oid() {
+        let primary = "https://a.example/repo.git";
+        let hedge = "https://b.example/repo.git";
+        let mock = Arc::new(
+            MockSyncContext::new()
+                .with_needed_oids(&["abc123"])
+                .url_provides(primary, &["abc123"])
+                .url_provides(hedge, &["abc123"])
+                .url_waits_for_release(primary),
+        );
+        let manager = Arc::new(ThrottleManager::new(5, 100));
+        let task_mock = mock.clone();
+        let task_manager = manager.clone();
+        let task = tokio::spawn(async move {
+            sync_identifier_from_urls(
+                &*task_mock,
+                "repo",
+                vec![primary.to_string(), hedge.to_string()],
+                &task_manager,
+            )
+            .await
+        });
+        mock.wait_for_fetches(1).await;
+        tokio::time::advance(REPOSITORY_FETCH_HEDGE_DELAY).await;
+        mock.wait_for_fetches(2).await;
+        assert_eq!(mock.process_call_count(), 1, "hedge should supply the OID");
+        mock.release_fetch(primary);
+        assert_eq!(task.await.unwrap().fetched, 1);
+        assert_eq!(mock.process_call_count(), 1, "same OID is promoted once");
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn empty_primary_does_not_hide_a_productive_hedge() {
+        let primary = "https://a.example/repo.git";
+        let productive_hedge = "https://b.example/repo.git";
+        let mock = Arc::new(
+            MockSyncContext::new()
+                .with_needed_oids(&["abc123"])
+                .url_provides(productive_hedge, &["abc123"])
+                .url_waits_for_release(primary),
+        );
+        let manager = Arc::new(ThrottleManager::new(5, 100));
+        let task_mock = mock.clone();
+        let task_manager = manager.clone();
+        let task = tokio::spawn(async move {
+            sync_identifier_from_urls(
+                &*task_mock,
+                "repo",
+                vec![primary.to_string(), productive_hedge.to_string()],
+                &task_manager,
+            )
+            .await
+        });
+        mock.wait_for_fetches(1).await;
+        tokio::time::advance(REPOSITORY_FETCH_HEDGE_DELAY).await;
+        mock.wait_for_fetches(2).await;
+        mock.release_fetch(primary);
+        assert_eq!(task.await.unwrap().fetched, 1);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn failed_pair_falls_back_to_a_third_source() {
+        let primary = "https://a.example/repo.git";
+        let hedge = "https://b.example/repo.git";
+        let fallback = "https://c.example/repo.git";
+        let mock = Arc::new(
+            MockSyncContext::new()
+                .with_needed_oids(&["abc123"])
+                .url_should_fail(primary)
+                .url_should_fail(hedge)
+                .url_waits_for_release(primary)
+                .url_provides(fallback, &["abc123"]),
+        );
+        let manager = Arc::new(ThrottleManager::new(5, 100));
+        let task_mock = mock.clone();
+        let task_manager = manager.clone();
+        let task = tokio::spawn(async move {
+            sync_identifier_from_urls(
+                &*task_mock,
+                "repo",
+                vec![primary.to_string(), hedge.to_string(), fallback.to_string()],
+                &task_manager,
+            )
+            .await
+        });
+        mock.wait_for_fetches(1).await;
+        tokio::time::advance(REPOSITORY_FETCH_HEDGE_DELAY).await;
+        mock.wait_for_fetches(2).await;
+        mock.release_fetch(primary);
+
+        let outcome = task.await.unwrap();
+        assert_eq!(outcome.fetched, 1);
+        assert_eq!(
+            outcome.attempted_urls,
+            vec![primary.to_string(), hedge.to_string(), fallback.to_string()]
+        );
+        assert_eq!(mock.process_call_count(), 1);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn shared_object_observation_is_not_treated_as_source_attribution() {
+        let primary = "https://a.example/repo.git";
+        let observer = "https://b.example/repo.git";
+        let mock = Arc::new(
+            MockSyncContext::new()
+                .with_needed_oids(&["abc123"])
+                .url_provides(primary, &["abc123"])
+                .url_waits_after_fetch(primary)
+                .url_reports_shared_objects(observer),
+        );
+        let manager = Arc::new(ThrottleManager::new(5, 100));
+        let task_mock = mock.clone();
+        let task_manager = manager.clone();
+        let task = tokio::spawn(async move {
+            sync_identifier_from_urls(
+                &*task_mock,
+                "repo",
+                vec![primary.to_string(), observer.to_string()],
+                &task_manager,
+            )
+            .await
+        });
+
+        mock.wait_for_fetches(1).await;
+        tokio::time::advance(REPOSITORY_FETCH_HEDGE_DELAY).await;
+        mock.wait_for_fetches(2).await;
+        mock.release_fetch(primary);
+        assert_eq!(task.await.unwrap().fetched, 1);
+        assert_eq!(mock.process_call_count(), 1);
+        // The observer returned the primary's shared object. There is
+        // deliberately no winner score or reorder API to receive that
+        // ambiguous result.
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn two_slow_fetches_never_run_a_third_concurrently() {
+        let first = "https://a.example/repo.git";
+        let second = "https://b.example/repo.git";
+        let third = "https://c.example/repo.git";
+        let mock = Arc::new(
+            MockSyncContext::new()
+                .with_needed_oids(&["abc123"])
+                .url_waits_for_release(first)
+                .url_waits_for_release(second),
+        );
+        let manager = Arc::new(ThrottleManager::new(5, 100));
+        let task_mock = mock.clone();
+        let task_manager = manager.clone();
+        let task = tokio::spawn(async move {
+            sync_identifier_from_urls(
+                &*task_mock,
+                "repo",
+                vec![first.to_string(), second.to_string(), third.to_string()],
+                &task_manager,
+            )
+            .await
+        });
+        mock.wait_for_fetches(1).await;
+        tokio::time::advance(REPOSITORY_FETCH_HEDGE_DELAY).await;
+        mock.wait_for_fetches(2).await;
+        tokio::time::advance(REPOSITORY_FETCH_HEDGE_DELAY * 10).await;
+        assert_eq!(mock.fetch_log().len(), 2);
+        mock.release_fetch(first);
+        mock.release_fetch(second);
+        task.await.unwrap();
+        assert!(mock.fetch_log().iter().any(|url| url == third));
+    }
+
+    #[tokio::test]
+    async fn different_repository_paths_fetch_independently() {
+        let first_url = "https://a.example/one.git";
+        let second_url = "https://b.example/two.git";
+        let first = Arc::new(
+            MockSyncContext::new()
+                .with_target_repo("/tmp/repo-one")
+                .with_needed_oids(&["one"])
+                .url_waits_for_release(first_url),
+        );
+        let second = Arc::new(
+            MockSyncContext::new()
+                .with_target_repo("/tmp/repo-two")
+                .with_needed_oids(&["two"])
+                .url_waits_for_release(second_url),
+        );
+        let manager = Arc::new(ThrottleManager::new(5, 100));
+
+        let first_task = {
+            let ctx = first.clone();
+            let manager = manager.clone();
+            tokio::spawn(async move {
+                sync_identifier_from_url(&*ctx, "one", first_url, &manager).await
+            })
+        };
+        let second_task = {
+            let ctx = second.clone();
+            let manager = manager.clone();
+            tokio::spawn(async move {
+                sync_identifier_from_url(&*ctx, "two", second_url, &manager).await
+            })
+        };
+        first.wait_for_fetches(1).await;
+        second.wait_for_fetches(1).await;
+        first.release_fetch(first_url);
+        second.release_fetch(second_url);
+        first_task.await.unwrap();
+        second_task.await.unwrap();
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn same_domain_alternative_is_not_a_hedge() {
+        let primary = "https://same.example/one.git";
+        let alternative = "https://same.example/two.git";
+        let mock = Arc::new(
+            MockSyncContext::new()
+                .with_needed_oids(&["abc123"])
+                .url_waits_for_release(primary),
+        );
+        let manager = Arc::new(ThrottleManager::new(5, 100));
+        let task_mock = mock.clone();
+        let task_manager = manager.clone();
+        let task = tokio::spawn(async move {
+            sync_identifier_from_urls(
+                &*task_mock,
+                "repo",
+                vec![primary.to_string(), alternative.to_string()],
+                &task_manager,
+            )
+            .await
+        });
+        mock.wait_for_fetches(1).await;
+        tokio::time::advance(REPOSITORY_FETCH_HEDGE_DELAY * 2).await;
+        assert_eq!(mock.fetch_log(), vec![primary.to_string()]);
+        mock.release_fetch(primary);
+        task.await.unwrap();
+    }
+
     #[tokio::test]
     async fn get_throttled_domains_returns_only_throttled_with_untried() {
         let mock = MockSyncContext::new()

src/purgatory/sync/throttle.rs

@@ -18,9 +18,10 @@
 use dashmap::DashMap;
 use indexmap::IndexMap;
 use std::collections::{HashSet, VecDeque};
-use std::sync::{Arc, Mutex, OnceLock};
+use std::path::{Path, PathBuf};
+use std::sync::{Arc, Mutex, OnceLock, Weak};
 use std::time::{Duration, Instant};
-use tokio::sync::Notify;
+use tokio::sync::{Mutex as AsyncMutex, Notify, OwnedMutexGuard};
 use tracing::debug;
 
 use super::context::SyncContext;
@@ -39,6 +40,7 @@ const PRESSURE_RECHECK_INTERVAL: Duration = Duration::from_millis(100);
 // Sustained CPU throttling can alternate healthy and pressured samples. Keep
 // the operational signal without turning that expected oscillation into spam.
 const PRESSURE_LOG_INTERVAL: Duration = Duration::from_secs(60);
+const REPOSITORY_COORDINATOR_CLEANUP_THRESHOLD: usize = 4096;
 
 fn grow_healthy_allowance(current: usize) -> usize {
     // Pressure counters describe work that has already run. Grow gradually so
@@ -347,6 +349,35 @@ impl DomainThrottle {
         recent_count < self.max_per_minute as usize
     }
 
+    /// Reserve capacity while holding the domain mutex.
+    ///
+    /// Combining the capacity check and accounting increment closes the race
+    /// where several callers could all observe the last slot before any of
+    /// them recorded their request.
+    fn try_start_request(&mut self) -> bool {
+        self.cleanup_request_times();
+        if !self.has_capacity() {
+            return false;
+        }
+        self.start_request();
+        true
+    }
+
+    /// Time until rate-window capacity can become available. `None` means the
+    /// caller is blocked only by concurrent work and should await a release.
+    fn retry_after(&mut self) -> Option<Duration> {
+        self.cleanup_request_times();
+        if self.in_flight >= self.max_concurrent {
+            return None;
+        }
+        if self.request_times.len() < self.max_per_minute as usize {
+            return Some(Duration::ZERO);
+        }
+        self.request_times.front().map(|started| {
+            Duration::from_secs(60).saturating_sub(Instant::now().duration_since(*started))
+        })
+    }
+
     /// Check if there are any identifiers in the queue.
     pub fn has_queued_work(&self) -> bool {
         !self.queue.is_empty()
@@ -366,7 +397,10 @@ impl DomainThrottle {
     pub fn complete_request(&mut self) {
         self.in_flight = self.in_flight.saturating_sub(1);
 
-        // Clean old timestamps outside the 60-second window
+        self.cleanup_request_times();
+    }
+
+    fn cleanup_request_times(&mut self) {
         let now = Instant::now();
         let window = Duration::from_secs(60);
         while self
@@ -485,6 +519,15 @@ pub struct ThrottleManager {
     /// Resource-pressure admission gate shared across every remote domain.
     pressure_gate: Arc<BackgroundPressureGate>,
 
+    /// Wakes fetch passes waiting for a per-domain slot. The domain state is
+    /// still checked under its mutex, so a shared notifier is sufficient and
+    /// a wake for another domain is only a harmless extra check.
+    domain_capacity: Arc<Notify>,
+
+    /// One coordinator lock per local object database. Waiting callers join
+    /// the active wave and re-check demand after it completes.
+    repository_fetches: DashMap<PathBuf, Weak<AsyncMutex<()>>>,
+
     /// Sync context for processing queued identifiers.
     /// Set once at startup via `set_context()`.
     ctx: OnceLock<Arc<dyn SyncContext>>,
@@ -494,6 +537,19 @@ pub struct ThrottleManager {
     git_naughty_list: OnceLock<Arc<NaughtyListTracker>>,
 }
 
+/// Cancellation-safe ownership of one remote Git pass's admission.
+pub(super) struct FetchPassPermit {
+    manager: Arc<ThrottleManager>,
+    domain: String,
+    _global: BackgroundPressurePermit,
+}
+
+impl Drop for FetchPassPermit {
+    fn drop(&mut self) {
+        self.manager.release_domain_request(&self.domain);
+    }
+}
+
 impl ThrottleManager {
     /// Create a new throttle manager with the specified limits.
     ///
@@ -512,15 +568,76 @@ impl ThrottleManager {
             max_concurrent_per_domain: max_concurrent,
             max_per_minute_per_domain: max_per_minute,
             pressure_gate,
+            domain_capacity: Arc::new(Notify::new()),
+            repository_fetches: DashMap::new(),
             ctx: OnceLock::new(),
             git_naughty_list: OnceLock::new(),
         }
     }
 
-    /// Wait for process-wide background Git capacity.
+    pub(super) async fn coordinate_repository(&self, path: &Path) -> OwnedMutexGuard<()> {
+        if self.repository_fetches.len() >= REPOSITORY_COORDINATOR_CLEANUP_THRESHOLD {
+            self.repository_fetches
+                .retain(|_, lock| lock.strong_count() > 0);
+        }
+        let mut entry = self
+            .repository_fetches
+            .entry(path.to_path_buf())
+            .or_insert_with(Weak::new);
+        let lock = match entry.upgrade() {
+            Some(lock) => lock,
+            None => {
+                let lock = Arc::new(AsyncMutex::new(()));
+                *entry = Arc::downgrade(&lock);
+                lock
+            }
+        };
+        drop(entry);
+        lock.lock_owned().await
+    }
+
+    /// Atomically admit one complete remote Git pass.
     ///
-    /// The owned permit spans the complete `fetch_oids` pass (ls-remote plus
-    /// any fetches) and releases automatically on every return or cancellation.
+    /// The returned permit spans `ls-remote` and every fetch needed by the
+    /// pass. It owns both process-wide pressure admission and the per-domain
+    /// reservation, and releases both on every return, panic, or cancellation.
+    /// Keeping this as the sole admission API also ensures that extracting the
+    /// `ls-remote` phase cannot accidentally bypass domain accounting.
+    pub(super) async fn acquire_fetch_pass(self: &Arc<Self>, domain: &str) -> FetchPassPermit {
+        loop {
+            // Register before checking capacity so a release between the
+            // failed check and the await cannot be lost.
+            let notified = self.domain_capacity.notified();
+            let global = self.pressure_gate.acquire().await;
+
+            let retry_after = {
+                let entry = self.get_or_create_throttle(domain);
+                let mut throttle = entry.lock().unwrap();
+                if throttle.try_start_request() {
+                    return FetchPassPermit {
+                        manager: self.clone(),
+                        domain: domain.to_string(),
+                        _global: global,
+                    };
+                }
+                throttle.retry_after()
+            };
+
+            // Do not pin scarce process-wide capacity while this particular
+            // remote domain is full.
+            drop(global);
+            if let Some(delay) = retry_after {
+                tokio::select! {
+                    _ = notified => {}
+                    _ = tokio::time::sleep(delay) => {}
+                }
+            } else {
+                notified.await;
+            }
+        }
+    }
+
+    #[cfg(test)]
     pub(super) async fn acquire_global_request(&self) -> BackgroundPressurePermit {
         self.pressure_gate.acquire().await
     }
@@ -582,9 +699,9 @@ impl ThrottleManager {
         self.throttles.get(domain).unwrap()
     }
 
-    /// Record that a request is starting for a domain.
-    ///
-    /// Increments in-flight count and records timestamp for rate limiting.
+    /// Direct accounting hook retained only for tests that arrange a domain's
+    /// pre-existing saturation before exercising URL selection.
+    #[cfg(test)]
     pub fn start_request(&self, domain: &str) {
         let entry = self.get_or_create_throttle(domain);
         let mut throttle = entry.lock().unwrap();
@@ -610,7 +727,7 @@ impl ThrottleManager {
     ///
     /// # Arguments
     /// * `domain` - The domain that completed a request
-    pub fn complete_request(self: &Arc<Self>, domain: &str) {
+    fn release_domain_request(self: &Arc<Self>, domain: &str) {
         let should_trigger = {
             if let Some(entry) = self.throttles.get(domain) {
                 let mut throttle = entry.lock().unwrap();
@@ -621,6 +738,8 @@ impl ThrottleManager {
             }
         };
 
+        self.domain_capacity.notify_waiters();
+
         if should_trigger {
             self.try_process_next(domain);
         }
@@ -1006,6 +1125,56 @@ mod tests {
         assert!(manager.throttles.contains_key("example.com"));
     }
 
+    #[tokio::test]
+    async fn fetch_pass_admission_atomically_holds_the_domain_slot() {
+        let manager = Arc::new(ThrottleManager::new(1, 100));
+        let first = manager.acquire_fetch_pass("example.com").await;
+
+        let waiting_manager = manager.clone();
+        let mut waiting =
+            tokio::spawn(async move { waiting_manager.acquire_fetch_pass("example.com").await });
+        assert!(
+            tokio::time::timeout(Duration::from_millis(50), &mut waiting)
+                .await
+                .is_err(),
+            "a second pass must not race through the same domain's last slot"
+        );
+
+        drop(first);
+        let _second = tokio::time::timeout(Duration::from_secs(1), waiting)
+            .await
+            .expect("dropping the first permit should wake the waiter")
+            .expect("waiter task should finish");
+
+        let throttle = manager.throttles.get("example.com").unwrap();
+        assert_eq!(throttle.lock().unwrap().request_times.len(), 2);
+    }
+
+    #[tokio::test]
+    async fn cancelling_fetch_pass_releases_domain_admission() {
+        let manager = Arc::new(ThrottleManager::new(1, 100));
+        let (acquired_tx, acquired_rx) = tokio::sync::oneshot::channel();
+        let holding_manager = manager.clone();
+        let holding = tokio::spawn(async move {
+            let _permit = holding_manager.acquire_fetch_pass("example.com").await;
+            let _ = acquired_tx.send(());
+            std::future::pending::<()>().await;
+        });
+        acquired_rx.await.expect("holder should acquire the permit");
+
+        holding.abort();
+        holding
+            .await
+            .expect_err("the holding task should be cancelled");
+
+        let _replacement = tokio::time::timeout(
+            Duration::from_secs(1),
+            manager.acquire_fetch_pass("example.com"),
+        )
+        .await
+        .expect("cancellation must return domain capacity");
+    }
+
     #[tokio::test]
     async fn initial_probe_bounds_work_before_pressure_is_observable() {
         let manager = Arc::new(ThrottleManager::new(5, 100));