1use crate::data::client::peer_xor_distance;
7use crate::data::client::Client;
8use crate::data::client::SettlementRefusals;
9use crate::data::client::PUT_TARGET_WIDTH;
10use crate::data::client::VERSIONED_QUOTE_PROBE_CEILING;
11use crate::data::error::{Error, Result};
12use ant_protocol::evm::{Amount, PaymentQuote};
13use ant_protocol::payment::calculate_price;
14use ant_protocol::payment::commitment::{
15 commitment_hash, verify_commitment_signature, StorageCommitment, MAX_COMMITMENT_KEY_COUNT,
16 MAX_COMMITMENT_SIDECAR_BYTES,
17};
18use ant_protocol::payment::{verify_quote_content, verify_quote_signature};
19use ant_protocol::transport::{
20 DHTNode, MultiAddr, P2PNode, PeerId, ResponderView, WitnessedCloseGroup,
21};
22use ant_protocol::{
23 client_update_required_message, compute_address, send_and_await_chunk_response, ChunkMessage,
24 ChunkMessageBody, ChunkQuoteRequest, ChunkQuoteRequestV2, ChunkQuoteResponse, ProtocolError,
25 CLOSE_GROUP_MAJORITY, CLOSE_GROUP_SIZE, CURRENT_SETTLEMENT_VERSION,
26};
27use futures::stream::{FuturesUnordered, StreamExt};
28use std::collections::{HashMap, HashSet};
29use std::sync::{Arc, Mutex};
30use std::time::Duration;
31use tracing::{debug, info, warn};
32
33const FAULT_TOLERANT_QUOTE_QUERY_MULTIPLIER: usize = 2;
38
39const WITNESSED_QUORUM_NUMERATOR: usize = 2;
43const WITNESSED_QUORUM_DENOMINATOR: usize = 3;
44
45const SINGLE_NODE_WITNESSED_VIEW_COUNT: usize = 20;
47
48const SINGLE_NODE_MIN_QUOTE_COUNT: usize = 1;
50
51const QUOTE_COLLECTION_TIMEOUT_SECS: u64 = 120;
55
56const ML_DSA_PUB_KEY_LEN: usize = 1952;
65
66type QuotedPeer = (
71 PeerId,
72 Vec<MultiAddr>,
73 PaymentQuote,
74 Amount,
75 Option<Vec<u8>>,
76);
77
78fn quote_binding_is_valid(peer_id: &PeerId, quote: &PaymentQuote) -> bool {
93 if quote.pub_key.len() != ML_DSA_PUB_KEY_LEN {
94 return false;
95 }
96 compute_address("e.pub_key) == *peer_id.as_bytes()
97}
98
99fn quote_commitment_binding_is_valid(
125 peer_id: &PeerId,
126 quote: &PaymentQuote,
127 commitment: &Option<Vec<u8>>,
128) -> std::result::Result<(), String> {
129 let count = quote.committed_key_count;
130 let pin = quote.commitment_pin;
131 match (count, pin.is_some()) {
132 (0, false) | (1.., true) => {}
133 (1.., false) => {
134 return Err(format!(
135 "committed_key_count={count} > 0 but commitment_pin is None (unauditable count)"
136 ));
137 }
138 (0, true) => {
139 return Err("committed_key_count=0 with a commitment_pin (incoherent baseline)".into());
140 }
141 }
142 if count > MAX_COMMITMENT_KEY_COUNT {
143 return Err(format!(
144 "committed_key_count={count} exceeds MAX_COMMITMENT_KEY_COUNT={MAX_COMMITMENT_KEY_COUNT}"
145 ));
146 }
147 let expected = calculate_price(count as usize);
149 if quote.price != expected {
150 return Err(format!(
151 "price {} does not equal calculate_price(committed_key_count={count}) = {expected}",
152 quote.price
153 ));
154 }
155
156 let Some(pin) = pin else {
158 return Ok(());
159 };
160
161 let Some(blob) = commitment else {
163 return Err(
164 "bound quote did not ship its commitment; the pin is unresolvable so the quote \
165 is dropped before payment"
166 .into(),
167 );
168 };
169 if blob.len() > MAX_COMMITMENT_SIDECAR_BYTES {
172 return Err(format!(
173 "shipped commitment is {} bytes, exceeds MAX_COMMITMENT_SIDECAR_BYTES={MAX_COMMITMENT_SIDECAR_BYTES}",
174 blob.len()
175 ));
176 }
177 let commitment: StorageCommitment = rmp_serde::from_slice(blob).map_err(|e| {
178 format!("shipped commitment did not deserialize as a StorageCommitment: {e}")
179 })?;
180
181 if compute_address(&commitment.sender_public_key) != *peer_id.as_bytes()
184 || commitment.sender_peer_id != *peer_id.as_bytes()
185 {
186 return Err("shipped commitment is not bound to the quoting peer".into());
187 }
188 if !verify_commitment_signature(&commitment) {
189 return Err("shipped commitment has an invalid signature".into());
190 }
191 if commitment_hash(&commitment) != Some(pin) {
192 return Err("shipped commitment does not hash to the quote's pin".into());
193 }
194 if commitment.key_count != count {
195 return Err(format!(
196 "shipped commitment attests key_count={} but the quote claims {count}",
197 commitment.key_count
198 ));
199 }
200 Ok(())
201}
202
203type ClassifiedQuote = std::result::Result<(PaymentQuote, Amount, Option<Vec<u8>>), Error>;
239fn classify_quote_response(
240 peer_id: &PeerId,
241 expected_content: &[u8; 32],
242 quote_bytes: &[u8],
243 already_stored: bool,
244 commitment: Option<Vec<u8>>,
245) -> ClassifiedQuote {
246 let payment_quote = rmp_serde::from_slice::<PaymentQuote>(quote_bytes).map_err(|e| {
247 Error::Serialization(format!("Failed to deserialize quote from {peer_id}: {e}"))
248 })?;
249
250 if !quote_binding_is_valid(peer_id, &payment_quote) {
255 let derived = compute_address(&payment_quote.pub_key);
256 warn!(
257 "Dropping response from {peer_id} — quote.pub_key BLAKE3 mismatch \
258 (peer is signing quotes with another peer's key); the storer \
259 would reject this proof"
260 );
261 return Err(Error::BadQuoteBinding {
262 peer_id: peer_id.to_string(),
263 detail: format!(
264 "BLAKE3(pub_key)={} pub_key_len={}",
265 hex::encode(derived),
266 payment_quote.pub_key.len(),
267 ),
268 });
269 }
270
271 if !verify_quote_content(&payment_quote, expected_content) {
277 return Err(Error::BadQuoteBinding {
278 peer_id: peer_id.to_string(),
279 detail: "quote content does not match the requested address".to_string(),
280 });
281 }
282 if !verify_quote_signature(&payment_quote) {
283 return Err(Error::BadQuoteBinding {
284 peer_id: peer_id.to_string(),
285 detail: "quote ML-DSA-65 signature is invalid".to_string(),
286 });
287 }
288
289 if let Err(detail) = quote_commitment_binding_is_valid(peer_id, &payment_quote, &commitment) {
295 warn!("Dropping response from {peer_id} — ADR-0004 binding invalid: {detail}");
296 return Err(Error::BadQuoteCommitment {
297 peer_id: peer_id.to_string(),
298 detail,
299 });
300 }
301
302 if already_stored {
303 debug!("Peer {peer_id} already has chunk");
304 return Err(Error::AlreadyStored);
305 }
306 let price = payment_quote.price;
307 debug!("Received quote from {peer_id}: price = {price}");
308 Ok((payment_quote, price, commitment))
309}
310
311fn drop_quotes_with_bad_bindings(quotes: &mut Vec<QuotedPeer>) -> usize {
314 let before = quotes.len();
315 quotes.retain(|(peer_id, _, quote, _, _)| {
316 if quote_binding_is_valid(peer_id, quote) {
317 true
318 } else {
319 warn!(
320 "Dropping quote from peer {peer_id} — quote.pub_key BLAKE3 mismatch \
321 (peer is signing quotes with another peer's key); the storer would \
322 reject this proof"
323 );
324 false
325 }
326 });
327 before - quotes.len()
328}
329
330#[allow(clippy::too_many_arguments)]
331async fn request_store_quote_from_peer(
332 node: Arc<P2PNode>,
333 peer_id: PeerId,
334 peer_addrs: Vec<MultiAddr>,
335 request_id: u64,
336 address: [u8; 32],
337 data_size: u64,
338 data_type: u32,
339 per_peer_timeout: Duration,
340 unversioned_peers: Arc<Mutex<HashSet<PeerId>>>,
341 versioned_capable: Arc<Mutex<HashSet<PeerId>>>,
342) -> StoreQuoteRequestResult {
343 let legacy_request = ChunkQuoteRequest {
344 address,
345 data_size,
346 data_type,
347 };
348
349 let known_legacy = !versioned_capable
359 .lock()
360 .is_ok_and(|peers| peers.contains(&peer_id))
361 && unversioned_peers
362 .lock()
363 .is_ok_and(|peers| peers.contains(&peer_id));
364
365 let body = if known_legacy {
369 ChunkMessageBody::QuoteRequest(legacy_request.clone())
370 } else {
371 let mut versioned_request = ChunkQuoteRequestV2::new(address, data_size);
372 versioned_request.data_type = data_type;
373 ChunkMessageBody::QuoteRequestV2(versioned_request)
374 };
375 let message = ChunkMessage { request_id, body };
376
377 let message_bytes = match message.encode() {
378 Ok(bytes) => bytes,
379 Err(e) => {
380 return (
381 peer_id,
382 peer_addrs,
383 Err(Error::Protocol(format!(
384 "Failed to encode quote request for {peer_id}: {e}"
385 ))),
386 );
387 }
388 };
389
390 let attempt_timeout = if known_legacy {
394 per_peer_timeout
395 } else {
396 per_peer_timeout.min(VERSIONED_QUOTE_PROBE_CEILING)
397 };
398
399 let result = send_and_await_chunk_response(
400 &node,
401 &peer_id,
402 message_bytes,
403 request_id,
404 attempt_timeout,
405 &peer_addrs,
406 |body| map_quote_response(&peer_id, &address, body),
407 |e| Error::Network(format!("Failed to send quote request to {peer_id}: {e}")),
408 || Error::Timeout(format!("Timeout waiting for quote from {peer_id}")),
409 )
410 .await;
411
412 let answered = match &result {
416 Ok(_) => true,
417 Err(e) => !is_version_unaware(e),
418 };
419 if !known_legacy && answered {
420 if let Ok(mut peers) = versioned_capable.lock() {
421 peers.insert(peer_id);
422 }
423 }
424
425 let result = match result {
429 Err(ref e) if is_version_unaware(e) && !known_legacy => {
430 let ever_answered = versioned_capable
439 .lock()
440 .is_ok_and(|peers| peers.contains(&peer_id));
441 if matches!(e, Error::Timeout(_)) && !ever_answered {
442 if let Ok(mut peers) = unversioned_peers.lock() {
443 peers.insert(peer_id);
444 }
445 }
446 let legacy = ChunkMessage {
447 request_id,
448 body: ChunkMessageBody::QuoteRequest(legacy_request),
449 };
450 match legacy.encode() {
451 Ok(legacy_bytes) => {
452 send_and_await_chunk_response(
453 &node,
454 &peer_id,
455 legacy_bytes,
456 request_id,
457 per_peer_timeout,
458 &peer_addrs,
459 |body| map_quote_response(&peer_id, &address, body),
460 |e| {
461 Error::Network(format!(
462 "Failed to send quote request to {peer_id}: {e}"
463 ))
464 },
465 || Error::Timeout(format!("Timeout waiting for quote from {peer_id}")),
466 )
467 .await
468 }
469 Err(e) => Err(Error::Protocol(format!(
470 "Failed to encode quote request for {peer_id}: {e}"
471 ))),
472 }
473 }
474 other => other,
475 };
476
477 (peer_id, peer_addrs, result)
478}
479
480fn map_quote_response(
487 peer_id: &PeerId,
488 address: &[u8; 32],
489 body: ChunkMessageBody,
490) -> Option<ClassifiedQuote> {
491 match body {
492 ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Success {
493 quote,
494 already_stored,
495 commitment,
496 }) => Some(classify_quote_response(
497 peer_id,
498 address,
499 "e,
500 already_stored,
501 commitment,
502 )),
503 ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error(
504 ProtocolError::ClientUpdateRequired {
505 client_settlement_version,
506 min_settlement_version,
507 },
508 )) => Some(Err(settlement_refusal_error(
509 peer_id,
510 client_settlement_version,
511 min_settlement_version,
512 ))),
513 ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error(
514 behind @ ProtocolError::StorerUpdateRequired { .. },
515 )) => Some(Err(Error::StorerUpdateRequired(behind.to_string()))),
516 ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error(e)) => Some(Err(
517 Error::Protocol(format!("Quote error from {peer_id}: {e}")),
518 )),
519 _ => None,
520 }
521}
522
523pub(super) fn settlement_refusal_error(
532 peer_id: &PeerId,
533 client_settlement_version: u32,
534 min_settlement_version: u32,
535) -> Error {
536 if client_settlement_version != CURRENT_SETTLEMENT_VERSION
537 || min_settlement_version <= client_settlement_version
538 {
539 return Error::Protocol(format!(
540 "Peer {peer_id} sent an incoherent settlement refusal (claimed this client is at \
541 version {client_settlement_version} needing {min_settlement_version}, but this \
542 client is at {CURRENT_SETTLEMENT_VERSION}); ignoring it"
543 ));
544 }
545 Error::ClientUpdateRequired(client_update_required_message(
546 client_settlement_version,
547 min_settlement_version,
548 ))
549}
550
551const _: () = crate::data::client::UNVERSIONED_RETRY_REQUIRES_MIN_V1;
560
561const fn is_version_unaware(error: &Error) -> bool {
562 matches!(error, Error::Network(_) | Error::Timeout(_))
563}
564
565#[allow(clippy::too_many_arguments)]
571fn record_store_quote_result(
572 peer_id: PeerId,
573 addrs: Vec<MultiAddr>,
574 quote_result: Result<(PaymentQuote, Amount, Option<Vec<u8>>)>,
575 address: &[u8; 32],
576 quotes: &mut Vec<StoreQuote>,
577 already_stored_peers: &mut Vec<(PeerId, [u8; 32])>,
578 failures: &mut Vec<String>,
579 bad_quote_count: &mut usize,
580 settlement_refusal: &mut Option<Error>,
581 refusals: &SettlementRefusals,
582) -> Result<()> {
583 match quote_result {
584 Ok((quote, price, commitment)) => {
585 quotes.push((peer_id, addrs, quote, price, commitment));
586 }
587 Err(Error::AlreadyStored) => {
588 info!("Peer {peer_id} reports chunk already stored");
589 let dist = peer_xor_distance(&peer_id, address);
590 already_stored_peers.push((peer_id, dist));
591 }
592 Err(e @ Error::ClientUpdateRequired(_)) => {
603 let Some(corroborated) = refusals.note(peer_id, &e.to_string()) else {
609 warn!("Peer {peer_id} refused this client's settlement version; awaiting corroboration");
610 failures.push(format!("{peer_id}: {e}"));
611 return Ok(());
612 };
613 let corroborators = refusals.corroborating_peers();
618 warn!(
619 "Settlement refusal corroborated by {} distinct peers [{}]; aborting before payment",
620 corroborators.len(),
621 corroborators.join(", ")
622 );
623 let verdict = Error::ClientUpdateRequired(corroborated);
624 if settlement_refusal.is_none() {
625 *settlement_refusal = Some(Error::ClientUpdateRequired(verdict.to_string()));
626 }
627 return Err(verdict);
628 }
629 Err(e) => {
630 if matches!(&e, Error::BadQuoteBinding { .. }) {
631 *bad_quote_count += 1;
632 }
633 warn!("Failed to get quote from {peer_id}: {e}");
634 failures.push(format!("{peer_id}: {e}"));
635 }
636 }
637 Ok(())
638}
639
640fn witnessed_quote_launch_budget(
641 successful_quotes: usize,
642 in_flight: usize,
643 remaining_peers: usize,
644) -> usize {
645 CLOSE_GROUP_SIZE
646 .saturating_sub(successful_quotes.saturating_add(in_flight))
647 .min(remaining_peers)
648}
649
650fn single_node_quote_query_count() -> usize {
651 CLOSE_GROUP_SIZE
652}
653
654fn fault_tolerant_quote_query_count() -> usize {
655 CLOSE_GROUP_SIZE * FAULT_TOLERANT_QUOTE_QUERY_MULTIPLIER
656}
657
658fn witnessed_close_group_quorum() -> usize {
659 (CLOSE_GROUP_SIZE * WITNESSED_QUORUM_NUMERATOR).div_ceil(WITNESSED_QUORUM_DENOMINATOR)
660}
661
662fn witnessed_close_group_quorum_for_missing_views(missing_views: usize) -> usize {
663 witnessed_close_group_quorum()
664 .saturating_sub(missing_views)
665 .max(1)
666}
667
668fn missing_witnessed_responder_views(witnessed: &WitnessedCloseGroup) -> usize {
669 witnessed
670 .initial_closest
671 .len()
672 .saturating_sub(witnessed.responder_views.len())
673}
674
675fn witnessed_close_group_quorum_for_transcript(witnessed: &WitnessedCloseGroup) -> usize {
676 witnessed_close_group_quorum_for_missing_views(missing_witnessed_responder_views(witnessed))
677}
678
679fn scope_witnessed_to_close_group(witnessed: &WitnessedCloseGroup) -> WitnessedCloseGroup {
688 let initial_closest: Vec<DHTNode> = witnessed
689 .initial_closest
690 .iter()
691 .take(CLOSE_GROUP_SIZE)
692 .cloned()
693 .collect();
694 let scope: HashSet<PeerId> = initial_closest.iter().map(|node| node.peer_id).collect();
695 let responder_views: Vec<ResponderView> = witnessed
696 .responder_views
697 .iter()
698 .filter(|view| scope.contains(&view.responder))
699 .cloned()
700 .collect();
701 WitnessedCloseGroup {
702 target: witnessed.target,
703 k: CLOSE_GROUP_SIZE,
704 initial_closest,
705 responder_views,
706 }
707}
708
709fn peer_list(peers: &[PeerId]) -> Vec<String> {
710 peers.iter().map(ToString::to_string).collect()
711}
712
713pub(crate) type StoreQuote = (
717 PeerId,
718 Vec<MultiAddr>,
719 PaymentQuote,
720 Amount,
721 Option<Vec<u8>>,
722);
723type StoreQuoteRequestResult = (
724 PeerId,
725 Vec<MultiAddr>,
726 Result<(PaymentQuote, Amount, Option<Vec<u8>>)>,
727);
728type VotersByPeer = HashMap<PeerId, HashSet<PeerId>>;
729type WitnessedVoteData = (HashMap<PeerId, DHTNode>, VotersByPeer, Vec<(PeerId, usize)>);
730
731pub(crate) struct StoreQuotePlan {
732 pub(crate) quotes: Vec<StoreQuote>,
733 pub(crate) put_peers: Vec<(PeerId, Vec<MultiAddr>)>,
734}
735
736#[derive(Debug, Clone)]
737struct WitnessedQuoteCandidate {
738 node: DHTNode,
739 votes: usize,
740 voters: HashSet<PeerId>,
741}
742
743#[derive(Debug, Clone)]
744struct WitnessedQuotePeer {
745 peer_id: PeerId,
746 addrs: Vec<MultiAddr>,
747 voters: HashSet<PeerId>,
748}
749
750#[derive(Debug, Clone)]
751struct WitnessedQuoteSelection {
752 quote_peers: Vec<WitnessedQuotePeer>,
753 initial_put_peers: Vec<(PeerId, Vec<MultiAddr>)>,
754 quorum: usize,
755}
756
757enum QuoteSelectionPolicy {
758 ClosestByDistance,
759 WitnessedMedianVoters {
760 voters_by_peer: VotersByPeer,
761 quorum: usize,
762 },
763}
764
765fn witnessed_initial_peers(witnessed: &WitnessedCloseGroup) -> Vec<String> {
766 witnessed
767 .initial_closest
768 .iter()
769 .map(|node| node.peer_id.to_string())
770 .collect()
771}
772
773fn witnessed_responder_views(witnessed: &WitnessedCloseGroup) -> Vec<String> {
774 witnessed
775 .responder_views
776 .iter()
777 .map(|view| {
778 let peers = view
779 .closest
780 .iter()
781 .map(|node| node.peer_id)
782 .collect::<Vec<_>>();
783 format!("{}=>{:?}", view.responder, peer_list(&peers))
784 })
785 .collect()
786}
787
788fn merge_witnessed_node(nodes: &mut HashMap<PeerId, DHTNode>, node: DHTNode) {
789 match nodes.entry(node.peer_id) {
790 std::collections::hash_map::Entry::Occupied(mut entry) => {
791 entry.get_mut().merge_from(node);
792 }
793 std::collections::hash_map::Entry::Vacant(entry) => {
794 entry.insert(node);
795 }
796 }
797}
798
799fn sort_vote_counts_by_distance(vote_counts: &mut [(PeerId, usize)], address: &[u8; 32]) {
800 vote_counts.sort_by(|left, right| {
801 peer_xor_distance(&left.0, address)
802 .cmp(&peer_xor_distance(&right.0, address))
803 .then_with(|| left.0.as_bytes().cmp(right.0.as_bytes()))
804 });
805}
806
807fn witnessed_vote_counts_and_nodes(
808 witnessed: &WitnessedCloseGroup,
809 address: &[u8; 32],
810) -> WitnessedVoteData {
811 let mut known_nodes = HashMap::new();
812 for node in &witnessed.initial_closest {
813 merge_witnessed_node(&mut known_nodes, node.clone());
814 }
815
816 let mut voters_by_peer: HashMap<PeerId, HashSet<PeerId>> = HashMap::new();
817 for view in &witnessed.responder_views {
818 let mut voted = HashSet::new();
819 for node in &view.closest {
820 merge_witnessed_node(&mut known_nodes, node.clone());
821 if voted.insert(node.peer_id) {
822 voters_by_peer
823 .entry(node.peer_id)
824 .or_default()
825 .insert(view.responder);
826 }
827 }
828 }
829
830 let mut vote_counts: Vec<(PeerId, usize)> = voters_by_peer
831 .iter()
832 .map(|(peer_id, voters)| (*peer_id, voters.len()))
833 .collect();
834 sort_vote_counts_by_distance(&mut vote_counts, address);
835 (known_nodes, voters_by_peer, vote_counts)
836}
837
838fn witnessed_consensus_candidates(
839 witnessed: &WitnessedCloseGroup,
840 address: &[u8; 32],
841 quorum: usize,
842) -> Vec<WitnessedQuoteCandidate> {
843 let (known_nodes, voters_by_peer, vote_counts) =
844 witnessed_vote_counts_and_nodes(witnessed, address);
845 let mut candidates = vote_counts
846 .iter()
847 .filter_map(|(peer_id, votes)| {
848 if *votes < quorum {
849 return None;
850 }
851 known_nodes.get(peer_id).cloned().and_then(|node| {
852 voters_by_peer
853 .get(peer_id)
854 .cloned()
855 .map(|voters| WitnessedQuoteCandidate {
856 node,
857 votes: *votes,
858 voters,
859 })
860 })
861 })
862 .collect::<Vec<_>>();
863
864 candidates.sort_by(|left, right| {
865 peer_xor_distance(&left.node.peer_id, address)
866 .cmp(&peer_xor_distance(&right.node.peer_id, address))
867 .then_with(|| right.votes.cmp(&left.votes))
868 .then_with(|| {
869 left.node
870 .peer_id
871 .as_bytes()
872 .cmp(right.node.peer_id.as_bytes())
873 })
874 });
875 candidates
876}
877
878fn witnessed_vote_counts(witnessed: &WitnessedCloseGroup, address: &[u8; 32]) -> Vec<String> {
879 let (_, _, vote_counts) = witnessed_vote_counts_and_nodes(witnessed, address);
880 vote_counts
881 .iter()
882 .map(|(peer_id, votes)| format!("{peer_id}:{votes}"))
883 .collect()
884}
885
886fn witnessed_consensus(
887 witnessed: &WitnessedCloseGroup,
888 address: &[u8; 32],
889 quorum: usize,
890) -> Vec<String> {
891 witnessed_consensus_candidates(witnessed, address, quorum)
892 .iter()
893 .map(|candidate| format!("{}:{}", candidate.node.peer_id, candidate.votes))
894 .collect()
895}
896
897fn witnessed_close_group_diagnostics(
898 address: &[u8; 32],
899 witnessed: &WitnessedCloseGroup,
900 quorum: usize,
901) -> String {
902 format!(
903 "target={}, initial={:?}, responder_views={:?}, vote_counts={:?}, quorum={}, final={:?}",
904 hex::encode(address),
905 witnessed_initial_peers(witnessed),
906 witnessed_responder_views(witnessed),
907 witnessed_vote_counts(witnessed, address),
908 quorum,
909 witnessed_consensus(witnessed, address, quorum)
910 )
911}
912
913fn witnessed_quote_selection_or_error(
914 address: &[u8; 32],
915 witnessed: &WitnessedCloseGroup,
916 required: usize,
917 quorum: usize,
918) -> Result<WitnessedQuoteSelection> {
919 let candidates = witnessed_consensus_candidates(witnessed, address, quorum);
920 if candidates.len() < required {
921 return Err(Error::InsufficientPeers(format!(
922 "Witnessed close group inconclusive before payment: got {}/{} quorum-recognised peers. {}",
923 candidates.len(),
924 required,
925 witnessed_close_group_diagnostics(address, witnessed, quorum)
926 )));
927 }
928
929 let initial_put_peers = witnessed
930 .initial_closest
931 .iter()
932 .take(CLOSE_GROUP_SIZE)
933 .map(|node| (node.peer_id, node.addresses_by_priority()))
934 .collect::<Vec<_>>();
935
936 if initial_put_peers.len() < CLOSE_GROUP_SIZE {
937 return Err(Error::InsufficientPeers(format!(
938 "Witnessed close group returned only {}/{} initial PUT peers before payment. {}",
939 initial_put_peers.len(),
940 CLOSE_GROUP_SIZE,
941 witnessed_close_group_diagnostics(address, witnessed, quorum)
942 )));
943 }
944
945 let quote_peers = candidates
946 .into_iter()
947 .map(|candidate| WitnessedQuotePeer {
948 peer_id: candidate.node.peer_id,
949 addrs: candidate.node.addresses_by_priority(),
950 voters: candidate.voters,
951 })
952 .collect();
953
954 Ok(WitnessedQuoteSelection {
955 quote_peers,
956 initial_put_peers,
957 quorum,
958 })
959}
960
961pub(crate) fn median_paid_quote_issuer(quotes: &[StoreQuote]) -> Option<(PeerId, Amount)> {
962 if quotes.is_empty() {
963 return None;
964 }
965
966 let median_quote_index = quotes.len() / 2;
967
968 let mut by_price: Vec<(usize, PeerId, Amount)> = quotes
969 .iter()
970 .enumerate()
971 .map(|(index, (peer_id, _, _, price, _))| (index, *peer_id, *price))
972 .collect();
973 by_price.sort_by_key(|(index, _, price)| (*price, *index));
974 by_price
975 .get(median_quote_index)
976 .map(|(_, peer_id, price)| (*peer_id, *price))
977}
978
979fn sort_quotes_by_distance(quotes: &mut [StoreQuote], address: &[u8; 32]) {
980 quotes.sort_by(|left, right| {
981 peer_xor_distance(&left.0, address)
982 .cmp(&peer_xor_distance(&right.0, address))
983 .then_with(|| left.0.as_bytes().cmp(right.0.as_bytes()))
984 });
985}
986
987fn median_paid_quote_issuer_for_indices(
988 quotes: &[StoreQuote],
989 indices: &[usize],
990) -> Option<(PeerId, Amount)> {
991 if indices.is_empty() {
992 return None;
993 }
994
995 let median_quote_index = indices.len() / 2;
996
997 let mut by_price: Vec<(usize, PeerId, Amount)> = indices
998 .iter()
999 .enumerate()
1000 .map(|(selected_index, quote_index)| {
1001 let (peer_id, _, _, price, _) = "es[*quote_index];
1002 (selected_index, *peer_id, *price)
1003 })
1004 .collect();
1005 by_price.sort_by_key(|(selected_index, _, price)| (*price, *selected_index));
1006 by_price
1007 .get(median_quote_index)
1008 .map(|(_, peer_id, price)| (*peer_id, *price))
1009}
1010
1011fn median_issuer_voter_support(
1012 quotes: &[StoreQuote],
1013 indices: &[usize],
1014 voters_by_peer: &VotersByPeer,
1015) -> Option<(PeerId, usize)> {
1016 let (median_peer_id, _) = median_paid_quote_issuer_for_indices(quotes, indices)?;
1017 let voters = voters_by_peer.get(&median_peer_id)?;
1018 Some((median_peer_id, voters.len()))
1019}
1020
1021fn visit_quote_subsets<F>(
1022 quote_count: usize,
1023 subset_size: usize,
1024 start_index: usize,
1025 current: &mut Vec<usize>,
1026 visit: &mut F,
1027) where
1028 F: FnMut(&[usize]),
1029{
1030 if current.len() == subset_size {
1031 visit(current);
1032 return;
1033 }
1034
1035 let remaining = subset_size - current.len();
1036 let last_start = quote_count - remaining;
1037 for index in start_index..=last_start {
1038 current.push(index);
1039 visit_quote_subsets(quote_count, subset_size, index + 1, current, visit);
1040 current.pop();
1041 }
1042}
1043
1044fn select_closest_quotes(mut quotes: Vec<StoreQuote>, address: &[u8; 32]) -> Vec<StoreQuote> {
1045 sort_quotes_by_distance(&mut quotes, address);
1046 quotes.truncate(CLOSE_GROUP_SIZE);
1047 quotes
1048}
1049
1050fn select_witnessed_median_voter_quotes(
1051 mut quotes: Vec<StoreQuote>,
1052 address: &[u8; 32],
1053 voters_by_peer: &VotersByPeer,
1054 required_support: usize,
1055) -> Option<Vec<StoreQuote>> {
1056 if quotes.is_empty() {
1057 return None;
1058 }
1059
1060 sort_quotes_by_distance(&mut quotes, address);
1061
1062 let max_quote_count = single_node_quote_query_count().min(quotes.len());
1063 for quote_count in (SINGLE_NODE_MIN_QUOTE_COUNT..=max_quote_count).rev() {
1064 let mut best_indices: Option<(usize, Vec<usize>)> = None;
1065 let mut current_indices = Vec::with_capacity(quote_count);
1066 visit_quote_subsets(
1067 quotes.len(),
1068 quote_count,
1069 0,
1070 &mut current_indices,
1071 &mut |indices| {
1072 let Some((_, support)) =
1073 median_issuer_voter_support("es, indices, voters_by_peer)
1074 else {
1075 return;
1076 };
1077 if support < required_support {
1078 return;
1079 }
1080 match &best_indices {
1081 Some((best_support, best)) if *best_support > support => {}
1082 Some((best_support, best))
1083 if *best_support == support && best.as_slice() <= indices => {}
1084 _ => best_indices = Some((support, indices.to_vec())),
1085 }
1086 },
1087 );
1088
1089 if let Some((_, indices)) = best_indices {
1090 return Some(
1091 indices
1092 .into_iter()
1093 .map(|index| quotes[index].clone())
1094 .collect(),
1095 );
1096 }
1097 }
1098
1099 None
1100}
1101
1102fn put_peers_with_median_voters_first(
1103 quotes: &[StoreQuote],
1104 put_peers: &[(PeerId, Vec<MultiAddr>)],
1105 voters_by_peer: &VotersByPeer,
1106 required_support: usize,
1107) -> Option<Vec<(PeerId, Vec<MultiAddr>)>> {
1108 let (median_peer_id, _) = median_paid_quote_issuer(quotes)?;
1109 let voters = voters_by_peer.get(&median_peer_id)?;
1110
1111 let mut supporting_peers = Vec::new();
1112 let mut fallback_peers = Vec::new();
1113 for (peer_id, addrs) in put_peers {
1114 let peer = (*peer_id, addrs.clone());
1115 if voters.contains(peer_id) {
1116 supporting_peers.push(peer);
1117 } else {
1118 fallback_peers.push(peer);
1119 }
1120 }
1121
1122 if supporting_peers.len() < required_support {
1123 return None;
1124 }
1125
1126 supporting_peers.extend(fallback_peers);
1127 Some(supporting_peers)
1128}
1129
1130impl Client {
1131 pub async fn get_store_quotes(
1146 &self,
1147 address: &[u8; 32],
1148 data_size: u64,
1149 data_type: u32,
1150 ) -> Result<Vec<StoreQuote>> {
1151 Ok(self
1152 .get_store_quote_plan(address, data_size, data_type)
1153 .await?
1154 .quotes)
1155 }
1156
1157 pub(crate) async fn get_store_quote_plan(
1164 &self,
1165 address: &[u8; 32],
1166 data_size: u64,
1167 data_type: u32,
1168 ) -> Result<StoreQuotePlan> {
1169 let witnessed_selection = self.select_witnessed_quote_selection(address).await?;
1170 let voters_by_peer: VotersByPeer = witnessed_selection
1171 .quote_peers
1172 .iter()
1173 .map(|peer| (peer.peer_id, peer.voters.clone()))
1174 .collect();
1175 let remote_peers = witnessed_selection
1176 .quote_peers
1177 .into_iter()
1178 .map(|peer| (peer.peer_id, peer.addrs))
1179 .collect();
1180 let initial_put_peers = witnessed_selection.initial_put_peers;
1181 let quorum = witnessed_selection.quorum;
1182 let quotes = self
1183 .collect_store_quotes_from_remote_peers(
1184 address,
1185 data_size,
1186 data_type,
1187 remote_peers,
1188 QuoteSelectionPolicy::WitnessedMedianVoters {
1189 voters_by_peer: voters_by_peer.clone(),
1190 quorum,
1191 },
1192 )
1193 .await?;
1194 let put_peers = put_peers_with_median_voters_first(
1195 "es,
1196 &initial_put_peers,
1197 &voters_by_peer,
1198 quorum,
1199 )
1200 .ok_or_else(|| {
1201 Error::InsufficientPeers(format!(
1202 "Collected {} witnessed quotes, but fewer than {} initial witness PUT peers \
1203 voted for the paid median issuer for {}",
1204 quotes.len(),
1205 quorum,
1206 hex::encode(address)
1207 ))
1208 })?;
1209
1210 Ok(StoreQuotePlan { quotes, put_peers })
1211 }
1212
1213 pub(crate) async fn get_store_quotes_with_fault_tolerance(
1220 &self,
1221 address: &[u8; 32],
1222 data_size: u64,
1223 data_type: u32,
1224 ) -> Result<Vec<StoreQuote>> {
1225 let peer_query_count = fault_tolerant_quote_query_count();
1226 let remote_peers = self
1227 .network()
1228 .find_closest_peers(address, peer_query_count)
1229 .await?;
1230
1231 self.collect_store_quotes_from_remote_peers(
1232 address,
1233 data_size,
1234 data_type,
1235 remote_peers,
1236 QuoteSelectionPolicy::ClosestByDistance,
1237 )
1238 .await
1239 }
1240
1241 async fn select_witnessed_quote_selection(
1242 &self,
1243 address: &[u8; 32],
1244 ) -> Result<WitnessedQuoteSelection> {
1245 let close_group_query_count = single_node_quote_query_count();
1248 let required_quotes = SINGLE_NODE_MIN_QUOTE_COUNT;
1249 let witnessed = match self
1255 .network()
1256 .find_witnessed_close_group_with_view_count(
1257 address,
1258 PUT_TARGET_WIDTH,
1259 SINGLE_NODE_WITNESSED_VIEW_COUNT,
1260 )
1261 .await
1262 {
1263 Ok(witnessed) => witnessed,
1264 Err(wide_err) => {
1265 debug!(
1266 target = %hex::encode(address),
1267 "Wide witnessed lookup ({PUT_TARGET_WIDTH}) failed ({wide_err}); \
1268 retrying at close-group width ({close_group_query_count})"
1269 );
1270 self.network()
1271 .find_witnessed_close_group_with_view_count(
1272 address,
1273 close_group_query_count,
1274 SINGLE_NODE_WITNESSED_VIEW_COUNT,
1275 )
1276 .await
1277 .map_err(|e| {
1278 Error::InsufficientPeers(format!(
1279 "Witnessed close group lookup failed before payment for target {}: {e}",
1280 hex::encode(address)
1281 ))
1282 })?
1283 }
1284 };
1285 let witnessed_quote = scope_witnessed_to_close_group(&witnessed);
1288 let base_quorum = witnessed_close_group_quorum();
1289 let missing_views = missing_witnessed_responder_views(&witnessed_quote);
1290 let quorum = witnessed_close_group_quorum_for_transcript(&witnessed_quote);
1291
1292 if missing_views > 0 {
1293 warn!(
1294 target = %hex::encode(address),
1295 initial = witnessed_quote.initial_closest.len(),
1296 responder_views = witnessed_quote.responder_views.len(),
1297 missing_views = missing_views,
1298 base_quorum = base_quorum,
1299 adjusted_quorum = quorum,
1300 "Witnessed close group transcript is missing responder views; lowering SNP witness quorum"
1301 );
1302 }
1303
1304 debug!(
1305 target = %hex::encode(address),
1306 quorum = quorum,
1307 view_count = SINGLE_NODE_WITNESSED_VIEW_COUNT,
1308 initial = ?witnessed_initial_peers(&witnessed_quote),
1309 responder_views = ?witnessed_responder_views(&witnessed_quote),
1310 vote_counts = ?witnessed_vote_counts(&witnessed_quote, address),
1311 final_witnessed_set = ?witnessed_consensus(&witnessed_quote, address, quorum),
1312 "Witnessed close group selected for SNP quote collection"
1313 );
1314
1315 let mut selection =
1316 witnessed_quote_selection_or_error(address, &witnessed_quote, required_quotes, quorum)?;
1317 selection.initial_put_peers = witnessed
1321 .initial_closest
1322 .iter()
1323 .take(PUT_TARGET_WIDTH)
1324 .map(|node| (node.peer_id, node.addresses_by_priority()))
1325 .collect();
1326 Ok(selection)
1327 }
1328
1329 #[allow(clippy::too_many_lines)]
1330 async fn collect_store_quotes_from_remote_peers(
1331 &self,
1332 address: &[u8; 32],
1333 data_size: u64,
1334 data_type: u32,
1335 remote_peers: Vec<(PeerId, Vec<MultiAddr>)>,
1336 quote_selection_policy: QuoteSelectionPolicy,
1337 ) -> Result<Vec<StoreQuote>> {
1338 let peer_query_count = remote_peers.len();
1339
1340 let node = self.network().node();
1341
1342 debug!(
1343 "Requesting quotes from up to {peer_query_count} peers for address {} (size: {data_size})",
1344 hex::encode(address)
1345 );
1346
1347 let (min_quote_count, target_quote_count, staged_witnessed_collection) =
1348 match "e_selection_policy {
1349 QuoteSelectionPolicy::ClosestByDistance => {
1350 (CLOSE_GROUP_SIZE, CLOSE_GROUP_SIZE, false)
1351 }
1352 QuoteSelectionPolicy::WitnessedMedianVoters { .. } => (
1353 SINGLE_NODE_MIN_QUOTE_COUNT,
1354 single_node_quote_query_count(),
1355 true,
1356 ),
1357 };
1358 let target_quote_count = target_quote_count.min(peer_query_count);
1359
1360 if remote_peers.len() < min_quote_count {
1361 return Err(Error::InsufficientPeers(format!(
1362 "Found {} peers, need {min_quote_count}",
1363 remote_peers.len(),
1364 )));
1365 }
1366 debug_assert!(peer_query_count >= min_quote_count);
1367
1368 let per_peer_timeout = Duration::from_secs(self.config().quote_timeout_secs);
1369 let overall_timeout = Duration::from_secs(QUOTE_COLLECTION_TIMEOUT_SECS);
1370
1371 let mut quotes = Vec::with_capacity(peer_query_count);
1375 let mut already_stored_peers: Vec<(PeerId, [u8; 32])> = Vec::new();
1376 let mut failures: Vec<String> = Vec::new();
1377
1378 let mut bad_quote_count = 0usize;
1383
1384 let mut settlement_refusal: Option<Error> = None;
1388 let refusals = self.settlement_refusals();
1389
1390 if staged_witnessed_collection {
1391 let mut quote_futures = FuturesUnordered::new();
1392 let mut next_peer_index = 0usize;
1393 let collect_result: std::result::Result<std::result::Result<(), Error>, _> =
1394 tokio::time::timeout(overall_timeout, async {
1395 loop {
1396 let launch_count = if quotes.len() >= target_quote_count {
1403 0
1404 } else {
1405 witnessed_quote_launch_budget(
1406 quotes.len(),
1407 quote_futures.len(),
1408 remote_peers.len().saturating_sub(next_peer_index),
1409 )
1410 };
1411 for _ in 0..launch_count {
1412 let (peer_id, peer_addrs) = &remote_peers[next_peer_index];
1413 next_peer_index += 1;
1414 quote_futures.push(request_store_quote_from_peer(
1415 node.clone(),
1416 *peer_id,
1417 peer_addrs.clone(),
1418 self.next_request_id(),
1419 *address,
1420 data_size,
1421 data_type,
1422 per_peer_timeout,
1423 self.unversioned_quote_peers(),
1424 self.versioned_quote_capable_handle(),
1425 ));
1426 }
1427
1428 if quote_futures.is_empty() {
1429 break;
1430 }
1431
1432 let Some((peer_id, addrs, quote_result)) = quote_futures.next().await
1433 else {
1434 break;
1435 };
1436 record_store_quote_result(
1437 peer_id,
1438 addrs,
1439 quote_result,
1440 address,
1441 &mut quotes,
1442 &mut already_stored_peers,
1443 &mut failures,
1444 &mut bad_quote_count,
1445 &mut settlement_refusal,
1446 &refusals,
1447 )?;
1448 }
1449 Ok(())
1450 })
1451 .await;
1452
1453 match collect_result {
1454 Err(_elapsed) => {
1455 warn!(
1456 "Quote collection timed out after {overall_timeout:?} for address {}",
1457 hex::encode(address)
1458 );
1459 }
1460 Ok(Err(e)) => return Err(e),
1461 Ok(Ok(())) => {}
1462 }
1463 if let Some(refusal) = settlement_refusal.take() {
1466 return Err(refusal);
1467 }
1468 } else {
1469 let mut quote_futures = FuturesUnordered::new();
1473
1474 for (peer_id, peer_addrs) in &remote_peers {
1475 quote_futures.push(request_store_quote_from_peer(
1476 node.clone(),
1477 *peer_id,
1478 peer_addrs.clone(),
1479 self.next_request_id(),
1480 *address,
1481 data_size,
1482 data_type,
1483 per_peer_timeout,
1484 self.unversioned_quote_peers(),
1485 self.versioned_quote_capable_handle(),
1486 ));
1487 }
1488
1489 let collect_result: std::result::Result<std::result::Result<(), Error>, _> =
1490 tokio::time::timeout(overall_timeout, async {
1491 while let Some((peer_id, addrs, quote_result)) = quote_futures.next().await {
1492 record_store_quote_result(
1493 peer_id,
1494 addrs,
1495 quote_result,
1496 address,
1497 &mut quotes,
1498 &mut already_stored_peers,
1499 &mut failures,
1500 &mut bad_quote_count,
1501 &mut settlement_refusal,
1502 &refusals,
1503 )?;
1504 }
1505 Ok(())
1506 })
1507 .await;
1508
1509 match collect_result {
1510 Err(_elapsed) => {
1511 warn!(
1512 "Quote collection timed out after {overall_timeout:?} for address {}",
1513 hex::encode(address)
1514 );
1515 }
1519 Ok(Err(e)) => return Err(e),
1520 Ok(Ok(())) => {}
1521 }
1522 if let Some(refusal) = settlement_refusal.take() {
1525 return Err(refusal);
1526 }
1527 }
1528
1529 let bad_dropped = drop_quotes_with_bad_bindings(&mut quotes);
1535 if bad_dropped > 0 {
1536 warn!(
1537 "Defensive filter dropped {bad_dropped} quotes with mismatched peer bindings \
1538 for address {} — the per-peer handler should have caught these earlier \
1539 (this indicates an upstream regression)",
1540 hex::encode(address),
1541 );
1542 bad_quote_count += bad_dropped;
1543 }
1544
1545 if !already_stored_peers.is_empty() {
1547 let mut all_peers_by_distance: Vec<(bool, [u8; 32])> = Vec::new();
1548 for (peer_id, _, _, _, _) in "es {
1549 all_peers_by_distance.push((false, peer_xor_distance(peer_id, address)));
1550 }
1551 for (_, dist) in &already_stored_peers {
1552 all_peers_by_distance.push((true, *dist));
1553 }
1554 all_peers_by_distance.sort_by_key(|a| a.1);
1555
1556 let close_group_stored = all_peers_by_distance
1557 .iter()
1558 .take(CLOSE_GROUP_SIZE)
1559 .filter(|(is_stored, _)| *is_stored)
1560 .count();
1561
1562 if close_group_stored >= CLOSE_GROUP_MAJORITY {
1563 debug!(
1564 "Chunk {} already stored ({close_group_stored}/{CLOSE_GROUP_SIZE} close-group peers confirm)",
1565 hex::encode(address)
1566 );
1567 return Err(Error::AlreadyStored);
1568 }
1569 }
1570
1571 let already_stored_count = already_stored_peers.len();
1572 let failure_count = failures.len();
1573 let quote_count = quotes.len();
1574 let total_responses = quote_count + failure_count + already_stored_count;
1575
1576 if quotes.len() >= min_quote_count {
1577 let selected_quotes = match quote_selection_policy {
1578 QuoteSelectionPolicy::ClosestByDistance => select_closest_quotes(quotes, address),
1579 QuoteSelectionPolicy::WitnessedMedianVoters {
1580 voters_by_peer,
1581 quorum,
1582 } => select_witnessed_median_voter_quotes(quotes, address, &voters_by_peer, quorum)
1583 .ok_or_else(|| {
1584 Error::InsufficientPeers(format!(
1585 "Got {quote_count} quotes, need at least {min_quote_count} whose paid \
1586 median issuer is recognised by at least {} \
1587 selected witness peers ({total_responses} responses: \
1588 {already_stored_count} already_stored, {failure_count} failed \
1589 including {bad_quote_count} with mismatched peer bindings). \
1590 Failures: [{}]",
1591 quorum,
1592 failures.join("; ")
1593 ))
1594 })?,
1595 };
1596
1597 info!(
1598 "Collected {} quotes for address {} ({total_responses} responses: \
1599 {quote_count} ok, {already_stored_count} already_stored, {failure_count} failed, \
1600 {bad_quote_count} bad-binding)",
1601 selected_quotes.len(),
1602 hex::encode(address),
1603 );
1604 return Ok(selected_quotes);
1605 }
1606
1607 Err(Error::InsufficientPeers(format!(
1608 "Got {quote_count} quotes, need {min_quote_count} ({total_responses} responses: \
1609 {already_stored_count} already_stored, {failure_count} failed including \
1610 {bad_quote_count} with mismatched peer bindings). Failures: [{}]",
1611 failures.join("; ")
1612 )))
1613 }
1614}
1615
1616#[cfg(test)]
1617#[allow(clippy::unwrap_used, clippy::expect_used)]
1618mod tests {
1619 use super::*;
1631 use ant_protocol::evm::RewardsAddress;
1632 use ant_protocol::pqc::ops::{MlDsaOperations, MlDsaPublicKey};
1633 use ant_protocol::transport::{DHTNode, MlDsa65, ResponderView, WitnessedCloseGroup};
1634 use std::time::SystemTime;
1635 use xor_name::XorName;
1636
1637 struct Keypair {
1639 peer_id: PeerId,
1640 pub_key_bytes: Vec<u8>,
1641 secret_key_bytes: Vec<u8>,
1642 }
1643
1644 fn gen_keypair() -> Keypair {
1645 let ml_dsa = MlDsa65::new();
1646 let (pub_key, sk) = ml_dsa.generate_keypair().expect("ML-DSA-65 keygen");
1647 let pub_key_bytes = pub_key.as_bytes().to_vec();
1648 let peer_id = PeerId::from_bytes(compute_address(&pub_key_bytes));
1649 Keypair {
1650 peer_id,
1651 pub_key_bytes,
1652 secret_key_bytes: sk.as_bytes().to_vec(),
1653 }
1654 }
1655
1656 fn signed_baseline_quote(content: [u8; 32]) -> (PeerId, PaymentQuote) {
1660 use ant_protocol::pqc::ops::MlDsaSecretKey;
1661 let kp = gen_keypair();
1662 let mut quote = PaymentQuote {
1663 content: XorName(content),
1664 timestamp: SystemTime::UNIX_EPOCH,
1665 price: calculate_price(0),
1666 rewards_address: RewardsAddress::new([0u8; 20]),
1667 pub_key: kp.pub_key_bytes.clone(),
1668 signature: Vec::new(),
1669 committed_key_count: 0,
1670 commitment_pin: None,
1671 };
1672 let ml_dsa = MlDsa65::new();
1673 let sk = MlDsaSecretKey::from_bytes(&kp.secret_key_bytes).expect("sk");
1674 let msg = quote.bytes_for_sig();
1675 quote.signature = ml_dsa.sign(&sk, &msg).expect("sign").as_bytes().to_vec();
1676 (kp.peer_id, quote)
1677 }
1678
1679 fn good_quote_real() -> QuotedPeer {
1687 let kp = gen_keypair();
1688 let quote = PaymentQuote {
1689 content: XorName([0u8; 32]),
1690 timestamp: SystemTime::UNIX_EPOCH,
1691 price: calculate_price(0),
1692 rewards_address: RewardsAddress::new([0u8; 20]),
1693 pub_key: kp.pub_key_bytes,
1694 signature: Vec::new(),
1695 committed_key_count: 0,
1696 commitment_pin: None,
1697 };
1698 (kp.peer_id, Vec::new(), quote, calculate_price(0), None)
1699 }
1700
1701 fn bad_quote_real() -> QuotedPeer {
1706 let claimed = gen_keypair();
1707 let signing = gen_keypair();
1708 assert_ne!(claimed.pub_key_bytes, signing.pub_key_bytes);
1709 assert_ne!(claimed.peer_id.as_bytes(), signing.peer_id.as_bytes());
1710 let quote = PaymentQuote {
1711 content: XorName([0u8; 32]),
1712 timestamp: SystemTime::UNIX_EPOCH,
1713 price: calculate_price(0),
1714 rewards_address: RewardsAddress::new([0u8; 20]),
1715 pub_key: signing.pub_key_bytes,
1716 signature: Vec::new(),
1717 committed_key_count: 0,
1718 commitment_pin: None,
1719 };
1720 (claimed.peer_id, Vec::new(), quote, calculate_price(0), None)
1721 }
1722
1723 fn witnessed_test_node(seed: u8) -> DHTNode {
1724 DHTNode {
1725 peer_id: PeerId::from_bytes([seed; 32]),
1726 addresses: Vec::new(),
1727 address_types: Vec::new(),
1728 distance: None,
1729 reliability: 1.0,
1730 }
1731 }
1732
1733 fn witnessed_test_nodes(seeds: &[u8]) -> Vec<DHTNode> {
1734 seeds.iter().copied().map(witnessed_test_node).collect()
1735 }
1736
1737 fn witnessed_test_view(responder: u8, closest: &[u8]) -> ResponderView {
1738 ResponderView {
1739 responder: PeerId::from_bytes([responder; 32]),
1740 closest: witnessed_test_nodes(closest),
1741 }
1742 }
1743
1744 fn synthetic_peer(seed: u8) -> PeerId {
1745 PeerId::from_bytes([seed; 32])
1746 }
1747
1748 fn synthetic_quote(
1749 seed: u8,
1750 price: u64,
1751 ) -> (
1752 PeerId,
1753 Vec<MultiAddr>,
1754 PaymentQuote,
1755 Amount,
1756 Option<Vec<u8>>,
1757 ) {
1758 let amount = Amount::from(price);
1759 let quote = PaymentQuote {
1760 content: XorName([0u8; 32]),
1761 timestamp: SystemTime::UNIX_EPOCH,
1762 price: amount,
1763 rewards_address: RewardsAddress::new([0u8; 20]),
1764 pub_key: Vec::new(),
1765 signature: Vec::new(),
1766 committed_key_count: 0,
1767 commitment_pin: None,
1768 };
1769 (synthetic_peer(seed), Vec::new(), quote, amount, None)
1770 }
1771
1772 fn synthetic_voters(seeds: &[u8]) -> HashSet<PeerId> {
1773 seeds.iter().copied().map(synthetic_peer).collect()
1774 }
1775
1776 fn quote_peer_seeds(quotes: &[StoreQuote]) -> Vec<u8> {
1777 quotes
1778 .iter()
1779 .map(|(peer_id, _, _, _, _)| peer_id.as_bytes()[0])
1780 .collect()
1781 }
1782
1783 fn put_peer_seeds(peers: &[(PeerId, Vec<MultiAddr>)]) -> Vec<u8> {
1784 peers
1785 .iter()
1786 .map(|(peer_id, _)| peer_id.as_bytes()[0])
1787 .collect()
1788 }
1789
1790 fn put_peers_from_seeds(seeds: &[u8]) -> Vec<(PeerId, Vec<MultiAddr>)> {
1791 seeds
1792 .iter()
1793 .copied()
1794 .map(|seed| (synthetic_peer(seed), Vec::new()))
1795 .collect()
1796 }
1797
1798 fn storer_binding_would_accept(peer_id: &PeerId, quote: &PaymentQuote) -> bool {
1807 if MlDsaPublicKey::from_bytes("e.pub_key).is_err() {
1808 return false;
1809 }
1810 compute_address("e.pub_key) == *peer_id.as_bytes()
1811 }
1812
1813 #[test]
1818 fn binding_accepts_real_self_consistent_keypair() {
1819 let (peer_id, _, quote, _, _) = good_quote_real();
1820 assert!(quote_binding_is_valid(&peer_id, "e));
1823 assert!(storer_binding_would_accept(&peer_id, "e));
1825 }
1826
1827 #[test]
1828 fn binding_rejects_real_crossed_keypair() {
1829 let (peer_id, _, quote, _, _) = bad_quote_real();
1830 assert!(!quote_binding_is_valid(&peer_id, "e));
1831 assert!(!storer_binding_would_accept(&peer_id, "e));
1832 }
1833
1834 #[test]
1835 fn binding_rejects_oversize_pubkey() {
1836 let oversized = vec![0u8; ML_DSA_PUB_KEY_LEN + 1];
1840 let peer_id = PeerId::from_bytes(compute_address(&oversized));
1841 let quote = PaymentQuote {
1842 content: XorName([0u8; 32]),
1843 timestamp: SystemTime::UNIX_EPOCH,
1844 price: Amount::ZERO,
1845 rewards_address: RewardsAddress::new([0u8; 20]),
1846 pub_key: oversized,
1847 signature: Vec::new(),
1848 committed_key_count: 0,
1849 commitment_pin: None,
1850 };
1851 assert_eq!(compute_address("e.pub_key), *peer_id.as_bytes());
1854 assert!(
1855 !quote_binding_is_valid(&peer_id, "e),
1856 "predicate must reject oversize pub_key even when BLAKE3 happens to match"
1857 );
1858 assert!(!storer_binding_would_accept(&peer_id, "e));
1859 }
1860
1861 #[test]
1862 fn binding_rejects_undersize_pubkey() {
1863 let undersized = vec![0u8; ML_DSA_PUB_KEY_LEN - 1];
1864 let peer_id = PeerId::from_bytes(compute_address(&undersized));
1865 let quote = PaymentQuote {
1866 content: XorName([0u8; 32]),
1867 timestamp: SystemTime::UNIX_EPOCH,
1868 price: Amount::ZERO,
1869 rewards_address: RewardsAddress::new([0u8; 20]),
1870 pub_key: undersized,
1871 signature: Vec::new(),
1872 committed_key_count: 0,
1873 commitment_pin: None,
1874 };
1875 assert!(!quote_binding_is_valid(&peer_id, "e));
1876 assert!(!storer_binding_would_accept(&peer_id, "e));
1877 }
1878
1879 #[test]
1884 fn quote_query_counts_keep_single_node_close_group_only() {
1885 assert_eq!(single_node_quote_query_count(), CLOSE_GROUP_SIZE);
1886 assert_eq!(SINGLE_NODE_MIN_QUOTE_COUNT, 1);
1887 assert_eq!(SINGLE_NODE_WITNESSED_VIEW_COUNT, 20);
1888 assert!(SINGLE_NODE_WITNESSED_VIEW_COUNT > single_node_quote_query_count());
1889 assert_eq!(witnessed_close_group_quorum(), 5);
1890 assert_eq!(witnessed_close_group_quorum_for_missing_views(0), 5);
1891 assert_eq!(witnessed_close_group_quorum_for_missing_views(1), 4);
1892 assert_eq!(witnessed_close_group_quorum_for_missing_views(2), 3);
1893 assert_eq!(
1894 fault_tolerant_quote_query_count(),
1895 CLOSE_GROUP_SIZE * FAULT_TOLERANT_QUOTE_QUERY_MULTIPLIER
1896 );
1897 assert!(fault_tolerant_quote_query_count() > single_node_quote_query_count());
1898 }
1899
1900 #[test]
1901 fn witnessed_quote_launch_budget_keeps_exact_quote_window() {
1902 assert_eq!(
1903 witnessed_quote_launch_budget(0, 0, CLOSE_GROUP_SIZE * 2),
1904 CLOSE_GROUP_SIZE,
1905 "initial SNP quote fetch should launch the closest seven peers"
1906 );
1907 assert_eq!(
1908 witnessed_quote_launch_budget(1, CLOSE_GROUP_SIZE - 1, CLOSE_GROUP_SIZE),
1909 0,
1910 "a successful quote should not launch an extra fallback"
1911 );
1912 assert_eq!(
1913 witnessed_quote_launch_budget(0, CLOSE_GROUP_SIZE - 1, CLOSE_GROUP_SIZE),
1914 1,
1915 "a failed in-flight quote should launch the next closest fallback"
1916 );
1917 assert_eq!(
1918 witnessed_quote_launch_budget(CLOSE_GROUP_SIZE - 1, 0, 3),
1919 1,
1920 "only one more peer is needed for the seventh quote"
1921 );
1922 assert_eq!(
1923 witnessed_quote_launch_budget(0, 0, CLOSE_GROUP_SIZE - 1),
1924 CLOSE_GROUP_SIZE - 1,
1925 "launch budget is capped by remaining candidates"
1926 );
1927 }
1928
1929 #[test]
1930 fn witnessed_candidates_sort_by_xor_distance_then_votes() {
1931 let address = [0u8; 32];
1932 let witnessed = WitnessedCloseGroup {
1933 target: address,
1934 k: CLOSE_GROUP_SIZE,
1935 initial_closest: witnessed_test_nodes(&[1, 2, 3, 4, 5, 6, 7]),
1936 responder_views: vec![
1937 witnessed_test_view(1, &[1, 9]),
1938 witnessed_test_view(2, &[1, 9]),
1939 witnessed_test_view(3, &[1, 9]),
1940 witnessed_test_view(4, &[1, 9]),
1941 witnessed_test_view(5, &[1, 9]),
1942 witnessed_test_view(6, &[9]),
1943 witnessed_test_view(7, &[9]),
1944 ],
1945 };
1946
1947 let candidates =
1948 witnessed_consensus_candidates(&witnessed, &address, witnessed_close_group_quorum());
1949
1950 assert_eq!(
1951 candidates
1952 .iter()
1953 .map(|candidate| candidate.node.peer_id.as_bytes()[0])
1954 .collect::<Vec<_>>(),
1955 vec![1, 9],
1956 "XOR closeness must be the primary sort before quote collection"
1957 );
1958 }
1959
1960 fn ascending_seeds(count: usize) -> Vec<u8> {
1962 (1..=count)
1963 .map(|n| u8::try_from(n).expect("test seed fits in u8"))
1964 .collect()
1965 }
1966
1967 #[test]
1968 fn scope_witnessed_to_close_group_matches_native_close_group_query() {
1969 const RESPONDED_IN_SCOPE: usize = 5;
1973 const OUT_OF_SCOPE_RESPONDERS: usize = 2;
1975
1976 let address = [0u8; 32];
1977 let close_seeds = ascending_seeds(CLOSE_GROUP_SIZE);
1978 let view_closest = [1, 2, 8, 9];
1981 let in_scope_views = || -> Vec<ResponderView> {
1982 ascending_seeds(RESPONDED_IN_SCOPE)
1983 .into_iter()
1984 .map(|responder| witnessed_test_view(responder, &view_closest))
1985 .collect()
1986 };
1987
1988 let mut wide_views = in_scope_views();
1992 for offset in 1..=OUT_OF_SCOPE_RESPONDERS {
1993 let responder =
1994 u8::try_from(CLOSE_GROUP_SIZE + offset).expect("out-of-scope seed fits in u8");
1995 wide_views.push(witnessed_test_view(responder, &[1, 2, 3]));
1996 }
1997 let wide = WitnessedCloseGroup {
1998 target: address,
1999 k: PUT_TARGET_WIDTH,
2000 initial_closest: witnessed_test_nodes(&ascending_seeds(PUT_TARGET_WIDTH)),
2001 responder_views: wide_views,
2002 };
2003
2004 let native = WitnessedCloseGroup {
2007 target: address,
2008 k: CLOSE_GROUP_SIZE,
2009 initial_closest: witnessed_test_nodes(&close_seeds),
2010 responder_views: in_scope_views(),
2011 };
2012
2013 let scoped = scope_witnessed_to_close_group(&wide);
2014
2015 assert_eq!(scoped.target, wide.target);
2017 assert_eq!(scoped.k, CLOSE_GROUP_SIZE);
2018 assert_eq!(
2019 scoped
2020 .initial_closest
2021 .iter()
2022 .map(|node| node.peer_id.as_bytes()[0])
2023 .collect::<Vec<_>>(),
2024 close_seeds,
2025 "initial set must be the closest CLOSE_GROUP_SIZE, in order"
2026 );
2027
2028 assert_eq!(
2031 scoped
2032 .responder_views
2033 .iter()
2034 .map(|view| view.responder.as_bytes()[0])
2035 .collect::<Vec<_>>(),
2036 ascending_seeds(RESPONDED_IN_SCOPE),
2037 "only responders inside the close group survive"
2038 );
2039 assert_eq!(
2040 scoped.responder_views[0]
2041 .closest
2042 .iter()
2043 .map(|node| node.peer_id.as_bytes()[0])
2044 .collect::<Vec<_>>(),
2045 view_closest.to_vec(),
2046 "a surviving view's closest set must be preserved verbatim"
2047 );
2048
2049 assert_eq!(
2052 missing_witnessed_responder_views(&scoped),
2053 missing_witnessed_responder_views(&native),
2054 );
2055 let quorum = witnessed_close_group_quorum_for_transcript(&scoped);
2056 assert_eq!(quorum, witnessed_close_group_quorum_for_transcript(&native));
2057 let candidate_seeds = |group: &WitnessedCloseGroup| {
2058 witnessed_consensus_candidates(group, &address, quorum)
2059 .iter()
2060 .map(|candidate| candidate.node.peer_id.as_bytes()[0])
2061 .collect::<Vec<_>>()
2062 };
2063 assert_eq!(
2064 candidate_seeds(&scoped),
2065 candidate_seeds(&native),
2066 "scoped consensus must match a native close-group query"
2067 );
2068 }
2069
2070 #[test]
2071 fn witnessed_quote_peers_error_is_typed_and_pre_payment_when_consensus_is_short() {
2072 let address = [0u8; 32];
2073 let responder_views = (1..=7)
2074 .map(|responder| witnessed_test_view(responder, &[1, 2, 3, 4]))
2075 .collect();
2076 let witnessed = WitnessedCloseGroup {
2077 target: address,
2078 k: CLOSE_GROUP_SIZE,
2079 initial_closest: witnessed_test_nodes(&[1, 2, 3, 4, 5, 6, 7]),
2080 responder_views,
2081 };
2082
2083 let err = witnessed_quote_selection_or_error(
2084 &address,
2085 &witnessed,
2086 CLOSE_GROUP_SIZE,
2087 witnessed_close_group_quorum(),
2088 )
2089 .expect_err("short witnessed consensus must fail before payment");
2090
2091 match err {
2092 Error::InsufficientPeers(message) => {
2093 assert!(message.contains("before payment"));
2094 assert!(message.contains("vote_counts"));
2095 assert!(message.contains("quorum"));
2096 }
2097 other => panic!("expected typed InsufficientPeers error, got {other:?}"),
2098 }
2099 }
2100
2101 #[test]
2102 fn witnessed_quote_selection_accepts_one_quorum_recognised_candidate() {
2103 let address = [0u8; 32];
2104 let witnessed = WitnessedCloseGroup {
2105 target: address,
2106 k: CLOSE_GROUP_SIZE,
2107 initial_closest: witnessed_test_nodes(&[1, 2, 3, 4, 5, 6, 7]),
2108 responder_views: (1..=7)
2109 .map(|responder| witnessed_test_view(responder, &[1]))
2110 .collect(),
2111 };
2112
2113 let selection = witnessed_quote_selection_or_error(
2114 &address,
2115 &witnessed,
2116 SINGLE_NODE_MIN_QUOTE_COUNT,
2117 witnessed_close_group_quorum(),
2118 )
2119 .expect("one quorum-recognised candidate is enough before payment");
2120
2121 assert_eq!(
2122 selection
2123 .quote_peers
2124 .iter()
2125 .map(|peer| peer.peer_id.as_bytes()[0])
2126 .collect::<Vec<_>>(),
2127 vec![1]
2128 );
2129 assert_eq!(
2130 put_peer_seeds(&selection.initial_put_peers),
2131 vec![1, 2, 3, 4, 5, 6, 7]
2132 );
2133 }
2134
2135 #[test]
2136 fn witnessed_quote_peers_include_quorum_fallback_candidates() {
2137 const EXTRA_QUORUM_CANDIDATES: usize = 1;
2138
2139 let address = [0u8; 32];
2140 let witnessed = WitnessedCloseGroup {
2141 target: address,
2142 k: CLOSE_GROUP_SIZE,
2143 initial_closest: witnessed_test_nodes(&[1, 2, 3, 4, 5, 6, 7]),
2144 responder_views: vec![
2145 witnessed_test_view(1, &[1, 2, 3, 4, 5, 6, 7]),
2146 witnessed_test_view(2, &[1, 2, 3, 4, 5, 6, 8]),
2147 witnessed_test_view(3, &[1, 2, 3, 4, 5, 7, 8]),
2148 witnessed_test_view(4, &[1, 2, 3, 4, 6, 7, 8]),
2149 witnessed_test_view(5, &[1, 2, 3, 5, 6, 7, 8]),
2150 witnessed_test_view(6, &[1, 2, 4, 5, 6, 7, 8]),
2151 witnessed_test_view(7, &[1, 3, 4, 5, 6, 7, 8]),
2152 ],
2153 };
2154
2155 let selection = witnessed_quote_selection_or_error(
2156 &address,
2157 &witnessed,
2158 CLOSE_GROUP_SIZE,
2159 witnessed_close_group_quorum(),
2160 )
2161 .expect("fallback candidates should be retained for quote collection");
2162
2163 assert_eq!(
2164 selection.quote_peers.len(),
2165 CLOSE_GROUP_SIZE + EXTRA_QUORUM_CANDIDATES
2166 );
2167 assert_eq!(
2168 selection
2169 .quote_peers
2170 .iter()
2171 .map(|peer| peer.peer_id.as_bytes()[0])
2172 .collect::<Vec<_>>(),
2173 vec![1, 2, 3, 4, 5, 6, 7, 8]
2174 );
2175 assert_eq!(
2176 put_peer_seeds(&selection.initial_put_peers),
2177 vec![1, 2, 3, 4, 5, 6, 7]
2178 );
2179 }
2180
2181 #[test]
2182 fn witnessed_quote_peers_lower_quorum_for_missing_responder_views() {
2183 let address = [0u8; 32];
2184 let witnessed = WitnessedCloseGroup {
2185 target: address,
2186 k: CLOSE_GROUP_SIZE,
2187 initial_closest: witnessed_test_nodes(&[1, 2, 3, 4, 5, 6, 7]),
2188 responder_views: vec![
2189 witnessed_test_view(1, &[1, 2, 3, 4, 5, 6, 7]),
2190 witnessed_test_view(2, &[1, 2, 3, 4, 5, 6, 8]),
2191 witnessed_test_view(3, &[1, 2, 3, 4, 5, 7, 8]),
2192 witnessed_test_view(4, &[1, 2, 3, 4, 6, 7, 8]),
2193 witnessed_test_view(5, &[1, 2, 3, 5, 6, 7, 8]),
2194 witnessed_test_view(6, &[1, 2, 4, 5, 6, 7, 8]),
2195 ],
2196 };
2197 let quorum = witnessed_close_group_quorum_for_transcript(&witnessed);
2198
2199 assert_eq!(missing_witnessed_responder_views(&witnessed), 1);
2200 assert_eq!(quorum, 4);
2201
2202 let selection =
2203 witnessed_quote_selection_or_error(&address, &witnessed, CLOSE_GROUP_SIZE, quorum)
2204 .expect(
2205 "one missing responder view should lower quorum and still select candidates",
2206 );
2207
2208 assert_eq!(
2209 selection
2210 .quote_peers
2211 .iter()
2212 .map(|peer| peer.peer_id.as_bytes()[0])
2213 .collect::<Vec<_>>(),
2214 vec![1, 2, 3, 4, 5, 6, 7, 8]
2215 );
2216 assert_eq!(selection.quorum, quorum);
2217 }
2218
2219 #[test]
2220 fn witnessed_quote_selection_keeps_closest_set_with_median_voter_quorum() {
2221 const MEDIAN_ISSUER_SEED: u8 = 7;
2222 const FAR_SUPPORTING_VOTER_SEED: u8 = 20;
2223 const UNSUCCESSFUL_SUPPORTING_VOTER_SEED: u8 = 21;
2224
2225 let address = [0u8; 32];
2226 let quotes = vec![
2227 synthetic_quote(1, 10),
2228 synthetic_quote(2, 20),
2229 synthetic_quote(3, 30),
2230 synthetic_quote(6, 50),
2231 synthetic_quote(MEDIAN_ISSUER_SEED, 40),
2232 synthetic_quote(8, 60),
2233 synthetic_quote(9, 70),
2234 synthetic_quote(FAR_SUPPORTING_VOTER_SEED, 80),
2235 ];
2236 let mut voters_by_peer = HashMap::new();
2237 voters_by_peer.insert(
2238 synthetic_peer(MEDIAN_ISSUER_SEED),
2239 synthetic_voters(&[
2240 1,
2241 2,
2242 3,
2243 MEDIAN_ISSUER_SEED,
2244 FAR_SUPPORTING_VOTER_SEED,
2245 UNSUCCESSFUL_SUPPORTING_VOTER_SEED,
2246 ]),
2247 );
2248
2249 let quorum = witnessed_close_group_quorum();
2250 let selected =
2251 select_witnessed_median_voter_quotes(quotes, &address, &voters_by_peer, quorum)
2252 .expect("a supported close-group quote set should be selected");
2253
2254 assert_eq!(quote_peer_seeds(&selected), vec![1, 2, 3, 6, 7, 8, 9]);
2255 let (median_peer_id, _) =
2256 median_paid_quote_issuer(&selected).expect("selected quotes have a median");
2257 assert_eq!(median_peer_id, synthetic_peer(MEDIAN_ISSUER_SEED));
2258 assert!(voters_by_peer[&median_peer_id].len() >= quorum);
2259 }
2260
2261 #[test]
2262 fn witnessed_quote_selection_uses_direct_median_witness_recognition() {
2263 const MEDIAN_ISSUER_SEED: u8 = 7;
2264
2265 let address = [0u8; 32];
2266 let quotes = vec![
2267 synthetic_quote(1, 10),
2268 synthetic_quote(2, 20),
2269 synthetic_quote(3, 30),
2270 synthetic_quote(4, 50),
2271 synthetic_quote(MEDIAN_ISSUER_SEED, 40),
2272 synthetic_quote(8, 60),
2273 synthetic_quote(9, 70),
2274 ];
2275 let mut voters_by_peer = HashMap::new();
2276 voters_by_peer.insert(
2277 synthetic_peer(MEDIAN_ISSUER_SEED),
2278 synthetic_voters(&[20, 21, 22, 23, 24]),
2279 );
2280
2281 let quorum = witnessed_close_group_quorum();
2282 let selected =
2283 select_witnessed_median_voter_quotes(quotes, &address, &voters_by_peer, quorum)
2284 .expect("direct witness recognition should support the paid median issuer");
2285
2286 let (median_peer_id, _) =
2287 median_paid_quote_issuer(&selected).expect("selected quotes have a median");
2288 let selected_peers = selected
2289 .iter()
2290 .map(|(peer_id, _, _, _, _)| *peer_id)
2291 .collect::<HashSet<_>>();
2292 assert_eq!(median_peer_id, synthetic_peer(MEDIAN_ISSUER_SEED));
2293 assert_eq!(
2294 voters_by_peer[&median_peer_id]
2295 .intersection(&selected_peers)
2296 .count(),
2297 0,
2298 "recognising witnesses need not also be selected quote issuers"
2299 );
2300 assert_eq!(voters_by_peer[&median_peer_id].len(), quorum);
2301 }
2302
2303 #[test]
2304 fn witnessed_quote_selection_allows_single_required_quote() {
2305 const QUOTE_ISSUER_SEED: u8 = 7;
2306
2307 let address = [0u8; 32];
2308 let quotes = vec![
2309 synthetic_quote(QUOTE_ISSUER_SEED, 10),
2310 synthetic_quote(1, 20),
2311 synthetic_quote(2, 30),
2312 ];
2313 let mut voters_by_peer = HashMap::new();
2314 voters_by_peer.insert(
2315 synthetic_peer(QUOTE_ISSUER_SEED),
2316 synthetic_voters(&[1, 2, 3, 4, 5]),
2317 );
2318
2319 let selected = select_witnessed_median_voter_quotes(
2320 quotes,
2321 &address,
2322 &voters_by_peer,
2323 witnessed_close_group_quorum(),
2324 )
2325 .expect("one quorum-supported quote is enough for SNP payment");
2326
2327 assert_eq!(quote_peer_seeds(&selected), vec![QUOTE_ISSUER_SEED]);
2328 let (median_peer_id, _) =
2329 median_paid_quote_issuer(&selected).expect("single quote is its own median");
2330 assert_eq!(median_peer_id, synthetic_peer(QUOTE_ISSUER_SEED));
2331 }
2332
2333 #[test]
2334 fn witnessed_quote_selection_rejects_median_without_witness_quorum() {
2335 const MEDIAN_ISSUER_SEED: u8 = 7;
2336
2337 let address = [0u8; 32];
2338 let quotes = vec![
2339 synthetic_quote(1, 10),
2340 synthetic_quote(2, 20),
2341 synthetic_quote(3, 30),
2342 synthetic_quote(6, 50),
2343 synthetic_quote(MEDIAN_ISSUER_SEED, 40),
2344 synthetic_quote(8, 60),
2345 synthetic_quote(9, 70),
2346 synthetic_quote(10, 80),
2347 ];
2348 let mut voters_by_peer = HashMap::new();
2349 voters_by_peer.insert(
2350 synthetic_peer(MEDIAN_ISSUER_SEED),
2351 synthetic_voters(&[1, 2, 3, 20]),
2352 );
2353
2354 let selected = select_witnessed_median_voter_quotes(
2355 quotes,
2356 &address,
2357 &voters_by_peer,
2358 witnessed_close_group_quorum(),
2359 );
2360
2361 assert!(
2362 selected.is_none(),
2363 "the selector must not return a paid quote set when fewer than the \
2364 witnessed median voter quorum recognised the paid median issuer"
2365 );
2366 }
2367
2368 #[test]
2369 fn put_peers_prioritise_median_voters_without_reordering_quotes() {
2370 const MEDIAN_ISSUER_SEED: u8 = 7;
2371
2372 let quotes = vec![
2373 synthetic_quote(1, 10),
2374 synthetic_quote(2, 20),
2375 synthetic_quote(3, 30),
2376 synthetic_quote(4, 50),
2377 synthetic_quote(5, 60),
2378 synthetic_quote(6, 70),
2379 synthetic_quote(MEDIAN_ISSUER_SEED, 40),
2380 ];
2381 let mut voters_by_peer = HashMap::new();
2382 voters_by_peer.insert(
2383 synthetic_peer(MEDIAN_ISSUER_SEED),
2384 synthetic_voters(&[3, 4, 5, 6, MEDIAN_ISSUER_SEED]),
2385 );
2386
2387 let put_candidates = put_peers_from_seeds(&[1, 2, 3, 4, 5, 6, 7]);
2388 let put_peers = put_peers_with_median_voters_first(
2389 "es,
2390 &put_candidates,
2391 &voters_by_peer,
2392 witnessed_close_group_quorum(),
2393 )
2394 .expect("median voters should produce an ordered PUT set");
2395
2396 assert_eq!(quote_peer_seeds("es), vec![1, 2, 3, 4, 5, 6, 7]);
2397 let (median_peer_id, _) =
2398 median_paid_quote_issuer("es).expect("selected quotes have a median");
2399 assert_eq!(median_peer_id, synthetic_peer(MEDIAN_ISSUER_SEED));
2400 assert_eq!(put_peer_seeds(&put_peers), vec![3, 4, 5, 6, 7, 1, 2]);
2401 }
2402
2403 #[test]
2404 fn filter_drops_only_bad_bindings_and_leaves_storer_acceptable_quotes() {
2405 let mut quotes = vec![
2406 good_quote_real(),
2407 bad_quote_real(),
2408 good_quote_real(),
2409 bad_quote_real(),
2410 good_quote_real(),
2411 ];
2412
2413 let dropped = drop_quotes_with_bad_bindings(&mut quotes);
2414
2415 assert_eq!(dropped, 2, "two crossed-key quotes must be dropped");
2416 assert_eq!(quotes.len(), 3, "three real-key quotes must remain");
2417
2418 for (peer_id, _, quote, _, _) in "es {
2424 assert!(
2425 storer_binding_would_accept(peer_id, quote),
2426 "every retained quote must satisfy the full storer-side spec"
2427 );
2428 }
2429 }
2430
2431 #[test]
2432 fn filter_is_noop_when_all_quotes_are_storer_acceptable() {
2433 let mut quotes: Vec<_> = (0..5).map(|_| good_quote_real()).collect();
2434 let before = quotes.len();
2435 let dropped = drop_quotes_with_bad_bindings(&mut quotes);
2436 assert_eq!(dropped, 0);
2437 assert_eq!(quotes.len(), before);
2438 for (peer_id, _, quote, _, _) in "es {
2439 assert!(storer_binding_would_accept(peer_id, quote));
2440 }
2441 }
2442
2443 #[test]
2444 fn filter_drops_all_when_every_responder_is_bad() {
2445 let mut quotes: Vec<_> = (0..fault_tolerant_quote_query_count())
2450 .map(|_| bad_quote_real())
2451 .collect();
2452 let dropped = drop_quotes_with_bad_bindings(&mut quotes);
2453 assert_eq!(dropped, fault_tolerant_quote_query_count());
2454 assert!(quotes.is_empty());
2455 }
2456
2457 #[test]
2458 fn filter_preserves_quote_payload_byte_for_byte() {
2459 let (peer_id, addrs, original_quote, amount, commitment) = good_quote_real();
2464 let mut quotes = vec![(
2465 peer_id,
2466 addrs.clone(),
2467 original_quote.clone(),
2468 amount,
2469 commitment,
2470 )];
2471 let _ = drop_quotes_with_bad_bindings(&mut quotes);
2472
2473 let (kept_peer, kept_addrs, kept_quote, kept_amount, _kept_commitment) =
2474 quotes.pop().expect("the good quote must survive filtering");
2475 assert_eq!(kept_peer.as_bytes(), peer_id.as_bytes());
2476 assert_eq!(kept_addrs.len(), addrs.len());
2477 assert_eq!(kept_amount, amount);
2478 assert_eq!(kept_quote.pub_key, original_quote.pub_key);
2479 assert_eq!(kept_quote.signature, original_quote.signature);
2480 assert_eq!(kept_quote.content.0, original_quote.content.0);
2481 assert_eq!(kept_quote.timestamp, original_quote.timestamp);
2482 assert_eq!(kept_quote.price, original_quote.price);
2483 assert_eq!(kept_quote.rewards_address, original_quote.rewards_address);
2484 }
2485
2486 #[test]
2517 fn repro_apr_30_storer_would_have_rejected_pre_filter_and_accepts_post_filter() {
2518 let over_query_count = fault_tolerant_quote_query_count();
2519 let mut quotes: Vec<_> = (0..over_query_count - 1)
2520 .map(|_| good_quote_real())
2521 .collect();
2522 quotes.insert(over_query_count / 2, bad_quote_real());
2525 assert_eq!(quotes.len(), over_query_count);
2526
2527 let storer_would_reject_count = quotes
2529 .iter()
2530 .filter(|(p, _, q, _, _)| !storer_binding_would_accept(p, q))
2531 .count();
2532 assert_eq!(
2533 storer_would_reject_count, 1,
2534 "exactly one quote (the crossed-key one) must be rejected by the storer spec"
2535 );
2536
2537 let dropped = drop_quotes_with_bad_bindings(&mut quotes);
2539 assert_eq!(dropped, 1, "exactly the crossed-key quote must be filtered");
2540
2541 for (peer_id, _, quote, _, _) in "es {
2543 assert!(
2544 storer_binding_would_accept(peer_id, quote),
2545 "every post-filter quote must be accepted by the storer spec — \
2546 this is what the filter guarantees before any quote set is used"
2547 );
2548 }
2549
2550 assert!(
2552 quotes.len() >= CLOSE_GROUP_SIZE,
2553 "after filtering, at least CLOSE_GROUP_SIZE good quotes must remain \
2554 so a fault-tolerant probe can still return a full close group"
2555 );
2556 }
2557
2558 #[test]
2563 fn filter_leaves_short_set_when_too_many_bad_peers() {
2564 let good_count = CLOSE_GROUP_SIZE - 1;
2565 let bad_count = fault_tolerant_quote_query_count() - good_count;
2566 let mut quotes: Vec<_> = std::iter::repeat_with(bad_quote_real)
2567 .take(bad_count)
2568 .chain(std::iter::repeat_with(good_quote_real).take(good_count))
2569 .collect();
2570
2571 let dropped = drop_quotes_with_bad_bindings(&mut quotes);
2572 assert_eq!(dropped, bad_count);
2573 assert!(
2574 quotes.len() < CLOSE_GROUP_SIZE,
2575 "this is the precondition for InsufficientPeers downstream"
2576 );
2577 for (peer_id, _, quote, _, _) in "es {
2579 assert!(storer_binding_would_accept(peer_id, quote));
2580 }
2581 }
2582
2583 fn serialize_quote(quote: &PaymentQuote) -> Vec<u8> {
2598 rmp_serde::to_vec(quote).expect("serialize quote")
2599 }
2600
2601 #[test]
2602 fn classifier_accepts_real_self_consistent_quote() {
2603 let content = [7u8; 32];
2606 let (peer_id, quote) = signed_baseline_quote(content);
2607 let bytes = serialize_quote("e);
2608 let result = classify_quote_response(&peer_id, &content, &bytes, false, None);
2609 match result {
2610 Ok((q, price, commitment)) => {
2611 assert_eq!(q.pub_key, quote.pub_key);
2612 assert_eq!(price, quote.price);
2613 assert!(commitment.is_none(), "baseline quote ships no commitment");
2614 }
2615 Err(e) => panic!("expected Ok, got {e}"),
2616 }
2617 }
2618
2619 #[test]
2620 fn classifier_rejects_quote_with_invalid_signature() {
2621 let content = [7u8; 32];
2624 let (peer_id, mut quote) = signed_baseline_quote(content);
2625 quote.signature = vec![0u8; quote.signature.len()]; let bytes = serialize_quote("e);
2627 let result = classify_quote_response(&peer_id, &content, &bytes, false, None);
2628 assert!(
2629 matches!(result, Err(Error::BadQuoteBinding { .. })),
2630 "a quote with an invalid signature must be rejected; got {result:?}"
2631 );
2632 }
2633
2634 #[test]
2635 fn classifier_rejects_quote_for_wrong_content() {
2636 let (peer_id, quote) = signed_baseline_quote([7u8; 32]);
2638 let bytes = serialize_quote("e);
2639 let result = classify_quote_response(&peer_id, &[9u8; 32], &bytes, false, None);
2640 assert!(
2641 matches!(result, Err(Error::BadQuoteBinding { .. })),
2642 "a quote for the wrong content must be rejected; got {result:?}"
2643 );
2644 }
2645
2646 #[test]
2647 fn classifier_rejects_crossed_keypair_with_typed_error() {
2648 let (peer_id, _, quote, _, _) = bad_quote_real();
2649 let bytes = serialize_quote("e);
2650 let result = classify_quote_response(&peer_id, &[0u8; 32], &bytes, false, None);
2651 match result {
2652 Err(Error::BadQuoteBinding {
2653 peer_id: pid,
2654 detail,
2655 }) => {
2656 assert_eq!(pid, peer_id.to_string());
2657 assert!(
2658 detail.contains("BLAKE3(pub_key)="),
2659 "diagnostic detail must include the derived peer id: {detail}"
2660 );
2661 }
2662 other => panic!("expected BadQuoteBinding for crossed-key quote, got {other:?}"),
2663 }
2664 }
2665
2666 #[test]
2677 fn classifier_rejects_already_stored_vote_from_bad_binding_peer() {
2678 let (peer_id, _, quote, _, _) = bad_quote_real();
2679 let bytes = serialize_quote("e);
2680 let result = classify_quote_response(&peer_id, &[0u8; 32], &bytes, true, None);
2682 assert!(
2683 matches!(result, Err(Error::BadQuoteBinding { .. })),
2684 "crossed-key peer must be classified BadQuoteBinding even when \
2685 voting already_stored=true; got {result:?}"
2686 );
2687 }
2688
2689 #[test]
2692 fn classifier_honours_already_stored_vote_from_good_binding_peer() {
2693 let content = [7u8; 32];
2694 let (peer_id, quote) = signed_baseline_quote(content);
2695 let bytes = serialize_quote("e);
2696 let result = classify_quote_response(&peer_id, &content, &bytes, true, None);
2697 assert!(
2698 matches!(result, Err(Error::AlreadyStored)),
2699 "honest peer's already_stored vote must be honoured; got {result:?}"
2700 );
2701 }
2702
2703 #[test]
2704 fn classifier_returns_serialization_error_on_bad_bytes() {
2705 let (peer_id, _, _, _, _) = good_quote_real();
2706 let garbage = b"this is not a valid msgpack PaymentQuote".to_vec();
2707 let result = classify_quote_response(&peer_id, &[0u8; 32], &garbage, false, None);
2708 assert!(
2709 matches!(result, Err(Error::Serialization(_))),
2710 "garbage bytes must produce a Serialization error; got {result:?}"
2711 );
2712 }
2713
2714 #[test]
2717 fn classifier_verdict_matches_storer_binding_spec_for_mixed_responders() {
2718 let content = [7u8; 32];
2719 let mut responders: Vec<(PeerId, PaymentQuote)> =
2720 (0..12).map(|_| signed_baseline_quote(content)).collect();
2721 for _ in 0..4 {
2722 let (p, _, q, _, _) = bad_quote_real();
2723 responders.push((p, q));
2724 }
2725
2726 for (peer_id, quote) in &responders {
2727 let bytes = serialize_quote(quote);
2728 let storer_verdict = storer_binding_would_accept(peer_id, quote);
2729 let classifier_verdict =
2730 classify_quote_response(peer_id, &content, &bytes, false, None).is_ok();
2731 assert_eq!(
2732 classifier_verdict, storer_verdict,
2733 "classifier and storer-binding-spec must agree on every responder \
2734 (peer_id={}, storer={storer_verdict}, classifier={classifier_verdict})",
2735 peer_id
2736 );
2737 }
2738 }
2739
2740 fn any_peer() -> PeerId {
2756 PeerId::from_bytes([0u8; 32])
2757 }
2758
2759 fn quote_with_binding(
2761 committed_key_count: u32,
2762 commitment_pin: Option<[u8; 32]>,
2763 price: Amount,
2764 ) -> PaymentQuote {
2765 PaymentQuote {
2766 content: XorName([0u8; 32]),
2767 timestamp: SystemTime::UNIX_EPOCH,
2768 price,
2769 rewards_address: RewardsAddress::new([0u8; 20]),
2770 pub_key: Vec::new(),
2771 signature: Vec::new(),
2772 committed_key_count,
2773 commitment_pin,
2774 }
2775 }
2776
2777 fn signed_commitment(kp: &Keypair, root: [u8; 32], key_count: u32) -> StorageCommitment {
2783 use ant_protocol::payment::commitment::DOMAIN_COMMITMENT;
2784 use ant_protocol::pqc::api::{ml_dsa_65, MlDsaSecretKey as ApiSecretKey, MlDsaVariant};
2785 let peer = compute_address(&kp.pub_key_bytes);
2786 let mut payload = Vec::with_capacity(32 + 4 + 32 + 4 + kp.pub_key_bytes.len());
2787 payload.extend_from_slice(&root);
2788 payload.extend_from_slice(&key_count.to_le_bytes());
2789 payload.extend_from_slice(&peer);
2790 payload.extend_from_slice(&(kp.pub_key_bytes.len() as u32).to_le_bytes());
2791 payload.extend_from_slice(&kp.pub_key_bytes);
2792 let sk = ApiSecretKey::from_bytes(MlDsaVariant::MlDsa65, &kp.secret_key_bytes)
2793 .expect("api secret key");
2794 let signature = ml_dsa_65()
2795 .sign_with_context(&sk, &payload, DOMAIN_COMMITMENT)
2796 .expect("sign commitment")
2797 .to_bytes();
2798 StorageCommitment {
2799 root,
2800 key_count,
2801 sender_peer_id: peer,
2802 sender_public_key: kp.pub_key_bytes.clone(),
2803 signature,
2804 }
2805 }
2806
2807 #[test]
2808 fn binding_baseline_ok_only_at_baseline_price() {
2809 let q = quote_with_binding(0, None, calculate_price(0));
2811 assert!(quote_commitment_binding_is_valid(&any_peer(), &q, &None).is_ok());
2812
2813 let q = quote_with_binding(0, None, calculate_price(500));
2816 assert!(quote_commitment_binding_is_valid(&any_peer(), &q, &None).is_err());
2817 }
2818
2819 #[test]
2820 fn binding_rejects_incoherent_shapes() {
2821 let q = quote_with_binding(500, None, calculate_price(500));
2823 assert!(quote_commitment_binding_is_valid(&any_peer(), &q, &None).is_err());
2824 let q = quote_with_binding(0, Some([9u8; 32]), calculate_price(0));
2826 assert!(quote_commitment_binding_is_valid(&any_peer(), &q, &None).is_err());
2827 }
2828
2829 #[test]
2830 fn binding_rejects_count_above_cap() {
2831 let over = MAX_COMMITMENT_KEY_COUNT + 1;
2832 let q = quote_with_binding(over, Some([9u8; 32]), calculate_price(over as usize));
2833 assert!(
2834 quote_commitment_binding_is_valid(&any_peer(), &q, &Some(vec![1u8; 16])).is_err(),
2835 "a count above MAX_COMMITMENT_KEY_COUNT must be rejected before payment"
2836 );
2837 }
2838
2839 #[test]
2840 fn binding_rejects_on_curve_wrong_count() {
2841 let q = quote_with_binding(500, Some([9u8; 32]), calculate_price(499));
2844 assert!(quote_commitment_binding_is_valid(&any_peer(), &q, &Some(vec![1u8; 16])).is_err());
2845 }
2846
2847 #[test]
2848 fn binding_rejects_bound_quote_without_shipped_commitment() {
2849 let q = quote_with_binding(500, Some([9u8; 32]), calculate_price(500));
2852 assert!(
2853 quote_commitment_binding_is_valid(&any_peer(), &q, &None).is_err(),
2854 "a bound quote missing its commitment must be rejected"
2855 );
2856 }
2857
2858 #[test]
2859 fn binding_rejects_unparseable_and_peer_unbound_commitment() {
2860 let q = quote_with_binding(500, Some([9u8; 32]), calculate_price(500));
2863 assert!(
2864 quote_commitment_binding_is_valid(&any_peer(), &q, &Some(vec![0xFF; 8])).is_err(),
2865 "an unparseable commitment must be rejected before payment"
2866 );
2867
2868 let bogus = StorageCommitment {
2874 root: [1u8; 32],
2875 key_count: 500,
2876 sender_peer_id: [2u8; 32], sender_public_key: vec![3u8; 1952],
2878 signature: vec![4u8; 3293],
2879 };
2880 let blob = rmp_serde::to_vec(&bogus).expect("serialize bogus commitment");
2881 assert!(
2882 quote_commitment_binding_is_valid(&any_peer(), &q, &Some(blob)).is_err(),
2883 "a commitment not bound to the quoting peer must be rejected before payment"
2884 );
2885 }
2886
2887 #[test]
2888 fn binding_rejects_commitment_with_invalid_signature() {
2889 let kp = gen_keypair();
2893 let mut commitment = signed_commitment(&kp, [6u8; 32], 500);
2894 commitment.signature[0] ^= 0xFF; let pin = commitment_hash(&commitment).expect("hash");
2898 let blob = rmp_serde::to_vec(&commitment).expect("serialize commitment");
2899 let q = quote_with_binding(500, Some(pin), calculate_price(500));
2900 let res = quote_commitment_binding_is_valid(&kp.peer_id, &q, &Some(blob));
2901 let err = res.expect_err("commitment with an invalid signature must be rejected");
2902 assert!(
2903 err.contains("signature"),
2904 "should fail at the signature check: {err}"
2905 );
2906 }
2907
2908 #[test]
2909 fn binding_rejects_commitment_that_does_not_hash_to_pin() {
2910 let kp = gen_keypair();
2914 let commitment = signed_commitment(&kp, [5u8; 32], 500);
2915 let wrong_pin = [0xAB; 32];
2916 assert_ne!(commitment_hash(&commitment), Some(wrong_pin));
2917 let blob = rmp_serde::to_vec(&commitment).expect("serialize commitment");
2918 let q = quote_with_binding(500, Some(wrong_pin), calculate_price(500));
2919 let res = quote_commitment_binding_is_valid(&kp.peer_id, &q, &Some(blob));
2920 let err = res.expect_err("commitment that does not hash to the pin must be rejected");
2921 assert!(
2922 err.contains("hash"),
2923 "should fail at the hash==pin check: {err}"
2924 );
2925 }
2926
2927 #[test]
2928 fn binding_rejects_count_disagreeing_with_commitment() {
2929 let kp = gen_keypair();
2934 let commitment = signed_commitment(&kp, [7u8; 32], 400);
2935 let pin = commitment_hash(&commitment).expect("hash");
2936 let blob = rmp_serde::to_vec(&commitment).expect("serialize commitment");
2937 let q = quote_with_binding(500, Some(pin), calculate_price(500));
2938 let res = quote_commitment_binding_is_valid(&kp.peer_id, &q, &Some(blob));
2939 let err = res.expect_err("a quote count disagreeing with the commitment must be rejected");
2940 assert!(
2941 err.contains("key_count") || err.contains("attests"),
2942 "should fail at the count==key_count check: {err}"
2943 );
2944 }
2945
2946 #[test]
2947 fn binding_rejects_oversized_commitment_before_parsing() {
2948 let q = quote_with_binding(500, Some([9u8; 32]), calculate_price(500));
2951 let huge = Some(vec![0u8; MAX_COMMITMENT_SIDECAR_BYTES + 1]);
2952 assert!(
2953 quote_commitment_binding_is_valid(&any_peer(), &q, &huge).is_err(),
2954 "an oversized commitment blob must be rejected before payment"
2955 );
2956 }
2957
2958 #[test]
2959 fn classifier_drops_off_curve_quote_with_typed_error() {
2960 use ant_protocol::pqc::ops::MlDsaSecretKey;
2965 let content = [7u8; 32];
2966 let kp = gen_keypair();
2967 let mut quote = PaymentQuote {
2968 content: XorName(content),
2969 timestamp: SystemTime::UNIX_EPOCH,
2970 price: calculate_price(500),
2972 rewards_address: RewardsAddress::new([0u8; 20]),
2973 pub_key: kp.pub_key_bytes.clone(),
2974 signature: Vec::new(),
2975 committed_key_count: 0,
2976 commitment_pin: None,
2977 };
2978 let ml_dsa = MlDsa65::new();
2979 let sk = MlDsaSecretKey::from_bytes(&kp.secret_key_bytes).expect("sk");
2980 quote.signature = ml_dsa
2981 .sign(&sk, "e.bytes_for_sig())
2982 .expect("sign")
2983 .as_bytes()
2984 .to_vec();
2985 let bytes = serialize_quote("e);
2986 let result = classify_quote_response(&kp.peer_id, &content, &bytes, false, None);
2987 assert!(
2988 matches!(result, Err(Error::BadQuoteCommitment { .. })),
2989 "off-curve quote must be dropped as BadQuoteCommitment; got {result:?}"
2990 );
2991 }
2992
2993 #[test]
2998 fn an_update_refusal_is_surfaced_with_its_upgrade_instruction() {
2999 let peer_id = PeerId::from_bytes([0x42; 32]);
3000 let refusal = ProtocolError::ClientUpdateRequired {
3001 client_settlement_version: CURRENT_SETTLEMENT_VERSION,
3002 min_settlement_version: CURRENT_SETTLEMENT_VERSION.saturating_add(1),
3003 };
3004
3005 let mapped = map_quote_response(
3006 &peer_id,
3007 &[0x11; 32],
3008 ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error(refusal)),
3009 );
3010
3011 match mapped {
3012 Some(Err(Error::ClientUpdateRequired(msg))) => {
3013 assert!(msg.contains("ant update"), "{msg}");
3014 assert!(msg.contains("nothing was charged"), "{msg}");
3015 }
3016 other => panic!("expected ClientUpdateRequired, got: {other:?}"),
3017 }
3018 }
3019
3020 #[test]
3026 fn only_silence_triggers_the_legacy_retry() {
3027 assert!(is_version_unaware(&Error::Timeout("no answer".into())));
3028 assert!(is_version_unaware(&Error::Network("send failed".into())));
3029
3030 assert!(!is_version_unaware(&Error::ClientUpdateRequired(
3031 "too old".into()
3032 )));
3033 assert!(!is_version_unaware(&Error::StorerUpdateRequired(
3037 "node behind".into()
3038 )));
3039 assert!(!is_version_unaware(&Error::Protocol(
3040 "quote error from peer".into()
3041 )));
3042 }
3043
3044 #[test]
3049 fn a_storer_that_is_behind_is_not_reported_as_the_clients_fault() {
3050 let peer_id = PeerId::from_bytes([0x43; 32]);
3051 let mapped = map_quote_response(
3052 &peer_id,
3053 &[0x11; 32],
3054 ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error(
3055 ProtocolError::StorerUpdateRequired {
3056 client_settlement_version: 2,
3057 node_settlement_version: 1,
3058 },
3059 )),
3060 );
3061
3062 match mapped {
3063 Some(Err(Error::StorerUpdateRequired(msg))) => {
3064 assert!(msg.contains("use a different storer"), "{msg}");
3065 assert!(!msg.contains("ant update"), "{msg}");
3066 }
3067 other => panic!("expected StorerUpdateRequired, got: {other:?}"),
3068 }
3069 }
3070
3071 #[test]
3076 fn a_refusal_aborts_quote_collection_instead_of_counting_as_one_bad_peer() {
3077 let mut quotes = Vec::new();
3078 let mut already_stored = Vec::new();
3079 let mut failures = Vec::new();
3080 let mut bad_quotes = 0usize;
3081 let mut refusal_slot: Option<Error> = None;
3082
3083 let refusals = SettlementRefusals::default();
3084 let mut refuse =
3085 |peer: u8, failures: &mut Vec<String>, slot: &mut Option<Error>| -> Result<()> {
3086 record_store_quote_result(
3087 PeerId::from_bytes([peer; 32]),
3088 Vec::new(),
3089 Err(Error::ClientUpdateRequired(
3090 "too old, run ant update".into(),
3091 )),
3092 &[0x11; 32],
3093 &mut quotes,
3094 &mut already_stored,
3095 failures,
3096 &mut bad_quotes,
3097 slot,
3098 &refusals,
3099 )
3100 };
3101
3102 let first = refuse(0x44, &mut failures, &mut refusal_slot);
3105 assert!(first.is_ok(), "one peer must not abort, got {first:?}");
3106 assert_eq!(failures.len(), 1);
3107 assert!(refusal_slot.is_none());
3108
3109 let second = refuse(0x45, &mut failures, &mut refusal_slot);
3111 assert!(
3112 matches!(second, Err(Error::ClientUpdateRequired(_))),
3113 "a corroborated refusal must propagate, got {second:?}"
3114 );
3115 }
3116
3117 #[test]
3120 fn a_storer_that_is_behind_does_not_abort_collection() {
3121 let mut quotes = Vec::new();
3122 let mut already_stored = Vec::new();
3123 let mut failures = Vec::new();
3124 let mut bad_quotes = 0usize;
3125 let mut refusal_slot: Option<Error> = None;
3126
3127 let outcome = record_store_quote_result(
3128 PeerId::from_bytes([0x45; 32]),
3129 Vec::new(),
3130 Err(Error::StorerUpdateRequired("node behind".into())),
3131 &[0x11; 32],
3132 &mut quotes,
3133 &mut already_stored,
3134 &mut failures,
3135 &mut bad_quotes,
3136 &mut refusal_slot,
3137 &SettlementRefusals::default(),
3138 );
3139
3140 assert!(
3141 outcome.is_ok(),
3142 "a lagging storer must be skipped, not fatal"
3143 );
3144 assert_eq!(failures.len(), 1, "and it should be recorded as a skip");
3145 assert!(
3146 refusal_slot.is_none(),
3147 "a node being behind is not a verdict about this client"
3148 );
3149 }
3150
3151 #[test]
3160 fn a_refusal_is_recorded_where_the_collection_timeout_cannot_discard_it() {
3161 let mut quotes = Vec::new();
3162 let mut already_stored = Vec::new();
3163 let mut failures = Vec::new();
3164 let mut bad_quotes = 0usize;
3165 let mut refusal_slot: Option<Error> = None;
3166
3167 let refusals = SettlementRefusals::default();
3168 for peer in [0x46u8, 0x47u8] {
3169 let _ = record_store_quote_result(
3170 PeerId::from_bytes([peer; 32]),
3171 Vec::new(),
3172 Err(Error::ClientUpdateRequired(
3173 "too old, run ant update".into(),
3174 )),
3175 &[0x11; 32],
3176 &mut quotes,
3177 &mut already_stored,
3178 &mut failures,
3179 &mut bad_quotes,
3180 &mut refusal_slot,
3181 &refusals,
3182 );
3183 }
3184
3185 match refusal_slot {
3186 Some(Error::ClientUpdateRequired(msg)) => {
3187 assert!(msg.contains("ant update"), "{msg}");
3188 }
3189 other => panic!("refusal must outlive the collection state, got {other:?}"),
3190 }
3191 }
3192
3193 #[test]
3200 fn meeting_the_target_stops_launching_without_stopping_collection() {
3201 assert!(
3202 witnessed_quote_launch_budget(0, 0, 32) > 0,
3203 "collection must start"
3204 );
3205 assert_eq!(witnessed_quote_launch_budget(CLOSE_GROUP_SIZE, 0, 32), 0);
3206 assert_eq!(
3207 witnessed_quote_launch_budget(CLOSE_GROUP_SIZE.saturating_add(1), 0, 32),
3208 0
3209 );
3210 assert_eq!(witnessed_quote_launch_budget(0, CLOSE_GROUP_SIZE, 32), 0);
3213 }
3214
3215 #[test]
3229 fn a_peer_that_cannot_answer_a_versioned_quote_is_only_probed_once() {
3230 let peers: Arc<Mutex<HashSet<PeerId>>> = Arc::new(Mutex::new(HashSet::new()));
3231 let legacy_peer = PeerId::from_bytes([0x51; 32]);
3232 let fresh_peer = PeerId::from_bytes([0x52; 32]);
3233
3234 let known = |p: &PeerId| peers.lock().expect("cache lock").contains(p);
3235
3236 assert!(!known(&legacy_peer));
3238
3239 peers.lock().expect("cache lock").insert(legacy_peer);
3241
3242 assert!(known(&legacy_peer));
3244 assert!(!known(&fresh_peer));
3246 }
3247
3248 #[test]
3256 fn only_silence_is_evidence_worth_caching() {
3257 assert!(matches!(
3258 Error::Timeout("no answer".into()),
3259 Error::Timeout(_)
3260 ));
3261 assert!(!matches!(
3262 Error::Network("send failed".into()),
3263 Error::Timeout(_)
3264 ));
3265 assert!(is_version_unaware(&Error::Network("send failed".into())));
3268 assert!(is_version_unaware(&Error::Timeout("no answer".into())));
3269 }
3270
3271 #[test]
3278 fn a_lone_peer_cannot_condemn_the_client() {
3279 let refusals = SettlementRefusals::default();
3280 assert!(
3281 refusals
3282 .note(PeerId::from_bytes([0x61; 32]), "too old")
3283 .is_none(),
3284 "one peer is not corroboration"
3285 );
3286 assert!(refusals.corroborated().is_none());
3287 assert!(refusals
3289 .note(PeerId::from_bytes([0x61; 32]), "too old")
3290 .is_none());
3291 assert!(refusals.corroborated().is_none());
3292 }
3293
3294 #[test]
3297 fn a_second_peer_makes_the_refusal_terminal_and_it_stays_latched() {
3298 let refusals = SettlementRefusals::default();
3299 refusals.note(PeerId::from_bytes([0x62; 32]), "run ant update");
3300 let verdict = refusals.note(PeerId::from_bytes([0x63; 32]), "run ant update");
3301
3302 assert!(verdict.is_some_and(|m| m.contains("ant update")));
3303 assert!(refusals
3307 .corroborated()
3308 .is_some_and(|m| m.contains("ant update")));
3309 }
3310
3311 #[test]
3315 fn an_incoherent_refusal_is_treated_as_a_bad_peer() {
3316 let peer_id = PeerId::from_bytes([0x64; 32]);
3317
3318 let wrong_echo = settlement_refusal_error(
3320 &peer_id,
3321 CURRENT_SETTLEMENT_VERSION.saturating_add(7),
3322 CURRENT_SETTLEMENT_VERSION.saturating_add(8),
3323 );
3324 assert!(matches!(wrong_echo, Error::Protocol(_)), "{wrong_echo:?}");
3325
3326 let no_gap = settlement_refusal_error(
3328 &peer_id,
3329 CURRENT_SETTLEMENT_VERSION,
3330 CURRENT_SETTLEMENT_VERSION,
3331 );
3332 assert!(matches!(no_gap, Error::Protocol(_)), "{no_gap:?}");
3333
3334 let real = settlement_refusal_error(
3336 &peer_id,
3337 CURRENT_SETTLEMENT_VERSION,
3338 CURRENT_SETTLEMENT_VERSION.saturating_add(1),
3339 );
3340 match real {
3341 Error::ClientUpdateRequired(msg) => assert!(msg.contains("ant update"), "{msg}"),
3342 other => panic!("expected ClientUpdateRequired, got {other:?}"),
3343 }
3344 }
3345}