diff --git a/CHANGELOG.md b/CHANGELOG.md index a091502..530675b 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/explanation/grasp-02-proactive-sync.md b/docs/explanation/grasp-02-proactive-sync.md index c8fec8b..da7bc6f 100644 --- a/docs/explanation/grasp-02-proactive-sync.md +++ b/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` diff --git a/docs/explanation/monitoring.md b/docs/explanation/monitoring.md index 70954f4..0ad5df1 100644 --- a/docs/explanation/monitoring.md +++ b/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 diff --git a/docs/explanation/sync-scaling-constraints.md b/docs/explanation/sync-scaling-constraints.md index 62045e4..0163242 100644 --- a/docs/explanation/sync-scaling-constraints.md +++ b/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. --- diff --git a/src/sync/health.rs b/src/sync/health.rs index 7c47e40..46a5e7d 100644 --- a/src/sync/health.rs +++ b/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 diff --git a/src/sync/relay_connection.rs b/src/sync/relay_connection.rs index 14bc011..4446b4b 100644 --- a/src/sync/relay_connection.rs +++ b/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 { advertised.unwrap_or(FALLBACK_SUBSCRIPTION_BUDGET) } @@ -48,6 +66,91 @@ fn usable_subscription_slots(advertised: Option) -> 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, + last_start: Option, + last_limit_signal: Option, + paused_until: Option, +} + +/// 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, +} + +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 { + self.state + .lock() + .expect("query pacing state poisoned") + .interval + } +} + #[cfg(test)] fn historic_slot_allowance(advertised: Option, live_slots: usize) -> usize { usable_subscription_slots(advertised) @@ -309,6 +412,9 @@ pub struct RelayConnection { nip77_warning_logged: std::sync::Arc, /// Whether this relay supports NIP-77 negentropy (0 = unknown, 2 = confirmed not supported) nip77_supported: std::sync::Arc, + /// 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, /// Consecutive transient negentropy failures (drives the cooldown schedule) nip77_transient_failures: std::sync::Arc, /// 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, } 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) { 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