diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5e4684f5..d22d5761 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -92,7 +92,7 @@ jobs: anvil --version forge --version - name: Run E2E tests (serial) - run: cargo test -p ant-core --test e2e_chunk --test e2e_data --test e2e_file --test e2e_payment --test e2e_security --test e2e_cost_estimate -- --test-threads=1 + run: cargo test -p ant-core --features test-utils --test e2e_chunk --test e2e_data --test e2e_file --test e2e_payment --test e2e_security --test e2e_cost_estimate -- --test-threads=1 test-merkle: name: Merkle E2E (${{ matrix.os }}) diff --git a/ant-core/Cargo.toml b/ant-core/Cargo.toml index c5a1f215..11a6441c 100644 --- a/ant-core/Cargo.toml +++ b/ant-core/Cargo.toml @@ -85,6 +85,9 @@ default = [] # Enable `LocalDevnet` (ant-core/src/node/devnet.rs) which wraps # `ant_node::devnet::Devnet` and an Anvil EVM testnet. devnet = ["dep:ant-node"] +# Expose test-only client seams (e.g. forcing the ADR-0002 extended PUT +# fallback) used by the e2e/integration test suite. +test-utils = [] [dev-dependencies] # Test infrastructure in tests/support/mod.rs spawns real ant-nodes diff --git a/ant-core/src/data/client/batch.rs b/ant-core/src/data/client/batch.rs index 3470b80d..ef3b0397 100644 --- a/ant-core/src/data/client/batch.rs +++ b/ant-core/src/data/client/batch.rs @@ -43,7 +43,11 @@ pub struct PreparedChunk { pub content: Bytes, /// Content address (BLAKE3 hash). pub address: XorName, - /// Closest peers from quote collection — PUT targets for close-group replication. + /// Ordered PUT targets from quote planning. + /// + /// Kept under the legacy `quoted_peers` name for API compatibility; the + /// list can include non-quoted fallback peers beyond the quoted close + /// group. pub quoted_peers: Vec<(PeerId, Vec)>, /// Payment structure (quotes sorted, median selected, not yet paid on-chain). pub payment: SingleNodePayment, @@ -58,7 +62,11 @@ pub struct PaidChunk { pub content: Bytes, /// Content address (BLAKE3 hash). pub address: XorName, - /// Closest peers from quote collection — PUT targets for close-group replication. + /// Ordered PUT targets from quote planning. + /// + /// Kept under the legacy `quoted_peers` name for API compatibility; the + /// list can include non-quoted fallback peers beyond the quoted close + /// group. pub quoted_peers: Vec<(PeerId, Vec)>, /// Serialized [`PaymentProof`] bytes. pub proof_bytes: Vec, @@ -256,7 +264,8 @@ impl Client { }; let quotes_with_peers = quote_plan.quotes; - // Capture all quoted peers for close-group replication. + // Capture the ordered PUT target set for replication. This can be + // wider than the peers that supplied the paid quotes. let quoted_peers = quote_plan.put_peers; // Build peer_quotes for ProofOfPayment + quotes for SingleNodePayment. diff --git a/ant-core/src/data/client/chunk.rs b/ant-core/src/data/client/chunk.rs index 8f7ac9f6..8e33b5cc 100644 --- a/ant-core/src/data/client/chunk.rs +++ b/ant-core/src/data/client/chunk.rs @@ -13,18 +13,80 @@ use ant_protocol::transport::{MultiAddr, PeerId}; use ant_protocol::{ compute_address, detect_proof_type, send_and_await_chunk_response, ChunkGetRequest, ChunkGetResponse, ChunkMessage, ChunkMessageBody, ChunkPutRequest, ChunkPutResponse, DataChunk, - ProofType, XorName, CLOSE_GROUP_MAJORITY, + ProofType, ProtocolError, XorName, CLOSE_GROUP_MAJORITY, }; use bytes::Bytes; use futures::stream::{self, FuturesUnordered, StreamExt}; use std::collections::HashMap; -use std::future::Future; use std::time::{Duration, Instant}; use tracing::{debug, info, warn}; /// Data type identifier for chunks (used in quote requests). const CHUNK_DATA_TYPE: u32 = 0; +/// Why a single-peer PUT was declined. Drives the surfaced aggregate error +/// and keeps the store AIMD limiter honest — only a transport shortfall is a +/// "client is sending too fast" signal (V2-468); a node that responds with a +/// structured rejection is an application-level decline (ADR-0002). +#[derive(Clone, Copy)] +enum PutRejection { + /// Node is out of storage (`ProtocolError::StorageFailed`) — try a + /// further peer. + Full, + /// Payment did not clear the node's local price floor, or the proof's + /// issuers are not close enough in this peer's view + /// (`ProtocolError::PaymentFailed`), or the node asked for more than was + /// paid (`ChunkPutResponse::PaymentRequired` → [`Error::Payment`]) — skip + /// this peer, do not re-quote. + PriceFloor, + /// Some other structured remote rejection. + OtherRemote, + /// Transport/timeout failure — the node did not respond. + Transport, +} + +/// Classify a failed single-peer PUT (ADR-0002). A `RemotePut` carries the +/// node's structured `ProtocolError`; a `PaymentRequired` response surfaces as +/// [`Error::Payment`]; anything else is a transport failure. +fn classify_put_failure(error: &Error) -> PutRejection { + match error { + Error::RemotePut { source, .. } => match source { + ProtocolError::StorageFailed(_) => PutRejection::Full, + ProtocolError::PaymentFailed(_) => PutRejection::PriceFloor, + _ => PutRejection::OtherRemote, + }, + // A `PaymentRequired` PUT response (the node wants more than was paid) + // arrives as `Error::Payment`. It is a structured application-level + // decline — skip the peer and advance fallback, exactly like a + // price-floor `PaymentFailed` — not a transport shortfall, so it must + // not push the store AIMD limiter down (ADR-0002 / V2-468). + Error::Payment(_) => PutRejection::PriceFloor, + _ => PutRejection::Transport, + } +} + +/// Decide the error for a close-group store that fell short of quorum. +/// +/// When every failure was an application-level decline — the node responded +/// (full / price-floor / `PaymentRequired` / other remote rejection) and there +/// was **no** transport failure — return the representative application error so +/// the shortfall classifies as `ApplicationError` and does not push the store +/// AIMD limiter down as a false capacity signal (ADR-0002 / V2-468). A shortfall +/// that included any transport failure is a genuine capacity signal and surfaces +/// as `InsufficientPeers` (classified `NetworkError`). +fn put_shortfall_error( + transport: usize, + first_app_rejection: Option, + insufficient_peers_message: String, +) -> Error { + if transport == 0 { + if let Some(app_rejection) = first_app_rejection { + return app_rejection; + } + } + Error::InsufficientPeers(insufficient_peers_message) +} + /// Result of one sweep over a chunk's close group. /// /// Either we got the chunk from some peer, or every peer in the group @@ -66,7 +128,7 @@ struct CloseGroupOutcome { /// reach" plus "4 non-storers," not data loss. /// /// 2. *Well-sampled*: at least `CLOSE_GROUP_MAJORITY` peers were -/// queried. `close_group_peers` (via `find_closest_peers`) accepts +/// queried. `closest_peers` (via `find_closest_peers`) accepts /// any non-empty DHT result, so a thin/under-sampled walk can return /// 1 or 2 peers. A `1/1` or `3/3` NotFound from such a walk is NOT /// authoritative — the real replica majority may sit entirely @@ -259,12 +321,51 @@ impl Client { } } - /// Store a chunk to `CLOSE_GROUP_MAJORITY` peers from the quoted set. + /// Test-only: pay for `content`, then store it with `dead_count` + /// unreachable peers prepended to the real put-target set. + /// + /// Every initial send hits a dead peer and fails, so the store can only + /// reach quorum by falling back through the real put-targets (the closest-K + /// set the quote plan already returned), reusing the same `ProofOfPayment`. + /// Pass `dead_count >= CLOSE_GROUP_MAJORITY` so a full quorum's worth of + /// replacements must come from the fallback; a success proves the fallback + /// works end-to-end. + /// + /// # Errors + /// + /// Returns an error if payment fails or quorum cannot be reached. + #[cfg(feature = "test-utils")] + pub async fn chunk_put_with_dead_initial_peers( + &self, + content: Bytes, + dead_count: usize, + ) -> Result { + let address = compute_address(&content); + let data_size = u64::try_from(content.len()) + .map_err(|e| Error::InvalidData(format!("content size too large: {e}")))?; + let (proof, real_peers) = self + .pay_for_storage(&address, data_size, CHUNK_DATA_TYPE) + .await?; + // Unreachable peers (random id, no addresses) first: every initial send + // fails, so quorum can only be reached by falling back through the real + // put-target set that follows. + let mut peers: Vec<(PeerId, Vec)> = (0..dead_count) + .map(|_| (PeerId::random(), Vec::new())) + .collect(); + peers.extend(real_peers); + self.chunk_put_to_close_group(content, proof, &peers).await + } + + /// Store a chunk to `CLOSE_GROUP_MAJORITY` peers, falling back past full or + /// over-priced members of the supplied put-target set (ADR-0002). /// - /// Initially sends the PUT concurrently to the first - /// `CLOSE_GROUP_MAJORITY` peers. If any fail, falls back to the - /// remaining peers in the quoted set until majority is reached or - /// all peers are exhausted. + /// Sends the PUT concurrently to the first `CLOSE_GROUP_MAJORITY` peers. On + /// each failure it advances to the next peer in `peers` — which the caller + /// supplies as the chunk's closest ~K neighbourhood, so no further DHT + /// lookup is needed. Every peer reuses the same payment proof: a node + /// accepts it as long as one of the proof's quote issuers is within that + /// peer's own local closest view, so the client never needs to re-quote or + /// re-pay to route around a full node. /// /// # Errors /// @@ -280,26 +381,32 @@ impl Client { let initial_count = peers.len().min(CLOSE_GROUP_MAJORITY); let (initial_peers, fallback_peers) = peers.split_at(initial_count); + let mut fallback_iter = fallback_peers.iter(); let mut put_futures = FuturesUnordered::new(); for (peer_id, addrs) in initial_peers { - put_futures.push(self.spawn_chunk_put(content.clone(), proof.clone(), peer_id, addrs)); + put_futures.push(self.spawn_chunk_put( + content.clone(), + proof.clone(), + *peer_id, + addrs.clone(), + )); } let mut success_count = 0usize; let mut failures: Vec = Vec::new(); - // Distinguish the *cause* of a quorum shortfall so it feeds the - // store AIMD limiter correctly (V2-468). If every failure was a - // structured remote application rejection (`Error::RemotePut` — the - // node responded and declined: pool-rejected / quote-stale / - // disk-full), the shortfall is not evidence the client is sending - // too fast and must not push the limiter down. Anything else - // (transport failure, or a different error) keeps it a real - // capacity signal. Hold the first remote rejection as the - // representative reason to surface when the shortfall is app-only. - let mut had_non_rejection_failure = false; - let mut first_remote_rejection: Option = None; - let mut fallback_iter = fallback_peers.iter(); + // Tally the *cause* of each failure. The store AIMD limiter must only be + // pushed down by a transport shortfall (V2-468): a node that responds — + // a structured `RemotePut` decline, or `PaymentRequired` surfacing as + // `Error::Payment` — declined at the application layer and is not + // evidence the client is sending too fast. The per-cause counts also + // surface a legible aggregate reason; hold the first application-level + // rejection as the representative error. + let mut full = 0usize; + let mut price_floor = 0usize; + let mut other_remote = 0usize; + let mut transport = 0usize; + let mut first_app_rejection: Option = None; while let Some((peer_id, result)) = put_futures.next().await { match result { @@ -316,14 +423,25 @@ impl Client { Err(e) => { warn!("Failed to store chunk on {peer_id}: {e}"); failures.push(format!("{peer_id}: {e}")); - if matches!(e, Error::RemotePut { .. }) { - if first_remote_rejection.is_none() { - first_remote_rejection = Some(e); - } - } else { - had_non_rejection_failure = true; + match classify_put_failure(&e) { + PutRejection::Full => full += 1, + PutRejection::PriceFloor => price_floor += 1, + PutRejection::OtherRemote => other_remote += 1, + PutRejection::Transport => transport += 1, + } + // An application-level decline is `RemotePut` (a structured + // node rejection) or `Error::Payment` (`PaymentRequired`): + // capture the first so an all-application shortfall surfaces + // as `ApplicationError`, not `InsufficientPeers` + // (`NetworkError`), and never suppresses the limiter. + if matches!(e, Error::RemotePut { .. } | Error::Payment(_)) + && first_app_rejection.is_none() + { + first_app_rejection = Some(e); } + // Advance to the next peer in the put-target set, reusing + // the same proof. if let Some((fb_peer, fb_addrs)) = fallback_iter.next() { debug!( "Falling back to peer {fb_peer} for chunk {}", @@ -332,46 +450,43 @@ impl Client { put_futures.push(self.spawn_chunk_put( content.clone(), proof.clone(), - fb_peer, - fb_addrs, + *fb_peer, + fb_addrs.clone(), )); } } } } - // Quorum not reached. If the only failures were structured remote - // rejections, surface a representative `RemotePut` (classifies - // `ApplicationError`, still recoverable in the merkle retry path) - // so the shortfall doesn't suppress the store limiter. Otherwise - // it's a real capacity shortfall. - if !had_non_rejection_failure { - if let Some(remote_rejection) = first_remote_rejection { - return Err(remote_rejection); - } - } - - Err(Error::InsufficientPeers(format!( - "Stored on {success_count} peers, need {CLOSE_GROUP_MAJORITY}. Failures: [{}]", + // Quorum not reached. An application-only shortfall surfaces the + // representative app error (so it doesn't suppress the limiter); a + // shortfall with any transport failure is a real capacity signal. + let aggregate = format!( + "Stored on {success_count} peers, need {CLOSE_GROUP_MAJORITY} \ + (full: {full}, price-floor: {price_floor}, other-rejection: {other_remote}, \ + transport: {transport}). Failures: [{}]", failures.join("; ") - ))) + ); + Err(put_shortfall_error( + transport, + first_app_rejection, + aggregate, + )) } - /// Spawn a chunk PUT future for a single peer. - fn spawn_chunk_put<'a>( - &'a self, + /// Build a chunk PUT future for a single peer. Takes owned peer data so + /// the future can outlive a fallback queue entry popped per iteration. + async fn spawn_chunk_put( + &self, content: Bytes, proof: Vec, - peer_id: &'a PeerId, - addrs: &'a [MultiAddr], - ) -> impl Future)> + 'a { - let peer_id_owned = *peer_id; - async move { - let result = self - .chunk_put_with_proof(content, proof, &peer_id_owned, addrs) - .await; - (peer_id_owned, result) - } + peer_id: PeerId, + addrs: Vec, + ) -> (PeerId, Result) { + let result = self + .chunk_put_with_proof(content, proof, &peer_id, &addrs) + .await; + (peer_id, result) } /// Store a chunk on the Autonomi network with a pre-built payment proof. @@ -518,7 +633,7 @@ impl Client { let addr_hex = hex::encode(address); // First attempt against the current close-group view. A - // lookup/transport error here (e.g. close_group_peers' DHT walk + // lookup/transport error here (e.g. closest_peers' DHT walk // momentarily returning an error, or InsufficientPeers from a // thin routing table) is NOT fatal: fall through to the retry // path exactly as a non-authoritative miss would. Otherwise one @@ -915,6 +1030,62 @@ mod tests { /// Last byte position in the test XOR distance arrays. const TEST_DISTANCE_TAIL_INDEX: usize = TEST_XORNAME_BYTE_LEN - 1; + #[test] + fn classify_put_failure_maps_remote_and_transport_reasons() { + let remote = |source| Error::RemotePut { + address: "test-addr".to_string(), + source, + }; + assert!(matches!( + classify_put_failure(&remote(ProtocolError::StorageFailed("full".to_string()))), + PutRejection::Full + )); + assert!(matches!( + classify_put_failure(&remote(ProtocolError::PaymentFailed( + "below floor".to_string() + ))), + PutRejection::PriceFloor + )); + assert!(matches!( + classify_put_failure(&remote(ProtocolError::Internal("boom".to_string()))), + PutRejection::OtherRemote + )); + // A `PaymentRequired` PUT response surfaces as `Error::Payment` and is an + // application-level decline, not a transport shortfall (ADR-0002). + assert!(matches!( + classify_put_failure(&Error::Payment("Payment required: more".to_string())), + PutRejection::PriceFloor + )); + assert!(matches!( + classify_put_failure(&Error::Timeout("no response".to_string())), + PutRejection::Transport + )); + } + + #[test] + fn put_shortfall_surfaces_app_error_only_without_transport_failure() { + let app = || Error::Payment("Payment required: more".to_string()); + let msg = || "shortfall".to_string(); + + // Every failure was an application-level decline (e.g. all peers asked + // for more payment): surface the app error so the limiter isn't driven + // down as a false capacity signal (ADR-0002 / V2-468). + assert!(matches!( + put_shortfall_error(0, Some(app()), msg()), + Error::Payment(_) + )); + // A transport failure in the mix: a genuine capacity shortfall. + assert!(matches!( + put_shortfall_error(1, Some(app()), msg()), + Error::InsufficientPeers(_) + )); + // No application rejection captured at all (pure transport): capacity. + assert!(matches!( + put_shortfall_error(0, None, msg()), + Error::InsufficientPeers(_) + )); + } + fn chunk_peer_get_result(peer_seed: u8, distance_tail: u8) -> ChunkPeerGetResult { let mut xor_distance = [0; TEST_XORNAME_BYTE_LEN]; xor_distance[TEST_DISTANCE_TAIL_INDEX] = distance_tail; diff --git a/ant-core/src/data/client/file.rs b/ant-core/src/data/client/file.rs index 05366cc1..12d0ddc3 100644 --- a/ant-core/src/data/client/file.rs +++ b/ant-core/src/data/client/file.rs @@ -2578,7 +2578,7 @@ impl Client { hex::encode(addr) )) })?; - let peers = self.close_group_peers(&addr).await?; + let peers = self.put_target_peers(&addr).await?; observe_op( &limiter, || async move { self.chunk_put_to_close_group(content, proof, &peers).await }, diff --git a/ant-core/src/data/client/merkle.rs b/ant-core/src/data/client/merkle.rs index 450537e0..7957c4f6 100644 --- a/ant-core/src/data/client/merkle.rs +++ b/ant-core/src/data/client/merkle.rs @@ -854,7 +854,7 @@ impl Client { hex::encode(addr) )) })?; - let peers = self.close_group_peers(&addr).await?; + let peers = self.put_target_peers(&addr).await?; observe_op( &limiter, || async move { self.chunk_put_to_close_group(content, proof, &peers).await }, diff --git a/ant-core/src/data/client/mod.rs b/ant-core/src/data/client/mod.rs index 807e28b5..5e54009a 100644 --- a/ant-core/src/data/client/mod.rs +++ b/ant-core/src/data/client/mod.rs @@ -28,6 +28,15 @@ use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; use tracing::debug; +/// Width of the chunk PUT-target set (initial writes plus fallback): the +/// closest `PUT_TARGET_WIDTH` peers to the address. +/// +/// Mirrors the node-side `K_BUCKET_SIZE` / `PAID_QUOTE_ISSUER_CLOSENESS_WIDTH` +/// (20): a node accepts a reused payment proof only when one of the proof's +/// closest-`CLOSE_GROUP_SIZE` quote issuers is within its own local 20-closest, +/// so trying peers past this width is pointless. +pub(crate) const PUT_TARGET_WIDTH: usize = 20; + /// Classify a `data::error::Error` into a controller `Outcome`. /// /// Capacity signals (Timeout / NetworkError) drive the controller @@ -508,16 +517,17 @@ impl Client { self.next_request_id.fetch_add(1, Ordering::Relaxed) } - /// Return all peers in the close group for a target address. + /// Return the chunk PUT-target set: the closest [`PUT_TARGET_WIDTH`] peers + /// to the address, each paired with its known network addresses. /// - /// Queries the DHT for the closest peers by XOR distance. - /// Returns each peer paired with its known network addresses. - pub(crate) async fn close_group_peers( + /// Used by the merkle store path, which — unlike single-node payment — has + /// no witnessed put-target list to forward, so it fetches the closest-K + /// neighbourhood locally. + pub(crate) async fn put_target_peers( &self, target: &XorName, ) -> Result)>> { - self.closest_peers(target, self.config().close_group_size) - .await + self.closest_peers(target, PUT_TARGET_WIDTH).await } /// Return the requested number of closest peers for a target address. diff --git a/ant-core/src/data/client/payment.rs b/ant-core/src/data/client/payment.rs index 3452d599..8bcad918 100644 --- a/ant-core/src/data/client/payment.rs +++ b/ant-core/src/data/client/payment.rs @@ -23,7 +23,7 @@ impl Client { /// Pay for storage and return the serialized payment proof bytes. /// /// This orchestrates the full payment flow: - /// 1. Collect `CLOSE_GROUP_SIZE` quotes from the witnessed close group + /// 1. Collect `CLOSE_GROUP_SIZE` quotes plus ordered PUT targets /// 2. Build `SingleNodePayment` using node-reported prices (median 3x, others 0) /// 3. Pay on-chain via the wallet /// 4. Serialize `PaymentProof` with transaction hashes @@ -32,9 +32,10 @@ impl Client { /// /// Returns an error if the wallet is not set, quotes cannot be collected, /// on-chain payment fails, or serialization fails. - /// Returns `(proof_bytes, quoted_peers)`. `quoted_peers` are the - /// `CLOSE_GROUP_SIZE` peers that provided quotes — callers should store - /// the chunk to at least `CLOSE_GROUP_MAJORITY` of these peers. + /// Returns `(proof_bytes, put_targets)`. The peer list is the ordered PUT + /// target set from quote planning: it starts with peers expected to accept + /// the paid proof and can include non-quoted fallback peers beyond the + /// quoted close group. pub async fn pay_for_storage( &self, address: &[u8; 32], @@ -59,7 +60,8 @@ impl Client { ) })?; - // Capture all quoted peers for replication by the caller. + // Capture the ordered PUT target set for replication by the caller. + // This can be wider than the peers that supplied the paid quotes. let quoted_peers = quote_plan.put_peers; // 2. Build peer_quotes for ProofOfPayment + quotes for SingleNodePayment. diff --git a/ant-core/src/data/client/quote.rs b/ant-core/src/data/client/quote.rs index f4bc38ea..68ff69b7 100644 --- a/ant-core/src/data/client/quote.rs +++ b/ant-core/src/data/client/quote.rs @@ -5,9 +5,12 @@ use crate::data::client::peer_xor_distance; use crate::data::client::Client; +use crate::data::client::PUT_TARGET_WIDTH; use crate::data::error::{Error, Result}; use ant_protocol::evm::{Amount, PaymentQuote}; -use ant_protocol::transport::{DHTNode, MultiAddr, P2PNode, PeerId, WitnessedCloseGroup}; +use ant_protocol::transport::{ + DHTNode, MultiAddr, P2PNode, PeerId, ResponderView, WitnessedCloseGroup, +}; use ant_protocol::{ compute_address, send_and_await_chunk_response, ChunkMessage, ChunkMessageBody, ChunkQuoteRequest, ChunkQuoteResponse, CLOSE_GROUP_MAJORITY, CLOSE_GROUP_SIZE, @@ -287,6 +290,36 @@ fn witnessed_close_group_quorum_for_transcript(witnessed: &WitnessedCloseGroup) witnessed_close_group_quorum_for_missing_views(missing_witnessed_responder_views(witnessed)) } +/// Restrict a witnessed transcript to its closest `CLOSE_GROUP_SIZE` peers. +/// +/// The witnessed query is widened to `PUT_TARGET_WIDTH` peers so we +/// have addresses for the full PUT-target set, but the consensus/quorum/quote +/// logic must still run on the close group only. Keeping just the closest-7 +/// initial peers and the responder views contributed by those peers leaves the +/// `missing_witnessed_responder_views` math — and the quorum derived from it — +/// byte-for-byte identical to a `CLOSE_GROUP_SIZE`-wide query. +fn scope_witnessed_to_close_group(witnessed: &WitnessedCloseGroup) -> WitnessedCloseGroup { + let initial_closest: Vec = witnessed + .initial_closest + .iter() + .take(CLOSE_GROUP_SIZE) + .cloned() + .collect(); + let scope: HashSet = initial_closest.iter().map(|node| node.peer_id).collect(); + let responder_views: Vec = witnessed + .responder_views + .iter() + .filter(|view| scope.contains(&view.responder)) + .cloned() + .collect(); + WitnessedCloseGroup { + target: witnessed.target, + k: CLOSE_GROUP_SIZE, + initial_closest, + responder_views, + } +} + fn peer_list(peers: &[PeerId]) -> Vec { peers.iter().map(ToString::to_string).collect() } @@ -801,30 +834,56 @@ impl Client { &self, address: &[u8; 32], ) -> Result { + // The quote/quorum/consensus scope is the closest CLOSE_GROUP_SIZE. let required = single_node_quote_query_count(); - let witnessed = self + // Contact the closest PUT_TARGET_WIDTH peers directly so the whole + // PUT-target set's addresses arrive in this single query. A network + // with fewer than that near the target can't satisfy the wide lookup, + // so fall back to the close-group width — the upload still proceeds with + // a narrower (but valid) PUT-target set rather than failing. + let witnessed = match self .network() .find_witnessed_close_group_with_view_count( address, - required, + PUT_TARGET_WIDTH, SINGLE_NODE_WITNESSED_VIEW_COUNT, ) .await - .map_err(|e| { - Error::InsufficientPeers(format!( - "Witnessed close group lookup failed before payment for target {}: {e}", - hex::encode(address) - )) - })?; + { + Ok(witnessed) => witnessed, + Err(wide_err) => { + debug!( + target = %hex::encode(address), + "Wide witnessed lookup ({PUT_TARGET_WIDTH}) failed ({wide_err}); \ + retrying at close-group width ({required})" + ); + self.network() + .find_witnessed_close_group_with_view_count( + address, + required, + SINGLE_NODE_WITNESSED_VIEW_COUNT, + ) + .await + .map_err(|e| { + Error::InsufficientPeers(format!( + "Witnessed close group lookup failed before payment for target {}: {e}", + hex::encode(address) + )) + })? + } + }; + // Run quoting/quorum on the closest CLOSE_GROUP_SIZE only, so payment + // semantics are unaffected by the wider PUT query. + let witnessed_quote = scope_witnessed_to_close_group(&witnessed); let base_quorum = witnessed_close_group_quorum(); - let missing_views = missing_witnessed_responder_views(&witnessed); - let quorum = witnessed_close_group_quorum_for_transcript(&witnessed); + let missing_views = missing_witnessed_responder_views(&witnessed_quote); + let quorum = witnessed_close_group_quorum_for_transcript(&witnessed_quote); if missing_views > 0 { warn!( target = %hex::encode(address), - initial = witnessed.initial_closest.len(), - responder_views = witnessed.responder_views.len(), + initial = witnessed_quote.initial_closest.len(), + responder_views = witnessed_quote.responder_views.len(), missing_views = missing_views, base_quorum = base_quorum, adjusted_quorum = quorum, @@ -836,14 +895,25 @@ impl Client { target = %hex::encode(address), quorum = quorum, view_count = SINGLE_NODE_WITNESSED_VIEW_COUNT, - initial = ?witnessed_initial_peers(&witnessed), - responder_views = ?witnessed_responder_views(&witnessed), - vote_counts = ?witnessed_vote_counts(&witnessed, address), - final_witnessed_set = ?witnessed_consensus(&witnessed, address, quorum), + initial = ?witnessed_initial_peers(&witnessed_quote), + responder_views = ?witnessed_responder_views(&witnessed_quote), + vote_counts = ?witnessed_vote_counts(&witnessed_quote, address), + final_witnessed_set = ?witnessed_consensus(&witnessed_quote, address, quorum), "Witnessed close group selected for SNP quote collection" ); - witnessed_quote_selection_or_error(address, &witnessed, required, quorum) + let mut selection = + witnessed_quote_selection_or_error(address, &witnessed_quote, required, quorum)?; + // Widen the PUT-target set to the closest PUT_TARGET_WIDTH + // directly-contacted peers; the quote set above stays the closest + // CLOSE_GROUP_SIZE. The same proof is reused on all of them. + selection.initial_put_peers = witnessed + .initial_closest + .iter() + .take(PUT_TARGET_WIDTH) + .map(|node| (node.peer_id, node.addresses_by_priority())) + .collect(); + Ok(selection) } #[allow(clippy::too_many_lines)] @@ -1385,6 +1455,116 @@ mod tests { ); } + /// Ascending seeds `1..=count`, each a valid `u8` peer seed. + fn ascending_seeds(count: usize) -> Vec { + (1..=count) + .map(|n| u8::try_from(n).expect("test seed fits in u8")) + .collect() + } + + #[test] + fn scope_witnessed_to_close_group_matches_native_close_group_query() { + // How many of the closest-`CLOSE_GROUP_SIZE` responders returned a view. + // The remainder are "missing", so the scoped transcript also exercises + // the missing-views quorum adjustment. + const RESPONDED_IN_SCOPE: usize = 5; + // Responders past the close group whose views scoping must drop. + const OUT_OF_SCOPE_RESPONDERS: usize = 2; + + let address = [0u8; 32]; + let close_seeds = ascending_seeds(CLOSE_GROUP_SIZE); + // Each view's closest list mixes in-group (1, 2) and far (8, 9) peers so + // candidate selection is non-trivial and must survive scoping verbatim. + let view_closest = [1, 2, 8, 9]; + let in_scope_views = || -> Vec { + ascending_seeds(RESPONDED_IN_SCOPE) + .into_iter() + .map(|responder| witnessed_test_view(responder, &view_closest)) + .collect() + }; + + // A wide PUT_TARGET_WIDTH-peer transcript, ordered closest-first (seed n + // == PeerId [n; 32], whose XOR distance to the zero address is n). Two + // responders past the close group (out of scope) must be dropped. + let mut wide_views = in_scope_views(); + for offset in 1..=OUT_OF_SCOPE_RESPONDERS { + let responder = + u8::try_from(CLOSE_GROUP_SIZE + offset).expect("out-of-scope seed fits in u8"); + wide_views.push(witnessed_test_view(responder, &[1, 2, 3])); + } + let wide = WitnessedCloseGroup { + target: address, + k: PUT_TARGET_WIDTH, + initial_closest: witnessed_test_nodes(&ascending_seeds(PUT_TARGET_WIDTH)), + responder_views: wide_views, + }; + + // The hand-built equivalent: a native CLOSE_GROUP_SIZE-wide query with + // the same in-scope responders. + let native = WitnessedCloseGroup { + target: address, + k: CLOSE_GROUP_SIZE, + initial_closest: witnessed_test_nodes(&close_seeds), + responder_views: in_scope_views(), + }; + + let scoped = scope_witnessed_to_close_group(&wide); + + // Target preserved; k and the initial set collapse to the close group. + assert_eq!(scoped.target, wide.target); + assert_eq!(scoped.k, CLOSE_GROUP_SIZE); + assert_eq!( + scoped + .initial_closest + .iter() + .map(|node| node.peer_id.as_bytes()[0]) + .collect::>(), + close_seeds, + "initial set must be the closest CLOSE_GROUP_SIZE, in order" + ); + + // Out-of-close-group responder views are dropped; the in-scope ones keep + // their closest lists untouched (scoping filters by responder only). + assert_eq!( + scoped + .responder_views + .iter() + .map(|view| view.responder.as_bytes()[0]) + .collect::>(), + ascending_seeds(RESPONDED_IN_SCOPE), + "only responders inside the close group survive" + ); + assert_eq!( + scoped.responder_views[0] + .closest + .iter() + .map(|node| node.peer_id.as_bytes()[0]) + .collect::>(), + view_closest.to_vec(), + "a surviving view's closest set must be preserved verbatim" + ); + + // The quorum math and candidate consensus run on the close group only + // and are byte-for-byte identical to the native CLOSE_GROUP_SIZE query. + assert_eq!( + missing_witnessed_responder_views(&scoped), + missing_witnessed_responder_views(&native), + ); + let quorum = witnessed_close_group_quorum_for_transcript(&scoped); + assert_eq!(quorum, witnessed_close_group_quorum_for_transcript(&native)); + let candidate_seeds = |group: &WitnessedCloseGroup| { + witnessed_consensus_candidates(group, &address, quorum) + .iter() + .map(|candidate| candidate.node.peer_id.as_bytes()[0]) + .collect::>() + }; + assert_eq!( + candidate_seeds(&scoped), + candidate_seeds(&native), + "scoped consensus must match a native close-group query" + ); + } + #[test] fn witnessed_quote_peers_error_is_typed_and_pre_payment_when_consensus_is_short() { let address = [0u8; 32]; diff --git a/ant-core/tests/e2e_chunk.rs b/ant-core/tests/e2e_chunk.rs index 412f005f..e5c04f1e 100644 --- a/ant-core/tests/e2e_chunk.rs +++ b/ant-core/tests/e2e_chunk.rs @@ -50,6 +50,54 @@ async fn test_chunk_put_get_round_trip() { testnet.teardown().await; } +/// ADR-0002: when initial PUT targets fail, the client falls back to the +/// chunk's genuine next-closest peers — reusing the same `ProofOfPayment`, +/// without re-quoting or re-paying — and still reaches quorum. +/// +/// Forces `DEAD_INITIAL_PEERS` (= `CLOSE_GROUP_MAJORITY`) unreachable initial +/// peers, so every initial send fails and a full quorum's worth of stores must +/// come from the *real* next-closest peers in the put-target set. A successful, +/// retrievable store proves the fallback reached those real peers on the single +/// proof produced by one payment. +/// +/// Scope: failures here are driven by *unreachable* peers (transport), which +/// exercises the fallback walk + single-proof reuse + quorum end-to-end. The +/// *classification* that routes a full node's `StorageFailed` and a price-floor +/// `PaymentFailed`/`PaymentRequired` into the same fallback is covered by the +/// `classify_put_failure_*` unit test. A live close-group node returning +/// `StorageFailed` (a genuinely full member) needs a node-side capacity hook the +/// in-process MiniTestnet does not expose — tracked as a follow-up. +#[cfg(feature = "test-utils")] +#[tokio::test(flavor = "multi_thread")] +#[serial] +async fn test_chunk_put_extended_fallback_reaches_quorum() { + /// Unreachable initial peers to inject; equals `CLOSE_GROUP_MAJORITY` so a + /// full quorum must be served by the extended fallback. + const DEAD_INITIAL_PEERS: usize = 4; + + let (client, testnet) = setup().await; + + let content = Bytes::from("adr-0002 extended fallback payload"); + let address = compute_address(&content); + + let stored = client + .chunk_put_with_dead_initial_peers(content.clone(), DEAD_INITIAL_PEERS) + .await + .expect("extended fallback should reach quorum despite dead initial peers"); + assert_eq!(stored, address, "stored address should be BLAKE3(content)"); + + // Retrievable -> quorum was genuinely met via the fallback peers. + let retrieved = client + .chunk_get(&address) + .await + .expect("chunk_get should succeed"); + let chunk = retrieved.expect("chunk should be found after fallback store"); + assert_eq!(chunk.content.as_ref(), content.as_ref()); + + drop(client); + testnet.teardown().await; +} + #[tokio::test(flavor = "multi_thread")] #[serial] async fn test_chunk_put_duplicate_is_idempotent() { diff --git a/docs/adr/ADR-0002-client-fallback-for-full-node-shunning.md b/docs/adr/ADR-0002-client-fallback-for-full-node-shunning.md new file mode 100644 index 00000000..0f1adf8f --- /dev/null +++ b/docs/adr/ADR-0002-client-fallback-for-full-node-shunning.md @@ -0,0 +1,144 @@ +# ADR-0002: Client-side fallback and diagnostics for full-node shunning + +- **Status:** Proposed +- **Date:** 2026-06-25 +- **Decision owners:** Mick +- **Reviewers:** +- **Supersedes:** none +- **Superseded by:** none +- **Related:** saorsa-node ADR-0003 (node-side full-node detection, penalisation, and eviction); ADR-0001 (adopt ADRs) + +## Context + +Full storage nodes are rejecting client uploads. When a chunk's close group +contains a full node, the client's put to that node fails and the upload can fall +short of write quorum, even though the network as a whole still has capacity. This +ADR covers the **client's** role in the wider full-node-shunning plan; the node and +membership roles are covered by saorsa-node ADR-0003. + +The decision rests on behaviour verified directly in the node and client code, not +on assumption: + +- The client selects peers via a DHT lookup of the `CLOSE_GROUP_SIZE = 7` closest, + and write quorum is `CLOSE_GROUP_MAJORITY = 4` (`ant-core/src/data/client/chunk.rs:308`). + Both constants come from `ant-protocol` and are not client-configurable. +- Today the client only falls back **among the 7 quoted peers** + (`ant-core/src/data/client/chunk.rs:273-358`). +- A node accepts a client put from **any** peer whose own local 20-closest view of + the address includes one of the proof's quote issuers; it does **not** require the + receiving node to have been quoted (`saorsa-node` `src/payment/verifier.rs:811-837, + 942-1003`; `src/storage/handler.rs:283-285`). The issuer-closeness width is + `PAID_QUOTE_ISSUER_CLOSENESS_WIDTH = K_BUCKET_SIZE = 20`, nearly 3× the close + group. **Consequence: the same `ProofOfPayment` is reusable by further peers + within that 20-wide window — fallback needs no re-quote and no re-pay.** +- A full node returns a **distinct** `ProtocolError::StorageFailed` *before* payment + verification (`saorsa-node` `src/storage/handler.rs:274-281`), whereas a price-floor + shortfall returns a `Payment` error. The client currently **flattens both** into + `Error::RemotePut` (`ant-core/src/data/client/chunk.rs:417-443`), losing the + distinction it needs to respond correctly. +- GET returns on **first success** and is read-safe while at least one queried close + peer holds the chunk; it does **not** expand its walk beyond K + (`ant-core/src/data/client/chunk.rs:483-623`). + +## Decision Drivers + +- Unblock uploads when a *minority* of the close group is full, with no protocol or + payment changes. +- Make the cause of a put failure legible so the client picks the right response + (fall back, skip, or retry) instead of one opaque error. +- Stay strictly within the node's verified acceptance rules; never assume behaviour + the node does not implement. +- Stay read-safe in the common case and flag the near-capacity boundary explicitly + rather than silently relying on it. + +## Considered Options + +1. **Do nothing on the client; rely solely on node-side eviction to reshape close + groups.** Rejected: leaves an immediate failure during the eviction convergence + window and wastes the already-available 20-wide acceptance headroom. +2. **Re-quote and re-pay a fresh close group whenever a put fails.** Rejected + outright, and a deliberate non-goal: it is unnecessary now that the proof is known + to be reusable, and **we do not re-quote or top-up payment** under any failure. A + peer whose local floor the existing payment does not clear is simply skipped. +3. **Extend client fallback to next-closest peers within the issuer-closeness + window, reusing the existing proof, with error-class-aware handling (chosen).** + +## Decision + +- **Classify node rejections** instead of collapsing them into one `RemotePut`: + `StorageFailed` (full) → fall back to a further peer; `Payment`/price-floor → this + peer wants more than was paid, so **skip it and advance fallback** (we do **not** + re-quote or top-up payment, and do not retry the same peer); transport/timeout → + bounded retry. +- **Extend `chunk_put_to_close_group` fallback** from the 7 quoted peers to the + next-closest peers from the same DHT walk, **reusing the same `ProofOfPayment`**, + bounded by the `K_BUCKET_SIZE = 20` issuer-closeness window — beyond it the node + provably rejects ("issuer not among this node's local K=20 closest"). Fallback is + **best-effort**: whether a given far peer accepts depends on its own routing view + and local price floor, which the client cannot see. +- **Quote/pay the quorum-witnessed close group, not the raw initial list.** The + `CLOSE_GROUP_SIZE` peers that are quoted and paid are the *witnessed consensus* + close group: the peers a quorum of the closest responders agree are closest to the + address, taken in XOR order. This is intentionally **not** strictly the querying + node's initial closest-`CLOSE_GROUP_SIZE` list — the witnessed/consensus model + exists precisely so that a stale or biased local view cannot unilaterally pick the + pay set, so a consensus peer surfaced in responder views may stand in for a stale + initial peer. The widened put-target query (closest `20`) only *enlarges the put set + and the responder pool*; the quote/quorum transcript is still scoped to + `CLOSE_GROUP_SIZE`, leaving payment byte-for-byte unchanged. +- **Keep write quorum at `CLOSE_GROUP_MAJORITY = 4`.** Fallback changes *which* peers + satisfy quorum, not the threshold. +- **Read availability:** rely on close-group convergence driven by node/membership + eviction (saorsa-node ADR-0003) as the primary guarantee; keep GET-on-first-success + plus the existing single retry. Do **not** widen the GET walk now. Explicitly flag + the near-capacity read-gap — a chunk that lands *entirely* on fallback peers outside + the queried 7 is unreadable until convergence shifts those peers into the close + group — and gate any future GET-walk widening behind observed GET shortfalls in the + near-capacity regime. + +## Consequences + +### Positive + +- Uploads blocked by a single (or minority) full close peer now reach quorum via + fallback, with no protocol, payment, or node change required. +- Failure causes are legible: full, under-paid, and transport failures get distinct, + correct client responses. + +### Negative / Trade-offs + +- Fallback is best-effort and **bounded at the 20-wide window**; if more than + `20 − 4` of the nearest peers refuse, quorum still cannot be met — the same + near-capacity boundary the whole plan is bounded by. +- A **per-node price floor** can reject a fully-paid put even with free disk; because + we do not re-quote or top-up, such a peer is simply skipped — so a neighbourhood + priced above the paid median can still cause a quorum shortfall this ADR does not + resolve. +- The near-capacity read-gap remains until convergence; we accept it for now rather + than widen GET prematurely. +- More client-side branching and a retry/fallback budget to tune. + +### Neutral / Operational + +- Quorum and close-group constants remain owned by `ant-protocol`; this ADR does not + change them. +- Adds one tunable: the fallback peer budget (how far down the next-closest list the + client will try before giving up). + +## Validation + +- In a minority-full testnet, uploads that previously failed on a single full close + peer reach quorum via fallback **without re-quoting or re-paying**. +- Tests required before this ADR is Accepted: error classification maps + `StorageFailed` / `Payment` / transport correctly; fallback stops at the 20-window; + quorum still requires 4 successes; a price-floor rejection skips the peer and + advances fallback without re-quoting or retrying the same peer; GET stays read-safe + while ≥1 queried close peer holds the chunk. +- Re-open trigger: if observed GET shortfalls reveal a persistent read-gap in the + near-capacity regime, revisit widening the GET walk. + +## 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 this one.