From b8e7dd41f029b1658cb2200268a6517601116f48 Mon Sep 17 00:00:00 2001 From: grumbach Date: Thu, 15 Oct 2026 18:08:23 +0900 Subject: [PATCH 1/2] feat(replication)!: answer a refused fetch with busy instead of silence A fetch responder admits two outstanding requests per requesting peer and twelve overall, and refuses the rest at admission without a reply. The requester then waits its whole 30 s fetch timeout, and saorsa-core charges the responder a connection-timeout trust event it did not earn. As measured for #252 from the responders' cumulative admission counters, this was 36.6% of inbound fetch requests on 0.21.0 and 44.3% on 0.22.0, and the WARN line counted 3.38M refusals in the 25 hours of the 2026-10-11 join wave. A requester now says it understands a busy reply, and a responder that refuses such a request answers `FetchResponse::Busy` at once, within bounds. Wire: `FetchRequest` keeps its serialized layout. A requester that understands busy replies appends a 5-byte capability trailer ("antf" and a flags byte) after the postcard message. Older nodes decode with `from_bytes`, which ignores it, so they refuse silently as before. A current node reads it with `take_from_bytes`. `FetchResponse::Busy` is appended after the existing variants, whose discriminants are unchanged, and is only sent in answer to a request with the trailer, so an older requester never receives it. Responder: a busy reply needs a free send permit (48), the per-peer allowance (two back to back, then one a second, at most two in flight) and the global one (32 back to back, then 16 a second; every message is signed, so about 5 KiB each). It is sent off the serial message loop, abandoned after 2 s and raced against shutdown. Anything outside those bounds is the old silence. The refusal WARN line keeps its text and counters and gains `capability` and `reply` fields; new cumulative counters report each busy-reply outcome. Requester: the fetch worker's requests ask for busy replies. A busy answer for the requested key carries no trust event. The attempt waits 5-10 s (jittered, never past its own fetch timeout) while holding its fetch slot, then ends as a source failure, so the queue moves on exactly as after a timeout. The busy holder is put last when choosing where to start or retry a key while its cooldown runs: the wait, doubling with each further busy answer up to 60 s. Serving once resets that doubling, and so do 10 minutes without a busy answer. Merge after #252: its per-source share removes most per-peer over-share, and the refusals that remain are what this answers. A test-only fetch serve gate, the same one #252 adds, makes the end-to-end tests deterministic. ADR-0025 records the decision. BREAKING CHANGE: `replication::protocol::FetchRequest` gains a public `busy_reply` field (not serialized) and `FetchResponse` gains a `Busy` variant, so code constructing the struct or matching the enum exhaustively must be updated. The wire format stays compatible with older nodes in both directions. --- docs/adr/ADR-0025-fetch-busy-reply.md | 209 ++++++++ docs/adr/README.md | 1 + src/replication/audit_metrics.rs | 121 +++++ src/replication/fetch_busy.rs | 745 ++++++++++++++++++++++++++ src/replication/mod.rs | 534 +++++++++++++++++- src/replication/protocol.rs | 380 ++++++++++++- src/replication/scheduling.rs | 21 +- tests/e2e/fetch_busy_ab.rs | 299 +++++++++++ tests/e2e/fetch_busy_reply.rs | 266 +++++++++ tests/e2e/mod.rs | 6 + tests/e2e/replication.rs | 15 +- 11 files changed, 2557 insertions(+), 40 deletions(-) create mode 100644 docs/adr/ADR-0025-fetch-busy-reply.md create mode 100644 src/replication/fetch_busy.rs create mode 100644 tests/e2e/fetch_busy_ab.rs create mode 100644 tests/e2e/fetch_busy_reply.rs diff --git a/docs/adr/ADR-0025-fetch-busy-reply.md b/docs/adr/ADR-0025-fetch-busy-reply.md new file mode 100644 index 00000000..091d97c8 --- /dev/null +++ b/docs/adr/ADR-0025-fetch-busy-reply.md @@ -0,0 +1,209 @@ +# ADR-0025: Tell a fetch requester "busy" instead of refusing silently + +- **Status:** Proposed +- **Date:** 2026-10-15 +- **Decision owners:** +- **Reviewers:** +- **Supersedes:** none +- **Superseded by:** none +- **Related:** [ADR-0005](ADR-0005-replication-repair-hardening.md) (the repair fetch pipeline), + [ADR-0017](ADR-0017-bounded-fresh-offers-and-copy-free-sends.md) (responder admission bounds), + ant-node #252 / ADR-0024 (requesters stay within the responder's per-peer share) + +## Context + +A replication fetch is a request/response exchange. The responder admits at most two +outstanding fetches per requesting peer (`FETCH_RESPONDER_MAX_OUTSTANDING_PER_PEER`) and twelve +overall (`FETCH_RESPONDER_MAX_OUTSTANDING`). A request past either limit is refused at admission: +the responder logs `Fetch response dropped at admission` and sends nothing back. + +Silence is expensive for both sides: + +- the requester waits its whole fetch timeout (30 s) before moving on to the next holder; +- saorsa-core records a `ConnectionTimeout` trust event against the responder, which did nothing + wrong; +- the requester learns nothing about why, so it keeps choosing that holder for new keys. + +On the production fleet this was the common case, not an edge. As measured for #252 from the +responders' own cumulative admission counters, 36.6% of inbound fetch requests on 0.21.0 and 44.3% +on 0.22.0 were refused at admission. In the 25 hours around a wave of 177 newly installed nodes +filling their ranges (2026-10-11 20:00 to 2026-10-12 21:00 UTC), the WARN line counted 3.38M +refusals, 95.2% of them to the new nodes. + +ant-node #252 removes most of the per-peer over-share by keeping a requester's own fetches within +each answering holder's share. It cannot remove every refusal: a request the requester has +already timed out on still holds its share at the responder until the responder sheds it, the +twelve-slot pool can be full of other peers' requests, and older requesters keep over-sharing +until they upgrade. Its ADR names a busy reply, which is a wire change, as the remaining piece. + +## Decision Drivers + +- Most peers run older versions for weeks after any release. An old node must decode the same + messages it always did and never be sent a `Busy` it cannot decode. +- A delivered `Busy` must not cost an honest responder trust; a refusal that cannot be answered + stays the old silence. +- The added reply traffic must be bounded: no request storm from requesters retrying on every busy + answer, and no echo amplification at responders. +- No new requester scheduling state that would collide with #252's per-source scheduler. + +## Considered Options + +1. **Keep silence; rely on #252 alone.** Leaves every remaining refusal at 30 s and a false + trust charge. +2. **Reply with the existing `FetchResponse::Error`.** Decodable by old requesters, but they + charge the responder an `ApplicationFailure` for it and fail over at once, so it trades a + timeout for an immediate hop storm. +3. **Add a field to `FetchRequest`.** Old responders would ignore a trailing field, but a current + responder would have to read old requests without it, which postcard reports as an error, not a + missing field. A hand-written tolerant deserializer can paper over that, but it cannot tell a + missing flag from a malformed one and only stays correct while `FetchRequest` is the last value + in every message that carries it. +4. **A capability trailer after the whole message, and an appended `FetchResponse::Busy` + variant (chosen).** +5. **Busy with requester-side retry rotation and budgets.** Reviewed and rejected: hopping on every + busy answer multiplies requests about tenfold in the worst case, competes with `tried` state, + and charging at budget expiry punishes honest overloaded holders once per key per cycle. + +## Decision + +We will let a requester say it understands a busy reply, and have a responder that refuses such a +request at admission say `Busy` at once, within bounds, instead of staying silent. + +**Wire.** `FetchRequest` keeps its serialized layout (`key`). A requester that understands busy +replies appends the 5-byte fetch capability trailer, `antf` and a flags byte with bit 0 set, +immediately after the postcard message. `ReplicationMessage::decode` uses `take_from_bytes` and +sets `FetchRequest::busy_reply` (not serialized) only when that trailer follows a fetch request; +bytes after the flags byte are reserved and ignored. Older nodes decode with `from_bytes`, which +ignores everything after the message, so they read the key and refuse silently as before. +`FetchResponse::Busy { key }` is appended after `Success`, `NotFound` and `Error`, whose +discriminants are unchanged, and is only ever sent in answer to a request carrying the trailer. + +**Responder.** Every saorsa-core message is signed with ML-DSA-65, so a busy reply costs a +signature and about 5 KiB on the wire, not a few bytes. On an admission refusal of a request that +asked, the responder takes a busy-reply send permit (48 at once), then a grant from +`FetchBusyReplyLimiter`: per peer, two back to back and then one a second, with at most two in +flight; overall, 32 back to back and then 16 a second (at most 48 replies, about 250 KiB, in any +one second; about 85 KiB/s sustained). Both rate allowances are tested and spent together under +one lock, a denial spends neither, and the per-peer table is capped at 4096 entries, forgetting +only peers whose allowance has refilled and that have nothing in flight. Within all bounds it +sends `Busy` on the detached task tracker, abandons the send after 2 s and races shutdown; +otherwise it refuses silently, exactly as before. The `Fetch response dropped at admission` +WARN line and its counters are unchanged, with `capability=asked|not_asked` and +`reply=scheduled|none` fields added. + +**Requester.** Every fetch request this node's fetch worker sends asks for busy replies. A `Busy` for the requested key carries +no trust event. The attempt then waits 5-10 s (uniform jitter), never past its own fetch timeout, +still holding its fetch slot, and ends as `SourceFailed`, so the queue moves on to the next untried +holder exactly as after a timeout. `SourceBusyTracker` remembers the busy holder: a new key starts, +and a failed key retries, on the first holder in preference order that is not cooling down, +falling back to the first when all are. It reorders and never withholds, so it lowers how often a +busy holder is asked but is not a hard per-holder bound; the hard bound is the busy wait per fetch +slot. The cooldown is the full jittered wait, doubling with each further busy answer before the +holder next serves this requester, up to 60 s, and is never shortened while it runs; once it runs +out the holder is no longer put last. Serving once clears the holder's record, so its next busy +answer starts the doubling again; a record with no busy answer for 10 minutes does the same. When +the 4096-entry table is full, a holder that is no longer cooling is forgotten first. A `Busy` for any other key is +treated as malformed, like every other key mismatch. + +**Order.** This is merged after #252. #252's per-source share is what removes most per-peer +over-share and bounds how many fetches one requester has open at one holder; without it, a +requester with many fetch slots still over-shares, and most of those refusals exceed the per-peer +busy allowance and stay silent. On rebase, #252's source choice also skips cooling holders. + +## Consequences + +### Positive + +- When a refused requester receives `Busy`, it moves on after 5-10 s instead of 30 s without + charging the responder. +- A holder that just refused is not handed this requester's next keys while another holder that is + not cooling exists. +- Old nodes decode exactly the messages they always did: a request carries 5 bytes after the + message that they never read, and they are never sent `Busy`. Rollback is a release without the + change. +- With #252, each busy wait keeps one attempt counted against that holder's two-request share for + its duration, reducing new requests and stopping them when both share slots are occupied; the + busy answer also counts as an answer for #252's silent-source logic. + +### Negative / Trade-offs + +- New outbound traffic at responders: at most 16 signed busy replies a second sustained, about + 85 KiB/s, and 48 (about 250 KiB) in any one second, plus that many signatures. Today a refusal + sends nothing. Each refused request already cost its sender the same signed message. +- A requester asks a refusing holder up to once per 5 s per fetch slot instead of once per 30 s: + at most `C / 5 s` small requests with `C` hardware-thread fetch slots, against `C / 30 s` today, + in the case where every holder refuses everything. Capping `C` is a separate decision. +- Refusals beyond the per-peer busy allowance stay silent and are still charged by saorsa-core. + Without #252, a requester that over-shares by more than two requests at once still meets some + silence; the simulation below gives the residual. +- Busy is trust-neutral. A holder that tries to answer busy to everything can still be charged when + a request did not ask, a reply was suppressed, or a scheduled send did not complete; otherwise it + is passed over for new keys for up to 60 s at a time and remains subject to storage audits. + Penalising "never serves" would need a service audit with reserved capacity, which is not part + of this decision. +- Sixteen identities each spending their per-peer second can use up the global allowance and + deny busy replies to everyone else, and stalled peers can hold the 48 send permits for up to + 2 s each. What they win is the old silence for those peers, not a new failure. The global + allowance is kept small on purpose, because it is what bounds the signed bytes a full node + spends on refusals. + +### Neutral / Operational + +- New counters on the `replication responder admission summary (cumulative)` line: + `fetch_busy_reply_{not_asked,send_pool_full,rate_limited,peer_in_flight,sent,send_failed,timed_out,cancelled}`, + `fetch_busy_received`, `fetch_busy_cooling_skips` and `fetch_busy_sources_tracked`. + `not_asked` against the rest is the share of refused requests that did not ask for a busy reply: + a proxy, and an upper bound, for how much refused traffic still comes from older versions (a + request from a current node can also not ask, through the API or a malformed trailer). + +## Validation + +- Unit tests: + - mirrors of the previous types decode current requests, and the unchanged answers 0-2; + - a current node reads an older request as not asking; + - trailer parsing: exact, wrong magic, truncated, unknown bits, reserved bytes, padding; + - an older requester cannot decode `Busy`; + - limiter: burst, refill, global cap, no spend on a denial, in-flight limit, bounded table; + - the busy wait's range and its cap at the fetch timeout; + - tracker: cooldown, escalation, cap, never shortened, reset, expiry, bounded table; + - a retry passes over a cooling holder but still takes it last; + - a busy reply is not counted as served bytes. +- End to end on a local network, made deterministic with a test-only fetch serve gate (the same + one #252 adds): + - a refused request that asks is told `Busy` at once; + - one that does not ask is refused silently; + - the held requests are served once the gate opens; + - the requester's own fetch worker, refused by a full holder, records the holder as cooling, + keeps the attempt for at least the minimum busy wait, then gives the key up, and the holder's + trust at the requester never drops. +- Measured on real nodes against upstream main with the same test-only gate, in + `tests/e2e/fetch_busy_ab.rs` (opt-in): + - 20 requesters each within their two-request share, the first of five holders with its + 12-slot pool full for 2 s (28 refusals in both builds), which is the refusal left after #252: + all 40 chunks in 10.5 s, no holder charged, against 60.4 s with 14 requesters charging the + first holder and 4 the second; + - one requester with 8 fetch slots over-asking three holders, which #252 removes: 39.7 s, 39.4 s + and 36.7 s with only the first holder charged (4 charges), against 60.2 s, 90.3 s and 90.2 s with + up to all three charged. +- A discrete-event simulation of the 177-node join wave and 24 more scenarios (dead holders, + holders that always answer busy, saturated pools, mixed versions, 1-64 fetch slots, + single-holder keys, limiter and wait sensitivity), comparing today, this change, #252-style + capping, and both. It reproduces the real-node run of this change and overstates the cost of + today's silence, so it gives directions, not numbers. In it no key is stranded in any scenario. + With nobody answering busy it behaves exactly as today. With #252-style capping, the busy reply + removes every false charge in the background-load, overload, always-busy and dense scenarios; + what remains elsewhere comes from requests a slow holder sheds unanswered after the fetch timeout + and from requesters that do not ask. Without capping it raises requests per chunk + by up to 12% in some scenarios and lowers it by up to 17% in others (with a 1-2 s wait instead of +5-10 s it rises 16%, which is why the wait is longer), and keys with a single holder go back to +verification about twice as often. +- In production: + - `fetch_busy_reply_sent` against `fetch_admission_dropped_*`; + - the fall in `connection_timeout` trust events between fleet nodes; + - `fetch_busy_reply_rate_limited` and `send_pool_full` staying small. + + Revisit the 5-10 s wait and the allowances against measured admission hold times. + +## Notes for AI-assisted work + +AI tools may help draft this ADR, but **must not mark it Accepted without human review**. Accepted ADRs are immutable: create a new superseding ADR rather than editing an Accepted ADR. diff --git a/docs/adr/README.md b/docs/adr/README.md index b58e9a32..092768d4 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -40,3 +40,4 @@ See [`TOOLING.md`](./TOOLING.md) for `adrs`, `adr-kit`, and AI harness setup. - [ADR-0018: Pointer ownership transfer by final redirection](./ADR-0018-pointer-transfer-by-final-redirection.md) - [ADR-0019: Pointer audits serve the record round 1 bound](./ADR-0019-pointer-audits-serve-the-record-round-one-bound.md) - [ADR-0022: Remove the LMDB Chunk Store](./ADR-0022-remove-the-lmdb-chunk-store.md) +- [ADR-0025: Tell a fetch requester "busy" instead of refusing silently](./ADR-0025-fetch-busy-reply.md) diff --git a/src/replication/audit_metrics.rs b/src/replication/audit_metrics.rs index de8f77a1..0989d5b2 100644 --- a/src/replication/audit_metrics.rs +++ b/src/replication/audit_metrics.rs @@ -400,6 +400,21 @@ static RESPONDER_ADMISSION_DROPS: [AtomicU64; N_RESPONDER_CLASSES * N_ADMISSION_ static RESPONDER_STALENESS_SHEDS: [AtomicU64; N_RESPONDER_CLASSES] = [const { AtomicU64::new(0) }; N_RESPONDER_CLASSES]; +/// Fetch admission refusals by what became of the busy reply, indexed by +/// [`FetchBusyReplyOutcome::index`]. +static FETCH_BUSY_REPLIES: [AtomicU64; FetchBusyReplyOutcome::COUNT] = + [const { AtomicU64::new(0) }; FetchBusyReplyOutcome::COUNT]; + +/// Busy answers this node received to its own fetches. +static FETCH_BUSY_RECEIVED: AtomicU64 = AtomicU64::new(0); + +/// Times this node, choosing where to start or retry a fetch, passed over a +/// holder that had recently answered it busy. +static FETCH_BUSY_COOLING_SKIPS: AtomicU64 = AtomicU64::new(0); + +/// Holders this node currently remembers as having answered it busy. +static FETCH_BUSY_SOURCES_TRACKED: AtomicU64 = AtomicU64::new(0); + static SERIAL_QUEUE_OVERFLOW_DROPS: [AtomicU64; protocol::N_REPLICATION_VARIANTS] = [const { AtomicU64::new(0) }; protocol::N_REPLICATION_VARIANTS]; @@ -532,6 +547,98 @@ pub fn record_responder_staleness_shed(class: ReplicationResponderClass) { } } +/// What became of a fetch refused at admission, as far as a busy reply goes. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum FetchBusyReplyOutcome { + /// The requester did not ask for busy replies (an older node): silence, + /// as before. + NotAsked, + /// Every busy-reply send permit was in use: silence. + SendPoolFull, + /// Over the per-peer or global busy-reply rate: silence. + RateLimited, + /// A busy reply to the same peer was still being sent: silence. + PeerInFlight, + /// The busy reply was handed to the transport. + Sent, + /// The transport refused the busy reply. + SendFailed, + /// The transport did not take the busy reply in time; abandoned. + TimedOut, + /// Shutdown began before the busy reply was sent. + Cancelled, +} + +impl FetchBusyReplyOutcome { + const COUNT: usize = 8; + + const fn index(self) -> usize { + match self { + Self::NotAsked => 0, + Self::SendPoolFull => 1, + Self::RateLimited => 2, + Self::PeerInFlight => 3, + Self::Sent => 4, + Self::SendFailed => 5, + Self::TimedOut => 6, + Self::Cancelled => 7, + } + } +} + +/// Count what became of one fetch refusal's busy reply. +pub fn record_fetch_busy_reply(outcome: FetchBusyReplyOutcome) { + if let Some(counter) = FETCH_BUSY_REPLIES.get(outcome.index()) { + counter.fetch_add(1, Ordering::Relaxed); + } +} + +/// Fetch refusals with this busy-reply outcome since process start. +#[must_use] +#[cfg_attr(not(feature = "logging"), allow(dead_code))] +pub fn fetch_busy_replies_total(outcome: FetchBusyReplyOutcome) -> u64 { + FETCH_BUSY_REPLIES + .get(outcome.index()) + .map_or(0, |counter| counter.load(Ordering::Relaxed)) +} + +/// Count a busy answer to one of this node's own fetches. +pub fn record_fetch_busy_received() { + FETCH_BUSY_RECEIVED.fetch_add(1, Ordering::Relaxed); +} + +/// Busy answers to this node's own fetches since process start. +#[must_use] +#[cfg_attr(not(feature = "logging"), allow(dead_code))] +pub fn fetch_busy_received_total() -> u64 { + FETCH_BUSY_RECEIVED.load(Ordering::Relaxed) +} + +/// Count a fetch started or retried on a later holder because an earlier one +/// had recently answered busy. +pub fn record_fetch_busy_cooling_skip() { + FETCH_BUSY_COOLING_SKIPS.fetch_add(1, Ordering::Relaxed); +} + +/// Fetches started or retried past a cooling holder since process start. +#[must_use] +#[cfg_attr(not(feature = "logging"), allow(dead_code))] +pub fn fetch_busy_cooling_skips_total() -> u64 { + FETCH_BUSY_COOLING_SKIPS.load(Ordering::Relaxed) +} + +/// Set how many holders this node remembers as having answered it busy. +pub fn set_fetch_busy_sources_tracked(count: usize) { + FETCH_BUSY_SOURCES_TRACKED.store(u64::try_from(count).unwrap_or(u64::MAX), Ordering::Relaxed); +} + +/// Holders this node remembers as having answered it busy. +#[must_use] +#[cfg_attr(not(feature = "logging"), allow(dead_code))] +pub fn fetch_busy_sources_tracked() -> u64 { + FETCH_BUSY_SOURCES_TRACKED.load(Ordering::Relaxed) +} + /// One class's admission pressure, read together for a summary line. #[cfg_attr(not(feature = "logging"), allow(dead_code))] struct ResponderAdmissionSnapshot { @@ -640,6 +747,20 @@ pub fn log_responder_admission_summary() { fetch_admission_dropped_global_pool = fetch.global_pool, fetch_admission_dropped_per_peer_share = fetch.per_peer_share, fetch_staleness_shed = fetch.staleness_shed, + fetch_busy_reply_not_asked = fetch_busy_replies_total(FetchBusyReplyOutcome::NotAsked), + fetch_busy_reply_send_pool_full = + fetch_busy_replies_total(FetchBusyReplyOutcome::SendPoolFull), + fetch_busy_reply_rate_limited = + fetch_busy_replies_total(FetchBusyReplyOutcome::RateLimited), + fetch_busy_reply_peer_in_flight = + fetch_busy_replies_total(FetchBusyReplyOutcome::PeerInFlight), + fetch_busy_reply_sent = fetch_busy_replies_total(FetchBusyReplyOutcome::Sent), + fetch_busy_reply_send_failed = fetch_busy_replies_total(FetchBusyReplyOutcome::SendFailed), + fetch_busy_reply_timed_out = fetch_busy_replies_total(FetchBusyReplyOutcome::TimedOut), + fetch_busy_reply_cancelled = fetch_busy_replies_total(FetchBusyReplyOutcome::Cancelled), + fetch_busy_received = fetch_busy_received_total(), + fetch_busy_cooling_skips = fetch_busy_cooling_skips_total(), + fetch_busy_sources_tracked = fetch_busy_sources_tracked(), verification_admission_dropped_global_pool = verification.global_pool, verification_admission_dropped_per_peer_share = verification.per_peer_share, verification_staleness_shed = verification.staleness_shed, diff --git a/src/replication/fetch_busy.rs b/src/replication/fetch_busy.rs new file mode 100644 index 00000000..e6be386f --- /dev/null +++ b/src/replication/fetch_busy.rs @@ -0,0 +1,745 @@ +//! The fetch busy reply: what each side does when a responder cannot admit a +//! fetch. +//! +//! A responder that refuses a fetch at admission used to stay silent. The +//! requester then waited its whole fetch timeout, and saorsa-core charged the +//! responder a connection timeout it had not earned. A requester that says it +//! understands [`FetchResponse::Busy`] is now told at once instead. +//! +//! Two bounds keep the new reply from costing more than the silence it +//! replaces: +//! +//! - **Responder:** a busy reply is a send the node never made before, caused +//! by a request it is refusing, and like every message it is signed, so it +//! costs a signature and about 5 KiB on the wire. [`FetchBusyReplyLimiter`] +//! caps those sends per peer and overall, and how many to one peer may be in +//! flight at once. +//! Past any cap a refusal is silent again, which is exactly what every +//! refusal was before. +//! - **Requester:** a busy answer is not a reason to ask again at once. +//! [`busy_wait`] keeps the attempt, and so its fetch slot, for a few seconds +//! before the key moves on, so a requester asks at most once per wait per +//! slot instead of as fast as answers come back. The attempt then ends as a +//! source failure, exactly as the silent timeout did, minus the trust charge. +//! [`SourceBusyTracker`] also puts a busy source last when choosing where to +//! start or retry a key, for longer each time it stays busy, so a requester +//! does not keep picking a holder that just said it cannot take more. +//! +//! [`FetchResponse::Busy`]: crate::replication::protocol::FetchResponse::Busy + +use std::collections::HashMap; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use parking_lot::Mutex; +use saorsa_core::identity::PeerId; + +use super::audit_metrics; + +/// Shortest time a requester waits after a busy answer before moving on. +/// +/// A responder refuses because its admitted fetches, each up to a 4 MiB chunk +/// sent over a home upload link, are still being served. A share frees in +/// seconds, not milliseconds, so asking again sooner would mostly be refused +/// again. +pub const FETCH_BUSY_WAIT_MIN: Duration = Duration::from_secs(5); + +/// Longest time a requester waits after a busy answer before moving on. +/// +/// Waits are spread uniformly between the two bounds, so requesters refused +/// together do not all come back together. +pub const FETCH_BUSY_WAIT_MAX: Duration = Duration::from_secs(10); + +/// The jitter a requester adds to [`FETCH_BUSY_WAIT_MIN`]: drawn uniformly from +/// zero to this, inclusive. +pub const FETCH_BUSY_WAIT_SPREAD: Duration = + FETCH_BUSY_WAIT_MAX.saturating_sub(FETCH_BUSY_WAIT_MIN); + +/// Longest a responder spends handing one busy reply to the transport. +/// +/// A reply that cannot be handed over in this long is abandoned: the requester +/// then gets the silence it would have had anyway. Without it, a peer that is +/// slow to accept data could hold a send permit indefinitely. +pub const FETCH_BUSY_REPLY_SEND_TIMEOUT: Duration = Duration::from_secs(2); + +/// Busy replies one peer may receive back to back, and the most to it that +/// may be in flight at once. +/// +/// The same as the per-peer fetch admission share, so a requester that sent +/// one share too many at once hears about each refusal. +const PEER_BURST: u32 = 2; + +/// Sustained rate of busy replies to one peer: one per this interval. +/// +/// A requester that waits [`FETCH_BUSY_WAIT_MIN`] after each busy answer never +/// comes near it. A peer that ignores the answers and keeps sending gets one +/// reply a second, and silence for the rest. +const PEER_INTERVAL: Duration = Duration::from_secs(1); + +/// Busy replies this node may send back to back across all peers. +/// +/// Every message is signed, which adds about 5 KiB to the wire and a +/// signature's worth of CPU, so the global allowance is what bounds the cost. +/// Burst and rate together allow at most 48 replies, about 250 KiB, in any one +/// second, and 16 a second, about 85 KiB/s, sustained. The burst lets a node +/// that becomes full answer a whole pool's worth of peers at once. +const GLOBAL_BURST: u32 = 32; + +/// Sustained rate of busy replies across all peers: one per this interval, +/// 16 a second. +const GLOBAL_INTERVAL: Duration = Duration::from_micros(62_500); + +/// Most peers whose busy-reply allowance is tracked at once. +/// +/// A peer whose allowance has fully refilled and has no reply in flight +/// carries no information and is dropped first. If every tracked peer is still +/// being limited, a new peer gets silence rather than an entry, so the table +/// cannot grow without bound. +const MAX_TRACKED_PEERS: usize = 4096; + +/// Longest a source that keeps answering busy is put last. +const MAX_SOURCE_COOLDOWN: Duration = Duration::from_secs(60); + +/// Busy answers after which a source's cooldown stops doubling: the wait, +/// then two, four and eight times it, capped at [`MAX_SOURCE_COOLDOWN`]. +const MAX_COOLDOWN_DOUBLINGS: u32 = 3; + +/// How long a source's busy record is kept after its last busy answer. Long +/// enough to remember a holder that stays busy across many cooldowns; short +/// enough that one busy spell is not held against it for good. +const SOURCE_RECORD_TTL: Duration = Duration::from_secs(600); + +/// Most sources whose busy record is kept at once. +const MAX_TRACKED_SOURCES: usize = 4096; + +/// How long to wait after a busy answer, given `remaining`, the time left +/// before the attempt's fetch timeout would have expired. +/// +/// The wait is [`FETCH_BUSY_WAIT_MIN`] plus `jitter`, which the caller draws +/// from `0..=FETCH_BUSY_WAIT_SPREAD` and which is clamped to that range. It +/// never outlasts `remaining`, so a busy answer never leaves a key waiting +/// longer than the silent refusal it replaces. (A fetch timeout shorter than +/// the minimum wait, which only tests configure, shortens the wait with it.) +#[must_use] +pub fn busy_wait(remaining: Duration, jitter: Duration) -> Duration { + FETCH_BUSY_WAIT_MIN + .saturating_add(jitter.min(FETCH_BUSY_WAIT_SPREAD)) + .min(remaining) +} + +/// A generic cell rate limiter: `burst` events back to back, then one per +/// `interval`. +/// +/// It keeps only the theoretical arrival time of the next event, so it needs no +/// floating point and no refill timer. +#[derive(Debug, Default, Clone, Copy)] +struct Gcra { + /// When the allowance is next fully refilled; `None` until first used. + tat: Option, +} + +impl Gcra { + /// The arrival time a further event would move this limiter to, if it is + /// allowed now. + fn admit(self, now: Instant, interval: Duration, burst: u32) -> Option { + let tat = self.tat.map_or(now, |tat| tat.max(now)); + let next = tat.checked_add(interval)?; + (next.saturating_duration_since(now) <= interval.saturating_mul(burst)).then_some(next) + } + + /// Whether the allowance has fully refilled. + fn is_idle(self, now: Instant) -> bool { + self.tat.is_none_or(|tat| tat <= now) + } +} + +/// One peer's busy-reply allowance. +#[derive(Debug, Default, Clone, Copy)] +struct PeerAllowance { + rate: Gcra, + /// Busy replies to this peer being sent now. + in_flight: u32, +} + +impl PeerAllowance { + /// Forgetting it loses nothing: its allowance is full and nothing is in + /// flight. + fn is_idle(self, now: Instant) -> bool { + self.in_flight == 0 && self.rate.is_idle(now) + } +} + +#[derive(Debug, Default)] +struct LimiterState { + global: Gcra, + peers: HashMap, +} + +/// Why a busy reply was not allowed. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BusyReplyDenied { + /// Over the per-peer or global rate, or the table of peers is full. + RateLimited, + /// The most busy replies to this peer that may be in flight are still + /// being sent. + PeerInFlight, +} + +/// Bounds how many busy replies a responder sends: per peer and overall, by +/// rate, and how many to one peer may be in flight at once. +/// +/// Synchronous and short, so it can be asked on the serial message loop. +#[derive(Debug, Default)] +pub struct FetchBusyReplyLimiter { + state: Mutex, +} + +impl FetchBusyReplyLimiter { + /// A limiter with every allowance full. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Allow one busy reply to `peer` at `now`, if its allowance, the global + /// one and its in-flight limit all permit. Both rate allowances are spent + /// together on a grant and neither on a denial. The returned grant holds + /// one of the peer's in-flight places until it is dropped, which the caller + /// does when the send ends. + /// + /// # Errors + /// + /// Returns why the reply is not allowed. + pub fn try_acquire( + self: &Arc, + peer: &PeerId, + now: Instant, + ) -> Result { + let mut state = self.state.lock(); + if !state.peers.contains_key(peer) && state.peers.len() >= MAX_TRACKED_PEERS { + state.peers.retain(|_, allowance| !allowance.is_idle(now)); + if state.peers.len() >= MAX_TRACKED_PEERS { + return Err(BusyReplyDenied::RateLimited); + } + } + let allowance = state.peers.get(peer).copied().unwrap_or_default(); + if allowance.in_flight >= PEER_BURST { + return Err(BusyReplyDenied::PeerInFlight); + } + let peer_next = allowance + .rate + .admit(now, PEER_INTERVAL, PEER_BURST) + .ok_or(BusyReplyDenied::RateLimited)?; + let global_next = state + .global + .admit(now, GLOBAL_INTERVAL, GLOBAL_BURST) + .ok_or(BusyReplyDenied::RateLimited)?; + state.global.tat = Some(global_next); + state.peers.insert( + *peer, + PeerAllowance { + rate: Gcra { + tat: Some(peer_next), + }, + in_flight: allowance.in_flight.saturating_add(1), + }, + ); + Ok(BusyReplyGrant { + limiter: Arc::clone(self), + peer: *peer, + }) + } + + fn release(&self, peer: &PeerId) { + if let Some(allowance) = self.state.lock().peers.get_mut(peer) { + allowance.in_flight = allowance.in_flight.saturating_sub(1); + } + } + + /// How many peers are tracked now. + #[cfg(test)] + fn tracked_peers(&self) -> usize { + self.state.lock().peers.len() + } +} + +/// A granted busy reply. Holds one of the peer's in-flight places until +/// dropped. +#[derive(Debug)] +pub struct BusyReplyGrant { + limiter: Arc, + peer: PeerId, +} + +impl Drop for BusyReplyGrant { + fn drop(&mut self) { + self.limiter.release(&self.peer); + } +} + +/// One source's recent busy answers to this requester. +#[derive(Debug, Clone, Copy)] +struct SourceBusy { + /// Busy answers since it last served this requester. + strikes: u32, + /// Until when it is put last. + cooling_until: Instant, + /// When it last answered busy. + last_busy: Instant, +} + +impl SourceBusy { + fn is_cooling(self, now: Instant) -> bool { + now < self.cooling_until + } + + fn expired(self, now: Instant) -> bool { + !self.is_cooling(now) && now.saturating_duration_since(self.last_busy) >= SOURCE_RECORD_TTL + } +} + +/// Which sources recently answered this requester busy, and until when they +/// should be asked last. +/// +/// A busy answer is not a fault, so nothing here touches trust. It is local +/// scheduling: a holder that just refused is the worst place to send the next +/// fetch, and one that keeps refusing is put last for longer each time. +/// Serving once clears its record. It orders sources and never withholds one: +/// a key whose every holder is cooling still starts on the first of them, and +/// then waits out that holder's busy answer like any other. So per requester +/// it lowers how often a busy holder is asked; the hard bound is the busy wait +/// per fetch slot. Synchronous and short, so it is safe to consult while +/// holding the fetch queues lock. +#[derive(Debug, Default)] +pub struct SourceBusyTracker { + sources: Mutex>, +} + +impl SourceBusyTracker { + /// A tracker that knows no busy source. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Record a busy answer from `source` at `now`. `wait` is the busy wait + /// before any cap from the attempt's own timeout. Returns how long the + /// source is now put last: `wait` the first time, doubling with each further + /// busy answer before it serves again, up to a minute. A + /// cooldown already running is never shortened. + pub fn record_busy(&self, source: &PeerId, now: Instant, wait: Duration) -> Duration { + let mut sources = self.sources.lock(); + if sources + .get(source) + .is_some_and(|record| record.expired(now)) + { + sources.remove(source); + } + if !sources.contains_key(source) && sources.len() >= MAX_TRACKED_SOURCES { + sources.retain(|_, record| !record.expired(now)); + if sources.len() >= MAX_TRACKED_SOURCES { + // Prefer to forget a source that is not cooling. Forgetting + // one that is ends its cooldown early; the busy wait still + // bounds how often it is asked. + let victim = sources + .iter() + .min_by_key(|(_, record)| (record.is_cooling(now), record.cooling_until)) + .map(|(peer, _)| *peer); + if let Some(peer) = victim { + sources.remove(&peer); + } + } + } + let record = sources.entry(*source).or_insert(SourceBusy { + strikes: 0, + cooling_until: now, + last_busy: now, + }); + record.strikes = record.strikes.saturating_add(1); + let doublings = record.strikes.saturating_sub(1).min(MAX_COOLDOWN_DOUBLINGS); + let cooldown = wait + .saturating_mul(1u32 << doublings) + .min(MAX_SOURCE_COOLDOWN); + let until = now.checked_add(cooldown).unwrap_or(now); + record.cooling_until = record.cooling_until.max(until); + record.last_busy = now; + audit_metrics::set_fetch_busy_sources_tracked(sources.len()); + cooldown + } + + /// `source` served this requester: forget its busy record. + pub fn record_served(&self, source: &PeerId) { + let mut sources = self.sources.lock(); + if sources.remove(source).is_some() { + audit_metrics::set_fetch_busy_sources_tracked(sources.len()); + } + } + + /// Whether `source` should be asked last at `now`. + #[must_use] + pub fn is_cooling(&self, source: &PeerId, now: Instant) -> bool { + self.sources + .lock() + .get(source) + .is_some_and(|record| record.is_cooling(now)) + } + + /// Of `candidates`, in their order of preference, the first that is not + /// cooling at `now`, or the first of them when all are. Counts a skip when + /// that passes over the first candidate. + pub fn choose<'a, I>(&self, candidates: I, now: Instant) -> Option<&'a PeerId> + where + I: IntoIterator, + { + let tracked = self.sources.lock(); + let mut first = None; + for candidate in candidates { + let first_seen = *first.get_or_insert(candidate); + if tracked + .get(candidate) + .is_none_or(|record| !record.is_cooling(now)) + { + if !std::ptr::eq(first_seen, candidate) { + audit_metrics::record_fetch_busy_cooling_skip(); + } + return Some(candidate); + } + } + first + } + + /// When `source` last answered busy, if it is still remembered. + #[must_use] + pub fn last_busy(&self, source: &PeerId) -> Option { + self.sources + .lock() + .get(source) + .map(|record| record.last_busy) + } + + /// How many sources are tracked now. + #[must_use] + pub fn len(&self) -> usize { + self.sources.lock().len() + } + + /// Whether no source is tracked. + #[must_use] + pub fn is_empty(&self) -> bool { + self.sources.lock().is_empty() + } +} + +#[cfg(test)] +#[allow(clippy::unwrap_used, clippy::expect_used)] +mod tests { + use super::*; + + fn peer(i: u32) -> PeerId { + let mut bytes = [0u8; 32]; + for (byte, value) in bytes.iter_mut().zip(i.to_le_bytes()) { + *byte = value; + } + PeerId::from_bytes(bytes) + } + + fn limiter() -> Arc { + Arc::new(FetchBusyReplyLimiter::new()) + } + + /// Grant and immediately end the send, as a reply that went out at once. + fn grant(limiter: &Arc, p: &PeerId, now: Instant) -> bool { + limiter.try_acquire(p, now).is_ok() + } + + #[test] + fn busy_wait_spans_the_jitter_range() { + let remaining = Duration::from_secs(30); + assert_eq!(busy_wait(remaining, Duration::ZERO), FETCH_BUSY_WAIT_MIN); + assert_eq!( + busy_wait(remaining, FETCH_BUSY_WAIT_SPREAD), + FETCH_BUSY_WAIT_MAX + ); + assert_eq!( + busy_wait(remaining, Duration::from_secs(2)), + FETCH_BUSY_WAIT_MIN + Duration::from_secs(2) + ); + // Jitter past the spread is clamped, not added. + assert_eq!( + busy_wait(remaining, Duration::from_secs(3600)), + FETCH_BUSY_WAIT_MAX + ); + } + + #[test] + fn busy_wait_never_outlasts_the_fetch_timeout() { + assert_eq!( + busy_wait(Duration::from_secs(3), FETCH_BUSY_WAIT_SPREAD), + Duration::from_secs(3) + ); + assert_eq!( + busy_wait(Duration::ZERO, FETCH_BUSY_WAIT_SPREAD), + Duration::ZERO + ); + } + + #[test] + fn a_peer_gets_its_burst_then_one_reply_per_interval() { + let limiter = limiter(); + let p = peer(1); + let t0 = Instant::now(); + for _ in 0..PEER_BURST { + assert!(grant(&limiter, &p, t0)); + } + assert_eq!( + limiter.try_acquire(&p, t0).err(), + Some(BusyReplyDenied::RateLimited) + ); + assert!(!grant(&limiter, &p, t0 + PEER_INTERVAL / 2)); + assert!(grant(&limiter, &p, t0 + PEER_INTERVAL)); + assert!(!grant(&limiter, &p, t0 + PEER_INTERVAL)); + // A long quiet spell refills only up to the burst, never beyond it. + let later = t0 + PEER_INTERVAL * 100; + for _ in 0..PEER_BURST { + assert!(grant(&limiter, &p, later)); + } + assert!(!grant(&limiter, &p, later)); + } + + #[test] + fn at_most_a_burst_in_flight_per_peer() { + let limiter = limiter(); + let p = peer(1); + let t0 = Instant::now(); + let held: Vec = (0..PEER_BURST) + .map(|_| limiter.try_acquire(&p, t0).expect("within the burst")) + .collect(); + // Long enough later that the rate allows more, but the sends have not + // ended: the in-flight limit refuses. + let later = t0 + PEER_INTERVAL * 10; + assert_eq!( + limiter.try_acquire(&p, later).err(), + Some(BusyReplyDenied::PeerInFlight) + ); + // Another peer is not affected. + assert!(grant(&limiter, &peer(2), later)); + drop(held); + assert!(grant(&limiter, &p, later)); + } + + #[test] + fn one_peer_exhausting_its_allowance_leaves_others_theirs() { + let limiter = limiter(); + let t0 = Instant::now(); + for _ in 0..10 { + let _ = grant(&limiter, &peer(1), t0); + } + assert!(grant(&limiter, &peer(2), t0)); + } + + #[test] + fn the_global_allowance_caps_every_peer_together() { + let limiter = limiter(); + let t0 = Instant::now(); + let granted = (0..1_000u32) + .filter(|i| grant(&limiter, &peer(*i), t0)) + .count(); + assert_eq!(granted, usize::try_from(GLOBAL_BURST).unwrap()); + // Refills at the global rate: one second later, one second's worth. + let t1 = t0 + Duration::from_secs(1); + let granted = (1_000..2_000u32) + .filter(|i| grant(&limiter, &peer(*i), t1)) + .count(); + let per_second = Duration::from_secs(1).as_nanos() / GLOBAL_INTERVAL.as_nanos(); + assert_eq!(granted, usize::try_from(per_second).unwrap()); + } + + #[test] + fn a_global_refusal_spends_nothing_from_the_peer() { + let limiter = limiter(); + let t0 = Instant::now(); + for i in 0..GLOBAL_BURST { + assert!(grant(&limiter, &peer(i), t0)); + } + let p = peer(100_000); + assert!(!grant(&limiter, &p, t0)); + assert_eq!( + limiter.tracked_peers(), + usize::try_from(GLOBAL_BURST).unwrap() + ); + // Once the global allowance refills, the peer still has its whole burst. + let t1 = t0 + Duration::from_secs(1); + for _ in 0..PEER_BURST { + assert!(grant(&limiter, &p, t1)); + } + } + + #[test] + fn the_peer_table_stays_bounded_and_forgets_only_idle_peers() { + let limiter = limiter(); + let mut now = Instant::now(); + let mut next_peer = 0u32; + // Fill the table with replies still in flight, pacing to stay inside + // the global allowance. + let mut held = Vec::new(); + while held.len() < MAX_TRACKED_PEERS { + match limiter.try_acquire(&peer(next_peer), now) { + Ok(grant) => { + held.push(grant); + next_peer += 1; + } + Err(_) => now += GLOBAL_INTERVAL, + } + } + assert_eq!(limiter.tracked_peers(), MAX_TRACKED_PEERS); + // No entry can be forgotten, so a newcomer is refused rather than + // tracked, although the global allowance has long refilled. + let newcomer = peer(u32::MAX); + let later = now + Duration::from_secs(10); + assert_eq!( + limiter.try_acquire(&newcomer, later).err(), + Some(BusyReplyDenied::RateLimited) + ); + assert_eq!(limiter.tracked_peers(), MAX_TRACKED_PEERS); + // Once those replies end, idle entries make room. + drop(held); + assert!(grant(&limiter, &newcomer, later)); + assert!(limiter.tracked_peers() <= MAX_TRACKED_PEERS); + } + + #[test] + fn a_peer_with_a_reply_in_flight_is_never_forgotten() { + let limiter = limiter(); + let t0 = Instant::now(); + let held: Vec = (0..PEER_BURST) + .map(|_| limiter.try_acquire(&peer(0), t0).expect("grant")) + .collect(); + let mut now = t0; + for i in 1..u32::try_from(MAX_TRACKED_PEERS).unwrap() { + while !grant(&limiter, &peer(i), now) { + now += GLOBAL_INTERVAL; + } + } + // Long after, everyone else is idle and a full table sheds them, but + // never the peer whose reply is still in flight. + let later = now + Duration::from_secs(3600); + assert!(grant(&limiter, &peer(u32::MAX), later)); + assert_eq!( + limiter.try_acquire(&peer(0), later).err(), + Some(BusyReplyDenied::PeerInFlight) + ); + drop(held); + assert!(grant(&limiter, &peer(0), later)); + } + + const WAIT: Duration = Duration::from_secs(5); + + #[test] + fn a_busy_source_is_put_last_while_it_cools() { + let tracker = SourceBusyTracker::new(); + let (a, b, c) = (peer(1), peer(2), peer(3)); + let sources = [a, b, c]; + let t0 = Instant::now(); + assert_eq!(tracker.choose(&sources, t0), Some(&a)); + assert_eq!(tracker.record_busy(&a, t0, WAIT), WAIT); + assert!(tracker.is_cooling(&a, t0)); + assert_eq!(tracker.choose(&sources, t0), Some(&b)); + // Cooling ends after the wait. + assert_eq!(tracker.choose(&sources, t0 + WAIT), Some(&a)); + } + + #[test] + fn when_every_source_cools_the_first_is_still_chosen() { + let tracker = SourceBusyTracker::new(); + let (a, b) = (peer(1), peer(2)); + let t0 = Instant::now(); + tracker.record_busy(&a, t0, WAIT); + tracker.record_busy(&b, t0, WAIT); + assert_eq!(tracker.choose(&[a, b], t0), Some(&a)); + assert_eq!(tracker.choose(&[], t0), None); + } + + #[test] + fn a_source_that_stays_busy_cools_longer_up_to_the_cap() { + let tracker = SourceBusyTracker::new(); + let a = peer(1); + let t0 = Instant::now(); + let cooldowns: Vec = (0..6).map(|_| tracker.record_busy(&a, t0, WAIT)).collect(); + assert_eq!( + cooldowns, + vec![WAIT, WAIT * 2, WAIT * 4, WAIT * 8, WAIT * 8, WAIT * 8] + ); + // A longer wait reaches the cap. + let b = peer(2); + let long = Duration::from_secs(10); + for _ in 0..4 { + tracker.record_busy(&b, t0, long); + } + assert_eq!(tracker.record_busy(&b, t0, long), MAX_SOURCE_COOLDOWN); + } + + #[test] + fn a_running_cooldown_is_never_shortened() { + let tracker = SourceBusyTracker::new(); + let a = peer(1); + let t0 = Instant::now(); + for _ in 0..4 { + tracker.record_busy(&a, t0, Duration::from_secs(10)); + } + // Cooling until t0 + 60 s now. A later busy answer with a tiny wait + // would end far sooner; the running cooldown stands. + tracker.record_busy(&a, t0 + Duration::from_secs(1), Duration::from_millis(1)); + assert!(tracker.is_cooling(&a, t0 + Duration::from_secs(59))); + } + + #[test] + fn serving_once_clears_the_record() { + let tracker = SourceBusyTracker::new(); + let a = peer(1); + let t0 = Instant::now(); + for _ in 0..3 { + tracker.record_busy(&a, t0, WAIT); + } + tracker.record_served(&a); + assert!(!tracker.is_cooling(&a, t0)); + assert!(tracker.is_empty()); + assert_eq!(tracker.record_busy(&a, t0, WAIT), WAIT); + } + + #[test] + fn a_stale_record_starts_over() { + let tracker = SourceBusyTracker::new(); + let a = peer(1); + let t0 = Instant::now(); + for _ in 0..3 { + tracker.record_busy(&a, t0, WAIT); + } + // Long after its last busy answer, the next one counts as the first. + let later = t0 + SOURCE_RECORD_TTL + MAX_SOURCE_COOLDOWN; + assert_eq!(tracker.record_busy(&a, later, WAIT), WAIT); + } + + #[test] + fn the_source_table_stays_bounded_and_spares_cooling_sources() { + let tracker = SourceBusyTracker::new(); + let t0 = Instant::now(); + // One source whose cooldown has ended, the rest still cooling. + tracker.record_busy(&peer(0), t0, Duration::from_millis(1)); + let t1 = t0 + Duration::from_secs(1); + let max = u32::try_from(MAX_TRACKED_SOURCES).unwrap(); + for i in 1..max { + tracker.record_busy(&peer(i), t1, WAIT); + } + assert_eq!(tracker.len(), MAX_TRACKED_SOURCES); + let newcomer = peer(u32::MAX); + tracker.record_busy(&newcomer, t1, WAIT); + assert_eq!(tracker.len(), MAX_TRACKED_SOURCES); + assert!(tracker.is_cooling(&newcomer, t1)); + // The one that had stopped cooling made room; every cooling one stayed. + assert!((1..max).all(|i| tracker.is_cooling(&peer(i), t1))); + // Long after, expired records make room before anything is displaced. + let later = t1 + SOURCE_RECORD_TTL + MAX_SOURCE_COOLDOWN; + tracker.record_busy(&peer(u32::MAX - 1), later, WAIT); + assert_eq!(tracker.len(), 1); + } +} diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 62937413..7288160e 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -22,6 +22,7 @@ pub mod bootstrap; pub mod commitment; pub mod commitment_state; pub mod config; +pub mod fetch_busy; pub mod fresh; pub mod neighbor_sync; pub mod paid_list; @@ -42,6 +43,8 @@ use std::fmt; use std::num::NonZeroUsize; use std::ops::ControlFlow; use std::path::{Path, PathBuf}; +#[cfg(any(test, feature = "test-utils"))] +use std::sync::atomic::AtomicBool; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use std::time::{Duration, Instant, SystemTime}; @@ -56,7 +59,7 @@ use parking_lot::Mutex; use rand::rngs::StdRng; use rand::{Rng, SeedableRng}; use tokio::sync::broadcast::error::RecvError; -use tokio::sync::{mpsc, Notify, RwLock, Semaphore}; +use tokio::sync::{mpsc, Notify, OwnedSemaphorePermit, RwLock, Semaphore}; use tokio::task::JoinHandle; use tokio_util::sync::CancellationToken; use tokio_util::task::TaskTracker; @@ -72,7 +75,7 @@ use crate::pointer::{FinalStateWitness, PointerService}; use crate::replication::audit::AuditTickResult; use crate::replication::audit_coordinator::AuditChallengeCoordinator; use crate::replication::audit_metrics::{ - AuditResponderClass, AuditResponderDropReason, AuditResponderMetrics, + AuditResponderClass, AuditResponderDropReason, AuditResponderMetrics, FetchBusyReplyOutcome, ReplicationResponderClass, ResponderAdmissionCeiling, }; use crate::replication::commitment::{commitment_hash, StorageCommitment}; @@ -88,6 +91,10 @@ use crate::replication::config::{ SUBTREE_AUDIT_PROTOCOL_ID, SUBTREE_ROUND1_WORK_BURST_BYTES, SUBTREE_ROUND1_WORK_REFILL_BYTES_PER_SEC, SUBTREE_SESSION_TTL, }; +use crate::replication::fetch_busy::{ + busy_wait, BusyReplyDenied, BusyReplyGrant, FetchBusyReplyLimiter, SourceBusyTracker, + FETCH_BUSY_REPLY_SEND_TIMEOUT, FETCH_BUSY_WAIT_SPREAD, +}; use crate::replication::paid_list::PaidList; use crate::replication::protocol::{ FreshReplicationResponse, NeighborSyncResponse, ReplicationMessage, ReplicationMessageBody, @@ -1129,6 +1136,14 @@ const FETCH_RESPONDER_MAX_OUTSTANDING: usize = FETCH_RESPONDER_WORKER_LIMIT * 4; /// of the three workers available to other peers under a single-source flood. const FETCH_RESPONDER_MAX_OUTSTANDING_PER_PEER: u32 = 2; +/// Maximum busy replies being sent at once. +/// +/// [`FetchBusyReplyLimiter`] bounds how often busy replies are sent; this +/// bounds how many sends are open at once, so peers that are slow to take a +/// reply cannot pile up send tasks. Past it, a refusal falls back to the +/// silence every refusal got before busy replies existed. +const FETCH_BUSY_REPLY_MAX_OUTSTANDING: usize = FETCH_RESPONDER_MAX_OUTSTANDING * 4; + /// Maximum verification batches served concurrently. /// /// Point lookups are fast, but a batch can contain 8,192 of them. Two @@ -1787,6 +1802,20 @@ pub struct ReplicationEngine { fetch_responder_admission_semaphore: Arc, /// Per-source fetch responder counts for flood-fair admission. fetch_responder_inflight: Arc>>, + /// Test-only hold on serving admitted fetch requests. + #[cfg(any(test, feature = "test-utils"))] + fetch_serve_gate: Arc, + /// Test-only count of this engine's fetch refusals because its admission + /// pool was full. + #[cfg(any(test, feature = "test-utils"))] + fetch_pool_full_refusals: Arc, + /// Permits for busy replies being sent to fetch requests refused at + /// admission. + fetch_busy_reply_semaphore: Arc, + /// Per-peer and global rate bound on those busy replies. + fetch_busy_reply_limiter: Arc, + /// Sources that recently answered this node's own fetches busy. + fetch_busy_sources: Arc, /// Worker permits for lookup-heavy verification batches. verification_responder_worker_semaphore: Arc, /// Admission permits bounding verification workers and queued waiters. @@ -1940,6 +1969,13 @@ impl ReplicationEngine { FETCH_RESPONDER_MAX_OUTSTANDING, )), fetch_responder_inflight: Arc::new(RwLock::new(HashMap::new())), + #[cfg(any(test, feature = "test-utils"))] + fetch_serve_gate: Arc::new(FetchServeGate::default()), + #[cfg(any(test, feature = "test-utils"))] + fetch_pool_full_refusals: Arc::new(AtomicU64::new(0)), + fetch_busy_reply_semaphore: Arc::new(Semaphore::new(FETCH_BUSY_REPLY_MAX_OUTSTANDING)), + fetch_busy_reply_limiter: Arc::new(FetchBusyReplyLimiter::new()), + fetch_busy_sources: Arc::new(SourceBusyTracker::new()), verification_responder_worker_semaphore: Arc::new(Semaphore::new( VERIFICATION_RESPONDER_WORKER_LIMIT, )), @@ -2172,6 +2208,53 @@ impl ReplicationEngine { queues.enqueue_fetch(key, distance, sources) } + /// Test-only: hold this engine's admitted fetch requests before they are + /// served, until [`Self::open_fetch_serve_gate_for_test`]. + #[cfg(any(test, feature = "test-utils"))] + pub fn close_fetch_serve_gate_for_test(&self) { + self.fetch_serve_gate.close(); + } + + /// Test-only: release this engine's held fetch requests and stop holding + /// new ones. + #[cfg(any(test, feature = "test-utils"))] + pub fn open_fetch_serve_gate_for_test(&self) { + self.fetch_serve_gate.open(); + } + + /// Test-only: whether this engine puts `source` last when choosing where + /// to start or retry a fetch, because it recently answered busy. + #[cfg(any(test, feature = "test-utils"))] + #[must_use] + pub fn fetch_source_cooling_for_test(&self, source: &PeerId) -> bool { + self.fetch_busy_sources.is_cooling(source, Instant::now()) + } + + /// Test-only: when `source` last answered this engine's fetches busy, if + /// it is still remembered. + #[cfg(any(test, feature = "test-utils"))] + #[must_use] + pub fn fetch_source_last_busy_for_test(&self, source: &PeerId) -> Option { + self.fetch_busy_sources.last_busy(source) + } + + /// Test-only: fetch requests this engine has admitted and not yet finished + /// serving. + #[cfg(any(test, feature = "test-utils"))] + #[must_use] + pub fn fetch_admitted_for_test(&self) -> usize { + FETCH_RESPONDER_MAX_OUTSTANDING + .saturating_sub(self.fetch_responder_admission_semaphore.available_permits()) + } + + /// Test-only: fetch requests this engine has refused because its + /// admission pool was full. + #[cfg(any(test, feature = "test-utils"))] + #[must_use] + pub fn fetch_pool_full_refusals_for_test(&self) -> u64 { + self.fetch_pool_full_refusals.load(Ordering::Relaxed) + } + /// Test-only: whether `key` is still tracked in any fetch-pipeline stage /// (pending verification, fetch queue, or in-flight fetch). #[cfg(any(test, feature = "test-utils"))] @@ -2434,6 +2517,12 @@ impl ReplicationEngine { } } + // A test may shut down with the fetch serve gate still closed. The + // requests it holds wait on the gate before they reach a worker pool, + // so release them, or the drain below never finishes. + #[cfg(any(test, feature = "test-utils"))] + self.fetch_serve_gate.open(); + // Close the responder worker pools. Detached tasks still queued for a // worker take the `Err` arm of `acquire_owned()` and exit immediately // rather than waiting for a slot that will never free up — without @@ -3063,6 +3152,12 @@ impl ReplicationEngine { let fetch_responder_admission_semaphore = Arc::clone(&self.fetch_responder_admission_semaphore); let fetch_responder_inflight = Arc::clone(&self.fetch_responder_inflight); + #[cfg(any(test, feature = "test-utils"))] + let fetch_serve_gate = Arc::clone(&self.fetch_serve_gate); + #[cfg(any(test, feature = "test-utils"))] + let fetch_pool_full_refusals = Arc::clone(&self.fetch_pool_full_refusals); + let fetch_busy_reply_semaphore = Arc::clone(&self.fetch_busy_reply_semaphore); + let fetch_busy_reply_limiter = Arc::clone(&self.fetch_busy_reply_limiter); let verification_responder_worker_semaphore = Arc::clone(&self.verification_responder_worker_semaphore); let verification_responder_admission_semaphore = @@ -3119,6 +3214,12 @@ impl ReplicationEngine { fetch_responder_worker_semaphore, fetch_responder_admission_semaphore, fetch_responder_inflight, + #[cfg(any(test, feature = "test-utils"))] + fetch_serve_gate, + #[cfg(any(test, feature = "test-utils"))] + fetch_pool_full_refusals, + fetch_busy_reply_semaphore, + fetch_busy_reply_limiter, verification_responder_worker_semaphore, verification_responder_admission_semaphore, verification_responder_inflight, @@ -3881,6 +3982,7 @@ impl ReplicationEngine { let is_bootstrapping = Arc::clone(&self.is_bootstrapping); let bootstrap_complete_notify = Arc::clone(&self.bootstrap_complete_notify); let detached_task_tracker = self.detached_task_tracker.clone(); + let busy_sources = Arc::clone(&self.fetch_busy_sources); let concurrency = max_parallel_fetch(); info!("Fetch worker concurrency set to {concurrency} (hardware threads)"); @@ -3899,7 +4001,11 @@ impl ReplicationEngine { break; }; let fetch_key = candidate.key; - let Some(&source) = candidate.sources.first() else { + // A source that just answered this node busy is the + // worst place to start a new key; start on the first + // that is not cooling down, or the first if all are. + let Some(&source) = busy_sources.choose(&candidate.sources, Instant::now()) + else { warn!( "Fetch candidate {} has no sources; requeueing for verification", hex::encode(fetch_key) @@ -3915,6 +4021,7 @@ impl ReplicationEngine { let p2p = Arc::clone(&p2p); let storage = Arc::clone(&storage); let config = Arc::clone(&config); + let busy_sources = Arc::clone(&busy_sources); let token = shutdown.clone(); let tracker = detached_task_tracker.clone(); in_flight.push(Box::pin(async move { @@ -3930,7 +4037,7 @@ impl ReplicationEngine { result: FetchResult::SourceFailed, }, outcome = execute_single_fetch( - p2p, storage, config, fetch_key, source, + p2p, storage, config, busy_sources, fetch_key, source, ) => outcome, } }); @@ -3969,6 +4076,7 @@ impl ReplicationEngine { &key, &outcome.result, config.verification_request_timeout, + &busy_sources, ) { FetchFollowUp::Terminal => true, FetchFollowUp::RequeuedForVerification => false, @@ -3977,6 +4085,7 @@ impl ReplicationEngine { let p2p = Arc::clone(&p2p); let storage = Arc::clone(&storage); let config = Arc::clone(&config); + let busy_sources = Arc::clone(&busy_sources); let token = shutdown.clone(); let tracker = detached_task_tracker.clone(); let fetch_key = key; @@ -3990,7 +4099,12 @@ impl ReplicationEngine { result: FetchResult::SourceFailed, }, outcome = execute_single_fetch( - p2p, storage, config, fetch_key, next_peer, + p2p, + storage, + config, + busy_sources, + fetch_key, + next_peer, ) => outcome, } }); @@ -4589,6 +4703,15 @@ struct ReplicationMessageHandlerContext { fetch_responder_worker_semaphore: Arc, fetch_responder_admission_semaphore: Arc, fetch_responder_inflight: Arc>>, + /// Test-only hold on serving admitted fetch requests. + #[cfg(any(test, feature = "test-utils"))] + fetch_serve_gate: Arc, + /// Test-only count of this engine's fetch refusals because its admission + /// pool was full. + #[cfg(any(test, feature = "test-utils"))] + fetch_pool_full_refusals: Arc, + fetch_busy_reply_semaphore: Arc, + fetch_busy_reply_limiter: Arc, verification_responder_worker_semaphore: Arc, verification_responder_admission_semaphore: Arc, verification_responder_inflight: Arc>>, @@ -7112,6 +7235,45 @@ pub fn verification_requests_for_key_from_for_test(requester: &PeerId, key: &Xor }) } +/// Test-only: holds an engine's admitted fetch requests before they are served. +/// +/// A held request keeps its admission slot and its requester's per-peer share, +/// exactly as a request waiting for a worker does, so a test can read what a +/// requester has outstanding without racing the speed at which the host serves +/// a chunk. One per engine, so closing it affects no other node or test. +#[cfg(any(test, feature = "test-utils"))] +#[derive(Debug, Default)] +struct FetchServeGate { + closed: AtomicBool, + opened: Notify, +} + +#[cfg(any(test, feature = "test-utils"))] +impl FetchServeGate { + fn close(&self) { + self.closed.store(true, Ordering::Release); + } + + fn open(&self) { + self.closed.store(false, Ordering::Release); + self.opened.notify_waiters(); + } + + /// Wait while the gate is closed. The wakeup is registered before the flag + /// is read, so an open that lands between the two is not missed. + async fn wait_open(&self) { + loop { + let opened = self.opened.notified(); + tokio::pin!(opened); + opened.as_mut().enable(); + if !self.closed.load(Ordering::Acquire) { + return; + } + opened.await; + } + } +} + async fn handle_verification_request( source: &PeerId, request: &protocol::VerificationRequest, @@ -7351,11 +7513,42 @@ async fn dispatch_fetch_request( ReplicationResponderClass::Fetch, failure.reason.into(), ); + #[cfg(any(test, feature = "test-utils"))] + if failure.reason == ResponderRejectReason::GlobalPoolFull { + ctx.fetch_pool_full_refusals.fetch_add(1, Ordering::Relaxed); + } + let busy_reply = fetch_busy_reply_permit(&source, request.busy_reply, ctx); + let capability = if request.busy_reply { + "asked" + } else { + "not_asked" + }; + let reply = if busy_reply.is_some() { + "scheduled" + } else { + "none" + }; + // Same line and counter as before busy replies existed, so the + // existing refusal counts carry on. `capability` says whether the + // requester asked for a busy reply, and `reply` whether one was + // scheduled; whether it was delivered is in the busy-reply counters. warn!( responder_class = "fetch", source = %source, + capability, + reply, "Fetch response dropped at admission: {failure}" ); + if let Some((permit, grant)) = busy_reply { + send_fetch_busy_reply( + source, + request.key, + ctx, + request_id, + rr_message_id, + (permit, grant), + ); + } return Ok(()); } }; @@ -7365,8 +7558,12 @@ async fn dispatch_fetch_request( let p2p_node = Arc::clone(&ctx.p2p_node); let request_timeout = ctx.config.fetch_request_timeout; let rr_message_id = rr_message_id.map(ToOwned::to_owned); + #[cfg(any(test, feature = "test-utils"))] + let serve_gate = Arc::clone(&ctx.fetch_serve_gate); ctx.detached_task_tracker.spawn(async move { let _guard = guard; + #[cfg(any(test, feature = "test-utils"))] + serve_gate.wait_open().await; let Ok(_worker_permit) = worker_semaphore.acquire_owned().await else { debug!( responder_class = "fetch", @@ -7402,6 +7599,91 @@ async fn dispatch_fetch_request( Ok(()) } +/// Whether a fetch refused at admission may be answered `Busy`: the send +/// permit and the limiter's grant to do it with. +/// +/// Only a requester that asked for busy replies gets one, and only while a +/// send permit is free, within the limiter's per-peer and global allowances, +/// and while fewer than the limiter's in-flight limit of busy replies to that +/// peer (two) are still being sent. Every other +/// refusal stays silent, exactly as all of them were before busy replies +/// existed. +fn fetch_busy_reply_permit( + source: &PeerId, + busy_reply: bool, + ctx: &ReplicationMessageHandlerContext, +) -> Option<(OwnedSemaphorePermit, BusyReplyGrant)> { + if !busy_reply { + audit_metrics::record_fetch_busy_reply(FetchBusyReplyOutcome::NotAsked); + return None; + } + // The send permit first: if none is free, the reply would not be sent, so + // the rate allowance is left unspent. + let Ok(permit) = Arc::clone(&ctx.fetch_busy_reply_semaphore).try_acquire_owned() else { + audit_metrics::record_fetch_busy_reply(FetchBusyReplyOutcome::SendPoolFull); + return None; + }; + match ctx + .fetch_busy_reply_limiter + .try_acquire(source, Instant::now()) + { + Ok(grant) => Some((permit, grant)), + Err(denied) => { + audit_metrics::record_fetch_busy_reply(match denied { + BusyReplyDenied::RateLimited => FetchBusyReplyOutcome::RateLimited, + BusyReplyDenied::PeerInFlight => FetchBusyReplyOutcome::PeerInFlight, + }); + None + } + } +} + +/// Tell a requester its fetch was refused at admission, off the serial message +/// loop. +/// +/// The send runs on the detached task tracker, holding the send permit and +/// one of the peer's in-flight places until it ends, so a slow or stalled peer +/// can neither block other replication traffic nor hold more permits than its +/// in-flight limit (two). It +/// is abandoned after [`FETCH_BUSY_REPLY_SEND_TIMEOUT`], and it races the +/// engine's shutdown, so `shutdown()` never waits on a peer that is slow to +/// take the reply. +fn send_fetch_busy_reply( + source: PeerId, + key: XorName, + ctx: &ReplicationMessageHandlerContext, + request_id: u64, + rr_message_id: Option<&str>, + held: (OwnedSemaphorePermit, BusyReplyGrant), +) { + let p2p_node = Arc::clone(&ctx.p2p_node); + let shutdown = ctx.shutdown.clone(); + let rr_message_id = rr_message_id.map(ToOwned::to_owned); + ctx.detached_task_tracker.spawn(async move { + let _held = held; + let delivery = tokio::time::timeout( + FETCH_BUSY_REPLY_SEND_TIMEOUT, + send_replication_response_checked( + &source, + &p2p_node, + request_id, + ReplicationMessageBody::FetchResponse(protocol::FetchResponse::Busy { key }), + rr_message_id.as_deref(), + None, + ), + ); + let outcome = tokio::select! { + () = shutdown.cancelled() => FetchBusyReplyOutcome::Cancelled, + sent = delivery => match sent { + Ok(true) => FetchBusyReplyOutcome::Sent, + Ok(false) => FetchBusyReplyOutcome::SendFailed, + Err(_) => FetchBusyReplyOutcome::TimedOut, + }, + }; + audit_metrics::record_fetch_busy_reply(outcome); + }); +} + fn request_is_stale(received_at: Instant, timeout: Duration) -> bool { received_at.elapsed() >= timeout } @@ -7666,15 +7948,7 @@ async fn send_replication_response_checked( return false; } }; - // V2-684: per-peer served-bytes attribution for the heavy serve paths. - // `FetchResponse` carries ~99% of served bytes; `NeighborSyncResponse` is - // included for completeness. Other response variants (verification/audit/ - // commitment) are intentionally excluded — the round-2 audit reply is now a - // few-KB verified slice (V2-685), not a full-chunk transfer, so it is light. - if matches!( - msg.body, - ReplicationMessageBody::FetchResponse(_) | ReplicationMessageBody::NeighborSyncResponse(_) - ) { + if counts_as_served(&msg.body) { protocol::record_served(peer, encoded.len()); } let protocol = protocol.unwrap_or_else(|| response_protocol_for(&msg.body)); @@ -7692,6 +7966,26 @@ async fn send_replication_response_checked( true } +/// Whether a response's bytes count towards the peer it is sent to in the +/// per-peer served-bytes attribution (V2-684). +/// +/// That attribution covers the heavy serve paths. `FetchResponse` carries ~99% +/// of served bytes; `NeighborSyncResponse` is included for completeness. Other +/// response variants (verification/audit/commitment) are intentionally +/// excluded — the round-2 audit reply is now a few-KB verified slice (V2-685), +/// not a full-chunk transfer, so it is light. A busy reply serves nothing, so +/// it is left out too. +const fn counts_as_served(body: &ReplicationMessageBody) -> bool { + matches!( + body, + ReplicationMessageBody::FetchResponse( + protocol::FetchResponse::Success { .. } + | protocol::FetchResponse::NotFound { .. } + | protocol::FetchResponse::Error { .. } + ) | ReplicationMessageBody::NeighborSyncResponse(_) + ) +} + async fn record_sent_replica_hints( peer: &PeerId, hints: &[neighbor_sync::SentReplicaHint], @@ -9013,6 +9307,7 @@ fn apply_fetch_result( key: &XorName, result: &FetchResult, verification_retry_after: Duration, + busy_sources: &SourceBusyTracker, ) -> FetchFollowUp { match result { FetchResult::Stored | FetchResult::NoLongerResponsible | FetchResult::AlreadyHeld => { @@ -9020,7 +9315,12 @@ fn apply_fetch_result( FetchFollowUp::Terminal } FetchResult::IntegrityFailed | FetchResult::SourceFailed => { - if let Some(next_peer) = q.retry_fetch(key) { + // The next untried source in preference order, passing over any + // that recently answered this node busy unless all of them have. + let now = Instant::now(); + if let Some(next_peer) = + q.retry_fetch_with(key, |untried| busy_sources.choose(untried, now).copied()) + { FetchFollowUp::RetryFrom(next_peer) } else if q.requeue_fetch_for_verification(key, verification_retry_after) { FetchFollowUp::RequeuedForVerification @@ -9090,6 +9390,7 @@ async fn execute_single_fetch( p2p_node: Arc, storage: Arc, config: Arc, + busy_sources: Arc, key: XorName, source: PeerId, ) -> FetchOutcome { @@ -9147,7 +9448,10 @@ async fn execute_single_fetch( }; } - let request = protocol::FetchRequest { key }; + let request = protocol::FetchRequest { + key, + busy_reply: true, + }; let msg = ReplicationMessage { request_id: rand::thread_rng().gen::(), body: ReplicationMessageBody::FetchRequest(request), @@ -9164,6 +9468,7 @@ async fn execute_single_fetch( } }; + let sent_at = Instant::now(); let result = p2p_node .send_request( &source, @@ -9257,6 +9562,10 @@ async fn execute_single_fetch( }; } + // It served a valid chunk, so any earlier busy answers no + // longer count against it when choosing where to start. + busy_sources.record_served(&source); + // Responsibility can lapse during the network round-trip. // The bandwidth is already spent; declining the write is // what still avoids the disk write and the later @@ -9325,8 +9634,46 @@ async fn execute_single_fetch( result: FetchResult::SourceFailed, } } + ReplicationMessageBody::FetchResponse(protocol::FetchResponse::Busy { + key: resp_key, + }) if resp_key == key => { + // The source refused at admission and said so, where it + // used to stay silent. That is an honest, prompt answer, so + // it carries no trust event. Waiting here, still holding + // this fetch slot, before the key moves on is what keeps a + // busy answer from becoming a fast retry loop: this slot asks + // at most once per wait, where silence made it once per fetch + // timeout. The attempt then ends exactly as that timeout + // did, so nothing downstream changes. The worker races this + // whole task against shutdown, so the wait does not delay it. + audit_metrics::record_fetch_busy_received(); + let jitter = + rand::thread_rng().gen_range(Duration::ZERO..=FETCH_BUSY_WAIT_SPREAD); + // The holder is put last for at least the full jittered + // wait, even when this attempt's own deadline cuts the + // sleep short. + let full_wait = busy_wait(Duration::MAX, jitter); + let remaining = config + .fetch_request_timeout + .saturating_sub(sent_at.elapsed()); + let wait = full_wait.min(remaining); + let cooldown = busy_sources.record_busy(&source, Instant::now(), full_wait); + debug!( + "Fetch: source {source} busy for {}; moving on in {} ms, \ + passing it over for new keys for {} ms", + hex::encode(key), + wait.as_millis(), + cooldown.as_millis() + ); + tokio::time::sleep(wait).await; + FetchOutcome { + key, + result: FetchResult::SourceFailed, + } + } _ => { - // Unexpected message type — treat as malformed. + // Unexpected message type, or a busy answer for another + // key — treat as malformed. p2p_node .report_trust_event( &source, @@ -13287,6 +13634,7 @@ mod tests { request_id: 1, body: ReplicationMessageBody::FetchRequest(protocol::FetchRequest { key: test_key(1), + busy_reply: true, }), }, rr_message_id: None, @@ -13779,8 +14127,13 @@ mod tests { drive_key_in_flight(&mut q, key, vec![test_peer(0x01), test_peer(0x02)]); assert_eq!(q.retry_reserved_slot_count(), 1); - let follow_up = - apply_fetch_result(&mut q, &key, &FetchResult::NoLongerResponsible, RETRY_AFTER); + let follow_up = apply_fetch_result( + &mut q, + &key, + &FetchResult::NoLongerResponsible, + RETRY_AFTER, + &SourceBusyTracker::new(), + ); assert_eq!( follow_up, @@ -13815,7 +14168,13 @@ mod tests { let key = test_key(0xCD); drive_key_in_flight(&mut q, key, vec![test_peer(0x03)]); - let follow_up = apply_fetch_result(&mut q, &key, &result, RETRY_AFTER); + let follow_up = apply_fetch_result( + &mut q, + &key, + &result, + RETRY_AFTER, + &SourceBusyTracker::new(), + ); assert_eq!(follow_up, FetchFollowUp::Terminal); assert!(!q.contains_key(&key)); @@ -13833,7 +14192,13 @@ mod tests { let second = test_peer(0x05); drive_key_in_flight(&mut q, key, vec![first, second]); - let follow_up = apply_fetch_result(&mut q, &key, &FetchResult::SourceFailed, RETRY_AFTER); + let follow_up = apply_fetch_result( + &mut q, + &key, + &FetchResult::SourceFailed, + RETRY_AFTER, + &SourceBusyTracker::new(), + ); assert_eq!( follow_up, FetchFollowUp::RetryFrom(second), @@ -13841,7 +14206,13 @@ mod tests { ); assert_eq!(q.in_flight_count(), 1, "retry keeps the key in flight"); - let follow_up = apply_fetch_result(&mut q, &key, &FetchResult::SourceFailed, RETRY_AFTER); + let follow_up = apply_fetch_result( + &mut q, + &key, + &FetchResult::SourceFailed, + RETRY_AFTER, + &SourceBusyTracker::new(), + ); assert_eq!( follow_up, FetchFollowUp::RequeuedForVerification, @@ -13855,6 +14226,91 @@ mod tests { ); } + /// A retry passes over an untried source that recently answered busy when + /// another untried source is free, and still takes it when it is the only + /// one left: busy holders go last, never out. + #[test] + fn source_failure_retries_on_a_holder_that_is_not_cooling_first() { + const RETRY_AFTER: Duration = Duration::from_mins(1); + + let mut q = ReplicationQueues::new(); + let key = test_key(0xE1); + let (first, cooling, free) = (test_peer(0x11), test_peer(0x12), test_peer(0x13)); + drive_key_in_flight(&mut q, key, vec![first, cooling, free]); + let busy_sources = SourceBusyTracker::new(); + busy_sources.record_busy(&cooling, Instant::now(), Duration::from_mins(1)); + + let follow_up = apply_fetch_result( + &mut q, + &key, + &FetchResult::SourceFailed, + RETRY_AFTER, + &busy_sources, + ); + assert_eq!(follow_up, FetchFollowUp::RetryFrom(free)); + + let follow_up = apply_fetch_result( + &mut q, + &key, + &FetchResult::SourceFailed, + RETRY_AFTER, + &busy_sources, + ); + assert_eq!( + follow_up, + FetchFollowUp::RetryFrom(cooling), + "a cooling holder is still tried once it is the last untried one" + ); + + let follow_up = apply_fetch_result( + &mut q, + &key, + &FetchResult::SourceFailed, + RETRY_AFTER, + &busy_sources, + ); + assert_eq!(follow_up, FetchFollowUp::RequeuedForVerification); + } + + /// A busy reply serves nothing, so it never counts as served bytes; the + /// answers that did before still do. + #[test] + fn a_busy_reply_is_not_counted_as_served() { + let key = test_key(0x77); + assert!(!counts_as_served(&ReplicationMessageBody::FetchResponse( + protocol::FetchResponse::Busy { key } + ))); + for response in [ + protocol::FetchResponse::Success { + key, + data: vec![1, 2, 3], + }, + protocol::FetchResponse::NotFound { key }, + protocol::FetchResponse::Error { + key, + reason: "io".to_string(), + }, + ] { + assert!(counts_as_served(&ReplicationMessageBody::FetchResponse( + response + ))); + } + assert!(counts_as_served( + &ReplicationMessageBody::NeighborSyncResponse(protocol::NeighborSyncResponse { + replica_hints: vec![], + paid_hints: vec![], + bootstrapping: false, + rejected_keys: vec![], + commitment: None, + }) + )); + assert!(!counts_as_served( + &ReplicationMessageBody::VerificationResponse(protocol::VerificationResponse { + results: vec![], + }) + )); + } + #[test] fn source_failure_without_retry_metadata_is_terminal() { const RETRY_AFTER: Duration = Duration::from_mins(1); @@ -13868,7 +14324,13 @@ mod tests { let candidate = q.dequeue_fetch().expect("enqueued key must dequeue"); q.start_dequeued_fetch(candidate, source); - let follow_up = apply_fetch_result(&mut q, &key, &FetchResult::SourceFailed, RETRY_AFTER); + let follow_up = apply_fetch_result( + &mut q, + &key, + &FetchResult::SourceFailed, + RETRY_AFTER, + &SourceBusyTracker::new(), + ); assert_eq!(follow_up, FetchFollowUp::Terminal); assert!(!q.contains_key(&key)); @@ -13883,7 +14345,13 @@ mod tests { let source = test_peer(0x0B); drive_key_in_flight(&mut q, key, vec![source, test_peer(0x0C)]); - let follow_up = apply_fetch_result(&mut q, &key, &FetchResult::AlreadyHeld, RETRY_AFTER); + let follow_up = apply_fetch_result( + &mut q, + &key, + &FetchResult::AlreadyHeld, + RETRY_AFTER, + &SourceBusyTracker::new(), + ); assert_eq!( follow_up, @@ -13912,8 +14380,13 @@ mod tests { let third = test_peer(0x09); drive_key_in_flight(&mut q, key, vec![first, second, third]); - let follow_up = - apply_fetch_result(&mut q, &key, &FetchResult::LocalWriteFailed, RETRY_AFTER); + let follow_up = apply_fetch_result( + &mut q, + &key, + &FetchResult::LocalWriteFailed, + RETRY_AFTER, + &SourceBusyTracker::new(), + ); assert_eq!( follow_up, @@ -13946,8 +14419,13 @@ mod tests { let candidate = q.dequeue_fetch().expect("enqueued key must dequeue"); q.start_dequeued_fetch(candidate, source); - let follow_up = - apply_fetch_result(&mut q, &key, &FetchResult::LocalWriteFailed, RETRY_AFTER); + let follow_up = apply_fetch_result( + &mut q, + &key, + &FetchResult::LocalWriteFailed, + RETRY_AFTER, + &SourceBusyTracker::new(), + ); assert_eq!(follow_up, FetchFollowUp::Terminal); assert!(!q.contains_key(&key)); diff --git a/src/replication/protocol.rs b/src/replication/protocol.rs index ebb4644a..d98eec80 100644 --- a/src/replication/protocol.rs +++ b/src/replication/protocol.rs @@ -22,6 +22,35 @@ use super::config::MAX_SUBTREE_AUDIT_MESSAGE_SIZE; /// Used in [`AuditResponse::Digests`] for keys the peer does not hold. pub const ABSENT_KEY_DIGEST: [u8; 32] = [0u8; 32]; +/// Opens the fetch capability trailer that may follow an encoded +/// [`FetchRequest`] message. +/// +/// A requester states what it understands here, after the end of the postcard +/// message, because every older node decodes with postcard's `from_bytes`, +/// which ignores whatever follows the message. A field inside the message +/// would instead shift or break their decoding. The magic keeps a stray byte +/// after a message from ever reading as a capability. +pub const FETCH_CAPABILITY_MAGIC: [u8; 4] = *b"antf"; + +/// Bit in the fetch capability flags byte: the requester understands +/// [`FetchResponse::Busy`]. +pub const FETCH_CAPABILITY_BUSY_REPLY: u8 = 0b0000_0001; + +/// Length of the fetch capability trailer: the magic and one flags byte. +const FETCH_CAPABILITY_TRAILER_LEN: usize = FETCH_CAPABILITY_MAGIC.len() + 1; + +/// The fetch capability flags in `trailer`, the bytes after a decoded message. +/// +/// Zero unless `trailer` opens with [`FETCH_CAPABILITY_MAGIC`] and has a flags +/// byte after it. Bytes after the flags byte are reserved and ignored, so a +/// later version can extend the trailer without this one rejecting it. +fn fetch_capability_flags(trailer: &[u8]) -> u8 { + match trailer.strip_prefix(FETCH_CAPABILITY_MAGIC.as_slice()) { + Some([flags, ..]) => *flags, + _ => 0, + } +} + // --------------------------------------------------------------------------- // Top-level envelope // --------------------------------------------------------------------------- @@ -68,8 +97,24 @@ impl ReplicationMessage { // The buffer is sized exactly, and an oversized body is refused // before anything is allocated. Chunk-carrying bodies run to several // MiB and are held while they are sent. - let max_size = ceiling_for(family_of_variant(self.body.variant_index())); - let bytes = encode_exact(self, max_size).map_err(|e| match e { + // A fetch request that asks for busy replies carries the capability + // trailer after the message, so its message must leave room for it + // under the same ceiling the decoder applies to the whole input. + let asks_busy_reply = matches!( + &self.body, + ReplicationMessageBody::FetchRequest(FetchRequest { + busy_reply: true, + .. + }) + ); + let trailer_len = if asks_busy_reply { + FETCH_CAPABILITY_TRAILER_LEN + } else { + 0 + }; + let max_size = + ceiling_for(family_of_variant(self.body.variant_index())).saturating_sub(trailer_len); + let mut bytes = encode_exact(self, max_size).map_err(|e| match e { ExactEncodeError::Serialize(e) => { ReplicationProtocolError::SerializationFailed(e.to_string()) } @@ -81,6 +126,14 @@ impl ReplicationMessage { } })?; + // The capability trailer goes immediately after the message, where + // older decoders never look. + if asks_busy_reply { + bytes.reserve_exact(trailer_len); + bytes.extend_from_slice(&FETCH_CAPABILITY_MAGIC); + bytes.push(FETCH_CAPABILITY_BUSY_REPLY); + } + // V2-623: cumulative per-variant tx accounting. Every replication send // funnels through here, so this is the single tx choke point. record_tx(&self.body, bytes.len()); @@ -120,8 +173,14 @@ impl ReplicationMessage { max_size, }); } - let message: Self = postcard::from_bytes(data) + // `take_from_bytes` rather than `from_bytes` only to see what follows + // the message. Both ignore it otherwise, so nothing that decoded before + // decodes differently now. + let (mut message, trailer): (Self, &[u8]) = postcard::take_from_bytes(data) .map_err(|e| ReplicationProtocolError::DeserializationFailed(e.to_string()))?; + if let ReplicationMessageBody::FetchRequest(request) = &mut message.body { + request.busy_reply = fetch_capability_flags(trailer) & FETCH_CAPABILITY_BUSY_REPLY != 0; + } // V2-623: cumulative per-variant rx accounting. Every replication // receive funnels through here, so this is the single rx choke point. @@ -977,6 +1036,18 @@ pub struct VerificationResponse { pub struct FetchRequest { /// The key of the record to fetch. pub key: XorName, + /// Whether the requester understands [`FetchResponse::Busy`]. + /// + /// Not part of the serialized struct, whose layout is unchanged since the + /// first release: it travels in the fetch capability trailer that + /// [`ReplicationMessage::encode`] appends after the whole message, and + /// [`ReplicationMessage::decode`] reads back. Older nodes decode with + /// postcard's `from_bytes`, which stops at the end of the message and never + /// sees the trailer, so they keep refusing silently. A responder answers + /// `Busy` only when this is set, so an older requester, which never sends + /// the trailer, is never sent a variant it cannot decode. + #[serde(skip)] + pub busy_reply: bool, } /// Response to a fetch request. @@ -1004,6 +1075,16 @@ pub enum FetchResponse { /// Human-readable error description. reason: String, }, + /// The responder could not admit the request right now and did not + /// serve it. Nothing is implied about whether it holds the record. + /// + /// APPENDED so the discriminants of the variants above are unchanged. Sent + /// only in answer to a [`FetchRequest`] with `busy_reply` set; an older + /// requester never sets it, so it is never sent a value it cannot decode. + Busy { + /// The requested key. + key: XorName, + }, } // --------------------------------------------------------------------------- @@ -1740,7 +1821,10 @@ mod tests { paid_list_check_indices: vec![], }), ReplicationMessageBody::VerificationResponse(VerificationResponse { results: vec![] }), - ReplicationMessageBody::FetchRequest(FetchRequest { key: z }), + ReplicationMessageBody::FetchRequest(FetchRequest { + key: z, + busy_reply: true, + }), ReplicationMessageBody::FetchResponse(FetchResponse::NotFound { key: z }), ReplicationMessageBody::AuditChallenge(AuditChallenge { challenge_id: 1, @@ -2392,7 +2476,10 @@ mod tests { fn fetch_request_roundtrip() { let msg = ReplicationMessage { request_id: 9, - body: ReplicationMessageBody::FetchRequest(FetchRequest { key: [0x40; 32] }), + body: ReplicationMessageBody::FetchRequest(FetchRequest { + key: [0x40; 32], + busy_reply: true, + }), }; let encoded = msg.encode().expect("encode should succeed"); let decoded = ReplicationMessage::decode(&encoded).expect("decode should succeed"); @@ -2400,6 +2487,7 @@ mod tests { assert_eq!(decoded.request_id, 9); if let ReplicationMessageBody::FetchRequest(req) = decoded.body { assert_eq!(req.key, [0x40; 32]); + assert!(req.busy_reply); } else { panic!("expected FetchRequest"); } @@ -2471,6 +2559,283 @@ mod tests { } } + #[test] + fn fetch_response_busy_roundtrip() { + let msg = ReplicationMessage { + request_id: 13, + body: ReplicationMessageBody::FetchResponse(FetchResponse::Busy { key: [0x71; 32] }), + }; + let encoded = msg.encode().expect("encode should succeed"); + let decoded = ReplicationMessage::decode(&encoded).expect("decode should succeed"); + + assert_eq!(decoded.request_id, 13); + if let ReplicationMessageBody::FetchResponse(FetchResponse::Busy { key }) = decoded.body { + assert_eq!(key, [0x71; 32]); + } else { + panic!("expected FetchResponse::Busy"); + } + } + + // -- the fetch busy reply across versions ------------------------------- + // + // `Legacy*` mirror the fetch types and the envelope as every release + // before the busy reply encoded them. The seven unit variants stand in for + // the body variants before `FetchRequest`; only their count matters, since + // postcard writes an enum as its variant index followed by the payload. + + #[derive(Serialize, Deserialize)] + struct LegacyFetchRequest { + key: XorName, + } + + #[derive(Serialize, Deserialize)] + enum LegacyFetchResponse { + Success { + key: XorName, + #[serde(with = "serde_bytes")] + data: Vec, + }, + NotFound { + key: XorName, + }, + Error { + key: XorName, + reason: String, + }, + } + + #[derive(Serialize, Deserialize)] + enum LegacyBody { + V0, + V1, + V2, + V3, + V4, + V5, + V6, + FetchRequest(LegacyFetchRequest), + FetchResponse(LegacyFetchResponse), + } + + #[derive(Serialize, Deserialize)] + struct LegacyMessage { + request_id: u64, + body: LegacyBody, + } + + fn legacy_fetch_request_bytes(key: XorName) -> Vec { + postcard::to_stdvec(&LegacyMessage { + request_id: 21, + body: LegacyBody::FetchRequest(LegacyFetchRequest { key }), + }) + .expect("encode") + } + + /// The mirror must match the real layout, or every test below proves + /// nothing. A current request is an older one's bytes exactly, plus the + /// capability trailer when it asks for a busy reply. + #[test] + fn legacy_mirror_matches_the_current_layout() { + let legacy = legacy_fetch_request_bytes([0x22; 32]); + let plain = ReplicationMessage { + request_id: 21, + body: ReplicationMessageBody::FetchRequest(FetchRequest { + key: [0x22; 32], + busy_reply: false, + }), + } + .encode() + .expect("encode"); + assert_eq!(plain, legacy); + + let asking = ReplicationMessage { + request_id: 21, + body: ReplicationMessageBody::FetchRequest(FetchRequest { + key: [0x22; 32], + busy_reply: true, + }), + } + .encode() + .expect("encode"); + let mut expected = legacy; + expected.extend_from_slice(&FETCH_CAPABILITY_MAGIC); + expected.push(FETCH_CAPABILITY_BUSY_REPLY); + assert_eq!(asking, expected); + } + + /// An older responder decodes a current request: the trailer comes after + /// the message, which `from_bytes` never reads, so it sees the key and + /// keeps refusing silently. + #[test] + fn older_responder_decodes_a_busy_aware_request() { + let encoded = ReplicationMessage { + request_id: 22, + body: ReplicationMessageBody::FetchRequest(FetchRequest { + key: [0x23; 32], + busy_reply: true, + }), + } + .encode() + .expect("encode"); + let decoded: LegacyMessage = postcard::from_bytes(&encoded).expect("older decoder accepts"); + assert_eq!(decoded.request_id, 22); + match decoded.body { + LegacyBody::FetchRequest(LegacyFetchRequest { key }) => assert_eq!(key, [0x23; 32]), + _ => panic!("expected the legacy FetchRequest variant"), + } + } + + /// A current responder decodes an older requester's request, which has no + /// trailer, as one that did not ask for a busy reply. + #[test] + fn older_request_decodes_without_busy_reply() { + let decoded = ReplicationMessage::decode(&legacy_fetch_request_bytes([0x24; 32])) + .expect("current decoder accepts an older request"); + match decoded.body { + ReplicationMessageBody::FetchRequest(FetchRequest { key, busy_reply }) => { + assert_eq!(key, [0x24; 32]); + assert!(!busy_reply); + } + _ => panic!("expected FetchRequest"), + } + } + + /// Only the magic followed by a flags byte with the busy bit set asks for a + /// busy reply. Unknown flag bits and bytes after the flags are reserved and + /// ignored; anything else after the message reads as no capability, as it + /// always has. + #[test] + fn only_the_trailer_with_the_busy_bit_asks_for_a_busy_reply() { + let magic = FETCH_CAPABILITY_MAGIC.to_vec(); + let with = |tail: &[u8]| { + let mut t = magic.clone(); + t.extend_from_slice(tail); + t + }; + let base = legacy_fetch_request_bytes([0x25; 32]); + for (trailer, expected) in [ + (vec![], false), + (with(&[FETCH_CAPABILITY_BUSY_REPLY]), true), + (with(&[0]), false), + (with(&[0b1111_1111]), true), + (with(&[0b1111_1110]), false), + (with(&[FETCH_CAPABILITY_BUSY_REPLY, 0xAA, 0xBB]), true), + (magic.clone(), false), + (magic.get(..3).expect("short magic").to_vec(), false), + ( + vec![b'a', b'n', b't', b'g', FETCH_CAPABILITY_BUSY_REPLY], + false, + ), + (vec![FETCH_CAPABILITY_BUSY_REPLY], false), + (vec![0u8; 64], false), + ] { + let mut bytes = base.clone(); + bytes.extend_from_slice(&trailer); + let decoded = ReplicationMessage::decode(&bytes).expect("decodes"); + match decoded.body { + ReplicationMessageBody::FetchRequest(req) => { + assert_eq!(req.key, [0x25; 32]); + assert_eq!(req.busy_reply, expected, "trailer {trailer:?}"); + } + _ => panic!("expected FetchRequest"), + } + } + } + + /// The trailer means something only after a fetch request. After any other + /// body it is ignored exactly like every other trailing byte. + #[test] + fn other_bodies_ignore_a_capability_trailer() { + let mut bytes = ReplicationMessage { + request_id: 25, + body: ReplicationMessageBody::FetchResponse(FetchResponse::NotFound { + key: [0x29; 32], + }), + } + .encode() + .expect("encode"); + bytes.extend_from_slice(&FETCH_CAPABILITY_MAGIC); + bytes.push(FETCH_CAPABILITY_BUSY_REPLY); + let decoded = ReplicationMessage::decode(&bytes).expect("decodes"); + assert_eq!(decoded.request_id, 25); + assert!(matches!( + decoded.body, + ReplicationMessageBody::FetchResponse(FetchResponse::NotFound { key }) if key == [0x29; 32] + )); + } + + /// A request cut short inside the key is still malformed. + #[test] + fn a_truncated_key_is_still_rejected() { + let bytes = legacy_fetch_request_bytes([0x26; 32]); + let truncated = bytes.get(..bytes.len() - 1).expect("non-empty"); + assert!(ReplicationMessage::decode(truncated).is_err()); + } + + /// The pre-existing answers keep their discriminants, so an older requester + /// still decodes every answer it is ever sent. + #[test] + fn older_requester_decodes_the_unchanged_answers() { + for (body, check) in [ + ( + FetchResponse::Success { + key: [0x27; 32], + data: vec![1, 2, 3], + }, + 0u8, + ), + (FetchResponse::NotFound { key: [0x27; 32] }, 1), + ( + FetchResponse::Error { + key: [0x27; 32], + reason: "io".to_string(), + }, + 2, + ), + ] { + let encoded = ReplicationMessage { + request_id: 23, + body: ReplicationMessageBody::FetchResponse(body), + } + .encode() + .expect("encode"); + let decoded: LegacyMessage = + postcard::from_bytes(&encoded).expect("older decoder accepts"); + let got = match decoded.body { + LegacyBody::FetchResponse(LegacyFetchResponse::Success { key, data }) => { + assert_eq!(key, [0x27; 32]); + assert_eq!(data, vec![1, 2, 3]); + 0 + } + LegacyBody::FetchResponse(LegacyFetchResponse::NotFound { key }) => { + assert_eq!(key, [0x27; 32]); + 1 + } + LegacyBody::FetchResponse(LegacyFetchResponse::Error { key, reason }) => { + assert_eq!(key, [0x27; 32]); + assert_eq!(reason, "io"); + 2 + } + _ => panic!("expected a legacy FetchResponse"), + }; + assert_eq!(got, check); + } + } + + /// Why the responder gates `Busy` on the request's flag: an older requester + /// cannot decode it, and an undecodable answer costs the responder a trust + /// event at that requester. + #[test] + fn older_requester_cannot_decode_busy() { + let encoded = ReplicationMessage { + request_id: 24, + body: ReplicationMessageBody::FetchResponse(FetchResponse::Busy { key: [0x28; 32] }), + } + .encode() + .expect("encode"); + assert!(postcard::from_bytes::(&encoded).is_err()); + } + // === Audit roundtrips === #[test] @@ -2620,7 +2985,10 @@ mod tests { // fetched chunk needs. let mut core_body = ReplicationMessage { request_id: 1, - body: ReplicationMessageBody::FetchRequest(FetchRequest { key: [0u8; 32] }), + body: ReplicationMessageBody::FetchRequest(FetchRequest { + key: [0u8; 32], + busy_reply: false, + }), } .encode() .expect("a small core body encodes"); diff --git a/src/replication/scheduling.rs b/src/replication/scheduling.rs index e296606e..5ce83aa3 100644 --- a/src/replication/scheduling.rs +++ b/src/replication/scheduling.rs @@ -1311,14 +1311,29 @@ impl ReplicationQueues { /// Returns the next source peer if one is available, or `None` if all /// sources have been exhausted. pub fn retry_fetch(&mut self, key: &XorName) -> Option { + self.retry_fetch_with(key, |untried| untried.first().copied()) + } + + /// Like [`Self::retry_fetch`], with `choose` picking the next source from + /// the untried ones, given in source-preference order. A pick that is not + /// among them is ignored and the first untried source is used instead, so + /// `choose` can reorder the untried sources but never revive a tried one. + pub fn retry_fetch_with(&mut self, key: &XorName, choose: F) -> Option + where + F: FnOnce(&[PeerId]) -> Option, + { let entry = self.in_flight_fetch.get_mut(key)?; entry.tried.insert(entry.source); - let next = entry + let untried: Vec = entry .all_sources .iter() - .find(|p| !entry.tried.contains(p)) - .copied(); + .filter(|p| !entry.tried.contains(p)) + .copied() + .collect(); + let next = choose(&untried) + .filter(|pick| untried.contains(pick)) + .or_else(|| untried.first().copied()); if let Some(next_peer) = next { entry.source = next_peer; diff --git a/tests/e2e/fetch_busy_ab.rs b/tests/e2e/fetch_busy_ab.rs new file mode 100644 index 00000000..739bbb10 --- /dev/null +++ b/tests/e2e/fetch_busy_ab.rs @@ -0,0 +1,299 @@ +//! Opt-in measurement, not a pass/fail gate: how long one node takes to fetch +//! a batch of chunks whose holders it over-asks, and what that costs the +//! holders in the requester's trust table. +//! +//! The requester fetches every chunk with the same three holders in the same +//! order, so its fetch worker, one slot per hardware thread, sends more than a +//! holder's two-request share to the first holder at once. That is the +//! over-share the busy reply answers. The file uses only test hooks that exist +//! on `main` too, so the same file measures a build without the busy reply and +//! a build with it. +//! +//! A second measurement, [`fetch_busy_ab_pool`], has every requester stay +//! within its share, as #252 makes them, and fills the first holder's +//! twelve-slot pool instead: the refusals that remain after #252. +//! +//! Run with `ANT_FETCH_BUSY_AB=1 cargo test --test e2e --features test-utils +//! fetch_busy_ab -- --nocapture --test-threads=1`. Without the variable both +//! return at once. The pool measurement uses the test-only fetch serve gate, +//! which a build without the busy reply needs patched in to compare. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use super::testnet::TestNetworkConfig; +use super::TestHarness; +use ant_node::client::compute_address; +use ant_node::ReplicationConfig; +use serial_test::serial; +use std::time::{Duration, Instant}; + +const REQUESTER_INDEX: usize = 4; +const HOLDER_INDICES: [usize; 3] = [1, 2, 3]; +const CHUNKS: usize = 48; +const CHUNK_LEN: usize = 256 * 1024; +const FILL_DEADLINE: Duration = Duration::from_mins(15); +const POLL: Duration = Duration::from_millis(100); + +#[tokio::test] +#[serial] +async fn fetch_busy_ab() { + if std::env::var("ANT_FETCH_BUSY_AB").is_err() { + return; + } + let harness = TestHarness::setup_minimal().await.expect("setup"); + harness.warmup_dht().await.expect("warmup"); + + let holders: Vec<_> = HOLDER_INDICES + .iter() + .map(|&i| { + *harness + .test_node(i) + .and_then(|n| n.p2p_node.as_ref()) + .expect("holder p2p") + .peer_id() + }) + .collect(); + + let mut keys = Vec::with_capacity(CHUNKS); + for c in 0..CHUNKS { + let content: Vec = (0..CHUNK_LEN) + .map(|i| u8::try_from((i * 31 + c * 7) % 251).unwrap()) + .collect(); + let key = compute_address(&content); + for &h in &HOLDER_INDICES { + harness + .test_node(h) + .and_then(|n| n.ant_protocol.as_ref()) + .expect("holder protocol") + .storage() + .put(&key, &content) + .await + .expect("put"); + } + keys.push(key); + } + + let requester = harness.test_node(REQUESTER_INDEX).expect("requester"); + let p2p = requester.p2p_node.as_ref().expect("requester p2p"); + let engine = requester + .replication_engine + .as_ref() + .expect("requester engine"); + let storage = requester + .ant_protocol + .as_ref() + .expect("requester protocol") + .storage(); + + let trust_before: Vec = holders.iter().map(|h| p2p.peer_trust(h)).collect(); + let started = Instant::now(); + let mut enqueued = 0usize; + for key in &keys { + if engine.enqueue_fetch_for_test(*key, holders.clone()).await { + enqueued += 1; + } + } + + let mut stored = 0usize; + while started.elapsed() < FILL_DEADLINE { + stored = 0; + for key in &keys { + if storage.exists(key).unwrap_or(false) { + stored += 1; + } + } + if stored == CHUNKS { + break; + } + tokio::time::sleep(POLL).await; + } + let elapsed = started.elapsed(); + let trust_after: Vec = holders.iter().map(|h| p2p.peer_trust(h)).collect(); + + println!( + "FETCH_BUSY_AB slots={} chunks={CHUNKS} enqueued={enqueued} stored={stored} \ + elapsed_ms={} trust_before={trust_before:?} trust_after={trust_after:?}", + std::thread::available_parallelism().map_or(0, std::num::NonZero::get), + elapsed.as_millis(), + ); + + harness.teardown().await.expect("teardown"); +} + +/// Nodes in the pool measurement's network. +const POOL_NODES: usize = 25; +/// A close group this wide makes every node responsible for every key, so +/// each requester stores what it fetches. +const POOL_CLOSE_GROUP: usize = 24; +/// The holders, in the source order every chunk uses. +const POOL_HOLDERS: [usize; 5] = [3, 4, 5, 6, 7]; +/// Chunks per requester: its per-peer share, so it never over-asks a holder. +const POOL_KEYS_PER_REQUESTER: usize = 2; +const POOL_CHUNK_LEN: usize = 64 * 1024; +/// How long the first holder keeps its pool full, once it is full, before it +/// serves what it holds. +const POOL_GATE_HOLD: Duration = Duration::from_secs(2); +/// A holder's fetch admission pool: `FETCH_RESPONDER_MAX_OUTSTANDING`. +const POOL_SLOTS: usize = 12; +/// Bound on reaching the precondition before measuring. +const POOL_SETUP_DEADLINE: Duration = Duration::from_secs(60); + +#[tokio::test] +#[serial] +async fn fetch_busy_ab_pool() { + if std::env::var("ANT_FETCH_BUSY_AB").is_err() { + return; + } + let config = TestNetworkConfig { + node_count: POOL_NODES, + replication_config: Some(ReplicationConfig { + close_group_size: POOL_CLOSE_GROUP, + ..ReplicationConfig::default() + }), + ..TestNetworkConfig::default() + }; + let harness = TestHarness::setup_with_config(config).await.expect("setup"); + harness.warmup_dht().await.expect("warmup"); + + let holders: Vec<_> = POOL_HOLDERS + .iter() + .map(|&i| { + *harness + .test_node(i) + .and_then(|n| n.p2p_node.as_ref()) + .expect("holder p2p") + .peer_id() + }) + .collect(); + let requesters: Vec = (0..POOL_NODES) + .filter(|i| !POOL_HOLDERS.contains(i)) + .collect(); + + let mut keys_by_requester = Vec::with_capacity(requesters.len()); + for (r, _) in requesters.iter().enumerate() { + let mut keys = Vec::with_capacity(POOL_KEYS_PER_REQUESTER); + for k in 0..POOL_KEYS_PER_REQUESTER { + let content: Vec = (0..POOL_CHUNK_LEN) + .map(|i| u8::try_from((i * 13 + r * 101 + k * 7) % 251).unwrap()) + .collect(); + let key = compute_address(&content); + for &h in &POOL_HOLDERS { + harness + .test_node(h) + .and_then(|n| n.ant_protocol.as_ref()) + .expect("holder protocol") + .storage() + .put(&key, &content) + .await + .expect("put"); + } + keys.push(key); + } + keys_by_requester.push(keys); + } + + let first_holder = POOL_HOLDERS + .first() + .and_then(|&i| harness.test_node(i)) + .and_then(|n| n.replication_engine.as_ref()) + .expect("first holder engine"); + first_holder.close_fetch_serve_gate_for_test(); + + let trust_of = |i: usize| -> Vec { + let p2p = harness + .test_node(i) + .and_then(|n| n.p2p_node.as_ref()) + .expect("requester p2p"); + holders.iter().map(|h| p2p.peer_trust(h)).collect() + }; + let trust_before: Vec> = requesters.iter().map(|&i| trust_of(i)).collect(); + + let total = requesters.len() * POOL_KEYS_PER_REQUESTER; + let refusals_before = first_holder.fetch_pool_full_refusals_for_test(); + let started = Instant::now(); + let mut enqueued = 0usize; + for (&i, keys) in requesters.iter().zip(&keys_by_requester) { + let engine = harness + .test_node(i) + .and_then(|n| n.replication_engine.as_ref()) + .expect("requester engine"); + for key in keys { + if engine.enqueue_fetch_for_test(*key, holders.clone()).await { + enqueued += 1; + } + } + } + assert_eq!( + enqueued, total, + "every key must be queued for the measurement" + ); + + // The measurement is only about a full pool: wait until the first holder + // holds a whole pool's worth, keep it full, and require that it refused + // the rest before letting it serve. + let setup_deadline = Instant::now() + POOL_SETUP_DEADLINE; + while first_holder.fetch_admitted_for_test() < POOL_SLOTS { + assert!( + Instant::now() < setup_deadline, + "the first holder never filled its pool: {} admitted", + first_holder.fetch_admitted_for_test() + ); + tokio::time::sleep(POLL).await; + } + tokio::time::sleep(POOL_GATE_HOLD).await; + let refused_while_full = first_holder + .fetch_pool_full_refusals_for_test() + .saturating_sub(refusals_before); + assert!( + refused_while_full > 0, + "the first holder refused nothing for a full pool while it was full" + ); + first_holder.open_fetch_serve_gate_for_test(); + let mut stored = 0usize; + while started.elapsed() < FILL_DEADLINE { + stored = 0; + for (&i, keys) in requesters.iter().zip(&keys_by_requester) { + let storage = harness + .test_node(i) + .and_then(|n| n.ant_protocol.as_ref()) + .expect("requester protocol") + .storage(); + for key in keys { + if storage.exists(key).unwrap_or(false) { + stored += 1; + } + } + } + if stored == total { + break; + } + tokio::time::sleep(POLL).await; + } + let elapsed = started.elapsed(); + + // Trust lost per holder, summed over requesters. Each connection-timeout + // charge from a neutral 0.5 costs about 0.06. + let trust_after: Vec> = requesters.iter().map(|&i| trust_of(i)).collect(); + let drops = |h: usize| { + trust_before + .iter() + .zip(&trust_after) + .filter_map(move |(b, a)| Some(b.get(h)? - a.get(h)?)) + }; + let lost: Vec = (0..holders.len()) + .map(|h| drops(h).map(|d| d.max(0.0)).sum()) + .collect(); + let charged: Vec = (0..holders.len()) + .map(|h| drops(h).filter(|d| *d > 0.0).count()) + .collect(); + + println!( + "FETCH_BUSY_AB_POOL requesters={} keys={total} enqueued={enqueued} stored={stored} \ + refused_while_full={refused_while_full} elapsed_ms={} \ + requesters_charging_each_holder={charged:?} trust_lost_per_holder={lost:?}", + requesters.len(), + elapsed.as_millis(), + ); + + harness.teardown().await.expect("teardown"); +} diff --git a/tests/e2e/fetch_busy_reply.rs b/tests/e2e/fetch_busy_reply.rs new file mode 100644 index 00000000..5419888f --- /dev/null +++ b/tests/e2e/fetch_busy_reply.rs @@ -0,0 +1,266 @@ +//! A fetch responder answers `Busy` to a refused request that asked for it, +//! stays silent to one that did not, and a requester takes a busy answer +//! without charging the holder. +//! +//! The unit tests pin the wire format, the limiter, the wait and the queue +//! transitions. What they cannot show is the whole path between two real +//! nodes: the capability trailer surviving the transport, admission refusing +//! past the per-peer share, the reply riding the request/response channel +//! back, and the requester's fetch worker acting on it. +//! +//! The holder's fetch serve gate makes every case deterministic. With the gate +//! closed, an admitted request is held before it is served, keeping its share, +//! so of three concurrent requests from one peer exactly two are admitted and +//! exactly one is refused, whatever order they arrive in. + +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +use super::TestHarness; +use ant_node::client::compute_address; +use ant_node::replication::config::REPLICATION_PROTOCOL_ID; +use ant_node::replication::fetch_busy::{FETCH_BUSY_WAIT_MAX, FETCH_BUSY_WAIT_MIN}; +use ant_node::replication::protocol::{ + FetchRequest, FetchResponse, ReplicationMessage, ReplicationMessageBody, +}; +use futures::stream::{FuturesUnordered, StreamExt}; +use saorsa_core::identity::PeerId; +use saorsa_core::P2PNode; +use serial_test::serial; +use std::time::{Duration, Instant}; + +const HOLDER_INDEX: usize = 3; +const REQUESTER_INDEX: usize = 4; +/// Requests admitted per requesting peer: the responder's per-peer share. +const SHARE: usize = 2; +/// Long enough that a held request is never timed out before the test opens +/// the gate; a timeout would charge the holder and spoil the trust checks. +const HELD_TIMEOUT: Duration = Duration::from_secs(60); +/// How long a request that should be refused silently is given. +const SILENCE_TIMEOUT: Duration = Duration::from_secs(3); +/// Bound on how long anything that must happen promptly may take. +const PROMPTLY: Duration = Duration::from_secs(10); +const POLL: Duration = Duration::from_millis(50); +const CHUNK_LEN: usize = 64 * 1024; + +#[derive(Debug, Clone, PartialEq, Eq)] +enum Answer { + Success, + Busy, + Silent, + Other, +} + +async fn fetch( + sender: &P2PNode, + holder: &PeerId, + key: [u8; 32], + busy_reply: bool, + timeout: Duration, +) -> Answer { + let msg = ReplicationMessage { + request_id: rand_request_id(), + body: ReplicationMessageBody::FetchRequest(FetchRequest { key, busy_reply }), + }; + let Ok(response) = sender + .send_request( + holder, + REPLICATION_PROTOCOL_ID, + msg.encode().expect("encode"), + timeout, + ) + .await + else { + return Answer::Silent; + }; + match ReplicationMessage::decode(&response.data).map(|m| m.body) { + Ok(ReplicationMessageBody::FetchResponse(FetchResponse::Success { key: k, .. })) + if k == key => + { + Answer::Success + } + Ok(ReplicationMessageBody::FetchResponse(FetchResponse::Busy { key: k })) if k == key => { + Answer::Busy + } + _ => Answer::Other, + } +} + +fn rand_request_id() -> u64 { + use std::sync::atomic::{AtomicU64, Ordering}; + static NEXT: AtomicU64 = AtomicU64::new(9_000); + NEXT.fetch_add(1, Ordering::Relaxed) +} + +struct Setup { + harness: TestHarness, + key: [u8; 32], + holder: PeerId, +} + +async fn setup() -> Setup { + let harness = TestHarness::setup_minimal().await.expect("setup"); + harness.warmup_dht().await.expect("warmup"); + let holder_node = harness.test_node(HOLDER_INDEX).expect("holder"); + let content: Vec = (0..CHUNK_LEN) + .map(|i| u8::try_from(i % 251).unwrap()) + .collect(); + let key = compute_address(&content); + holder_node + .ant_protocol + .as_ref() + .expect("holder protocol") + .storage() + .put(&key, &content) + .await + .expect("put"); + let holder = *holder_node.p2p_node.as_ref().expect("holder p2p").peer_id(); + holder_node + .replication_engine + .as_ref() + .expect("holder engine") + .close_fetch_serve_gate_for_test(); + Setup { + harness, + key, + holder, + } +} + +fn open_gate(harness: &TestHarness) { + harness + .test_node(HOLDER_INDEX) + .and_then(|n| n.replication_engine.as_ref()) + .expect("holder engine") + .open_fetch_serve_gate_for_test(); +} + +/// Of `SHARE + 1` concurrent requests that ask for busy replies, the one the +/// holder refuses is answered `Busy` at once, the held ones are served once +/// the gate opens, and a request that does not ask is refused silently. (The +/// per-peer and global busy allowances are pinned exactly by the limiter's +/// unit tests; over a real network their timing is not controllable.) +#[tokio::test] +#[serial] +async fn a_refused_fetch_is_answered_busy_only_when_it_asks() { + let Setup { + harness, + key, + holder, + } = setup().await; + let requester = harness + .test_node(REQUESTER_INDEX) + .and_then(|n| n.p2p_node.as_ref()) + .expect("requester p2p"); + + let mut pending: FuturesUnordered<_> = (0..=SHARE) + .map(|_| fetch(requester, &holder, key, true, HELD_TIMEOUT)) + .collect(); + let started = Instant::now(); + let first = tokio::time::timeout(PROMPTLY, pending.next()) + .await + .expect("the refused request is answered promptly") + .expect("a request"); + assert_eq!(first, Answer::Busy, "the refused request is told busy"); + assert!(started.elapsed() < PROMPTLY); + + // The share is still held, so a request that does not ask for busy + // replies is refused, and silently, exactly as before busy replies. + assert_eq!( + fetch(requester, &holder, key, false, SILENCE_TIMEOUT).await, + Answer::Silent + ); + + open_gate(&harness); + let rest: Vec = tokio::time::timeout(PROMPTLY, pending.collect()) + .await + .expect("held requests are served once the gate opens"); + assert_eq!(rest, vec![Answer::Success; SHARE]); + + harness.teardown().await.expect("teardown"); +} + +/// The requester's own fetch worker, refused by a holder whose share is full, +/// gets a busy answer, puts that holder last for later fetches, waits before +/// giving the key up, and charges the holder nothing. Before busy replies the +/// same refusal was silent and cost the holder a connection-timeout trust +/// event at the requester. +#[tokio::test] +#[serial] +async fn the_requester_takes_a_busy_answer_without_charging_the_holder() { + let Setup { + harness, + key, + holder, + } = setup().await; + let requester_node = harness.test_node(REQUESTER_INDEX).expect("requester"); + let requester = requester_node.p2p_node.as_ref().expect("requester p2p"); + let engine = requester_node + .replication_engine + .as_ref() + .expect("requester engine"); + let trust_before = requester.peer_trust(&holder); + + // Fill this requester's share at the holder: two held, one told busy. + let mut held: FuturesUnordered<_> = (0..=SHARE) + .map(|_| fetch(requester, &holder, key, true, HELD_TIMEOUT)) + .collect(); + let first = tokio::time::timeout(PROMPTLY, held.next()) + .await + .expect("the refused request is answered promptly") + .expect("a request"); + assert_eq!(first, Answer::Busy); + assert!(!engine.fetch_source_cooling_for_test(&holder)); + + // Trust is sampled throughout, so a charge cannot hide behind a later + // unrelated success: any dip below the starting value fails. + let mut lowest_trust = trust_before; + let sample = |lowest: &mut f64| *lowest = lowest.min(requester.peer_trust(&holder)); + + // The fetch worker's request is refused too, and told busy. + assert!(engine.enqueue_fetch_for_test(key, vec![holder]).await); + let deadline = Instant::now() + PROMPTLY; + while !engine.fetch_source_cooling_for_test(&holder) { + assert!( + Instant::now() < deadline, + "the fetch worker never recorded the busy answer" + ); + sample(&mut lowest_trust); + tokio::time::sleep(POLL).await; + } + + // The attempt holds its slot through the busy wait, then gives the key up + // (its only source refused), exactly as a timeout would have, sooner. The + // wait is timed from when the busy answer was recorded, just before it + // starts, to when this fetch attempt is first seen gone, watched with the + // gate still closed so nothing else delays the observation. + let busy_at = engine + .fetch_source_last_busy_for_test(&holder) + .expect("the busy answer was recorded"); + let deadline = Instant::now() + FETCH_BUSY_WAIT_MAX + PROMPTLY; + while engine.fetch_queued_or_in_flight_for_test(&key).await { + assert!( + Instant::now() < deadline, + "the fetch attempt outlived its busy wait" + ); + sample(&mut lowest_trust); + tokio::time::sleep(POLL).await; + } + let attempt_lasted = busy_at.elapsed(); + assert!( + attempt_lasted >= FETCH_BUSY_WAIT_MIN, + "the key was given up {attempt_lasted:?} after the busy answer, before the busy wait" + ); + + open_gate(&harness); + let rest: Vec = tokio::time::timeout(PROMPTLY, held.collect()) + .await + .expect("held requests are served once the gate opens"); + assert_eq!(rest, vec![Answer::Success; SHARE]); + sample(&mut lowest_trust); + assert!( + lowest_trust >= trust_before, + "a busy answer must not cost the holder trust: {trust_before} fell to {lowest_trust}" + ); + + harness.teardown().await.expect("teardown"); +} diff --git a/tests/e2e/mod.rs b/tests/e2e/mod.rs index ec2a74c7..e375b108 100644 --- a/tests/e2e/mod.rs +++ b/tests/e2e/mod.rs @@ -69,6 +69,12 @@ mod security_attacks; #[cfg(test)] mod subtree_audit_testnet; +#[cfg(test)] +mod fetch_busy_ab; + +#[cfg(test)] +mod fetch_busy_reply; + #[cfg(test)] mod fetch_local_write_guard; diff --git a/tests/e2e/replication.rs b/tests/e2e/replication.rs index 6ae3fb5b..06855160 100644 --- a/tests/e2e/replication.rs +++ b/tests/e2e/replication.rs @@ -1203,7 +1203,10 @@ async fn test_fetch_request_returns_record() { .expect("put"); // Send fetch request from B to A - let request = FetchRequest { key: address }; + let request = FetchRequest { + key: address, + busy_reply: false, + }; let msg = ReplicationMessage { request_id: 99, body: ReplicationMessageBody::FetchRequest(request), @@ -2030,7 +2033,10 @@ async fn test_fetch_not_found() { let peer_a = *p2p_a.peer_id(); let missing_key = [0xEE; 32]; - let request = FetchRequest { key: missing_key }; + let request = FetchRequest { + key: missing_key, + busy_reply: false, + }; let msg = ReplicationMessage { request_id: 77, body: ReplicationMessageBody::FetchRequest(request), @@ -2366,7 +2372,10 @@ async fn test_fetch_returns_error_for_corrupt_key() { let peer_a = *p2p_a.peer_id(); let fake_key = [0x00; 32]; - let request = FetchRequest { key: fake_key }; + let request = FetchRequest { + key: fake_key, + busy_reply: false, + }; let msg = ReplicationMessage { request_id: 4000, body: ReplicationMessageBody::FetchRequest(request), From 6abba5a79a7f04be40e11aa12c4a638148989ace Mon Sep 17 00:00:00 2001 From: grumbach Date: Fri, 16 Oct 2026 15:57:36 +0900 Subject: [PATCH 2/2] fix(replication): log the cooldown a busy holder actually gets `SourceBusyTracker::record_busy` never shortens a cooldown that is already running, but it returned the newly computed cooldown, so when an earlier, longer one was still running the requester's debug line reported less than the holder was really put last for. It now returns what is left of the cooldown in force. ant-node only logs the return value, so scheduling and wire behaviour do not change; callers of the public method now get the remaining cooldown in force. The never-shortened test now checks the value too. The same debug line said the holder was passed over "for new keys", but the cooldown applies both when a key starts and when it moves on to another holder. It now says the holder is put last. ADR-0025: - the context adds the refusals #252 leaves at holders it currently counts as silent, and says #252's ADR defers a busy reply rather than naming it as the remaining piece; - the trade-off on holders that answer busy to everything says they are put behind every holder that is not cooling, on retries too, and names the `fetch_busy_reply_send_pool_full` counter in full; - the requester section states the cooldown as 1, 2, 4 and 8 times the jittered wait drawn for each busy answer, capped at 60 s, says that a later busy answer never shortens a running cooldown, and gives the table's eviction order when every holder in it is cooling; - the request-rate trade-off is given for any fetch timeout, not only the default 30 s; - the requester end-to-end test is described as filling that requester's own two-request share at the holder; - `fetch_busy_sources_tracked` is listed as a gauge, apart from the cumulative counters; - the positive consequence notes that the busy wait can be cut short by the attempt's own fetch timeout; - the rejected retry-rotation option states its cost as the mechanism (one request per round trip instead of one per fetch timeout) rather than an estimate; - the order section records that this change and #252 conflict and have not been built or measured together yet; - the real-node measurements are rerun on upstream main a6e3388e, and the local simulation, which is not published, is named only as design input rather than offered as validation. --- docs/adr/ADR-0025-fetch-busy-reply.md | 100 ++++++++++++++------------ src/replication/fetch_busy.rs | 16 +++-- src/replication/mod.rs | 2 +- 3 files changed, 65 insertions(+), 53 deletions(-) diff --git a/docs/adr/ADR-0025-fetch-busy-reply.md b/docs/adr/ADR-0025-fetch-busy-reply.md index 091d97c8..f6f5a8c3 100644 --- a/docs/adr/ADR-0025-fetch-busy-reply.md +++ b/docs/adr/ADR-0025-fetch-busy-reply.md @@ -33,8 +33,11 @@ refusals, 95.2% of them to the new nodes. ant-node #252 removes most of the per-peer over-share by keeping a requester's own fetches within each answering holder's share. It cannot remove every refusal: a request the requester has already timed out on still holds its share at the responder until the responder sheds it, the -twelve-slot pool can be full of other peers' requests, and older requesters keep over-sharing -until they upgrade. Its ADR names a busy reply, which is a wire change, as the remaining piece. +twelve-slot pool can be full of other peers' requests, a holder #252 currently counts as silent +may be sent requests past its share until an answer clears the mark or the mark expires and the +holder is probed under the share again, and older requesters keep over-sharing until they +upgrade. #252's description leaves the first case to a busy reply, and its ADR defers that reply +as a wire change with its own rollout. ## Decision Drivers @@ -61,8 +64,9 @@ until they upgrade. Its ADR names a busy reply, which is a wire change, as the r 4. **A capability trailer after the whole message, and an appended `FetchResponse::Busy` variant (chosen).** 5. **Busy with requester-side retry rotation and budgets.** Reviewed and rejected: hopping on every - busy answer multiplies requests about tenfold in the worst case, competes with `tried` state, - and charging at budget expiry punishes honest overloaded holders once per key per cycle. + busy answer turns each refusal into an immediate new request, so a requester whose holders all + refuse asks once per round trip instead of once per fetch timeout; it competes with `tried` + state; and charging at budget expiry punishes honest overloaded holders once per key per cycle. ## Decision @@ -98,24 +102,28 @@ holder exactly as after a timeout. `SourceBusyTracker` remembers the busy holder and a failed key retries, on the first holder in preference order that is not cooling down, falling back to the first when all are. It reorders and never withholds, so it lowers how often a busy holder is asked but is not a hard per-holder bound; the hard bound is the busy wait per fetch -slot. The cooldown is the full jittered wait, doubling with each further busy answer before the -holder next serves this requester, up to 60 s, and is never shortened while it runs; once it runs -out the holder is no longer put last. Serving once clears the holder's record, so its next busy -answer starts the doubling again; a record with no busy answer for 10 minutes does the same. When -the 4096-entry table is full, a holder that is no longer cooling is forgotten first. A `Busy` for any other key is -treated as malformed, like every other key mismatch. +slot. The cooldown is 1, 2, 4, then 8 times the full jittered wait drawn for that busy answer, +the multiplier rising with each further busy answer before the holder next serves this +requester, capped at 60 s, and a later busy answer never shortens a running cooldown; once it +runs out the holder is no longer put last. Serving once clears the holder's record, so its next +busy answer starts the multiplier again; a record with no busy answer for 10 minutes does the +same. When the 4096-entry table is full, a holder that is no longer cooling is forgotten first, +or, if every holder in it is cooling, the one whose cooldown ends soonest. A `Busy` for any +other key is treated as malformed, like every other key mismatch. **Order.** This is merged after #252. #252's per-source share is what removes most per-peer -over-share and bounds how many fetches one requester has open at one holder; without it, a -requester with many fetch slots still over-shares, and most of those refusals exceed the per-peer -busy allowance and stay silent. On rebase, #252's source choice also skips cooling holders. +over-share and bounds how many fetches one requester has open at one answering holder; without +it, a requester with many fetch slots still over-shares, and most of those refusals exceed the +per-peer busy allowance and stay silent. On rebase, #252's source choice also puts cooling holders +last. The two changes conflict in `src/replication/mod.rs` and have not yet been built or measured +together; the gates and the real-node measurements below will be rerun on the rebased change. ## Consequences ### Positive -- When a refused requester receives `Busy`, it moves on after 5-10 s instead of 30 s without - charging the responder. +- When a refused requester receives `Busy`, it moves on after 5-10 s, or sooner if its own fetch + timeout runs out first, instead of 30 s, without charging the responder. - A holder that just refused is not handed this requester's next keys while another holder that is not cooling exists. - Old nodes decode exactly the messages they always did: a request carries 5 bytes after the @@ -130,15 +138,18 @@ busy allowance and stay silent. On rebase, #252's source choice also skips cooli - New outbound traffic at responders: at most 16 signed busy replies a second sustained, about 85 KiB/s, and 48 (about 250 KiB) in any one second, plus that many signatures. Today a refusal sends nothing. Each refused request already cost its sender the same signed message. -- A requester asks a refusing holder up to once per 5 s per fetch slot instead of once per 30 s: - at most `C / 5 s` small requests with `C` hardware-thread fetch slots, against `C / 30 s` today, - in the case where every holder refuses everything. Capping `C` is a separate decision. +- A requester asks a refusing holder up to once per busy wait per fetch slot instead of once per + fetch timeout `T`: at most `C / min(5 s, T)` small requests with `C` hardware-thread fetch + slots, against `C / T` today, in the case where every holder refuses everything; with the + default 30 s timeout, `C / 5 s` against `C / 30 s`. Capping `C` is a separate decision. - Refusals beyond the per-peer busy allowance stay silent and are still charged by saorsa-core. Without #252, a requester that over-shares by more than two requests at once still meets some - silence; the simulation below gives the residual. + silence. - Busy is trust-neutral. A holder that tries to answer busy to everything can still be charged when - a request did not ask, a reply was suppressed, or a scheduled send did not complete; otherwise it - is passed over for new keys for up to 60 s at a time and remains subject to storage audits. + a request did not ask, a reply was suppressed, or a scheduled send did not complete; otherwise, + when this requester starts or retries a key, it is put behind every holder that is not cooling + (when all are cooling, the first is still tried), for up to 60 s at a time, and remains subject + to storage audits. Penalising "never serves" would need a service audit with reserved capacity, which is not part of this decision. - Sixteen identities each spending their per-peer second can use up the global allowance and @@ -149,9 +160,11 @@ busy allowance and stay silent. On rebase, #252's source choice also skips cooli ### Neutral / Operational -- New counters on the `replication responder admission summary (cumulative)` line: +- New fields on the `replication responder admission summary (cumulative)` line: the cumulative + counters `fetch_busy_reply_{not_asked,send_pool_full,rate_limited,peer_in_flight,sent,send_failed,timed_out,cancelled}`, - `fetch_busy_received`, `fetch_busy_cooling_skips` and `fetch_busy_sources_tracked`. + `fetch_busy_received` and `fetch_busy_cooling_skips`, and the gauge `fetch_busy_sources_tracked`, + the current size of the requester's busy-holder table. `not_asked` against the rest is the share of refused requests that did not ask for a busy reply: a proxy, and an upper bound, for how much refused traffic still comes from older versions (a request from a current node can also not ask, through the API or a malformed trailer). @@ -165,7 +178,8 @@ busy allowance and stay silent. On rebase, #252's source choice also skips cooli - an older requester cannot decode `Busy`; - limiter: burst, refill, global cap, no spend on a denial, in-flight limit, bounded table; - the busy wait's range and its cap at the fetch timeout; - - tracker: cooldown, escalation, cap, never shortened, reset, expiry, bounded table; + - tracker: cooldown, escalation, cap, not shortened by a later busy answer, reset, expiry, + bounded table; - a retry passes over a cooling holder but still takes it last; - a busy reply is not counted as served bytes. - End to end on a local network, made deterministic with a test-only fetch serve gate (the same @@ -173,34 +187,26 @@ busy allowance and stay silent. On rebase, #252's source choice also skips cooli - a refused request that asks is told `Busy` at once; - one that does not ask is refused silently; - the held requests are served once the gate opens; - - the requester's own fetch worker, refused by a full holder, records the holder as cooling, - keeps the attempt for at least the minimum busy wait, then gives the key up, and the holder's - trust at the requester never drops. -- Measured on real nodes against upstream main with the same test-only gate, in + - the requester's own fetch worker, refused because its two-request share at the holder is + full, records the holder as cooling, keeps the attempt for at least the minimum busy wait, + then gives the key up, and the holder's trust at the requester never drops. +- Measured on real nodes against upstream main a6e3388e with the same test-only gate, in `tests/e2e/fetch_busy_ab.rs` (opt-in): - 20 requesters each within their two-request share, the first of five holders with its - 12-slot pool full for 2 s (28 refusals in both builds), which is the refusal left after #252: - all 40 chunks in 10.5 s, no holder charged, against 60.4 s with 14 requesters charging the - first holder and 4 the second; - - one requester with 8 fetch slots over-asking three holders, which #252 removes: 39.7 s, 39.4 s - and 36.7 s with only the first holder charged (4 charges), against 60.2 s, 90.3 s and 90.2 s with - up to all three charged. -- A discrete-event simulation of the 177-node join wave and 24 more scenarios (dead holders, - holders that always answer busy, saturated pools, mixed versions, 1-64 fetch slots, - single-holder keys, limiter and wait sensitivity), comparing today, this change, #252-style - capping, and both. It reproduces the real-node run of this change and overstates the cost of - today's silence, so it gives directions, not numbers. In it no key is stranded in any scenario. - With nobody answering busy it behaves exactly as today. With #252-style capping, the busy reply - removes every false charge in the background-load, overload, always-busy and dense scenarios; - what remains elsewhere comes from requests a slow holder sheds unanswered after the fetch timeout - and from requesters that do not ask. Without capping it raises requests per chunk - by up to 12% in some scenarios and lowers it by up to 17% in others (with a 1-2 s wait instead of -5-10 s it rises 16%, which is why the wait is longer), and keys with a single holder go back to -verification about twice as often. + 12-slot pool full for 2 s (26 refusals with this change, 28 on main), one of the refusal + cases left after #252: all 40 chunks in 10.5 s, no holder charged, against 90.3 s with 14, + 9, 9, 4 and 3 requesters charging the five holders (one run each; on the previous main + 8d25f0c3 it was 60.4 s with 14 and 4 requesters charging the first two); + - one requester with 8 fetch slots over-asking three holders, which #252 removes: 38.1 s, 39.2 s + and 37.2 s with only the first holder charged (4 charges), against 90.2 s, 90.2 s and 90.3 s + with all three charged (6 charges on the first, 1 to 4 on the others). +- A local, unpublished discrete-event model of the 177-node join wave and 24 more scenarios + informed the design, including the 5-10 s wait. Its source and raw output are not part of this + change, so its results are not offered as validation. - In production: - `fetch_busy_reply_sent` against `fetch_admission_dropped_*`; - the fall in `connection_timeout` trust events between fleet nodes; - - `fetch_busy_reply_rate_limited` and `send_pool_full` staying small. + - `fetch_busy_reply_rate_limited` and `fetch_busy_reply_send_pool_full` staying small. Revisit the 5-10 s wait and the allowances against measured admission hold times. diff --git a/src/replication/fetch_busy.rs b/src/replication/fetch_busy.rs index e6be386f..b651b304 100644 --- a/src/replication/fetch_busy.rs +++ b/src/replication/fetch_busy.rs @@ -325,8 +325,9 @@ impl SourceBusyTracker { /// Record a busy answer from `source` at `now`. `wait` is the busy wait /// before any cap from the attempt's own timeout. Returns how long the /// source is now put last: `wait` the first time, doubling with each further - /// busy answer before it serves again, up to a minute. A - /// cooldown already running is never shortened. + /// busy answer before it serves again, up to a minute. A cooldown already + /// running is never shortened, so when it outlasts the new one, what is + /// left of it is returned. pub fn record_busy(&self, source: &PeerId, now: Instant, wait: Duration) -> Duration { let mut sources = self.sources.lock(); if sources @@ -363,8 +364,9 @@ impl SourceBusyTracker { let until = now.checked_add(cooldown).unwrap_or(now); record.cooling_until = record.cooling_until.max(until); record.last_busy = now; + let cooling_for = record.cooling_until.saturating_duration_since(now); audit_metrics::set_fetch_busy_sources_tracked(sources.len()); - cooldown + cooling_for } /// `source` served this requester: forget its busy record. @@ -687,8 +689,12 @@ mod tests { tracker.record_busy(&a, t0, Duration::from_secs(10)); } // Cooling until t0 + 60 s now. A later busy answer with a tiny wait - // would end far sooner; the running cooldown stands. - tracker.record_busy(&a, t0 + Duration::from_secs(1), Duration::from_millis(1)); + // would end far sooner; the running cooldown stands, and what is left + // of it is what the caller is told. + assert_eq!( + tracker.record_busy(&a, t0 + Duration::from_secs(1), Duration::from_millis(1)), + Duration::from_secs(59) + ); assert!(tracker.is_cooling(&a, t0 + Duration::from_secs(59))); } diff --git a/src/replication/mod.rs b/src/replication/mod.rs index 7288160e..3d694c2f 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -9660,7 +9660,7 @@ async fn execute_single_fetch( let cooldown = busy_sources.record_busy(&source, Instant::now(), full_wait); debug!( "Fetch: source {source} busy for {}; moving on in {} ms, \ - passing it over for new keys for {} ms", + putting it last for {} ms", hex::encode(key), wait.as_millis(), cooldown.as_millis()