From dec92a0332d1c0c4200cc31eb27a72aa201f006e Mon Sep 17 00:00:00 2001 From: Warm Beer Date: Mon, 15 Jun 2026 09:56:17 +0200 Subject: [PATCH 1/3] feat(client): apply witnessed quote policy locally Consume the witnessed close-group transcript from saorsa-core and compute quorum, vote counts, and fallback quote candidates in ant-client. Quote collection now keeps all quorum-recognised candidates available for reachability fallback, then pays the closest successful close group. SemVer: feature change; no public ant-client API break expected. --- Cargo.lock | 6 +- ant-core/Cargo.toml | 14 +- ant-core/src/data/client/quote.rs | 224 ++++++++++++++++++++++++------ ant-core/src/data/network.rs | 14 +- 4 files changed, 198 insertions(+), 60 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 4e51ac8d..71767eaa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -893,7 +893,7 @@ dependencies = [ [[package]] name = "ant-node" version = "0.12.1-rc.6" -source = "git+https://github.com/WithAutonomi/ant-node?branch=rc-2026.6.2#9ba96e24db172264028fd7a9cc39bb062018324d" +source = "git+https://github.com/WithAutonomi/ant-node?branch=feat%2Fwitnessed-transcript-policy#cd56dca9123a1af4bcde39c97ca26f6c8a52be30" dependencies = [ "ant-protocol", "blake3", @@ -942,7 +942,7 @@ dependencies = [ [[package]] name = "ant-protocol" version = "2.1.4-rc.2" -source = "git+https://github.com/WithAutonomi/ant-protocol?branch=rc-2026.6.2#af8fadc828266c52169bccd106f3ebfd9e409b36" +source = "git+https://github.com/WithAutonomi/ant-protocol?branch=feat%2Fwitnessed-transcript-policy#d73d7656b4fcc7f96d06dfb98c7b9155edd50e08" dependencies = [ "blake3", "bytes", @@ -5249,7 +5249,7 @@ dependencies = [ [[package]] name = "saorsa-core" version = "0.25.1-rc.2" -source = "git+https://github.com/saorsa-labs/saorsa-core?branch=rc-2026.6.2#f2dab914367b687342cc758b0cc8c2b10d824f31" +source = "git+https://github.com/saorsa-labs/saorsa-core?branch=feat%2Fwitnessed-transcript-policy#d6877658b3b8248aeaeb7f422effd7d8b5499f98" dependencies = [ "anyhow", "async-trait", diff --git a/ant-core/Cargo.toml b/ant-core/Cargo.toml index 74f2b64a..d9532c93 100644 --- a/ant-core/Cargo.toml +++ b/ant-core/Cargo.toml @@ -37,7 +37,7 @@ 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 = { git = "https://github.com/WithAutonomi/ant-protocol", branch = "rc-2026.6.2" } +ant-protocol = { git = "https://github.com/WithAutonomi/ant-protocol", branch = "feat/witnessed-transcript-policy" } xor_name = "5" self_encryption = "0.36" futures = "0.3" @@ -61,11 +61,11 @@ sysinfo = { version = "0.32", default-features = false, features = ["system"] } # Must track the same `saorsa-core` / `ant-protocol` line as the # `ant-protocol` pin above — a version skew pulls a second copy of # `saorsa-core` into the graph and makes `ant_node`'s and `ant_protocol`'s -# `MultiAddr` mutually incompatible in `node/devnet.rs`. During an RC the -# runtime `ant-protocol` pin above points at a git rc branch, so this -# ant-node must point at the matching ant-node rc branch (carrying the same -# saorsa-core / ant-protocol lineage) rather than a released version. -ant-node = { git = "https://github.com/WithAutonomi/ant-node", branch = "rc-2026.6.2", optional = true } +# `MultiAddr` mutually incompatible in `node/devnet.rs`. While the runtime +# `ant-protocol` pin above points at a git branch, this ant-node must point at +# the matching ant-node branch carrying the same saorsa-core / ant-protocol +# lineage rather than a released version. +ant-node = { git = "https://github.com/WithAutonomi/ant-node", branch = "feat/witnessed-transcript-policy", optional = true } tracing-subscriber = { version = "0.3", features = ["env-filter"] } [target.'cfg(unix)'.dependencies] @@ -93,7 +93,7 @@ devnet = ["dep:ant-node"] # always compile even without the `devnet` feature. Pinned to the same # version as the runtime dep so there is a single ant-node / # saorsa-core version across the whole graph. -ant-node = { git = "https://github.com/WithAutonomi/ant-node", branch = "rc-2026.6.2" } +ant-node = { git = "https://github.com/WithAutonomi/ant-node", branch = "feat/witnessed-transcript-policy" } serial_test = "3" anyhow = "1" alloy = { version = "1.6", features = ["node-bindings"] } diff --git a/ant-core/src/data/client/quote.rs b/ant-core/src/data/client/quote.rs index 6d89eca2..ecbf4618 100644 --- a/ant-core/src/data/client/quote.rs +++ b/ant-core/src/data/client/quote.rs @@ -7,12 +7,13 @@ use crate::data::client::peer_xor_distance; use crate::data::client::Client; use crate::data::error::{Error, Result}; use ant_protocol::evm::{Amount, PaymentQuote}; -use ant_protocol::transport::{MultiAddr, PeerId, WitnessedCloseGroup}; +use ant_protocol::transport::{DHTNode, MultiAddr, PeerId, WitnessedCloseGroup}; use ant_protocol::{ compute_address, send_and_await_chunk_response, ChunkMessage, ChunkMessageBody, ChunkQuoteRequest, ChunkQuoteResponse, CLOSE_GROUP_MAJORITY, CLOSE_GROUP_SIZE, }; use futures::stream::{FuturesUnordered, StreamExt}; +use std::collections::{HashMap, HashSet}; use std::time::Duration; use tracing::{debug, info, warn}; @@ -170,6 +171,12 @@ fn peer_list(peers: &[PeerId]) -> Vec { peers.iter().map(ToString::to_string).collect() } +#[derive(Debug, Clone)] +struct WitnessedQuoteCandidate { + node: DHTNode, + votes: usize, +} + fn witnessed_initial_peers(witnessed: &WitnessedCloseGroup) -> Vec { witnessed .initial_closest @@ -182,38 +189,128 @@ fn witnessed_responder_views(witnessed: &WitnessedCloseGroup) -> Vec { witnessed .responder_views .iter() - .map(|view| format!("{}=>{:?}", view.responder, peer_list(&view.closest))) + .map(|view| { + let peers = view + .closest + .iter() + .map(|node| node.peer_id) + .collect::>(); + format!("{}=>{:?}", view.responder, peer_list(&peers)) + }) .collect() } -fn witnessed_vote_counts(witnessed: &WitnessedCloseGroup) -> Vec { - witnessed - .vote_counts +fn merge_witnessed_node(nodes: &mut HashMap, node: DHTNode) { + match nodes.entry(node.peer_id) { + std::collections::hash_map::Entry::Occupied(mut entry) => { + entry.get_mut().merge_from(node); + } + std::collections::hash_map::Entry::Vacant(entry) => { + entry.insert(node); + } + } +} + +fn sort_vote_counts_by_distance(vote_counts: &mut [(PeerId, usize)], address: &[u8; 32]) { + vote_counts.sort_by(|left, right| { + peer_xor_distance(&left.0, address) + .cmp(&peer_xor_distance(&right.0, address)) + .then_with(|| left.0.as_bytes().cmp(right.0.as_bytes())) + }); +} + +fn witnessed_vote_counts_and_nodes( + witnessed: &WitnessedCloseGroup, + address: &[u8; 32], +) -> (HashMap, Vec<(PeerId, usize)>) { + let mut known_nodes = HashMap::new(); + for node in &witnessed.initial_closest { + merge_witnessed_node(&mut known_nodes, node.clone()); + } + + let mut vote_counts_by_peer = HashMap::new(); + for view in &witnessed.responder_views { + let mut voted = HashSet::new(); + for node in &view.closest { + merge_witnessed_node(&mut known_nodes, node.clone()); + if voted.insert(node.peer_id) { + *vote_counts_by_peer.entry(node.peer_id).or_insert(0) += 1; + } + } + } + + let mut vote_counts: Vec<(PeerId, usize)> = vote_counts_by_peer.into_iter().collect(); + sort_vote_counts_by_distance(&mut vote_counts, address); + (known_nodes, vote_counts) +} + +fn witnessed_consensus_candidates( + witnessed: &WitnessedCloseGroup, + address: &[u8; 32], + quorum: usize, +) -> Vec { + let (known_nodes, vote_counts) = witnessed_vote_counts_and_nodes(witnessed, address); + let mut candidates = vote_counts + .iter() + .filter_map(|(peer_id, votes)| { + if *votes < quorum { + return None; + } + known_nodes + .get(peer_id) + .cloned() + .map(|node| WitnessedQuoteCandidate { + node, + votes: *votes, + }) + }) + .collect::>(); + + candidates.sort_by(|left, right| { + peer_xor_distance(&left.node.peer_id, address) + .cmp(&peer_xor_distance(&right.node.peer_id, address)) + .then_with(|| { + left.node + .peer_id + .as_bytes() + .cmp(right.node.peer_id.as_bytes()) + }) + }); + candidates +} + +fn witnessed_vote_counts(witnessed: &WitnessedCloseGroup, address: &[u8; 32]) -> Vec { + let (_, vote_counts) = witnessed_vote_counts_and_nodes(witnessed, address); + vote_counts .iter() .map(|(peer_id, votes)| format!("{peer_id}:{votes}")) .collect() } -fn witnessed_consensus(witnessed: &WitnessedCloseGroup) -> Vec { - witnessed - .consensus +fn witnessed_consensus( + witnessed: &WitnessedCloseGroup, + address: &[u8; 32], + quorum: usize, +) -> Vec { + witnessed_consensus_candidates(witnessed, address, quorum) .iter() - .map(|node| format!("{}:{}", node.node.peer_id, node.votes)) + .map(|candidate| format!("{}:{}", candidate.node.peer_id, candidate.votes)) .collect() } fn witnessed_close_group_diagnostics( address: &[u8; 32], witnessed: &WitnessedCloseGroup, + quorum: usize, ) -> String { format!( "target={}, initial={:?}, responder_views={:?}, vote_counts={:?}, quorum={}, final={:?}", hex::encode(address), witnessed_initial_peers(witnessed), witnessed_responder_views(witnessed), - witnessed_vote_counts(witnessed), - witnessed.quorum, - witnessed_consensus(witnessed) + witnessed_vote_counts(witnessed, address), + quorum, + witnessed_consensus(witnessed, address, quorum) ) } @@ -221,20 +318,20 @@ fn witnessed_quote_peers_or_error( address: &[u8; 32], witnessed: &WitnessedCloseGroup, required: usize, + quorum: usize, ) -> Result)>> { - if witnessed.consensus.len() < required { + let candidates = witnessed_consensus_candidates(witnessed, address, quorum); + if candidates.len() < required { return Err(Error::InsufficientPeers(format!( "Witnessed close group inconclusive before payment: got {}/{} quorum-recognised peers. {}", - witnessed.consensus.len(), + candidates.len(), required, - witnessed_close_group_diagnostics(address, witnessed) + witnessed_close_group_diagnostics(address, witnessed, quorum) ))); } - Ok(witnessed - .consensus - .iter() - .take(required) + Ok(candidates + .into_iter() .map(|candidate| { ( candidate.node.peer_id, @@ -265,9 +362,11 @@ pub(crate) fn median_paid_quote_issuer( impl Client { /// Get storage quotes from the closest peers for a given address. /// - /// Builds a quorum-witnessed close group of exactly `CLOSE_GROUP_SIZE` - /// peers, requests quotes from all of them concurrently, and returns those - /// responders sorted by XOR distance. + /// Builds a quorum-witnessed candidate set with at least + /// `CLOSE_GROUP_SIZE` peers, requests quotes from all of them concurrently, + /// and returns the closest `CLOSE_GROUP_SIZE` successful responders sorted + /// by XOR distance. Farther quorum-recognised candidates are used only as + /// fallbacks when closer candidates do not return usable quotes. /// /// Returns `Error::AlreadyStored` early if `CLOSE_GROUP_MAJORITY` peers /// report the chunk is already stored. @@ -316,7 +415,7 @@ impl Client { let quorum = witnessed_close_group_quorum(); let witnessed = self .network() - .find_witnessed_close_group(address, required, quorum) + .find_witnessed_close_group(address, required) .await .map_err(|e| { Error::InsufficientPeers(format!( @@ -327,15 +426,15 @@ impl Client { debug!( target = %hex::encode(address), - quorum = witnessed.quorum, + quorum = quorum, initial = ?witnessed_initial_peers(&witnessed), responder_views = ?witnessed_responder_views(&witnessed), - vote_counts = ?witnessed_vote_counts(&witnessed), - final_witnessed_set = ?witnessed_consensus(&witnessed), + vote_counts = ?witnessed_vote_counts(&witnessed, address), + final_witnessed_set = ?witnessed_consensus(&witnessed, address, quorum), "Witnessed close group selected for SNP quote collection" ); - witnessed_quote_peers_or_error(address, &witnessed, required) + witnessed_quote_peers_or_error(address, &witnessed, required, quorum) } #[allow(clippy::too_many_lines)] @@ -575,9 +674,7 @@ mod tests { use super::*; use ant_protocol::evm::RewardsAddress; use ant_protocol::pqc::ops::{MlDsaOperations, MlDsaPublicKey}; - use ant_protocol::transport::{ - ConsensusNode, DHTNode, MlDsa65, ResponderView, WitnessedCloseGroup, - }; + use ant_protocol::transport::{DHTNode, MlDsa65, ResponderView, WitnessedCloseGroup}; use std::time::SystemTime; use xor_name::XorName; @@ -647,6 +744,13 @@ mod tests { seeds.iter().copied().map(witnessed_test_node).collect() } + fn witnessed_test_view(responder: u8, closest: &[u8]) -> ResponderView { + ResponderView { + responder: PeerId::from_bytes([responder; 32]), + closest: witnessed_test_nodes(closest), + } + } + /// Independent re-implementation of the storer-side binding spec /// (`ant-node/src/payment/verifier.rs::validate_peer_bindings` + /// `peer_id_from_public_key_bytes`): @@ -742,25 +846,23 @@ mod tests { #[test] fn witnessed_quote_peers_error_is_typed_and_pre_payment_when_consensus_is_short() { let address = [0u8; 32]; - let consensus: Vec = witnessed_test_nodes(&[1, 2, 3, 4, 5, 6]) - .into_iter() - .map(|node| ConsensusNode { node, votes: 5 }) + let responder_views = (1..=7) + .map(|responder| witnessed_test_view(responder, &[1, 2, 3, 4])) .collect(); let witnessed = WitnessedCloseGroup { target: address, k: CLOSE_GROUP_SIZE, - quorum: witnessed_close_group_quorum(), initial_closest: witnessed_test_nodes(&[1, 2, 3, 4, 5, 6, 7]), - responder_views: vec![ResponderView { - responder: PeerId::from_bytes([1; 32]), - closest: vec![PeerId::from_bytes([1; 32])], - }], - vote_counts: vec![(PeerId::from_bytes([1; 32]), 5)], - consensus, + responder_views, }; - let err = witnessed_quote_peers_or_error(&address, &witnessed, CLOSE_GROUP_SIZE) - .expect_err("short witnessed consensus must fail before payment"); + let err = witnessed_quote_peers_or_error( + &address, + &witnessed, + CLOSE_GROUP_SIZE, + witnessed_close_group_quorum(), + ) + .expect_err("short witnessed consensus must fail before payment"); match err { Error::InsufficientPeers(message) => { @@ -772,6 +874,44 @@ mod tests { } } + #[test] + fn witnessed_quote_peers_include_quorum_fallback_candidates() { + const EXTRA_QUORUM_CANDIDATES: usize = 1; + + let address = [0u8; 32]; + let witnessed = WitnessedCloseGroup { + target: address, + k: CLOSE_GROUP_SIZE, + initial_closest: witnessed_test_nodes(&[1, 2, 3, 4, 5, 6, 7]), + responder_views: vec![ + witnessed_test_view(1, &[1, 2, 3, 4, 5, 6, 7]), + witnessed_test_view(2, &[1, 2, 3, 4, 5, 6, 8]), + witnessed_test_view(3, &[1, 2, 3, 4, 5, 7, 8]), + witnessed_test_view(4, &[1, 2, 3, 4, 6, 7, 8]), + witnessed_test_view(5, &[1, 2, 3, 5, 6, 7, 8]), + witnessed_test_view(6, &[1, 2, 4, 5, 6, 7, 8]), + witnessed_test_view(7, &[1, 3, 4, 5, 6, 7, 8]), + ], + }; + + let peers = witnessed_quote_peers_or_error( + &address, + &witnessed, + CLOSE_GROUP_SIZE, + witnessed_close_group_quorum(), + ) + .expect("fallback candidates should be retained for quote collection"); + + assert_eq!(peers.len(), CLOSE_GROUP_SIZE + EXTRA_QUORUM_CANDIDATES); + assert_eq!( + peers + .iter() + .map(|(peer_id, _)| peer_id.as_bytes()[0]) + .collect::>(), + vec![1, 2, 3, 4, 5, 6, 7, 8] + ); + } + #[test] fn filter_drops_only_bad_bindings_and_leaves_storer_acceptable_quotes() { let mut quotes = vec![ diff --git a/ant-core/src/data/network.rs b/ant-core/src/data/network.rs index ab52c3a5..53e9015c 100644 --- a/ant-core/src/data/network.rs +++ b/ant-core/src/data/network.rs @@ -131,26 +131,24 @@ impl Network { .collect()) } - /// Find a quorum-witnessed close group for a target address. + /// Find a witnessed close-group transcript for a target address. /// /// The underlying DHT method returns the initial client K, each responder's - /// self-inclusive closest-K view, vote counts, and the final - /// quorum-recognised set ordered by pure XOR distance. + /// self-inclusive closest-K node view, and enough trusted node records for + /// callers to apply their own quorum and fallback policy. /// /// # Errors /// - /// Returns an error if the DHT lookup itself fails. The returned witnessed - /// group may still be inconclusive; callers should check - /// `WitnessedCloseGroup::is_complete`. + /// Returns an error if the DHT lookup itself fails. The returned transcript + /// may still be inconclusive; callers should evaluate it before payment. pub async fn find_witnessed_close_group( &self, target: &[u8; 32], count: usize, - quorum: usize, ) -> Result { self.node .dht() - .find_witnessed_close_group(target, count, quorum) + .find_witnessed_close_group(target, count) .await .map_err(|e| Error::Network(format!("DHT witnessed close-group lookup failed: {e}"))) } From 652f62ea7285a467975235d7bab4de4bb5fabd7c Mon Sep 17 00:00:00 2001 From: Warm Beer Date: Mon, 15 Jun 2026 12:14:07 +0200 Subject: [PATCH 2/3] fix(client): require median quote voter majority Select the closest witnessed SNP quote set whose paid median issuer is recognised by a close-group majority of the selected peers. This keeps fallback quote candidates available without paying a median issuer that the PUT majority may reject. SemVer: bug fix; no public ant-client API break expected. --- ant-core/src/data/client/quote.rs | 354 ++++++++++++++++++++++++++---- 1 file changed, 315 insertions(+), 39 deletions(-) diff --git a/ant-core/src/data/client/quote.rs b/ant-core/src/data/client/quote.rs index ecbf4618..c7ef959d 100644 --- a/ant-core/src/data/client/quote.rs +++ b/ant-core/src/data/client/quote.rs @@ -171,10 +171,27 @@ fn peer_list(peers: &[PeerId]) -> Vec { peers.iter().map(ToString::to_string).collect() } +type StoreQuote = (PeerId, Vec, PaymentQuote, Amount); +type VotersByPeer = HashMap>; +type WitnessedVoteData = (HashMap, VotersByPeer, Vec<(PeerId, usize)>); + #[derive(Debug, Clone)] struct WitnessedQuoteCandidate { node: DHTNode, votes: usize, + voters: HashSet, +} + +#[derive(Debug, Clone)] +struct WitnessedQuotePeer { + peer_id: PeerId, + addrs: Vec, + voters: HashSet, +} + +enum QuoteSelectionPolicy { + ClosestByDistance, + WitnessedMedianVoters { voters_by_peer: VotersByPeer }, } fn witnessed_initial_peers(witnessed: &WitnessedCloseGroup) -> Vec { @@ -222,26 +239,32 @@ fn sort_vote_counts_by_distance(vote_counts: &mut [(PeerId, usize)], address: &[ fn witnessed_vote_counts_and_nodes( witnessed: &WitnessedCloseGroup, address: &[u8; 32], -) -> (HashMap, Vec<(PeerId, usize)>) { +) -> WitnessedVoteData { let mut known_nodes = HashMap::new(); for node in &witnessed.initial_closest { merge_witnessed_node(&mut known_nodes, node.clone()); } - let mut vote_counts_by_peer = HashMap::new(); + let mut voters_by_peer: HashMap> = HashMap::new(); for view in &witnessed.responder_views { let mut voted = HashSet::new(); for node in &view.closest { merge_witnessed_node(&mut known_nodes, node.clone()); if voted.insert(node.peer_id) { - *vote_counts_by_peer.entry(node.peer_id).or_insert(0) += 1; + voters_by_peer + .entry(node.peer_id) + .or_default() + .insert(view.responder); } } } - let mut vote_counts: Vec<(PeerId, usize)> = vote_counts_by_peer.into_iter().collect(); + let mut vote_counts: Vec<(PeerId, usize)> = voters_by_peer + .iter() + .map(|(peer_id, voters)| (*peer_id, voters.len())) + .collect(); sort_vote_counts_by_distance(&mut vote_counts, address); - (known_nodes, vote_counts) + (known_nodes, voters_by_peer, vote_counts) } fn witnessed_consensus_candidates( @@ -249,20 +272,24 @@ fn witnessed_consensus_candidates( address: &[u8; 32], quorum: usize, ) -> Vec { - let (known_nodes, vote_counts) = witnessed_vote_counts_and_nodes(witnessed, address); + let (known_nodes, voters_by_peer, vote_counts) = + witnessed_vote_counts_and_nodes(witnessed, address); let mut candidates = vote_counts .iter() .filter_map(|(peer_id, votes)| { if *votes < quorum { return None; } - known_nodes - .get(peer_id) - .cloned() - .map(|node| WitnessedQuoteCandidate { - node, - votes: *votes, - }) + known_nodes.get(peer_id).cloned().and_then(|node| { + voters_by_peer + .get(peer_id) + .cloned() + .map(|voters| WitnessedQuoteCandidate { + node, + votes: *votes, + voters, + }) + }) }) .collect::>(); @@ -280,7 +307,7 @@ fn witnessed_consensus_candidates( } fn witnessed_vote_counts(witnessed: &WitnessedCloseGroup, address: &[u8; 32]) -> Vec { - let (_, vote_counts) = witnessed_vote_counts_and_nodes(witnessed, address); + let (_, _, vote_counts) = witnessed_vote_counts_and_nodes(witnessed, address); vote_counts .iter() .map(|(peer_id, votes)| format!("{peer_id}:{votes}")) @@ -319,7 +346,7 @@ fn witnessed_quote_peers_or_error( witnessed: &WitnessedCloseGroup, required: usize, quorum: usize, -) -> Result)>> { +) -> Result> { let candidates = witnessed_consensus_candidates(witnessed, address, quorum); if candidates.len() < required { return Err(Error::InsufficientPeers(format!( @@ -332,11 +359,10 @@ fn witnessed_quote_peers_or_error( Ok(candidates .into_iter() - .map(|candidate| { - ( - candidate.node.peer_id, - candidate.node.addresses_by_priority(), - ) + .map(|candidate| WitnessedQuotePeer { + peer_id: candidate.node.peer_id, + addrs: candidate.node.addresses_by_priority(), + voters: candidate.voters, }) .collect()) } @@ -359,14 +385,129 @@ pub(crate) fn median_paid_quote_issuer( .map(|(_, peer_id, price)| (*peer_id, *price)) } +fn sort_quotes_by_distance(quotes: &mut [StoreQuote], address: &[u8; 32]) { + quotes.sort_by(|left, right| { + peer_xor_distance(&left.0, address) + .cmp(&peer_xor_distance(&right.0, address)) + .then_with(|| left.0.as_bytes().cmp(right.0.as_bytes())) + }); +} + +fn median_paid_quote_issuer_for_indices( + quotes: &[StoreQuote], + indices: &[usize], +) -> Option<(PeerId, Amount)> { + if indices.len() <= MEDIAN_QUOTE_INDEX { + return None; + } + + let mut by_price: Vec<(usize, PeerId, Amount)> = indices + .iter() + .enumerate() + .map(|(selected_index, quote_index)| { + let (peer_id, _, _, price) = "es[*quote_index]; + (selected_index, *peer_id, *price) + }) + .collect(); + by_price.sort_by_key(|(selected_index, _, price)| (*price, *selected_index)); + by_price + .get(MEDIAN_QUOTE_INDEX) + .map(|(_, peer_id, price)| (*peer_id, *price)) +} + +fn median_issuer_voter_support( + quotes: &[StoreQuote], + indices: &[usize], + voters_by_peer: &VotersByPeer, +) -> Option<(PeerId, usize)> { + let (median_peer_id, _) = median_paid_quote_issuer_for_indices(quotes, indices)?; + let voters = voters_by_peer.get(&median_peer_id)?; + let support = indices + .iter() + .filter(|quote_index| voters.contains("es[**quote_index].0)) + .count(); + Some((median_peer_id, support)) +} + +fn visit_quote_subsets( + quote_count: usize, + subset_size: usize, + start_index: usize, + current: &mut Vec, + visit: &mut F, +) where + F: FnMut(&[usize]), +{ + if current.len() == subset_size { + visit(current); + return; + } + + let remaining = subset_size - current.len(); + let last_start = quote_count - remaining; + for index in start_index..=last_start { + current.push(index); + visit_quote_subsets(quote_count, subset_size, index + 1, current, visit); + current.pop(); + } +} + +fn select_closest_quotes(mut quotes: Vec, address: &[u8; 32]) -> Vec { + sort_quotes_by_distance(&mut quotes, address); + quotes.truncate(CLOSE_GROUP_SIZE); + quotes +} + +fn select_witnessed_median_voter_quotes( + mut quotes: Vec, + address: &[u8; 32], + voters_by_peer: &VotersByPeer, +) -> Option> { + if quotes.len() < CLOSE_GROUP_SIZE { + return None; + } + + sort_quotes_by_distance(&mut quotes, address); + + let mut best_indices: Option> = None; + let mut current_indices = Vec::with_capacity(CLOSE_GROUP_SIZE); + visit_quote_subsets( + quotes.len(), + CLOSE_GROUP_SIZE, + 0, + &mut current_indices, + &mut |indices| { + let Some((_, support)) = median_issuer_voter_support("es, indices, voters_by_peer) + else { + return; + }; + if support < CLOSE_GROUP_MAJORITY { + return; + } + match &best_indices { + Some(best) if best.as_slice() <= indices => {} + _ => best_indices = Some(indices.to_vec()), + } + }, + ); + + best_indices.map(|indices| { + indices + .into_iter() + .map(|index| quotes[index].clone()) + .collect() + }) +} + impl Client { /// Get storage quotes from the closest peers for a given address. /// /// Builds a quorum-witnessed candidate set with at least /// `CLOSE_GROUP_SIZE` peers, requests quotes from all of them concurrently, - /// and returns the closest `CLOSE_GROUP_SIZE` successful responders sorted - /// by XOR distance. Farther quorum-recognised candidates are used only as - /// fallbacks when closer candidates do not return usable quotes. + /// and returns the closest supported `CLOSE_GROUP_SIZE` successful + /// responders sorted by XOR distance. Farther quorum-recognised candidates + /// are used only as fallbacks when needed to make the paid median issuer + /// locally acceptable to a close-group majority. /// /// Returns `Error::AlreadyStored` early if `CLOSE_GROUP_MAJORITY` peers /// report the chunk is already stored. @@ -380,9 +521,23 @@ impl Client { data_size: u64, data_type: u32, ) -> Result, PaymentQuote, Amount)>> { - let remote_peers = self.select_witnessed_quote_peers(address).await?; - self.collect_store_quotes_from_remote_peers(address, data_size, data_type, remote_peers) - .await + let witnessed_peers = self.select_witnessed_quote_peers(address).await?; + let voters_by_peer = witnessed_peers + .iter() + .map(|peer| (peer.peer_id, peer.voters.clone())) + .collect(); + let remote_peers = witnessed_peers + .into_iter() + .map(|peer| (peer.peer_id, peer.addrs)) + .collect(); + self.collect_store_quotes_from_remote_peers( + address, + data_size, + data_type, + remote_peers, + QuoteSelectionPolicy::WitnessedMedianVoters { voters_by_peer }, + ) + .await } /// Get storage quotes with the previous over-query behaviour. @@ -403,14 +558,20 @@ impl Client { .find_closest_peers(address, peer_query_count) .await?; - self.collect_store_quotes_from_remote_peers(address, data_size, data_type, remote_peers) - .await + self.collect_store_quotes_from_remote_peers( + address, + data_size, + data_type, + remote_peers, + QuoteSelectionPolicy::ClosestByDistance, + ) + .await } async fn select_witnessed_quote_peers( &self, address: &[u8; 32], - ) -> Result)>> { + ) -> Result> { let required = single_node_quote_query_count(); let quorum = witnessed_close_group_quorum(); let witnessed = self @@ -444,6 +605,7 @@ impl Client { data_size: u64, data_type: u32, remote_peers: Vec<(PeerId, Vec)>, + quote_selection_policy: QuoteSelectionPolicy, ) -> Result, PaymentQuote, Amount)>> { let peer_query_count = remote_peers.len(); @@ -630,22 +792,32 @@ impl Client { let total_responses = quote_count + failure_count + already_stored_count; if quotes.len() >= CLOSE_GROUP_SIZE { - // Sort by XOR distance to target, keep the closest CLOSE_GROUP_SIZE. - quotes.sort_by(|a, b| { - let dist_a = peer_xor_distance(&a.0, address); - let dist_b = peer_xor_distance(&b.0, address); - dist_a.cmp(&dist_b) - }); - quotes.truncate(CLOSE_GROUP_SIZE); + let selected_quotes = match quote_selection_policy { + QuoteSelectionPolicy::ClosestByDistance => select_closest_quotes(quotes, address), + QuoteSelectionPolicy::WitnessedMedianVoters { voters_by_peer } => { + select_witnessed_median_voter_quotes(quotes, address, &voters_by_peer) + .ok_or_else(|| { + Error::InsufficientPeers(format!( + "Got {quote_count} quotes, need {CLOSE_GROUP_SIZE} whose paid \ + median issuer is recognised by at least {CLOSE_GROUP_MAJORITY} \ + selected witness peers ({total_responses} responses: \ + {already_stored_count} already_stored, {failure_count} failed \ + including {bad_quote_count} with mismatched peer bindings). \ + Failures: [{}]", + failures.join("; ") + )) + })? + } + }; info!( "Collected {} quotes for address {} ({total_responses} responses: \ {quote_count} ok, {already_stored_count} already_stored, {failure_count} failed, \ {bad_quote_count} bad-binding)", - quotes.len(), + selected_quotes.len(), hex::encode(address), ); - return Ok(quotes); + return Ok(selected_quotes); } Err(Error::InsufficientPeers(format!( @@ -751,6 +923,34 @@ mod tests { } } + fn synthetic_peer(seed: u8) -> PeerId { + PeerId::from_bytes([seed; 32]) + } + + fn synthetic_quote(seed: u8, price: u64) -> (PeerId, Vec, PaymentQuote, Amount) { + let amount = Amount::from(price); + let quote = PaymentQuote { + content: XorName([0u8; 32]), + timestamp: SystemTime::UNIX_EPOCH, + price: amount, + rewards_address: RewardsAddress::new([0u8; 20]), + pub_key: Vec::new(), + signature: Vec::new(), + }; + (synthetic_peer(seed), Vec::new(), quote, amount) + } + + fn synthetic_voters(seeds: &[u8]) -> HashSet { + seeds.iter().copied().map(synthetic_peer).collect() + } + + fn quote_peer_seeds(quotes: &[(PeerId, Vec, PaymentQuote, Amount)]) -> Vec { + quotes + .iter() + .map(|(peer_id, _, _, _)| peer_id.as_bytes()[0]) + .collect() + } + /// Independent re-implementation of the storer-side binding spec /// (`ant-node/src/payment/verifier.rs::validate_peer_bindings` + /// `peer_id_from_public_key_bytes`): @@ -906,12 +1106,88 @@ mod tests { assert_eq!( peers .iter() - .map(|(peer_id, _)| peer_id.as_bytes()[0]) + .map(|peer| peer.peer_id.as_bytes()[0]) .collect::>(), vec![1, 2, 3, 4, 5, 6, 7, 8] ); } + #[test] + fn witnessed_quote_selection_keeps_closest_set_with_median_voter_majority() { + const MEDIAN_ISSUER_SEED: u8 = 7; + const FAR_SUPPORTING_VOTER_SEED: u8 = 20; + const UNSUCCESSFUL_SUPPORTING_VOTER_SEED: u8 = 21; + + let address = [0u8; 32]; + let quotes = vec![ + synthetic_quote(1, 10), + synthetic_quote(2, 20), + synthetic_quote(3, 30), + synthetic_quote(6, 50), + synthetic_quote(MEDIAN_ISSUER_SEED, 40), + synthetic_quote(8, 60), + synthetic_quote(9, 70), + synthetic_quote(FAR_SUPPORTING_VOTER_SEED, 80), + ]; + let mut voters_by_peer = HashMap::new(); + voters_by_peer.insert( + synthetic_peer(MEDIAN_ISSUER_SEED), + synthetic_voters(&[ + 1, + 2, + 3, + FAR_SUPPORTING_VOTER_SEED, + UNSUCCESSFUL_SUPPORTING_VOTER_SEED, + ]), + ); + + let selected = select_witnessed_median_voter_quotes(quotes, &address, &voters_by_peer) + .expect("a supported close-group quote set should be selected"); + + assert_eq!(quote_peer_seeds(&selected), vec![1, 2, 3, 6, 7, 8, 20]); + let (median_peer_id, _) = + median_paid_quote_issuer(&selected).expect("selected quotes have a median"); + assert_eq!(median_peer_id, synthetic_peer(MEDIAN_ISSUER_SEED)); + let selected_peers = selected + .iter() + .map(|(peer_id, _, _, _)| *peer_id) + .collect::>(); + let support = voters_by_peer[&median_peer_id] + .intersection(&selected_peers) + .count(); + assert_eq!(support, CLOSE_GROUP_MAJORITY); + } + + #[test] + fn witnessed_quote_selection_rejects_median_without_selected_voter_majority() { + const MEDIAN_ISSUER_SEED: u8 = 7; + + let address = [0u8; 32]; + let quotes = vec![ + synthetic_quote(1, 10), + synthetic_quote(2, 20), + synthetic_quote(3, 30), + synthetic_quote(6, 50), + synthetic_quote(MEDIAN_ISSUER_SEED, 40), + synthetic_quote(8, 60), + synthetic_quote(9, 70), + synthetic_quote(10, 80), + ]; + let mut voters_by_peer = HashMap::new(); + voters_by_peer.insert( + synthetic_peer(MEDIAN_ISSUER_SEED), + synthetic_voters(&[1, 2, 3, 20, 21]), + ); + + let selected = select_witnessed_median_voter_quotes(quotes, &address, &voters_by_peer); + + assert!( + selected.is_none(), + "the selector must not return a paid quote set when fewer than \ + CLOSE_GROUP_MAJORITY supporting witness peers produced usable quotes" + ); + } + #[test] fn filter_drops_only_bad_bindings_and_leaves_storer_acceptable_quotes() { let mut quotes = vec![ From 10e75c4a8218da5cbce8d25bddefa80d9ce3a1a7 Mon Sep 17 00:00:00 2001 From: Warm Beer Date: Mon, 15 Jun 2026 14:51:32 +0200 Subject: [PATCH 3/3] fix(client): stabilise witnessed quote PUT targets Keep proof quote order stable while ordering PUT targets so the initial store wave favours peers that voted for the paid median issuer. Wire the in-process E2E protocol through AntProtocol::attach_p2p_node and use ant-node's test-only paid close-group override for the local client/storage-node topology. SemVer: bug fix; no public ant-client API break expected. --- ant-core/Cargo.toml | 2 +- ant-core/src/data/client/batch.rs | 12 +-- ant-core/src/data/client/payment.rs | 10 +- ant-core/src/data/client/quote.rs | 157 +++++++++++++++++++++++----- ant-core/tests/support/mod.rs | 41 ++++++-- 5 files changed, 176 insertions(+), 46 deletions(-) diff --git a/ant-core/Cargo.toml b/ant-core/Cargo.toml index d9532c93..c8cd7a0a 100644 --- a/ant-core/Cargo.toml +++ b/ant-core/Cargo.toml @@ -93,7 +93,7 @@ devnet = ["dep:ant-node"] # always compile even without the `devnet` feature. Pinned to the same # version as the runtime dep so there is a single ant-node / # saorsa-core version across the whole graph. -ant-node = { git = "https://github.com/WithAutonomi/ant-node", branch = "feat/witnessed-transcript-policy" } +ant-node = { git = "https://github.com/WithAutonomi/ant-node", branch = "feat/witnessed-transcript-policy", features = ["test-utils"] } serial_test = "3" anyhow = "1" alloy = { version = "1.6", features = ["node-bindings"] } diff --git a/ant-core/src/data/client/batch.rs b/ant-core/src/data/client/batch.rs index 09c3b4b8..3470b80d 100644 --- a/ant-core/src/data/client/batch.rs +++ b/ant-core/src/data/client/batch.rs @@ -243,23 +243,21 @@ impl Client { let data_size = u64::try_from(content.len()) .map_err(|e| Error::InvalidData(format!("content size too large: {e}")))?; - let quotes_with_peers = match self - .get_store_quotes(&address, data_size, DATA_TYPE_CHUNK) + let quote_plan = match self + .get_store_quote_plan(&address, data_size, DATA_TYPE_CHUNK) .await { - Ok(quotes) => quotes, + Ok(plan) => plan, Err(Error::AlreadyStored) => { debug!("Chunk {} already stored, skipping", hex::encode(address)); return Ok(None); } Err(e) => return Err(e), }; + let quotes_with_peers = quote_plan.quotes; // Capture all quoted peers for close-group replication. - let quoted_peers: Vec<(PeerId, Vec)> = quotes_with_peers - .iter() - .map(|(peer_id, addrs, _, _)| (*peer_id, addrs.clone())) - .collect(); + let quoted_peers = quote_plan.put_peers; // Build peer_quotes for ProofOfPayment + quotes for SingleNodePayment. // Use node-reported prices directly — no contract price fetch needed. diff --git a/ant-core/src/data/client/payment.rs b/ant-core/src/data/client/payment.rs index 6f734e04..3452d599 100644 --- a/ant-core/src/data/client/payment.rs +++ b/ant-core/src/data/client/payment.rs @@ -48,7 +48,10 @@ impl Client { debug!("Collecting quotes for address {}", hex::encode(address)); // 1. Collect quotes from network - let quotes_with_peers = self.get_store_quotes(address, data_size, data_type).await?; + let quote_plan = self + .get_store_quote_plan(address, data_size, data_type) + .await?; + let quotes_with_peers = quote_plan.quotes; let median_quote_issuer = median_paid_quote_issuer("es_with_peers).ok_or_else(|| { Error::Payment( @@ -57,10 +60,7 @@ impl Client { })?; // Capture all quoted peers for replication by the caller. - let quoted_peers: Vec<(PeerId, Vec)> = quotes_with_peers - .iter() - .map(|(peer_id, addrs, _, _)| (*peer_id, addrs.clone())) - .collect(); + let quoted_peers = quote_plan.put_peers; // 2. Build peer_quotes for ProofOfPayment + quotes for SingleNodePayment. // Use node-reported prices directly — no contract price fetch needed. diff --git a/ant-core/src/data/client/quote.rs b/ant-core/src/data/client/quote.rs index c7ef959d..c5771427 100644 --- a/ant-core/src/data/client/quote.rs +++ b/ant-core/src/data/client/quote.rs @@ -167,14 +167,23 @@ fn witnessed_close_group_quorum() -> usize { (CLOSE_GROUP_SIZE * WITNESSED_QUORUM_NUMERATOR).div_ceil(WITNESSED_QUORUM_DENOMINATOR) } +fn witnessed_median_voter_quorum() -> usize { + witnessed_close_group_quorum() +} + fn peer_list(peers: &[PeerId]) -> Vec { peers.iter().map(ToString::to_string).collect() } -type StoreQuote = (PeerId, Vec, PaymentQuote, Amount); +pub(crate) type StoreQuote = (PeerId, Vec, PaymentQuote, Amount); type VotersByPeer = HashMap>; type WitnessedVoteData = (HashMap, VotersByPeer, Vec<(PeerId, usize)>); +pub(crate) struct StoreQuotePlan { + pub(crate) quotes: Vec, + pub(crate) put_peers: Vec<(PeerId, Vec)>, +} + #[derive(Debug, Clone)] struct WitnessedQuoteCandidate { node: DHTNode, @@ -469,8 +478,9 @@ fn select_witnessed_median_voter_quotes( sort_quotes_by_distance(&mut quotes, address); - let mut best_indices: Option> = None; + let mut best_indices: Option<(usize, Vec)> = None; let mut current_indices = Vec::with_capacity(CLOSE_GROUP_SIZE); + let required_support = witnessed_median_voter_quorum(); visit_quote_subsets( quotes.len(), CLOSE_GROUP_SIZE, @@ -481,17 +491,19 @@ fn select_witnessed_median_voter_quotes( else { return; }; - if support < CLOSE_GROUP_MAJORITY { + if support < required_support { return; } match &best_indices { - Some(best) if best.as_slice() <= indices => {} - _ => best_indices = Some(indices.to_vec()), + Some((best_support, best)) if *best_support > support => {} + Some((best_support, best)) + if *best_support == support && best.as_slice() <= indices => {} + _ => best_indices = Some((support, indices.to_vec())), } }, ); - best_indices.map(|indices| { + best_indices.map(|(_, indices)| { indices .into_iter() .map(|index| quotes[index].clone()) @@ -499,15 +511,41 @@ fn select_witnessed_median_voter_quotes( }) } +fn put_peers_with_median_voters_first( + quotes: &[StoreQuote], + voters_by_peer: &VotersByPeer, +) -> Option)>> { + let (median_peer_id, _) = median_paid_quote_issuer(quotes)?; + let voters = voters_by_peer.get(&median_peer_id)?; + + let mut supporting_peers = Vec::new(); + let mut fallback_peers = Vec::new(); + for (peer_id, addrs, _, _) in quotes { + let peer = (*peer_id, addrs.clone()); + if voters.contains(peer_id) { + supporting_peers.push(peer); + } else { + fallback_peers.push(peer); + } + } + + if supporting_peers.len() < witnessed_median_voter_quorum() { + return None; + } + + supporting_peers.extend(fallback_peers); + Some(supporting_peers) +} + impl Client { /// Get storage quotes from the closest peers for a given address. /// /// Builds a quorum-witnessed candidate set with at least /// `CLOSE_GROUP_SIZE` peers, requests quotes from all of them concurrently, /// and returns the closest supported `CLOSE_GROUP_SIZE` successful - /// responders sorted by XOR distance. Farther quorum-recognised candidates - /// are used only as fallbacks when needed to make the paid median issuer - /// locally acceptable to a close-group majority. + /// responders. When multiple sets are possible, the client prefers the + /// one with the strongest paid-median voter support, then the closest + /// peers by XOR distance. /// /// Returns `Error::AlreadyStored` early if `CLOSE_GROUP_MAJORITY` peers /// report the chunk is already stored. @@ -521,8 +559,26 @@ impl Client { data_size: u64, data_type: u32, ) -> Result, PaymentQuote, Amount)>> { + Ok(self + .get_store_quote_plan(address, data_size, data_type) + .await? + .quotes) + } + + /// Get storage quotes plus PUT targets ordered for paid-median acceptance. + /// + /// Quote order is preserved for proof construction because tied quote + /// prices rely on stable median selection. PUT target order is separate: + /// peers that voted for the paid median issuer are placed first so the + /// initial write wave is locally acceptable to a storage majority. + pub(crate) async fn get_store_quote_plan( + &self, + address: &[u8; 32], + data_size: u64, + data_type: u32, + ) -> Result { let witnessed_peers = self.select_witnessed_quote_peers(address).await?; - let voters_by_peer = witnessed_peers + let voters_by_peer: VotersByPeer = witnessed_peers .iter() .map(|peer| (peer.peer_id, peer.voters.clone())) .collect(); @@ -530,14 +586,29 @@ impl Client { .into_iter() .map(|peer| (peer.peer_id, peer.addrs)) .collect(); - self.collect_store_quotes_from_remote_peers( - address, - data_size, - data_type, - remote_peers, - QuoteSelectionPolicy::WitnessedMedianVoters { voters_by_peer }, - ) - .await + let quotes = self + .collect_store_quotes_from_remote_peers( + address, + data_size, + data_type, + remote_peers, + QuoteSelectionPolicy::WitnessedMedianVoters { + voters_by_peer: voters_by_peer.clone(), + }, + ) + .await?; + let put_peers = + put_peers_with_median_voters_first("es, &voters_by_peer).ok_or_else(|| { + Error::InsufficientPeers(format!( + "Collected {} witnessed quotes, but fewer than {} \ + selected PUT peers voted for the paid median issuer for {}", + quotes.len(), + witnessed_median_voter_quorum(), + hex::encode(address) + )) + })?; + + Ok(StoreQuotePlan { quotes, put_peers }) } /// Get storage quotes with the previous over-query behaviour. @@ -799,11 +870,12 @@ impl Client { .ok_or_else(|| { Error::InsufficientPeers(format!( "Got {quote_count} quotes, need {CLOSE_GROUP_SIZE} whose paid \ - median issuer is recognised by at least {CLOSE_GROUP_MAJORITY} \ + median issuer is recognised by at least {} \ selected witness peers ({total_responses} responses: \ {already_stored_count} already_stored, {failure_count} failed \ including {bad_quote_count} with mismatched peer bindings). \ Failures: [{}]", + witnessed_median_voter_quorum(), failures.join("; ") )) })? @@ -951,6 +1023,13 @@ mod tests { .collect() } + fn put_peer_seeds(peers: &[(PeerId, Vec)]) -> Vec { + peers + .iter() + .map(|(peer_id, _)| peer_id.as_bytes()[0]) + .collect() + } + /// Independent re-implementation of the storer-side binding spec /// (`ant-node/src/payment/verifier.rs::validate_peer_bindings` + /// `peer_id_from_public_key_bytes`): @@ -1113,7 +1192,7 @@ mod tests { } #[test] - fn witnessed_quote_selection_keeps_closest_set_with_median_voter_majority() { + fn witnessed_quote_selection_keeps_closest_set_with_median_voter_quorum() { const MEDIAN_ISSUER_SEED: u8 = 7; const FAR_SUPPORTING_VOTER_SEED: u8 = 20; const UNSUCCESSFUL_SUPPORTING_VOTER_SEED: u8 = 21; @@ -1136,6 +1215,7 @@ mod tests { 1, 2, 3, + MEDIAN_ISSUER_SEED, FAR_SUPPORTING_VOTER_SEED, UNSUCCESSFUL_SUPPORTING_VOTER_SEED, ]), @@ -1155,11 +1235,11 @@ mod tests { let support = voters_by_peer[&median_peer_id] .intersection(&selected_peers) .count(); - assert_eq!(support, CLOSE_GROUP_MAJORITY); + assert_eq!(support, witnessed_median_voter_quorum()); } #[test] - fn witnessed_quote_selection_rejects_median_without_selected_voter_majority() { + fn witnessed_quote_selection_rejects_median_without_selected_voter_quorum() { const MEDIAN_ISSUER_SEED: u8 = 7; let address = [0u8; 32]; @@ -1183,11 +1263,40 @@ mod tests { assert!( selected.is_none(), - "the selector must not return a paid quote set when fewer than \ - CLOSE_GROUP_MAJORITY supporting witness peers produced usable quotes" + "the selector must not return a paid quote set when fewer than the \ + witnessed median voter quorum produced usable quotes" ); } + #[test] + fn put_peers_prioritise_median_voters_without_reordering_quotes() { + const MEDIAN_ISSUER_SEED: u8 = 7; + + let quotes = vec![ + synthetic_quote(1, 10), + synthetic_quote(2, 20), + synthetic_quote(3, 30), + synthetic_quote(4, 50), + synthetic_quote(5, 60), + synthetic_quote(6, 70), + synthetic_quote(MEDIAN_ISSUER_SEED, 40), + ]; + let mut voters_by_peer = HashMap::new(); + voters_by_peer.insert( + synthetic_peer(MEDIAN_ISSUER_SEED), + synthetic_voters(&[3, 4, 5, 6, MEDIAN_ISSUER_SEED]), + ); + + let put_peers = put_peers_with_median_voters_first("es, &voters_by_peer) + .expect("median voters should produce an ordered PUT set"); + + assert_eq!(quote_peer_seeds("es), vec![1, 2, 3, 4, 5, 6, 7]); + let (median_peer_id, _) = + median_paid_quote_issuer("es).expect("selected quotes have a median"); + assert_eq!(median_peer_id, synthetic_peer(MEDIAN_ISSUER_SEED)); + assert_eq!(put_peer_seeds(&put_peers), vec![3, 4, 5, 6, 7, 1, 2]); + } + #[test] fn filter_drops_only_bad_bindings_and_leaves_storer_acceptable_quotes() { let mut quotes = vec![ diff --git a/ant-core/tests/support/mod.rs b/ant-core/tests/support/mod.rs index 3731843a..4a168c81 100644 --- a/ant-core/tests/support/mod.rs +++ b/ant-core/tests/support/mod.rs @@ -27,6 +27,7 @@ use ant_node::storage::{AntProtocol, LmdbStorage, LmdbStorageConfig}; // Wire / transport / EVM types: route through ant-protocol so the test // harness exercises the same surface the client does. use ant_protocol::evm::{testnet::Testnet, Network as EvmNetwork, RewardsAddress, Wallet}; +use ant_protocol::pqc::ops::{MlDsaOperations, MlDsaSecretKey}; use ant_protocol::transport::{ CoreNodeConfig, IPDiversityConfig, MlDsa65, MultiAddr, NodeIdentity, P2PEvent, P2PNode, }; @@ -197,6 +198,32 @@ impl MiniTestnet { sleep(Duration::from_millis(500)).await; } + // The in-process E2E harness builds clients from one of the storage + // nodes. Saorsa's witnessed client lookup filters that local peer out, + // while node-side payment verification uses a self-inclusive local + // close-group view. In tiny random testnets that can make an otherwise + // valid paid median issuer fall just outside a storer's local top-7. + // Keep the production live-DHT check in normal builds, but use + // ant-node's test-only override here so these client E2Es exercise + // payment/proof/storage behaviour without depending on that topology + // artifact. + let paid_quote_close_group_override: Vec<[u8; 32]> = nodes + .iter() + .filter_map(|test_node| { + test_node + .p2p_node + .as_ref() + .map(|p2p_node| *p2p_node.peer_id().as_bytes()) + }) + .collect(); + for test_node in &nodes { + if let Some(protocol) = &test_node.protocol { + protocol + .payment_verifier_arc() + .set_paid_quote_close_group_for_tests(paid_quote_close_group_override.clone()); + } + } + // Approve token spend for the unified payment vault contract let vault_address = evm_network.payment_vault_address(); wallet @@ -290,22 +317,15 @@ impl MiniTestnet { local_rewards_address: rewards_address, }; let payment_verifier = Arc::new(PaymentVerifier::new(payment_config)); - // Wire the P2P node into the verifier so the merkle pay-yourself - // closeness check can do its DHT lookup. Without this, the - // verifier fail-closes on every merkle payment (PR #77 defense). - payment_verifier.attach_p2p_node(Arc::clone(&node)); let metrics_tracker = QuotingMetricsTracker::new(TEST_MAX_RECORDS); let mut quote_generator = QuoteGenerator::new(rewards_address, metrics_tracker); // Wire ML-DSA-65 signing so quotes are properly signed and verifiable let pub_key_bytes = identity.public_key().as_bytes().to_vec(); let sk_bytes = identity.secret_key_bytes().to_vec(); - let sk = { - use ant_protocol::pqc::ops::MlDsaSecretKey; - MlDsaSecretKey::from_bytes(&sk_bytes).expect("deserialize ML-DSA-65 secret key") - }; + let sk = + { MlDsaSecretKey::from_bytes(&sk_bytes).expect("deserialize ML-DSA-65 secret key") }; quote_generator.set_signer(pub_key_bytes, move |msg| { - use ant_protocol::pqc::ops::MlDsaOperations; let ml_dsa = MlDsa65::new(); ml_dsa .sign(&sk, msg) @@ -318,6 +338,9 @@ impl MiniTestnet { payment_verifier, Arc::new(quote_generator), )); + // Wire the P2P node into the protocol so direct PUT storage-admission + // and payment closeness checks use the node's live DHT view. + protocol.attach_p2p_node(Arc::clone(&node)); // Start message handler loop let handler_node = Arc::clone(&node);