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..f6f5a8c3 --- /dev/null +++ b/docs/adr/ADR-0025-fetch-busy-reply.md @@ -0,0 +1,215 @@ +# 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, 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 + +- 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 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 + +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 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 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, 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 + 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 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. +- 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, + 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 + 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 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` 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). + +## 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, 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 + 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 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 (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 `fetch_busy_reply_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..b651b304 --- /dev/null +++ b/src/replication/fetch_busy.rs @@ -0,0 +1,751 @@ +//! 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, 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 + .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; + let cooling_for = record.cooling_until.saturating_duration_since(now); + audit_metrics::set_fetch_busy_sources_tracked(sources.len()); + cooling_for + } + + /// `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, 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))); + } + + #[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..3d694c2f 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, \ + putting it last 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),