diff --git a/Cargo.lock b/Cargo.lock index 1eb3d36..c31fb46 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -838,7 +838,7 @@ version = "0.7.0" dependencies = [ "alloy", "ant-node", - "ant-protocol", + "ant-protocol 2.3.3 (git+https://github.com/grumbach/ant-protocol?branch=reapply%2Fpr-23-settlement-version)", "anyhow", "async-stream", "axum", @@ -895,7 +895,7 @@ version = "0.17.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "722c2b8c9c867e0c78c20f3a82ae7244530f76caf78e16e8cebb049149bbc5e7" dependencies = [ - "ant-protocol", + "ant-protocol 2.3.3 (registry+https://github.com/rust-lang/crates.io-index)", "bao", "blake3", "bytes", @@ -959,6 +959,24 @@ dependencies = [ "tracing", ] +[[package]] +name = "ant-protocol" +version = "2.3.3" +source = "git+https://github.com/grumbach/ant-protocol?branch=reapply%2Fpr-23-settlement-version#a22897d7a3048c41f8c419975274151c9ac9a843" +dependencies = [ + "blake3", + "bytes", + "evmlib", + "hex", + "postcard", + "rmp-serde", + "saorsa-core", + "saorsa-pqc 0.5.1", + "serde", + "tokio", + "tracing", +] + [[package]] name = "anyhow" version = "1.0.102" @@ -3339,7 +3357,7 @@ dependencies = [ "js-sys", "log", "wasm-bindgen", - "windows-core 0.57.0", + "windows-core 0.58.0", ] [[package]] diff --git a/ant-core/Cargo.toml b/ant-core/Cargo.toml index 94d6e85..1807962 100644 --- a/ant-core/Cargo.toml +++ b/ant-core/Cargo.toml @@ -37,7 +37,10 @@ tower-http = { version = "0.6.8", features = ["cors"] } # under `ant_protocol::{evm, transport, pqc}`. This is the ONE pin for # those three deps — do not add direct evmlib/saorsa-core/saorsa-pqc # deps here or the version can skew between ant-client and ant-node. -ant-protocol = "2.3.3" +# Branch pin while the settlement-version wire types are in review +# (WithAutonomi/ant-protocol#25). Swap back to a published `ant-protocol` +# version pin once that PR merges and the release train publishes it. +ant-protocol = { git = "https://github.com/grumbach/ant-protocol", branch = "reapply/pr-23-settlement-version" } xor_name = "5" self_encryption = "0.36" futures = "0.3" diff --git a/ant-core/src/data/client/batch.rs b/ant-core/src/data/client/batch.rs index 9a62b4c..a86a2e8 100644 --- a/ant-core/src/data/client/batch.rs +++ b/ant-core/src/data/client/batch.rs @@ -352,6 +352,16 @@ impl Client { /// /// Returns an error if quote collection or payment construction fails. pub async fn prepare_chunk_payment(&self, content: Bytes) -> Result> { + // A refusal established by any earlier upload on this client stops this + // one before it quotes. The verdict is about this build, not about one + // operation, so the wave path has to honour it exactly as the + // single-node path does in `pay_for_storage`. Checked here as well as + // at the spend because a prepared chunk is also what the external + // signer is handed, and handing one out is telling a user to pay. + if let Some(refusal) = self.corroborated_settlement_refusal() { + return Err(Error::ClientUpdateRequired(refusal)); + } + let address = compute_address(&content); let data_size = u64::try_from(content.len()) .map_err(|e| Error::InvalidData(format!("content size too large: {e}")))?; @@ -422,6 +432,13 @@ impl Client { return Ok((Vec::new(), "0".to_string(), 0)); } + // Re-checked immediately before the spend, not only at preparation. + // Waves are pipelined, so chunks quoted for wave N+1 while wave N is + // still storing can have been prepared before a refusal landed. + if let Some(refusal) = self.corroborated_settlement_refusal() { + return Err(Error::ClientUpdateRequired(refusal)); + } + let wallet = self.require_wallet()?; // Compute total storage cost from the prepared chunks before paying. diff --git a/ant-core/src/data/client/file.rs b/ant-core/src/data/client/file.rs index 8e51c51..988a6fd 100644 --- a/ant-core/src/data/client/file.rs +++ b/ant-core/src/data/client/file.rs @@ -998,6 +998,56 @@ fn fold_single_wave( } } +/// Shape a corroborated settlement refusal that landed on a wave **after** an +/// earlier wave had already paid and stored. +/// +/// The refusal is terminal for this build, but by wave two or later it is no +/// longer true that nothing was charged: the single-node path pays each wave +/// before storing it, so the earlier waves' spend has settled on-chain. Bare +/// `ClientUpdateRequired` would report the upload as costing nothing and drop +/// the stored set a resume needs, so the refusal is surfaced as a +/// `PartialUpload` carrying the real spend and the stored chunks, with the +/// storer's upgrade instruction kept in the reason. Every remaining chunk is +/// listed as failed: none of it was quoted, let alone paid for. +#[allow(clippy::too_many_arguments)] +fn settlement_refusal_after_paid_waves( + refusal: &str, + wave_num: usize, + wave_count: usize, + stored_addresses: Vec<[u8; 32]>, + total_stored: usize, + remaining: &[[u8; 32]], + total_chunks: usize, + total_storage: Amount, + total_gas: u128, +) -> Error { + let remaining_count = remaining.len(); + let refused_note = format!( + "not quoted: storers refused this client's settlement version at wave \ + {wave_num}/{wave_count}" + ); + let failed: Vec<([u8; 32], String)> = remaining + .iter() + .map(|addr| (*addr, refused_note.clone())) + .collect(); + Error::PartialUpload { + stored: stored_addresses, + stored_count: total_stored, + failed, + failed_count: remaining_count, + total_chunks, + spend: Box::new(PartialUploadSpend { + storage_cost_atto: total_storage.to_string(), + gas_cost_wei: total_gas, + }), + reason: format!( + "storers refused this client's settlement version at wave {wave_num}/{wave_count}: \ + the {total_stored} chunk(s) in earlier wave(s) were already paid for and stored, \ + and the remaining {remaining_count} chunk(s) were neither quoted nor paid. {refusal}" + ), + } +} + /// Check that the spill directory has enough free space for the spilled chunks. /// /// `file_size` is the source file's byte count. We require @@ -3243,9 +3293,9 @@ impl Client { } // Fold this wave's result. A quorum shortfall (`PartialUpload`) is // recoverable and its parts are returned to be recorded here; - // genuinely fatal errors propagate via `?` and abort the file, as in + // genuinely fatal errors abort the file, as in // `upload_merkle_from_spill`. - let outcome = fold_single_wave( + let outcome = match fold_single_wave( self.batch_upload_chunks_with_events( wave_data, progress, @@ -3254,7 +3304,29 @@ impl Client { resume_key, ) .await, - )?; + ) { + Ok(outcome) => outcome, + // A corroborated settlement refusal is terminal, but on any wave + // after the first it lands after earlier waves have paid and + // stored. Bare, it would report the upload as costing nothing + // and lose the stored set; carry both instead. On the first + // wave nothing has been paid, so the bare refusal — whose + // wording says nothing was charged — is exactly right. + Err(Error::ClientUpdateRequired(refusal)) if wave_idx > 0 => { + return Err(settlement_refusal_after_paid_waves( + &refusal, + wave_num, + wave_count, + stored_addresses, + total_stored, + &addresses[wave_idx * UPLOAD_WAVE_SIZE..], + total_chunks, + total_storage, + total_gas, + )); + } + Err(e) => return Err(e), + }; if !outcome.failed.is_empty() { warn!( @@ -4798,6 +4870,65 @@ mod tests { ); } + /// A settlement refusal on a later wave must not be reported as if nothing + /// was charged: the earlier waves paid before storing. The refusal is + /// reshaped into a `PartialUpload` that carries the real spend, the stored + /// set (for resume), every un-quoted chunk as failed, and the storer's + /// upgrade instruction in the reason. + #[test] + fn settlement_refusal_after_paid_waves_carries_spend_and_upgrade_instruction() { + let refusal = "your client is too old to pay the current storage rate. Run `ant update`"; + let stored = vec![[1u8; 32], [2u8; 32]]; + let remaining = [[3u8; 32], [4u8; 32], [5u8; 32]]; + + let err = settlement_refusal_after_paid_waves( + refusal, + 2, + 3, + stored.clone(), + stored.len(), + &remaining, + 5, + Amount::from(700u64), + 13, + ); + + let Error::PartialUpload { + stored: got_stored, + stored_count, + failed, + failed_count, + total_chunks, + spend, + reason, + } = err + else { + panic!("expected PartialUpload, got: {err:?}"); + }; + assert_eq!(got_stored, stored); + assert_eq!(stored_count, 2); + assert_eq!(failed_count, 3); + assert_eq!(total_chunks, 5); + // Every un-quoted chunk is listed, none of them as "stored". + let failed_addrs: Vec<[u8; 32]> = failed.iter().map(|(a, _)| *a).collect(); + assert_eq!(failed_addrs, remaining.to_vec()); + assert!(failed.iter().all(|(_, why)| why.contains("not quoted"))); + // The spend is what the earlier waves actually paid, not zero. + assert_eq!(spend.storage_cost_atto, "700"); + assert_eq!(spend.gas_cost_wei, 13); + // The user learns both facts: earlier waves paid, and how to upgrade. + assert!(reason.contains("wave 2/3"), "reason: {reason}"); + assert!( + reason.contains("2 chunk(s) in earlier wave(s) were already paid"), + "reason: {reason}" + ); + assert!( + reason.contains("3 chunk(s) were neither quoted nor paid"), + "reason: {reason}" + ); + assert!(reason.contains(refusal), "reason: {reason}"); + } + #[test] fn partition_addresses_by_proof_handles_all_or_nothing() { let a = [5u8; 32]; diff --git a/ant-core/src/data/client/merkle.rs b/ant-core/src/data/client/merkle.rs index 089ff38..29e882a 100644 --- a/ant-core/src/data/client/merkle.rs +++ b/ant-core/src/data/client/merkle.rs @@ -7,6 +7,7 @@ use crate::data::client::adaptive::{observe_op, Outcome}; use crate::data::client::classify_error; use crate::data::client::file::UploadEvent; +use crate::data::client::quote::settlement_refusal_error; use crate::data::client::Client; use crate::data::error::{Error, Result}; use ant_protocol::evm::{ @@ -23,12 +24,14 @@ use ant_protocol::payment::{ use ant_protocol::transport::PeerId; use ant_protocol::{ compute_address, send_and_await_chunk_response, ChunkMessage, ChunkMessageBody, - MerkleCandidateQuoteRequest, MerkleCandidateQuoteResponse, + MerkleCandidateQuoteRequest, MerkleCandidateQuoteRequestV2, MerkleCandidateQuoteResponse, + ProtocolError, }; use bytes::Bytes; use futures::stream::{self, FuturesUnordered, StreamExt}; use rand::Rng; use std::collections::{HashMap, VecDeque}; +use std::sync::Arc; use std::time::Duration; use tokio::sync::mpsc; use tracing::{debug, info, warn}; @@ -162,6 +165,131 @@ fn pool_commitment_with_payment_multiplier( Ok(commitment) } +/// Turn a merkle candidate quote response into the candidate it carries, or +/// the error that explains why there is none. +/// +/// Shared by the versioned request and its legacy retry so the two cannot +/// interpret the same response differently. +/// +/// `ClientUpdateRequired` is lifted out of the generic protocol-error case +/// deliberately. It is the one rejection that is about this client rather than +/// about this request, it already carries wording aimed at the person running +/// the upload, and it must not be retried in a shape that would get a quote +/// anyway. +/// Which failure a fully drained set of candidate pools should report. +/// +/// Pools are drained rather than short-circuited, so more than one can fail. +/// A refusal outranks an ordinary failure: if one pool runs out of peers while +/// another declares this client unable to settle, the second is the answer that +/// keeps the caller from falling back to a payment path and spending. +#[derive(Default)] +struct PoolVerdict { + refusal: Option, + first_failure: Option, +} + +impl PoolVerdict { + fn note(&mut self, e: Error) { + match e { + e @ Error::ClientUpdateRequired(_) => { + if self.refusal.is_none() { + self.refusal = Some(e); + } + } + e => { + if self.first_failure.is_none() { + self.first_failure = Some(e); + } + } + } + } + + fn into_error(self) -> Option { + self.refusal.or(self.first_failure) + } +} + +fn map_merkle_candidate_response( + peer_id: PeerId, + body: ChunkMessageBody, +) -> Option>)>> { + match body { + ChunkMessageBody::MerkleCandidateQuoteResponse(MerkleCandidateQuoteResponse::Success { + candidate_node, + commitment, + }) => match rmp_serde::from_slice::(&candidate_node) { + Ok(node) => Some(Ok((node, commitment))), + Err(e) => Some(Err(Error::Serialization(format!( + "Failed to deserialize candidate node from {peer_id}: {e}" + )))), + }, + // Validated, not taken at face value. A refusal is unauthenticated and + // two of them latch this client for its whole run, so the merkle path + // applies the same coherence check as the single-node one: the echoed + // version must be ours and the stated minimum must genuinely exceed it. + // Without it, two faulty or hostile candidates could deny every upload + // with a refusal that does not describe this client at all. + ChunkMessageBody::MerkleCandidateQuoteResponse(MerkleCandidateQuoteResponse::Error( + ProtocolError::ClientUpdateRequired { + client_settlement_version, + min_settlement_version, + }, + )) => Some(Err(settlement_refusal_error( + &peer_id, + client_settlement_version, + min_settlement_version, + ))), + ChunkMessageBody::MerkleCandidateQuoteResponse(MerkleCandidateQuoteResponse::Error( + behind @ ProtocolError::StorerUpdateRequired { .. }, + )) => Some(Err(Error::StorerUpdateRequired(behind.to_string()))), + ChunkMessageBody::MerkleCandidateQuoteResponse(MerkleCandidateQuoteResponse::Error(e)) => { + Some(Err(Error::Protocol(format!( + "Merkle quote error from {peer_id}: {e}" + )))) + } + _ => None, + } +} + +/// Should this failure be retried as an unversioned request? +/// +/// A storer built before the settlement version existed cannot decode the +/// versioned request, so it never replies and the send fails at the transport +/// layer. Any response this client *recognises*, refusal included, means the +/// storer understood us, and retrying that in an older shape would defeat the +/// gate. +/// +/// A reply that decodes but carries a body the mapper does not recognise is +/// not distinguished: the wait continues and ends in a timeout, so it takes +/// the fallback too. That is no more than a peer could achieve by saying +/// nothing, which is why it is bounded by the cutover guard rather than +/// special-cased here. +/// +/// # This is a downgrade path, and it is only safe because nothing can be +/// refused yet +/// +/// Silence is **not** proof that a peer cannot parse the request. A dropped +/// response, packet loss, an overloaded peer, or one deliberately discarding +/// versioned requests all look identical from here. So this predicate can be +/// provoked, and a peer that provokes it gets asked again without a version. +/// +/// That is harmless only while no client can be refused on version grounds, +/// which is exactly the case while `MIN_SUPPORTED_SETTLEMENT_VERSION` is the +/// first declarable version. The moment the minimum is raised, this becomes a +/// way to obtain a quote the gate meant to withhold, and then to burn a +/// payment against it. +/// +/// The guard below turns "remember to delete the fallback before raising the +/// minimum" from a comment into a build failure. It lives in +/// [`crate::data::client::UNVERSIONED_RETRY_REQUIRES_MIN_V1`] and is referenced +/// from every fallback site, including the independent single-node one in +/// `quote.rs`, so deleting one path cannot silently leave the other unguarded. +const _: () = crate::data::client::UNVERSIONED_RETRY_REQUIRES_MIN_V1; + +const fn is_version_unaware(error: &Error) -> bool { + matches!(error, Error::Network(_) | Error::Timeout(_)) +} + /// Payment mode for uploads. #[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)] #[serde(rename_all = "snake_case")] @@ -561,6 +689,13 @@ impl Client { data_type: u32, data_size: u64, ) -> Result { + // A refusal established by any earlier upload on this client stops + // this one before it spends. The verdict is about this build, not + // about one operation, so an upload that started after another had + // already been told the settlement rules are wrong must not pay. + if let Some(refusal) = self.corroborated_settlement_refusal() { + return Err(Error::ClientUpdateRequired(refusal)); + } let chunk_count = addresses.len(); if chunk_count < 2 { return Err(Error::Payment( @@ -775,6 +910,13 @@ impl Client { data_type: u32, data_size: u64, ) -> Result { + // A refusal established by any earlier upload on this client stops + // this one before it spends. The verdict is about this build, not + // about one operation, so an upload that started after another had + // already been told the settlement rules are wrong must not pay. + if let Some(refusal) = self.corroborated_settlement_refusal() { + return Err(Error::ClientUpdateRequired(refusal)); + } ensure_single_merkle_tree_batch(addresses.len())?; let chunk_count = addresses.len(); @@ -917,6 +1059,33 @@ impl Client { // First sub-batch failed, nothing paid yet -- propagate directly. return Err(e); } + // A storer saying this client cannot settle is terminal + // even mid-batch. Folding it into a partial result would + // report success, hide the upgrade instruction, and leave + // the caller to rediscover the same refusal on its next + // upload. Every remaining sub-batch would be refused the + // same way, so there is nothing to salvage by continuing. + if matches!(e, Error::ClientUpdateRequired(_)) { + // Do NOT return Err here. The caller writes the receipt + // cache only on the Ok path, so failing the call would + // discard proofs for sub-batches whose payment has + // already settled on-chain and cannot be undone. That is + // the very destruction this work exists to stop. + // + // The verdict is not lost by returning the partial + // result: it was latched client-wide when it was + // corroborated, so the next quote collection refuses + // before spending anything and the upgrade instruction + // reaches the user from there. + warn!( + "Merkle sub-batch {}/{total_sub_batches}: storers refused this \ + client's settlement version. Returning {} proofs from \ + already-paid sub-batches so that spend is not stranded; the \ + refusal is latched and will stop the next payment.", + i + 1, + all_proofs.len() + ); + } // Return partial result so caller can still store already-paid chunks. warn!( "Merkle sub-batch {}/{total_sub_batches} failed: {e}. \ @@ -973,9 +1142,27 @@ impl Client { }); } + // Drained rather than short-circuited. Returning on the first error + // would drop `pool_futures`, cancelling every pool still in flight, and + // one of those may be carrying the second refusal that corroborates a + // verdict about this client. A refusal lost that way is a refusal the + // latch never sees, and the caller can fall back to wave payment and + // spend. This is the same rule the single-node collector follows: stop + // making progress, but never drop a verdict that is already in flight. + // + // A refusal outranks an ordinary failure for the same reason. If one + // pool runs out of peers while another declares this client unable to + // settle, the second is the answer worth returning. let mut pools = Vec::with_capacity(midpoint_proofs.len()); + let mut verdict = PoolVerdict::default(); while let Some(result) = pool_futures.next().await { - pools.push(result?); + match result { + Ok(pool) => pools.push(pool), + Err(e) => verdict.note(e), + } + } + if let Some(e) = verdict.into_error() { + return Err(e); } Ok(pools) @@ -1022,17 +1209,51 @@ impl Client { let mut candidate_futures = FuturesUnordered::new(); + let unversioned_peers = self.unversioned_quote_peers(); + let versioned_capable = self.versioned_quote_capable_handle(); + for (peer_id, peer_addrs) in &remote_peers { let request_id = self.next_request_id(); - let request = MerkleCandidateQuoteRequest { + // A peer that already failed to answer a versioned request is + // asked in the legacy shape directly. Re-probing costs a full + // per-peer timeout every time, and a merkle pool asks sixteen + // candidates per pool, so against a fleet that predates the + // versioned requests the probes dominate the run: measured on this + // suite it went from ~24 minutes to past the 60-minute CI cap. + // The capable set wins. The two sets are updated under separate + // locks, so a slow probe can insert into the legacy set after a + // concurrent request has already proved the peer capable; letting + // capability win makes that interleaving harmless instead of + // permanently preferring the legacy shape. + let known_legacy = !versioned_capable + .lock() + .is_ok_and(|peers| peers.contains(peer_id)) + && unversioned_peers + .lock() + .is_ok_and(|peers| peers.contains(peer_id)); + + let legacy_request = MerkleCandidateQuoteRequest { address: *address, data_type, data_size, merkle_payment_timestamp, }; + + // Declare the settlement version so a storer can turn us away + // before we pay, rather than refusing the payment afterwards. let message = ChunkMessage { request_id, - body: ChunkMessageBody::MerkleCandidateQuoteRequest(request), + body: if known_legacy { + ChunkMessageBody::MerkleCandidateQuoteRequest(legacy_request.clone()) + } else { + ChunkMessageBody::MerkleCandidateQuoteRequestV2( + MerkleCandidateQuoteRequestV2::new( + *address, + data_size, + merkle_payment_timestamp, + ), + ) + }, }; let message_bytes = match message.encode() { @@ -1043,41 +1264,53 @@ impl Client { } }; + // Fallback for the mixed fleet. A storer that predates the + // versioned request cannot decode it and simply never answers, so + // without this every quote would fail until the whole network had + // upgraded. Retried only on a transport-level failure, never on a + // refusal: see `is_version_unaware` below. + // + // Delete this, and the second request id, once the fleet is known + // to answer V2. + let legacy_request_id = self.next_request_id(); + let legacy_message_bytes = match (ChunkMessage { + request_id: legacy_request_id, + body: ChunkMessageBody::MerkleCandidateQuoteRequest(legacy_request), + }) + .encode() + { + Ok(bytes) => bytes, + Err(e) => { + warn!("Failed to encode legacy merkle candidate request for {peer_id}: {e}"); + continue; + } + }; + let peer_id_clone = *peer_id; let addrs_clone = peer_addrs.clone(); let node_clone = node.clone(); + let peers_handle = Arc::clone(&unversioned_peers); + let capable_handle = Arc::clone(&versioned_capable); let fut = async move { + // First contact waits only long enough to learn whether the + // peer can parse the shape. A peer already known to be legacy + // gets the caller's full patience, because that request is the + // real one rather than a probe. + let attempt_timeout = if known_legacy { + timeout + } else { + timeout.min(crate::data::client::VERSIONED_QUOTE_PROBE_CEILING) + }; + let result = send_and_await_chunk_response( &node_clone, &peer_id_clone, message_bytes, request_id, - timeout, + attempt_timeout, &addrs_clone, - |body| match body { - ChunkMessageBody::MerkleCandidateQuoteResponse( - MerkleCandidateQuoteResponse::Success { - candidate_node, - commitment, - }, - ) => { - match rmp_serde::from_slice::( - &candidate_node, - ) { - Ok(node) => Some(Ok((node, commitment))), - Err(e) => Some(Err(Error::Serialization(format!( - "Failed to deserialize candidate node from {peer_id_clone}: {e}" - )))), - } - } - ChunkMessageBody::MerkleCandidateQuoteResponse( - MerkleCandidateQuoteResponse::Error(e), - ) => Some(Err(Error::Protocol(format!( - "Merkle quote error from {peer_id_clone}: {e}" - )))), - _ => None, - }, + |body| map_merkle_candidate_response(peer_id_clone, body), |e| { Error::Network(format!( "Failed to send merkle candidate request to {peer_id_clone}: {e}" @@ -1091,6 +1324,70 @@ impl Client { ) .await; + // Any answer at all to the versioned shape proves the peer + // can parse it, so it can never later be demoted. + // Any recognised answer proves the peer parsed the versioned + // shape, including a structured error. Only silence and send + // failures leave the question open. + let answered = match &result { + Ok(_) => true, + Err(e) => !is_version_unaware(e), + }; + if !known_legacy && answered { + if let Ok(mut peers) = capable_handle.lock() { + peers.insert(peer_id_clone); + } + } + + // Silence from a storer means it could not decode the + // versioned request, so ask again in the shape it understands. + // A storer that answered with a refusal is NOT retried: it + // understood us and said no, and asking again without the + // version would talk it into quoting a client that cannot pay. + let result = match result { + Err(ref e) if is_version_unaware(e) && !known_legacy => { + // Only silence is evidence the peer cannot parse the + // shape. A send failure means the request never + // arrived and teaches nothing, so it must not strand + // the peer in the legacy shape for the whole session. + // Nor may a peer that has answered a versioned request + // before be demoted by one lost response. + let ever_answered = capable_handle + .lock() + .is_ok_and(|peers| peers.contains(&peer_id_clone)); + if matches!(e, Error::Timeout(_)) && !ever_answered { + if let Ok(mut peers) = peers_handle.lock() { + peers.insert(peer_id_clone); + } + } + debug!( + "Peer {peer_id_clone} did not answer a versioned merkle quote; \ + retrying in the legacy shape" + ); + send_and_await_chunk_response( + &node_clone, + &peer_id_clone, + legacy_message_bytes, + legacy_request_id, + timeout, + &addrs_clone, + |body| map_merkle_candidate_response(peer_id_clone, body), + |e| { + Error::Network(format!( + "Failed to send merkle candidate request to {peer_id_clone}: {e}" + )) + }, + || { + Error::Timeout(format!( + "Timeout waiting for merkle candidate from {peer_id_clone}" + )) + }, + ) + .await + } + other => other, + }; + (peer_id_clone, result) }; @@ -1170,6 +1467,33 @@ impl Client { } valid.push((candidate_peer, candidate)); } + // A storer that has explicitly declared this client + // incompatible ends the batch, but only once enough distinct + // peers agree. Collecting a corroborated refusal as one more + // failed peer would let the pool fill from the remaining + // sixteen and go on to pay, which is the burn this mechanism + // exists to prevent; acting on a single peer's unauthenticated + // word would instead let one hostile responder deny every + // upload. + Err(e @ Error::ClientUpdateRequired(_)) => { + if let Some(corroborated) = + self.note_settlement_refusal(peer_id, &e.to_string()) + { + // Name the corroborators — the below-quorum branch logs + // one peer at a time, so without this line a reader + // cannot tell a two-peer verdict from a one-peer + // misfire (V2-1109). + let corroborators = self.settlement_refusals().corroborating_peers(); + warn!( + "Settlement refusal corroborated by {} distinct peers [{}]; aborting before payment", + corroborators.len(), + corroborators.join(", ") + ); + return Err(Error::ClientUpdateRequired(corroborated)); + } + warn!("Merkle candidate {peer_id} refused this client's settlement version; awaiting corroboration"); + failures.push(format!("{peer_id}: {e}")); + } Err(e) => { debug!("Failed to get merkle candidate from {peer_id}: {e}"); failures.push(format!("{peer_id}: {e}")); @@ -3267,4 +3591,101 @@ mod tests { assert!(outcome.failed_addresses.is_empty()); assert!(outcome.fatal.is_none()); } + + // ========================================================================= + // Settlement refusals on the merkle path + // ========================================================================= + + /// A refusal is unauthenticated and two of them latch this client for the + /// rest of its run, so the merkle path must apply the same coherence check + /// the single-node path applies. Without it, two faulty or hostile + /// candidates could deny every upload with a refusal that is not about this + /// client at all. + #[test] + fn a_merkle_refusal_that_does_not_describe_this_client_is_ignored() { + use ant_protocol::CURRENT_SETTLEMENT_VERSION; + + let peer_id = PeerId::from_bytes([0x71; 32]); + let refusal = |client: u32, min: u32| { + map_merkle_candidate_response( + peer_id, + ChunkMessageBody::MerkleCandidateQuoteResponse( + MerkleCandidateQuoteResponse::Error(ProtocolError::ClientUpdateRequired { + client_settlement_version: client, + min_settlement_version: min, + }), + ), + ) + }; + + // Echoes a version that is not ours: the peer is talking about someone + // else's request. + let wrong_echo = refusal( + CURRENT_SETTLEMENT_VERSION.saturating_add(7), + CURRENT_SETTLEMENT_VERSION.saturating_add(8), + ); + assert!( + matches!(wrong_echo, Some(Err(Error::Protocol(_)))), + "{wrong_echo:?}" + ); + + // States a minimum this client already meets, so there is nothing to + // refuse. + let no_gap = refusal(CURRENT_SETTLEMENT_VERSION, CURRENT_SETTLEMENT_VERSION); + assert!( + matches!(no_gap, Some(Err(Error::Protocol(_)))), + "{no_gap:?}" + ); + + // A coherent one is believed, and still carries the upgrade wording. + match refusal( + CURRENT_SETTLEMENT_VERSION, + CURRENT_SETTLEMENT_VERSION.saturating_add(1), + ) { + Some(Err(Error::ClientUpdateRequired(msg))) => { + assert!(msg.contains("ant update"), "{msg}"); + } + other => panic!("expected ClientUpdateRequired, got {other:?}"), + } + } + + /// Pools are drained rather than cancelled on the first error, so more than + /// one can fail. A refusal has to win: reporting the ordinary failure + /// instead would hide the one verdict that stops the caller falling back to + /// another payment path and spending. + #[test] + fn a_refusal_outranks_an_ordinary_pool_failure() { + let refusal = || Error::ClientUpdateRequired("run ant update".to_string()); + let ordinary = || Error::InsufficientPeers("need 16, got 2".to_string()); + + // Ordinary failure first, refusal second. + let mut verdict = PoolVerdict::default(); + verdict.note(ordinary()); + verdict.note(refusal()); + assert!( + matches!(verdict.into_error(), Some(Error::ClientUpdateRequired(_))), + "a later refusal must still outrank an earlier failure" + ); + + // Refusal first, ordinary failure second. Order must not matter. + let mut verdict = PoolVerdict::default(); + verdict.note(refusal()); + verdict.note(ordinary()); + assert!( + matches!(verdict.into_error(), Some(Error::ClientUpdateRequired(_))), + "an earlier refusal must not be displaced by a later failure" + ); + + // With no refusal, the first ordinary failure is what the caller sees. + let mut verdict = PoolVerdict::default(); + verdict.note(ordinary()); + verdict.note(Error::Protocol("second".to_string())); + assert!( + matches!(verdict.into_error(), Some(Error::InsufficientPeers(_))), + "the first ordinary failure is the one reported" + ); + + // Every pool succeeded, so there is nothing to report. + assert!(PoolVerdict::default().into_error().is_none()); + } } diff --git a/ant-core/src/data/client/mod.rs b/ant-core/src/data/client/mod.rs index d45b005..9572e5c 100644 --- a/ant-core/src/data/client/mod.rs +++ b/ant-core/src/data/client/mod.rs @@ -23,9 +23,11 @@ use crate::data::peer_cache; use ant_protocol::evm::Wallet; use ant_protocol::transport::{MultiAddr, P2PNode, PeerId}; use ant_protocol::{XorName, CLOSE_GROUP_SIZE}; +use std::collections::HashSet; use std::path::PathBuf; use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; +use std::sync::Mutex; use tracing::debug; /// Width of the chunk PUT-target set (initial writes plus fallback): the @@ -37,6 +39,226 @@ use tracing::debug; /// so trying peers past this width is pointless. pub(crate) const PUT_TARGET_WIDTH: usize = 20; +/// Ceiling on how long to wait for a peer to answer a settlement-versioned +/// quote request before falling back to the unversioned shape. +/// +/// A storer that predates the versioned request cannot decode it and never +/// replies, so the only way to find out is to wait. +/// +/// Note what is being waited for. This is **not** a cheap parse check: a peer +/// that understands the request runs the whole quote handler, queueing, +/// storage reads, pricing and signing, so the answer takes as long as any +/// quote takes. Abandoning the wait does not cancel that work either, it just +/// stops listening and adds a duplicate legacy request on top. So the wait has +/// to be sized for a real quote, not for a ping. +/// +/// The full timeout is what made this expensive. The merkle E2E suite runs +/// with `quote_timeout_secs = 120`, so every probe against a fleet on the +/// published node cost two minutes and the suite blew the 60-minute CI cap. +/// +/// # It must never bind in production +/// +/// **Keep this at or above the largest production `quote_timeout_secs`**, +/// currently 10s. That is not a tuning preference, it is the safety property. +/// +/// This wait is the only window in which a peer can refuse. Abandoning it +/// early does not merely mislabel a slow peer: it drops the refusal, so it +/// never counts toward corroboration, never sets the latch, and the legacy +/// request it raced can return a perfectly good quote the client then pays +/// against. +/// +/// The two fallbacks lose it by different mechanisms. The merkle path sends +/// its legacy request under a *new* request id (`merkle.rs`), so a refusal +/// arriving after the ceiling is answering a request nobody is listening to +/// and is discarded on the id mismatch. The single-node path reuses the same +/// id (`quote.rs`), so a late refusal is seen only if it happens to arrive +/// after the retry has resubscribed; the await that would have matched it is +/// already gone, and anything landing in the gap between the two waits is +/// lost. Neither path observes a late refusal reliably, which is what the +/// ceiling exists to prevent. +/// +/// A shorter ceiling was tried, at 5s, to bring the slower CI runner under the +/// cap. It would have turned every legitimate 5-to-10 second refusal in +/// production into exactly that silent downgrade. The never-demote rule does +/// not help: it stops a peer being *cached* as legacy after it has answered +/// once, but it does not stop the request in flight from falling back. And the +/// compile-time guard does not help either, because it binds future builds +/// while the clients at risk are the ones already released. +/// +/// So the ceiling exists solely to bound configurations that set a timeout far +/// above any real answer time, which in practice means test harnesses. Above +/// 10s it never binds on a production client, and nothing real is truncated. +/// +/// The cost of leaving it here is roughly two minutes per merkle E2E test +/// while the suite's devnet still speaks the pre-versioned dialect. That is a +/// consequence of the temporary fork-branch protocol pin, not of the design: +/// once the fleet under test can answer a versioned request there are no +/// probes to pay for, and the suite returns to its baseline. +pub(crate) const VERSIONED_QUOTE_PROBE_CEILING: std::time::Duration = + std::time::Duration::from_secs(15); + +/// How many distinct peers must refuse this client's settlement version before +/// the refusal is believed and uploads stop. +/// +/// Nothing authenticates a refusal, so one peer's word cannot be enough: a +/// single hostile or misconfigured storer answering `ClientUpdateRequired` to +/// everything would otherwise deny every upload, turning an over-query design +/// that tolerates many bad peers into one that tolerates none. +/// +/// Two is deliberately low. A genuine incompatibility reaches it instantly, +/// because every peer enforcing the newer rule refuses, and a client queries +/// far more than two. An attacker has to control two of the peers a given +/// request happens to reach, which is a materially different proposition from +/// controlling one. +pub(crate) const SETTLEMENT_REFUSAL_QUORUM: usize = 2; + +/// Compile-time cutover guard shared by **every** unversioned quote retry. +/// +/// Both quote paths fall back to an unversioned request when a peer stays +/// silent, because a storer built before the settlement version existed cannot +/// decode the versioned one. Silence is not proof of that, though: a dropped +/// response, packet loss, an overloaded peer, or one deliberately discarding +/// versioned requests are indistinguishable from the client side. So the +/// fallback is a downgrade path and can be provoked. +/// +/// It is not only literal silence, either. `send_and_await_chunk_response` +/// keeps waiting when a reply decodes but carries a body the mapper does not +/// recognise, so a peer answering with an unexpected variant also lands on the +/// timeout and takes this path. That grants no capability beyond staying +/// silent, which is why it is bounded rather than special-cased, but a comment +/// claiming "any structured response prevents the retry" would be wrong. +/// +/// It is safe only while **no refusal of either kind is possible**, and that +/// means bounding both constants, not just the minimum. +/// +/// Raising `MIN` is the obvious hazard: a client below it would be refused, +/// and the retry hands it a quote anyway. Raising `CURRENT` is the subtler +/// one. As soon as some node runs a newer `CURRENT` than another, the older +/// node refuses newer clients with `StorerUpdateRequired` precisely because it +/// cannot promise to honour their payment. A client that retries such a peer +/// unversioned gets that unhonourable quote, and for a settlement change that +/// is not a pure increase the payment is then rejected after it has settled. +/// Guarding only `MIN` would leave that route open. +/// +/// So the guard requires both to still be at the first declarable version. +/// +/// This bounds **future builds** only. A client binary already in the field +/// carries whatever fallback it shipped with, and no source change reaches it; +/// that is inherent to shipping software and is why the storer still verifies +/// every payment it is actually offered. +/// +/// Each fallback site references this constant so the guard cannot be orphaned +/// by deleting one path and forgetting the other. Retiring the fallbacks means +/// deleting this constant and every reference to it, which the compiler then +/// points at one by one. +pub(crate) const UNVERSIONED_RETRY_REQUIRES_MIN_V1: () = assert!( + ant_protocol::MIN_SUPPORTED_SETTLEMENT_VERSION == 1 + && ant_protocol::CURRENT_SETTLEMENT_VERSION == 1, + "an unversioned quote retry is still compiled in: it is a downgrade path around \ + both the too-old and the node-behind refusals, so delete every fallback site \ + before raising MIN_SUPPORTED_SETTLEMENT_VERSION or CURRENT_SETTLEMENT_VERSION" +); + +/// Distinct peers that have refused this client's settlement version, plus the +/// wording of the first refusal. +/// +/// A separate type rather than a field so the corroboration rule can be tested +/// on its own: it is the piece that decides whether an upload stops, and it +/// has to hold against both a lone lying peer and a genuine incompatibility. +#[derive(Clone, Default)] +pub(crate) struct SettlementRefusals { + inner: Arc, Option)>>, +} + +impl SettlementRefusals { + /// Record a refusal from `peer_id`, returning the wording once + /// [`SETTLEMENT_REFUSAL_QUORUM`] distinct peers agree and `None` below it. + pub(crate) fn note(&self, peer_id: PeerId, message: &str) -> Option { + let mut guard = self.inner.lock().ok()?; + let (peers, wording) = &mut *guard; + peers.insert(peer_id); + if wording.is_none() { + *wording = Some(message.to_string()); + } + (peers.len() >= SETTLEMENT_REFUSAL_QUORUM) + .then(|| wording.clone()) + .flatten() + } + + /// The corroborated refusal, if one has been established. + pub(crate) fn corroborated(&self) -> Option { + let guard = self.inner.lock().ok()?; + let (peers, wording) = &*guard; + (peers.len() >= SETTLEMENT_REFUSAL_QUORUM) + .then(|| wording.clone()) + .flatten() + } + + /// The distinct refusing peers recorded so far, rendered for logging, + /// sorted so the line is deterministic. + /// + /// Exists so the terminal abort can NAME its corroborators. Without it the + /// quorum is unverifiable from outside: "awaiting corroboration" is only + /// logged below the quorum and the terminal error carries no peer ids, so + /// a log reader sees one warned peer followed by an abort whether the + /// quorum counted two distinct peers or misfired on one — which is exactly + /// the ambiguity the V2-1109 testnet run hit (0/463 aborts showed a second + /// peer, because the second peer was structurally unloggable). + pub(crate) fn corroborating_peers(&self) -> Vec { + let Ok(guard) = self.inner.lock() else { + return Vec::new(); + }; + let (peers, _) = &*guard; + let mut ids: Vec = peers.iter().map(|p| format!("{p}")).collect(); + ids.sort(); + ids + } +} + +#[cfg(test)] +mod settlement_refusal_tests { + use super::*; + + fn peer(seed: u8) -> PeerId { + PeerId::from_bytes([seed; 32]) + } + + /// The question the V2-1109 run could not answer from logs: does one peer + /// refusing many times count as many? It must not — the quorum is over + /// DISTINCT peers, and a lone hostile or misconfigured storer repeating + /// itself must never become a verdict about this build. + #[test] + fn one_peer_refusing_many_times_never_reaches_the_quorum() { + let refusals = SettlementRefusals::default(); + for _ in 0..50 { + assert!( + refusals.note(peer(7), "run ant update").is_none(), + "a single peer's repeated refusals must stay below the quorum" + ); + } + assert!(refusals.corroborated().is_none()); + assert_eq!(refusals.corroborating_peers().len(), 1); + } + + /// The terminal log line must be able to name both corroborators, so the + /// distinct-peer property is verifiable from the outside. + #[test] + fn corroborating_peers_names_every_distinct_refuser() { + let refusals = SettlementRefusals::default(); + assert!(refusals.note(peer(1), "run ant update").is_none()); + assert!(refusals.note(peer(2), "run ant update").is_some()); + + let ids = refusals.corroborating_peers(); + assert_eq!(ids.len(), 2); + assert_ne!(ids[0], ids[1]); + assert_eq!(ids, { + let mut sorted = ids.clone(); + sorted.sort(); + sorted + }); + } +} + /// Classify a `data::error::Error` into a controller `Outcome`. /// /// Capacity signals (Timeout / NetworkError) drive the controller @@ -64,6 +286,9 @@ pub(crate) const PUT_TARGET_WIDTH: usize = 20; /// - `RemotePut` -> `ApplicationError` (the remote node responded with a /// structured rejection — the transport succeeded, so the node declined /// at the application layer; not a local capacity signal) +/// - `ClientUpdateRequired` -> `ApplicationError` (the storer refused to quote +/// a client that settles under superseded rules — a terminal verdict about +/// this build, not about link capacity, and no retry rate clears it) /// - `CloseGroupShortfall` -> `ApplicationError` (a quorum shortfall caused /// by close-group dial/relay churn with no PUT-response timeouts — remote /// peer churn, not local backpressure; a timeout-bearing shortfall keeps @@ -93,6 +318,12 @@ pub(crate) fn classify_error(err: &Error) -> Outcome { | Error::InsufficientDiskSpace(_) | Error::CostEstimationInconclusive(_) | Error::Cancelled(_) + // The storer parsed our request and refused it on its merits, over a + // working link. Sending fewer requests would not help, and treating it + // as congestion would quietly shrink the limiter for the rest of the + // run on the basis of a fault no retry can clear. + | Error::ClientUpdateRequired(_) + | Error::StorerUpdateRequired(_) | Error::BadQuoteBinding { .. } | Error::BadQuoteCommitment { .. } // An external-signer merkle batch larger than one tree can hold — @@ -383,6 +614,64 @@ pub struct Client { persist_path: Option, /// Path for the persistent client peer cache. `None` disables the cache. peer_cache_path: Option, + /// Peers that did not answer a settlement-versioned quote request, and are + /// therefore asked in the legacy shape from now on. + /// + /// Without this the probe cost is paid on **every** request rather than + /// roughly once per peer. Measured on the merkle E2E suite against a fleet + /// that predates the versioned requests, re-probing took the run from ~24 + /// minutes to over 60, because each of the sixteen candidates per pool sat + /// out a full `quote_timeout_secs` before the fallback. + /// + /// Roughly, not exactly: concurrent first contacts are not coalesced, so + /// several in-flight requests can all miss the cache for the same peer and + /// each probe it once before any of them records the answer. Observed at + /// about two probes per peer on a 35-node devnet. Single-flighting them + /// would remove the duplicates but not the wall-clock cost, which is set + /// by how many *sequential* quote rounds an upload performs rather than by + /// how many probes each round contains. + /// + /// Process-local and never persisted. A peer that upgrades mid-run keeps + /// being asked in the legacy shape until the next start, which is + /// acceptable while the legacy shape still gets a quote, and stops + /// mattering when the fallback is deleted (see + /// [`UNVERSIONED_RETRY_REQUIRES_MIN_V1`]). + /// + /// Entries are only ever added for a peer that has **never** answered a + /// versioned request. Without that condition a single lost response would + /// pin an upgraded peer to the legacy shape for the rest of the session, + /// turning one dropped packet into a standing downgrade; with it, a peer + /// that has shown it understands the versioned shape can never be demoted. + /// + /// A peer that has never answered can still get itself asked without a + /// version by staying silent, exactly as it could through the fallback + /// alone. Remembering the answer makes that cheaper to sustain, so it is + /// not a new capability but it is a wider one, and the compile-time guard + /// requires the whole path to be gone before any refusal is possible. + unversioned_quote_peers: Arc>>, + /// Peers observed answering a settlement-versioned request. Never + /// downgraded, however they behave later. + versioned_capable_peers: Arc>>, + /// Distinct peers that have refused this client on settlement-version + /// grounds, and the wording of the first such refusal. + /// + /// Client-wide and sticky, for two reasons that pull in opposite + /// directions and are both real. + /// + /// It must outlive one operation, because the verdict is about this + /// **build**, not this upload. Held in a single collector's local state, a + /// refusal observed by one in-flight upload says nothing to another that + /// is about to submit a payment, and merkle payments cannot be undone. + /// + /// It must not fire on one peer's say-so, because nothing authenticates a + /// refusal. A single hostile or confused peer answering + /// `ClientUpdateRequired` to every query would otherwise abort every + /// upload the client attempts, converting an over-query design that + /// tolerates many bad peers into one that tolerates none. So a refusal + /// becomes terminal only once [`SETTLEMENT_REFUSAL_QUORUM`] distinct peers + /// agree, which a genuine incompatibility reaches immediately (every + /// upgraded peer refuses) and a lone attacker cannot reach at all. + settlement_refusals: SettlementRefusals, } impl Client { @@ -409,6 +698,9 @@ impl Client { evm_network: None, chunk_cache: ChunkCache::default(), next_request_id: AtomicU64::new(1), + unversioned_quote_peers: Arc::new(Mutex::new(HashSet::new())), + versioned_capable_peers: Arc::new(Mutex::new(HashSet::new())), + settlement_refusals: SettlementRefusals::default(), controller, persist_path, peer_cache_path, @@ -444,6 +736,9 @@ impl Client { evm_network: None, chunk_cache: ChunkCache::default(), next_request_id: AtomicU64::new(1), + unversioned_quote_peers: Arc::new(Mutex::new(HashSet::new())), + versioned_capable_peers: Arc::new(Mutex::new(HashSet::new())), + settlement_refusals: SettlementRefusals::default(), controller, persist_path, peer_cache_path: None, @@ -562,6 +857,51 @@ impl Client { self.next_request_id.fetch_add(1, Ordering::Relaxed) } + /// Handle to the set of peers that cannot answer a settlement-versioned + /// quote request, shared with the per-peer request futures on both quote + /// paths. + /// + /// Callers read it before choosing a request shape and insert into it when + /// a peer stays silent. A poisoned lock is treated as "nothing known", so + /// the worst case is a wasted probe rather than a silently skipped version + /// declaration. + pub(crate) fn unversioned_quote_peers(&self) -> Arc>> { + Arc::clone(&self.unversioned_quote_peers) + } + + /// Handle to the set of peers already seen answering a versioned request. + /// Consulted before demoting a peer, so a lost response cannot strand an + /// upgraded peer in the legacy shape. + pub(crate) fn versioned_quote_capable_handle(&self) -> Arc>> { + Arc::clone(&self.versioned_capable_peers) + } + + /// Record that `peer_id` refused this client's settlement version, and + /// report whether enough distinct peers now agree for it to be believed. + /// + /// Returns the refusal wording once [`SETTLEMENT_REFUSAL_QUORUM`] is met, + /// and `None` below it, so a lone peer is treated as a peer fault rather + /// than a verdict about this build. + pub(crate) fn note_settlement_refusal(&self, peer_id: PeerId, message: &str) -> Option { + self.settlement_refusals.note(peer_id, message) + } + + /// The corroborated refusal, if this client has already been told by + /// enough peers that it cannot settle. + /// + /// Checked before spending money. The verdict concerns this build rather + /// than any one upload, so an upload that starts after another has already + /// established it must not proceed to pay. + pub(crate) fn corroborated_settlement_refusal(&self) -> Option { + self.settlement_refusals.corroborated() + } + + /// Handle to the shared refusal tracker, for collectors that run outside + /// `&self`. + pub(crate) fn settlement_refusals(&self) -> SettlementRefusals { + self.settlement_refusals.clone() + } + /// Return the chunk PUT-target set: the closest [`PUT_TARGET_WIDTH`] peers /// to the address, each paired with its known network addresses. /// @@ -823,6 +1163,8 @@ mod tests { | Error::BadQuoteCommitment { .. } | Error::MerkleBatchTooLarge { .. } | Error::RemotePut { .. } + | Error::ClientUpdateRequired(_) + | Error::StorerUpdateRequired(_) | Error::CloseGroupShortfall(_) => (), }; } diff --git a/ant-core/src/data/client/payment.rs b/ant-core/src/data/client/payment.rs index 582f46f..8d2f430 100644 --- a/ant-core/src/data/client/payment.rs +++ b/ant-core/src/data/client/payment.rs @@ -46,6 +46,13 @@ impl Client { data_size: u64, data_type: u32, ) -> Result<(Vec, Vec<(PeerId, Vec)>)> { + // A refusal established by any earlier upload on this client stops + // this one before it spends. The verdict is about this build, not + // about one operation. + if let Some(refusal) = self.corroborated_settlement_refusal() { + return Err(Error::ClientUpdateRequired(refusal)); + } + // Wallet is required for the on-chain payment step (step 4 below). // Check early so we don't waste time collecting quotes for a misconfigured client. let wallet = self.require_wallet()?; diff --git a/ant-core/src/data/client/quote.rs b/ant-core/src/data/client/quote.rs index 87e19b3..cbce575 100644 --- a/ant-core/src/data/client/quote.rs +++ b/ant-core/src/data/client/quote.rs @@ -5,7 +5,9 @@ use crate::data::client::peer_xor_distance; use crate::data::client::Client; +use crate::data::client::SettlementRefusals; use crate::data::client::PUT_TARGET_WIDTH; +use crate::data::client::VERSIONED_QUOTE_PROBE_CEILING; use crate::data::error::{Error, Result}; use ant_protocol::evm::{Amount, PaymentQuote}; use ant_protocol::payment::calculate_price; @@ -18,12 +20,13 @@ 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, + client_update_required_message, compute_address, send_and_await_chunk_response, ChunkMessage, + ChunkMessageBody, ChunkQuoteRequest, ChunkQuoteRequestV2, ChunkQuoteResponse, ProtocolError, + CLOSE_GROUP_MAJORITY, CLOSE_GROUP_SIZE, CURRENT_SETTLEMENT_VERSION, }; use futures::stream::{FuturesUnordered, StreamExt}; use std::collections::{HashMap, HashSet}; -use std::sync::Arc; +use std::sync::{Arc, Mutex}; use std::time::Duration; use tracing::{debug, info, warn}; @@ -229,13 +232,17 @@ fn quote_commitment_binding_is_valid( /// On success the returned commitment is the opaque signed-commitment blob the /// node shipped with the quote (`None` for a baseline quote), to be forwarded /// as a sidecar in the PUT bundle. +/// +/// A quote as this module hands it on: the quote itself, the price to settle, +/// and the opaque signed commitment it was priced against. +type ClassifiedQuote = std::result::Result<(PaymentQuote, Amount, Option>), Error>; fn classify_quote_response( peer_id: &PeerId, expected_content: &[u8; 32], quote_bytes: &[u8], already_stored: bool, commitment: Option>, -) -> std::result::Result<(PaymentQuote, Amount, Option>), Error> { +) -> ClassifiedQuote { let payment_quote = rmp_serde::from_slice::(quote_bytes).map_err(|e| { Error::Serialization(format!("Failed to deserialize quote from {peer_id}: {e}")) })?; @@ -330,16 +337,42 @@ async fn request_store_quote_from_peer( data_size: u64, data_type: u32, per_peer_timeout: Duration, + unversioned_peers: Arc>>, + versioned_capable: Arc>>, ) -> StoreQuoteRequestResult { - let request = ChunkQuoteRequest { + let legacy_request = ChunkQuoteRequest { address, data_size, data_type, }; - let message = ChunkMessage { - request_id, - body: ChunkMessageBody::QuoteRequest(request), + + // A peer that already failed to answer a versioned request is asked in the + // legacy shape directly. Re-probing costs a full per-peer timeout every + // time, and against a fleet that predates the versioned requests that is + // paid on every quote: measured on the merkle E2E suite it took the run + // from ~24 minutes to past the 60-minute cap. + // The capable set wins. The two sets are updated under separate locks, so + // a slow probe can insert into the legacy set after a concurrent request + // has already proved the peer capable; letting capability win makes that + // interleaving harmless instead of permanently preferring the legacy shape. + let known_legacy = !versioned_capable + .lock() + .is_ok_and(|peers| peers.contains(&peer_id)) + && unversioned_peers + .lock() + .is_ok_and(|peers| peers.contains(&peer_id)); + + // Declare the settlement version so a storer can turn us away before we + // pay. See `merkle.rs` for why the legacy retry below exists and when it + // can be deleted. + let body = if known_legacy { + ChunkMessageBody::QuoteRequest(legacy_request.clone()) + } else { + let mut versioned_request = ChunkQuoteRequestV2::new(address, data_size); + versioned_request.data_type = data_type; + ChunkMessageBody::QuoteRequestV2(versioned_request) }; + let message = ChunkMessage { request_id, body }; let message_bytes = match message.encode() { Ok(bytes) => bytes, @@ -354,38 +387,186 @@ async fn request_store_quote_from_peer( } }; + // A first-contact probe waits only long enough to learn whether the peer + // can parse the shape; a peer already known to be legacy is asked with the + // caller's full patience because that request is the real one. + let attempt_timeout = if known_legacy { + per_peer_timeout + } else { + per_peer_timeout.min(VERSIONED_QUOTE_PROBE_CEILING) + }; + let result = send_and_await_chunk_response( &node, &peer_id, message_bytes, request_id, - per_peer_timeout, + attempt_timeout, &peer_addrs, - |body| match body { - ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Success { - quote, - already_stored, - commitment, - }) => Some(classify_quote_response( - &peer_id, - &address, - "e, - already_stored, - commitment, - )), - ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error(e)) => Some(Err( - Error::Protocol(format!("Quote error from {peer_id}: {e}")), - )), - _ => None, - }, + |body| map_quote_response(&peer_id, &address, body), |e| Error::Network(format!("Failed to send quote request to {peer_id}: {e}")), || Error::Timeout(format!("Timeout waiting for quote from {peer_id}")), ) .await; + // Any recognised answer proves the peer parsed the versioned shape, + // including a structured error. Only silence and send failures leave the + // question open. + let answered = match &result { + Ok(_) => true, + Err(e) => !is_version_unaware(e), + }; + if !known_legacy && answered { + if let Ok(mut peers) = versioned_capable.lock() { + peers.insert(peer_id); + } + } + + // Only a storer that could not decode the versioned request is asked + // again in the legacy shape. One that answered has understood us, and a + // refusal must stay a refusal. + let result = match result { + Err(ref e) if is_version_unaware(e) && !known_legacy => { + // Remember it, so the next request to this peer skips the probe. + // Only silence counts as evidence: a send failure means the + // request never arrived, which says nothing about whether the peer + // could have parsed it, and caching that would strand a peer in + // the legacy shape for the rest of the session over one flaky send. + // A peer that has answered a versioned request before is never + // demoted: one lost response would otherwise pin an upgraded peer + // to the legacy shape for the rest of the session. + let ever_answered = versioned_capable + .lock() + .is_ok_and(|peers| peers.contains(&peer_id)); + if matches!(e, Error::Timeout(_)) && !ever_answered { + if let Ok(mut peers) = unversioned_peers.lock() { + peers.insert(peer_id); + } + } + let legacy = ChunkMessage { + request_id, + body: ChunkMessageBody::QuoteRequest(legacy_request), + }; + match legacy.encode() { + Ok(legacy_bytes) => { + send_and_await_chunk_response( + &node, + &peer_id, + legacy_bytes, + request_id, + per_peer_timeout, + &peer_addrs, + |body| map_quote_response(&peer_id, &address, body), + |e| { + Error::Network(format!( + "Failed to send quote request to {peer_id}: {e}" + )) + }, + || Error::Timeout(format!("Timeout waiting for quote from {peer_id}")), + ) + .await + } + Err(e) => Err(Error::Protocol(format!( + "Failed to encode quote request for {peer_id}: {e}" + ))), + } + } + other => other, + }; + (peer_id, peer_addrs, result) } +/// Turn a quote response into the quote it carries, or the error explaining +/// why there is none. Shared by the versioned request and its legacy retry. +/// +/// `ClientUpdateRequired` is separated from the generic protocol error because +/// it is terminal: it must reach the user with its own wording rather than +/// being counted as one more peer that failed to quote. +fn map_quote_response( + peer_id: &PeerId, + address: &[u8; 32], + body: ChunkMessageBody, +) -> Option { + match body { + ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Success { + quote, + already_stored, + commitment, + }) => Some(classify_quote_response( + peer_id, + address, + "e, + already_stored, + commitment, + )), + ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error( + ProtocolError::ClientUpdateRequired { + client_settlement_version, + min_settlement_version, + }, + )) => Some(Err(settlement_refusal_error( + peer_id, + client_settlement_version, + min_settlement_version, + ))), + ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error( + behind @ ProtocolError::StorerUpdateRequired { .. }, + )) => Some(Err(Error::StorerUpdateRequired(behind.to_string()))), + ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error(e)) => Some(Err( + Error::Protocol(format!("Quote error from {peer_id}: {e}")), + )), + _ => None, + } +} + +/// Turn a peer's `ClientUpdateRequired` into an error, rejecting one that does +/// not describe this client. +/// +/// A refusal is unauthenticated, so the least this client can do is check that +/// the peer is talking about the request it actually sent: the echoed version +/// must be ours, and the stated minimum must genuinely exceed it. A peer that +/// fails either is confused or lying, and is treated as an ordinary bad peer +/// rather than as evidence about this build. +pub(super) fn settlement_refusal_error( + peer_id: &PeerId, + client_settlement_version: u32, + min_settlement_version: u32, +) -> Error { + if client_settlement_version != CURRENT_SETTLEMENT_VERSION + || min_settlement_version <= client_settlement_version + { + return Error::Protocol(format!( + "Peer {peer_id} sent an incoherent settlement refusal (claimed this client is at \ + version {client_settlement_version} needing {min_settlement_version}, but this \ + client is at {CURRENT_SETTLEMENT_VERSION}); ignoring it" + )); + } + Error::ClientUpdateRequired(client_update_required_message( + client_settlement_version, + min_settlement_version, + )) +} + +/// Did this failure look like a storer that cannot parse a versioned request, +/// as opposed to one that parsed it and refused? See the merkle path for the +/// full reasoning. +/// +/// This is the single-node fallback, independent of the merkle one, so it +/// carries its own reference to the shared cutover guard. Without it, deleting +/// the merkle fallback would take the only build-time check with it and leave +/// this downgrade path live. +const _: () = crate::data::client::UNVERSIONED_RETRY_REQUIRES_MIN_V1; + +const fn is_version_unaware(error: &Error) -> bool { + matches!(error, Error::Network(_) | Error::Timeout(_)) +} + +/// Fold one peer's quote result into the collection state. +/// +/// Returns `Err` only when collection must stop outright: a storer has said +/// this client cannot settle, and no number of further quotes makes paying +/// safe. #[allow(clippy::too_many_arguments)] fn record_store_quote_result( peer_id: PeerId, @@ -396,7 +577,9 @@ fn record_store_quote_result( already_stored_peers: &mut Vec<(PeerId, [u8; 32])>, failures: &mut Vec, bad_quote_count: &mut usize, -) { + settlement_refusal: &mut Option, + refusals: &SettlementRefusals, +) -> Result<()> { match quote_result { Ok((quote, price, commitment)) => { quotes.push((peer_id, addrs, quote, price, commitment)); @@ -406,6 +589,43 @@ fn record_store_quote_result( let dist = peer_xor_distance(&peer_id, address); already_stored_peers.push((peer_id, dist)); } + // A storer that has explicitly declared this client incompatible ends + // quote collection here. Recording it as one more failed peer would + // let the remaining peers supply a quorum and go on to pay, which is + // the burn this mechanism exists to prevent, and would bury the + // upgrade instruction among ordinary per-peer failures. + // Recorded as well as returned. The caller runs this inside an overall + // timeout, and if that timeout fires the returned error is thrown away + // with the rest of the collection state. The verdict must outlive it: + // a client told it cannot settle must not pay, whatever else happened + // during collection. + Err(e @ Error::ClientUpdateRequired(_)) => { + // One peer's word is not a verdict about this build: nothing + // authenticates a refusal, so a single hostile peer answering this + // to everything would deny every upload. Believe it once enough + // distinct peers agree, which a genuine incompatibility reaches at + // once because every enforcing peer refuses. + let Some(corroborated) = refusals.note(peer_id, &e.to_string()) else { + warn!("Peer {peer_id} refused this client's settlement version; awaiting corroboration"); + failures.push(format!("{peer_id}: {e}")); + return Ok(()); + }; + // Name the corroborators. This is the only place the quorum is + // externally verifiable: the below-quorum branch above logs one + // peer at a time, so without this line a reader cannot tell a + // two-peer verdict from a one-peer misfire (V2-1109). + let corroborators = refusals.corroborating_peers(); + warn!( + "Settlement refusal corroborated by {} distinct peers [{}]; aborting before payment", + corroborators.len(), + corroborators.join(", ") + ); + let verdict = Error::ClientUpdateRequired(corroborated); + if settlement_refusal.is_none() { + *settlement_refusal = Some(Error::ClientUpdateRequired(verdict.to_string())); + } + return Err(verdict); + } Err(e) => { if matches!(&e, Error::BadQuoteBinding { .. }) { *bad_quote_count += 1; @@ -414,6 +634,7 @@ fn record_store_quote_result( failures.push(format!("{peer_id}: {e}")); } } + Ok(()) } fn witnessed_quote_launch_budget( @@ -1160,17 +1381,33 @@ impl Client { // network-broken) and the user benefits from seeing them called out. let mut bad_quote_count = 0usize; + // A storer's verdict that this client cannot settle, kept outside the + // collection loops so neither the overall timeout nor an early exit + // can discard it. Checked before any quote plan is built. + let mut settlement_refusal: Option = None; + let refusals = self.settlement_refusals(); + if staged_witnessed_collection { let mut quote_futures = FuturesUnordered::new(); let mut next_peer_index = 0usize; let collect_result: std::result::Result, _> = tokio::time::timeout(overall_timeout, async { loop { - let launch_count = witnessed_quote_launch_budget( - quotes.len(), - quote_futures.len(), - remote_peers.len().saturating_sub(next_peer_index), - ); + // Stop launching once the target is met, but keep + // draining below. Peers already in flight may yet + // declare this client unable to settle, and dropping + // that verdict because faster peers filled the quota + // would make it depend on response order. The surplus + // quotes are discarded; a refusal among them is not. + let launch_count = if quotes.len() >= target_quote_count { + 0 + } else { + witnessed_quote_launch_budget( + quotes.len(), + quote_futures.len(), + remote_peers.len().saturating_sub(next_peer_index), + ) + }; for _ in 0..launch_count { let (peer_id, peer_addrs) = &remote_peers[next_peer_index]; next_peer_index += 1; @@ -1183,10 +1420,12 @@ impl Client { data_size, data_type, per_peer_timeout, + self.unversioned_quote_peers(), + self.versioned_quote_capable_handle(), )); } - if quotes.len() >= target_quote_count || quote_futures.is_empty() { + if quote_futures.is_empty() { break; } @@ -1203,7 +1442,9 @@ impl Client { &mut already_stored_peers, &mut failures, &mut bad_quote_count, - ); + &mut settlement_refusal, + &refusals, + )?; } Ok(()) }) @@ -1219,6 +1460,11 @@ impl Client { Ok(Err(e)) => return Err(e), Ok(Ok(())) => {} } + // Outranks the timeout: a refusal says paying is unsafe no matter + // how many quotes were gathered before the clock ran out. + if let Some(refusal) = settlement_refusal.take() { + return Err(refusal); + } } else { // Merkle preflight keeps the previous behaviour: query the full // over-query set concurrently because those quote responses are @@ -1235,6 +1481,8 @@ impl Client { data_size, data_type, per_peer_timeout, + self.unversioned_quote_peers(), + self.versioned_quote_capable_handle(), )); } @@ -1250,7 +1498,9 @@ impl Client { &mut already_stored_peers, &mut failures, &mut bad_quote_count, - ); + &mut settlement_refusal, + &refusals, + )?; } Ok(()) }) @@ -1269,6 +1519,11 @@ impl Client { Ok(Err(e)) => return Err(e), Ok(Ok(())) => {} } + // Outranks the timeout: a refusal says paying is unsafe no matter + // how many quotes were gathered before the clock ran out. + if let Some(refusal) = settlement_refusal.take() { + return Err(refusal); + } } // Defensive double-check: the per-peer handler already filters @@ -2734,4 +2989,357 @@ mod tests { "off-curve quote must be dropped as BadQuoteCommitment; got {result:?}" ); } + + /// A storer's refusal must arrive as its own terminal error, carrying the + /// storer's wording. Folding it into the generic protocol error would bury + /// the upgrade instruction among ordinary per-peer quote failures, which + /// is the outcome this whole change exists to avoid. + #[test] + fn an_update_refusal_is_surfaced_with_its_upgrade_instruction() { + let peer_id = PeerId::from_bytes([0x42; 32]); + let refusal = ProtocolError::ClientUpdateRequired { + client_settlement_version: CURRENT_SETTLEMENT_VERSION, + min_settlement_version: CURRENT_SETTLEMENT_VERSION.saturating_add(1), + }; + + let mapped = map_quote_response( + &peer_id, + &[0x11; 32], + ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error(refusal)), + ); + + match mapped { + Some(Err(Error::ClientUpdateRequired(msg))) => { + assert!(msg.contains("ant update"), "{msg}"); + assert!(msg.contains("nothing was charged"), "{msg}"); + } + other => panic!("expected ClientUpdateRequired, got: {other:?}"), + } + } + + /// The legacy retry exists for storers that cannot parse a versioned + /// request, which are silent. A storer that answered has understood us, so + /// retrying its refusal without the version would talk it into quoting a + /// client that cannot pay. That is the exact failure this change removes, + /// so the predicate deciding it is pinned. + #[test] + fn only_silence_triggers_the_legacy_retry() { + assert!(is_version_unaware(&Error::Timeout("no answer".into()))); + assert!(is_version_unaware(&Error::Network("send failed".into()))); + + assert!(!is_version_unaware(&Error::ClientUpdateRequired( + "too old".into() + ))); + // A storer that says it is the old side has understood the request. + // Retrying it unversioned would obtain a quote from a peer that cannot + // verify the resulting payment, which is a burn. + assert!(!is_version_unaware(&Error::StorerUpdateRequired( + "node behind".into() + ))); + assert!(!is_version_unaware(&Error::Protocol( + "quote error from peer".into() + ))); + } + + /// A storer declaring itself the old side is an ordinary skippable peer, + /// not a client fault. Surfacing it as `ClientUpdateRequired` would tell an + /// up-to-date user to upgrade, and during a client-first rollout it would + /// tell that to nearly everyone. + #[test] + fn a_storer_that_is_behind_is_not_reported_as_the_clients_fault() { + let peer_id = PeerId::from_bytes([0x43; 32]); + let mapped = map_quote_response( + &peer_id, + &[0x11; 32], + ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error( + ProtocolError::StorerUpdateRequired { + client_settlement_version: 2, + node_settlement_version: 1, + }, + )), + ); + + match mapped { + Some(Err(Error::StorerUpdateRequired(msg))) => { + assert!(msg.contains("use a different storer"), "{msg}"); + assert!(!msg.contains("ant update"), "{msg}"); + } + other => panic!("expected StorerUpdateRequired, got: {other:?}"), + } + } + + /// The refusal has to stop quote collection, not join the failure list. + /// If it is merely recorded, the remaining peers can still form a quorum + /// and the upload proceeds to pay, which is exactly the burn the gate is + /// meant to prevent. + #[test] + fn a_refusal_aborts_quote_collection_instead_of_counting_as_one_bad_peer() { + let mut quotes = Vec::new(); + let mut already_stored = Vec::new(); + let mut failures = Vec::new(); + let mut bad_quotes = 0usize; + let mut refusal_slot: Option = None; + + let refusals = SettlementRefusals::default(); + let mut refuse = + |peer: u8, failures: &mut Vec, slot: &mut Option| -> Result<()> { + record_store_quote_result( + PeerId::from_bytes([peer; 32]), + Vec::new(), + Err(Error::ClientUpdateRequired( + "too old, run ant update".into(), + )), + &[0x11; 32], + &mut quotes, + &mut already_stored, + failures, + &mut bad_quotes, + slot, + &refusals, + ) + }; + + // One peer is not corroboration: recorded as an ordinary bad peer, so a + // single hostile responder cannot deny every upload. + let first = refuse(0x44, &mut failures, &mut refusal_slot); + assert!(first.is_ok(), "one peer must not abort, got {first:?}"); + assert_eq!(failures.len(), 1); + assert!(refusal_slot.is_none()); + + // A second, distinct peer makes it a verdict about this build. + let second = refuse(0x45, &mut failures, &mut refusal_slot); + assert!( + matches!(second, Err(Error::ClientUpdateRequired(_))), + "a corroborated refusal must propagate, got {second:?}" + ); + } + + /// A storer being behind must NOT abort. Otherwise one lagging peer in the + /// close group fails an upload that the rest of the group could serve. + #[test] + fn a_storer_that_is_behind_does_not_abort_collection() { + let mut quotes = Vec::new(); + let mut already_stored = Vec::new(); + let mut failures = Vec::new(); + let mut bad_quotes = 0usize; + let mut refusal_slot: Option = None; + + let outcome = record_store_quote_result( + PeerId::from_bytes([0x45; 32]), + Vec::new(), + Err(Error::StorerUpdateRequired("node behind".into())), + &[0x11; 32], + &mut quotes, + &mut already_stored, + &mut failures, + &mut bad_quotes, + &mut refusal_slot, + &SettlementRefusals::default(), + ); + + assert!( + outcome.is_ok(), + "a lagging storer must be skipped, not fatal" + ); + assert_eq!(failures.len(), 1, "and it should be recorded as a skip"); + assert!( + refusal_slot.is_none(), + "a node being behind is not a verdict about this client" + ); + } + + /// The refusal must survive the overall collection timeout. + /// + /// The collector runs inside `tokio::time::timeout`, and its elapsed arm + /// deliberately falls through so quotes gathered from fast peers stay + /// usable. That arm would otherwise discard a refusal observed just before + /// the clock ran out, and the upload would pay anyway. Recording the + /// verdict in a slot that outlives the timeout is what prevents it, so the + /// slot is what gets tested. + #[test] + fn a_refusal_is_recorded_where_the_collection_timeout_cannot_discard_it() { + let mut quotes = Vec::new(); + let mut already_stored = Vec::new(); + let mut failures = Vec::new(); + let mut bad_quotes = 0usize; + let mut refusal_slot: Option = None; + + let refusals = SettlementRefusals::default(); + for peer in [0x46u8, 0x47u8] { + let _ = record_store_quote_result( + PeerId::from_bytes([peer; 32]), + Vec::new(), + Err(Error::ClientUpdateRequired( + "too old, run ant update".into(), + )), + &[0x11; 32], + &mut quotes, + &mut already_stored, + &mut failures, + &mut bad_quotes, + &mut refusal_slot, + &refusals, + ); + } + + match refusal_slot { + Some(Error::ClientUpdateRequired(msg)) => { + assert!(msg.contains("ant update"), "{msg}"); + } + other => panic!("refusal must outlive the collection state, got {other:?}"), + } + } + + /// Once the quote target is met the collector stops launching new peers + /// but keeps draining those already in flight, so a refusal cannot be + /// missed merely because faster peers filled the quota first. + /// + /// The launch budget enforces the first half, and the drain relies on it + /// reaching zero to terminate rather than recruiting forever. + #[test] + fn meeting_the_target_stops_launching_without_stopping_collection() { + assert!( + witnessed_quote_launch_budget(0, 0, 32) > 0, + "collection must start" + ); + assert_eq!(witnessed_quote_launch_budget(CLOSE_GROUP_SIZE, 0, 32), 0); + assert_eq!( + witnessed_quote_launch_budget(CLOSE_GROUP_SIZE.saturating_add(1), 0, 32), + 0 + ); + // In-flight peers count against the budget, so draining them does not + // pull in replacements. + assert_eq!(witnessed_quote_launch_budget(0, CLOSE_GROUP_SIZE, 32), 0); + } + + /// The probe must be paid per peer, not per request. + /// + /// A storer that predates the versioned request never answers it, so the + /// client eats the probe wait before falling back. Without remembering the + /// answer that cost lands on every quote, and a merkle pool asks sixteen + /// candidates. Measured on the merkle E2E suite against a fleet on + /// published ant-node, re-probing took the run from ~24 minutes to past + /// the 60-minute CI cap. + /// + /// The cache is consulted, not enforced: concurrent first contacts are not + /// single-flighted, so the same peer can be probed by a few in-flight + /// requests before any of them records the answer. What the cache + /// guarantees is that later rounds do not re-probe. + #[test] + fn a_peer_that_cannot_answer_a_versioned_quote_is_only_probed_once() { + let peers: Arc>> = Arc::new(Mutex::new(HashSet::new())); + let legacy_peer = PeerId::from_bytes([0x51; 32]); + let fresh_peer = PeerId::from_bytes([0x52; 32]); + + let known = |p: &PeerId| peers.lock().expect("cache lock").contains(p); + + // First contact: nothing known, so the versioned request is sent. + assert!(!known(&legacy_peer)); + + // Silence records the peer. + peers.lock().expect("cache lock").insert(legacy_peer); + + // Second contact skips the probe entirely. + assert!(known(&legacy_peer)); + // and does not tar every other peer with the same brush. + assert!(!known(&fresh_peer)); + } + + /// Only silence is evidence that a peer cannot parse the versioned shape. + /// + /// Both a timeout and a send failure trigger the legacy retry, but they + /// mean different things: a send failure says the request never arrived, + /// so it teaches nothing about the peer's capabilities. Caching it would + /// strand that peer in the legacy shape for the rest of the session over + /// one flaky send. + #[test] + fn only_silence_is_evidence_worth_caching() { + assert!(matches!( + Error::Timeout("no answer".into()), + Error::Timeout(_) + )); + assert!(!matches!( + Error::Network("send failed".into()), + Error::Timeout(_) + )); + // Both still take the fallback, so a send failure is retried rather + // than left to fail outright. + assert!(is_version_unaware(&Error::Network("send failed".into()))); + assert!(is_version_unaware(&Error::Timeout("no answer".into()))); + } + + /// One peer cannot condemn the client. + /// + /// Nothing authenticates a refusal, so a single hostile or misconfigured + /// storer answering `ClientUpdateRequired` to everything would otherwise + /// abort every upload. That turns an over-query design which tolerates + /// many bad peers into one that tolerates none. + #[test] + fn a_lone_peer_cannot_condemn_the_client() { + let refusals = SettlementRefusals::default(); + assert!( + refusals + .note(PeerId::from_bytes([0x61; 32]), "too old") + .is_none(), + "one peer is not corroboration" + ); + assert!(refusals.corroborated().is_none()); + // The same peer repeating itself is still one peer. + assert!(refusals + .note(PeerId::from_bytes([0x61; 32]), "too old") + .is_none()); + assert!(refusals.corroborated().is_none()); + } + + /// A genuine incompatibility reaches the threshold at once, because every + /// peer enforcing the newer rule refuses. + #[test] + fn a_second_peer_makes_the_refusal_terminal_and_it_stays_latched() { + let refusals = SettlementRefusals::default(); + refusals.note(PeerId::from_bytes([0x62; 32]), "run ant update"); + let verdict = refusals.note(PeerId::from_bytes([0x63; 32]), "run ant update"); + + assert!(verdict.is_some_and(|m| m.contains("ant update"))); + // Latched: an upload starting later must see it before it spends, + // which is the whole point of holding it on the client rather than in + // one collector's local state. + assert!(refusals + .corroborated() + .is_some_and(|m| m.contains("ant update"))); + } + + /// A refusal that does not describe this client is a confused or lying + /// peer, not evidence about this build, and must not count toward the + /// threshold. + #[test] + fn an_incoherent_refusal_is_treated_as_a_bad_peer() { + let peer_id = PeerId::from_bytes([0x64; 32]); + + // Claims to be about some other client version. + let wrong_echo = settlement_refusal_error( + &peer_id, + CURRENT_SETTLEMENT_VERSION.saturating_add(7), + CURRENT_SETTLEMENT_VERSION.saturating_add(8), + ); + assert!(matches!(wrong_echo, Error::Protocol(_)), "{wrong_echo:?}"); + + // Claims a minimum that our version already satisfies. + let no_gap = settlement_refusal_error( + &peer_id, + CURRENT_SETTLEMENT_VERSION, + CURRENT_SETTLEMENT_VERSION, + ); + assert!(matches!(no_gap, Error::Protocol(_)), "{no_gap:?}"); + + // A coherent one is believed, and carries the upgrade instruction. + let real = settlement_refusal_error( + &peer_id, + CURRENT_SETTLEMENT_VERSION, + CURRENT_SETTLEMENT_VERSION.saturating_add(1), + ); + match real { + Error::ClientUpdateRequired(msg) => assert!(msg.contains("ant update"), "{msg}"), + other => panic!("expected ClientUpdateRequired, got {other:?}"), + } + } } diff --git a/ant-core/src/data/error.rs b/ant-core/src/data/error.rs index 552277f..4a38d9c 100644 --- a/ant-core/src/data/error.rs +++ b/ant-core/src/data/error.rs @@ -99,6 +99,30 @@ pub enum Error { #[error("insufficient peers: {0}")] InsufficientPeers(String), + /// The network refused to quote this client because it settles payments + /// under superseded rules. + /// + /// Deliberately terminal. A client that reaches this would pay an amount + /// every storer rejects, and merkle payments are not refundable, so + /// retrying or falling back to an older request shape would convert a + /// clean refusal into destroyed money. The message is the storer's own + /// wording, which already tells the user how to upgrade and that nothing + /// has been charged. + #[error("{0}")] + ClientUpdateRequired(String), + + /// A storer declined to quote because *it* settles under older rules than + /// this client. + /// + /// The opposite of [`Self::ClientUpdateRequired`] and deliberately not + /// terminal. Nothing is wrong with this client, so the upload should use a + /// different peer and say nothing to the user. During a client-first + /// rollout most of the fleet is briefly in this state. If too few peers + /// remain the operation fails for lack of quotes, which is the correct + /// outcome: it fails before any payment rather than after. + #[error("{0}")] + StorerUpdateRequired(String), + /// BLS signature verification failed. #[error("signature verification failed: {0}")] SignatureVerification(String),