Skip to main content

ant_core/data/client/
quote.rs

1//! Quote and payment operations.
2//!
3//! Handles requesting storage quotes from network nodes and
4//! managing payment for data storage.
5
6use 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
33/// Fault-tolerant quote collection asks one extra close group of peers and
34/// keeps the closest successful `CLOSE_GROUP_SIZE` responders. This remains
35/// useful for merkle preflight probes, but single-node payments deliberately
36/// ask only the actual close group.
37const FAULT_TOLERANT_QUOTE_QUERY_MULTIPLIER: usize = 2;
38
39/// Witnessed close-group quorum as a fraction of the initial close group.
40/// For today's `CLOSE_GROUP_SIZE = 7`, this yields the requested 5-of-7
41/// quorum.
42const WITNESSED_QUORUM_NUMERATOR: usize = 2;
43const WITNESSED_QUORUM_DENOMINATOR: usize = 3;
44
45/// Number of closest nodes each initial witnessed responder contributes.
46const SINGLE_NODE_WITNESSED_VIEW_COUNT: usize = 20;
47
48/// Minimum quote count accepted by the single-node payment path.
49const SINGLE_NODE_MIN_QUOTE_COUNT: usize = 1;
50
51/// Overall timeout for collecting quote responses. Must accommodate
52/// connect_with_fallback cascade (direct 5s + hole-punch 15s×3 + relay 30s ≈
53/// 80s) plus the per-peer quote timeout.
54const QUOTE_COLLECTION_TIMEOUT_SECS: u64 = 120;
55
56/// ML-DSA-65 public key length in bytes. Mirrors the same value defined as
57/// `pub const ML_DSA_65_PUBLIC_KEY_SIZE` in `saorsa-pqc::pqc::types`, which
58/// the storer's `peer_id_from_public_key_bytes` enforces. We keep a local
59/// copy here rather than adding a direct `saorsa-pqc` dep — the constant
60/// is FIPS-mandated for ML-DSA-65 and won't change unless we change variant.
61///
62/// TODO: switch to `saorsa_pqc::pqc::types::ML_DSA_65_PUBLIC_KEY_SIZE` once
63/// `ant-protocol` re-exports it (`pqc::ops::ML_DSA_65_PUBLIC_KEY_SIZE`).
64const ML_DSA_PUB_KEY_LEN: usize = 1952;
65
66/// One collected quote: the responding peer, its addresses, the signed quote,
67/// the price it demands, and (ADR-0004) the opaque signed-commitment blob the
68/// node shipped alongside the quote (`None` for a baseline quote), to be
69/// forwarded as a sidecar in the PUT bundle.
70type QuotedPeer = (
71    PeerId,
72    Vec<MultiAddr>,
73    PaymentQuote,
74    Amount,
75    Option<Vec<u8>>,
76);
77
78/// Check that a quote's `pub_key` is well-formed and BLAKE3-hashes to the
79/// claimed `peer_id`.
80///
81/// The storer node enforces both constraints in `ant-node/src/payment/verifier.rs`
82/// (via `peer_id_from_public_key_bytes` and `validate_peer_bindings`): every
83/// quote inside a `ProofOfPayment` must (a) have a 1952-byte `pub_key` parsable
84/// as ML-DSA-65 and (b) satisfy `BLAKE3(pub_key) == peer_id`. A single quote
85/// failing either check causes the storer to reject the entire close-group
86/// proof and burn the chunk's payment.
87///
88/// This is the cheap structural pre-check. ADR-0004 additionally has the client
89/// run `verify_quote_content` + `verify_quote_signature` (the full ML-DSA check)
90/// in [`classify_quote_response`] before paying, so a quote the storer would
91/// reject never gets paid.
92fn 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(&quote.pub_key) == *peer_id.as_bytes()
97}
98
99/// ADR-0004 client-side resolve-before-pay gate — "the client pays nothing it
100/// cannot resolve", the ceiling's load-bearing wall (ADR §"The client pays
101/// nothing it cannot resolve").
102///
103/// Runs the **full** binding check before paying, identical to the storer's,
104/// using the shared `ant-protocol` commitment type + verifier so client and
105/// node can never disagree:
106/// 1. **Shape.** `(0, None)` baseline or `(n>0, Some(pin))` bound; the mixed
107///    shapes `(n>0, None)` (unauditable count) and `(0, Some)` (incoherent
108///    baseline) are rejected.
109/// 2. **Cap.** `committed_key_count <= MAX_COMMITMENT_KEY_COUNT` — a count a
110///    commitment could never legitimately attest is rejected before pricing.
111/// 3. **Forced price.** `price == calculate_price(committed_key_count)`, by
112///    exact recomputation with the shared `calculate_price` — never inverted.
113/// 4. **Resolution (bound quotes).** The shipped commitment must: parse as a
114///    `StorageCommitment`, be bound to the quoting peer
115///    (`BLAKE3(sender_public_key) == sender_peer_id`), have a valid ML-DSA-65
116///    signature, hash to the quote's `commitment_pin`
117///    (`commitment_hash == pin`), and attest exactly the claimed count
118///    (`key_count == committed_key_count`). A withheld, unparseable, wrong-pin,
119///    mis-bound, or count-mismatched commitment is unresolvable → the quote is
120///    dropped before payment.
121///
122/// Returns `Ok(())` if the binding fully resolves, or `Err(detail)` naming the
123/// rule that failed.
124fn 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    // Forced price: exact recomputation, never inversion.
148    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    // Baseline `(0, None)` pins nothing — fully resolved by the checks above.
157    let Some(pin) = pin else {
158        return Ok(());
159    };
160
161    // Bound quote: the commitment MUST have arrived and MUST resolve the pin.
162    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    // Cap before parsing: bound the deserialize work a malicious responder can
170    // force, and never forward an oversized blob in the PUT bundle.
171    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    // Peer binding: the commitment must belong to the quoting peer, exactly as
182    // the storer derives a candidate's peer id (`BLAKE3(pub_key)`).
183    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
203/// Classification of a `ChunkQuoteResponse::Success` body for a single peer.
204///
205/// Mirrors the storer-side `validate_peer_bindings` check from
206/// `ant-node/src/payment/verifier.rs` — the cheap BLAKE3 binding —
207/// so we drop misbehaving peers' quotes before payment.
208///
209/// ADR-0004: the client now ALSO runs the storer's `verify_quote_content` and
210/// `verify_quote_signature` (ML-DSA-65) before paying, so "the client pays
211/// nothing it cannot resolve" covers the quote's own validity too, not just the
212/// commitment binding. This matches what the merkle path already does
213/// client-side and costs ~1 ms × CLOSE_GROUP_SIZE per chunk — accepted, since
214/// paying a quote the storer then rejects burns the on-chain payment.
215///
216/// Pulling the logic out of the async closure lets us unit-test the
217/// primary defense (not just the post-collect defensive filter).
218///
219/// # Returns
220///
221/// - `Ok((quote, price))` — the response is honoured as a quote.
222/// - `Err(Error::AlreadyStored)` — the peer claims the chunk is already
223///   present AND the quote it provided binds to its peer ID. Vote counts.
224/// - `Err(Error::BadQuoteBinding { .. })` — bad binding (mirrors the
225///   storer-side rejection). Outer collector counts these via the typed
226///   variant (no string matching).
227/// - `Err(Error::BadQuoteCommitment { .. })` — ADR-0004 forced-price binding
228///   failed (price off the curve, incoherent shape, or a bound quote that did
229///   not ship its commitment); dropped before payment like a bad binding.
230/// - `Err(Error::Serialization(...))` — the quote bytes did not deserialize.
231///
232/// On success the returned commitment is the opaque signed-commitment blob the
233/// node shipped with the quote (`None` for a baseline quote), to be forwarded
234/// as a sidecar in the PUT bundle.
235///
236/// A quote as this module hands it on: the quote itself, the price to settle,
237/// and the opaque signed commitment it was priced against.
238type 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    // Peer binding: BLAKE3(pub_key) must equal peer_id. This is the
251    // exact mitigation Chris and the AI investigation requested for the
252    // 2026-04-30 production failure: drop crossed-key peers before they
253    // poison the close-group ProofOfPayment.
254    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    // ADR-0004 "the client runs the full binding check": verify the quote's OWN
272    // ML-DSA-65 signature and that it is for THIS content, before paying —
273    // exactly what the storer checks and what the merkle path already does
274    // client-side. A quote with a valid pub_key binding but a bad signature or
275    // wrong content would otherwise be paid and then rejected by the storer.
276    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    // ADR-0004 forced-price gate: drop a quote whose price is not exactly the
290    // public formula of its committed count, whose (count, pin) shape is
291    // incoherent, or which is bound but did not ship its commitment. The storer
292    // re-runs the arithmetic and would reject the bundle; we drop it here so we
293    // never pay a quote we cannot resolve.
294    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
311/// Drop quotes whose `pub_key` does not BLAKE3-hash to the peer that supplied
312/// them. Logs each dropped quote at WARN.
313fn 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    // A peer that already failed to answer a versioned request is asked in the
350    // legacy shape directly. Re-probing costs a full per-peer timeout every
351    // time, and against a fleet that predates the versioned requests that is
352    // paid on every quote: measured on the merkle E2E suite it took the run
353    // from ~24 minutes to past the 60-minute cap.
354    // The capable set wins. The two sets are updated under separate locks, so
355    // a slow probe can insert into the legacy set after a concurrent request
356    // has already proved the peer capable; letting capability win makes that
357    // interleaving harmless instead of permanently preferring the legacy shape.
358    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    // Declare the settlement version so a storer can turn us away before we
366    // pay. See `merkle.rs` for why the legacy retry below exists and when it
367    // can be deleted.
368    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    // A first-contact probe waits only long enough to learn whether the peer
391    // can parse the shape; a peer already known to be legacy is asked with the
392    // caller's full patience because that request is the real one.
393    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    // Any recognised answer proves the peer parsed the versioned shape,
413    // including a structured error. Only silence and send failures leave the
414    // question open.
415    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    // Only a storer that could not decode the versioned request is asked
426    // again in the legacy shape. One that answered has understood us, and a
427    // refusal must stay a refusal.
428    let result = match result {
429        Err(ref e) if is_version_unaware(e) && !known_legacy => {
430            // Remember it, so the next request to this peer skips the probe.
431            // Only silence counts as evidence: a send failure means the
432            // request never arrived, which says nothing about whether the peer
433            // could have parsed it, and caching that would strand a peer in
434            // the legacy shape for the rest of the session over one flaky send.
435            // A peer that has answered a versioned request before is never
436            // demoted: one lost response would otherwise pin an upgraded peer
437            // to the legacy shape for the rest of the session.
438            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
480/// Turn a quote response into the quote it carries, or the error explaining
481/// why there is none. Shared by the versioned request and its legacy retry.
482///
483/// `ClientUpdateRequired` is separated from the generic protocol error because
484/// it is terminal: it must reach the user with its own wording rather than
485/// being counted as one more peer that failed to quote.
486fn 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            &quote,
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
523/// Turn a peer's `ClientUpdateRequired` into an error, rejecting one that does
524/// not describe this client.
525///
526/// A refusal is unauthenticated, so the least this client can do is check that
527/// the peer is talking about the request it actually sent: the echoed version
528/// must be ours, and the stated minimum must genuinely exceed it. A peer that
529/// fails either is confused or lying, and is treated as an ordinary bad peer
530/// rather than as evidence about this build.
531pub(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
551/// Did this failure look like a storer that cannot parse a versioned request,
552/// as opposed to one that parsed it and refused? See the merkle path for the
553/// full reasoning.
554///
555/// This is the single-node fallback, independent of the merkle one, so it
556/// carries its own reference to the shared cutover guard. Without it, deleting
557/// the merkle fallback would take the only build-time check with it and leave
558/// this downgrade path live.
559const _: () = 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/// Fold one peer's quote result into the collection state.
566///
567/// Returns `Err` only when collection must stop outright: a storer has said
568/// this client cannot settle, and no number of further quotes makes paying
569/// safe.
570#[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        // A storer that has explicitly declared this client incompatible ends
593        // quote collection here. Recording it as one more failed peer would
594        // let the remaining peers supply a quorum and go on to pay, which is
595        // the burn this mechanism exists to prevent, and would bury the
596        // upgrade instruction among ordinary per-peer failures.
597        // Recorded as well as returned. The caller runs this inside an overall
598        // timeout, and if that timeout fires the returned error is thrown away
599        // with the rest of the collection state. The verdict must outlive it:
600        // a client told it cannot settle must not pay, whatever else happened
601        // during collection.
602        Err(e @ Error::ClientUpdateRequired(_)) => {
603            // One peer's word is not a verdict about this build: nothing
604            // authenticates a refusal, so a single hostile peer answering this
605            // to everything would deny every upload. Believe it once enough
606            // distinct peers agree, which a genuine incompatibility reaches at
607            // once because every enforcing peer refuses.
608            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            // Name the corroborators. This is the only place the quorum is
614            // externally verifiable: the below-quorum branch above logs one
615            // peer at a time, so without this line a reader cannot tell a
616            // two-peer verdict from a one-peer misfire (V2-1109).
617            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
679/// Restrict a witnessed transcript to its closest `CLOSE_GROUP_SIZE` peers.
680///
681/// The witnessed query is widened to `PUT_TARGET_WIDTH` peers so we
682/// have addresses for the full PUT-target set, but the consensus/quorum/quote
683/// logic must still run on the close group only. Keeping just the closest-7
684/// initial peers and the responder views contributed by those peers leaves the
685/// `missing_witnessed_responder_views` math — and the quorum derived from it —
686/// byte-for-byte identical to a `CLOSE_GROUP_SIZE`-wide query.
687fn 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
713/// One collected store quote, carrying (ADR-0004) the opaque signed-commitment
714/// sidecar the node shipped with its quote (`None` for a baseline quote), to be
715/// forwarded in the PUT bundle and cross-checked by storers.
716pub(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, _) = &quotes[*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(&quotes, 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    /// Get storage quotes from the closest peers for a given address.
1132    ///
1133    /// Builds a quorum-witnessed candidate set, still attempts to collect the
1134    /// close-group quote count, and returns the largest supported successful
1135    /// quote set. The single-node path now only requires one valid quote to
1136    /// proceed, but still pays the median quote from the selected set when more
1137    /// quotes were successfully fetched.
1138    ///
1139    /// Returns `Error::AlreadyStored` early if `CLOSE_GROUP_MAJORITY` peers
1140    /// report the chunk is already stored.
1141    ///
1142    /// # Errors
1143    ///
1144    /// Returns an error if insufficient quotes can be collected.
1145    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    /// Get storage quotes plus PUT targets ordered for paid-median acceptance.
1158    ///
1159    /// Quote order is preserved for proof construction because tied quote
1160    /// prices rely on stable median selection. PUT target order is separate:
1161    /// peers that voted for the paid median issuer are placed first so the
1162    /// initial write wave is locally acceptable to a storage majority.
1163    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            &quotes,
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    /// Get storage quotes with the previous over-query behaviour.
1214    ///
1215    /// Merkle preflight uses quote responses only as an already-stored probe;
1216    /// the actual payment still happens through merkle candidate pools. Keep
1217    /// the extra peer buffer there so merkle upload behaviour remains
1218    /// unchanged when a few peers are slow or return unusable quote bindings.
1219    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        // Query the close-group width, but single-node payment now only needs
1246        // one valid witnessed quote to proceed.
1247        let close_group_query_count = single_node_quote_query_count();
1248        let required_quotes = SINGLE_NODE_MIN_QUOTE_COUNT;
1249        // Contact the closest PUT_TARGET_WIDTH peers directly so the whole
1250        // PUT-target set's addresses arrive in this single query. A network
1251        // with fewer than that near the target can't satisfy the wide lookup,
1252        // so fall back to the close-group width — the upload still proceeds with
1253        // a narrower (but valid) PUT-target set rather than failing.
1254        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        // Run quoting/quorum on the closest CLOSE_GROUP_SIZE only, so payment
1286        // semantics are unaffected by the wider PUT query.
1287        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        // Widen the PUT-target set to the closest PUT_TARGET_WIDTH
1318        // directly-contacted peers; the quote set above stays the closest
1319        // CLOSE_GROUP_SIZE. The same proof is reused on all of them.
1320        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 &quote_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        // Collect quote responses. SNP/witnessed collection deliberately tries
1372        // the closest witnessed peers first and only falls back to further
1373        // witnessed peers when a closer peer fails to produce a usable quote.
1374        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        // Track storer-rejecting peers separately (binding, content, signature
1379        // failures) so we can surface their count in diagnostics — they're a
1380        // special class of failure (peer misconfigured or hostile, not
1381        // network-broken) and the user benefits from seeing them called out.
1382        let mut bad_quote_count = 0usize;
1383
1384        // A storer's verdict that this client cannot settle, kept outside the
1385        // collection loops so neither the overall timeout nor an early exit
1386        // can discard it. Checked before any quote plan is built.
1387        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                        // Stop launching once the target is met, but keep
1397                        // draining below. Peers already in flight may yet
1398                        // declare this client unable to settle, and dropping
1399                        // that verdict because faster peers filled the quota
1400                        // would make it depend on response order. The surplus
1401                        // quotes are discarded; a refusal among them is not.
1402                        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            // Outranks the timeout: a refusal says paying is unsafe no matter
1464            // how many quotes were gathered before the clock ran out.
1465            if let Some(refusal) = settlement_refusal.take() {
1466                return Err(refusal);
1467            }
1468        } else {
1469            // Merkle preflight keeps the previous behaviour: query the full
1470            // over-query set concurrently because those quote responses are
1471            // only used as an already-stored probe.
1472            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                    // Fall through to check if we have enough quotes despite timeout.
1516                    // The timeout fires when slow peers haven't responded yet, but we
1517                    // may already have enough successful quotes from fast peers.
1518                }
1519                Ok(Err(e)) => return Err(e),
1520                Ok(Ok(())) => {}
1521            }
1522            // Outranks the timeout: a refusal says paying is unsafe no matter
1523            // how many quotes were gathered before the clock ran out.
1524            if let Some(refusal) = settlement_refusal.take() {
1525                return Err(refusal);
1526            }
1527        }
1528
1529        // Defensive double-check: the per-peer handler already filters
1530        // bad-binding responses into `failures`, but if any path slipped a bad
1531        // quote into `quotes` (e.g. a future refactor) this catches it before
1532        // we sort by distance and return. `bad_dropped` should be 0 in normal
1533        // operation; non-zero indicates an upstream regression worth investigating.
1534        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        // Check already-stored: only count votes from the closest CLOSE_GROUP_SIZE peers.
1546        if !already_stored_peers.is_empty() {
1547            let mut all_peers_by_distance: Vec<(bool, [u8; 32])> = Vec::new();
1548            for (peer_id, _, _, _, _) in &quotes {
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    //! Test fixtures use real ML-DSA-65 keypairs (1952-byte public keys), the
1620    //! same key material that ships on the wire. The "bad" quote is built by
1621    //! **swapping** the public key field with a different real keypair's
1622    //! public key — the exact shape produced by the Apr 30 production
1623    //! failure (an operator running two co-located identities with crossed
1624    //! quote-signing keys). Signatures are not exercised here because this
1625    //! filter only mirrors `validate_peer_bindings` (BLAKE3 binding); see
1626    //! the doc-comment on `quote_binding_is_valid` for why
1627    //! `verify_quote_signature` and `verify_quote_content` are deliberately
1628    //! NOT mirrored.
1629
1630    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    /// A real ML-DSA-65 keypair plus its derived peer ID.
1638    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    /// Build a PROPERLY-SIGNED baseline quote for `content`, signed by a real
1657    /// ML-DSA-65 key whose `BLAKE3(pub_key)` is the returned peer id. Passes the
1658    /// client's full classifier gate (binding + content + signature + price).
1659    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    /// Build a quote tuple whose `pub_key` correctly hashes to its peer_id.
1680    /// Signature is left empty: this filter does not verify signatures.
1681    ///
1682    /// The quote is a valid ADR-0004 **baseline**: `(0, None)` priced at
1683    /// `calculate_price(0)`, so it passes the forced-price gate in
1684    /// `classify_quote_response`. The 5th tuple element is the (absent)
1685    /// commitment sidecar.
1686    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    /// Build a quote tuple where the quote carries a different keypair's
1702    /// `pub_key` than the peer_id derives from. Mirrors the production
1703    /// failure shape: peer A advertised on the transport, but the quote
1704    /// carries peer B's key.
1705    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    /// Independent re-implementation of the storer-side binding spec
1799    /// (`ant-node/src/payment/verifier.rs::validate_peer_bindings` +
1800    /// `peer_id_from_public_key_bytes`):
1801    /// (a) `pub_key` parses as ML-DSA-65 (length 1952), and
1802    /// (b) `BLAKE3(pub_key) == peer_id`.
1803    ///
1804    /// Re-derived from spec, NOT delegating to `quote_binding_is_valid`,
1805    /// so cross-checks are not "function == itself".
1806    fn storer_binding_would_accept(peer_id: &PeerId, quote: &PaymentQuote) -> bool {
1807        if MlDsaPublicKey::from_bytes(&quote.pub_key).is_err() {
1808            return false;
1809        }
1810        compute_address(&quote.pub_key) == *peer_id.as_bytes()
1811    }
1812
1813    // ============================================================
1814    // Tests for `quote_binding_is_valid` (the predicate)
1815    // ============================================================
1816
1817    #[test]
1818    fn binding_accepts_real_self_consistent_keypair() {
1819        let (peer_id, _, quote, _, _) = good_quote_real();
1820        // Property under test: the predicate accepts a quote whose pub_key
1821        // genuinely belongs to the claimed peer.
1822        assert!(quote_binding_is_valid(&peer_id, &quote));
1823        // Cross-check against the independent full storer-spec implementation.
1824        assert!(storer_binding_would_accept(&peer_id, &quote));
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, &quote));
1831        assert!(!storer_binding_would_accept(&peer_id, &quote));
1832    }
1833
1834    #[test]
1835    fn binding_rejects_oversize_pubkey() {
1836        // A pub_key longer than ML-DSA-65 (1952 bytes) must be rejected
1837        // even if BLAKE3 happens to agree, because the storer rejects on
1838        // length first via `peer_id_from_public_key_bytes`.
1839        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        // BLAKE3(pub_key) DOES equal the peer_id we constructed, so the
1852        // bare hash check would pass — but the length guard must reject.
1853        assert_eq!(compute_address(&quote.pub_key), *peer_id.as_bytes());
1854        assert!(
1855            !quote_binding_is_valid(&peer_id, &quote),
1856            "predicate must reject oversize pub_key even when BLAKE3 happens to match"
1857        );
1858        assert!(!storer_binding_would_accept(&peer_id, &quote));
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, &quote));
1876        assert!(!storer_binding_would_accept(&peer_id, &quote));
1877    }
1878
1879    // ============================================================
1880    // Tests for the filter (`drop_quotes_with_bad_bindings`)
1881    // ============================================================
1882
1883    #[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    /// Ascending seeds `1..=count`, each a valid `u8` peer seed.
1961    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        // How many of the closest-`CLOSE_GROUP_SIZE` responders returned a view.
1970        // The remainder are "missing", so the scoped transcript also exercises
1971        // the missing-views quorum adjustment.
1972        const RESPONDED_IN_SCOPE: usize = 5;
1973        // Responders past the close group whose views scoping must drop.
1974        const OUT_OF_SCOPE_RESPONDERS: usize = 2;
1975
1976        let address = [0u8; 32];
1977        let close_seeds = ascending_seeds(CLOSE_GROUP_SIZE);
1978        // Each view's closest list mixes in-group (1, 2) and far (8, 9) peers so
1979        // candidate selection is non-trivial and must survive scoping verbatim.
1980        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        // A wide PUT_TARGET_WIDTH-peer transcript, ordered closest-first (seed n
1989        // == PeerId [n; 32], whose XOR distance to the zero address is n). Two
1990        // responders past the close group (out of scope) must be dropped.
1991        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        // The hand-built equivalent: a native CLOSE_GROUP_SIZE-wide query with
2005        // the same in-scope responders.
2006        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        // Target preserved; k and the initial set collapse to the close group.
2016        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        // Out-of-close-group responder views are dropped; the in-scope ones keep
2029        // their closest lists untouched (scoping filters by responder only).
2030        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        // The quorum math and candidate consensus run on the close group only
2050        // and are byte-for-byte identical to the native CLOSE_GROUP_SIZE query.
2051        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            &quotes,
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(&quotes), vec![1, 2, 3, 4, 5, 6, 7]);
2397        let (median_peer_id, _) =
2398            median_paid_quote_issuer(&quotes).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        // Cross-checked invariant: every retained quote would be accepted by
2419        // a storer running the full spec. The defensive filter only checks
2420        // the binding, so this asserts the binding-only filter is correct
2421        // for binding-only failures (other failure modes are filtered by
2422        // the per-peer classifier upstream).
2423        for (peer_id, _, quote, _, _) in &quotes {
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 &quotes {
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        // The "all hostile" case: every peer returned a bad binding. The
2446        // patch should leave us with zero quotes (not panic, not skip the
2447        // filter, not return malformed quotes). The caller then surfaces
2448        // InsufficientPeers.
2449        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        // After filtering, the retained quotes must be untouched — pub_key,
2460        // signature, content, timestamp, price, rewards_address. The patch
2461        // is a filter, not a transformation; this test catches any future
2462        // regression that mutates a retained quote.
2463        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    // ============================================================
2487    // The Apr 30 production-failure repro
2488    // ============================================================
2489
2490    /// Repro of the production failure from 2026-04-30 testnet runs.
2491    ///
2492    /// An external operator on `75.48.86.24` ran two co-located ant-node
2493    /// identities (peer `0755ecb55b…` and peer `073db92f…`) that crossed
2494    /// their quote-signing keys. Every chunk whose XOR-closest set happened
2495    /// to include peer `0755ecb5` got a payment proof with one malformed
2496    /// quote, and the storer's `validate_peer_bindings` rejected the
2497    /// entire close-group proof — burning the chunk's payment.
2498    ///
2499    /// This test proves the fault-tolerant quote path still fixes that failure
2500    /// shape:
2501    ///
2502    /// 1. We assemble `2x CLOSE_GROUP_SIZE` real ML-DSA-65 quotes — the same
2503    ///    buffer merkle preflight and merkle-mode estimates retain for probes.
2504    /// 2. One of them is a *crossed-key* quote — the production failure shape.
2505    /// 3. We run an independent `storer_would_accept` check (re-derived from
2506    ///    the storer spec, not from `quote_binding_is_valid`) over the
2507    ///    pre-filter set; we confirm the bad peer is rejected, proving the
2508    ///    storer **would** burn the chunk's payment if we proceeded unfiltered.
2509    /// 4. We run `drop_quotes_with_bad_bindings`.
2510    /// 5. We re-run `storer_would_accept` over the post-filter set; we confirm
2511    ///    EVERY remaining quote would be accepted, proving the filtered set
2512    ///    will not trigger the `validate_peer_bindings` rejection that caused
2513    ///    the Apr 30 outage.
2514    /// 6. We confirm the post-filter set has at least `CLOSE_GROUP_SIZE`
2515    ///    quotes — the over-query buffer (2x) is sufficient.
2516    #[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        // Splice the crossed-key quote in the middle (mirrors the random
2523        // position the bad peer takes in the DHT-returned closest set).
2524        quotes.insert(over_query_count / 2, bad_quote_real());
2525        assert_eq!(quotes.len(), over_query_count);
2526
2527        // Step 1: prove the storer would reject the pre-filter set.
2528        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        // Step 2: run the patched filter.
2538        let dropped = drop_quotes_with_bad_bindings(&mut quotes);
2539        assert_eq!(dropped, 1, "exactly the crossed-key quote must be filtered");
2540
2541        // Step 3: prove the storer would accept every survivor under the FULL spec.
2542        for (peer_id, _, quote, _, _) in &quotes {
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        // Step 4: prove the over-query buffer is sufficient to refill.
2551        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    /// When more than the over-query buffer of peers misbehave, the filter
2559    /// must NOT silently produce a short proof. The downstream caller in
2560    /// `get_store_quotes` must see fewer than `CLOSE_GROUP_SIZE` survivors
2561    /// and return `InsufficientPeers`.
2562    #[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        // Sanity: every survivor is storer-acceptable under the full spec.
2578        for (peer_id, _, quote, _, _) in &quotes {
2579            assert!(storer_binding_would_accept(peer_id, quote));
2580        }
2581    }
2582
2583    // ============================================================
2584    // Tests for the per-peer response classifier (the PRIMARY defense).
2585    //
2586    // These tests exercise the production code path that runs inside
2587    // get_store_quotes' per-peer async closure. The defensive
2588    // `drop_quotes_with_bad_bindings` is a second line of defence —
2589    // these tests make sure the FIRST line is what actually catches
2590    // misbehaving peers in production. Without these, a regression
2591    // that removes the per-peer check could be masked by the post-
2592    // collect filter and pass the rest of the suite.
2593    // ============================================================
2594
2595    /// Helper: serialize a `PaymentQuote` to bytes the way the wire layer
2596    /// does (rmp_serde / msgpack), to feed into `classify_quote_response`.
2597    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        // A properly-signed baseline quote for the requested content passes the
2604        // full client gate (binding + content + signature + price).
2605        let content = [7u8; 32];
2606        let (peer_id, quote) = signed_baseline_quote(content);
2607        let bytes = serialize_quote(&quote);
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        // A quote whose pub_key binds correctly but whose signature is bogus is
2622        // dropped BEFORE payment (the storer would reject it and burn the pay).
2623        let content = [7u8; 32];
2624        let (peer_id, mut quote) = signed_baseline_quote(content);
2625        quote.signature = vec![0u8; quote.signature.len()]; // corrupt the signature
2626        let bytes = serialize_quote(&quote);
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        // A validly-signed quote for a DIFFERENT address is dropped before pay.
2637        let (peer_id, quote) = signed_baseline_quote([7u8; 32]);
2638        let bytes = serialize_quote(&quote);
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(&quote);
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    /// CRITICAL: a misbehaving peer that votes `already_stored=true` must
2667    /// NOT be allowed to influence the close-group "already stored"
2668    /// majority decision. The bind-check runs before the AlreadyStored
2669    /// short-circuit, so a crossed-key peer voting "already stored" is
2670    /// classified as `BadQuoteBinding`, not `AlreadyStored`.
2671    ///
2672    /// This locks in a specific reviewer concern from round 1:
2673    ///   "A peer with a crossed/garbage signing key could simply respond
2674    ///   already_stored=true and its vote enters already_stored_peers
2675    ///   unfiltered."
2676    #[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(&quote);
2680        // The peer claims already_stored=true, but its quote has a crossed key.
2681        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    /// An honest peer's `already_stored=true` vote IS honoured (after
2690    /// passing the bind-check). This is the contrast to the test above.
2691    #[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(&quote);
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    /// Cross-validate the classifier's binding verdict against the
2715    /// independent storer-spec re-derivation across mixed responders.
2716    #[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    // ============================================================
2741    // ADR-0004: quote_commitment_binding_is_valid (forced-price gate)
2742    //
2743    // Mirrors the storer-side `binding_violation` in
2744    // `ant-node/src/payment/verifier.rs`. The client runs this before
2745    // paying so it never pays a quote the storer's arithmetic gate would
2746    // reject. The client now runs the FULL check (shape, cap, exact price,
2747    // and for bound quotes: parse + peer-binding + signature + hash==pin +
2748    // count==key_count) using the shared ant-protocol commitment type, so an
2749    // unresolvable/forged commitment is never paid. A live resolve against a
2750    // REAL signed commitment is proven in the e2e suite (e2e_adr0004.rs).
2751    // ============================================================
2752
2753    /// A throwaway peer id for tests that fail BEFORE commitment resolution
2754    /// (shape/cap/price checks don't depend on the peer).
2755    fn any_peer() -> PeerId {
2756        PeerId::from_bytes([0u8; 32])
2757    }
2758
2759    /// Build a quote carrying a specific `(count, pin, price)` binding.
2760    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    /// Build a VALIDLY-SIGNED `StorageCommitment` bound to `kp`'s peer id, so a
2778    /// test can pass peer-binding + signature and isolate the `hash == pin` and
2779    /// `count == key_count` sub-checks. Mirrors ant-node's commitment signing:
2780    /// the canonical payload (`root || key_count(LE) || peer_id || pk_len(LE) ||
2781    /// pub_key`) signed under `DOMAIN_COMMITMENT`.
2782    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        // (0, None) priced at calculate_price(0) is the valid baseline.
2810        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        // (0, None) priced above baseline is rejected — the forged-shape
2814        // bypass (strip the pin, charge more than the empty-node price).
2815        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        // count > 0 but no pin: unauditable.
2822        let q = quote_with_binding(500, None, calculate_price(500));
2823        assert!(quote_commitment_binding_is_valid(&any_peer(), &q, &None).is_err());
2824        // count 0 but a pin: incoherent baseline.
2825        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        // Priced for 499 but claims count 500 — on a real price curve but the
2842        // wrong count. Rejected at the exact-price check, before resolution.
2843        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        // A bound quote whose commitment did not arrive is unresolvable, so it
2850        // is dropped before payment even though its price is on the curve.
2851        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        // A bound quote whose shipped commitment is garbage (doesn't even
2861        // deserialize) is rejected — the client never pays an unresolvable pin.
2862        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        // A well-formed StorageCommitment that is NOT bound to the quoting peer
2869        // (its sender_peer_id / pubkey don't derive the peer id) is rejected at
2870        // the peer-binding check. The signature / hash==pin / count==key_count
2871        // sub-checks are covered by the dedicated tests below, which pass
2872        // peer-binding first so each isolates exactly one sub-check.
2873        let bogus = StorageCommitment {
2874            root: [1u8; 32],
2875            key_count: 500,
2876            sender_peer_id: [2u8; 32], // not the quoting peer
2877            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        // Correctly-bound commitment (passes peer-binding) but with a corrupted
2890        // signature: must be rejected at the signature check. Deleting that check
2891        // would let a peer attest any (root, key_count) without holding the key.
2892        let kp = gen_keypair();
2893        let mut commitment = signed_commitment(&kp, [6u8; 32], 500);
2894        commitment.signature[0] ^= 0xFF; // still 3293 bytes, no longer valid
2895                                         // Pin the (corrupted) commitment so the hash==pin check would pass; the
2896                                         // only thing wrong is the signature, isolating that sub-check.
2897        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        // Validly-signed, correctly-bound commitment, but the quote pins a
2911        // DIFFERENT hash: must be rejected. Deleting the hash==pin check would
2912        // let a peer ship any commitment it holds for a pin it doesn't back.
2913        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        // Validly-signed, correctly-bound, correctly-pinned commitment attesting
2930        // key_count=400, but the quote claims 500 (priced on-curve for 500):
2931        // must be rejected. Deleting the count==key_count check would let a peer
2932        // price against an inflated count while committing to fewer keys.
2933        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        // A bound quote shipping a blob larger than the sidecar cap is rejected
2949        // before any deserialize attempt (DoS guard on the hot path).
2950        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        // End-to-end through the classifier: a VALIDLY-SIGNED, correctly-bound
2961        // quote for the right content, but with an off-curve price, is dropped
2962        // as BadQuoteCommitment (the forced-price extraction guard fires after
2963        // the quote's own signature/content checks pass).
2964        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            // claims baseline shape but charges a non-baseline price
2971            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, &quote.bytes_for_sig())
2982            .expect("sign")
2983            .as_bytes()
2984            .to_vec();
2985        let bytes = serialize_quote(&quote);
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    /// A storer's refusal must arrive as its own terminal error, carrying the
2994    /// storer's wording. Folding it into the generic protocol error would bury
2995    /// the upgrade instruction among ordinary per-peer quote failures, which
2996    /// is the outcome this whole change exists to avoid.
2997    #[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    /// The legacy retry exists for storers that cannot parse a versioned
3021    /// request, which are silent. A storer that answered has understood us, so
3022    /// retrying its refusal without the version would talk it into quoting a
3023    /// client that cannot pay. That is the exact failure this change removes,
3024    /// so the predicate deciding it is pinned.
3025    #[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        // A storer that says it is the old side has understood the request.
3034        // Retrying it unversioned would obtain a quote from a peer that cannot
3035        // verify the resulting payment, which is a burn.
3036        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    /// A storer declaring itself the old side is an ordinary skippable peer,
3045    /// not a client fault. Surfacing it as `ClientUpdateRequired` would tell an
3046    /// up-to-date user to upgrade, and during a client-first rollout it would
3047    /// tell that to nearly everyone.
3048    #[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    /// The refusal has to stop quote collection, not join the failure list.
3072    /// If it is merely recorded, the remaining peers can still form a quorum
3073    /// and the upload proceeds to pay, which is exactly the burn the gate is
3074    /// meant to prevent.
3075    #[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        // One peer is not corroboration: recorded as an ordinary bad peer, so a
3103        // single hostile responder cannot deny every upload.
3104        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        // A second, distinct peer makes it a verdict about this build.
3110        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    /// A storer being behind must NOT abort. Otherwise one lagging peer in the
3118    /// close group fails an upload that the rest of the group could serve.
3119    #[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    /// The refusal must survive the overall collection timeout.
3152    ///
3153    /// The collector runs inside `tokio::time::timeout`, and its elapsed arm
3154    /// deliberately falls through so quotes gathered from fast peers stay
3155    /// usable. That arm would otherwise discard a refusal observed just before
3156    /// the clock ran out, and the upload would pay anyway. Recording the
3157    /// verdict in a slot that outlives the timeout is what prevents it, so the
3158    /// slot is what gets tested.
3159    #[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    /// Once the quote target is met the collector stops launching new peers
3194    /// but keeps draining those already in flight, so a refusal cannot be
3195    /// missed merely because faster peers filled the quota first.
3196    ///
3197    /// The launch budget enforces the first half, and the drain relies on it
3198    /// reaching zero to terminate rather than recruiting forever.
3199    #[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        // In-flight peers count against the budget, so draining them does not
3211        // pull in replacements.
3212        assert_eq!(witnessed_quote_launch_budget(0, CLOSE_GROUP_SIZE, 32), 0);
3213    }
3214
3215    /// The probe must be paid per peer, not per request.
3216    ///
3217    /// A storer that predates the versioned request never answers it, so the
3218    /// client eats the probe wait before falling back. Without remembering the
3219    /// answer that cost lands on every quote, and a merkle pool asks sixteen
3220    /// candidates. Measured on the merkle E2E suite against a fleet on
3221    /// published ant-node, re-probing took the run from ~24 minutes to past
3222    /// the 60-minute CI cap.
3223    ///
3224    /// The cache is consulted, not enforced: concurrent first contacts are not
3225    /// single-flighted, so the same peer can be probed by a few in-flight
3226    /// requests before any of them records the answer. What the cache
3227    /// guarantees is that later rounds do not re-probe.
3228    #[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        // First contact: nothing known, so the versioned request is sent.
3237        assert!(!known(&legacy_peer));
3238
3239        // Silence records the peer.
3240        peers.lock().expect("cache lock").insert(legacy_peer);
3241
3242        // Second contact skips the probe entirely.
3243        assert!(known(&legacy_peer));
3244        // and does not tar every other peer with the same brush.
3245        assert!(!known(&fresh_peer));
3246    }
3247
3248    /// Only silence is evidence that a peer cannot parse the versioned shape.
3249    ///
3250    /// Both a timeout and a send failure trigger the legacy retry, but they
3251    /// mean different things: a send failure says the request never arrived,
3252    /// so it teaches nothing about the peer's capabilities. Caching it would
3253    /// strand that peer in the legacy shape for the rest of the session over
3254    /// one flaky send.
3255    #[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        // Both still take the fallback, so a send failure is retried rather
3266        // than left to fail outright.
3267        assert!(is_version_unaware(&Error::Network("send failed".into())));
3268        assert!(is_version_unaware(&Error::Timeout("no answer".into())));
3269    }
3270
3271    /// One peer cannot condemn the client.
3272    ///
3273    /// Nothing authenticates a refusal, so a single hostile or misconfigured
3274    /// storer answering `ClientUpdateRequired` to everything would otherwise
3275    /// abort every upload. That turns an over-query design which tolerates
3276    /// many bad peers into one that tolerates none.
3277    #[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        // The same peer repeating itself is still one peer.
3288        assert!(refusals
3289            .note(PeerId::from_bytes([0x61; 32]), "too old")
3290            .is_none());
3291        assert!(refusals.corroborated().is_none());
3292    }
3293
3294    /// A genuine incompatibility reaches the threshold at once, because every
3295    /// peer enforcing the newer rule refuses.
3296    #[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        // Latched: an upload starting later must see it before it spends,
3304        // which is the whole point of holding it on the client rather than in
3305        // one collector's local state.
3306        assert!(refusals
3307            .corroborated()
3308            .is_some_and(|m| m.contains("ant update")));
3309    }
3310
3311    /// A refusal that does not describe this client is a confused or lying
3312    /// peer, not evidence about this build, and must not count toward the
3313    /// threshold.
3314    #[test]
3315    fn an_incoherent_refusal_is_treated_as_a_bad_peer() {
3316        let peer_id = PeerId::from_bytes([0x64; 32]);
3317
3318        // Claims to be about some other client version.
3319        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        // Claims a minimum that our version already satisfies.
3327        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        // A coherent one is believed, and carries the upgrade instruction.
3335        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}