fix(sync): retry events lost by incomplete historic-sync batches

Production logs after deploying a8964bb to gitnostr.com showed ~41
incomplete negentropy retries and 20 batches completing with partial
results within six minutes, some batches missing hundreds of events.
Negentropy reconciliation identifies event IDs missing locally, but a
relay's exact-ID response can return only a subset (or nothing on the
retry). Batches without repository/root-event metadata - the generic
Layer 1 announcements batch - cannot build a semantic REQ+EOSE
fallback, so handle_eose finalized them "with partial results" and
dropped the missing IDs entirely. Nothing retried them until the next
daily sync up to 25 hours later, leaving repository announcements and
their dependencies absent indefinitely.

Missing IDs from a batch that finalizes incomplete are now registered
in a per-relay recovery index (sync::missing_events), and the existing
sync maintenance timer refetches them over the relay's live connection
with bounded exponential backoff (30s doubling to 15min, one in-flight
attempt per relay, 300 IDs per fetch). Network I/O runs outside the
sync actor lock. Startup remains non-blocking: the batch still
finalizes as failed, the relay transitions to
ConnectedHistoricSyncFailures, and traffic is served while recovery
runs in the background.

Semantics:
- progress clears only the IDs actually recovered and resets backoff;
- duplicate incomplete responses merge into the pending set without
  duplicating work;
- IDs satisfied by live sync or user submission are cleared on the
  next tick without consuming attempt budget;
- attempts against a disconnected relay are deferred, not counted, so
  an unavailable relay neither expires its work nor loops tightly;
- 12 consecutive zero-progress attempts expire the pending IDs with an
  explicit warning; the relay stays observably degraded until the
  daily sync re-discovers the gap;
- full recovery promotes the relay back to Connected unless an
  unrelated batch failure was observed for it;
- nothing persists across restarts: historic sync re-runs from scratch
  and re-detects any still-missing events, so incomplete work is never
  falsely reported as complete.

Also fixes the retry-subscription-failure path, which confirmed an
incomplete batch without marking it failed (falsely reporting
Connected), and bounds the previously unbounded missing_ids log arrays
to a five-ID sample.

Regression coverage: a new censoring WebSocket proxy fixture sits
between a syncing relay and a real ngit-grasp bootstrap relay,
forwarding NIP-77 frames unchanged while withholding chosen EVENT
frames. The integration test reproduces the full production sequence
(subset response, zero-progress retry, no semantic fallback,
ConnectedHistoricSyncFailures) and proves the withheld event is
recovered and the relay promoted to Connected once the event becomes
available - without a restart and while live sync continues unstarved.
Unit tests cover registration dedupe, partial clears, backoff growth
and cap, explicit expiry, deferral, and health-restoration poisoning.
Full cargo test suite passes.

tokio-tungstenite was added as a dev-dependency for the proxy fixture;
it was already present transitively in Cargo.lock, so no Nix hash
updates are required (crates.io dependency under cargoLock).

Deliberately out of scope: durable persistence of pending recovery
work, retrying missing IDs against other relays, outbound-target
policy changes, and broader logging cleanup.

Confirms the closed issue
nostr:nevent1qqs94up6nnkzjlz4fcy5tesh8yxvr63xqjhg79etmc573fuunjt0qeqpz3mhxue69uhhyetvv9ujumn8d96zuer9wc5tdht6

CHANGELOG.md

@@ -32,6 +32,26 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed historic sync silently abandoning events a relay identified but failed
+  to deliver. Negentropy reconciliation finds event IDs missing locally, but a
+  relay's exact-ID response can return only a subset; for batches without
+  repository/root-event metadata (the Layer 1 announcements batch) no semantic
+  REQ+EOSE fallback exists, and production logs showed such batches completing
+  "with partial results" — dropping the missing IDs until the next daily sync
+  up to 25 hours later. The still-missing IDs are now kept in a per-relay
+  recovery index and refetched over the existing connection with bounded
+  exponential backoff (30s doubling to 15min, one in-flight attempt per relay,
+  300 IDs per fetch). Progress clears only the recovered IDs and resets the
+  backoff; duplicate incomplete responses merge without duplicating work; IDs
+  satisfied by live sync are cleared without consuming attempts; 12
+  consecutive zero-progress attempts expire the pending IDs explicitly, and
+  the relay stays observably degraded (`ConnectedHistoricSyncFailures`) until
+  the daily sync. Startup remains non-blocking throughout, and a relay whose
+  missing events are all recovered — with no unrelated batch failures — is now
+  promoted back to `Connected` instead of reporting failures forever. Also
+  fixed the retry-subscription-failure path confirming an incomplete batch as
+  successful, and bounded the previously unbounded `missing_ids` log arrays to
+  a five-ID sample.
 - Fixed malformed client messages tearing down the whole WebSocket connection.
   A single unparseable message - in production, requests carrying invalid event
   IDs that fail with `Invalid input length 64` - closed the session, forcing

Cargo.lock

@@ -1299,6 +1299,7 @@ dependencies = [
  "tar",
  "tempfile",
  "tokio",
+ "tokio-tungstenite",
  "tracing",
  "tracing-subscriber",
 ]

Cargo.toml

@@ -83,6 +83,7 @@ tempfile = "3"
 grasp-audit = { path = "grasp-audit", version = "0.2.0" }
 tempfile = "3"
 reqwest = { version = "0.13", default-features = false, features = ["native-tls"] }
+tokio-tungstenite = "0.28.0"
 
 [lib]
 name = "ngit_grasp"

docs/explanation/grasp-02-proactive-sync.md

@@ -292,7 +292,7 @@ Each layer creates one or more `PendingBatch` entries tracked in `PendingSyncInd
 
 **Why the double-check?** There's an async gap between receiving EOSE and the self-subscriber processing events to create Layer 2/3 filters. The 6-second wait (5s batch window + 1s buffer) ensures we don't prematurely mark sync complete while Layer 2/3 batches are being created.
 
-**Batch Failure Tracking**: When negentropy retry protection triggers (relay returns zero requested events on retry), the batch is marked as `failed = true`. This causes the relay to transition to `ConnectedHistoricSyncFailures` instead of `Connected`, signaling that live sync is active but historic sync is incomplete.
+**Batch Failure Tracking**: When negentropy retry protection triggers (relay returns zero requested events on retry) and no fallback subscriptions can be created, the batch is marked as `failed = true`. This causes the relay to transition to `ConnectedHistoricSyncFailures` instead of `Connected`, signaling that live sync is active but historic sync is incomplete. The event IDs the relay failed to deliver are not dropped with the batch: they are registered for bounded background recovery (see "Missing-Event Recovery for Incomplete Batches" below), and a relay whose pending IDs are all eventually recovered — with no unrelated batch failures — is promoted back to `Connected`.
 
 **Metrics tracking**: The `ngit_sync_relay_connected` metric shows:
 - `0` = Disconnected
@@ -391,6 +391,8 @@ The sync system uses three background tasks that run continuously:
 
 **Why a separate timer?** Purgatory announcements are never saved to the database, so the SelfSubscriber never sees them. The timer bridges this gap, ensuring state events are synced for repos that may still receive git data.
 
+The same tick also drives missing-event recovery for incomplete historic sync batches (see "Missing-Event Recovery for Incomplete Batches"). Recovery attempts are backed off per relay, so the tick stays cheap when nothing is due.
+
 ---
 
 ## Core Architecture: Live vs Historic Sync
@@ -800,6 +802,43 @@ separate from the SDK's initial-response and idle timers: a relay that keeps an
 exchange active without completing it cannot hold the sync actor indefinitely.
 Reaching the deadline uses the same unsupported-relay fallback path.
 
+### Missing-Event Recovery for Incomplete Batches
+
+Negentropy reconciliation can identify event IDs a relay holds, only for the
+relay's exact-ID REQ response to return a subset of them (result limits,
+truncation) or nothing at all. The batch flow retries once with an ID-based
+subscription and then falls back to semantic REQ+EOSE filters built from the
+batch's repos/root events. The generic Layer 1 announcements batch carries no
+such metadata, so no semantic fallback exists for it: in production this
+finalized the batch with partial results and silently dropped the missing IDs
+until the next daily sync (23-25h later).
+
+Missing IDs from a batch that finalizes incomplete are instead registered in a
+per-relay recovery index (`sync::missing_events`) and retried by the sync
+maintenance timer:
+
+- **Non-blocking**: the batch still finalizes (as failed) and the relay keeps
+  serving traffic; recovery runs in the background over the existing relay
+  connection, outside the sync actor lock.
+- **Bounded and backed off**: attempts are per relay with exponential backoff
+  (30s base doubling up to 15min; sub-second in `NGIT_TEST`), one in-flight
+  attempt per relay, at most 300 IDs per fetch. One persistently incomplete
+  relay cannot starve other relays or later batches.
+- **Progress-aware**: a successful attempt clears only the IDs actually
+  recovered and resets the backoff; duplicate incomplete responses merge into
+  the existing pending set. IDs that arrive by other means (live sync, user
+  submission) are cleared on the next tick without consuming attempt budget.
+- **Explicit expiry**: after 12 consecutive zero-progress attempts the relay's
+  pending IDs are dropped with a warning, and the relay stays in
+  `ConnectedHistoricSyncFailures` until the daily sync re-discovers the gap.
+  Attempts against a disconnected relay are deferred, not counted, so an
+  unavailable relay neither expires its pending work nor loops tightly.
+- **Honest status**: full recovery promotes the relay from
+  `ConnectedHistoricSyncFailures` back to `Connected` — but only when no
+  unrelated batch failure was observed for that relay. Nothing is persisted
+  across restarts; a restart re-runs historic sync, which re-detects any
+  still-missing events.
+
 ### Integration with Rejected Events Index
 
 The rejected events index prevents wasteful re-fetching during negentropy sync by excluding rejected event IDs from the reconciliation process:

src/sync/missing_events.rs

@@ -0,0 +1,593 @@
+//! Missing-Event Recovery for Incomplete Historic Sync Batches
+//!
+//! NIP-77 negentropy reconciliation identifies event IDs a relay holds that
+//! are missing locally. Those IDs are fetched with exact-ID REQs, but relays
+//! can return only a subset (result limits, truncation) or nothing at all.
+//! When the batch also lacks repository/root-event metadata, no semantic
+//! REQ+EOSE fallback can be constructed and the batch is finalized with
+//! partial results.
+//!
+//! This module keeps the still-missing IDs represented after such a batch is
+//! finalized, so the sync maintenance timer can retry them with bounded,
+//! exponentially backed-off exact-ID fetches. It is a per-relay bookkeeping
+//! structure only: all network work is driven by `SyncManager`.
+//!
+//! ## Policy
+//!
+//! - Attempts are scheduled per relay with exponential backoff
+//!   (30s base doubling up to 15min in production; sub-second in `NGIT_TEST`).
+//! - Any progress (at least one ID recovered) resets the backoff; the pending
+//!   set shrinks monotonically, so this cannot loop forever.
+//! - After [`MAX_ATTEMPTS_WITHOUT_PROGRESS`] consecutive attempts with zero
+//!   progress the relay's pending IDs are expired: they are dropped with an
+//!   explicit warning and the relay remains in
+//!   `ConnectedHistoricSyncFailures` until the next daily sync re-discovers
+//!   the gap. Nothing is persisted across restarts - a restart re-runs
+//!   historic sync from scratch, which re-detects any still-missing events.
+//! - Recovery state never blocks startup: batches still finalize (as failed)
+//!   and the relay serves traffic while attempts continue in the background.
+//! - A relay whose pending set empties through actual recovery can be
+//!   promoted back to `Connected`, unless an unrelated batch failure was
+//!   observed for that relay (the entry is then marked as unable to restore
+//!   health, so the degraded status stays visible).
+
+use std::collections::{HashMap, HashSet};
+use std::time::{Duration, Instant};
+
+use nostr_sdk::prelude::EventId;
+
+/// Maximum IDs fetched per recovery attempt, matching the exact-ID chunk size
+/// used by historic sync. Larger pending sets recover across several attempts.
+pub const MAX_RECOVERY_IDS_PER_ATTEMPT: usize = 300;
+
+/// Consecutive zero-progress attempts before a relay's pending IDs expire.
+pub const MAX_ATTEMPTS_WITHOUT_PROGRESS: u32 = 12;
+
+/// Cap on remembered source batch IDs (for logging only).
+const MAX_TRACKED_SOURCE_BATCHES: usize = 16;
+
+fn in_test_mode() -> bool {
+    std::env::var("NGIT_TEST").as_deref() == Ok("1")
+}
+
+fn base_backoff() -> Duration {
+    if in_test_mode() {
+        Duration::from_millis(500)
+    } else {
+        Duration::from_secs(30)
+    }
+}
+
+fn max_backoff() -> Duration {
+    if in_test_mode() {
+        Duration::from_secs(5)
+    } else {
+        Duration::from_secs(15 * 60)
+    }
+}
+
+fn backoff_for(attempts_without_progress: u32) -> Duration {
+    let base = base_backoff();
+    let doubled = base.saturating_mul(1u32 << attempts_without_progress.min(16));
+    doubled.min(max_backoff())
+}
+
+/// Pending recovery work for one relay.
+#[derive(Debug)]
+struct RelayRecoveryState {
+    /// Event IDs the relay reported but has not yet delivered.
+    missing: HashSet<EventId>,
+    /// Consecutive attempts that recovered nothing. Reset on progress.
+    attempts_without_progress: u32,
+    /// Total attempts issued, for logging.
+    total_attempts: u32,
+    /// Earliest time the next attempt may run.
+    next_attempt_at: Instant,
+    /// One recovery fetch in flight per relay at a time.
+    in_flight: bool,
+    /// False when an unrelated batch failure means full recovery of the
+    /// pending IDs would not make the relay's historic sync complete.
+    can_restore_health: bool,
+    /// Batch IDs that contributed missing IDs, for logging (bounded).
+    source_batches: Vec<u64>,
+    /// Total IDs recovered so far, for logging.
+    total_recovered: usize,
+}
+
+/// Result of registering missing IDs for a relay.
+#[derive(Debug, PartialEq, Eq)]
+pub struct RegisterOutcome {
+    /// IDs that were not already pending.
+    pub newly_added: usize,
+    /// Total pending IDs for the relay after registration.
+    pub pending_total: usize,
+}
+
+/// A recovery fetch reserved by [`MissingEventRecoveryIndex::begin_attempt`].
+#[derive(Debug)]
+pub struct RecoveryAttempt {
+    /// IDs to fetch in this attempt (bounded).
+    pub ids: Vec<EventId>,
+    /// 1-based attempt number for the relay, for logging.
+    pub attempt_number: u32,
+}
+
+/// Outcome of completing a recovery attempt or a local-satisfaction check.
+#[derive(Debug, PartialEq, Eq)]
+pub enum AttemptOutcome {
+    /// Every pending ID for the relay has been recovered.
+    FullyRecovered {
+        recovered: usize,
+        total_recovered: usize,
+        can_restore_health: bool,
+    },
+    /// Some IDs were recovered; the rest are rescheduled.
+    PartiallyRecovered {
+        recovered: usize,
+        remaining: usize,
+        next_attempt_in: Duration,
+    },
+    /// Nothing was recovered; the next attempt is scheduled with backoff.
+    RetryScheduled {
+        attempt: u32,
+        remaining: usize,
+        next_attempt_in: Duration,
+    },
+    /// The zero-progress attempt budget is exhausted; pending IDs dropped.
+    Expired { remaining: usize, attempts: u32 },
+}
+
+/// Per-relay index of events that historic sync failed to fetch.
+#[derive(Debug, Default)]
+pub struct MissingEventRecoveryIndex {
+    relays: HashMap<String, RelayRecoveryState>,
+}
+
+impl MissingEventRecoveryIndex {
+    /// Record IDs a finalized batch failed to fetch from `relay_url`.
+    ///
+    /// Duplicate registrations merge into the existing pending set, so
+    /// repeated incomplete responses cannot create duplicate work.
+    /// `relay_already_degraded` poisons health restoration when the relay had
+    /// unrelated historic-sync failures before this entry existed.
+    pub fn register(
+        &mut self,
+        relay_url: &str,
+        batch_id: u64,
+        ids: impl IntoIterator<Item = EventId>,
+        relay_already_degraded: bool,
+        now: Instant,
+    ) -> RegisterOutcome {
+        let entry =
+            self.relays
+                .entry(relay_url.to_string())
+                .or_insert_with(|| RelayRecoveryState {
+                    missing: HashSet::new(),
+                    attempts_without_progress: 0,
+                    total_attempts: 0,
+                    next_attempt_at: now + base_backoff(),
+                    in_flight: false,
+                    can_restore_health: !relay_already_degraded,
+                    source_batches: Vec::new(),
+                    total_recovered: 0,
+                });
+
+        if relay_already_degraded {
+            entry.can_restore_health = false;
+        }
+        if entry.source_batches.len() < MAX_TRACKED_SOURCE_BATCHES
+            && !entry.source_batches.contains(&batch_id)
+        {
+            entry.source_batches.push(batch_id);
+        }
+
+        let before = entry.missing.len();
+        entry.missing.extend(ids);
+        RegisterOutcome {
+            newly_added: entry.missing.len() - before,
+            pending_total: entry.missing.len(),
+        }
+    }
+
+    /// Note a failed batch confirmation for `relay_url`.
+    ///
+    /// If the batch did not contribute this relay's pending IDs, the failure
+    /// is unrelated, so recovering every pending ID must not promote the
+    /// relay back to a healthy status.
+    pub fn note_failed_batch(&mut self, relay_url: &str, batch_id: u64) {
+        if let Some(entry) = self.relays.get_mut(relay_url) {
+            if !entry.source_batches.contains(&batch_id) {
+                entry.can_restore_health = false;
+            }
+        }
+    }
+
+    /// Relays with pending IDs and no attempt currently in flight.
+    pub fn idle_relays(&self) -> Vec<String> {
+        self.relays
+            .iter()
+            .filter(|(_, entry)| !entry.in_flight)
+            .map(|(relay, _)| relay.clone())
+            .collect()
+    }
+
+    /// Pending IDs for a relay, if any attempt is not in flight.
+    pub fn pending_ids(&self, relay_url: &str) -> Option<Vec<EventId>> {
+        let entry = self.relays.get(relay_url)?;
+        if entry.in_flight {
+            return None;
+        }
+        Some(entry.missing.iter().copied().collect())
+    }
+
+    /// Remove IDs that were satisfied outside recovery (live sync, user
+    /// submission). Not counted as an attempt. Returns `FullyRecovered` when
+    /// the pending set empties.
+    pub fn clear_satisfied(
+        &mut self,
+        relay_url: &str,
+        satisfied: &[EventId],
+    ) -> Option<AttemptOutcome> {
+        let entry = self.relays.get_mut(relay_url)?;
+        let before = entry.missing.len();
+        for id in satisfied {
+            entry.missing.remove(id);
+        }
+        let recovered = before - entry.missing.len();
+        entry.total_recovered += recovered;
+        if recovered > 0 {
+            entry.attempts_without_progress = 0;
+        }
+        if entry.missing.is_empty() {
+            let entry = self
+                .relays
+                .remove(relay_url)
+                .expect("entry present just above");
+            return Some(AttemptOutcome::FullyRecovered {
+                recovered,
+                total_recovered: entry.total_recovered,
+                can_restore_health: entry.can_restore_health,
+            });
+        }
+        None
+    }
+
+    /// Reserve a recovery fetch for a relay whose backoff deadline passed.
+    pub fn begin_attempt(&mut self, relay_url: &str, now: Instant) -> Option<RecoveryAttempt> {
+        let entry = self.relays.get_mut(relay_url)?;
+        if entry.in_flight || entry.missing.is_empty() || now < entry.next_attempt_at {
+            return None;
+        }
+        entry.in_flight = true;
+        entry.total_attempts += 1;
+        Some(RecoveryAttempt {
+            ids: entry
+                .missing
+                .iter()
+                .take(MAX_RECOVERY_IDS_PER_ATTEMPT)
+                .copied()
+                .collect(),
+            attempt_number: entry.total_attempts,
+        })
+    }
+
+    /// Push the next attempt out without consuming attempt budget. Used when
+    /// the relay has no live connection, so a disconnected relay can neither
+    /// expire its pending IDs nor spin in a tight loop.
+    pub fn defer_attempt(&mut self, relay_url: &str, now: Instant) {
+        if let Some(entry) = self.relays.get_mut(relay_url) {
+            if !entry.in_flight {
+                entry.next_attempt_at = now + base_backoff();
+            }
+        }
+    }
+
+    /// Complete a reserved attempt: clear only the recovered IDs, then either
+    /// finish, reschedule with backoff, or expire by policy.
+    ///
+    /// Returns `None` when the relay's entry disappeared while the attempt was
+    /// in flight (daily sync reset or intentional relay removal).
+    pub fn complete_attempt(
+        &mut self,
+        relay_url: &str,
+        recovered: &HashSet<EventId>,
+        now: Instant,
+    ) -> Option<AttemptOutcome> {
+        let entry = self.relays.get_mut(relay_url)?;
+        entry.in_flight = false;
+
+        let before = entry.missing.len();
+        entry.missing.retain(|id| !recovered.contains(id));
+        let recovered_count = before - entry.missing.len();
+        entry.total_recovered += recovered_count;
+
+        if entry.missing.is_empty() {
+            let entry = self
+                .relays
+                .remove(relay_url)
+                .expect("entry present just above");
+            return Some(AttemptOutcome::FullyRecovered {
+                recovered: recovered_count,
+                total_recovered: entry.total_recovered,
+                can_restore_health: entry.can_restore_health,
+            });
+        }
+
+        if recovered_count > 0 {
+            entry.attempts_without_progress = 0;
+            let delay = backoff_for(0);
+            entry.next_attempt_at = now + delay;
+            return Some(AttemptOutcome::PartiallyRecovered {
+                recovered: recovered_count,
+                remaining: entry.missing.len(),
+                next_attempt_in: delay,
+            });
+        }
+
+        entry.attempts_without_progress += 1;
+        if entry.attempts_without_progress >= MAX_ATTEMPTS_WITHOUT_PROGRESS {
+            let entry = self
+                .relays
+                .remove(relay_url)
+                .expect("entry present just above");
+            return Some(AttemptOutcome::Expired {
+                remaining: entry.missing.len(),
+                attempts: entry.total_attempts,
+            });
+        }
+        let delay = backoff_for(entry.attempts_without_progress);
+        entry.next_attempt_at = now + delay;
+        Some(AttemptOutcome::RetryScheduled {
+            attempt: entry.total_attempts,
+            remaining: entry.missing.len(),
+            next_attempt_in: delay,
+        })
+    }
+
+    /// Drop all pending IDs for a relay (daily sync reset or relay removal).
+    /// Returns the number of IDs dropped.
+    pub fn clear_relay(&mut self, relay_url: &str) -> usize {
+        self.relays
+            .remove(relay_url)
+            .map(|entry| entry.missing.len())
+            .unwrap_or(0)
+    }
+
+    /// Batch IDs that contributed a relay's pending IDs, for logging.
+    pub fn source_batches(&self, relay_url: &str) -> Vec<u64> {
+        self.relays
+            .get(relay_url)
+            .map(|entry| entry.source_batches.clone())
+            .unwrap_or_default()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn id(byte: u8) -> EventId {
+        EventId::from_slice(&[byte; 32]).expect("valid event id")
+    }
+
+    fn registered(index: &mut MissingEventRecoveryIndex, ids: &[EventId]) -> Instant {
+        let now = Instant::now();
+        index.register("ws://relay", 1, ids.iter().copied(), false, now);
+        now
+    }
+
+    fn due(now: Instant) -> Instant {
+        now + Duration::from_secs(24 * 3600)
+    }
+
+    #[test]
+    fn duplicate_registrations_do_not_duplicate_pending_work() {
+        let mut index = MissingEventRecoveryIndex::default();
+        let now = registered(&mut index, &[id(1), id(2)]);
+        let outcome = index.register("ws://relay", 2, [id(2), id(3)], false, now);
+        assert_eq!(
+            outcome,
+            RegisterOutcome {
+                newly_added: 1,
+                pending_total: 3,
+            }
+        );
+        assert_eq!(index.pending_ids("ws://relay").unwrap().len(), 3);
+    }
+
+    #[test]
+    fn attempts_wait_for_backoff_deadline_and_single_flight() {
+        let mut index = MissingEventRecoveryIndex::default();
+        let now = registered(&mut index, &[id(1)]);
+
+        assert!(
+            index.begin_attempt("ws://relay", now).is_none(),
+            "attempt before the backoff deadline must not run"
+        );
+        let attempt = index
+            .begin_attempt("ws://relay", due(now))
+            .expect("due attempt");
+        assert_eq!(attempt.ids, vec![id(1)]);
+        assert_eq!(attempt.attempt_number, 1);
+        assert!(
+            index.begin_attempt("ws://relay", due(now)).is_none(),
+            "only one attempt may be in flight per relay"
+        );
+        assert!(
+            index.pending_ids("ws://relay").is_none(),
+            "in-flight relays are skipped by local-satisfaction checks"
+        );
+    }
+
+    #[test]
+    fn successful_attempt_clears_only_recovered_ids_and_resets_backoff() {
+        let mut index = MissingEventRecoveryIndex::default();
+        let now = registered(&mut index, &[id(1), id(2), id(3)]);
+
+        // Two failed attempts grow the backoff.
+        for _ in 0..2 {
+            index.begin_attempt("ws://relay", due(now)).unwrap();
+            index
+                .complete_attempt("ws://relay", &HashSet::new(), now)
+                .unwrap();
+        }
+
+        index.begin_attempt("ws://relay", due(now)).unwrap();
+        let outcome = index
+            .complete_attempt("ws://relay", &HashSet::from([id(2)]), now)
+            .unwrap();
+        assert_eq!(
+            outcome,
+            AttemptOutcome::PartiallyRecovered {
+                recovered: 1,
+                remaining: 2,
+                next_attempt_in: backoff_for(0),
+            },
+            "progress must clear only the recovered ID and reset the backoff"
+        );
+        let mut remaining = index.pending_ids("ws://relay").unwrap();
+        remaining.sort();
+        assert_eq!(remaining, vec![id(1), id(3)]);
+    }
+
+    #[test]
+    fn zero_progress_attempts_back_off_exponentially_with_a_cap() {
+        assert!(backoff_for(1) > backoff_for(0));
+        assert!(backoff_for(2) > backoff_for(1));
+        assert_eq!(backoff_for(30), max_backoff());
+    }
+
+    #[test]
+    fn zero_progress_budget_expires_pending_ids_explicitly() {
+        let mut index = MissingEventRecoveryIndex::default();
+        let now = registered(&mut index, &[id(1), id(2)]);
+
+        for attempt in 1..MAX_ATTEMPTS_WITHOUT_PROGRESS {
+            index.begin_attempt("ws://relay", due(now)).unwrap();
+            let outcome = index
+                .complete_attempt("ws://relay", &HashSet::new(), now)
+                .unwrap();
+            assert!(
+                matches!(outcome, AttemptOutcome::RetryScheduled { .. }),
+                "attempt {attempt} should reschedule, got {outcome:?}"
+            );
+        }
+
+        index.begin_attempt("ws://relay", due(now)).unwrap();
+        let outcome = index
+            .complete_attempt("ws://relay", &HashSet::new(), now)
+            .unwrap();
+        assert_eq!(
+            outcome,
+            AttemptOutcome::Expired {
+                remaining: 2,
+                attempts: MAX_ATTEMPTS_WITHOUT_PROGRESS,
+            }
+        );
+        assert!(
+            index.pending_ids("ws://relay").is_none(),
+            "expired relays carry no pending work"
+        );
+    }
+
+    #[test]
+    fn full_recovery_reports_whether_health_can_be_restored() {
+        let mut index = MissingEventRecoveryIndex::default();
+        let now = registered(&mut index, &[id(1)]);
+        index.begin_attempt("ws://relay", due(now)).unwrap();
+        let outcome = index
+            .complete_attempt("ws://relay", &HashSet::from([id(1)]), now)
+            .unwrap();
+        assert_eq!(
+            outcome,
+            AttemptOutcome::FullyRecovered {
+                recovered: 1,
+                total_recovered: 1,
+                can_restore_health: true,
+            }
+        );
+    }
+
+    #[test]
+    fn unrelated_failures_poison_health_restoration() {
+        let mut index = MissingEventRecoveryIndex::default();
+        let now = registered(&mut index, &[id(1)]);
+
+        // Batch 1 registered the IDs; its own failed confirmation is related.
+        index.note_failed_batch("ws://relay", 1);
+        // Batch 7 never registered recovery work: unrelated failure.
+        index.note_failed_batch("ws://relay", 7);
+
+        index.begin_attempt("ws://relay", due(now)).unwrap();
+        let outcome = index
+            .complete_attempt("ws://relay", &HashSet::from([id(1)]), now)
+            .unwrap();
+        assert_eq!(
+            outcome,
+            AttemptOutcome::FullyRecovered {
+                recovered: 1,
+                total_recovered: 1,
+                can_restore_health: false,
+            }
+        );
+    }
+
+    #[test]
+    fn registration_on_an_already_degraded_relay_poisons_health_restoration() {
+        let mut index = MissingEventRecoveryIndex::default();
+        let now = Instant::now();
+        index.register("ws://relay", 1, [id(1)], true, now);
+        index.begin_attempt("ws://relay", due(now)).unwrap();
+        let outcome = index
+            .complete_attempt("ws://relay", &HashSet::from([id(1)]), now)
+            .unwrap();
+        assert!(matches!(
+            outcome,
+            AttemptOutcome::FullyRecovered {
+                can_restore_health: false,
+                ..
+            }
+        ));
+    }
+
+    #[test]
+    fn locally_satisfied_ids_clear_promptly_without_consuming_attempts() {
+        let mut index = MissingEventRecoveryIndex::default();
+        registered(&mut index, &[id(1), id(2)]);
+
+        assert_eq!(index.clear_satisfied("ws://relay", &[id(1)]), None);
+        assert_eq!(index.pending_ids("ws://relay").unwrap(), vec![id(2)]);
+
+        let outcome = index.clear_satisfied("ws://relay", &[id(2)]).unwrap();
+        assert!(matches!(outcome, AttemptOutcome::FullyRecovered { .. }));
+    }
+
+    #[test]
+    fn completing_an_attempt_after_relay_reset_is_a_no_op() {
+        let mut index = MissingEventRecoveryIndex::default();
+        let now = registered(&mut index, &[id(1)]);
+        index.begin_attempt("ws://relay", due(now)).unwrap();
+        assert_eq!(index.clear_relay("ws://relay"), 1);
+        assert!(index
+            .complete_attempt("ws://relay", &HashSet::new(), now)
+            .is_none());
+    }
+
+    #[test]
+    fn deferred_attempts_do_not_consume_the_zero_progress_budget() {
+        let mut index = MissingEventRecoveryIndex::default();
+        let now = registered(&mut index, &[id(1)]);
+        index.defer_attempt("ws://relay", due(now));
+        assert!(
+            index.begin_attempt("ws://relay", due(now)).is_none(),
+            "deferral must push the next attempt out"
+        );
+        let attempt = index
+            .begin_attempt("ws://relay", due(due(now)))
+            .expect("attempt after deferral window");
+        assert_eq!(
+            attempt.attempt_number, 1,
+            "deferrals are not counted as attempts"
+        );
+    }
+}

src/sync/mod.rs

@@ -16,6 +16,7 @@ pub mod algorithms;
 pub mod filters;
 pub mod health;
 pub mod metrics;
+pub mod missing_events;
 pub mod naughty_list;
 pub mod rejected_index;
 pub mod relay_connection;
@@ -752,6 +753,11 @@ async fn run_daily_timer(
 /// - Sync-path announcements: registered here within one interval of arriving.
 /// - User-submitted purgatory announcements: the SelfSubscriber never sees them
 ///   (they're rejected from DB), so this timer is the only registration path.
+///
+/// The same tick also drives bounded recovery of events that relays reported
+/// during negentropy reconciliation but failed to deliver on exact-ID fetches
+/// (see [`missing_events`]). Recovery attempts are backed off per relay, so
+/// the tick itself stays cheap when nothing is due.
 async fn run_purgatory_announcement_sync(
     sync_manager: Arc<Mutex<SyncManager>>,
     mut shutdown_rx: broadcast::Receiver<()>,
@@ -766,6 +772,7 @@ async fn run_purgatory_announcement_sync(
             _ = tokio::time::sleep(interval) => {
                 let mut manager = sync_manager.lock().await;
                 manager.sync_purgatory_announcements_to_index().await;
+                manager.tick_missing_event_recovery().await;
             }
             _ = shutdown_rx.recv() => {
                 tracing::debug!("Purgatory announcement sync timer received shutdown signal");
@@ -958,6 +965,10 @@ pub struct SyncManager {
     rejected_relay_targets: HashSet<String>,
     /// Last exact-ID dependency recovery attempt, used to bound retries.
     dependency_refetch_attempts: Arc<std::sync::Mutex<HashMap<EventId, Instant>>>,
+    /// Events relays reported during negentropy reconciliation but failed to
+    /// deliver on exact-ID fetches. Retried with bounded backoff by the sync
+    /// maintenance timer instead of being dropped with their failed batch.
+    missing_event_recovery: Arc<std::sync::Mutex<missing_events::MissingEventRecoveryIndex>>,
     /// Last dependency pass for each purgatory announcement.
     purgatory_dependency_attempts: HashMap<EventId, Instant>,
     /// Temporary source relays retained while rejected dependencies are recovered.
@@ -1059,6 +1070,9 @@ impl SyncManager {
             connections: HashMap::new(),
             rejected_relay_targets: HashSet::new(),
             dependency_refetch_attempts: Arc::new(std::sync::Mutex::new(HashMap::new())),
+            missing_event_recovery: Arc::new(std::sync::Mutex::new(
+                missing_events::MissingEventRecoveryIndex::default(),
+            )),
             purgatory_dependency_attempts: HashMap::new(),
             dependency_relay_deadlines: HashMap::new(),
             health_tracker: Arc::new(RelayHealthTracker::new(config)),
@@ -1327,7 +1341,7 @@ impl SyncManager {
                             retry_count = retry_count,
                             requested_count = requested_count,
                             missing_count = missing.len(),
-                            missing_ids = ?missing.iter().map(|id| id.to_hex()).collect::<Vec<_>>(),
+                            missing_ids_sample = ?missing.iter().take(5).map(|id| id.to_hex()).collect::<Vec<_>>(),
                             "Negentropy retry made no progress - relay returned zero requested events. \
                              Marking relay as not supporting negentropy and falling back to REQ+EOSE."
                         );
@@ -1422,12 +1436,33 @@ impl SyncManager {
                             // Early return - batch not complete yet, waiting for REQ+EOSE EOSE
                             return;
                         } else {
-                            // Failed to create any fallback subscriptions, mark as failed
+                            // No fallback subscriptions could be created (for
+                            // announcement batches there is no semantic metadata
+                            // at all). Finalize the batch as failed but keep the
+                            // missing IDs represented so the maintenance timer
+                            // retries them with bounded backoff instead of
+                            // forgetting them until the next daily sync.
+                            let relay_already_degraded = {
+                                let index = self.relay_sync_index.read().await;
+                                index
+                                    .get(&relay_url_for_fallback)
+                                    .map(|state| state.historic_sync_had_failures)
+                                    .unwrap_or(false)
+                            };
+                            let register_outcome =
+                                self.missing_event_recovery.lock().unwrap().register(
+                                    &relay_url_for_fallback,
+                                    batch_id,
+                                    missing.iter().copied(),
+                                    relay_already_degraded,
+                                    Instant::now(),
+                                );
                             tracing::error!(
                                 relay = %relay_url_for_fallback,
                                 batch_id = batch_id,
                                 missing_count = missing_count,
-                                "Failed to create REQ+EOSE fallback subscriptions - completing batch with partial results"
+                                pending_recovery = register_outcome.pending_total,
+                                "Failed to create REQ+EOSE fallback subscriptions - completing batch with partial results; bounded missing-event recovery scheduled"
                             );
 
                             // Re-acquire lock to extract the batch
@@ -1457,7 +1492,7 @@ impl SyncManager {
                         requested_count = requested_count,
                         received_count = received_count,
                         missing_count = missing.len(),
-                        missing_ids = ?missing.iter().map(|id| id.to_hex()).collect::<Vec<_>>(),
+                        missing_ids_sample = ?missing.iter().take(5).map(|id| id.to_hex()).collect::<Vec<_>>(),
                         "Negentropy sync incomplete - relay returned fewer events than requested. \
                          This may indicate a relay limit on ID-based queries. \
                          Retrying missing events."
@@ -1523,20 +1558,39 @@ impl SyncManager {
                         // Early return - batch not complete yet, waiting for retry EOSE
                         return;
                     } else {
-                        // Failed to create retry subscriptions, log and continue to confirm
-                        // with partial results
+                        // Failed to create retry subscriptions. Finalize the
+                        // batch as failed (an incomplete batch must not be
+                        // reported as successfully complete) and keep the
+                        // missing IDs represented for bounded recovery.
+                        let relay_already_degraded = {
+                            let index = self.relay_sync_index.read().await;
+                            index
+                                .get(&relay_url_for_retry)
+                                .map(|state| state.historic_sync_had_failures)
+                                .unwrap_or(false)
+                        };
+                        let register_outcome =
+                            self.missing_event_recovery.lock().unwrap().register(
+                                &relay_url_for_retry,
+                                batch_id,
+                                missing.iter().copied(),
+                                relay_already_degraded,
+                                Instant::now(),
+                            );
                         tracing::error!(
                             relay = %relay_url_for_retry,
                             batch_id = batch_id,
                             missing_count = missing.len(),
-                            "Failed to retry missing events - confirming batch with partial results"
+                            pending_recovery = register_outcome.pending_total,
+                            "Failed to retry missing events - confirming batch with partial results; bounded missing-event recovery scheduled"
                         );
 
                         // Re-acquire lock to extract the batch
                         let mut pending = self.pending_sync_index.write().await;
                         if let Some(batches) = pending.get_mut(&relay_url_for_retry) {
                             if let Some(idx) = batches.iter().position(|b| b.batch_id == batch_id) {
-                                let completed_batch = batches.remove(idx);
+                                let mut completed_batch = batches.remove(idx);
+                                completed_batch.failed = true;
                                 if batches.is_empty() {
                                     pending.remove(&relay_url_for_retry);
                                 }
@@ -1642,6 +1696,288 @@ impl SyncManager {
         });
     }
 
+    /// Drive bounded recovery of events relays reported during negentropy
+    /// reconciliation but failed to deliver on exact-ID fetches.
+    ///
+    /// Runs on the sync maintenance timer. For each relay with pending IDs:
+    /// 1. IDs already present locally (live sync, user submission) are
+    ///    cleared promptly without consuming attempt budget.
+    /// 2. When the relay's backoff deadline has passed and it has a live
+    ///    connection, one bounded exact-ID fetch is spawned. Network I/O runs
+    ///    outside the sync actor lock, and per-relay single-flight keeps one
+    ///    persistently incomplete relay from starving other relays or later
+    ///    batches.
+    async fn tick_missing_event_recovery(&mut self) {
+        let idle_relays = self.missing_event_recovery.lock().unwrap().idle_relays();
+        if idle_relays.is_empty() {
+            return;
+        }
+
+        for relay_url in idle_relays {
+            let Some(pending) = self
+                .missing_event_recovery
+                .lock()
+                .unwrap()
+                .pending_ids(&relay_url)
+            else {
+                continue;
+            };
+
+            // 1. Clear IDs satisfied by other means.
+            let satisfied: Vec<EventId> = match self
+                .database
+                .query(Filter::new().ids(pending.iter().copied()))
+                .await
+            {
+                Ok(events) => events.into_iter().map(|event| event.id).collect(),
+                Err(_) => Vec::new(),
+            };
+            if !satisfied.is_empty() {
+                let outcome = self
+                    .missing_event_recovery
+                    .lock()
+                    .unwrap()
+                    .clear_satisfied(&relay_url, &satisfied);
+                tracing::info!(
+                    relay = %relay_url,
+                    satisfied = satisfied.len(),
+                    "Pending missing events satisfied by local arrivals"
+                );
+                if let Some(outcome) = outcome {
+                    Self::apply_recovery_outcome(
+                        &relay_url,
+                        outcome,
+                        &self.relay_sync_index,
+                        self.metrics.as_ref(),
+                    )
+                    .await;
+                    continue;
+                }
+            }
+
+            // 2. Issue a due attempt over a live connection only.
+            let connection_ready = {
+                let index = self.relay_sync_index.read().await;
+                index
+                    .get(&relay_url)
+                    .map(|state| {
+                        matches!(
+                            state.connection_status,
+                            ConnectionStatus::Syncing
+                                | ConnectionStatus::Connected
+                                | ConnectionStatus::ConnectedHistoricSyncFailures
+                        )
+                    })
+                    .unwrap_or(false)
+            };
+            let connection = self.connections.get(&relay_url).cloned();
+            let (Some(connection), true) = (connection, connection_ready) else {
+                // No usable connection: postpone without consuming the
+                // zero-progress budget so an unavailable relay can neither
+                // expire its pending IDs nor spin in a tight loop.
+                self.missing_event_recovery
+                    .lock()
+                    .unwrap()
+                    .defer_attempt(&relay_url, Instant::now());
+                continue;
+            };
+
+            let Some(attempt) = self
+                .missing_event_recovery
+                .lock()
+                .unwrap()
+                .begin_attempt(&relay_url, Instant::now())
+            else {
+                continue;
+            };
+
+            tokio::spawn(Self::run_missing_event_recovery_attempt(
+                relay_url,
+                connection,
+                attempt,
+                Arc::clone(&self.database),
+                self.write_policy.clone(),
+                self.local_relay.clone(),
+                Arc::clone(&self.rejected_events_index),
+                Arc::clone(&self.missing_event_recovery),
+                self.relay_sync_index.clone(),
+                self.metrics.clone(),
+            ));
+        }
+    }
+
+    /// One bounded exact-ID recovery fetch against a single relay.
+    ///
+    /// Runs outside the sync actor lock. Every event the relay returns is
+    /// passed through the normal write policy; an ID counts as recovered once
+    /// the relay has delivered the event, regardless of the policy verdict
+    /// (rejected events have their own re-processing machinery).
+    #[allow(clippy::too_many_arguments)]
+    async fn run_missing_event_recovery_attempt(
+        relay_url: String,
+        connection: RelayConnection,
+        attempt: missing_events::RecoveryAttempt,
+        database: SharedDatabase,
+        write_policy: Nip34WritePolicy,
+        local_relay: LocalRelay,
+        rejected_events_index: Arc<RejectedEventsIndex>,
+        missing_event_recovery: Arc<std::sync::Mutex<missing_events::MissingEventRecoveryIndex>>,
+        relay_index: RelaySyncIndex,
+        metrics: Option<SyncMetrics>,
+    ) {
+        let requested: HashSet<EventId> = attempt.ids.iter().copied().collect();
+        tracing::info!(
+            relay = %relay_url,
+            attempt = attempt.attempt_number,
+            requested = requested.len(),
+            "Retrying events missing from an incomplete historic sync batch"
+        );
+
+        let events = match connection
+            .fetch_events(
+                Filter::new().ids(attempt.ids.iter().copied()),
+                Duration::from_secs(10),
+            )
+            .await
+        {
+            Ok(events) => events,
+            Err(error) => {
+                tracing::warn!(
+                    relay = %relay_url,
+                    attempt = attempt.attempt_number,
+                    requested = requested.len(),
+                    error = %error,
+                    "Missing-event recovery fetch failed"
+                );
+                Vec::new()
+            }
+        };
+
+        let mut recovered = HashSet::new();
+        for event in events {
+            if !requested.contains(&event.id) {
+                continue;
+            }
+            let result = Self::process_event_static(
+                &event,
+                &relay_url,
+                &database,
+                &write_policy,
+                &local_relay,
+                &rejected_events_index,
+                crate::nostr::persistence::SaveContext::RelaySync,
+            )
+            .await;
+            if result == ProcessResult::Rejected {
+                tracing::debug!(
+                    relay = %relay_url,
+                    event_id = %event.id,
+                    "Recovered missing event was rejected by the write policy"
+                );
+            }
+            recovered.insert(event.id);
+        }
+
+        let outcome = missing_event_recovery.lock().unwrap().complete_attempt(
+            &relay_url,
+            &recovered,
+            Instant::now(),
+        );
+        let Some(outcome) = outcome else {
+            tracing::debug!(
+                relay = %relay_url,
+                "Recovery attempt finished after the relay's pending work was reset"
+            );
+            return;
+        };
+        Self::apply_recovery_outcome(&relay_url, outcome, &relay_index, metrics.as_ref()).await;
+    }
+
+    /// Log a recovery outcome and, on full recovery, restore relay health.
+    ///
+    /// A relay is only promoted from `ConnectedHistoricSyncFailures` back to
+    /// `Connected` when every registered missing ID was recovered and no
+    /// unrelated batch failure was observed for the relay.
+    async fn apply_recovery_outcome(
+        relay_url: &str,
+        outcome: missing_events::AttemptOutcome,
+        relay_index: &RelaySyncIndex,
+        metrics: Option<&SyncMetrics>,
+    ) {
+        use missing_events::AttemptOutcome;
+
+        match outcome {
+            AttemptOutcome::FullyRecovered {
+                recovered,
+                total_recovered,
+                can_restore_health,
+            } => {
+                tracing::info!(
+                    relay = %relay_url,
+                    recovered,
+                    total_recovered,
+                    can_restore_health,
+                    "All events missing from historic sync fully recovered"
+                );
+                if !can_restore_health {
+                    return;
+                }
+                let mut index = relay_index.write().await;
+                if let Some(state) = index.get_mut(relay_url) {
+                    state.historic_sync_had_failures = false;
+                    if state.connection_status == ConnectionStatus::ConnectedHistoricSyncFailures {
+                        state.connection_status = ConnectionStatus::Connected;
+                        tracing::info!(
+                            relay = %relay_url,
+                            "Historic sync failures resolved - relay promoted to Connected"
+                        );
+                        if let Some(metrics) = metrics {
+                            metrics
+                                .record_connection_status(relay_url, ConnectionStatus::Connected);
+                        }
+                    }
+                }
+            }
+            AttemptOutcome::PartiallyRecovered {
+                recovered,
+                remaining,
+                next_attempt_in,
+            } => {
+                tracing::info!(
+                    relay = %relay_url,
+                    recovered,
+                    remaining,
+                    next_attempt_in_secs = next_attempt_in.as_secs_f64(),
+                    "Partially recovered events missing from historic sync - retry scheduled"
+                );
+            }
+            AttemptOutcome::RetryScheduled {
+                attempt,
+                remaining,
+                next_attempt_in,
+            } => {
+                tracing::info!(
+                    relay = %relay_url,
+                    attempt,
+                    remaining,
+                    next_attempt_in_secs = next_attempt_in.as_secs_f64(),
+                    "No missing events recovered - retry scheduled with backoff"
+                );
+            }
+            AttemptOutcome::Expired {
+                remaining,
+                attempts,
+            } => {
+                tracing::warn!(
+                    relay = %relay_url,
+                    remaining,
+                    attempts,
+                    "Missing-event recovery expired by policy after repeated zero-progress attempts - relay remains ConnectedHistoricSyncFailures until daily sync"
+                );
+            }
+        }
+    }
+
     /// Confirm a completed batch by moving items to RelayState
     ///
     /// This method is used by both sync paths (REQ+EOSE and Negentropy) to
@@ -1716,6 +2052,12 @@ impl SyncManager {
             // Track if this batch failed (for ConnectedDegraded transition)
             if batch.failed {
                 state.historic_sync_had_failures = true;
+                // Failures unrelated to a relay's pending missing-event
+                // recovery mean full recovery must not restore its health.
+                self.missing_event_recovery
+                    .lock()
+                    .unwrap()
+                    .note_failed_batch(relay_url, batch_id);
                 tracing::warn!(
                     relay = %relay_url,
                     batch_id = batch_id,
@@ -1893,6 +2235,21 @@ impl SyncManager {
         // Unsubscribe all current subscriptions
         connection.unsubscribe_all().await;
 
+        // Daily sync re-discovers everything from scratch, superseding any
+        // pending missing-event recovery for this relay.
+        let dropped_recovery_ids = self
+            .missing_event_recovery
+            .lock()
+            .unwrap()
+            .clear_relay(relay_url);
+        if dropped_recovery_ids > 0 {
+            tracing::info!(
+                relay = %relay_url,
+                dropped_missing_ids = dropped_recovery_ids,
+                "Daily sync reset pending missing-event recovery"
+            );
+        }
+
         // Clear pending batches for this relay
         {
             let mut pending = self.pending_sync_index.write().await;
@@ -3562,6 +3919,12 @@ impl SyncManager {
                 );
             }
 
+            // Drop pending missing-event recovery along with the relay.
+            self.missing_event_recovery
+                .lock()
+                .unwrap()
+                .clear_relay(relay_url);
+
             // Update metrics - decrement connected count
             if let Some(ref metrics) = self.metrics {
                 metrics.dec_connected_count();

tests/common/censoring_proxy.rs

@@ -0,0 +1,245 @@
+//! Censoring WebSocket Proxy for Sync Tests
+//!
+//! A transparent WebSocket proxy that sits between a syncing relay and its
+//! bootstrap relay, forwarding every frame except `["EVENT", ...]` messages
+//! whose event ID is currently withheld.
+//!
+//! This simulates a relay that reports events during NIP-77 negentropy
+//! reconciliation (NEG-* frames pass through untouched, so the backend's
+//! full event set is visible to reconciliation) but fails to deliver some
+//! of those events on exact-ID REQ fetches — the production behaviour
+//! behind incomplete historic-sync batches.
+//!
+//! # Usage
+//!
+//! ```ignore
+//! let source = TestRelay::start().await;
+//! let proxy = CensoringProxy::start(source.url()).await;
+//! proxy.withhold(event.id);
+//! let syncing = TestRelay::start_with_sync(Some(proxy.url().into())).await;
+//! // ... syncing relay never receives `event` ...
+//! proxy.release(event.id);
+//! // ... the next fetch through the proxy can deliver it ...
+//! ```
+
+use std::collections::HashSet;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Arc, RwLock};
+
+use futures_util::{SinkExt, StreamExt};
+use nostr_sdk::prelude::EventId;
+use tokio::net::TcpListener;
+use tokio::sync::oneshot;
+use tokio_tungstenite::tungstenite::Message;
+
+/// WebSocket proxy that withholds selected EVENT frames from its backend.
+pub struct CensoringProxy {
+    url: String,
+    withheld: Arc<RwLock<HashSet<String>>>,
+    dropped: Arc<AtomicUsize>,
+    shutdown_tx: Option<oneshot::Sender<()>>,
+    handle: Option<tokio::task::JoinHandle<()>>,
+}
+
+impl CensoringProxy {
+    /// Start a proxy on a random loopback port, forwarding to `backend_url`.
+    pub async fn start(backend_url: &str) -> Self {
+        let listener = TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("CensoringProxy failed to bind");
+        let port = listener
+            .local_addr()
+            .expect("CensoringProxy local_addr")
+            .port();
+
+        let withheld: Arc<RwLock<HashSet<String>>> = Arc::new(RwLock::new(HashSet::new()));
+        let dropped = Arc::new(AtomicUsize::new(0));
+        let (shutdown_tx, mut shutdown_rx) = oneshot::channel::<()>();
+
+        let backend_url = backend_url.to_string();
+        let accept_withheld = withheld.clone();
+        let accept_dropped = dropped.clone();
+
+        let handle = tokio::spawn(async move {
+            loop {
+                tokio::select! {
+                    accepted = listener.accept() => {
+                        let Ok((stream, _)) = accepted else { break };
+                        let backend_url = backend_url.clone();
+                        let withheld = accept_withheld.clone();
+                        let dropped = accept_dropped.clone();
+                        tokio::spawn(async move {
+                            if let Err(error) =
+                                proxy_connection(stream, &backend_url, withheld, dropped).await
+                            {
+                                // Disconnects mid-test are expected; log for debugging only.
+                                eprintln!("CensoringProxy connection ended: {error}");
+                            }
+                        });
+                    }
+                    _ = &mut shutdown_rx => break,
+                }
+            }
+        });
+
+        Self {
+            url: format!("ws://127.0.0.1:{port}"),
+            withheld,
+            dropped,
+            shutdown_tx: Some(shutdown_tx),
+            handle: Some(handle),
+        }
+    }
+
+    /// The ws:// URL the syncing relay should use as its bootstrap relay.
+    pub fn url(&self) -> &str {
+        &self.url
+    }
+
+    /// Start withholding EVENT frames carrying this event ID.
+    pub fn withhold(&self, event_id: EventId) {
+        self.withheld
+            .write()
+            .expect("withheld lock poisoned")
+            .insert(event_id.to_hex());
+    }
+
+    /// Stop withholding this event ID. Frames sent by the backend after this
+    /// call pass through; already-dropped frames are not replayed.
+    pub fn release(&self, event_id: EventId) {
+        self.withheld
+            .write()
+            .expect("withheld lock poisoned")
+            .remove(&event_id.to_hex());
+    }
+
+    /// Number of EVENT frames dropped so far.
+    pub fn dropped_count(&self) -> usize {
+        self.dropped.load(Ordering::Relaxed)
+    }
+
+    /// Stop the proxy.
+    pub async fn stop(mut self) {
+        if let Some(tx) = self.shutdown_tx.take() {
+            let _ = tx.send(());
+        }
+        if let Some(handle) = self.handle.take() {
+            let _ = handle.await;
+        }
+    }
+}
+
+impl Drop for CensoringProxy {
+    fn drop(&mut self) {
+        if let Some(tx) = self.shutdown_tx.take() {
+            let _ = tx.send(());
+        }
+    }
+}
+
+/// Forward one client connection to the backend, censoring backend->client
+/// EVENT frames for withheld IDs.
+async fn proxy_connection(
+    client_stream: tokio::net::TcpStream,
+    backend_url: &str,
+    withheld: Arc<RwLock<HashSet<String>>>,
+    dropped: Arc<AtomicUsize>,
+) -> Result<(), String> {
+    let client_ws = tokio_tungstenite::accept_async(client_stream)
+        .await
+        .map_err(|e| format!("client handshake failed: {e}"))?;
+    let (backend_ws, _) = tokio_tungstenite::connect_async(backend_url)
+        .await
+        .map_err(|e| format!("backend connect failed: {e}"))?;
+
+    let (mut client_tx, mut client_rx) = client_ws.split();
+    let (mut backend_tx, mut backend_rx) = backend_ws.split();
+
+    // Client -> backend: forward untouched.
+    let upstream = async {
+        while let Some(message) = client_rx.next().await {
+            let message = message.map_err(|e| format!("client read: {e}"))?;
+            backend_tx
+                .send(message)
+                .await
+                .map_err(|e| format!("backend write: {e}"))?;
+        }
+        Ok::<(), String>(())
+    };
+
+    // Backend -> client: drop withheld EVENT frames, forward everything else
+    // (including NEG-MSG, EOSE, NOTICE, and control frames).
+    let downstream = async {
+        while let Some(message) = backend_rx.next().await {
+            let message = message.map_err(|e| format!("backend read: {e}"))?;
+            if let Message::Text(text) = &message {
+                let ids = withheld.read().expect("withheld lock poisoned");
+                if is_withheld_event_frame(text.as_str(), &ids) {
+                    drop(ids);
+                    dropped.fetch_add(1, Ordering::Relaxed);
+                    continue;
+                }
+            }
+            client_tx
+                .send(message)
+                .await
+                .map_err(|e| format!("client write: {e}"))?;
+        }
+        Ok::<(), String>(())
+    };
+
+    // Either side ending tears the whole proxied connection down.
+    tokio::select! {
+        result = upstream => result,
+        result = downstream => result,
+    }
+}
+
+/// True if `text` is a NIP-01 `["EVENT", <sub>, {..}]` frame whose event ID
+/// is in the withheld set.
+fn is_withheld_event_frame(text: &str, withheld: &HashSet<String>) -> bool {
+    if withheld.is_empty() {
+        return false;
+    }
+    let Ok(value) = serde_json::from_str::<serde_json::Value>(text) else {
+        return false;
+    };
+    let Some(array) = value.as_array() else {
+        return false;
+    };
+    if array.first().and_then(|v| v.as_str()) != Some("EVENT") {
+        return false;
+    }
+    array
+        .get(2)
+        .and_then(|event| event.get("id"))
+        .and_then(|id| id.as_str())
+        .is_some_and(|id| withheld.contains(id))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn withheld_event_frames_are_detected() {
+        let mut withheld = HashSet::new();
+        withheld.insert("ab".repeat(32));
+        let id = "ab".repeat(32);
+
+        let event_frame = format!(r#"["EVENT","sub",{{"id":"{id}","kind":1}}]"#);
+        assert!(is_withheld_event_frame(&event_frame, &withheld));
+
+        let other_frame = r#"["EVENT","sub",{"id":"cd","kind":1}]"#;
+        assert!(!is_withheld_event_frame(other_frame, &withheld));
+
+        let eose_frame = r#"["EOSE","sub"]"#;
+        assert!(!is_withheld_event_frame(eose_frame, &withheld));
+
+        let neg_frame = format!(r#"["NEG-MSG","sub","{id}"]"#);
+        assert!(
+            !is_withheld_event_frame(&neg_frame, &withheld),
+            "negentropy frames must pass through so reconciliation still reports the event"
+        );
+    }
+}

tests/common/mod.rs

@@ -2,6 +2,7 @@
 #![allow(dead_code)] // Test helpers may not be used in all test configurations
 #![allow(unused_imports)] // Re-exports may not be used in all test configurations
 
+pub mod censoring_proxy;
 pub mod git_server;
 pub mod mock_relay;
 pub mod nip09_helpers;

tests/sync.rs

@@ -33,6 +33,7 @@ mod common;
 mod sync {
     pub mod catchup;
     pub mod discovery;
+    pub mod historic_recovery;
     pub mod historic_sync;
     pub mod live_sync;
     pub mod maintainer_reprocessing;

tests/sync/historic_recovery.rs

@@ -0,0 +1,178 @@
+//! Historic Sync Missing-Event Recovery Tests
+//!
+//! Regression coverage for a production failure observed on gitnostr.com:
+//! NIP-77 negentropy reconciliation identifies event IDs missing locally, but
+//! the relay's exact-ID REQ response returns only a subset of them. For
+//! batches without repository or root-event metadata (the generic Layer-1
+//! announcements batch), no semantic REQ+EOSE fallback can be constructed, and
+//! the batch used to be finalized with partial results — silently dropping the
+//! missing IDs until the next daily sync (23-25h later).
+//!
+//! Production log sequence being reproduced:
+//! - "Negentropy sync incomplete - relay returned fewer events than requested"
+//! - "Cannot create semantic fallback filters - no repos or root_events in batch"
+//! - "Failed to create REQ+EOSE fallback subscriptions - completing batch with partial results"
+//! - "Batch failed - will transition to ConnectedHistoricSyncFailures instead of Connected"
+//!
+//! The test drives the real sync path end to end: a genuine ngit-grasp source
+//! relay (with real NIP-77 support) sits behind a censoring WebSocket proxy
+//! that withholds one event's EVENT frames while letting negentropy frames
+//! through, so reconciliation keeps reporting the event as available.
+
+use std::time::Duration;
+
+use nostr_sdk::prelude::*;
+
+use crate::common::censoring_proxy::CensoringProxy;
+use crate::common::sync_helpers::{fetch_metrics, send_to_relay, wait_for_event_on_relay};
+use crate::common::TestRelay;
+
+/// Build a kind 10317 (GitUserGraspList) event.
+///
+/// Kind 10317 is part of the Layer-1 announcements filter and is stored
+/// directly in the relay DB (no purgatory / git-data gating), which keeps this
+/// scenario focused on the sync path rather than announcement promotion.
+fn grasp_list_event(identifier: &str) -> Event {
+    EventBuilder::new(Kind::GitUserGraspList, "")
+        .tags(vec![Tag::identifier(identifier)])
+        .finalize(&Keys::generate())
+        .expect("Failed to sign grasp list event")
+}
+
+/// Read the `ngit_sync_relay_connected` gauge for the single tracked relay.
+async fn connection_status_gauge(relay_url: &str) -> Option<i64> {
+    let metrics = fetch_metrics(relay_url).await.ok()?;
+    for line in metrics.lines() {
+        if line.starts_with("ngit_sync_relay_connected{") {
+            return line
+                .split_whitespace()
+                .last()
+                .and_then(|value| value.parse::<f64>().ok())
+                .map(|value| value as i64);
+        }
+    }
+    None
+}
+
+/// Wait until the connection-status gauge reaches `expected`.
+async fn wait_for_connection_status(relay_url: &str, expected: i64, timeout: Duration) -> bool {
+    let deadline = tokio::time::Instant::now() + timeout;
+    loop {
+        if connection_status_gauge(relay_url).await == Some(expected) {
+            return true;
+        }
+        if tokio::time::Instant::now() >= deadline {
+            return false;
+        }
+        tokio::time::sleep(Duration::from_millis(300)).await;
+    }
+}
+
+/// Scenario:
+/// 1. Source relay holds two Layer-1 events; negentropy reports both missing.
+/// 2. The proxy withholds one of them, so the exact-ID fetch returns a subset
+///    and the ID-based retry returns nothing (zero progress).
+/// 3. The announcements batch has no repos/root_events, so no semantic
+///    fallback exists; the batch completes with failures and the relay
+///    transitions to ConnectedHistoricSyncFailures (gauge value 4).
+/// 4. Live sync for later unrelated events keeps working (no starvation).
+/// 5. The withheld event becomes available; bounded background recovery must
+///    fetch and store it without restarting the relay, after which the relay
+///    is promoted to Connected (gauge value 3).
+#[tokio::test]
+async fn incomplete_exact_id_fetch_recovers_after_event_becomes_available() {
+    // 1. Source relay with two Layer-1 events (kind 10317).
+    let source = TestRelay::start().await;
+    let delivered = grasp_list_event("grasp-list-delivered");
+    let withheld = grasp_list_event("grasp-list-withheld");
+    send_to_relay(&source, &delivered)
+        .await
+        .expect("send delivered event to source");
+    send_to_relay(&source, &withheld)
+        .await
+        .expect("send withheld event to source");
+
+    // 2. Censoring proxy in front of the source; withhold one event before
+    //    the syncing relay ever connects.
+    let proxy = CensoringProxy::start(source.url()).await;
+    proxy.withhold(withheld.id);
+
+    // 3. Syncing relay bootstraps through the proxy.
+    let syncing = TestRelay::start_with_sync(Some(proxy.url().to_string())).await;
+
+    // The non-withheld event flows through the initial historic sync.
+    assert!(
+        wait_for_event_on_relay(
+            syncing.url(),
+            Filter::new().id(delivered.id),
+            Duration::from_secs(20),
+        )
+        .await,
+        "delivered event should reach the syncing relay via historic sync"
+    );
+
+    // 4. The incomplete batch must complete with failures (status 4), not be
+    //    reported as fully synchronized (status 3).
+    assert!(
+        wait_for_connection_status(syncing.url(), 4, Duration::from_secs(30)).await,
+        "relay should report ConnectedHistoricSyncFailures while an event is withheld"
+    );
+    assert!(
+        proxy.dropped_count() >= 2,
+        "proxy should have censored the initial fetch and at least one retry, dropped: {}",
+        proxy.dropped_count()
+    );
+    assert!(
+        !wait_for_event_on_relay(
+            syncing.url(),
+            Filter::new().id(withheld.id),
+            Duration::from_secs(1),
+        )
+        .await,
+        "withheld event must not have reached the syncing relay yet"
+    );
+
+    // 5. Later unrelated work is not starved: a new live event flows through
+    //    the same proxied connection while the missing ID is still pending.
+    let live = grasp_list_event("grasp-list-live");
+    send_to_relay(&source, &live)
+        .await
+        .expect("send live event to source");
+    assert!(
+        wait_for_event_on_relay(
+            syncing.url(),
+            Filter::new().id(live.id),
+            Duration::from_secs(20),
+        )
+        .await,
+        "live sync should keep delivering unrelated events while recovery is pending"
+    );
+    assert_eq!(
+        connection_status_gauge(syncing.url()).await,
+        Some(4),
+        "relay must remain in ConnectedHistoricSyncFailures while the event is still withheld"
+    );
+
+    // 6. The withheld event becomes available. Bounded background recovery
+    //    must retry the missing ID and store the event without a restart.
+    proxy.release(withheld.id);
+    assert!(
+        wait_for_event_on_relay(
+            syncing.url(),
+            Filter::new().id(withheld.id),
+            Duration::from_secs(45),
+        )
+        .await,
+        "withheld event should be recovered after it becomes available (was the missing ID dropped?)"
+    );
+
+    // 7. Once every missing ID is recovered the relay is honestly complete.
+    assert!(
+        wait_for_connection_status(syncing.url(), 3, Duration::from_secs(30)).await,
+        "relay should transition to Connected after full recovery"
+    );
+
+    syncing.stop().await;
+    proxy.stop().await;
+    source.stop().await;
+}

tests/sync/mod.rs

@@ -128,6 +128,7 @@
 //! - `fetch_metrics()` - Prometheus metrics fetching
 
 // Test modules
+pub mod historic_recovery;
 pub mod historic_sync;
 pub mod catchup;
 pub mod discovery;