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

Production startup against relay.ngit.dev exhausted its 120-query-per-minute token bucket, cooled down for 65 seconds, then replayed the entire recovered workload fast enough to exhaust the bucket again. This repeated every roughly 67 seconds even though the subscription ledger correctly bounded simultaneous open subscriptions: turnover, rather than concurrency, was the missing dimension.

Teach each connection session to recognise the specific too-many-queries refusal, unwind queued starts during the existing cooldown, and pace subsequent live REQs, transient REQs, exact-ID fetches, and negentropy round starts through one shared gate. Recovery begins at 100 starts per minute, leaves slack below the ngit-grasp 120/minute serving default, and doubles its interval only after a distinct later refusal, capped at ten seconds.

The first production candidate showed that rust-nostr can send several charged NEG-MSG frames inside one admitted NIP-77 round and caused a later refusal despite paced round starts. Because the application cannot pace those internal frames, any query-rate refusal now selects paced REQ fallback for the remainder of that connection session. Reconnect resets both lessons.

Correctness assumes the refusal identifies a per-connection query-start budget; NIP-11 has no standard query-rate field from which to learn proactively. Concurrent-REQ and subscription-byte refusals remain separate because pacing them would conceal different capacity problems. This does not change the serving-side limit or add configuration.

Validation: cargo test --lib (657 passed); cargo test --test sync (89 passed, 1 ignored), including the standalone startup request-concurrency scenario; nix build .#ngit-grasp on both substantive candidate designs; a real LocalRelay CLOSED scenario; and production evidence that identified the otherwise invisible SDK-managed NIP-77 traffic.

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