Merge #a44ff1d9: fix(sync): adapt query starts after rate-limit exhaust…

fix(sync): adapt query starts after rate-limit exhaustion

nostr:nevent1qgsx2lyl2e4zvfadwcvkd9fkrcwczj7mf858hy85mwqclwgut8wpg2spz3mhxue69uhhyetvv9ujumn8d96zuer9wcq3yamnwvaz7tm8d96xummnw3ezucm0d5q3kamnwvaz7tmwva5hgtnyv9hxxmmwwashjer9wchxxmmdqqs2gnl3m9fv4zvge4mfc0rzfgpgfe7l4zemrr62crwsvfjcdp5u9uq9y99ms

PR-Author: DanConwayDev's Agent
nostr:npub1v47f74n2ycn66asev62nv8sas99akj0g0wg0fkup37u3ckwuzs4q7cwtp0

CoverNote:

Production showed a roughly 67-second oscillator against relay.ngit.dev: simultaneous subscriptions stayed bounded, but query turnover exhausted a separate 120-query-per-minute allowance after every cooldown. That allowance is a newly enabled rust-nostr 0.45 LocalRelay default, not an ecosystem floor: the preserved cross-relay survey in commit 9723ff4 found no other audited implementation enabling an equivalent query-specific per-connection default. A finite limit remains useful as one DoS layer, but a small per-connection bucket is weak protection on its own because connections can be multiplied.

Production iteration also showed that rust-nostr charges multiple SDK-managed NEG-MSG continuation frames inside one admitted NIP-77 round. This revision therefore paces all application-visible starts only after an explicit refusal and uses paced REQs for the rest of that session, because internal NEG-MSG frames cannot honor the gate. It intentionally does not proactively pace startup history: doing that safely requires request-class priority so live coverage is not delayed and is a separate follow-up.

Validation: 657 library tests, 89 sync integration tests (1 ignored), and nix build .#ngit-grasp. Production soak of the behaviorally identical pre-documentation tip: one expected startup refusal; immediate 600 ms pacing and session REQ fallback; zero further refusals across more than four former failure periods; four successful ReqEose batches advancing 13 state-only repositories; NRestarts=0. Final commit 86d19b7 changes documentation and comments only relative to that soaked runtime.

CHANGELOG.md

@@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Pace query starts after a relay reports its per-connection query-rate budget
+  is exhausted, preventing fixed cooldown recovery from replaying the same
+  fast historic-sync burst indefinitely. If an SDK-managed NIP-77 exchange
+  exhausts that budget, use paced REQs for the rest of the connection session
+  because the application cannot pace individual `NEG-MSG` frames. This is a
+  reactive compatibility backstop for an unadvertised limit; proactive pacing
+  of lower-priority historic work remains separate so live coverage is not
+  delayed.
+
 ## [2.1.0] - 2026-08-07
 
 ngit-grasp 2.1.0 substantially improves proactive repository-event sync

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

@@ -1062,6 +1062,17 @@ Probing -> previous state: Recovery REQs succeed
 Probing -> RateLimited: Recovery REQ is rate limited again
 ```
 
+The subscription ledger limits simultaneous work, while some relays also
+limit completed query operations per minute. A `too many queries` refusal
+therefore makes already-queued starts unwind during the 65-second cooldown so
+their work can be re-derived in priority order, then activates per-connection
+query-start pacing: 600 ms between starts initially, doubling for a distinct
+later episode up to 10 seconds. The pacing gate is shared by live and transient
+REQs, exact-ID fetches and NIP-77 round starts, and resets with the relay
+connection session. Any query-rate refusal disables NIP-77 for the rest of the
+session and falls back to paced REQs, because rust-nostr owns the internal
+`NEG-MSG` frames and the application cannot guarantee their pacing.
+
 ### Backoff Configuration
 
 - **Formula**: `base_backoff * 2^(failures-1)`, capped at `max_backoff`

docs/explanation/monitoring.md

@@ -146,6 +146,15 @@ The `ngit_sync_relay_status` metric tracks relay health:
 - `4` = **Dead** - 24h+ of continuous failures
 - `5` = **RateLimited** - Rate limit cooldown active (65s)
 
+After a `too many queries` response, `Activated adaptive query-start pacing`
+reports the learned per-connection interval. It begins at 600 ms and doubles
+only when a distinct later episode proves that pace too fast. Queued starts
+unwind during the existing 65-second cooldown before paced recovery;
+reconnecting starts a fresh session without pacing.
+`Falling back to paced REQs for the query-limited connection session` means
+NIP-77 remains skipped until reconnect because SDK-managed `NEG-MSG` traffic
+cannot be passed individually through the learned gate.
+
 ### Example Grafana Queries
 
 ```promql

docs/explanation/sync-scaling-constraints.md

@@ -181,6 +181,16 @@ strfry and Ditto have no native limiter at all, deferring to deployment
 infrastructure. The full survey with citations is preserved in this
 file's history (commit `9723ff4`).
 
+The 120-query allowance is a newly enabled rust-nostr 0.45 LocalRelay default,
+not a floor established by that survey. It is unusually restrictive: none of
+the other audited implementations enables an equivalent query-specific,
+per-connection default. A finite limit is still useful as one layer of DoS
+protection, but a small per-connection bucket is not sufficient protection by
+itself because a hostile client can multiply connections; per-IP admission and
+global resource bounds address that threat more directly. rust-nostr also
+charges SDK-managed NIP-77 `NEG-MSG` continuation frames to this same bucket,
+so one application-started reconciliation can consume multiple query tokens.
+
 NIP-11 describes hard relay limitations, not rate-limit algorithms. The
 standard fields relevant here are `max_limit` and
 `max_subscriptions`; it has no standard fields for simultaneous connections
@@ -190,6 +200,14 @@ whether it applies independently to each filter or to the merged REQ, which is
 why the source audit above remains necessary. Relay-specific extensions can
 add fields, but clients cannot assume common names or semantics.
 
+The implemented query-start pacer is therefore a reactive compatibility
+backstop: it remains inactive until an explicit `too many queries` response,
+then spaces all application-visible starts and selects REQ fallback because
+the application cannot pace individual `NEG-MSG` frames. Proactively being
+gentle with non-urgent historic work is desirable but separate. It requires
+request-class priority—live coverage first, dependency recovery next, bulk
+history last—rather than enabling this shared gate from connection startup.
+
 The client encodes this model in per-connection `RelayPaginationSession`
 state (`src/sync/mod.rs`). After EOSE it learns the largest raw page seen
 from that relay and computes `max(90, floor(0.9 × estimated_cap))`, where
@@ -379,6 +397,19 @@ So concurrency is not a free scaling axis; it is the residual of the ledger:
 - Rounds queue behind a per-connection semaphore; each completion releases
   the next. No timed batches or sleeps — throughput degrades smoothly instead
   of bursting into rejections.
+- The ledger bounds simultaneous resource use, not query starts over time. If
+  a relay returns `rate-limited: too many queries`, the current connection
+  first rejects locally queued starts for the existing 65-second cooldown so
+  their incomplete work can be re-derived in priority order, then learns a
+  shared query-start interval: 600 ms initially (100 starts/minute, below our
+  own 120/minute serving limit), doubling on a later rate-limit episode up to
+  10 seconds. Live, transient REQ, exact-ID fetch and NIP-77 round starts all
+  pass through that pacer. Relays that never report a query
+  rate limit remain unpaced, and reconnecting resets the per-session lesson.
+  After any query-budget refusal, NIP-77 is skipped for the rest of the session:
+  rust-nostr owns the internal `NEG-MSG` exchange, so the application cannot
+  guarantee that each charged frame passes through its pacing gate. Historic
+  recovery then uses the paced REQ path.
 - Transient REQ+EOSE subscriptions — historic sync groups, fallback
   filters, exact-ID fetches, retries, and pagination pages — retain a
   five-request class cap inside the shared ledger: a slot is acquired when
@@ -501,8 +532,10 @@ is worthwhile upstream work.
 **Cons:** reactive by construction (eats one rejection burst per relay per
 startup), needs heuristic NOTICE parsing as its *primary* control loop, and
 fixed pauses waste time on fast relays while still bursting slow ones.
-**Why not:** the semaphore ledger achieves the same containment continuously,
-with the heuristics demoted to backstop.
+**Why not:** fixed global batching would penalise every relay and cannot adapt
+to their different windows. The semaphore ledger continuously contains active
+resources; the implemented per-session pacer is activated only by an explicit
+query-rate refusal and then drains work smoothly at a learned rate.
 
 ---
 

src/sync/health.rs

@@ -35,7 +35,7 @@ const DEFAULT_MAX_BACKOFF_SECS: u64 = 3600;
 const DEFAULT_BASE_BACKOFF_SECS: u64 = 5;
 
 /// Rate limit cooldown duration in seconds (65 seconds = typical 60s limit + buffer)
-const RATE_LIMIT_COOLDOWN_SECS: u64 = 65;
+pub(super) const RATE_LIMIT_COOLDOWN_SECS: u64 = 65;
 
 /// Stability period after recovery before marking relay as fully healthy (5 minutes)
 /// A relay must maintain connection for this duration after failures before being marked Healthy

src/sync/relay_connection.rs

@@ -21,6 +21,7 @@ use std::future::Future;
 use std::time::Duration;
 use tokio::sync::mpsc;
 
+use super::health::RATE_LIMIT_COOLDOWN_SECS;
 use super::is_rate_limit_message;
 use crate::nostr::SharedDatabase;
 use crate::outbound::{OutboundTargetKind, OutboundTargetPolicy, RelayTargetSource};
@@ -40,6 +41,23 @@ const FALLBACK_SUBSCRIPTION_BUDGET: usize = 20;
 /// Control-plane safety slots never offered to live or historic consumers.
 const SUBSCRIPTION_RESERVED_MARGIN: usize = 2;
 
+/// Initial spacing after a relay reports that its per-connection query-rate
+/// allowance is exhausted.
+///
+/// NIP-11 cannot advertise query-rate windows. One start every 600 ms is 100
+/// starts/minute, leaving operational slack below rust-nostr 0.45's newly
+/// introduced 120/minute LocalRelay default. This is deliberately reactive:
+/// proactively pacing non-urgent history needs request-class priority so live
+/// coverage is not delayed, and belongs in a separate scheduler change.
+const QUERY_PACING_INITIAL_INTERVAL: Duration = Duration::from_millis(600);
+
+/// Bound adaptive slowdown so a restrictive relay still makes progress.
+const QUERY_PACING_MAX_INTERVAL: Duration = Duration::from_secs(10);
+
+/// Multiple refusals from one already-sent burst describe one episode. A new
+/// refusal after a full rate-limit window means the learned pace was too fast.
+const QUERY_RATE_LIMIT_EPISODE_GAP: Duration = Duration::from_secs(60);
+
 fn effective_subscription_budget(advertised: Option<usize>) -> usize {
     advertised.unwrap_or(FALLBACK_SUBSCRIPTION_BUDGET)
 }
@@ -48,6 +66,91 @@ fn usable_subscription_slots(advertised: Option<usize>) -> usize {
     effective_subscription_budget(advertised).saturating_sub(SUBSCRIPTION_RESERVED_MARGIN)
 }
 
+fn is_query_rate_limit_message(message: &str) -> bool {
+    message.to_ascii_lowercase().contains("too many queries")
+}
+
+#[derive(Debug, Default)]
+struct QueryPacingState {
+    interval: Option<Duration>,
+    last_start: Option<tokio::time::Instant>,
+    last_limit_signal: Option<tokio::time::Instant>,
+    paused_until: Option<tokio::time::Instant>,
+}
+
+/// Serialises query starts only after this connection has demonstrated a
+/// per-minute query limit. Simultaneous subscription capacity remains owned by
+/// the separate subscription ledger.
+#[derive(Debug, Default)]
+struct QueryStartPacer {
+    gate: tokio::sync::Mutex<()>,
+    state: std::sync::Mutex<QueryPacingState>,
+}
+
+impl QueryStartPacer {
+    fn record_rate_limit(&self) -> (Duration, bool) {
+        let now = tokio::time::Instant::now();
+        let mut state = self.state.lock().expect("query pacing state poisoned");
+        let new_episode = state
+            .last_limit_signal
+            .is_none_or(|last| now.duration_since(last) >= QUERY_RATE_LIMIT_EPISODE_GAP);
+        if new_episode {
+            state.interval = Some(match state.interval {
+                None => QUERY_PACING_INITIAL_INTERVAL,
+                Some(current) => current.saturating_mul(2).min(QUERY_PACING_MAX_INTERVAL),
+            });
+            state.paused_until = Some(now + Duration::from_secs(RATE_LIMIT_COOLDOWN_SECS));
+        }
+        state.last_limit_signal = Some(now);
+        (
+            state.interval.unwrap_or(QUERY_PACING_INITIAL_INTERVAL),
+            new_episode,
+        )
+    }
+
+    async fn wait_for_start(&self) -> Result<(), Duration> {
+        let _gate = self.gate.lock().await;
+        let now = tokio::time::Instant::now();
+        if let Some(paused_until) = self
+            .state
+            .lock()
+            .expect("query pacing state poisoned")
+            .paused_until
+        {
+            if now < paused_until {
+                return Err(paused_until.duration_since(now));
+            }
+        }
+        let deadline = {
+            let state = self.state.lock().expect("query pacing state poisoned");
+            state
+                .interval
+                .zip(state.last_start)
+                .map(|(interval, last)| last + interval)
+        };
+        if let Some(deadline) = deadline {
+            tokio::time::sleep_until(deadline).await;
+        }
+        let mut state = self.state.lock().expect("query pacing state poisoned");
+        if state.interval.is_some() {
+            state.last_start = Some(tokio::time::Instant::now());
+        }
+        Ok(())
+    }
+
+    fn reset(&self) {
+        *self.state.lock().expect("query pacing state poisoned") = QueryPacingState::default();
+    }
+
+    #[cfg(test)]
+    fn interval(&self) -> Option<Duration> {
+        self.state
+            .lock()
+            .expect("query pacing state poisoned")
+            .interval
+    }
+}
+
 #[cfg(test)]
 fn historic_slot_allowance(advertised: Option<usize>, live_slots: usize) -> usize {
     usable_subscription_slots(advertised)
@@ -309,6 +412,9 @@ pub struct RelayConnection {
     nip77_warning_logged: std::sync::Arc<std::sync::atomic::AtomicBool>,
     /// Whether this relay supports NIP-77 negentropy (0 = unknown, 2 = confirmed not supported)
     nip77_supported: std::sync::Arc<std::sync::atomic::AtomicU8>,
+    /// Whether an in-session NIP-77 round exhausted the relay's query-rate
+    /// budget. The SDK owns NEG-MSG exchange, so later work uses paced REQs.
+    nip77_query_rate_limited: std::sync::Arc<std::sync::atomic::AtomicBool>,
     /// Consecutive transient negentropy failures (drives the cooldown schedule)
     nip77_transient_failures: std::sync::Arc<std::sync::atomic::AtomicU32>,
     /// Deadline before which negentropy is not attempted (transient-failure cooldown)
@@ -330,9 +436,41 @@ pub struct RelayConnection {
     transient_req_permits_held: TransientReqPermitMap,
     /// Ledger slots held for persistent subscriptions until CLOSE/teardown.
     live_req_permits_held: LiveReqPermitMap,
+    /// Learned per-session spacing for query starts after a query-rate refusal.
+    query_start_pacer: std::sync::Arc<QueryStartPacer>,
 }
 
 impl RelayConnection {
+    async fn await_query_start(&self) -> Result<(), String> {
+        self.query_start_pacer
+            .wait_for_start()
+            .await
+            .map_err(|remaining| {
+                format!(
+                    "Query start deferred for {}: rate-limit cooldown {:.3}s remaining",
+                    self.url,
+                    remaining.as_secs_f64()
+                )
+            })
+    }
+
+    fn record_query_rate_limit(&self) {
+        let (interval, new_episode) = self.query_start_pacer.record_rate_limit();
+        if new_episode {
+            tracing::info!(
+                relay = %self.url,
+                query_start_interval_ms = interval.as_millis(),
+                "Activated adaptive query-start pacing"
+            );
+        }
+        if self.mark_negentropy_query_rate_limited() {
+            tracing::info!(
+                relay = %self.url,
+                "Falling back to paced REQs for the query-limited connection session"
+            );
+        }
+    }
+
     /// Normalize a relay URL to include a scheme (wss:// or ws://)
     ///
     /// If the URL already has a scheme, it's returned as-is.
@@ -381,6 +519,9 @@ impl RelayConnection {
             database: None,
             nip77_warning_logged: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
             nip77_supported: std::sync::Arc::new(std::sync::atomic::AtomicU8::new(0)),
+            nip77_query_rate_limited: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(
+                false,
+            )),
             nip77_transient_failures: std::sync::Arc::new(std::sync::atomic::AtomicU32::new(0)),
             nip77_cooldown_until: std::sync::Arc::new(std::sync::Mutex::new(None)),
             neg_diff_permits: std::sync::Arc::new(tokio::sync::Semaphore::new(
@@ -406,6 +547,7 @@ impl RelayConnection {
             live_req_permits_held: std::sync::Arc::new(std::sync::Mutex::new(
                 std::collections::HashMap::new(),
             )),
+            query_start_pacer: std::sync::Arc::new(QueryStartPacer::default()),
         }
     }
 
@@ -436,6 +578,9 @@ impl RelayConnection {
             database: Some(database),
             nip77_warning_logged: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)),
             nip77_supported: std::sync::Arc::new(std::sync::atomic::AtomicU8::new(0)),
+            nip77_query_rate_limited: std::sync::Arc::new(std::sync::atomic::AtomicBool::new(
+                false,
+            )),
             nip77_transient_failures: std::sync::Arc::new(std::sync::atomic::AtomicU32::new(0)),
             nip77_cooldown_until: std::sync::Arc::new(std::sync::Mutex::new(None)),
             neg_diff_permits: std::sync::Arc::new(tokio::sync::Semaphore::new(
@@ -461,6 +606,7 @@ impl RelayConnection {
             live_req_permits_held: std::sync::Arc::new(std::sync::Mutex::new(
                 std::collections::HashMap::new(),
             )),
+            query_start_pacer: std::sync::Arc::new(QueryStartPacer::default()),
         }
     }
 
@@ -638,6 +784,9 @@ impl RelayConnection {
     /// Configure the one per-session ledger before any subscriptions open.
     pub fn reset_subscription_budget(&self, advertised: Option<usize>) {
         self.clear_subscription_permits();
+        self.query_start_pacer.reset();
+        self.nip77_query_rate_limited
+            .store(false, std::sync::atomic::Ordering::Relaxed);
         let budget = effective_subscription_budget(advertised);
         let usable = usable_subscription_slots(advertised);
         self.subscription_usable_slots
@@ -1019,6 +1168,9 @@ impl RelayConnection {
                         }
                     }
                     RelayMessage::Notice(msg) => {
+                        if is_query_rate_limit_message(&msg) {
+                            self.record_query_rate_limit();
+                        }
                         // Check if this is a negentropy-related notice
                         let is_negentropy_notice = msg.contains("envelope")
                             || msg.contains("NEG-")
@@ -1047,6 +1199,9 @@ impl RelayConnection {
                         // this processor-facing path retains live restoration.
                         let subscription_id = subscription_id.into_owned();
                         let live_generation = self.release_live_req_permit(&subscription_id);
+                        if is_query_rate_limit_message(&msg) {
+                            self.record_query_rate_limit();
+                        }
                         if is_rate_limit_message(&msg) {
                             // The sync actor emits one canonical signal and owns
                             // cooldown deduplication for a rate-limit episode.
@@ -1227,6 +1382,10 @@ impl RelayConnection {
             "subscribe_filters called"
         );
 
+        // Acquire pacing before subscription/class permits: a learned remote
+        // query-rate window must not turn local capacity into queued sleepers.
+        self.await_query_start().await?;
+
         // Transient (auto-close) subscriptions share a bounded number of
         // per-connection slots so historic bursts queue instead of
         // exceeding relay subscription budgets. The permit is registered
@@ -1359,6 +1518,7 @@ impl RelayConnection {
         // Exact-ID purgatory recovery opens an ordinary transient REQ inside
         // nostr-sdk. It must share the same permit bound as historic pages,
         // fallbacks and retries rather than escaping the connection budget.
+        self.await_query_start().await?;
         if self.historic_capacity_consumed_by_live() {
             tracing::warn!(
                 relay = %self.url,
@@ -1647,6 +1807,14 @@ impl RelayConnection {
             return false;
         }
 
+        if self
+            .nip77_query_rate_limited
+            .load(std::sync::atomic::Ordering::Relaxed)
+        {
+            tracing::trace!(relay = %self.url, "Skipping negentropy - query-rate budget was exhausted this session");
+            return false;
+        }
+
         // Transient-failure cooldown: pause negentropy without ruling it out
         let cooldown_until = *self
             .nip77_cooldown_until
@@ -1674,6 +1842,18 @@ impl RelayConnection {
             .store(2, std::sync::atomic::Ordering::Relaxed);
     }
 
+    /// Fall back to paced REQs for the rest of this connection session.
+    ///
+    /// rust-nostr owns the messages within a NIP-77 round, while relays can
+    /// charge every NEG-MSG against the same query-rate bucket as REQ. We can
+    /// pace round starts but not those internal messages, so retrying NIP-77
+    /// after cooldown can deterministically exhaust the bucket again.
+    fn mark_negentropy_query_rate_limited(&self) -> bool {
+        !self
+            .nip77_query_rate_limited
+            .swap(true, std::sync::atomic::Ordering::Relaxed)
+    }
+
     /// Record a transient negentropy failure and start (or keep) a cooldown.
     ///
     /// Failures that arrive while a cooldown is already active come from
@@ -1747,6 +1927,7 @@ impl RelayConnection {
         // and relays count each open round against a per-connection budget
         // shared with live subscriptions (see MAX_CONCURRENT_NEG_DIFFS).
         // The permit is held for the whole round, including the timeout.
+        self.await_query_start().await?;
         if self.historic_capacity_consumed_by_live() {
             tracing::warn!(
                 relay = %self.url,
@@ -1852,6 +2033,9 @@ impl RelayConnection {
                 Ok(reconciliation)
             }
             Err(e) => {
+                if is_query_rate_limit_message(&e) {
+                    self.record_query_rate_limit();
+                }
                 match classify_negentropy_failure(&e) {
                     NegentropyFailure::Unsupported => {
                         self.mark_negentropy_unsupported();
@@ -2301,6 +2485,60 @@ mod tests {
         drain.abort();
     }
 
+    #[tokio::test]
+    async fn query_rate_closed_paces_later_wire_requests() {
+        let relay = LocalRelayBuilder::default().queries_per_minute(1).build();
+        relay.run().await.expect("start query-limited relay");
+        let connection = RelayConnection::new(
+            relay.url().await.to_string(),
+            Keys::generate(),
+            RelayTargetSource::OperatorConfigured,
+            OutboundTargetPolicy::default(),
+        );
+        connection
+            .connect(3)
+            .await
+            .expect("connect query-limited relay");
+        let (event_tx, mut event_rx) = tokio::sync::mpsc::channel(RELAY_EVENT_BUFFER_CAPACITY);
+        let event_loop = tokio::spawn(connection.clone().run_event_loop(event_tx));
+        let drain = tokio::spawn(async move { while event_rx.recv().await.is_some() {} });
+
+        let filter = Filter::new().kind(Kind::Custom(65_534));
+        connection
+            .subscribe_filter(filter.clone(), TransientRequestClass::HistoricPage)
+            .await
+            .expect("first query consumes the relay token");
+        let _ = connection
+            .subscribe_filter(filter.clone(), TransientRequestClass::HistoricPage)
+            .await;
+
+        tokio::time::timeout(Duration::from_secs(2), async {
+            while connection.query_start_pacer.interval().is_none() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("rate-limited CLOSED should activate pacing");
+        assert!(
+            !connection.supports_negentropy().await,
+            "query-limited sessions must avoid unpaced NEG-MSG traffic"
+        );
+
+        let error = connection
+            .subscribe_filter(filter.clone(), TransientRequestClass::HistoricPage)
+            .await
+            .expect_err("work queued during cooldown must unwind");
+        assert!(
+            error.contains("rate-limit cooldown"),
+            "unexpected deferred-query error: {error}"
+        );
+
+        connection.disconnect().await;
+        relay.shutdown();
+        event_loop.abort();
+        drain.abort();
+    }
+
     #[tokio::test]
     async fn fetch_events_reports_an_unregistered_exact_relay() {
         let connection = permissive_connection("ws://127.0.0.1:1", Keys::generate());
@@ -2930,6 +3168,97 @@ mod tests {
         );
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn query_pacer_activates_only_after_query_rate_limit() {
+        let pacer = std::sync::Arc::new(QueryStartPacer::default());
+        pacer
+            .wait_for_start()
+            .await
+            .expect("inactive pacer should admit immediately");
+        assert_eq!(pacer.interval(), None);
+
+        let (interval, new_episode) = pacer.record_rate_limit();
+        assert!(new_episode);
+        assert_eq!(interval, QUERY_PACING_INITIAL_INTERVAL);
+        assert!(pacer.wait_for_start().await.is_err());
+        tokio::time::advance(Duration::from_secs(RATE_LIMIT_COOLDOWN_SECS)).await;
+        pacer
+            .wait_for_start()
+            .await
+            .expect("first recovery query should start after cooldown");
+
+        let waiting = {
+            let pacer = std::sync::Arc::clone(&pacer);
+            tokio::spawn(async move { pacer.wait_for_start().await })
+        };
+        tokio::task::yield_now().await;
+        assert!(!waiting.is_finished());
+        tokio::time::advance(QUERY_PACING_INITIAL_INTERVAL - Duration::from_millis(1)).await;
+        tokio::task::yield_now().await;
+        assert!(!waiting.is_finished());
+        tokio::time::advance(Duration::from_millis(1)).await;
+        waiting
+            .await
+            .expect("paced query task should complete")
+            .expect("paced query start should be admitted");
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn query_pacer_deduplicates_bursts_and_slows_new_episodes() {
+        let pacer = QueryStartPacer::default();
+        assert_eq!(
+            pacer.record_rate_limit(),
+            (QUERY_PACING_INITIAL_INTERVAL, true)
+        );
+        assert_eq!(
+            pacer.record_rate_limit(),
+            (QUERY_PACING_INITIAL_INTERVAL, false)
+        );
+
+        tokio::time::advance(Duration::from_secs(RATE_LIMIT_COOLDOWN_SECS)).await;
+        pacer
+            .wait_for_start()
+            .await
+            .expect("a duplicate signal must not extend the cooldown");
+
+        tokio::time::advance(QUERY_RATE_LIMIT_EPISODE_GAP).await;
+        assert_eq!(
+            pacer.record_rate_limit(),
+            (QUERY_PACING_INITIAL_INTERVAL * 2, true)
+        );
+        pacer.reset();
+        assert_eq!(pacer.interval(), None);
+        pacer
+            .wait_for_start()
+            .await
+            .expect("session reset should clear the cooldown");
+    }
+
+    #[tokio::test]
+    async fn query_limited_negentropy_falls_back_until_session_reset() {
+        let connection = permissive_connection("ws://127.0.0.1:1", Keys::generate());
+        assert!(connection.supports_negentropy().await);
+        assert!(connection.mark_negentropy_query_rate_limited());
+        assert!(!connection.mark_negentropy_query_rate_limited());
+        assert!(!connection.supports_negentropy().await);
+
+        connection.reset_subscription_budget(None);
+        assert!(connection.supports_negentropy().await);
+    }
+
+    #[test]
+    fn query_pacing_classifier_excludes_subscription_capacity_limits() {
+        assert!(is_query_rate_limit_message(
+            "rate-limited: too many queries"
+        ));
+        assert!(!is_query_rate_limit_message(
+            "rate-limited: too many concurrent REQs"
+        ));
+        assert!(!is_query_rate_limit_message(
+            "rate-limited: active subscriptions exceed max size 1048576 bytes"
+        ));
+    }
+
     #[test]
     fn test_normalize_url_real_world_example() {
         // Test the exact case from the bug report