diff --git a/Cargo.lock b/Cargo.lock index 3ee7e5a8..18969f86 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -862,6 +862,7 @@ dependencies = [ "self_encryption", "semver 1.0.28", "serde", + "serde_bytes", "serde_json", "serial_test", "sha2", @@ -883,7 +884,7 @@ dependencies = [ [[package]] name = "ant-protocol" version = "2.4.0" -source = "git+https://github.com/WithAutonomi/ant-protocol?rev=1764950d7af880aa13af679ac8c2d0fcbcde0776#1764950d7af880aa13af679ac8c2d0fcbcde0776" +source = "git+https://github.com/WithAutonomi/ant-protocol?rev=a66ddcfb60d50a9ec07b9aee0d063a730a96b330#a66ddcfb60d50a9ec07b9aee0d063a730a96b330" dependencies = [ "blake3", "bytes", @@ -5324,7 +5325,7 @@ dependencies = [ [[package]] name = "saorsa-core" version = "0.27.4" -source = "git+https://github.com/WithAutonomi/saorsa-core?rev=02dd65fc4df68f326b8d7ed0bd0e3ffb6c29329c#02dd65fc4df68f326b8d7ed0bd0e3ffb6c29329c" +source = "git+https://github.com/WithAutonomi/saorsa-core?rev=53f1fc6fca13109f583dd951ee3eab42d7a37121#53f1fc6fca13109f583dd951ee3eab42d7a37121" dependencies = [ "anyhow", "async-trait", @@ -5392,7 +5393,7 @@ dependencies = [ [[package]] name = "saorsa-transport" version = "0.36.4" -source = "git+https://github.com/WithAutonomi/saorsa-transport?rev=3ed66a9c0880583ceab1e1550dce756e2716d111#3ed66a9c0880583ceab1e1550dce756e2716d111" +source = "git+https://github.com/WithAutonomi/saorsa-transport?rev=1766037ef159d80495515de65f96b3fd8fc94319#1766037ef159d80495515de65f96b3fd8fc94319" dependencies = [ "anyhow", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index f3610c9f..e189a326 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -107,6 +107,9 @@ page_size = "0.6" # Protocol serialization postcard = { version = "1.1.3", features = ["use-std"] } +# Byte-string encoding for chunk payloads in replication messages (same postcard +# wire layout as a u8 sequence, but serialized and sized in one memcpy pass). +serde_bytes = "0.11" bao = "0.13.1" # Shared portable browser profile. The native listener is enabled separately @@ -231,9 +234,9 @@ webrtc-direct = [ [patch.crates-io] saorsa-pqc = { git = "https://github.com/saorsa-labs/saorsa-pqc", rev = "29a2b272c4ee8b72c17102da00033dbfe1ddcd37" } evmlib = { git = "https://github.com/WithAutonomi/evmlib", rev = "ccd65f18bf3750af8b8c3cfe07548bc77e3c368d" } -ant-protocol = { git = "https://github.com/WithAutonomi/ant-protocol", rev = "1764950d7af880aa13af679ac8c2d0fcbcde0776" } -saorsa-core = { git = "https://github.com/WithAutonomi/saorsa-core", rev = "02dd65fc4df68f326b8d7ed0bd0e3ffb6c29329c" } -saorsa-transport = { git = "https://github.com/WithAutonomi/saorsa-transport", rev = "3ed66a9c0880583ceab1e1550dce756e2716d111" } +ant-protocol = { git = "https://github.com/WithAutonomi/ant-protocol", rev = "a66ddcfb60d50a9ec07b9aee0d063a730a96b330" } +saorsa-core = { git = "https://github.com/WithAutonomi/saorsa-core", rev = "53f1fc6fca13109f583dd951ee3eab42d7a37121" } +saorsa-transport = { git = "https://github.com/WithAutonomi/saorsa-transport", rev = "1766037ef159d80495515de65f96b3fd8fc94319" } [profile.release] lto = true diff --git a/docs/adr/ADR-0016-bounded-fresh-offers-and-copy-free-sends.md b/docs/adr/ADR-0016-bounded-fresh-offers-and-copy-free-sends.md new file mode 100644 index 00000000..ba3741c8 --- /dev/null +++ b/docs/adr/ADR-0016-bounded-fresh-offers-and-copy-free-sends.md @@ -0,0 +1,144 @@ +# ADR-0016: Bounded fresh-replication offers and copy-free message sends + +- **Status:** Proposed +- **Date:** 2026-09-22 +- **Decision owners:** +- **Reviewers:** +- **Supersedes:** none +- **Superseded by:** none +- **Related:** [ADR-0003](ADR-0003-full-node-detection-and-eviction.md) + (best-effort fresh delivery and possession checks), + [ADR-0005](ADR-0005-replication-repair-hardening.md), + ant-node `perf/replication-send-path`, saorsa-core `perf/replication-send-path`, + saorsa-transport `perf/replication-send-path`, + `ant-testnet/state/comparisons/web-support-memory-diag-0921/` (heap profiles + and per-minute live/RSS reports from the diagnosis) + +## Context + +Under sustained client uploads on a 60-node DigitalOcean testnet, individual +nodes grew from ~100 MiB to 1–2 GiB of resident memory within an hour and +kept growing for as long as writes continued. Heap profiles taken on the +running nodes attributed roughly 80% of live memory to one site: encoded +`FreshReplicationOffer` messages queued by the fresh-write drainer. + +The mechanism was structural rather than a leak: + +- Every accepted PUT was pushed to the drainer with the full chunk, and + `replicate_fresh` encoded the offer (chunk plus proof, up to ~4–5 MiB) + immediately, before any send permit was held. +- One send task per close-group peer then waited for one of + `MAX_CONCURRENT_REPLICATION_SENDS` (3) permits while pinning that encoded + buffer. Nothing bounded how many chunks could be waiting in that state. +- On a real network each send holds its permit for seconds (QUIC delivery + acknowledgement, retries, unreachable NAT peers), so a write rate above + the send rate grew the queue without limit. Loopback devnets never showed + it because sends complete instantly. + +Once the backlog was bounded, the profile showed the remaining cost per +in-flight send: the same frame existed as the caller's serialized message +*and* as the QUIC stream's copy for the whole transfer, plus transient +copies made while framing (payload clone per channel attempt, owned wire +message for signing, and a doubling `Vec` that left chunk-sized frames with +up to twice their length in capacity). + +## Decision Drivers + +- Node memory must stay bounded under any client write rate; replication + may be delayed by backpressure but must not be dropped. +- The change must not alter the wire format, storage format or payment + logic, so it can ship as a behavioural fix. +- Existing callers of the send APIs in saorsa-core and saorsa-transport must + keep compiling and behaving the same. + +## Considered Options + +1. Bound the fresh-write channel and drop or block PUT handling when full. + Rejected: either silently loses replication or blocks client responses + on network conditions. +2. Raise `MAX_CONCURRENT_REPLICATION_SENDS`. Rejected: only moves the + knee of the curve and increases bandwidth pressure on home links; the + queue behind the permits would still be unbounded. +3. Keep events small and take a bounded permit before materialising an + offer; separately remove the avoidable copies on the send path. Chosen. + +## Decision + +We will bound the number of encoded fresh offers that can exist at once and +make the send path hand a single owned buffer down to the QUIC stream: + +- `FreshWriteEvent` carries only the key and the payment proof, and fresh + replication runs as two stages. The fresh-write drainer never waits for + chunk back-pressure: for every event, at arrival rate, it records the key + in `PaidForList(self)` and sends `PaidNotify` to the paid close group — + the evidence later repair depends on — then forwards the event to the + offer dispatcher. The dispatcher is the only permit-gated stage: it + acquires a `MAX_PENDING_FRESH_OFFERS` (8) permit before it reads the + chunk back from storage (`get_raw`; the chunk was content-checked when + stored) and encodes it; the permit lives with the encoded offer until the + last per-peer send drops it. A backlog therefore waits as small queued + events, and at most ~40 MiB of encoded offers exist per node. Nothing is + dropped by back-pressure: both queues are unbounded and FIFO, every offer + is dispatched with the same fan-out, retries and delayed possession check, + and a failed read-back is retried `MAX_FRESH_READ_ATTEMPTS` times with the + permit released in between; only a chunk that is no longer stored is + skipped. +- The chunk moves into the offer rather than being copied, and + `ReplicationMessage::encode` serializes into an exactly-sized buffer. +- The encoded offer is shared as `Bytes`; saorsa-core's `send_message` + accepts `impl Into`, frames the payload through a borrowing + `WireMessageRef` (byte-identical to `WireMessage` on the wire) into an + exactly-sized frame, and passes that frame as `Bytes` to + saorsa-transport's new `send_bytes`, where the QUIC stream takes ownership + via `write_chunks` instead of copying it. + +## Consequences + +### Positive + +- Memory under write load is bounded by configuration: pending offers plus + the three in-flight sends, each held once, instead of growing with the + backlog. On the diagnostic fleets peak live memory fell from 1068 MiB to + 298 MiB (mimalloc build) and from 674 MiB to 262 MiB (jemalloc build) + after the backpressure change alone. +- Every large send node-wide (chunk GET responses included) stops paying + for a second copy of its frame during the transfer. +- No wire, storage or API break: `send(&[u8])` remains and copies once as + before; `Vec` callers of `send_message` convert without copying. + +### Negative / Trade-offs + +- Replication of a burst of writes is spread out in time rather than + encoded eagerly; the delayed possession check is scheduled after each + offer's sends are dispatched, so it shifts by the same amount. A chunk + fetched seconds after its upload can therefore have fewer replicas than + before (the 2026-09-22 comparison measured downloads of just-uploaded + files 8% slower). Paid-list evidence is not affected, and the previous + unbounded fan-out lost that evidence outright under load (2,795 + "paid notify dropped at admission" in one hour on the baseline fleet). +- The drainer re-reads each chunk from disk when its permit arrives, one + extra read per accepted write. + +### Neutral / Operational + +- `MAX_PENDING_FRESH_OFFERS` and `MAX_CONCURRENT_REPLICATION_SENDS` are + the two knobs; raising the first trades memory for burst absorption. +- The signing step still serializes the payload once to produce the signed + bytes; changing that would alter the signature input and is out of scope. + +## Validation + +- Unit tests: exact-capacity encoding of chunk-sized offers (ant-node) and + byte-for-byte equivalence of `WireMessageRef` with `WireMessage` + (saorsa-core); replication unit and e2e fresh-replication scenarios pass. +- Testnet evidence (2026-09-21): with the backpressure change, the node + that had reached 1051 MiB live memory stayed flat at 0.0 MiB/min with a + 150 MiB peak, and the worst bootstrap's queued offers dropped from 101 + (463 MiB) to 6 (17.7 MiB) in heap profiles. +- Review trigger: any change to fresh replication fan-out, send permits, or + the wire-message framing must re-run the memory diagnostics under + sustained uploads and confirm live memory stays bounded. + +## 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/src/replication/config.rs b/src/replication/config.rs index 9150d584..6aaec7b3 100644 --- a/src/replication/config.rs +++ b/src/replication/config.rs @@ -170,6 +170,28 @@ pub const SELF_LOOKUP_INTERVAL_MAX: Duration = Duration::from_secs(SELF_LOOKUP_I /// at most ~12 MB queued for the upload link at any instant. pub const MAX_CONCURRENT_REPLICATION_SENDS: usize = 3; +/// Maximum number of encoded fresh-replication offers held in memory. +/// +/// Each accepted write is encoded once (chunk plus proof, up to ~4 MB) and +/// that buffer stays alive until the last of its per-peer sends completes. +/// With only `MAX_CONCURRENT_REPLICATION_SENDS` transfers in flight, a write +/// rate above the network's send rate would otherwise queue an unbounded +/// number of encoded offers behind the send permits. The fresh-write drainer +/// takes one of these permits before it reads and encodes a chunk, so the +/// backlog waits as small queued events instead of chunk-sized buffers. +pub const MAX_PENDING_FRESH_OFFERS: usize = 8; + +/// How many times the offer dispatcher tries to read an accepted chunk back +/// from storage before giving up on its fresh offers. +/// +/// The chunk was stored moments earlier, so a failed read is a transient +/// fault (exhausted descriptors, an I/O hiccup) far more often than a lost +/// chunk; a lost chunk reports `None` and is skipped without retry. +pub const MAX_FRESH_READ_ATTEMPTS: u32 = 3; + +/// Pause before retrying a failed chunk read-back in the offer dispatcher. +pub const FRESH_READ_RETRY_DELAY: Duration = Duration::from_secs(1); + /// Maximum number of concurrent in-flight audit-responder tasks. /// /// The LIGHT audit-responder handlers — responsible-chunk audits and subtree diff --git a/src/replication/fresh.rs b/src/replication/fresh.rs index 80e9766e..1b3b156d 100644 --- a/src/replication/fresh.rs +++ b/src/replication/fresh.rs @@ -2,22 +2,27 @@ //! //! When a node accepts a newly written record with valid `PoP`: //! 1. Store locally (already done by chunk handler). -//! 2. Send fresh offers to `CLOSE_GROUP_SIZE` nearest peers (excluding self). -//! 3. Send `PaidNotify` to all peers in `PaidCloseGroup(K)`. +//! 2. Record the key in `PaidForList(self)` and send `PaidNotify` to every +//! peer in `PaidCloseGroup(K)` — immediately, never behind back-pressure. +//! 3. Send fresh offers to `CLOSE_GROUP_SIZE` nearest peers (excluding self), +//! bounded by the pending-offer permits so a write burst cannot pile up +//! chunk-sized buffers. use std::sync::Arc; use crate::logging::{debug, warn}; +use bytes::Bytes; use rand::Rng; use saorsa_core::identity::PeerId; use saorsa_core::P2PNode; -use tokio::sync::Semaphore; +use tokio::sync::{mpsc, OwnedSemaphorePermit, Semaphore}; use crate::ant_protocol::XorName; use crate::replication::config::{ ReplicationConfig, FRESH_REPLICATION_DELIVERY_MAX_RETRIES, REPLICATION_PROTOCOL_ID, }; use crate::replication::paid_list::PaidList; +use crate::replication::possession::PossessionCheckEvent; use crate::replication::protocol::{ FreshReplicationOffer, PaidNotify, ReplicationMessage, ReplicationMessageBody, }; @@ -26,48 +31,94 @@ use crate::replication::protocol::{ /// /// Sent from the chunk PUT handler to the replication engine via an /// unbounded channel so that the PUT response is not blocked by -/// replication fan-out. +/// replication fan-out. The event deliberately carries no chunk bytes: the +/// chunk is already on disk, and the offer dispatcher reads it back only +/// once it holds a pending-offer permit, so a replication backlog queues as +/// small events rather than chunk-sized buffers. pub struct FreshWriteEvent { /// Content-address of the stored chunk. pub key: XorName, - /// The chunk data. - pub data: Vec, /// Serialized proof-of-payment. pub payment_proof: Vec, } -/// Execute fresh replication for a newly accepted record. +/// A write whose paid-list evidence has been announced and whose chunk offer +/// is waiting for a pending-offer permit. Carries no chunk bytes. +pub(crate) struct FreshOfferEvent { + pub(crate) key: XorName, + pub(crate) payment_proof: Vec, + /// Storage read-backs attempted so far; see `MAX_FRESH_READ_ATTEMPTS`. + pub(crate) read_attempts: u32, +} + +/// Handles shared by everything that dispatches fresh offers, so the offer +/// dispatcher task and the direct entry point run one pipeline. +#[derive(Clone)] +pub(crate) struct FreshOfferContext { + pub(crate) p2p_node: Arc, + pub(crate) config: Arc, + /// Limits concurrent outbound chunk transfers across the engine. + pub(crate) send_semaphore: Arc, + /// Delayed possession checks (ADR-0003) are scheduled here once an + /// offer's sends are dispatched. + pub(crate) possession_check_tx: mpsc::UnboundedSender, +} + +/// An encoded fresh offer shared by the per-peer send tasks. /// -/// Sends fresh offers to close group members (with bounded delivery retries, -/// ADR-0003) and `PaidNotify` to `PaidCloseGroup`. Returns the close-group -/// peers responsible for the key (excluding self) so the caller can schedule -/// the delayed possession check; `PaidNotify` remains fire-and-forget. +/// The pending-offer permit is released together with the buffer, once the +/// last send task drops its reference, which caps how many encoded offers +/// can wait behind the send permits at `MAX_PENDING_FRESH_OFFERS`. The bytes +/// are shared with the transport as well: each send attempt hands out a +/// reference-counted handle rather than a copy. +struct EncodedOffer { + bytes: Bytes, + _pending: OwnedSemaphorePermit, +} + +/// Rules 6-8: record the paid key locally and announce it to +/// `PaidCloseGroup(K)`. /// -/// The `send_semaphore` limits how many outbound chunk transfers can be -/// in-flight concurrently across the entire replication engine, preventing -/// bandwidth saturation on home broadband connections. -pub async fn replicate_fresh( +/// This is the evidence peers need to repair the key later, so it runs the +/// moment a write is accepted and is never gated by the pending-offer permit +/// or the send semaphore; both messages are small metadata. +pub(crate) async fn announce_paid_write( key: &XorName, - data: &[u8], proof_of_payment: &[u8], + paid_list: &PaidList, p2p_node: &Arc, - paid_list: &Arc, config: &ReplicationConfig, - send_semaphore: &Arc, -) -> Vec { - let self_id = *p2p_node.peer_id(); - +) { // Rule 6: Node that validates PoP adds K to PaidForList(self). if let Err(e) = paid_list.insert(key).await { warn!("Failed to add key {} to PaidForList: {e}", hex::encode(key)); } + // Rules 7-8: PaidNotify to every member of PaidCloseGroup(K). + send_paid_notify(key, proof_of_payment, p2p_node, config).await; +} - // Rule 2-3: Send fresh offers to CLOSE_GROUP_SIZE nearest peers - // (excluding self). Use self-inclusive query to get the true close group, - // then filter self out. - let closest = p2p_node +/// Rules 2-3: send fresh offers to the close group and schedule the delayed +/// possession check (ADR-0003) for the responsible peers. +/// +/// `pending_offer` is the caller's permit from the pending-offer semaphore; +/// it is held with the encoded offer until the last per-peer send finishes. +/// `data` is taken by value so the chunk moves into the offer instead of +/// being copied. +pub(crate) async fn dispatch_fresh_offer( + ctx: &FreshOfferContext, + key: &XorName, + data: Vec, + proof_of_payment: &[u8], + pending_offer: OwnedSemaphorePermit, +) { + let self_id = *ctx.p2p_node.peer_id(); + + // Use the self-inclusive query to get the true close group, then filter + // self out. + let closest = ctx + .p2p_node .dht_manager() - .find_closest_nodes_local_with_self(key, config.close_group_size) + .find_closest_nodes_local_with_self(key, ctx.config.close_group_size) .await; let target_peers: Vec = closest .iter() @@ -77,7 +128,7 @@ pub async fn replicate_fresh( let offer = FreshReplicationOffer { key: *key, - data: data.to_vec(), + data, proof_of_payment: proof_of_payment.to_vec(), }; let request_id = rand::thread_rng().gen::(); @@ -86,22 +137,29 @@ pub async fn replicate_fresh( body: ReplicationMessageBody::FreshReplicationOffer(offer), }; - let Ok(encoded) = offer_msg.encode() else { + let encoded = offer_msg.encode(); + // Only the encoded bytes are needed from here on; release the chunk now + // rather than holding it alongside the encoding while sends are queued. + drop(offer_msg); + let Ok(encoded) = encoded else { warn!( "Failed to encode FreshReplicationOffer for {}", hex::encode(key), ); - return Vec::new(); + return; }; - // Share one encoded copy across the per-peer send tasks so a retry only - // re-materialises the buffer for the (consuming) send call, keeping the - // common single-attempt path at one clone per peer. - let encoded = Arc::new(encoded); + // One encoded copy serves every per-peer send task and every retry; the + // transport borrows it through `Bytes` instead of taking a copy. The + // pending-offer permit travels with the buffer. + let encoded = Arc::new(EncodedOffer { + bytes: Bytes::from(encoded), + _pending: pending_offer, + }); for peer in &target_peers { - let p2p = Arc::clone(p2p_node); - let data = Arc::clone(&encoded); + let p2p = Arc::clone(&ctx.p2p_node); + let offer = Arc::clone(&encoded); let peer_id = *peer; - let sem = Arc::clone(send_semaphore); + let sem = Arc::clone(&ctx.send_semaphore); tokio::spawn(async move { // Acquire a permit before sending — this caps the number of // concurrent outbound replication transfers across the engine. @@ -117,12 +175,7 @@ pub async fn replicate_fresh( let mut attempt = 0u32; loop { match p2p - .send_message( - &peer_id, - REPLICATION_PROTOCOL_ID, - data.as_ref().clone(), - &[], - ) + .send_message(&peer_id, REPLICATION_PROTOCOL_ID, offer.bytes.clone(), &[]) .await { Ok(()) => break, @@ -145,24 +198,28 @@ pub async fn replicate_fresh( }); } - // Rule 7-8: Send PaidNotify to every member of PaidCloseGroup(K). - // PaidNotify messages are small metadata (no chunk data), so they don't - // need semaphore gating. - send_paid_notify(key, proof_of_payment, p2p_node, config).await; - debug!( - "Fresh replication initiated for {} to {} peers + PaidNotify", + "Fresh replication initiated for {} to {} peers", hex::encode(key), target_peers.len() ); - target_peers + // Schedule the delayed possession check (ADR-0003) for the responsible + // close-group peers. A closed receiver (engine shutting down) is ignored. + if !target_peers.is_empty() { + let _ = ctx.possession_check_tx.send(PossessionCheckEvent { + key: *key, + peers: target_peers, + }); + } } /// Send `PaidNotify(K)` to every peer in `PaidCloseGroup(K)` (fire-and-forget). /// -/// Per Invariant 16: sender MUST attempt delivery to every member. -async fn send_paid_notify( +/// Per Invariant 16: sender MUST attempt delivery to every member. The +/// message is small metadata (no chunk data), so it is neither gated by the +/// send semaphore nor by the pending-offer permit. +pub(crate) async fn send_paid_notify( key: &XorName, proof_of_payment: &[u8], p2p_node: &Arc, @@ -188,7 +245,8 @@ async fn send_paid_notify( warn!("Failed to encode PaidNotify for {}", hex::encode(key)); return; }; - + // One buffer for every recipient; the sends only take handles. + let encoded = Bytes::from(encoded); for node in &paid_group { if node.peer_id == self_id { continue; diff --git a/src/replication/mod.rs b/src/replication/mod.rs index e245bd9d..cac1bbeb 100644 --- a/src/replication/mod.rs +++ b/src/replication/mod.rs @@ -77,12 +77,12 @@ use crate::replication::commitment_state::{ PeerCommitmentRecord, PersistedRetention, ResponderCommitmentState, GOSSIP_ANSWERABILITY_TTL, }; use crate::replication::config::{ - max_parallel_fetch, storage_admission_width, ReplicationConfig, MAX_AUDIT_RESPONSES_PER_PEER, - MAX_CONCURRENT_AUDIT_RESPONSES, MAX_CONCURRENT_REPLICATION_SENDS, - MAX_DIGEST_AUDIT_RESPONSES_PER_PEER, MAX_INCOMING_VERIFICATION_KEYS, - MAX_SUBTREE_ROUND1_PER_PEER, MAX_SUBTREE_SESSIONS, MAX_VERIFICATION_KEYS_PER_CYCLE, - REPLICATION_PROTOCOL_ID, SUBTREE_AUDIT_PROTOCOL_ID, SUBTREE_ROUND1_WORK_BURST_BYTES, - SUBTREE_ROUND1_WORK_REFILL_BYTES_PER_SEC, SUBTREE_SESSION_TTL, + max_parallel_fetch, storage_admission_width, ReplicationConfig, FRESH_READ_RETRY_DELAY, + MAX_AUDIT_RESPONSES_PER_PEER, MAX_CONCURRENT_AUDIT_RESPONSES, MAX_CONCURRENT_REPLICATION_SENDS, + MAX_DIGEST_AUDIT_RESPONSES_PER_PEER, MAX_FRESH_READ_ATTEMPTS, MAX_INCOMING_VERIFICATION_KEYS, + MAX_PENDING_FRESH_OFFERS, MAX_SUBTREE_ROUND1_PER_PEER, MAX_SUBTREE_SESSIONS, + MAX_VERIFICATION_KEYS_PER_CYCLE, REPLICATION_PROTOCOL_ID, SUBTREE_AUDIT_PROTOCOL_ID, + SUBTREE_ROUND1_WORK_BURST_BYTES, SUBTREE_ROUND1_WORK_REFILL_BYTES_PER_SEC, SUBTREE_SESSION_TTL, }; use crate::replication::paid_list::PaidList; use crate::replication::protocol::{ @@ -1746,6 +1746,9 @@ pub struct ReplicationEngine { /// Limits concurrent outbound replication sends to prevent bandwidth /// saturation on home broadband connections. send_semaphore: Arc, + /// Bounds how many encoded fresh offers can wait behind `send_semaphore`; + /// see [`MAX_PENDING_FRESH_OFFERS`]. + pending_offer_semaphore: Arc, /// Bounds concurrent IN-FLIGHT LIGHT audit-responder tasks (responsible-chunk /// audits + subtree slice round 2). The heavy subtree round 1 has its own /// tighter pool ([`SubtreeRound1Limiter`]). Those are spawned off the serial @@ -1812,9 +1815,15 @@ pub struct ReplicationEngine { subtree_round1: SubtreeRound1Limiter, /// Receiver for fresh-write events from the chunk PUT handler. /// - /// When present, `start()` spawns a drainer task that calls - /// `replicate_fresh` for each event. + /// When present, `start()` spawns the fresh-write drainer, which records + /// paid-list evidence for each event immediately and forwards it to the + /// offer dispatcher. fresh_write_rx: Option>, + /// Hand-off from the fresh-write drainer to the offer dispatcher, the only + /// permit-gated stage. Unbounded and FIFO, holding key + proof only. + fresh_offer_tx: mpsc::UnboundedSender, + /// Receiver paired with `fresh_offer_tx`; taken by the dispatcher task. + fresh_offer_rx: Option>, /// Sender for delayed possession-check events (ADR-0003). The fresh-write /// drainer pushes the responsible close-group peers here after each fresh /// replication; the possession-check scheduler drains the paired receiver. @@ -1878,6 +1887,7 @@ impl ReplicationEngine { let initial_neighbors = NeighborSyncState::new_cycle(Vec::new()); let config = Arc::new(config); let (possession_check_tx, possession_check_rx) = mpsc::unbounded_channel(); + let (fresh_offer_tx, fresh_offer_rx) = mpsc::unbounded_channel(); // ADR-0004: monetized-pin channel (verifier -> first-audit drainer). // Bounded (Amendment 2): every stage of the first-audit pipeline is @@ -1911,6 +1921,7 @@ impl ReplicationEngine { recent_provers: Arc::new(RwLock::new(RecentProvers::new())), sig_verify_attempts: Arc::new(RwLock::new(HashMap::new())), send_semaphore: Arc::new(Semaphore::new(MAX_CONCURRENT_REPLICATION_SENDS)), + pending_offer_semaphore: Arc::new(Semaphore::new(MAX_PENDING_FRESH_OFFERS)), audit_responder_semaphore: Arc::new(Semaphore::new(MAX_CONCURRENT_AUDIT_RESPONSES)), audit_responder_inflight: Arc::new(RwLock::new(HashMap::new())), audit_responder_metrics: Arc::new(AuditResponderMetrics::default()), @@ -1948,6 +1959,8 @@ impl ReplicationEngine { config.subtree_round1_max_concurrent, ), fresh_write_rx: Some(fresh_write_rx), + fresh_offer_tx, + fresh_offer_rx: Some(fresh_offer_rx), possession_check_tx, possession_check_rx: Some(possession_check_rx), monetized_pin_tx, @@ -2205,6 +2218,16 @@ impl ReplicationEngine { }) } + /// Test-only: pending-offer permits not currently held by an encoded + /// fresh offer. Equals [`MAX_PENDING_FRESH_OFFERS`] when no fresh + /// replication is in flight, which is how tests prove a burst of writes + /// drained without leaking a permit. + #[cfg(any(test, feature = "test-utils"))] + #[must_use] + pub fn pending_offer_permits_available(&self) -> usize { + self.pending_offer_semaphore.available_permits() + } + /// Start all background tasks. /// /// `dht_events` must be subscribed **before** `P2PNode::start()` so that @@ -2231,6 +2254,7 @@ impl ReplicationEngine { self.start_verification_worker(); self.start_bootstrap_sync(dht_events); self.start_fresh_write_drainer(); + self.start_fresh_offer_dispatcher(); self.start_possession_check_scheduler(); // ADR-0004: deterministic first audit of commitments that backed a // payment (surfaced by the verifier cross-check). @@ -2369,20 +2393,38 @@ impl ReplicationEngine { /// drainer; this direct entry point schedules here so callers (and tests) /// that drive replication directly still get the possession check. pub async fn replicate_fresh(&self, key: &XorName, data: &[u8], proof_of_payment: &[u8]) { - let peers = fresh::replicate_fresh( + fresh::announce_paid_write( key, - data, proof_of_payment, - &self.p2p_node, &self.paid_list, + &self.p2p_node, &self.config, - &self.send_semaphore, ) .await; - if !peers.is_empty() { - let _ = self - .possession_check_tx - .send(possession::PossessionCheckEvent { key: *key, peers }); + // The semaphore is never closed, so this only fails at shutdown. + let Ok(pending_offer) = Arc::clone(&self.pending_offer_semaphore) + .acquire_owned() + .await + else { + return; + }; + fresh::dispatch_fresh_offer( + &self.fresh_offer_context(), + key, + data.to_vec(), + proof_of_payment, + pending_offer, + ) + .await; + } + + /// Handles the offer dispatcher and the direct entry point share. + fn fresh_offer_context(&self) -> fresh::FreshOfferContext { + fresh::FreshOfferContext { + p2p_node: Arc::clone(&self.p2p_node), + config: Arc::clone(&self.config), + send_semaphore: Arc::clone(&self.send_semaphore), + possession_check_tx: self.possession_check_tx.clone(), } } @@ -2399,36 +2441,38 @@ impl ReplicationEngine { let p2p = Arc::clone(&self.p2p_node); let paid_list = Arc::clone(&self.paid_list); let config = Arc::clone(&self.config); - let send_semaphore = Arc::clone(&self.send_semaphore); - let possession_tx = self.possession_check_tx.clone(); + let offer_tx = self.fresh_offer_tx.clone(); let shutdown = self.shutdown.clone(); let handle = tokio::spawn(async move { loop { - tokio::select! { + let event = tokio::select! { () = shutdown.cancelled() => break, event = rx.recv() => { let Some(event) = event else { break }; - let peers = fresh::replicate_fresh( - &event.key, - &event.data, - &event.payment_proof, - &p2p, - &paid_list, - &config, - &send_semaphore, - ) - .await; - // Schedule the delayed possession check (ADR-0003) for - // the responsible close-group peers. A closed receiver - // (engine shutting down) is ignored. - if !peers.is_empty() { - let _ = possession_tx.send(possession::PossessionCheckEvent { - key: event.key, - peers, - }); - } + event } + }; + // Stage one never waits for chunk back-pressure: the paid-list + // entry and PaidNotify are what let peers repair the key later, + // so every queued write gets them at arrival rate. + fresh::announce_paid_write( + &event.key, + &event.payment_proof, + &paid_list, + &p2p, + &config, + ) + .await; + if offer_tx + .send(fresh::FreshOfferEvent { + key: event.key, + payment_proof: event.payment_proof, + read_attempts: 0, + }) + .is_err() + { + break; } } debug!("Fresh-write drainer shut down"); @@ -2436,6 +2480,88 @@ impl ReplicationEngine { self.task_handles.push(handle); } + /// Spawn the fresh-offer dispatcher: the only stage of fresh replication + /// that waits for a pending-offer permit. + /// + /// For each forwarded write it acquires a permit, reads the chunk back + /// from storage and dispatches the offers. A missing chunk is skipped; a + /// failed read is retried up to `MAX_FRESH_READ_ATTEMPTS` times with the + /// permit released in between, so a transient I/O fault does not lose the + /// write's replication. + fn start_fresh_offer_dispatcher(&mut self) { + let Some(mut rx) = self.fresh_offer_rx.take() else { + return; + }; + let storage = Arc::clone(&self.storage); + let pending_offer_semaphore = Arc::clone(&self.pending_offer_semaphore); + let offer_tx = self.fresh_offer_tx.clone(); + let ctx = self.fresh_offer_context(); + let shutdown = self.shutdown.clone(); + + let handle = tokio::spawn(async move { + loop { + let event = tokio::select! { + () = shutdown.cancelled() => break, + event = rx.recv() => { + let Some(event) = event else { break }; + event + } + }; + // Wait for a pending-offer permit before touching the chunk so a + // send backlog holds queued events, not encoded chunk buffers. + let pending_offer = tokio::select! { + () = shutdown.cancelled() => break, + permit = Arc::clone(&pending_offer_semaphore).acquire_owned() => { + let Ok(permit) = permit else { break }; + permit + } + }; + let key_hex = hex::encode(event.key); + // The chunk was content-checked when it was stored; the + // receiver validates the offer against its address anyway. + let data = match storage.get_raw(&event.key).await { + Ok(Some(data)) => data, + Ok(None) => { + debug!("Chunk {key_hex} no longer stored, skipping fresh replication"); + continue; + } + Err(e) => { + drop(pending_offer); + let attempts = event.read_attempts + 1; + if attempts >= MAX_FRESH_READ_ATTEMPTS { + warn!( + "Giving up fresh replication of {key_hex} after {attempts} failed reads: {e}" + ); + continue; + } + warn!( + "Failed to read chunk {key_hex} for fresh replication (attempt {attempts}): {e}" + ); + tokio::select! { + () = shutdown.cancelled() => break, + () = tokio::time::sleep(FRESH_READ_RETRY_DELAY) => {} + } + let _ = offer_tx.send(fresh::FreshOfferEvent { + read_attempts: attempts, + ..event + }); + continue; + } + }; + fresh::dispatch_fresh_offer( + &ctx, + &event.key, + data, + &event.payment_proof, + pending_offer, + ) + .await; + } + debug!("Fresh-offer dispatcher shut down"); + }); + self.task_handles.push(handle); + } + /// Spawn the possession-check scheduler (ADR-0003). /// /// Drains scheduled possession-check events and, for each, waits a diff --git a/src/replication/protocol.rs b/src/replication/protocol.rs index 3ef7d0fa..55b1fd0f 100644 --- a/src/replication/protocol.rs +++ b/src/replication/protocol.rs @@ -46,7 +46,19 @@ impl ReplicationMessage { /// Returns [`ReplicationProtocolError::SerializationFailed`] if postcard /// serialization fails. pub fn encode(&self) -> Result, ReplicationProtocolError> { - let bytes = postcard::to_stdvec(self) + // Size the buffer exactly up front. Chunk-carrying bodies run to + // several MiB, and a growing `Vec` would otherwise end up with up to + // twice the needed capacity, retained for as long as the encoded + // message is queued for sending. + let size = postcard::experimental::serialized_size(self) + .map_err(|e| ReplicationProtocolError::SerializationFailed(e.to_string()))?; + // The size is known before anything is allocated, so an oversized body + // is refused without serializing it first. + let max_size = ceiling_for(family_of_variant(self.body.variant_index())); + if size > max_size { + return Err(ReplicationProtocolError::MessageTooLarge { size, max_size }); + } + let bytes = postcard::to_extend(self, Vec::with_capacity(size)) .map_err(|e| ReplicationProtocolError::SerializationFailed(e.to_string()))?; // The same family ceiling the decoder applies, from the same table and @@ -66,13 +78,6 @@ impl ReplicationMessage { // the largest is a round-1 proof at the commitment // key-count cap, pinned under it with headroom by // `max_round1_proof_fits_the_audit_family_ceiling`. - let max_size = ceiling_for(family_of_variant(self.body.variant_index())); - if bytes.len() > max_size { - return Err(ReplicationProtocolError::MessageTooLarge { - size: bytes.len(), - max_size, - }); - } // V2-623: cumulative per-variant tx accounting. Every replication send // funnels through here, so this is the single tx choke point. @@ -797,9 +802,13 @@ pub(crate) fn log_served_peers_summary() { pub struct FreshReplicationOffer { /// The record key. pub key: XorName, - /// The record data. + /// The record data. Encoded as a byte string, which postcard lays out + /// exactly like a `u8` sequence, so the wire format is unchanged while + /// serialization and sizing copy the payload in one pass. + #[serde(with = "serde_bytes")] pub data: Vec, /// Proof of Payment (required, validated by receiver). + #[serde(with = "serde_bytes")] pub proof_of_payment: Vec, } @@ -829,6 +838,7 @@ pub struct PaidNotify { /// The record key. pub key: XorName, /// Proof of Payment for receiver-side verification. + #[serde(with = "serde_bytes")] pub proof_of_payment: Vec, } @@ -2429,6 +2439,75 @@ mod tests { assert_eq!(decoded.request_id, 7); } + /// `serde_bytes` must not change the wire layout: postcard encodes a byte + /// string and a `u8` sequence identically (varint length + raw bytes). + #[test] + fn byte_string_fields_encode_like_u8_sequences() { + #[derive(Serialize)] + struct PlainOffer { + key: XorName, + data: Vec, + proof_of_payment: Vec, + } + #[derive(Serialize)] + struct PlainNotify { + key: XorName, + proof_of_payment: Vec, + } + let data: Vec = (0..=255u8).cycle().take(70_000).collect(); + let proof = vec![9u8; 300]; + + let offer = FreshReplicationOffer { + key: [5; 32], + data: data.clone(), + proof_of_payment: proof.clone(), + }; + let plain_offer = PlainOffer { + key: [5; 32], + data: data.clone(), + proof_of_payment: proof.clone(), + }; + assert_eq!( + postcard::to_stdvec(&offer).unwrap(), + postcard::to_stdvec(&plain_offer).unwrap() + ); + let decoded: FreshReplicationOffer = + postcard::from_bytes(&postcard::to_stdvec(&plain_offer).unwrap()).unwrap(); + assert_eq!(decoded.data, data); + assert_eq!(decoded.proof_of_payment, proof); + + let notify = PaidNotify { + key: [6; 32], + proof_of_payment: proof.clone(), + }; + let plain_notify = PlainNotify { + key: [6; 32], + proof_of_payment: proof, + }; + assert_eq!( + postcard::to_stdvec(¬ify).unwrap(), + postcard::to_stdvec(&plain_notify).unwrap() + ); + } + + #[test] + fn encode_allocates_exactly_the_serialized_size() { + // A chunk-sized offer must not carry growth slack: the encoded buffer is + // shared by every per-peer send task for as long as it is queued. + let msg = ReplicationMessage { + request_id: 7, + body: ReplicationMessageBody::FreshReplicationOffer(FreshReplicationOffer { + key: [3; 32], + data: vec![0xAB; 3 * 1024 * 1024 + 123], + proof_of_payment: vec![1, 2, 3], + }), + }; + let encoded = msg.encode().unwrap(); + assert_eq!(encoded.capacity(), encoded.len()); + let decoded = ReplicationMessage::decode(&encoded).unwrap(); + assert_eq!(decoded.request_id, 7); + } + #[test] fn encode_rejects_oversized_message() { // Build a message whose serialized form exceeds the limit. diff --git a/src/storage/handler.rs b/src/storage/handler.rs index bedffa11..d2726e4a 100644 --- a/src/storage/handler.rs +++ b/src/storage/handler.rs @@ -357,6 +357,13 @@ impl AntProtocol { self.fresh_write_tx = Some(tx); } + /// The fresh-write sender, if one was set. Lets tests drive the + /// replication engine's fresh-write pipeline exactly as a PUT would. + #[cfg(any(test, feature = "test-utils"))] + pub fn fresh_write_sender(&self) -> Option> { + self.fresh_write_tx.clone() + } + /// Get the protocol identifier. #[must_use] pub fn protocol_id(&self) -> &'static str { @@ -733,14 +740,11 @@ impl AntProtocol { // fall back to the original proof rather than dropping the // replication entirely. let proof = Self::strip_commitment_sidecars(proof); - // `request.content` is now `bytes::Bytes`; FreshWriteEvent - // still carries the chunk as `Vec` for compatibility - // with the replication wire format, so materialise once - // here. Done only on the success path, where storage has - // already accepted the chunk. + // Storage has already accepted the chunk on this path, so + // the event carries only the key; the replication drainer + // reads the chunk back when it is ready to send it. let event = FreshWriteEvent { key: address, - data: request.content.to_vec(), payment_proof: proof, }; if tx.send(event).is_err() { diff --git a/tests/e2e/replication.rs b/tests/e2e/replication.rs index abf5f92b..814931bf 100644 --- a/tests/e2e/replication.rs +++ b/tests/e2e/replication.rs @@ -11,8 +11,10 @@ use ant_node::client::compute_address; use ant_node::replication::audit_coordinator::AuditChallengeCoordinator; use ant_node::replication::commitment_state::{BuiltCommitment, ResponderCommitmentState}; use ant_node::replication::config::{ - storage_admission_width, K_BUCKET_SIZE, REPLICATION_PROTOCOL_ID, + storage_admission_width, FRESH_READ_RETRY_DELAY, K_BUCKET_SIZE, MAX_PENDING_FRESH_OFFERS, + REPLICATION_PROTOCOL_ID, }; +use ant_node::replication::fresh::FreshWriteEvent; use ant_node::replication::protocol::{ compute_audit_digest, AuditChallenge, AuditResponse, FetchRequest, FetchResponse, FreshReplicationOffer, FreshReplicationResponse, NeighborSyncRequest, ReplicationMessage, @@ -21,11 +23,17 @@ use ant_node::replication::protocol::{ use ant_node::replication::pruning; use ant_node::replication::scheduling::ReplicationQueues; use ant_node::replication::types::{NeighborSyncState, RepairProofs}; +use ant_node::storage::file_store::CHUNKS_DIR_NAME; +use ant_node::storage::XorName; use ant_node::ReplicationConfig; use saorsa_core::identity::PeerId; use saorsa_core::{P2PNode, TrustEvent}; use serial_test::serial; use std::collections::HashSet; +use std::fs; +#[cfg(unix)] +use std::os::unix::fs::PermissionsExt; +use std::path::{Path, PathBuf}; use std::sync::Arc; use std::time::Duration; use tokio::sync::RwLock; @@ -48,6 +56,17 @@ const FULL_NODE_SHUN_POSSESSION_DELAY_MAX: Duration = Duration::from_millis(500) const DUMMY_PAYMENT_PROOF_LEN: usize = 64; /// Dummy proof byte used when a test only needs to reach pre-payment gates. const DUMMY_PAYMENT_PROOF_BYTE: u8 = 0x01; +/// First regular (non-bootstrap) node of the minimal harness; source of the +/// fresh-write pipeline tests. +const FRESH_PIPELINE_SOURCE_INDEX: usize = 3; +/// Writes queued at once by the saturation test: three times the pending-offer +/// budget, so the dispatcher must block on and recycle permits to drain it. +const FRESH_BURST_WRITES: usize = 3 * MAX_PENDING_FRESH_OFFERS; +/// Wait budget for the whole burst to replicate and release its permits. +const FRESH_BURST_TIMEOUT: Duration = Duration::from_secs(45); +/// File mode that makes a chunk unreadable, injecting a transient read fault. +#[cfg(unix)] +const UNREADABLE_FILE_MODE: u32 = 0o000; /// Minimal paid-list repair close group used by the deterministic repair e2e. const PAID_REPAIR_GROUP_SIZE: usize = 5; /// Storage threshold configured above majority so one holder is below quorum. @@ -256,6 +275,302 @@ async fn test_fresh_replication_propagates_to_close_group() { harness.teardown().await.expect("teardown"); } +/// The PUT-driven pipeline (fresh-write drainer → offer dispatcher) replicates +/// a queued write, and a write whose chunk is no longer stored is skipped +/// without stalling the pipeline or leaking a pending-offer permit. +#[tokio::test] +async fn fresh_write_pipeline_replicates_queued_writes_and_skips_missing_chunks() { + let harness = TestHarness::setup_minimal().await.expect("setup"); + harness.warmup_dht().await.expect("warmup"); + + let source = harness + .test_node(FRESH_PIPELINE_SOURCE_INDEX) + .expect("source node"); + let fresh_tx = source + .fresh_write_tx + .clone() + .expect("fresh-write sender wired by the harness"); + let address = store_paid_chunk( + &harness, + FRESH_PIPELINE_SOURCE_INDEX, + b"queued write through the fresh-write pipeline", + ) + .await; + + // A write whose chunk was never stored goes first: the dispatcher must + // skip it and carry on with the next event. + let missing = compute_address(b"never stored anywhere"); + fresh_tx + .send(FreshWriteEvent { + key: missing, + payment_proof: dummy_payment_proof(), + }) + .expect("queue missing write"); + fresh_tx + .send(FreshWriteEvent { + key: address, + payment_proof: dummy_payment_proof(), + }) + .expect("queue write"); + + assert!( + wait_until_replicated( + &harness, + FRESH_PIPELINE_SOURCE_INDEX, + &address, + PROPAGATION_TIMEOUT + ) + .await, + "queued write should have replicated through the fresh-write pipeline" + ); + + harness.teardown().await.expect("teardown"); +} + +/// Saturation: a burst of writes three times larger than the pending-offer +/// budget, queued in one go, all replicate. The dispatcher has to block on the +/// `MAX_PENDING_FRESH_OFFERS` semaphore and recycle permits to get through it, +/// and once the burst has drained every permit is back — no write was lost to +/// back-pressure and no permit leaked. +#[tokio::test] +async fn fresh_write_pipeline_drains_a_burst_larger_than_the_offer_budget() { + let harness = TestHarness::setup_minimal().await.expect("setup"); + harness.warmup_dht().await.expect("warmup"); + + let source = harness + .test_node(FRESH_PIPELINE_SOURCE_INDEX) + .expect("source node"); + let engine = source + .replication_engine + .as_ref() + .expect("replication engine"); + let fresh_tx = source + .fresh_write_tx + .clone() + .expect("fresh-write sender wired by the harness"); + + let mut addresses = Vec::with_capacity(FRESH_BURST_WRITES); + for i in 0..FRESH_BURST_WRITES { + let content = format!("fresh-write burst chunk {i}"); + addresses.push( + store_paid_chunk(&harness, FRESH_PIPELINE_SOURCE_INDEX, content.as_bytes()).await, + ); + } + assert_eq!( + engine.pending_offer_permits_available(), + MAX_PENDING_FRESH_OFFERS, + "all permits must be free before the burst" + ); + + for &address in &addresses { + fresh_tx + .send(FreshWriteEvent { + key: address, + payment_proof: dummy_payment_proof(), + }) + .expect("queue write"); + } + + let deadline = tokio::time::Instant::now() + FRESH_BURST_TIMEOUT; + let mut replicated: HashSet = HashSet::new(); + while tokio::time::Instant::now() < deadline && replicated.len() < addresses.len() { + for address in &addresses { + if !replicated.contains(address) + && stored_on_another_node(&harness, FRESH_PIPELINE_SOURCE_INDEX, address) + { + replicated.insert(*address); + } + } + tokio::time::sleep(PROPAGATION_POLL_INTERVAL).await; + } + assert_eq!( + replicated.len(), + addresses.len(), + "only {} of {} burst writes replicated within {FRESH_BURST_TIMEOUT:?}", + replicated.len(), + addresses.len() + ); + + // A permit is released when the last per-peer send of its offer finishes, + // which can trail the chunk landing on a peer by a moment. + while tokio::time::Instant::now() < deadline + && engine.pending_offer_permits_available() < MAX_PENDING_FRESH_OFFERS + { + tokio::time::sleep(PROPAGATION_POLL_INTERVAL).await; + } + assert_eq!( + engine.pending_offer_permits_available(), + MAX_PENDING_FRESH_OFFERS, + "the burst leaked a pending-offer permit" + ); + + harness.teardown().await.expect("teardown"); +} + +/// Retry: a chunk whose file cannot be read when its offer permit arrives is +/// not dropped. The dispatcher releases the permit, waits +/// `FRESH_READ_RETRY_DELAY` and reads again; once the fault has cleared the +/// write replicates. The fault is injected by making the chunk file +/// unreadable on disk and restored inside the retry window. +#[cfg(unix)] +#[tokio::test] +async fn fresh_write_pipeline_retries_a_transient_read_failure() { + let harness = TestHarness::setup_minimal().await.expect("setup"); + harness.warmup_dht().await.expect("warmup"); + + let source = harness + .test_node(FRESH_PIPELINE_SOURCE_INDEX) + .expect("source node"); + let storage = source.ant_protocol.as_ref().expect("protocol").storage(); + let fresh_tx = source + .fresh_write_tx + .clone() + .expect("fresh-write sender wired by the harness"); + let address = store_paid_chunk( + &harness, + FRESH_PIPELINE_SOURCE_INDEX, + b"chunk whose first read-back fails", + ) + .await; + + let chunk_file = chunk_file_path(storage.root_dir(), &address).expect("chunk file on disk"); + let readable = fs::metadata(&chunk_file) + .expect("chunk metadata") + .permissions(); + fs::set_permissions( + &chunk_file, + fs::Permissions::from_mode(UNREADABLE_FILE_MODE), + ) + .expect("make chunk unreadable"); + if fs::File::open(&chunk_file).is_ok() { + // File modes are not enforced for this user (root): the fault cannot + // be injected, so there is nothing to test here. + eprintln!("skipping: file modes are not enforced for this user"); + fs::set_permissions(&chunk_file, readable).expect("restore chunk mode"); + harness.teardown().await.expect("teardown"); + return; + } + assert!( + storage.exists(&address).unwrap_or(false), + "chunk is indexed before the fault is hit" + ); + + fresh_tx + .send(FreshWriteEvent { + key: address, + payment_proof: dummy_payment_proof(), + }) + .expect("queue write"); + + // A failed read marks the chunk suspect, which hides it from `exists`: + // that is the observable proof the dispatcher's first attempt hit the + // fault. Only then clear it, inside the retry delay. + assert!( + wait_until( + || !storage.exists(&address).unwrap_or(true), + PROPAGATION_TIMEOUT + ) + .await, + "dispatcher never attempted the faulty read" + ); + fs::set_permissions(&chunk_file, readable).expect("restore chunk mode"); + + assert!( + wait_until_replicated( + &harness, + FRESH_PIPELINE_SOURCE_INDEX, + &address, + FRESH_READ_RETRY_DELAY + PROPAGATION_TIMEOUT + ) + .await, + "write should have replicated on the retried read" + ); + assert!( + storage.exists(&address).unwrap_or(false), + "the successful retry clears the suspect mark on the source" + ); + + harness.teardown().await.expect("teardown"); +} + +/// Proof bytes for tests that only need to reach the pre-payment gates. +fn dummy_payment_proof() -> Vec { + vec![DUMMY_PAYMENT_PROOF_BYTE; DUMMY_PAYMENT_PROOF_LEN] +} + +/// Store `content` on `source_idx` and mark it paid on every node, so a fresh +/// offer for it is accepted wherever it lands. Returns the chunk's address. +async fn store_paid_chunk(harness: &TestHarness, source_idx: usize, content: &[u8]) -> XorName { + let address = compute_address(content); + harness + .test_node(source_idx) + .expect("source node") + .ant_protocol + .as_ref() + .expect("protocol") + .storage() + .put(&address, content) + .await + .expect("put"); + for i in 0..harness.node_count() { + if let Some(protocol) = harness + .test_node(i) + .and_then(|node| node.ant_protocol.as_ref()) + { + protocol.payment_verifier().cache_insert(address); + } + } + address +} + +/// Whether any node other than `source_idx` currently stores `address`. +fn stored_on_another_node(harness: &TestHarness, source_idx: usize, address: &XorName) -> bool { + (0..harness.node_count()) + .filter(|&i| i != source_idx) + .filter_map(|i| harness.test_node(i)) + .filter_map(|node| node.ant_protocol.as_ref()) + .any(|protocol| protocol.storage().exists(address).unwrap_or(false)) +} + +/// Poll until `address` is stored on a node other than `source_idx`, or +/// `budget` runs out. +async fn wait_until_replicated( + harness: &TestHarness, + source_idx: usize, + address: &XorName, + budget: Duration, +) -> bool { + wait_until( + || stored_on_another_node(harness, source_idx, address), + budget, + ) + .await +} + +/// Poll `condition` every `PROPAGATION_POLL_INTERVAL` until it holds or +/// `budget` runs out. +async fn wait_until(mut condition: impl FnMut() -> bool, budget: Duration) -> bool { + let deadline = tokio::time::Instant::now() + budget; + while tokio::time::Instant::now() < deadline { + if condition() { + return true; + } + tokio::time::sleep(PROPAGATION_POLL_INTERVAL).await; + } + false +} + +/// On-disk file of a stored chunk: the store shards `root/chunks/` into +/// subdirectories, so look through them for the file named after the address. +fn chunk_file_path(root_dir: &Path, address: &XorName) -> Option { + let file_name = hex::encode(address); + fs::read_dir(root_dir.join(CHUNKS_DIR_NAME)) + .ok()? + .filter_map(Result::ok) + .map(|shard| shard.path().join(&file_name)) + .find(|candidate| candidate.is_file()) +} + /// ADR-0003: the delayed possession check penalises a responsible peer that /// does NOT hold the chunk, and leaves a peer that DOES hold it unpenalised. /// diff --git a/tests/e2e/testnet.rs b/tests/e2e/testnet.rs index a082efd0..a06344c6 100644 --- a/tests/e2e/testnet.rs +++ b/tests/e2e/testnet.rs @@ -23,6 +23,7 @@ use ant_node::payment::{ QuotingMetricsTracker, }; use ant_node::replication::config::MAX_REPLICATION_MESSAGE_SIZE; +use ant_node::replication::fresh::FreshWriteEvent; use ant_node::storage::{AntProtocol, ChunkStore, ChunkStoreConfig}; use ant_node::{ReplicationConfig, ReplicationEngine}; use bytes::Bytes; @@ -425,6 +426,10 @@ pub struct TestNode { /// Shutdown token for the replication engine. pub replication_shutdown: Option, + + /// Sender feeding the replication engine's fresh-write pipeline, kept so + /// tests can queue writes exactly as the PUT handler does. + pub fresh_write_tx: Option>, } impl TestNode { @@ -1090,6 +1095,7 @@ impl TestNetwork { protocol_task: None, replication_engine: None, replication_shutdown: None, + fresh_write_tx: None, }) } @@ -1333,7 +1339,8 @@ impl TestNetwork { { let shutdown = CancellationToken::new(); let repl_config = self.config.replication_config.clone().unwrap_or_default(); - let (_fresh_tx, fresh_rx) = tokio::sync::mpsc::unbounded_channel(); + let (fresh_tx, fresh_rx) = tokio::sync::mpsc::unbounded_channel(); + node.fresh_write_tx = Some(fresh_tx); let node_identity = Arc::clone(id); match ReplicationEngine::new( repl_config,