fix(sync): drain mailbox history per relay

The global mailbox rotation made productive relays wait behind the full discovered relay inventory between every filter group. Fast relays therefore took hours to traverse a cycle even when each individual request completed in seconds.

Keep one worker per relay and immediately continue that relay after a successful partial cycle. Admit new relay workers one per maintenance pass and retain a 32-worker process-wide safety ceiling so peer-controlled NIP-65 inventories cannot create unbounded response accumulation.

Failure backoff, completed-cycle refresh, request pacing, pagination, subscription-ledger accounting, persistence policy, and connection retirement remain unchanged. This deliberately does not add physical connection sharding or reduce mailbox discovery scope.

Validated with cargo fmt --check, cargo clippy --all-targets --all-features -- -D warnings, the full cargo test workspace suite, focused owner-inbox live-sync coverage, and git diff --check.

docs/explanation/architecture.md

@@ -682,7 +682,9 @@ advertises `GRASP-03` only while the overlay is enabled. Exact thread
 provenance scopes each mailbox to its accepted roots. Accepted root authors'
 read/unmarked inboxes retain ordinary live/rotating GRASP-02 coverage. Wider
 participant mailboxes use independent, history-only `fetch_events` workers:
-at most one per relay and one new start per maintenance pass. They reuse the
+at most one per relay, 32 process-wide, and one newly admitted relay per
+maintenance pass. Once admitted, a successful relay drains its own stable
+filter cursor without waiting for a global relay rotation. They reuse the
 connection's ordinary pacing, subscription ledger, pagination and 30-second
 per-page terminal timeout without installing permanent participant
 subscriptions or coupling progress between relays. On public instances, each

docs/explanation/grasp-03-proactive-sync-plus.md

@@ -88,10 +88,12 @@ that retained state on the normal cadence.
 
 Root-author and repository owner/maintainer read/unmarked inboxes use ordinary
 GRASP-02 coverage. Wider participant mailboxes and owner/maintainer write-only
-outboxes use history-only workers. A maintenance pass starts at most one due
-history relay, while each relay may have at most one worker in flight. Relays
-do not share a mailbox lane or terminal state: a slow or unavailable relay
-cannot prevent another relay from progressing on a later pass.
+outboxes use history-only workers. A maintenance pass admits at most one new
+due history relay, while each admitted relay may have at most one worker in
+flight and drains its own filter cursor without returning to a global
+round-robin between successful groups. Up to 32 relay workers may run at once
+as a process-safety ceiling. Relays do not share a mailbox lane or terminal
+state: a slow or unavailable relay consumes only its own worker slot.
 
 Each worker selects one stable-sorted, byte-bounded filter and delegates its
 REQ lifecycle to the existing `RelayConnection::fetch_events` path. That path
@@ -99,9 +101,11 @@ owns per-relay request pacing, background priority, subscription-ledger
 capacity, EOSE/CLOSED handling and a 30-second timeout. The worker reuses the
 ordinary pagination state to continue through full historic pages until the
 filter is exhausted or a page makes no new progress. It then passes events
-through the normal write policy and persistence pipeline. No mailbox-specific
-pending-batch kind, EOSE hook, close API, watchdog or cross-relay coordinator
-is added.
+through the normal write policy and persistence pipeline. A successful group
+immediately starts that relay's next group until the cycle completes; a failed
+group advances the cursor but releases the worker for the five-minute retry
+delay. No mailbox-specific pending-batch kind, EOSE hook, close API, watchdog
+or cross-relay coordinator is added.
 
 Coverage uses accepted repository coordinates, root IDs, bounded recursive
 descendant IDs, and descendant address coordinates with `a`/`A`/`q` and
@@ -125,10 +129,11 @@ roots assigned to every mailbox. Shared relays and duplicate repository scopes
 are unioned before filter construction. Canonical live filters consume normal
 GRASP-02 subscription-ledger capacity; compatibility variants and recursive
 descendants retain the existing priority cutoff and history fallback. History
-still starts at most one due relay per maintenance pass, remains single-flight
-per relay, and fetches one filter group through the background request pacer
-before yielding. More distinct maintainer relays can therefore increase both
-persistent connection/subscription load and active history workers; the
+still admits at most one new relay per maintenance pass and remains
+single-flight per relay, but an admitted relay keeps its worker while its
+successful filter groups drain. More distinct maintainer relays can therefore
+increase both persistent connection/subscription load and active history
+workers up to the process-wide safety ceiling; the
 fixed-cardinality relay, cursor, worker, connection, and subscription metrics
 must be watched during a production soak.
 

docs/explanation/peer-controlled-state.md

@@ -39,7 +39,7 @@ larger retained set; it is not an excuse to copy arbitrary wire input.
 | Purgatory event maps and sync queue | write policy; promotion/cleanup owns removal | Time/external: entries are keyed by admitted event/repository identity, normally expire at 30 minutes, and soft-expired announcements at 24 hours. Queue entries deduplicate by identifier and disappear on completion or event expiry. | purgatory counts, queue and Git-process metrics |
 | Per-domain Git throttle queues | incomplete purgatory fetch; throttle manager owns drain | External: one entry per purgatory identifier/domain, merged on repeat. Completion, URL exhaustion, or purgatory expiry removes useful work. Request history is time-windowed. | domain/fetch logs and Git-process metrics |
 | Dependency retry attempts and temporary relays | rejected/purgatory dependency discovery; maintenance owns expiry | Time/external: event IDs are pruned against current purgatory input; temporary relays have explicit deadlines. Repeats overwrite timestamps. | retained dependency gauges |
-| NIP-65 discovery and mailbox probes | accepted root and descendant authors; ordinary root-inbox coverage plus discovery/probe scheduler own completion | External/session work: root-author read inboxes remain ordinary derived sync targets, while participant maps derive from accepted root threads. Identity discovery is globally single-flight. Mailbox history has at most one ordinary `fetch_events` worker per accepted relay; each page consumes that relay's pacing and ledger capacity and has a 30-second timeout. Starts are paced one per maintenance pass, progress on different relays is independent, and exclusive connections retire when idle. Numeric filter cursors and due times are in memory and bounded by desired mailbox relays. | discovery/probe logs and relay gauges |
+| NIP-65 discovery and mailbox probes | accepted root and descendant authors; ordinary root-inbox coverage plus discovery/probe scheduler own completion | External/session work: root-author read inboxes remain ordinary derived sync targets, while participant maps derive from accepted root threads. Identity discovery is globally single-flight. Mailbox history has at most one ordinary `fetch_events` worker per accepted relay and 32 process-wide; each page consumes that relay's pacing and ledger capacity and has a 30-second timeout. New relay workers are admitted one per maintenance pass, then each successful worker drains its own filter cursor until the cycle completes. Exclusive connections retire when idle. Numeric filter cursors and due times are in memory and bounded by desired mailbox relays. | discovery/probe logs and relay gauges |
 | Deferred consolidation | capacity refusal; final batch/reset/disconnect owns removal | External and deduplicated by relay. Final batch completion processes the set directly; no self-addressed notification queue remains. | retained deferred-consolidation gauge |
 | Descendant rotations and auxiliary live coverage | accepted root coverage; EOSE/CLOSED/disconnect/daily reset own transition | Session/external: one state object per derived relay and at most one rotating request in flight per relay. Coverage IDs consume ledger slots. | retained rotation gauge and terminal logs |
 | Health and naughty-list entries | connection failures; health checker owns recovery/expiry | External/time: one entry per canonical target, ordinary failures back off, persistent entries expire after 12 hours. Metrics expose only three fixed categories. | health gauges and aggregate naughty metrics |

src/sync/mod.rs

@@ -160,6 +160,14 @@ fn mailbox_probe_completion(
     (next_filter, completed_cycle, next_probe_in)
 }
 
+fn should_continue_mailbox_probe(succeeded: bool, completed_cycle: bool) -> bool {
+    succeeded && !completed_cycle
+}
+
+fn mailbox_worker_capacity_available(active_workers: usize) -> bool {
+    active_workers < MAX_CONCURRENT_MAILBOX_WORKERS
+}
+
 async fn fetch_mailbox_filter(
     connection: RelayConnection,
     filter: Filter,
@@ -2025,6 +2033,14 @@ const QUICK_RECONNECT_WINDOW_SECS: u64 = 15 * 60;
 /// exhaust network resources while keeping the sync actor responsive.
 const MAX_CONCURRENT_CONNECT_ATTEMPTS: usize = 8;
 
+/// Safety ceiling for independently progressing mailbox relays.
+///
+/// A relay owns at most one worker and drains its filter cursor without
+/// returning to the global relay rotation between slices. Keep a separate,
+/// generous process-wide ceiling so a peer-influenced NIP-65 inventory cannot
+/// create an unbounded number of simultaneous response accumulators.
+const MAX_CONCURRENT_MAILBOX_WORKERS: usize = 32;
+
 /// Maximum number of collection members rendered in an INFO-level log field.
 /// Counts remain authoritative; samples keep remotely influenced logs bounded.
 const LOG_COLLECTION_SAMPLE_SIZE: usize = 5;
@@ -2448,9 +2464,11 @@ async fn run_health_and_metrics_checker(
                 // to historic work when no slot is immediately available.
                 manager.schedule_nip65_discovery().await;
 
-                // 6. Start at most one participant mailbox history filter.
-                // Mailbox coverage is history-only: it does not turn every
-                // participant relay into a permanent live source.
+                // 6. Admit at most one new participant mailbox relay worker.
+                // Once admitted, that relay drains its own history-filter
+                // cursor independently, one request at a time. Mailbox
+                // coverage is history-only: it does not turn every participant
+                // relay into a permanent live source.
                 manager.schedule_mailbox_probe().await;
 
                 // 7. Check for naughty list expiration
@@ -6033,6 +6051,10 @@ impl SyncManager {
             return;
         }
 
+        if !mailbox_worker_capacity_available(self.nip65_discovery.mailbox_probes_in_flight.len()) {
+            return;
+        }
+
         let now = Instant::now();
         let due_relays: Vec<String> = due_mailbox_relays(
             &self.nip65_discovery.mailbox_roots,
@@ -6059,13 +6081,17 @@ impl SyncManager {
         let Some(relay) = select_due_mailbox_relay(&due_relays, &active_relays) else {
             return;
         };
+        self.start_mailbox_probe(relay, now).await;
+    }
+
+    async fn start_mailbox_probe(&mut self, relay: String, now: Instant) -> bool {
         if is_own_sync_target(&relay, &self.service_domain)
             || self.rejected_relay_targets.contains(&relay)
         {
             self.nip65_discovery
                 .mailbox_probe_next_at
                 .insert(relay, now + mailbox_probe_refresh_interval());
-            return;
+            return false;
         }
 
         if !self.connections.contains_key(&relay) {
@@ -6073,12 +6099,12 @@ impl SyncManager {
                 self.schedule_connect_relay(&relay).await;
             }
             self.defer_mailbox_relay(&relay);
-            return;
+            return false;
         }
 
         let Some(connection) = self.connections.get(&relay).cloned() else {
             self.defer_mailbox_relay(&relay);
-            return;
+            return false;
         };
         let socket_connected = connection.is_connected().await;
         let connection_status = self
@@ -6094,7 +6120,7 @@ impl SyncManager {
                 self.retire_idle_nip65_discovery_source(&relay).await;
             }
             self.defer_mailbox_relay(&relay);
-            return;
+            return false;
         }
 
         let root_scope = self
@@ -6127,7 +6153,7 @@ impl SyncManager {
                 .mailbox_probe_next_at
                 .insert(relay.clone(), now + mailbox_probe_refresh_interval());
             self.retire_idle_nip65_discovery_source(&relay).await;
-            return;
+            return false;
         };
         let filter_index = self
             .nip65_discovery
@@ -6139,7 +6165,7 @@ impl SyncManager {
         let filter = filters.swap_remove(filter_index);
         let Some(result_tx) = self.mailbox_probe_result_tx.clone() else {
             self.defer_mailbox_relay(&relay);
-            return;
+            return false;
         };
         self.nip65_discovery
             .mailbox_probes_in_flight
@@ -6162,6 +6188,7 @@ impl SyncManager {
             filter_count,
             "Started bounded participant mailbox fetch"
         );
+        true
     }
 
     fn defer_mailbox_relay(&mut self, relay: &str) {
@@ -6225,6 +6252,14 @@ impl SyncManager {
             next_probe_in_secs = next_probe_in.as_secs_f64(),
             "Participant mailbox fetch reached terminal state"
         );
+
+        if should_continue_mailbox_probe(succeeded, completed_cycle)
+            && self
+                .start_mailbox_probe(result.source_relay.clone(), Instant::now())
+                .await
+        {
+            return;
+        }
         self.retire_idle_nip65_discovery_source(&result.source_relay)
             .await;
     }
@@ -9939,11 +9974,32 @@ mod tests {
         assert_eq!(next_filter, 1);
         assert!(!completed_cycle);
         assert_eq!(retry_after, mailbox_probe_retry_interval());
+        assert!(!should_continue_mailbox_probe(false, completed_cycle));
 
         let (next_filter, completed_cycle, refresh_after) = mailbox_probe_completion(1, 2, true);
         assert_eq!(next_filter, 0);
         assert!(completed_cycle);
         assert_eq!(refresh_after, mailbox_probe_refresh_interval());
+        assert!(!should_continue_mailbox_probe(true, completed_cycle));
+    }
+
+    #[test]
+    fn mailbox_worker_continues_successful_partial_cycle() {
+        let (next_filter, completed_cycle, retry_after) = mailbox_probe_completion(0, 2, true);
+        assert_eq!(next_filter, 1);
+        assert!(!completed_cycle);
+        assert_eq!(retry_after, Duration::ZERO);
+        assert!(should_continue_mailbox_probe(true, completed_cycle));
+    }
+
+    #[test]
+    fn mailbox_worker_capacity_is_a_process_safety_ceiling() {
+        assert!(mailbox_worker_capacity_available(
+            MAX_CONCURRENT_MAILBOX_WORKERS - 1
+        ));
+        assert!(!mailbox_worker_capacity_available(
+            MAX_CONCURRENT_MAILBOX_WORKERS
+        ));
     }
 
     #[test]