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 crate::data::network::send_and_await_chunk_response;
13#[cfg(test)]
14use ant_protocol::compute_address;
15use ant_protocol::evm::{Amount, PaymentQuote};
16#[cfg(test)]
17use ant_protocol::payment::calculate_price;
18use ant_protocol::payment::commitment::StorageCommitment;
19#[cfg(test)]
20use ant_protocol::payment::commitment::{
21    commitment_hash, MAX_COMMITMENT_KEY_COUNT, MAX_COMMITMENT_SIDECAR_BYTES,
22};
23use ant_protocol::payment::verify_quote_signature;
24use ant_protocol::transport::{DHTNode, MultiAddr, PeerId, ResponderView, WitnessedCloseGroup};
25use ant_protocol::{
26    client_update_required_message, ChunkMessage, ChunkMessageBody, ChunkQuoteRequest,
27    ChunkQuoteRequestV2, ChunkQuoteResponse, ProtocolError, CLOSE_GROUP_SIZE,
28    CURRENT_SETTLEMENT_VERSION,
29};
30use futures::stream::{FuturesUnordered, StreamExt};
31use std::collections::{HashMap, HashSet};
32use std::sync::{Arc, Mutex};
33use std::time::Duration;
34use tracing::{debug, info, warn};
35
36/// Fault-tolerant quote collection asks one extra close group of peers and
37/// keeps the closest successful `CLOSE_GROUP_SIZE` responders. This remains
38/// useful for merkle preflight probes, but single-node payments deliberately
39/// ask only the actual close group.
40const FAULT_TOLERANT_QUOTE_QUERY_MULTIPLIER: usize = 2;
41
42use crate::quote_policy::{SINGLE_NODE_MIN_QUOTE_COUNT, SINGLE_NODE_WITNESSED_VIEW_COUNT};
43
44/// Overall timeout for collecting quote responses. Must accommodate
45/// connect_with_fallback cascade (direct 5s + hole-punch 15s×3 + relay 30s ≈
46/// 80s) plus the per-peer quote timeout.
47const QUOTE_COLLECTION_TIMEOUT_SECS: u64 = 120;
48
49/// ML-DSA-65 public key length in bytes. Mirrors the same value defined as
50/// `pub const ML_DSA_65_PUBLIC_KEY_SIZE` in `saorsa-pqc::pqc::types`, which
51/// the storer's `peer_id_from_public_key_bytes` enforces. We keep a local
52/// copy here rather than adding a direct `saorsa-pqc` dep — the constant
53/// is FIPS-mandated for ML-DSA-65 and won't change unless we change variant.
54///
55/// TODO: switch to `saorsa_pqc::pqc::types::ML_DSA_65_PUBLIC_KEY_SIZE` once
56/// `ant-protocol` re-exports it (`pqc::ops::ML_DSA_65_PUBLIC_KEY_SIZE`).
57#[cfg(test)]
58use crate::quote_validation::ML_DSA_PUB_KEY_LEN;
59
60/// One collected quote: the responding peer, its addresses, the signed quote,
61/// the price it demands, and (ADR-0004) the opaque signed-commitment blob the
62/// node shipped alongside the quote (`None` for a baseline quote), to be
63/// forwarded as a sidecar in the PUT bundle.
64type QuotedPeer = (
65    PeerId,
66    Vec<MultiAddr>,
67    PaymentQuote,
68    Amount,
69    Option<Vec<u8>>,
70);
71
72fn quote_fields(quote: &PaymentQuote) -> crate::quote_validation::QuoteFields<'_, Amount> {
73    crate::quote_validation::QuoteFields {
74        public_key: &quote.pub_key,
75        content: &quote.content.0,
76        price: quote.price,
77        committed_key_count: quote.committed_key_count,
78        commitment_pin: quote.commitment_pin,
79    }
80}
81
82/// Check that a quote's `pub_key` is well-formed and BLAKE3-hashes to the
83/// claimed `peer_id`.
84///
85/// The storer node enforces both constraints in `ant-node/src/payment/verifier.rs`
86/// (via `peer_id_from_public_key_bytes` and `validate_peer_bindings`): every
87/// quote inside a `ProofOfPayment` must (a) have a 1952-byte `pub_key` parsable
88/// as ML-DSA-65 and (b) satisfy `BLAKE3(pub_key) == peer_id`. A single quote
89/// failing either check causes the storer to reject the entire close-group
90/// proof and burn the chunk's payment.
91///
92/// This is the cheap structural pre-check. ADR-0004 additionally has the client
93/// run `verify_quote_content` + `verify_quote_signature` (the full ML-DSA check)
94/// in [`classify_quote_response`] before paying, so a quote the storer would
95/// reject never gets paid.
96fn quote_binding_is_valid(peer_id: &PeerId, quote: &PaymentQuote) -> bool {
97    crate::quote_validation::peer_binding_is_valid(peer_id.as_bytes(), &quote.pub_key)
98}
99
100/// ADR-0004 client-side resolve-before-pay gate — "the client pays nothing it
101/// cannot resolve", the ceiling's load-bearing wall (ADR §"The client pays
102/// nothing it cannot resolve").
103///
104/// Runs the **full** binding check before paying, identical to the storer's,
105/// using the shared `ant-protocol` commitment type + verifier so client and
106/// node can never disagree:
107/// 1. **Shape.** `(0, None)` baseline or `(n>0, Some(pin))` bound; the mixed
108///    shapes `(n>0, None)` (unauditable count) and `(0, Some)` (incoherent
109///    baseline) are rejected.
110/// 2. **Cap.** `committed_key_count <= MAX_COMMITMENT_KEY_COUNT` — a count a
111///    commitment could never legitimately attest is rejected before pricing.
112/// 3. **Forced price.** `price == calculate_price(committed_key_count)`, by
113///    exact recomputation with the shared `calculate_price` — never inverted.
114/// 4. **Resolution (bound quotes).** The shipped commitment must: parse as a
115///    `StorageCommitment`, be bound to the quoting peer
116///    (`BLAKE3(sender_public_key) == sender_peer_id`), have a valid ML-DSA-65
117///    signature, hash to the quote's `commitment_pin`
118///    (`commitment_hash == pin`), and attest exactly the claimed count
119///    (`key_count == committed_key_count`). A withheld, unparseable, wrong-pin,
120///    mis-bound, or count-mismatched commitment is unresolvable → the quote is
121///    dropped before payment.
122///
123/// Returns `Ok(())` if the binding fully resolves, or `Err(detail)` naming the
124/// rule that failed.
125#[cfg(test)]
126fn quote_commitment_binding_is_valid(
127    peer_id: &PeerId,
128    quote: &PaymentQuote,
129    commitment: &Option<Vec<u8>>,
130) -> std::result::Result<(), String> {
131    crate::quote_validation::validate_commitment_binding::<_, StorageCommitment>(
132        peer_id.as_bytes(),
133        &quote_fields(quote),
134        commitment.as_deref(),
135    )
136}
137
138/// Classification of a `ChunkQuoteResponse::Success` body for a single peer.
139///
140/// Mirrors the storer-side `validate_peer_bindings` check from
141/// `ant-node/src/payment/verifier.rs` — the cheap BLAKE3 binding —
142/// so we drop misbehaving peers' quotes before payment.
143///
144/// ADR-0004: the client now ALSO runs the storer's `verify_quote_content` and
145/// `verify_quote_signature` (ML-DSA-65) before paying, so "the client pays
146/// nothing it cannot resolve" covers the quote's own validity too, not just the
147/// commitment binding. This matches what the merkle path already does
148/// client-side and costs ~1 ms × CLOSE_GROUP_SIZE per chunk — accepted, since
149/// paying a quote the storer then rejects burns the on-chain payment.
150///
151/// Pulling the logic out of the async closure lets us unit-test the
152/// primary defense (not just the post-collect defensive filter).
153///
154/// # Returns
155///
156/// - `Ok((quote, price))` — the response is honoured as a quote.
157/// - `Err(Error::AlreadyStored)` — the peer claims the chunk is already
158///   present AND the quote it provided binds to its peer ID. Vote counts.
159/// - `Err(Error::BadQuoteBinding { .. })` — bad binding (mirrors the
160///   storer-side rejection). Outer collector counts these via the typed
161///   variant (no string matching).
162/// - `Err(Error::BadQuoteCommitment { .. })` — ADR-0004 forced-price binding
163///   failed (price off the curve, incoherent shape, or a bound quote that did
164///   not ship its commitment); dropped before payment like a bad binding.
165/// - `Err(Error::Serialization(...))` — the quote bytes did not deserialize.
166///
167/// On success the returned commitment is the opaque signed-commitment blob the
168/// node shipped with the quote (`None` for a baseline quote), to be forwarded
169/// as a sidecar in the PUT bundle.
170///
171/// A quote as this module hands it on: the quote itself, the price to settle,
172/// and the opaque signed commitment it was priced against.
173type ClassifiedQuote = std::result::Result<(PaymentQuote, Amount, Option<Vec<u8>>), Error>;
174pub(crate) fn classify_quote_response(
175    peer_id: &PeerId,
176    expected_content: &[u8; 32],
177    quote_bytes: &[u8],
178    already_stored: bool,
179    commitment: Option<Vec<u8>>,
180) -> ClassifiedQuote {
181    let payment_quote = rmp_serde::from_slice::<PaymentQuote>(quote_bytes).map_err(|e| {
182        Error::Serialization(format!("Failed to deserialize quote from {peer_id}: {e}"))
183    })?;
184
185    crate::quote_validation::validate_quote::<_, StorageCommitment>(
186        peer_id.as_bytes(),
187        expected_content,
188        &quote_fields(&payment_quote),
189        || verify_quote_signature(&payment_quote),
190        commitment.as_deref(),
191    )
192    .map_err(|error| match error {
193        crate::quote_validation::QuoteValidationError::Binding(detail) => Error::BadQuoteBinding {
194            peer_id: peer_id.to_string(),
195            detail,
196        },
197        crate::quote_validation::QuoteValidationError::Commitment(detail) => {
198            Error::BadQuoteCommitment {
199                peer_id: peer_id.to_string(),
200                detail,
201            }
202        }
203    })?;
204
205    if already_stored {
206        debug!("Peer {peer_id} already has chunk");
207        return Err(Error::AlreadyStored);
208    }
209    let price = payment_quote.price;
210    debug!("Received quote from {peer_id}: price = {price}");
211    Ok((payment_quote, price, commitment))
212}
213
214/// Drop quotes whose `pub_key` does not BLAKE3-hash to the peer that supplied
215/// them. Logs each dropped quote at WARN.
216fn drop_quotes_with_bad_bindings(quotes: &mut Vec<QuotedPeer>) -> usize {
217    let before = quotes.len();
218    quotes.retain(|(peer_id, _, quote, _, _)| {
219        if quote_binding_is_valid(peer_id, quote) {
220            true
221        } else {
222            warn!(
223                "Dropping quote from peer {peer_id} — quote.pub_key BLAKE3 mismatch \
224                 (peer is signing quotes with another peer's key); the storer would \
225                 reject this proof"
226            );
227            false
228        }
229    });
230    before - quotes.len()
231}
232
233#[allow(clippy::too_many_arguments)]
234async fn request_store_quote_from_peer(
235    node: crate::data::network::Network,
236    peer_id: PeerId,
237    peer_addrs: Vec<MultiAddr>,
238    request_id: u64,
239    address: [u8; 32],
240    data_size: u64,
241    data_type: u32,
242    per_peer_timeout: Duration,
243    unversioned_peers: Arc<Mutex<HashSet<PeerId>>>,
244    versioned_capable: Arc<Mutex<HashSet<PeerId>>>,
245) -> StoreQuoteRequestResult {
246    let legacy_request = ChunkQuoteRequest {
247        address,
248        data_size,
249        data_type,
250    };
251
252    // A peer that already failed to answer a versioned request is asked in the
253    // legacy shape directly. Re-probing costs a full per-peer timeout every
254    // time, and against a fleet that predates the versioned requests that is
255    // paid on every quote: measured on the merkle E2E suite it took the run
256    // from ~24 minutes to past the 60-minute cap.
257    // The capable set wins. The two sets are updated under separate locks, so
258    // a slow probe can insert into the legacy set after a concurrent request
259    // has already proved the peer capable; letting capability win makes that
260    // interleaving harmless instead of permanently preferring the legacy shape.
261    let known_legacy = !versioned_capable
262        .lock()
263        .is_ok_and(|peers| peers.contains(&peer_id))
264        && unversioned_peers
265            .lock()
266            .is_ok_and(|peers| peers.contains(&peer_id));
267
268    // Declare the settlement version so a storer can turn us away before we
269    // pay. See `merkle.rs` for why the legacy retry below exists and when it
270    // can be deleted.
271    let body = if known_legacy {
272        ChunkMessageBody::QuoteRequest(legacy_request.clone())
273    } else {
274        let mut versioned_request = ChunkQuoteRequestV2::new(address, data_size);
275        versioned_request.data_type = data_type;
276        ChunkMessageBody::QuoteRequestV2(versioned_request)
277    };
278    let message = ChunkMessage { request_id, body };
279
280    let message_bytes = match message.encode() {
281        Ok(bytes) => bytes,
282        Err(e) => {
283            return (
284                peer_id,
285                peer_addrs,
286                Err(Error::Protocol(format!(
287                    "Failed to encode quote request for {peer_id}: {e}"
288                ))),
289            );
290        }
291    };
292
293    // A first-contact probe waits only long enough to learn whether the peer
294    // can parse the shape; a peer already known to be legacy is asked with the
295    // caller's full patience because that request is the real one.
296    let attempt_timeout = if known_legacy {
297        per_peer_timeout
298    } else {
299        per_peer_timeout.min(VERSIONED_QUOTE_PROBE_CEILING)
300    };
301
302    let result = send_and_await_chunk_response(
303        &node,
304        &peer_id,
305        message_bytes,
306        request_id,
307        attempt_timeout,
308        &peer_addrs,
309        |body| map_quote_response(&peer_id, &address, body),
310        |e| Error::Network(format!("Failed to send quote request to {peer_id}: {e}")),
311        || Error::Timeout(format!("Timeout waiting for quote from {peer_id}")),
312    )
313    .await;
314
315    // Any recognised answer proves the peer parsed the versioned shape,
316    // including a structured error. Only silence and send failures leave the
317    // question open.
318    let answered = match &result {
319        Ok(_) => true,
320        Err(e) => !is_version_unaware(e),
321    };
322    if !known_legacy && answered {
323        if let Ok(mut peers) = versioned_capable.lock() {
324            peers.insert(peer_id);
325        }
326    }
327
328    // Only a storer that could not decode the versioned request is asked
329    // again in the legacy shape. One that answered has understood us, and a
330    // refusal must stay a refusal.
331    let result = match result {
332        Err(ref e) if is_version_unaware(e) && !known_legacy => {
333            // Remember it, so the next request to this peer skips the probe.
334            // Only silence counts as evidence: a send failure means the
335            // request never arrived, which says nothing about whether the peer
336            // could have parsed it, and caching that would strand a peer in
337            // the legacy shape for the rest of the session over one flaky send.
338            // A peer that has answered a versioned request before is never
339            // demoted: one lost response would otherwise pin an upgraded peer
340            // to the legacy shape for the rest of the session.
341            let ever_answered = versioned_capable
342                .lock()
343                .is_ok_and(|peers| peers.contains(&peer_id));
344            if matches!(e, Error::Timeout(_)) && !ever_answered {
345                if let Ok(mut peers) = unversioned_peers.lock() {
346                    peers.insert(peer_id);
347                }
348            }
349            let legacy = ChunkMessage {
350                request_id,
351                body: ChunkMessageBody::QuoteRequest(legacy_request),
352            };
353            match legacy.encode() {
354                Ok(legacy_bytes) => {
355                    send_and_await_chunk_response(
356                        &node,
357                        &peer_id,
358                        legacy_bytes,
359                        request_id,
360                        per_peer_timeout,
361                        &peer_addrs,
362                        |body| map_quote_response(&peer_id, &address, body),
363                        |e| {
364                            Error::Network(format!(
365                                "Failed to send quote request to {peer_id}: {e}"
366                            ))
367                        },
368                        || Error::Timeout(format!("Timeout waiting for quote from {peer_id}")),
369                    )
370                    .await
371                }
372                Err(e) => Err(Error::Protocol(format!(
373                    "Failed to encode quote request for {peer_id}: {e}"
374                ))),
375            }
376        }
377        other => other,
378    };
379
380    (peer_id, peer_addrs, result)
381}
382
383/// Turn a quote response into the quote it carries, or the error explaining
384/// why there is none. Shared by the versioned request and its legacy retry.
385///
386/// `ClientUpdateRequired` is separated from the generic protocol error because
387/// it is terminal: it must reach the user with its own wording rather than
388/// being counted as one more peer that failed to quote.
389fn map_quote_response(
390    peer_id: &PeerId,
391    address: &[u8; 32],
392    body: ChunkMessageBody,
393) -> Option<ClassifiedQuote> {
394    match body {
395        ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Success {
396            quote,
397            already_stored,
398            commitment,
399        }) => Some(classify_quote_response(
400            peer_id,
401            address,
402            &quote,
403            already_stored,
404            commitment,
405        )),
406        ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error(
407            ProtocolError::ClientUpdateRequired {
408                client_settlement_version,
409                min_settlement_version,
410            },
411        )) => Some(Err(settlement_refusal_error(
412            peer_id,
413            client_settlement_version,
414            min_settlement_version,
415        ))),
416        ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error(
417            behind @ ProtocolError::StorerUpdateRequired { .. },
418        )) => Some(Err(Error::StorerUpdateRequired(behind.to_string()))),
419        ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error(e)) => Some(Err(
420            Error::Protocol(format!("Quote error from {peer_id}: {e}")),
421        )),
422        _ => None,
423    }
424}
425
426/// Turn a peer's `ClientUpdateRequired` into an error, rejecting one that does
427/// not describe this client.
428///
429/// A refusal is unauthenticated, so the least this client can do is check that
430/// the peer is talking about the request it actually sent: the echoed version
431/// must be ours, and the stated minimum must genuinely exceed it. A peer that
432/// fails either is confused or lying, and is treated as an ordinary bad peer
433/// rather than as evidence about this build.
434pub(super) fn settlement_refusal_error(
435    peer_id: &PeerId,
436    client_settlement_version: u32,
437    min_settlement_version: u32,
438) -> Error {
439    if client_settlement_version != CURRENT_SETTLEMENT_VERSION
440        || min_settlement_version <= client_settlement_version
441    {
442        return Error::Protocol(format!(
443            "Peer {peer_id} sent an incoherent settlement refusal (claimed this client is at \
444             version {client_settlement_version} needing {min_settlement_version}, but this \
445             client is at {CURRENT_SETTLEMENT_VERSION}); ignoring it"
446        ));
447    }
448    Error::ClientUpdateRequired(client_update_required_message(
449        client_settlement_version,
450        min_settlement_version,
451    ))
452}
453
454/// Did this failure look like a storer that cannot parse a versioned request,
455/// as opposed to one that parsed it and refused? See the merkle path for the
456/// full reasoning.
457///
458/// This is the single-node fallback, independent of the merkle one, so it
459/// carries its own reference to the shared cutover guard. Without it, deleting
460/// the merkle fallback would take the only build-time check with it and leave
461/// this downgrade path live.
462const _: () = crate::data::client::UNVERSIONED_RETRY_REQUIRES_MIN_V1;
463
464const fn is_version_unaware(error: &Error) -> bool {
465    matches!(error, Error::Network(_) | Error::Timeout(_))
466}
467
468/// Fold one peer's quote result into the collection state.
469///
470/// Returns `Err` only when collection must stop outright: a storer has said
471/// this client cannot settle, and no number of further quotes makes paying
472/// safe.
473#[allow(clippy::too_many_arguments)]
474fn record_store_quote_result(
475    peer_id: PeerId,
476    addrs: Vec<MultiAddr>,
477    quote_result: Result<(PaymentQuote, Amount, Option<Vec<u8>>)>,
478    address: &[u8; 32],
479    quotes: &mut Vec<StoreQuote>,
480    already_stored_peers: &mut Vec<(PeerId, [u8; 32])>,
481    failures: &mut Vec<String>,
482    bad_quote_count: &mut usize,
483    settlement_refusal: &mut Option<Error>,
484    refusals: &SettlementRefusals,
485) -> Result<()> {
486    match quote_result {
487        Ok((quote, price, commitment)) => {
488            quotes.push((peer_id, addrs, quote, price, commitment));
489        }
490        Err(Error::AlreadyStored) => {
491            info!("Peer {peer_id} reports chunk already stored");
492            let dist = peer_xor_distance(&peer_id, address);
493            already_stored_peers.push((peer_id, dist));
494        }
495        // A storer that has explicitly declared this client incompatible ends
496        // quote collection here. Recording it as one more failed peer would
497        // let the remaining peers supply a quorum and go on to pay, which is
498        // the burn this mechanism exists to prevent, and would bury the
499        // upgrade instruction among ordinary per-peer failures.
500        // Recorded as well as returned. The caller runs this inside an overall
501        // timeout, and if that timeout fires the returned error is thrown away
502        // with the rest of the collection state. The verdict must outlive it:
503        // a client told it cannot settle must not pay, whatever else happened
504        // during collection.
505        Err(e @ Error::ClientUpdateRequired(_)) => {
506            // One peer's word is not a verdict about this build: nothing
507            // authenticates a refusal, so a single hostile peer answering this
508            // to everything would deny every upload. Believe it once enough
509            // distinct peers agree, which a genuine incompatibility reaches at
510            // once because every enforcing peer refuses.
511            let Some(corroborated) = refusals.note(peer_id, &e.to_string()) else {
512                warn!("Peer {peer_id} refused this client's settlement version; awaiting corroboration");
513                failures.push(format!("{peer_id}: {e}"));
514                return Ok(());
515            };
516            // Name the corroborators. This is the only place the quorum is
517            // externally verifiable: the below-quorum branch above logs one
518            // peer at a time, so without this line a reader cannot tell a
519            // two-peer verdict from a one-peer misfire (V2-1109).
520            let corroborators = refusals.corroborating_peers();
521            warn!(
522                "Settlement refusal corroborated by {} distinct peers [{}]; aborting before payment",
523                corroborators.len(),
524                corroborators.join(", ")
525            );
526            let verdict = Error::ClientUpdateRequired(corroborated);
527            if settlement_refusal.is_none() {
528                *settlement_refusal = Some(Error::ClientUpdateRequired(verdict.to_string()));
529            }
530            return Err(verdict);
531        }
532        Err(e) => {
533            if matches!(&e, Error::BadQuoteBinding { .. }) {
534                *bad_quote_count += 1;
535            }
536            warn!("Failed to get quote from {peer_id}: {e}");
537            failures.push(format!("{peer_id}: {e}"));
538        }
539    }
540    Ok(())
541}
542
543fn witnessed_quote_launch_budget(
544    successful_quotes: usize,
545    in_flight: usize,
546    remaining_peers: usize,
547) -> usize {
548    crate::quote_policy::quote_launch_budget(successful_quotes, in_flight, remaining_peers)
549}
550
551fn single_node_quote_query_count() -> usize {
552    CLOSE_GROUP_SIZE
553}
554
555fn fault_tolerant_quote_query_count() -> usize {
556    CLOSE_GROUP_SIZE * FAULT_TOLERANT_QUOTE_QUERY_MULTIPLIER
557}
558
559fn witnessed_close_group_quorum() -> usize {
560    crate::quote_policy::witness_quorum(0)
561}
562
563fn witnessed_close_group_quorum_for_missing_views(missing_views: usize) -> usize {
564    crate::quote_policy::witness_quorum(missing_views)
565}
566
567fn missing_witnessed_responder_views(witnessed: &WitnessedCloseGroup) -> usize {
568    witnessed
569        .initial_closest
570        .len()
571        .saturating_sub(witnessed.responder_views.len())
572}
573
574fn witnessed_close_group_quorum_for_transcript(witnessed: &WitnessedCloseGroup) -> usize {
575    witnessed_close_group_quorum_for_missing_views(missing_witnessed_responder_views(witnessed))
576}
577
578/// Restrict a witnessed transcript to its closest `CLOSE_GROUP_SIZE` peers.
579///
580/// The witnessed query is widened to `PUT_TARGET_WIDTH` peers so we
581/// have addresses for the full PUT-target set, but the consensus/quorum/quote
582/// logic must still run on the close group only. Keeping just the closest-7
583/// initial peers and the responder views contributed by those peers leaves the
584/// `missing_witnessed_responder_views` math — and the quorum derived from it —
585/// byte-for-byte identical to a `CLOSE_GROUP_SIZE`-wide query.
586fn scope_witnessed_to_close_group(witnessed: &WitnessedCloseGroup) -> WitnessedCloseGroup {
587    let initial_closest: Vec<DHTNode> = witnessed
588        .initial_closest
589        .iter()
590        .take(CLOSE_GROUP_SIZE)
591        .cloned()
592        .collect();
593    let keys = initial_closest
594        .iter()
595        .map(|node| node.peer_id)
596        .collect::<Vec<_>>();
597    let views = witnessed
598        .responder_views
599        .iter()
600        .map(|view| crate::quote_policy::ResponderView {
601            responder: view.responder,
602            closest: view.closest.iter().map(|node| node.peer_id).collect(),
603        })
604        .collect::<Vec<_>>();
605    let scoped = crate::quote_policy::scope_views(&keys, &views)
606        .into_iter()
607        .map(|view| view.responder)
608        .collect::<HashSet<_>>();
609    let responder_views: Vec<ResponderView> = witnessed
610        .responder_views
611        .iter()
612        .filter(|view| scoped.contains(&view.responder))
613        .cloned()
614        .collect();
615    WitnessedCloseGroup {
616        target: witnessed.target,
617        k: CLOSE_GROUP_SIZE,
618        initial_closest,
619        responder_views,
620    }
621}
622
623fn peer_list(peers: &[PeerId]) -> Vec<String> {
624    peers.iter().map(ToString::to_string).collect()
625}
626
627/// One collected store quote, carrying (ADR-0004) the opaque signed-commitment
628/// sidecar the node shipped with its quote (`None` for a baseline quote), to be
629/// forwarded in the PUT bundle and cross-checked by storers.
630pub(crate) type StoreQuote = (
631    PeerId,
632    Vec<MultiAddr>,
633    PaymentQuote,
634    Amount,
635    Option<Vec<u8>>,
636);
637type StoreQuoteRequestResult = (
638    PeerId,
639    Vec<MultiAddr>,
640    Result<(PaymentQuote, Amount, Option<Vec<u8>>)>,
641);
642type VotersByPeer = HashMap<PeerId, HashSet<PeerId>>;
643type WitnessedVoteData = (HashMap<PeerId, DHTNode>, VotersByPeer, Vec<(PeerId, usize)>);
644
645pub(crate) struct StoreQuotePlan {
646    pub(crate) quotes: Vec<StoreQuote>,
647    pub(crate) put_peers: Vec<(PeerId, Vec<MultiAddr>)>,
648}
649
650#[derive(Debug, Clone)]
651struct WitnessedQuoteCandidate {
652    node: DHTNode,
653    votes: usize,
654    voters: HashSet<PeerId>,
655}
656
657#[derive(Debug, Clone)]
658struct WitnessedQuotePeer {
659    peer_id: PeerId,
660    addrs: Vec<MultiAddr>,
661    voters: HashSet<PeerId>,
662}
663
664#[derive(Debug, Clone)]
665struct WitnessedQuoteSelection {
666    quote_peers: Vec<WitnessedQuotePeer>,
667    initial_put_peers: Vec<(PeerId, Vec<MultiAddr>)>,
668    quorum: usize,
669}
670
671enum QuoteSelectionPolicy {
672    ClosestByDistance,
673    WitnessedMedianVoters {
674        voters_by_peer: VotersByPeer,
675        quorum: usize,
676    },
677}
678
679fn witnessed_initial_peers(witnessed: &WitnessedCloseGroup) -> Vec<String> {
680    witnessed
681        .initial_closest
682        .iter()
683        .map(|node| node.peer_id.to_string())
684        .collect()
685}
686
687fn witnessed_responder_views(witnessed: &WitnessedCloseGroup) -> Vec<String> {
688    witnessed
689        .responder_views
690        .iter()
691        .map(|view| {
692            let peers = view
693                .closest
694                .iter()
695                .map(|node| node.peer_id)
696                .collect::<Vec<_>>();
697            format!("{}=>{:?}", view.responder, peer_list(&peers))
698        })
699        .collect()
700}
701
702fn merge_witnessed_node(nodes: &mut HashMap<PeerId, DHTNode>, node: DHTNode) {
703    match nodes.entry(node.peer_id) {
704        std::collections::hash_map::Entry::Occupied(mut entry) => {
705            entry.get_mut().merge_from(node);
706        }
707        std::collections::hash_map::Entry::Vacant(entry) => {
708            entry.insert(node);
709        }
710    }
711}
712
713fn sort_vote_counts_by_distance(vote_counts: &mut [(PeerId, usize)], address: &[u8; 32]) {
714    vote_counts.sort_by(|left, right| {
715        peer_xor_distance(&left.0, address)
716            .cmp(&peer_xor_distance(&right.0, address))
717            .then_with(|| left.0.as_bytes().cmp(right.0.as_bytes()))
718    });
719}
720
721fn witnessed_vote_counts_and_nodes(
722    witnessed: &WitnessedCloseGroup,
723    address: &[u8; 32],
724) -> WitnessedVoteData {
725    let mut known_nodes = HashMap::new();
726    for node in &witnessed.initial_closest {
727        merge_witnessed_node(&mut known_nodes, node.clone());
728    }
729
730    for view in &witnessed.responder_views {
731        for node in &view.closest {
732            merge_witnessed_node(&mut known_nodes, node.clone());
733        }
734    }
735    let views = witnessed
736        .responder_views
737        .iter()
738        .map(|view| crate::quote_policy::ResponderView {
739            responder: view.responder,
740            closest: view.closest.iter().map(|node| node.peer_id).collect(),
741        })
742        .collect::<Vec<_>>();
743    let voters_by_peer = crate::quote_policy::witness_votes(&views);
744
745    let mut vote_counts: Vec<(PeerId, usize)> = voters_by_peer
746        .iter()
747        .map(|(peer_id, voters)| (*peer_id, voters.len()))
748        .collect();
749    sort_vote_counts_by_distance(&mut vote_counts, address);
750    (known_nodes, voters_by_peer, vote_counts)
751}
752
753fn witnessed_consensus_candidates(
754    witnessed: &WitnessedCloseGroup,
755    address: &[u8; 32],
756    quorum: usize,
757) -> Vec<WitnessedQuoteCandidate> {
758    let (known_nodes, voters_by_peer, _) = witnessed_vote_counts_and_nodes(witnessed, address);
759    crate::quote_policy::consensus_peers(&voters_by_peer, address, quorum)
760        .into_iter()
761        .filter_map(|peer| {
762            known_nodes
763                .get(&peer)
764                .cloned()
765                .map(|node| WitnessedQuoteCandidate {
766                    node,
767                    votes: voters_by_peer[&peer].len(),
768                    voters: voters_by_peer[&peer].clone(),
769                })
770        })
771        .collect()
772}
773
774fn witnessed_vote_counts(witnessed: &WitnessedCloseGroup, address: &[u8; 32]) -> Vec<String> {
775    let (_, _, vote_counts) = witnessed_vote_counts_and_nodes(witnessed, address);
776    vote_counts
777        .iter()
778        .map(|(peer_id, votes)| format!("{peer_id}:{votes}"))
779        .collect()
780}
781
782fn witnessed_consensus(
783    witnessed: &WitnessedCloseGroup,
784    address: &[u8; 32],
785    quorum: usize,
786) -> Vec<String> {
787    witnessed_consensus_candidates(witnessed, address, quorum)
788        .iter()
789        .map(|candidate| format!("{}:{}", candidate.node.peer_id, candidate.votes))
790        .collect()
791}
792
793fn witnessed_close_group_diagnostics(
794    address: &[u8; 32],
795    witnessed: &WitnessedCloseGroup,
796    quorum: usize,
797) -> String {
798    format!(
799        "target={}, initial={:?}, responder_views={:?}, vote_counts={:?}, quorum={}, final={:?}",
800        hex::encode(address),
801        witnessed_initial_peers(witnessed),
802        witnessed_responder_views(witnessed),
803        witnessed_vote_counts(witnessed, address),
804        quorum,
805        witnessed_consensus(witnessed, address, quorum)
806    )
807}
808
809fn witnessed_quote_selection_or_error(
810    address: &[u8; 32],
811    witnessed: &WitnessedCloseGroup,
812    required: usize,
813    quorum: usize,
814) -> Result<WitnessedQuoteSelection> {
815    let candidates = witnessed_consensus_candidates(witnessed, address, quorum);
816    crate::quote_policy::validate_witnessed_peers(
817        witnessed.initial_closest.len(),
818        candidates.len(),
819        required,
820    )
821    .map_err(|error| {
822        Error::InsufficientPeers(format!(
823            "{error} {}",
824            witnessed_close_group_diagnostics(address, witnessed, quorum)
825        ))
826    })?;
827
828    let initial_put_peers = witnessed
829        .initial_closest
830        .iter()
831        .take(CLOSE_GROUP_SIZE)
832        .map(|node| (node.peer_id, node.addresses_by_priority()))
833        .collect::<Vec<_>>();
834
835    let quote_peers = candidates
836        .into_iter()
837        .map(|candidate| WitnessedQuotePeer {
838            peer_id: candidate.node.peer_id,
839            addrs: candidate.node.addresses_by_priority(),
840            voters: candidate.voters,
841        })
842        .collect();
843
844    Ok(WitnessedQuoteSelection {
845        quote_peers,
846        initial_put_peers,
847        quorum,
848    })
849}
850
851pub(crate) fn median_paid_quote_issuer(quotes: &[StoreQuote]) -> Option<(PeerId, Amount)> {
852    let prices = quotes
853        .iter()
854        .map(|(_, _, _, price, _)| *price)
855        .collect::<Vec<_>>();
856    let index = crate::payment_policy::median_quote_index(&prices)?;
857    let (peer_id, _, _, price, _) = &quotes[index];
858    Some((*peer_id, *price))
859}
860
861fn sort_quotes_by_distance(quotes: &mut [StoreQuote], address: &[u8; 32]) {
862    quotes.sort_by(|left, right| {
863        peer_xor_distance(&left.0, address)
864            .cmp(&peer_xor_distance(&right.0, address))
865            .then_with(|| left.0.as_bytes().cmp(right.0.as_bytes()))
866    });
867}
868
869fn select_closest_quotes(mut quotes: Vec<StoreQuote>, address: &[u8; 32]) -> Vec<StoreQuote> {
870    sort_quotes_by_distance(&mut quotes, address);
871    quotes.truncate(CLOSE_GROUP_SIZE);
872    quotes
873}
874
875fn select_witnessed_median_voter_quotes(
876    quotes: Vec<StoreQuote>,
877    address: &[u8; 32],
878    voters_by_peer: &VotersByPeer,
879    required_support: usize,
880) -> Option<Vec<StoreQuote>> {
881    let prices = quotes
882        .iter()
883        .map(|quote| (quote.0, quote.3))
884        .collect::<Vec<_>>();
885    let indices = crate::quote_policy::select_witnessed_quotes(
886        &prices,
887        address,
888        voters_by_peer,
889        required_support,
890    )?;
891    Some(
892        indices
893            .into_iter()
894            .map(|index| quotes[index].clone())
895            .collect(),
896    )
897}
898
899fn put_peers_with_median_voters_first(
900    quotes: &[StoreQuote],
901    put_peers: &[(PeerId, Vec<MultiAddr>)],
902    voters_by_peer: &VotersByPeer,
903    required_support: usize,
904) -> Option<Vec<(PeerId, Vec<MultiAddr>)>> {
905    let (median_peer_id, _) = median_paid_quote_issuer(quotes)?;
906    let keys = put_peers.iter().map(|peer| peer.0).collect::<Vec<_>>();
907    let ordered = crate::quote_policy::order_put_peers(
908        median_peer_id,
909        &keys,
910        voters_by_peer,
911        required_support,
912    )?;
913    let addresses = put_peers
914        .iter()
915        .map(|(peer, addrs)| (*peer, addrs))
916        .collect::<HashMap<_, _>>();
917    Some(
918        ordered
919            .into_iter()
920            .map(|peer| (peer, addresses[&peer].clone()))
921            .collect(),
922    )
923}
924
925impl Client {
926    /// Get storage quotes from the closest peers for a given address.
927    ///
928    /// Builds a quorum-witnessed candidate set, still attempts to collect the
929    /// close-group quote count, and returns the largest supported successful
930    /// quote set. The single-node path now only requires one valid quote to
931    /// proceed, but still pays the median quote from the selected set when more
932    /// quotes were successfully fetched.
933    ///
934    /// Returns `Error::AlreadyStored` early if `CLOSE_GROUP_MAJORITY` peers
935    /// report the chunk is already stored.
936    ///
937    /// # Errors
938    ///
939    /// Returns an error if insufficient quotes can be collected.
940    pub async fn get_store_quotes(
941        &self,
942        address: &[u8; 32],
943        data_size: u64,
944        data_type: u32,
945    ) -> Result<Vec<StoreQuote>> {
946        Ok(self
947            .get_store_quote_plan(address, data_size, data_type)
948            .await?
949            .quotes)
950    }
951
952    /// Get storage quotes plus PUT targets ordered for paid-median acceptance.
953    ///
954    /// Quote order is preserved for proof construction because tied quote
955    /// prices rely on stable median selection. PUT target order is separate:
956    /// peers that voted for the paid median issuer are placed first so the
957    /// initial write wave is locally acceptable to a storage majority.
958    pub(crate) async fn get_store_quote_plan(
959        &self,
960        address: &[u8; 32],
961        data_size: u64,
962        data_type: u32,
963    ) -> Result<StoreQuotePlan> {
964        let witnessed_selection = self.select_witnessed_quote_selection(address).await?;
965        let voters_by_peer: VotersByPeer = witnessed_selection
966            .quote_peers
967            .iter()
968            .map(|peer| (peer.peer_id, peer.voters.clone()))
969            .collect();
970        let remote_peers = witnessed_selection
971            .quote_peers
972            .into_iter()
973            .map(|peer| (peer.peer_id, peer.addrs))
974            .collect();
975        let initial_put_peers = witnessed_selection.initial_put_peers;
976        let quorum = witnessed_selection.quorum;
977        let quotes = self
978            .collect_store_quotes_from_remote_peers(
979                address,
980                data_size,
981                data_type,
982                remote_peers,
983                QuoteSelectionPolicy::WitnessedMedianVoters {
984                    voters_by_peer: voters_by_peer.clone(),
985                    quorum,
986                },
987            )
988            .await?;
989        let put_peers = put_peers_with_median_voters_first(
990            &quotes,
991            &initial_put_peers,
992            &voters_by_peer,
993            quorum,
994        )
995        .ok_or_else(|| {
996            Error::InsufficientPeers(format!(
997                "Collected {} witnessed quotes, but fewer than {} initial witness PUT peers \
998                 voted for the paid median issuer for {}",
999                quotes.len(),
1000                quorum,
1001                hex::encode(address)
1002            ))
1003        })?;
1004
1005        Ok(StoreQuotePlan { quotes, put_peers })
1006    }
1007
1008    /// Get storage quotes with the previous over-query behaviour.
1009    ///
1010    /// Merkle preflight uses quote responses only as an already-stored probe;
1011    /// the actual payment still happens through merkle candidate pools. Keep
1012    /// the extra peer buffer there so merkle upload behaviour remains
1013    /// unchanged when a few peers are slow or return unusable quote bindings.
1014    pub(crate) async fn get_store_quotes_with_fault_tolerance(
1015        &self,
1016        address: &[u8; 32],
1017        data_size: u64,
1018        data_type: u32,
1019    ) -> Result<Vec<StoreQuote>> {
1020        let peer_query_count = fault_tolerant_quote_query_count();
1021        let remote_peers = self
1022            .network()
1023            .find_closest_peers(address, peer_query_count)
1024            .await?;
1025
1026        self.collect_store_quotes_from_remote_peers(
1027            address,
1028            data_size,
1029            data_type,
1030            remote_peers,
1031            QuoteSelectionPolicy::ClosestByDistance,
1032        )
1033        .await
1034    }
1035
1036    async fn select_witnessed_quote_selection(
1037        &self,
1038        address: &[u8; 32],
1039    ) -> Result<WitnessedQuoteSelection> {
1040        // Query the close-group width, but single-node payment now only needs
1041        // one valid witnessed quote to proceed.
1042        let required_quotes = SINGLE_NODE_MIN_QUOTE_COUNT;
1043        // Contact the closest PUT_TARGET_WIDTH peers directly so the whole
1044        // PUT-target set's addresses arrive in this single query. A network
1045        // with fewer than that near the target can't satisfy the wide lookup,
1046        // so fall back to the close-group width — the upload still proceeds with
1047        // a narrower (but valid) PUT-target set rather than failing.
1048        let witnessed = crate::quote_policy::discover_put_peers(
1049            |width| async move {
1050                if width == CLOSE_GROUP_SIZE {
1051                    debug!(target = %hex::encode(address), "Retrying witnessed discovery at close-group width");
1052                }
1053                self.network().find_witnessed_close_group_with_view_count(
1054                    address, width, SINGLE_NODE_WITNESSED_VIEW_COUNT,
1055                ).await
1056            },
1057            |witnessed| witnessed.initial_closest.len(),
1058            |found, width| Error::InsufficientPeers(format!("Witnessed close group initial lookup found {found} peers, need {width}")),
1059        ).await.map_err(|e| Error::InsufficientPeers(format!(
1060            "Witnessed close group lookup failed before payment for target {}: {e}", hex::encode(address),
1061        )))?;
1062        // Run quoting/quorum on the closest CLOSE_GROUP_SIZE only, so payment
1063        // semantics are unaffected by the wider PUT query.
1064        let witnessed_quote = scope_witnessed_to_close_group(&witnessed);
1065        let base_quorum = witnessed_close_group_quorum();
1066        let missing_views = missing_witnessed_responder_views(&witnessed_quote);
1067        let quorum = witnessed_close_group_quorum_for_transcript(&witnessed_quote);
1068
1069        if missing_views > 0 {
1070            warn!(
1071                target = %hex::encode(address),
1072                initial = witnessed_quote.initial_closest.len(),
1073                responder_views = witnessed_quote.responder_views.len(),
1074                missing_views = missing_views,
1075                base_quorum = base_quorum,
1076                adjusted_quorum = quorum,
1077                "Witnessed close group transcript is missing responder views; lowering SNP witness quorum"
1078            );
1079        }
1080
1081        debug!(
1082            target = %hex::encode(address),
1083            quorum = quorum,
1084            view_count = SINGLE_NODE_WITNESSED_VIEW_COUNT,
1085            initial = ?witnessed_initial_peers(&witnessed_quote),
1086            responder_views = ?witnessed_responder_views(&witnessed_quote),
1087            vote_counts = ?witnessed_vote_counts(&witnessed_quote, address),
1088            final_witnessed_set = ?witnessed_consensus(&witnessed_quote, address, quorum),
1089            "Witnessed close group selected for SNP quote collection"
1090        );
1091
1092        let mut selection =
1093            witnessed_quote_selection_or_error(address, &witnessed_quote, required_quotes, quorum)?;
1094        // Widen the PUT-target set to the closest PUT_TARGET_WIDTH
1095        // directly-contacted peers; the quote set above stays the closest
1096        // CLOSE_GROUP_SIZE. The same proof is reused on all of them.
1097        selection.initial_put_peers = witnessed
1098            .initial_closest
1099            .iter()
1100            .take(PUT_TARGET_WIDTH)
1101            .map(|node| (node.peer_id, node.addresses_by_priority()))
1102            .collect();
1103        Ok(selection)
1104    }
1105
1106    #[allow(clippy::too_many_lines)]
1107    async fn collect_store_quotes_from_remote_peers(
1108        &self,
1109        address: &[u8; 32],
1110        data_size: u64,
1111        data_type: u32,
1112        remote_peers: Vec<(PeerId, Vec<MultiAddr>)>,
1113        quote_selection_policy: QuoteSelectionPolicy,
1114    ) -> Result<Vec<StoreQuote>> {
1115        let peer_query_count = remote_peers.len();
1116
1117        let node = self.network();
1118
1119        debug!(
1120            "Requesting quotes from up to {peer_query_count} peers for address {} (size: {data_size})",
1121            hex::encode(address)
1122        );
1123
1124        let (min_quote_count, target_quote_count, staged_witnessed_collection) =
1125            match &quote_selection_policy {
1126                QuoteSelectionPolicy::ClosestByDistance => {
1127                    (CLOSE_GROUP_SIZE, CLOSE_GROUP_SIZE, false)
1128                }
1129                QuoteSelectionPolicy::WitnessedMedianVoters { .. } => (
1130                    SINGLE_NODE_MIN_QUOTE_COUNT,
1131                    single_node_quote_query_count(),
1132                    true,
1133                ),
1134            };
1135        let target_quote_count = target_quote_count.min(peer_query_count);
1136
1137        if remote_peers.len() < min_quote_count {
1138            return Err(Error::InsufficientPeers(format!(
1139                "Found {} peers, need {min_quote_count}",
1140                remote_peers.len(),
1141            )));
1142        }
1143        debug_assert!(peer_query_count >= min_quote_count);
1144
1145        let per_peer_timeout = Duration::from_secs(self.config().quote_timeout_secs);
1146        let overall_timeout = Duration::from_secs(QUOTE_COLLECTION_TIMEOUT_SECS);
1147
1148        // Collect quote responses. SNP/witnessed collection deliberately tries
1149        // the closest witnessed peers first and only falls back to further
1150        // witnessed peers when a closer peer fails to produce a usable quote.
1151        let mut quotes = Vec::with_capacity(peer_query_count);
1152        let mut already_stored_peers: Vec<(PeerId, [u8; 32])> = Vec::new();
1153        let mut failures: Vec<String> = Vec::new();
1154
1155        // Track storer-rejecting peers separately (binding, content, signature
1156        // failures) so we can surface their count in diagnostics — they're a
1157        // special class of failure (peer misconfigured or hostile, not
1158        // network-broken) and the user benefits from seeing them called out.
1159        let mut bad_quote_count = 0usize;
1160
1161        // A storer's verdict that this client cannot settle, kept outside the
1162        // collection loops so neither the overall timeout nor an early exit
1163        // can discard it. Checked before any quote plan is built.
1164        let mut settlement_refusal: Option<Error> = None;
1165        let refusals = self.settlement_refusals();
1166
1167        if staged_witnessed_collection {
1168            let mut quote_futures = FuturesUnordered::new();
1169            let mut next_peer_index = 0usize;
1170            let collect_result: std::result::Result<std::result::Result<(), Error>, _> =
1171                crate::runtime::timeout(overall_timeout, async {
1172                    loop {
1173                        // Stop launching once the target is met, but keep
1174                        // draining below. Peers already in flight may yet
1175                        // declare this client unable to settle, and dropping
1176                        // that verdict because faster peers filled the quota
1177                        // would make it depend on response order. The surplus
1178                        // quotes are discarded; a refusal among them is not.
1179                        let launch_count = if quotes.len() >= target_quote_count {
1180                            0
1181                        } else {
1182                            witnessed_quote_launch_budget(
1183                                quotes.len(),
1184                                quote_futures.len(),
1185                                remote_peers.len().saturating_sub(next_peer_index),
1186                            )
1187                        };
1188                        for _ in 0..launch_count {
1189                            let (peer_id, peer_addrs) = &remote_peers[next_peer_index];
1190                            next_peer_index += 1;
1191                            quote_futures.push(request_store_quote_from_peer(
1192                                node.clone(),
1193                                *peer_id,
1194                                peer_addrs.clone(),
1195                                self.next_request_id(),
1196                                *address,
1197                                data_size,
1198                                data_type,
1199                                per_peer_timeout,
1200                                self.unversioned_quote_peers(),
1201                                self.versioned_quote_capable_handle(),
1202                            ));
1203                        }
1204
1205                        if quote_futures.is_empty() {
1206                            break;
1207                        }
1208
1209                        let Some((peer_id, addrs, quote_result)) = quote_futures.next().await
1210                        else {
1211                            break;
1212                        };
1213                        record_store_quote_result(
1214                            peer_id,
1215                            addrs,
1216                            quote_result,
1217                            address,
1218                            &mut quotes,
1219                            &mut already_stored_peers,
1220                            &mut failures,
1221                            &mut bad_quote_count,
1222                            &mut settlement_refusal,
1223                            &refusals,
1224                        )?;
1225                    }
1226                    Ok(())
1227                })
1228                .await;
1229
1230            match collect_result {
1231                Err(_elapsed) => {
1232                    warn!(
1233                        "Quote collection timed out after {overall_timeout:?} for address {}",
1234                        hex::encode(address)
1235                    );
1236                }
1237                Ok(Err(e)) => return Err(e),
1238                Ok(Ok(())) => {}
1239            }
1240            // Outranks the timeout: a refusal says paying is unsafe no matter
1241            // how many quotes were gathered before the clock ran out.
1242            if let Some(refusal) = settlement_refusal.take() {
1243                return Err(refusal);
1244            }
1245        } else {
1246            // Merkle preflight keeps the previous behaviour: query the full
1247            // over-query set concurrently because those quote responses are
1248            // only used as an already-stored probe.
1249            let mut quote_futures = FuturesUnordered::new();
1250
1251            for (peer_id, peer_addrs) in &remote_peers {
1252                quote_futures.push(request_store_quote_from_peer(
1253                    node.clone(),
1254                    *peer_id,
1255                    peer_addrs.clone(),
1256                    self.next_request_id(),
1257                    *address,
1258                    data_size,
1259                    data_type,
1260                    per_peer_timeout,
1261                    self.unversioned_quote_peers(),
1262                    self.versioned_quote_capable_handle(),
1263                ));
1264            }
1265
1266            let collect_result: std::result::Result<std::result::Result<(), Error>, _> =
1267                crate::runtime::timeout(overall_timeout, async {
1268                    while let Some((peer_id, addrs, quote_result)) = quote_futures.next().await {
1269                        record_store_quote_result(
1270                            peer_id,
1271                            addrs,
1272                            quote_result,
1273                            address,
1274                            &mut quotes,
1275                            &mut already_stored_peers,
1276                            &mut failures,
1277                            &mut bad_quote_count,
1278                            &mut settlement_refusal,
1279                            &refusals,
1280                        )?;
1281                    }
1282                    Ok(())
1283                })
1284                .await;
1285
1286            match collect_result {
1287                Err(_elapsed) => {
1288                    warn!(
1289                        "Quote collection timed out after {overall_timeout:?} for address {}",
1290                        hex::encode(address)
1291                    );
1292                    // Fall through to check if we have enough quotes despite timeout.
1293                    // The timeout fires when slow peers haven't responded yet, but we
1294                    // may already have enough successful quotes from fast peers.
1295                }
1296                Ok(Err(e)) => return Err(e),
1297                Ok(Ok(())) => {}
1298            }
1299            // Outranks the timeout: a refusal says paying is unsafe no matter
1300            // how many quotes were gathered before the clock ran out.
1301            if let Some(refusal) = settlement_refusal.take() {
1302                return Err(refusal);
1303            }
1304        }
1305
1306        // Defensive double-check: the per-peer handler already filters
1307        // bad-binding responses into `failures`, but if any path slipped a bad
1308        // quote into `quotes` (e.g. a future refactor) this catches it before
1309        // we sort by distance and return. `bad_dropped` should be 0 in normal
1310        // operation; non-zero indicates an upstream regression worth investigating.
1311        let bad_dropped = drop_quotes_with_bad_bindings(&mut quotes);
1312        if bad_dropped > 0 {
1313            warn!(
1314                "Defensive filter dropped {bad_dropped} quotes with mismatched peer bindings \
1315                 for address {} — the per-peer handler should have caught these earlier \
1316                 (this indicates an upstream regression)",
1317                hex::encode(address),
1318            );
1319            bad_quote_count += bad_dropped;
1320        }
1321
1322        if crate::quote_policy::already_stored(
1323            quotes.iter().map(|quote| quote.0),
1324            already_stored_peers.iter().map(|peer| peer.0),
1325            address,
1326        ) {
1327            debug!(
1328                "Chunk {} already stored on a close-group majority",
1329                hex::encode(address)
1330            );
1331            return Err(Error::AlreadyStored);
1332        }
1333
1334        let already_stored_count = already_stored_peers.len();
1335        let failure_count = failures.len();
1336        let quote_count = quotes.len();
1337        let total_responses = quote_count + failure_count + already_stored_count;
1338
1339        if quotes.len() >= min_quote_count {
1340            let selected_quotes = match quote_selection_policy {
1341                QuoteSelectionPolicy::ClosestByDistance => select_closest_quotes(quotes, address),
1342                QuoteSelectionPolicy::WitnessedMedianVoters {
1343                    voters_by_peer,
1344                    quorum,
1345                } => select_witnessed_median_voter_quotes(quotes, address, &voters_by_peer, quorum)
1346                    .ok_or_else(|| {
1347                        Error::InsufficientPeers(format!(
1348                            "Got {quote_count} quotes, need at least {min_quote_count} whose paid \
1349                                 median issuer is recognised by at least {} \
1350                                 selected witness peers ({total_responses} responses: \
1351                                 {already_stored_count} already_stored, {failure_count} failed \
1352                                 including {bad_quote_count} with mismatched peer bindings). \
1353                                 Failures: [{}]",
1354                            quorum,
1355                            failures.join("; ")
1356                        ))
1357                    })?,
1358            };
1359
1360            info!(
1361                "Collected {} quotes for address {} ({total_responses} responses: \
1362                 {quote_count} ok, {already_stored_count} already_stored, {failure_count} failed, \
1363                 {bad_quote_count} bad-binding)",
1364                selected_quotes.len(),
1365                hex::encode(address),
1366            );
1367            return Ok(selected_quotes);
1368        }
1369
1370        Err(Error::InsufficientPeers(format!(
1371            "Got {quote_count} quotes, need {min_quote_count} ({total_responses} responses: \
1372             {already_stored_count} already_stored, {failure_count} failed including \
1373             {bad_quote_count} with mismatched peer bindings). Failures: [{}]",
1374            failures.join("; ")
1375        )))
1376    }
1377}
1378
1379#[cfg(test)]
1380#[allow(clippy::unwrap_used, clippy::expect_used)]
1381mod tests {
1382    //! Test fixtures use real ML-DSA-65 keypairs (1952-byte public keys), the
1383    //! same key material that ships on the wire. The "bad" quote is built by
1384    //! **swapping** the public key field with a different real keypair's
1385    //! public key — the exact shape produced by the Apr 30 production
1386    //! failure (an operator running two co-located identities with crossed
1387    //! quote-signing keys). Signatures are not exercised here because this
1388    //! filter only mirrors `validate_peer_bindings` (BLAKE3 binding); see
1389    //! the doc-comment on `quote_binding_is_valid` for why
1390    //! `verify_quote_signature` and `verify_quote_content` are deliberately
1391    //! NOT mirrored.
1392
1393    use super::*;
1394    use ant_protocol::evm::RewardsAddress;
1395    use ant_protocol::pqc::ops::{MlDsaOperations, MlDsaPublicKey};
1396    use ant_protocol::transport::{DHTNode, MlDsa65, ResponderView, WitnessedCloseGroup};
1397    use std::time::SystemTime;
1398    use xor_name::XorName;
1399
1400    /// A real ML-DSA-65 keypair plus its derived peer ID.
1401    struct Keypair {
1402        peer_id: PeerId,
1403        pub_key_bytes: Vec<u8>,
1404        secret_key_bytes: Vec<u8>,
1405    }
1406
1407    fn gen_keypair() -> Keypair {
1408        let ml_dsa = MlDsa65::new();
1409        let (pub_key, sk) = ml_dsa.generate_keypair().expect("ML-DSA-65 keygen");
1410        let pub_key_bytes = pub_key.as_bytes().to_vec();
1411        let peer_id = PeerId::from_bytes(compute_address(&pub_key_bytes));
1412        Keypair {
1413            peer_id,
1414            pub_key_bytes,
1415            secret_key_bytes: sk.as_bytes().to_vec(),
1416        }
1417    }
1418
1419    /// Build a PROPERLY-SIGNED baseline quote for `content`, signed by a real
1420    /// ML-DSA-65 key whose `BLAKE3(pub_key)` is the returned peer id. Passes the
1421    /// client's full classifier gate (binding + content + signature + price).
1422    fn signed_baseline_quote(content: [u8; 32]) -> (PeerId, PaymentQuote) {
1423        use ant_protocol::pqc::ops::MlDsaSecretKey;
1424        let kp = gen_keypair();
1425        let mut quote = PaymentQuote {
1426            content: XorName(content),
1427            timestamp: SystemTime::UNIX_EPOCH,
1428            price: calculate_price(0),
1429            rewards_address: RewardsAddress::new([0u8; 20]),
1430            pub_key: kp.pub_key_bytes.clone(),
1431            signature: Vec::new(),
1432            committed_key_count: 0,
1433            commitment_pin: None,
1434        };
1435        let ml_dsa = MlDsa65::new();
1436        let sk = MlDsaSecretKey::from_bytes(&kp.secret_key_bytes).expect("sk");
1437        let msg = quote.bytes_for_sig();
1438        quote.signature = ml_dsa.sign(&sk, &msg).expect("sign").as_bytes().to_vec();
1439        (kp.peer_id, quote)
1440    }
1441
1442    /// Build a quote tuple whose `pub_key` correctly hashes to its peer_id.
1443    /// Signature is left empty: this filter does not verify signatures.
1444    ///
1445    /// The quote is a valid ADR-0004 **baseline**: `(0, None)` priced at
1446    /// `calculate_price(0)`, so it passes the forced-price gate in
1447    /// `classify_quote_response`. The 5th tuple element is the (absent)
1448    /// commitment sidecar.
1449    fn good_quote_real() -> QuotedPeer {
1450        let kp = gen_keypair();
1451        let quote = PaymentQuote {
1452            content: XorName([0u8; 32]),
1453            timestamp: SystemTime::UNIX_EPOCH,
1454            price: calculate_price(0),
1455            rewards_address: RewardsAddress::new([0u8; 20]),
1456            pub_key: kp.pub_key_bytes,
1457            signature: Vec::new(),
1458            committed_key_count: 0,
1459            commitment_pin: None,
1460        };
1461        (kp.peer_id, Vec::new(), quote, calculate_price(0), None)
1462    }
1463
1464    /// Build a quote tuple where the quote carries a different keypair's
1465    /// `pub_key` than the peer_id derives from. Mirrors the production
1466    /// failure shape: peer A advertised on the transport, but the quote
1467    /// carries peer B's key.
1468    fn bad_quote_real() -> QuotedPeer {
1469        let claimed = gen_keypair();
1470        let signing = gen_keypair();
1471        assert_ne!(claimed.pub_key_bytes, signing.pub_key_bytes);
1472        assert_ne!(claimed.peer_id.as_bytes(), signing.peer_id.as_bytes());
1473        let quote = PaymentQuote {
1474            content: XorName([0u8; 32]),
1475            timestamp: SystemTime::UNIX_EPOCH,
1476            price: calculate_price(0),
1477            rewards_address: RewardsAddress::new([0u8; 20]),
1478            pub_key: signing.pub_key_bytes,
1479            signature: Vec::new(),
1480            committed_key_count: 0,
1481            commitment_pin: None,
1482        };
1483        (claimed.peer_id, Vec::new(), quote, calculate_price(0), None)
1484    }
1485
1486    fn witnessed_test_node(seed: u8) -> DHTNode {
1487        DHTNode {
1488            peer_id: PeerId::from_bytes([seed; 32]),
1489            addresses: Vec::new(),
1490            address_types: Vec::new(),
1491            distance: None,
1492            reliability: 1.0,
1493            address_authority: None,
1494        }
1495    }
1496
1497    fn witnessed_test_nodes(seeds: &[u8]) -> Vec<DHTNode> {
1498        seeds.iter().copied().map(witnessed_test_node).collect()
1499    }
1500
1501    fn witnessed_test_view(responder: u8, closest: &[u8]) -> ResponderView {
1502        ResponderView {
1503            responder: PeerId::from_bytes([responder; 32]),
1504            closest: witnessed_test_nodes(closest),
1505        }
1506    }
1507
1508    fn synthetic_peer(seed: u8) -> PeerId {
1509        PeerId::from_bytes([seed; 32])
1510    }
1511
1512    fn synthetic_quote(
1513        seed: u8,
1514        price: u64,
1515    ) -> (
1516        PeerId,
1517        Vec<MultiAddr>,
1518        PaymentQuote,
1519        Amount,
1520        Option<Vec<u8>>,
1521    ) {
1522        let amount = Amount::from(price);
1523        let quote = PaymentQuote {
1524            content: XorName([0u8; 32]),
1525            timestamp: SystemTime::UNIX_EPOCH,
1526            price: amount,
1527            rewards_address: RewardsAddress::new([0u8; 20]),
1528            pub_key: Vec::new(),
1529            signature: Vec::new(),
1530            committed_key_count: 0,
1531            commitment_pin: None,
1532        };
1533        (synthetic_peer(seed), Vec::new(), quote, amount, None)
1534    }
1535
1536    fn synthetic_voters(seeds: &[u8]) -> HashSet<PeerId> {
1537        seeds.iter().copied().map(synthetic_peer).collect()
1538    }
1539
1540    fn quote_peer_seeds(quotes: &[StoreQuote]) -> Vec<u8> {
1541        quotes
1542            .iter()
1543            .map(|(peer_id, _, _, _, _)| peer_id.as_bytes()[0])
1544            .collect()
1545    }
1546
1547    fn put_peer_seeds(peers: &[(PeerId, Vec<MultiAddr>)]) -> Vec<u8> {
1548        peers
1549            .iter()
1550            .map(|(peer_id, _)| peer_id.as_bytes()[0])
1551            .collect()
1552    }
1553
1554    fn put_peers_from_seeds(seeds: &[u8]) -> Vec<(PeerId, Vec<MultiAddr>)> {
1555        seeds
1556            .iter()
1557            .copied()
1558            .map(|seed| (synthetic_peer(seed), Vec::new()))
1559            .collect()
1560    }
1561
1562    /// Independent re-implementation of the storer-side binding spec
1563    /// (`ant-node/src/payment/verifier.rs::validate_peer_bindings` +
1564    /// `peer_id_from_public_key_bytes`):
1565    /// (a) `pub_key` parses as ML-DSA-65 (length 1952), and
1566    /// (b) `BLAKE3(pub_key) == peer_id`.
1567    ///
1568    /// Re-derived from spec, NOT delegating to `quote_binding_is_valid`,
1569    /// so cross-checks are not "function == itself".
1570    fn storer_binding_would_accept(peer_id: &PeerId, quote: &PaymentQuote) -> bool {
1571        if MlDsaPublicKey::from_bytes(&quote.pub_key).is_err() {
1572            return false;
1573        }
1574        compute_address(&quote.pub_key) == *peer_id.as_bytes()
1575    }
1576
1577    // ============================================================
1578    // Tests for `quote_binding_is_valid` (the predicate)
1579    // ============================================================
1580
1581    #[test]
1582    fn binding_accepts_real_self_consistent_keypair() {
1583        let (peer_id, _, quote, _, _) = good_quote_real();
1584        // Property under test: the predicate accepts a quote whose pub_key
1585        // genuinely belongs to the claimed peer.
1586        assert!(quote_binding_is_valid(&peer_id, &quote));
1587        // Cross-check against the independent full storer-spec implementation.
1588        assert!(storer_binding_would_accept(&peer_id, &quote));
1589    }
1590
1591    #[test]
1592    fn binding_rejects_real_crossed_keypair() {
1593        let (peer_id, _, quote, _, _) = bad_quote_real();
1594        assert!(!quote_binding_is_valid(&peer_id, &quote));
1595        assert!(!storer_binding_would_accept(&peer_id, &quote));
1596    }
1597
1598    #[test]
1599    fn binding_rejects_oversize_pubkey() {
1600        // A pub_key longer than ML-DSA-65 (1952 bytes) must be rejected
1601        // even if BLAKE3 happens to agree, because the storer rejects on
1602        // length first via `peer_id_from_public_key_bytes`.
1603        let oversized = vec![0u8; ML_DSA_PUB_KEY_LEN + 1];
1604        let peer_id = PeerId::from_bytes(compute_address(&oversized));
1605        let quote = PaymentQuote {
1606            content: XorName([0u8; 32]),
1607            timestamp: SystemTime::UNIX_EPOCH,
1608            price: Amount::ZERO,
1609            rewards_address: RewardsAddress::new([0u8; 20]),
1610            pub_key: oversized,
1611            signature: Vec::new(),
1612            committed_key_count: 0,
1613            commitment_pin: None,
1614        };
1615        // BLAKE3(pub_key) DOES equal the peer_id we constructed, so the
1616        // bare hash check would pass — but the length guard must reject.
1617        assert_eq!(compute_address(&quote.pub_key), *peer_id.as_bytes());
1618        assert!(
1619            !quote_binding_is_valid(&peer_id, &quote),
1620            "predicate must reject oversize pub_key even when BLAKE3 happens to match"
1621        );
1622        assert!(!storer_binding_would_accept(&peer_id, &quote));
1623    }
1624
1625    #[test]
1626    fn binding_rejects_undersize_pubkey() {
1627        let undersized = vec![0u8; ML_DSA_PUB_KEY_LEN - 1];
1628        let peer_id = PeerId::from_bytes(compute_address(&undersized));
1629        let quote = PaymentQuote {
1630            content: XorName([0u8; 32]),
1631            timestamp: SystemTime::UNIX_EPOCH,
1632            price: Amount::ZERO,
1633            rewards_address: RewardsAddress::new([0u8; 20]),
1634            pub_key: undersized,
1635            signature: Vec::new(),
1636            committed_key_count: 0,
1637            commitment_pin: None,
1638        };
1639        assert!(!quote_binding_is_valid(&peer_id, &quote));
1640        assert!(!storer_binding_would_accept(&peer_id, &quote));
1641    }
1642
1643    // ============================================================
1644    // Tests for the filter (`drop_quotes_with_bad_bindings`)
1645    // ============================================================
1646
1647    #[test]
1648    fn quote_query_counts_keep_single_node_close_group_only() {
1649        assert_eq!(single_node_quote_query_count(), CLOSE_GROUP_SIZE);
1650        assert_eq!(SINGLE_NODE_MIN_QUOTE_COUNT, 1);
1651        assert_eq!(SINGLE_NODE_WITNESSED_VIEW_COUNT, 20);
1652        assert!(SINGLE_NODE_WITNESSED_VIEW_COUNT > single_node_quote_query_count());
1653        assert_eq!(witnessed_close_group_quorum(), 5);
1654        assert_eq!(witnessed_close_group_quorum_for_missing_views(0), 5);
1655        assert_eq!(witnessed_close_group_quorum_for_missing_views(1), 4);
1656        assert_eq!(witnessed_close_group_quorum_for_missing_views(2), 3);
1657        assert_eq!(
1658            fault_tolerant_quote_query_count(),
1659            CLOSE_GROUP_SIZE * FAULT_TOLERANT_QUOTE_QUERY_MULTIPLIER
1660        );
1661        assert!(fault_tolerant_quote_query_count() > single_node_quote_query_count());
1662    }
1663
1664    #[test]
1665    fn witnessed_quote_launch_budget_keeps_exact_quote_window() {
1666        assert_eq!(
1667            witnessed_quote_launch_budget(0, 0, CLOSE_GROUP_SIZE * 2),
1668            CLOSE_GROUP_SIZE,
1669            "initial SNP quote fetch should launch the closest seven peers"
1670        );
1671        assert_eq!(
1672            witnessed_quote_launch_budget(1, CLOSE_GROUP_SIZE - 1, CLOSE_GROUP_SIZE),
1673            0,
1674            "a successful quote should not launch an extra fallback"
1675        );
1676        assert_eq!(
1677            witnessed_quote_launch_budget(0, CLOSE_GROUP_SIZE - 1, CLOSE_GROUP_SIZE),
1678            1,
1679            "a failed in-flight quote should launch the next closest fallback"
1680        );
1681        assert_eq!(
1682            witnessed_quote_launch_budget(CLOSE_GROUP_SIZE - 1, 0, 3),
1683            1,
1684            "only one more peer is needed for the seventh quote"
1685        );
1686        assert_eq!(
1687            witnessed_quote_launch_budget(0, 0, CLOSE_GROUP_SIZE - 1),
1688            CLOSE_GROUP_SIZE - 1,
1689            "launch budget is capped by remaining candidates"
1690        );
1691    }
1692
1693    #[test]
1694    fn witnessed_candidates_sort_by_xor_distance_then_votes() {
1695        let address = [0u8; 32];
1696        let witnessed = WitnessedCloseGroup {
1697            target: address,
1698            k: CLOSE_GROUP_SIZE,
1699            initial_closest: witnessed_test_nodes(&[1, 2, 3, 4, 5, 6, 7]),
1700            responder_views: vec![
1701                witnessed_test_view(1, &[1, 9]),
1702                witnessed_test_view(2, &[1, 9]),
1703                witnessed_test_view(3, &[1, 9]),
1704                witnessed_test_view(4, &[1, 9]),
1705                witnessed_test_view(5, &[1, 9]),
1706                witnessed_test_view(6, &[9]),
1707                witnessed_test_view(7, &[9]),
1708            ],
1709        };
1710
1711        let candidates =
1712            witnessed_consensus_candidates(&witnessed, &address, witnessed_close_group_quorum());
1713
1714        assert_eq!(
1715            candidates
1716                .iter()
1717                .map(|candidate| candidate.node.peer_id.as_bytes()[0])
1718                .collect::<Vec<_>>(),
1719            vec![1, 9],
1720            "XOR closeness must be the primary sort before quote collection"
1721        );
1722    }
1723
1724    /// Ascending seeds `1..=count`, each a valid `u8` peer seed.
1725    fn ascending_seeds(count: usize) -> Vec<u8> {
1726        (1..=count)
1727            .map(|n| u8::try_from(n).expect("test seed fits in u8"))
1728            .collect()
1729    }
1730
1731    #[test]
1732    fn scope_witnessed_to_close_group_matches_native_close_group_query() {
1733        // How many of the closest-`CLOSE_GROUP_SIZE` responders returned a view.
1734        // The remainder are "missing", so the scoped transcript also exercises
1735        // the missing-views quorum adjustment.
1736        const RESPONDED_IN_SCOPE: usize = 5;
1737        // Responders past the close group whose views scoping must drop.
1738        const OUT_OF_SCOPE_RESPONDERS: usize = 2;
1739
1740        let address = [0u8; 32];
1741        let close_seeds = ascending_seeds(CLOSE_GROUP_SIZE);
1742        // Each view's closest list mixes in-group (1, 2) and far (8, 9) peers so
1743        // candidate selection is non-trivial and must survive scoping verbatim.
1744        let view_closest = [1, 2, 8, 9];
1745        let in_scope_views = || -> Vec<ResponderView> {
1746            ascending_seeds(RESPONDED_IN_SCOPE)
1747                .into_iter()
1748                .map(|responder| witnessed_test_view(responder, &view_closest))
1749                .collect()
1750        };
1751
1752        // A wide PUT_TARGET_WIDTH-peer transcript, ordered closest-first (seed n
1753        // == PeerId [n; 32], whose XOR distance to the zero address is n). Two
1754        // responders past the close group (out of scope) must be dropped.
1755        let mut wide_views = in_scope_views();
1756        for offset in 1..=OUT_OF_SCOPE_RESPONDERS {
1757            let responder =
1758                u8::try_from(CLOSE_GROUP_SIZE + offset).expect("out-of-scope seed fits in u8");
1759            wide_views.push(witnessed_test_view(responder, &[1, 2, 3]));
1760        }
1761        let wide = WitnessedCloseGroup {
1762            target: address,
1763            k: PUT_TARGET_WIDTH,
1764            initial_closest: witnessed_test_nodes(&ascending_seeds(PUT_TARGET_WIDTH)),
1765            responder_views: wide_views,
1766        };
1767
1768        // The hand-built equivalent: a native CLOSE_GROUP_SIZE-wide query with
1769        // the same in-scope responders.
1770        let native = WitnessedCloseGroup {
1771            target: address,
1772            k: CLOSE_GROUP_SIZE,
1773            initial_closest: witnessed_test_nodes(&close_seeds),
1774            responder_views: in_scope_views(),
1775        };
1776
1777        let scoped = scope_witnessed_to_close_group(&wide);
1778
1779        // Target preserved; k and the initial set collapse to the close group.
1780        assert_eq!(scoped.target, wide.target);
1781        assert_eq!(scoped.k, CLOSE_GROUP_SIZE);
1782        assert_eq!(
1783            scoped
1784                .initial_closest
1785                .iter()
1786                .map(|node| node.peer_id.as_bytes()[0])
1787                .collect::<Vec<_>>(),
1788            close_seeds,
1789            "initial set must be the closest CLOSE_GROUP_SIZE, in order"
1790        );
1791
1792        // Out-of-close-group responder views are dropped; the in-scope ones keep
1793        // their closest lists untouched (scoping filters by responder only).
1794        assert_eq!(
1795            scoped
1796                .responder_views
1797                .iter()
1798                .map(|view| view.responder.as_bytes()[0])
1799                .collect::<Vec<_>>(),
1800            ascending_seeds(RESPONDED_IN_SCOPE),
1801            "only responders inside the close group survive"
1802        );
1803        assert_eq!(
1804            scoped.responder_views[0]
1805                .closest
1806                .iter()
1807                .map(|node| node.peer_id.as_bytes()[0])
1808                .collect::<Vec<_>>(),
1809            view_closest.to_vec(),
1810            "a surviving view's closest set must be preserved verbatim"
1811        );
1812
1813        // The quorum math and candidate consensus run on the close group only
1814        // and are byte-for-byte identical to the native CLOSE_GROUP_SIZE query.
1815        assert_eq!(
1816            missing_witnessed_responder_views(&scoped),
1817            missing_witnessed_responder_views(&native),
1818        );
1819        let quorum = witnessed_close_group_quorum_for_transcript(&scoped);
1820        assert_eq!(quorum, witnessed_close_group_quorum_for_transcript(&native));
1821        let candidate_seeds = |group: &WitnessedCloseGroup| {
1822            witnessed_consensus_candidates(group, &address, quorum)
1823                .iter()
1824                .map(|candidate| candidate.node.peer_id.as_bytes()[0])
1825                .collect::<Vec<_>>()
1826        };
1827        assert_eq!(
1828            candidate_seeds(&scoped),
1829            candidate_seeds(&native),
1830            "scoped consensus must match a native close-group query"
1831        );
1832    }
1833
1834    #[test]
1835    fn witnessed_quote_peers_error_is_typed_and_pre_payment_when_consensus_is_short() {
1836        let address = [0u8; 32];
1837        let responder_views = (1..=7)
1838            .map(|responder| witnessed_test_view(responder, &[1, 2, 3, 4]))
1839            .collect();
1840        let witnessed = WitnessedCloseGroup {
1841            target: address,
1842            k: CLOSE_GROUP_SIZE,
1843            initial_closest: witnessed_test_nodes(&[1, 2, 3, 4, 5, 6, 7]),
1844            responder_views,
1845        };
1846
1847        let err = witnessed_quote_selection_or_error(
1848            &address,
1849            &witnessed,
1850            CLOSE_GROUP_SIZE,
1851            witnessed_close_group_quorum(),
1852        )
1853        .expect_err("short witnessed consensus must fail before payment");
1854
1855        match err {
1856            Error::InsufficientPeers(message) => {
1857                assert!(message.contains("before payment"));
1858                assert!(message.contains("vote_counts"));
1859                assert!(message.contains("quorum"));
1860            }
1861            other => panic!("expected typed InsufficientPeers error, got {other:?}"),
1862        }
1863    }
1864
1865    #[test]
1866    fn witnessed_quote_selection_accepts_one_quorum_recognised_candidate() {
1867        let address = [0u8; 32];
1868        let witnessed = WitnessedCloseGroup {
1869            target: address,
1870            k: CLOSE_GROUP_SIZE,
1871            initial_closest: witnessed_test_nodes(&[1, 2, 3, 4, 5, 6, 7]),
1872            responder_views: (1..=7)
1873                .map(|responder| witnessed_test_view(responder, &[1]))
1874                .collect(),
1875        };
1876
1877        let selection = witnessed_quote_selection_or_error(
1878            &address,
1879            &witnessed,
1880            SINGLE_NODE_MIN_QUOTE_COUNT,
1881            witnessed_close_group_quorum(),
1882        )
1883        .expect("one quorum-recognised candidate is enough before payment");
1884
1885        assert_eq!(
1886            selection
1887                .quote_peers
1888                .iter()
1889                .map(|peer| peer.peer_id.as_bytes()[0])
1890                .collect::<Vec<_>>(),
1891            vec![1]
1892        );
1893        assert_eq!(
1894            put_peer_seeds(&selection.initial_put_peers),
1895            vec![1, 2, 3, 4, 5, 6, 7]
1896        );
1897    }
1898
1899    #[test]
1900    fn witnessed_quote_peers_include_quorum_fallback_candidates() {
1901        const EXTRA_QUORUM_CANDIDATES: usize = 1;
1902
1903        let address = [0u8; 32];
1904        let witnessed = WitnessedCloseGroup {
1905            target: address,
1906            k: CLOSE_GROUP_SIZE,
1907            initial_closest: witnessed_test_nodes(&[1, 2, 3, 4, 5, 6, 7]),
1908            responder_views: vec![
1909                witnessed_test_view(1, &[1, 2, 3, 4, 5, 6, 7]),
1910                witnessed_test_view(2, &[1, 2, 3, 4, 5, 6, 8]),
1911                witnessed_test_view(3, &[1, 2, 3, 4, 5, 7, 8]),
1912                witnessed_test_view(4, &[1, 2, 3, 4, 6, 7, 8]),
1913                witnessed_test_view(5, &[1, 2, 3, 5, 6, 7, 8]),
1914                witnessed_test_view(6, &[1, 2, 4, 5, 6, 7, 8]),
1915                witnessed_test_view(7, &[1, 3, 4, 5, 6, 7, 8]),
1916            ],
1917        };
1918
1919        let selection = witnessed_quote_selection_or_error(
1920            &address,
1921            &witnessed,
1922            CLOSE_GROUP_SIZE,
1923            witnessed_close_group_quorum(),
1924        )
1925        .expect("fallback candidates should be retained for quote collection");
1926
1927        assert_eq!(
1928            selection.quote_peers.len(),
1929            CLOSE_GROUP_SIZE + EXTRA_QUORUM_CANDIDATES
1930        );
1931        assert_eq!(
1932            selection
1933                .quote_peers
1934                .iter()
1935                .map(|peer| peer.peer_id.as_bytes()[0])
1936                .collect::<Vec<_>>(),
1937            vec![1, 2, 3, 4, 5, 6, 7, 8]
1938        );
1939        assert_eq!(
1940            put_peer_seeds(&selection.initial_put_peers),
1941            vec![1, 2, 3, 4, 5, 6, 7]
1942        );
1943    }
1944
1945    #[test]
1946    fn witnessed_quote_peers_lower_quorum_for_missing_responder_views() {
1947        let address = [0u8; 32];
1948        let witnessed = WitnessedCloseGroup {
1949            target: address,
1950            k: CLOSE_GROUP_SIZE,
1951            initial_closest: witnessed_test_nodes(&[1, 2, 3, 4, 5, 6, 7]),
1952            responder_views: vec![
1953                witnessed_test_view(1, &[1, 2, 3, 4, 5, 6, 7]),
1954                witnessed_test_view(2, &[1, 2, 3, 4, 5, 6, 8]),
1955                witnessed_test_view(3, &[1, 2, 3, 4, 5, 7, 8]),
1956                witnessed_test_view(4, &[1, 2, 3, 4, 6, 7, 8]),
1957                witnessed_test_view(5, &[1, 2, 3, 5, 6, 7, 8]),
1958                witnessed_test_view(6, &[1, 2, 4, 5, 6, 7, 8]),
1959            ],
1960        };
1961        let quorum = witnessed_close_group_quorum_for_transcript(&witnessed);
1962
1963        assert_eq!(missing_witnessed_responder_views(&witnessed), 1);
1964        assert_eq!(quorum, 4);
1965
1966        let selection =
1967            witnessed_quote_selection_or_error(&address, &witnessed, CLOSE_GROUP_SIZE, quorum)
1968                .expect(
1969                    "one missing responder view should lower quorum and still select candidates",
1970                );
1971
1972        assert_eq!(
1973            selection
1974                .quote_peers
1975                .iter()
1976                .map(|peer| peer.peer_id.as_bytes()[0])
1977                .collect::<Vec<_>>(),
1978            vec![1, 2, 3, 4, 5, 6, 7, 8]
1979        );
1980        assert_eq!(selection.quorum, quorum);
1981    }
1982
1983    #[test]
1984    fn witnessed_quote_selection_keeps_closest_set_with_median_voter_quorum() {
1985        const MEDIAN_ISSUER_SEED: u8 = 7;
1986        const FAR_SUPPORTING_VOTER_SEED: u8 = 20;
1987        const UNSUCCESSFUL_SUPPORTING_VOTER_SEED: u8 = 21;
1988
1989        let address = [0u8; 32];
1990        let quotes = vec![
1991            synthetic_quote(1, 10),
1992            synthetic_quote(2, 20),
1993            synthetic_quote(3, 30),
1994            synthetic_quote(6, 50),
1995            synthetic_quote(MEDIAN_ISSUER_SEED, 40),
1996            synthetic_quote(8, 60),
1997            synthetic_quote(9, 70),
1998            synthetic_quote(FAR_SUPPORTING_VOTER_SEED, 80),
1999        ];
2000        let mut voters_by_peer = HashMap::new();
2001        voters_by_peer.insert(
2002            synthetic_peer(MEDIAN_ISSUER_SEED),
2003            synthetic_voters(&[
2004                1,
2005                2,
2006                3,
2007                MEDIAN_ISSUER_SEED,
2008                FAR_SUPPORTING_VOTER_SEED,
2009                UNSUCCESSFUL_SUPPORTING_VOTER_SEED,
2010            ]),
2011        );
2012
2013        let quorum = witnessed_close_group_quorum();
2014        let selected =
2015            select_witnessed_median_voter_quotes(quotes, &address, &voters_by_peer, quorum)
2016                .expect("a supported close-group quote set should be selected");
2017
2018        assert_eq!(quote_peer_seeds(&selected), vec![1, 2, 3, 6, 7, 8, 9]);
2019        let (median_peer_id, _) =
2020            median_paid_quote_issuer(&selected).expect("selected quotes have a median");
2021        assert_eq!(median_peer_id, synthetic_peer(MEDIAN_ISSUER_SEED));
2022        assert!(voters_by_peer[&median_peer_id].len() >= quorum);
2023    }
2024
2025    #[test]
2026    fn witnessed_quote_selection_uses_direct_median_witness_recognition() {
2027        const MEDIAN_ISSUER_SEED: u8 = 7;
2028
2029        let address = [0u8; 32];
2030        let quotes = vec![
2031            synthetic_quote(1, 10),
2032            synthetic_quote(2, 20),
2033            synthetic_quote(3, 30),
2034            synthetic_quote(4, 50),
2035            synthetic_quote(MEDIAN_ISSUER_SEED, 40),
2036            synthetic_quote(8, 60),
2037            synthetic_quote(9, 70),
2038        ];
2039        let mut voters_by_peer = HashMap::new();
2040        voters_by_peer.insert(
2041            synthetic_peer(MEDIAN_ISSUER_SEED),
2042            synthetic_voters(&[20, 21, 22, 23, 24]),
2043        );
2044
2045        let quorum = witnessed_close_group_quorum();
2046        let selected =
2047            select_witnessed_median_voter_quotes(quotes, &address, &voters_by_peer, quorum)
2048                .expect("direct witness recognition should support the paid median issuer");
2049
2050        let (median_peer_id, _) =
2051            median_paid_quote_issuer(&selected).expect("selected quotes have a median");
2052        let selected_peers = selected
2053            .iter()
2054            .map(|(peer_id, _, _, _, _)| *peer_id)
2055            .collect::<HashSet<_>>();
2056        assert_eq!(median_peer_id, synthetic_peer(MEDIAN_ISSUER_SEED));
2057        assert_eq!(
2058            voters_by_peer[&median_peer_id]
2059                .intersection(&selected_peers)
2060                .count(),
2061            0,
2062            "recognising witnesses need not also be selected quote issuers"
2063        );
2064        assert_eq!(voters_by_peer[&median_peer_id].len(), quorum);
2065    }
2066
2067    #[test]
2068    fn witnessed_quote_selection_allows_single_required_quote() {
2069        const QUOTE_ISSUER_SEED: u8 = 7;
2070
2071        let address = [0u8; 32];
2072        let quotes = vec![
2073            synthetic_quote(QUOTE_ISSUER_SEED, 10),
2074            synthetic_quote(1, 20),
2075            synthetic_quote(2, 30),
2076        ];
2077        let mut voters_by_peer = HashMap::new();
2078        voters_by_peer.insert(
2079            synthetic_peer(QUOTE_ISSUER_SEED),
2080            synthetic_voters(&[1, 2, 3, 4, 5]),
2081        );
2082
2083        let selected = select_witnessed_median_voter_quotes(
2084            quotes,
2085            &address,
2086            &voters_by_peer,
2087            witnessed_close_group_quorum(),
2088        )
2089        .expect("one quorum-supported quote is enough for SNP payment");
2090
2091        assert_eq!(quote_peer_seeds(&selected), vec![QUOTE_ISSUER_SEED]);
2092        let (median_peer_id, _) =
2093            median_paid_quote_issuer(&selected).expect("single quote is its own median");
2094        assert_eq!(median_peer_id, synthetic_peer(QUOTE_ISSUER_SEED));
2095    }
2096
2097    #[test]
2098    fn witnessed_quote_selection_rejects_median_without_witness_quorum() {
2099        const MEDIAN_ISSUER_SEED: u8 = 7;
2100
2101        let address = [0u8; 32];
2102        let quotes = vec![
2103            synthetic_quote(1, 10),
2104            synthetic_quote(2, 20),
2105            synthetic_quote(3, 30),
2106            synthetic_quote(6, 50),
2107            synthetic_quote(MEDIAN_ISSUER_SEED, 40),
2108            synthetic_quote(8, 60),
2109            synthetic_quote(9, 70),
2110            synthetic_quote(10, 80),
2111        ];
2112        let mut voters_by_peer = HashMap::new();
2113        voters_by_peer.insert(
2114            synthetic_peer(MEDIAN_ISSUER_SEED),
2115            synthetic_voters(&[1, 2, 3, 20]),
2116        );
2117
2118        let selected = select_witnessed_median_voter_quotes(
2119            quotes,
2120            &address,
2121            &voters_by_peer,
2122            witnessed_close_group_quorum(),
2123        );
2124
2125        assert!(
2126            selected.is_none(),
2127            "the selector must not return a paid quote set when fewer than the \
2128             witnessed median voter quorum recognised the paid median issuer"
2129        );
2130    }
2131
2132    #[test]
2133    fn put_peers_prioritise_median_voters_without_reordering_quotes() {
2134        const MEDIAN_ISSUER_SEED: u8 = 7;
2135
2136        let quotes = vec![
2137            synthetic_quote(1, 10),
2138            synthetic_quote(2, 20),
2139            synthetic_quote(3, 30),
2140            synthetic_quote(4, 50),
2141            synthetic_quote(5, 60),
2142            synthetic_quote(6, 70),
2143            synthetic_quote(MEDIAN_ISSUER_SEED, 40),
2144        ];
2145        let mut voters_by_peer = HashMap::new();
2146        voters_by_peer.insert(
2147            synthetic_peer(MEDIAN_ISSUER_SEED),
2148            synthetic_voters(&[3, 4, 5, 6, MEDIAN_ISSUER_SEED]),
2149        );
2150
2151        let put_candidates = put_peers_from_seeds(&[1, 2, 3, 4, 5, 6, 7]);
2152        let put_peers = put_peers_with_median_voters_first(
2153            &quotes,
2154            &put_candidates,
2155            &voters_by_peer,
2156            witnessed_close_group_quorum(),
2157        )
2158        .expect("median voters should produce an ordered PUT set");
2159
2160        assert_eq!(quote_peer_seeds(&quotes), vec![1, 2, 3, 4, 5, 6, 7]);
2161        let (median_peer_id, _) =
2162            median_paid_quote_issuer(&quotes).expect("selected quotes have a median");
2163        assert_eq!(median_peer_id, synthetic_peer(MEDIAN_ISSUER_SEED));
2164        assert_eq!(put_peer_seeds(&put_peers), vec![3, 4, 5, 6, 7, 1, 2]);
2165    }
2166
2167    #[test]
2168    fn filter_drops_only_bad_bindings_and_leaves_storer_acceptable_quotes() {
2169        let mut quotes = vec![
2170            good_quote_real(),
2171            bad_quote_real(),
2172            good_quote_real(),
2173            bad_quote_real(),
2174            good_quote_real(),
2175        ];
2176
2177        let dropped = drop_quotes_with_bad_bindings(&mut quotes);
2178
2179        assert_eq!(dropped, 2, "two crossed-key quotes must be dropped");
2180        assert_eq!(quotes.len(), 3, "three real-key quotes must remain");
2181
2182        // Cross-checked invariant: every retained quote would be accepted by
2183        // a storer running the full spec. The defensive filter only checks
2184        // the binding, so this asserts the binding-only filter is correct
2185        // for binding-only failures (other failure modes are filtered by
2186        // the per-peer classifier upstream).
2187        for (peer_id, _, quote, _, _) in &quotes {
2188            assert!(
2189                storer_binding_would_accept(peer_id, quote),
2190                "every retained quote must satisfy the full storer-side spec"
2191            );
2192        }
2193    }
2194
2195    #[test]
2196    fn filter_is_noop_when_all_quotes_are_storer_acceptable() {
2197        let mut quotes: Vec<_> = (0..5).map(|_| good_quote_real()).collect();
2198        let before = quotes.len();
2199        let dropped = drop_quotes_with_bad_bindings(&mut quotes);
2200        assert_eq!(dropped, 0);
2201        assert_eq!(quotes.len(), before);
2202        for (peer_id, _, quote, _, _) in &quotes {
2203            assert!(storer_binding_would_accept(peer_id, quote));
2204        }
2205    }
2206
2207    #[test]
2208    fn filter_drops_all_when_every_responder_is_bad() {
2209        // The "all hostile" case: every peer returned a bad binding. The
2210        // patch should leave us with zero quotes (not panic, not skip the
2211        // filter, not return malformed quotes). The caller then surfaces
2212        // InsufficientPeers.
2213        let mut quotes: Vec<_> = (0..fault_tolerant_quote_query_count())
2214            .map(|_| bad_quote_real())
2215            .collect();
2216        let dropped = drop_quotes_with_bad_bindings(&mut quotes);
2217        assert_eq!(dropped, fault_tolerant_quote_query_count());
2218        assert!(quotes.is_empty());
2219    }
2220
2221    #[test]
2222    fn filter_preserves_quote_payload_byte_for_byte() {
2223        // After filtering, the retained quotes must be untouched — pub_key,
2224        // signature, content, timestamp, price, rewards_address. The patch
2225        // is a filter, not a transformation; this test catches any future
2226        // regression that mutates a retained quote.
2227        let (peer_id, addrs, original_quote, amount, commitment) = good_quote_real();
2228        let mut quotes = vec![(
2229            peer_id,
2230            addrs.clone(),
2231            original_quote.clone(),
2232            amount,
2233            commitment,
2234        )];
2235        let _ = drop_quotes_with_bad_bindings(&mut quotes);
2236
2237        let (kept_peer, kept_addrs, kept_quote, kept_amount, _kept_commitment) =
2238            quotes.pop().expect("the good quote must survive filtering");
2239        assert_eq!(kept_peer.as_bytes(), peer_id.as_bytes());
2240        assert_eq!(kept_addrs.len(), addrs.len());
2241        assert_eq!(kept_amount, amount);
2242        assert_eq!(kept_quote.pub_key, original_quote.pub_key);
2243        assert_eq!(kept_quote.signature, original_quote.signature);
2244        assert_eq!(kept_quote.content.0, original_quote.content.0);
2245        assert_eq!(kept_quote.timestamp, original_quote.timestamp);
2246        assert_eq!(kept_quote.price, original_quote.price);
2247        assert_eq!(kept_quote.rewards_address, original_quote.rewards_address);
2248    }
2249
2250    // ============================================================
2251    // The Apr 30 production-failure repro
2252    // ============================================================
2253
2254    /// Repro of the production failure from 2026-04-30 testnet runs.
2255    ///
2256    /// An external operator on `75.48.86.24` ran two co-located ant-node
2257    /// identities (peer `0755ecb55b…` and peer `073db92f…`) that crossed
2258    /// their quote-signing keys. Every chunk whose XOR-closest set happened
2259    /// to include peer `0755ecb5` got a payment proof with one malformed
2260    /// quote, and the storer's `validate_peer_bindings` rejected the
2261    /// entire close-group proof — burning the chunk's payment.
2262    ///
2263    /// This test proves the fault-tolerant quote path still fixes that failure
2264    /// shape:
2265    ///
2266    /// 1. We assemble `2x CLOSE_GROUP_SIZE` real ML-DSA-65 quotes — the same
2267    ///    buffer merkle preflight and merkle-mode estimates retain for probes.
2268    /// 2. One of them is a *crossed-key* quote — the production failure shape.
2269    /// 3. We run an independent `storer_would_accept` check (re-derived from
2270    ///    the storer spec, not from `quote_binding_is_valid`) over the
2271    ///    pre-filter set; we confirm the bad peer is rejected, proving the
2272    ///    storer **would** burn the chunk's payment if we proceeded unfiltered.
2273    /// 4. We run `drop_quotes_with_bad_bindings`.
2274    /// 5. We re-run `storer_would_accept` over the post-filter set; we confirm
2275    ///    EVERY remaining quote would be accepted, proving the filtered set
2276    ///    will not trigger the `validate_peer_bindings` rejection that caused
2277    ///    the Apr 30 outage.
2278    /// 6. We confirm the post-filter set has at least `CLOSE_GROUP_SIZE`
2279    ///    quotes — the over-query buffer (2x) is sufficient.
2280    #[test]
2281    fn repro_apr_30_storer_would_have_rejected_pre_filter_and_accepts_post_filter() {
2282        let over_query_count = fault_tolerant_quote_query_count();
2283        let mut quotes: Vec<_> = (0..over_query_count - 1)
2284            .map(|_| good_quote_real())
2285            .collect();
2286        // Splice the crossed-key quote in the middle (mirrors the random
2287        // position the bad peer takes in the DHT-returned closest set).
2288        quotes.insert(over_query_count / 2, bad_quote_real());
2289        assert_eq!(quotes.len(), over_query_count);
2290
2291        // Step 1: prove the storer would reject the pre-filter set.
2292        let storer_would_reject_count = quotes
2293            .iter()
2294            .filter(|(p, _, q, _, _)| !storer_binding_would_accept(p, q))
2295            .count();
2296        assert_eq!(
2297            storer_would_reject_count, 1,
2298            "exactly one quote (the crossed-key one) must be rejected by the storer spec"
2299        );
2300
2301        // Step 2: run the patched filter.
2302        let dropped = drop_quotes_with_bad_bindings(&mut quotes);
2303        assert_eq!(dropped, 1, "exactly the crossed-key quote must be filtered");
2304
2305        // Step 3: prove the storer would accept every survivor under the FULL spec.
2306        for (peer_id, _, quote, _, _) in &quotes {
2307            assert!(
2308                storer_binding_would_accept(peer_id, quote),
2309                "every post-filter quote must be accepted by the storer spec — \
2310                 this is what the filter guarantees before any quote set is used"
2311            );
2312        }
2313
2314        // Step 4: prove the over-query buffer is sufficient to refill.
2315        assert!(
2316            quotes.len() >= CLOSE_GROUP_SIZE,
2317            "after filtering, at least CLOSE_GROUP_SIZE good quotes must remain \
2318             so a fault-tolerant probe can still return a full close group"
2319        );
2320    }
2321
2322    /// When more than the over-query buffer of peers misbehave, the filter
2323    /// must NOT silently produce a short proof. The downstream caller in
2324    /// `get_store_quotes` must see fewer than `CLOSE_GROUP_SIZE` survivors
2325    /// and return `InsufficientPeers`.
2326    #[test]
2327    fn filter_leaves_short_set_when_too_many_bad_peers() {
2328        let good_count = CLOSE_GROUP_SIZE - 1;
2329        let bad_count = fault_tolerant_quote_query_count() - good_count;
2330        let mut quotes: Vec<_> = std::iter::repeat_with(bad_quote_real)
2331            .take(bad_count)
2332            .chain(std::iter::repeat_with(good_quote_real).take(good_count))
2333            .collect();
2334
2335        let dropped = drop_quotes_with_bad_bindings(&mut quotes);
2336        assert_eq!(dropped, bad_count);
2337        assert!(
2338            quotes.len() < CLOSE_GROUP_SIZE,
2339            "this is the precondition for InsufficientPeers downstream"
2340        );
2341        // Sanity: every survivor is storer-acceptable under the full spec.
2342        for (peer_id, _, quote, _, _) in &quotes {
2343            assert!(storer_binding_would_accept(peer_id, quote));
2344        }
2345    }
2346
2347    // ============================================================
2348    // Tests for the per-peer response classifier (the PRIMARY defense).
2349    //
2350    // These tests exercise the production code path that runs inside
2351    // get_store_quotes' per-peer async closure. The defensive
2352    // `drop_quotes_with_bad_bindings` is a second line of defence —
2353    // these tests make sure the FIRST line is what actually catches
2354    // misbehaving peers in production. Without these, a regression
2355    // that removes the per-peer check could be masked by the post-
2356    // collect filter and pass the rest of the suite.
2357    // ============================================================
2358
2359    /// Helper: serialize a `PaymentQuote` to bytes the way the wire layer
2360    /// does (rmp_serde / msgpack), to feed into `classify_quote_response`.
2361    fn serialize_quote(quote: &PaymentQuote) -> Vec<u8> {
2362        rmp_serde::to_vec(quote).expect("serialize quote")
2363    }
2364
2365    #[test]
2366    fn classifier_accepts_real_self_consistent_quote() {
2367        // A properly-signed baseline quote for the requested content passes the
2368        // full client gate (binding + content + signature + price).
2369        let content = [7u8; 32];
2370        let (peer_id, quote) = signed_baseline_quote(content);
2371        let bytes = serialize_quote(&quote);
2372        let result = classify_quote_response(&peer_id, &content, &bytes, false, None);
2373        match result {
2374            Ok((q, price, commitment)) => {
2375                assert_eq!(q.pub_key, quote.pub_key);
2376                assert_eq!(price, quote.price);
2377                assert!(commitment.is_none(), "baseline quote ships no commitment");
2378            }
2379            Err(e) => panic!("expected Ok, got {e}"),
2380        }
2381    }
2382
2383    #[test]
2384    fn classifier_rejects_quote_with_invalid_signature() {
2385        // A quote whose pub_key binds correctly but whose signature is bogus is
2386        // dropped BEFORE payment (the storer would reject it and burn the pay).
2387        let content = [7u8; 32];
2388        let (peer_id, mut quote) = signed_baseline_quote(content);
2389        quote.signature = vec![0u8; quote.signature.len()]; // corrupt the signature
2390        let bytes = serialize_quote(&quote);
2391        let result = classify_quote_response(&peer_id, &content, &bytes, false, None);
2392        assert!(
2393            matches!(result, Err(Error::BadQuoteBinding { .. })),
2394            "a quote with an invalid signature must be rejected; got {result:?}"
2395        );
2396    }
2397
2398    #[test]
2399    fn classifier_rejects_quote_for_wrong_content() {
2400        // A validly-signed quote for a DIFFERENT address is dropped before pay.
2401        let (peer_id, quote) = signed_baseline_quote([7u8; 32]);
2402        let bytes = serialize_quote(&quote);
2403        let result = classify_quote_response(&peer_id, &[9u8; 32], &bytes, false, None);
2404        assert!(
2405            matches!(result, Err(Error::BadQuoteBinding { .. })),
2406            "a quote for the wrong content must be rejected; got {result:?}"
2407        );
2408    }
2409
2410    #[test]
2411    fn classifier_rejects_crossed_keypair_with_typed_error() {
2412        let (peer_id, _, quote, _, _) = bad_quote_real();
2413        let bytes = serialize_quote(&quote);
2414        let result = classify_quote_response(&peer_id, &[0u8; 32], &bytes, false, None);
2415        match result {
2416            Err(Error::BadQuoteBinding {
2417                peer_id: pid,
2418                detail,
2419            }) => {
2420                assert_eq!(pid, peer_id.to_string());
2421                assert!(
2422                    detail.contains("BLAKE3(pub_key)="),
2423                    "diagnostic detail must include the derived peer id: {detail}"
2424                );
2425            }
2426            other => panic!("expected BadQuoteBinding for crossed-key quote, got {other:?}"),
2427        }
2428    }
2429
2430    /// CRITICAL: a misbehaving peer that votes `already_stored=true` must
2431    /// NOT be allowed to influence the close-group "already stored"
2432    /// majority decision. The bind-check runs before the AlreadyStored
2433    /// short-circuit, so a crossed-key peer voting "already stored" is
2434    /// classified as `BadQuoteBinding`, not `AlreadyStored`.
2435    ///
2436    /// This locks in a specific reviewer concern from round 1:
2437    ///   "A peer with a crossed/garbage signing key could simply respond
2438    ///   already_stored=true and its vote enters already_stored_peers
2439    ///   unfiltered."
2440    #[test]
2441    fn classifier_rejects_already_stored_vote_from_bad_binding_peer() {
2442        let (peer_id, _, quote, _, _) = bad_quote_real();
2443        let bytes = serialize_quote(&quote);
2444        // The peer claims already_stored=true, but its quote has a crossed key.
2445        let result = classify_quote_response(&peer_id, &[0u8; 32], &bytes, true, None);
2446        assert!(
2447            matches!(result, Err(Error::BadQuoteBinding { .. })),
2448            "crossed-key peer must be classified BadQuoteBinding even when \
2449             voting already_stored=true; got {result:?}"
2450        );
2451    }
2452
2453    /// An honest peer's `already_stored=true` vote IS honoured (after
2454    /// passing the bind-check). This is the contrast to the test above.
2455    #[test]
2456    fn classifier_honours_already_stored_vote_from_good_binding_peer() {
2457        let content = [7u8; 32];
2458        let (peer_id, quote) = signed_baseline_quote(content);
2459        let bytes = serialize_quote(&quote);
2460        let result = classify_quote_response(&peer_id, &content, &bytes, true, None);
2461        assert!(
2462            matches!(result, Err(Error::AlreadyStored)),
2463            "honest peer's already_stored vote must be honoured; got {result:?}"
2464        );
2465    }
2466
2467    #[test]
2468    fn classifier_returns_serialization_error_on_bad_bytes() {
2469        let (peer_id, _, _, _, _) = good_quote_real();
2470        let garbage = b"this is not a valid msgpack PaymentQuote".to_vec();
2471        let result = classify_quote_response(&peer_id, &[0u8; 32], &garbage, false, None);
2472        assert!(
2473            matches!(result, Err(Error::Serialization(_))),
2474            "garbage bytes must produce a Serialization error; got {result:?}"
2475        );
2476    }
2477
2478    /// Cross-validate the classifier's binding verdict against the
2479    /// independent storer-spec re-derivation across mixed responders.
2480    #[test]
2481    fn classifier_verdict_matches_storer_binding_spec_for_mixed_responders() {
2482        let content = [7u8; 32];
2483        let mut responders: Vec<(PeerId, PaymentQuote)> =
2484            (0..12).map(|_| signed_baseline_quote(content)).collect();
2485        for _ in 0..4 {
2486            let (p, _, q, _, _) = bad_quote_real();
2487            responders.push((p, q));
2488        }
2489
2490        for (peer_id, quote) in &responders {
2491            let bytes = serialize_quote(quote);
2492            let storer_verdict = storer_binding_would_accept(peer_id, quote);
2493            let classifier_verdict =
2494                classify_quote_response(peer_id, &content, &bytes, false, None).is_ok();
2495            assert_eq!(
2496                classifier_verdict, storer_verdict,
2497                "classifier and storer-binding-spec must agree on every responder \
2498                 (peer_id={}, storer={storer_verdict}, classifier={classifier_verdict})",
2499                peer_id
2500            );
2501        }
2502    }
2503
2504    // ============================================================
2505    // ADR-0004: quote_commitment_binding_is_valid (forced-price gate)
2506    //
2507    // Mirrors the storer-side `binding_violation` in
2508    // `ant-node/src/payment/verifier.rs`. The client runs this before
2509    // paying so it never pays a quote the storer's arithmetic gate would
2510    // reject. The client now runs the FULL check (shape, cap, exact price,
2511    // and for bound quotes: parse + peer-binding + signature + hash==pin +
2512    // count==key_count) using the shared ant-protocol commitment type, so an
2513    // unresolvable/forged commitment is never paid. A live resolve against a
2514    // REAL signed commitment is proven in the e2e suite (e2e_adr0004.rs).
2515    // ============================================================
2516
2517    /// A throwaway peer id for tests that fail BEFORE commitment resolution
2518    /// (shape/cap/price checks don't depend on the peer).
2519    fn any_peer() -> PeerId {
2520        PeerId::from_bytes([0u8; 32])
2521    }
2522
2523    /// Build a quote carrying a specific `(count, pin, price)` binding.
2524    fn quote_with_binding(
2525        committed_key_count: u32,
2526        commitment_pin: Option<[u8; 32]>,
2527        price: Amount,
2528    ) -> PaymentQuote {
2529        PaymentQuote {
2530            content: XorName([0u8; 32]),
2531            timestamp: SystemTime::UNIX_EPOCH,
2532            price,
2533            rewards_address: RewardsAddress::new([0u8; 20]),
2534            pub_key: Vec::new(),
2535            signature: Vec::new(),
2536            committed_key_count,
2537            commitment_pin,
2538        }
2539    }
2540
2541    /// Build a VALIDLY-SIGNED `StorageCommitment` bound to `kp`'s peer id, so a
2542    /// test can pass peer-binding + signature and isolate the `hash == pin` and
2543    /// `count == key_count` sub-checks. Mirrors ant-node's commitment signing:
2544    /// the canonical payload (`root || key_count(LE) || peer_id || pk_len(LE) ||
2545    /// pub_key`) signed under `DOMAIN_COMMITMENT`.
2546    fn signed_commitment(kp: &Keypair, root: [u8; 32], key_count: u32) -> StorageCommitment {
2547        use ant_protocol::payment::commitment::DOMAIN_COMMITMENT;
2548        use ant_protocol::pqc::api::{ml_dsa_65, MlDsaSecretKey as ApiSecretKey, MlDsaVariant};
2549        let peer = compute_address(&kp.pub_key_bytes);
2550        let mut payload = Vec::with_capacity(32 + 4 + 32 + 4 + kp.pub_key_bytes.len());
2551        payload.extend_from_slice(&root);
2552        payload.extend_from_slice(&key_count.to_le_bytes());
2553        payload.extend_from_slice(&peer);
2554        payload.extend_from_slice(&(kp.pub_key_bytes.len() as u32).to_le_bytes());
2555        payload.extend_from_slice(&kp.pub_key_bytes);
2556        let sk = ApiSecretKey::from_bytes(MlDsaVariant::MlDsa65, &kp.secret_key_bytes)
2557            .expect("api secret key");
2558        let signature = ml_dsa_65()
2559            .sign_with_context(&sk, &payload, DOMAIN_COMMITMENT)
2560            .expect("sign commitment")
2561            .to_bytes();
2562        StorageCommitment {
2563            root,
2564            key_count,
2565            sender_peer_id: peer,
2566            sender_public_key: kp.pub_key_bytes.clone(),
2567            signature,
2568        }
2569    }
2570
2571    #[test]
2572    fn binding_baseline_ok_only_at_baseline_price() {
2573        // (0, None) priced at calculate_price(0) is the valid baseline.
2574        let q = quote_with_binding(0, None, calculate_price(0));
2575        assert!(quote_commitment_binding_is_valid(&any_peer(), &q, &None).is_ok());
2576
2577        // (0, None) priced above baseline is rejected — the forged-shape
2578        // bypass (strip the pin, charge more than the empty-node price).
2579        let q = quote_with_binding(0, None, calculate_price(500));
2580        assert!(quote_commitment_binding_is_valid(&any_peer(), &q, &None).is_err());
2581    }
2582
2583    #[test]
2584    fn binding_rejects_incoherent_shapes() {
2585        // count > 0 but no pin: unauditable.
2586        let q = quote_with_binding(500, None, calculate_price(500));
2587        assert!(quote_commitment_binding_is_valid(&any_peer(), &q, &None).is_err());
2588        // count 0 but a pin: incoherent baseline.
2589        let q = quote_with_binding(0, Some([9u8; 32]), calculate_price(0));
2590        assert!(quote_commitment_binding_is_valid(&any_peer(), &q, &None).is_err());
2591    }
2592
2593    #[test]
2594    fn binding_rejects_count_above_cap() {
2595        let over = MAX_COMMITMENT_KEY_COUNT + 1;
2596        let q = quote_with_binding(over, Some([9u8; 32]), calculate_price(over as usize));
2597        assert!(
2598            quote_commitment_binding_is_valid(&any_peer(), &q, &Some(vec![1u8; 16])).is_err(),
2599            "a count above MAX_COMMITMENT_KEY_COUNT must be rejected before payment"
2600        );
2601    }
2602
2603    #[test]
2604    fn binding_rejects_on_curve_wrong_count() {
2605        // Priced for 499 but claims count 500 — on a real price curve but the
2606        // wrong count. Rejected at the exact-price check, before resolution.
2607        let q = quote_with_binding(500, Some([9u8; 32]), calculate_price(499));
2608        assert!(quote_commitment_binding_is_valid(&any_peer(), &q, &Some(vec![1u8; 16])).is_err());
2609    }
2610
2611    #[test]
2612    fn binding_rejects_bound_quote_without_shipped_commitment() {
2613        // A bound quote whose commitment did not arrive is unresolvable, so it
2614        // is dropped before payment even though its price is on the curve.
2615        let q = quote_with_binding(500, Some([9u8; 32]), calculate_price(500));
2616        assert!(
2617            quote_commitment_binding_is_valid(&any_peer(), &q, &None).is_err(),
2618            "a bound quote missing its commitment must be rejected"
2619        );
2620    }
2621
2622    #[test]
2623    fn binding_rejects_unparseable_and_peer_unbound_commitment() {
2624        // A bound quote whose shipped commitment is garbage (doesn't even
2625        // deserialize) is rejected — the client never pays an unresolvable pin.
2626        let q = quote_with_binding(500, Some([9u8; 32]), calculate_price(500));
2627        assert!(
2628            quote_commitment_binding_is_valid(&any_peer(), &q, &Some(vec![0xFF; 8])).is_err(),
2629            "an unparseable commitment must be rejected before payment"
2630        );
2631
2632        // A well-formed StorageCommitment that is NOT bound to the quoting peer
2633        // (its sender_peer_id / pubkey don't derive the peer id) is rejected at
2634        // the peer-binding check. The signature / hash==pin / count==key_count
2635        // sub-checks are covered by the dedicated tests below, which pass
2636        // peer-binding first so each isolates exactly one sub-check.
2637        let bogus = StorageCommitment {
2638            root: [1u8; 32],
2639            key_count: 500,
2640            sender_peer_id: [2u8; 32], // not the quoting peer
2641            sender_public_key: vec![3u8; 1952],
2642            signature: vec![4u8; 3293],
2643        };
2644        let blob = rmp_serde::to_vec(&bogus).expect("serialize bogus commitment");
2645        assert!(
2646            quote_commitment_binding_is_valid(&any_peer(), &q, &Some(blob)).is_err(),
2647            "a commitment not bound to the quoting peer must be rejected before payment"
2648        );
2649    }
2650
2651    #[test]
2652    fn binding_rejects_commitment_with_invalid_signature() {
2653        // Correctly-bound commitment (passes peer-binding) but with a corrupted
2654        // signature: must be rejected at the signature check. Deleting that check
2655        // would let a peer attest any (root, key_count) without holding the key.
2656        let kp = gen_keypair();
2657        let mut commitment = signed_commitment(&kp, [6u8; 32], 500);
2658        commitment.signature[0] ^= 0xFF; // still 3293 bytes, no longer valid
2659                                         // Pin the (corrupted) commitment so the hash==pin check would pass; the
2660                                         // only thing wrong is the signature, isolating that sub-check.
2661        let pin = commitment_hash(&commitment).expect("hash");
2662        let blob = rmp_serde::to_vec(&commitment).expect("serialize commitment");
2663        let q = quote_with_binding(500, Some(pin), calculate_price(500));
2664        let res = quote_commitment_binding_is_valid(&kp.peer_id, &q, &Some(blob));
2665        let err = res.expect_err("commitment with an invalid signature must be rejected");
2666        assert!(
2667            err.contains("signature"),
2668            "should fail at the signature check: {err}"
2669        );
2670    }
2671
2672    #[test]
2673    fn binding_rejects_commitment_that_does_not_hash_to_pin() {
2674        // Validly-signed, correctly-bound commitment, but the quote pins a
2675        // DIFFERENT hash: must be rejected. Deleting the hash==pin check would
2676        // let a peer ship any commitment it holds for a pin it doesn't back.
2677        let kp = gen_keypair();
2678        let commitment = signed_commitment(&kp, [5u8; 32], 500);
2679        let wrong_pin = [0xAB; 32];
2680        assert_ne!(commitment_hash(&commitment), Some(wrong_pin));
2681        let blob = rmp_serde::to_vec(&commitment).expect("serialize commitment");
2682        let q = quote_with_binding(500, Some(wrong_pin), calculate_price(500));
2683        let res = quote_commitment_binding_is_valid(&kp.peer_id, &q, &Some(blob));
2684        let err = res.expect_err("commitment that does not hash to the pin must be rejected");
2685        assert!(
2686            err.contains("hash"),
2687            "should fail at the hash==pin check: {err}"
2688        );
2689    }
2690
2691    #[test]
2692    fn binding_rejects_count_disagreeing_with_commitment() {
2693        // Validly-signed, correctly-bound, correctly-pinned commitment attesting
2694        // key_count=400, but the quote claims 500 (priced on-curve for 500):
2695        // must be rejected. Deleting the count==key_count check would let a peer
2696        // price against an inflated count while committing to fewer keys.
2697        let kp = gen_keypair();
2698        let commitment = signed_commitment(&kp, [7u8; 32], 400);
2699        let pin = commitment_hash(&commitment).expect("hash");
2700        let blob = rmp_serde::to_vec(&commitment).expect("serialize commitment");
2701        let q = quote_with_binding(500, Some(pin), calculate_price(500));
2702        let res = quote_commitment_binding_is_valid(&kp.peer_id, &q, &Some(blob));
2703        let err = res.expect_err("a quote count disagreeing with the commitment must be rejected");
2704        assert!(
2705            err.contains("key_count") || err.contains("attests"),
2706            "should fail at the count==key_count check: {err}"
2707        );
2708    }
2709
2710    #[test]
2711    fn binding_rejects_oversized_commitment_before_parsing() {
2712        // A bound quote shipping a blob larger than the sidecar cap is rejected
2713        // before any deserialize attempt (DoS guard on the hot path).
2714        let q = quote_with_binding(500, Some([9u8; 32]), calculate_price(500));
2715        let huge = Some(vec![0u8; MAX_COMMITMENT_SIDECAR_BYTES + 1]);
2716        assert!(
2717            quote_commitment_binding_is_valid(&any_peer(), &q, &huge).is_err(),
2718            "an oversized commitment blob must be rejected before payment"
2719        );
2720    }
2721
2722    #[test]
2723    fn classifier_drops_off_curve_quote_with_typed_error() {
2724        // End-to-end through the classifier: a VALIDLY-SIGNED, correctly-bound
2725        // quote for the right content, but with an off-curve price, is dropped
2726        // as BadQuoteCommitment (the forced-price extraction guard fires after
2727        // the quote's own signature/content checks pass).
2728        use ant_protocol::pqc::ops::MlDsaSecretKey;
2729        let content = [7u8; 32];
2730        let kp = gen_keypair();
2731        let mut quote = PaymentQuote {
2732            content: XorName(content),
2733            timestamp: SystemTime::UNIX_EPOCH,
2734            // claims baseline shape but charges a non-baseline price
2735            price: calculate_price(500),
2736            rewards_address: RewardsAddress::new([0u8; 20]),
2737            pub_key: kp.pub_key_bytes.clone(),
2738            signature: Vec::new(),
2739            committed_key_count: 0,
2740            commitment_pin: None,
2741        };
2742        let ml_dsa = MlDsa65::new();
2743        let sk = MlDsaSecretKey::from_bytes(&kp.secret_key_bytes).expect("sk");
2744        quote.signature = ml_dsa
2745            .sign(&sk, &quote.bytes_for_sig())
2746            .expect("sign")
2747            .as_bytes()
2748            .to_vec();
2749        let bytes = serialize_quote(&quote);
2750        let result = classify_quote_response(&kp.peer_id, &content, &bytes, false, None);
2751        assert!(
2752            matches!(result, Err(Error::BadQuoteCommitment { .. })),
2753            "off-curve quote must be dropped as BadQuoteCommitment; got {result:?}"
2754        );
2755    }
2756
2757    /// A storer's refusal must arrive as its own terminal error, carrying the
2758    /// storer's wording. Folding it into the generic protocol error would bury
2759    /// the upgrade instruction among ordinary per-peer quote failures, which
2760    /// is the outcome this whole change exists to avoid.
2761    #[test]
2762    fn an_update_refusal_is_surfaced_with_its_upgrade_instruction() {
2763        let peer_id = PeerId::from_bytes([0x42; 32]);
2764        let refusal = ProtocolError::ClientUpdateRequired {
2765            client_settlement_version: CURRENT_SETTLEMENT_VERSION,
2766            min_settlement_version: CURRENT_SETTLEMENT_VERSION.saturating_add(1),
2767        };
2768
2769        let mapped = map_quote_response(
2770            &peer_id,
2771            &[0x11; 32],
2772            ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error(refusal)),
2773        );
2774
2775        match mapped {
2776            Some(Err(Error::ClientUpdateRequired(msg))) => {
2777                assert!(msg.contains("ant update"), "{msg}");
2778                assert!(msg.contains("nothing was charged"), "{msg}");
2779            }
2780            other => panic!("expected ClientUpdateRequired, got: {other:?}"),
2781        }
2782    }
2783
2784    /// The legacy retry exists for storers that cannot parse a versioned
2785    /// request, which are silent. A storer that answered has understood us, so
2786    /// retrying its refusal without the version would talk it into quoting a
2787    /// client that cannot pay. That is the exact failure this change removes,
2788    /// so the predicate deciding it is pinned.
2789    #[test]
2790    fn only_silence_triggers_the_legacy_retry() {
2791        assert!(is_version_unaware(&Error::Timeout("no answer".into())));
2792        assert!(is_version_unaware(&Error::Network("send failed".into())));
2793
2794        assert!(!is_version_unaware(&Error::ClientUpdateRequired(
2795            "too old".into()
2796        )));
2797        // A storer that says it is the old side has understood the request.
2798        // Retrying it unversioned would obtain a quote from a peer that cannot
2799        // verify the resulting payment, which is a burn.
2800        assert!(!is_version_unaware(&Error::StorerUpdateRequired(
2801            "node behind".into()
2802        )));
2803        assert!(!is_version_unaware(&Error::Protocol(
2804            "quote error from peer".into()
2805        )));
2806    }
2807
2808    /// A storer declaring itself the old side is an ordinary skippable peer,
2809    /// not a client fault. Surfacing it as `ClientUpdateRequired` would tell an
2810    /// up-to-date user to upgrade, and during a client-first rollout it would
2811    /// tell that to nearly everyone.
2812    #[test]
2813    fn a_storer_that_is_behind_is_not_reported_as_the_clients_fault() {
2814        let peer_id = PeerId::from_bytes([0x43; 32]);
2815        let mapped = map_quote_response(
2816            &peer_id,
2817            &[0x11; 32],
2818            ChunkMessageBody::QuoteResponse(ChunkQuoteResponse::Error(
2819                ProtocolError::StorerUpdateRequired {
2820                    client_settlement_version: 2,
2821                    node_settlement_version: 1,
2822                },
2823            )),
2824        );
2825
2826        match mapped {
2827            Some(Err(Error::StorerUpdateRequired(msg))) => {
2828                assert!(msg.contains("use a different storer"), "{msg}");
2829                assert!(!msg.contains("ant update"), "{msg}");
2830            }
2831            other => panic!("expected StorerUpdateRequired, got: {other:?}"),
2832        }
2833    }
2834
2835    /// The refusal has to stop quote collection, not join the failure list.
2836    /// If it is merely recorded, the remaining peers can still form a quorum
2837    /// and the upload proceeds to pay, which is exactly the burn the gate is
2838    /// meant to prevent.
2839    #[test]
2840    fn a_refusal_aborts_quote_collection_instead_of_counting_as_one_bad_peer() {
2841        let mut quotes = Vec::new();
2842        let mut already_stored = Vec::new();
2843        let mut failures = Vec::new();
2844        let mut bad_quotes = 0usize;
2845        let mut refusal_slot: Option<Error> = None;
2846
2847        let refusals = SettlementRefusals::default();
2848        let mut refuse =
2849            |peer: u8, failures: &mut Vec<String>, slot: &mut Option<Error>| -> Result<()> {
2850                record_store_quote_result(
2851                    PeerId::from_bytes([peer; 32]),
2852                    Vec::new(),
2853                    Err(Error::ClientUpdateRequired(
2854                        "too old, run ant update".into(),
2855                    )),
2856                    &[0x11; 32],
2857                    &mut quotes,
2858                    &mut already_stored,
2859                    failures,
2860                    &mut bad_quotes,
2861                    slot,
2862                    &refusals,
2863                )
2864            };
2865
2866        // One peer is not corroboration: recorded as an ordinary bad peer, so a
2867        // single hostile responder cannot deny every upload.
2868        let first = refuse(0x44, &mut failures, &mut refusal_slot);
2869        assert!(first.is_ok(), "one peer must not abort, got {first:?}");
2870        assert_eq!(failures.len(), 1);
2871        assert!(refusal_slot.is_none());
2872
2873        // A second, distinct peer makes it a verdict about this build.
2874        let second = refuse(0x45, &mut failures, &mut refusal_slot);
2875        assert!(
2876            matches!(second, Err(Error::ClientUpdateRequired(_))),
2877            "a corroborated refusal must propagate, got {second:?}"
2878        );
2879    }
2880
2881    /// A storer being behind must NOT abort. Otherwise one lagging peer in the
2882    /// close group fails an upload that the rest of the group could serve.
2883    #[test]
2884    fn a_storer_that_is_behind_does_not_abort_collection() {
2885        let mut quotes = Vec::new();
2886        let mut already_stored = Vec::new();
2887        let mut failures = Vec::new();
2888        let mut bad_quotes = 0usize;
2889        let mut refusal_slot: Option<Error> = None;
2890
2891        let outcome = record_store_quote_result(
2892            PeerId::from_bytes([0x45; 32]),
2893            Vec::new(),
2894            Err(Error::StorerUpdateRequired("node behind".into())),
2895            &[0x11; 32],
2896            &mut quotes,
2897            &mut already_stored,
2898            &mut failures,
2899            &mut bad_quotes,
2900            &mut refusal_slot,
2901            &SettlementRefusals::default(),
2902        );
2903
2904        assert!(
2905            outcome.is_ok(),
2906            "a lagging storer must be skipped, not fatal"
2907        );
2908        assert_eq!(failures.len(), 1, "and it should be recorded as a skip");
2909        assert!(
2910            refusal_slot.is_none(),
2911            "a node being behind is not a verdict about this client"
2912        );
2913    }
2914
2915    /// The refusal must survive the overall collection timeout.
2916    ///
2917    /// The collector runs inside `tokio::time::timeout`, and its elapsed arm
2918    /// deliberately falls through so quotes gathered from fast peers stay
2919    /// usable. That arm would otherwise discard a refusal observed just before
2920    /// the clock ran out, and the upload would pay anyway. Recording the
2921    /// verdict in a slot that outlives the timeout is what prevents it, so the
2922    /// slot is what gets tested.
2923    #[test]
2924    fn a_refusal_is_recorded_where_the_collection_timeout_cannot_discard_it() {
2925        let mut quotes = Vec::new();
2926        let mut already_stored = Vec::new();
2927        let mut failures = Vec::new();
2928        let mut bad_quotes = 0usize;
2929        let mut refusal_slot: Option<Error> = None;
2930
2931        let refusals = SettlementRefusals::default();
2932        for peer in [0x46u8, 0x47u8] {
2933            let _ = record_store_quote_result(
2934                PeerId::from_bytes([peer; 32]),
2935                Vec::new(),
2936                Err(Error::ClientUpdateRequired(
2937                    "too old, run ant update".into(),
2938                )),
2939                &[0x11; 32],
2940                &mut quotes,
2941                &mut already_stored,
2942                &mut failures,
2943                &mut bad_quotes,
2944                &mut refusal_slot,
2945                &refusals,
2946            );
2947        }
2948
2949        match refusal_slot {
2950            Some(Error::ClientUpdateRequired(msg)) => {
2951                assert!(msg.contains("ant update"), "{msg}");
2952            }
2953            other => panic!("refusal must outlive the collection state, got {other:?}"),
2954        }
2955    }
2956
2957    /// Once the quote target is met the collector stops launching new peers
2958    /// but keeps draining those already in flight, so a refusal cannot be
2959    /// missed merely because faster peers filled the quota first.
2960    ///
2961    /// The launch budget enforces the first half, and the drain relies on it
2962    /// reaching zero to terminate rather than recruiting forever.
2963    #[test]
2964    fn meeting_the_target_stops_launching_without_stopping_collection() {
2965        assert!(
2966            witnessed_quote_launch_budget(0, 0, 32) > 0,
2967            "collection must start"
2968        );
2969        assert_eq!(witnessed_quote_launch_budget(CLOSE_GROUP_SIZE, 0, 32), 0);
2970        assert_eq!(
2971            witnessed_quote_launch_budget(CLOSE_GROUP_SIZE.saturating_add(1), 0, 32),
2972            0
2973        );
2974        // In-flight peers count against the budget, so draining them does not
2975        // pull in replacements.
2976        assert_eq!(witnessed_quote_launch_budget(0, CLOSE_GROUP_SIZE, 32), 0);
2977    }
2978
2979    /// The probe must be paid per peer, not per request.
2980    ///
2981    /// A storer that predates the versioned request never answers it, so the
2982    /// client eats the probe wait before falling back. Without remembering the
2983    /// answer that cost lands on every quote, and a merkle pool asks sixteen
2984    /// candidates. Measured on the merkle E2E suite against a fleet on
2985    /// published ant-node, re-probing took the run from ~24 minutes to past
2986    /// the 60-minute CI cap.
2987    ///
2988    /// The cache is consulted, not enforced: concurrent first contacts are not
2989    /// single-flighted, so the same peer can be probed by a few in-flight
2990    /// requests before any of them records the answer. What the cache
2991    /// guarantees is that later rounds do not re-probe.
2992    #[test]
2993    fn a_peer_that_cannot_answer_a_versioned_quote_is_only_probed_once() {
2994        let peers: Arc<Mutex<HashSet<PeerId>>> = Arc::new(Mutex::new(HashSet::new()));
2995        let legacy_peer = PeerId::from_bytes([0x51; 32]);
2996        let fresh_peer = PeerId::from_bytes([0x52; 32]);
2997
2998        let known = |p: &PeerId| peers.lock().expect("cache lock").contains(p);
2999
3000        // First contact: nothing known, so the versioned request is sent.
3001        assert!(!known(&legacy_peer));
3002
3003        // Silence records the peer.
3004        peers.lock().expect("cache lock").insert(legacy_peer);
3005
3006        // Second contact skips the probe entirely.
3007        assert!(known(&legacy_peer));
3008        // and does not tar every other peer with the same brush.
3009        assert!(!known(&fresh_peer));
3010    }
3011
3012    /// Only silence is evidence that a peer cannot parse the versioned shape.
3013    ///
3014    /// Both a timeout and a send failure trigger the legacy retry, but they
3015    /// mean different things: a send failure says the request never arrived,
3016    /// so it teaches nothing about the peer's capabilities. Caching it would
3017    /// strand that peer in the legacy shape for the rest of the session over
3018    /// one flaky send.
3019    #[test]
3020    fn only_silence_is_evidence_worth_caching() {
3021        assert!(matches!(
3022            Error::Timeout("no answer".into()),
3023            Error::Timeout(_)
3024        ));
3025        assert!(!matches!(
3026            Error::Network("send failed".into()),
3027            Error::Timeout(_)
3028        ));
3029        // Both still take the fallback, so a send failure is retried rather
3030        // than left to fail outright.
3031        assert!(is_version_unaware(&Error::Network("send failed".into())));
3032        assert!(is_version_unaware(&Error::Timeout("no answer".into())));
3033    }
3034
3035    /// One peer cannot condemn the client.
3036    ///
3037    /// Nothing authenticates a refusal, so a single hostile or misconfigured
3038    /// storer answering `ClientUpdateRequired` to everything would otherwise
3039    /// abort every upload. That turns an over-query design which tolerates
3040    /// many bad peers into one that tolerates none.
3041    #[test]
3042    fn a_lone_peer_cannot_condemn_the_client() {
3043        let refusals = SettlementRefusals::default();
3044        assert!(
3045            refusals
3046                .note(PeerId::from_bytes([0x61; 32]), "too old")
3047                .is_none(),
3048            "one peer is not corroboration"
3049        );
3050        assert!(refusals.corroborated().is_none());
3051        // The same peer repeating itself is still one peer.
3052        assert!(refusals
3053            .note(PeerId::from_bytes([0x61; 32]), "too old")
3054            .is_none());
3055        assert!(refusals.corroborated().is_none());
3056    }
3057
3058    /// A genuine incompatibility reaches the threshold at once, because every
3059    /// peer enforcing the newer rule refuses.
3060    #[test]
3061    fn a_second_peer_makes_the_refusal_terminal_and_it_stays_latched() {
3062        let refusals = SettlementRefusals::default();
3063        refusals.note(PeerId::from_bytes([0x62; 32]), "run ant update");
3064        let verdict = refusals.note(PeerId::from_bytes([0x63; 32]), "run ant update");
3065
3066        assert!(verdict.is_some_and(|m| m.contains("ant update")));
3067        // Latched: an upload starting later must see it before it spends,
3068        // which is the whole point of holding it on the client rather than in
3069        // one collector's local state.
3070        assert!(refusals
3071            .corroborated()
3072            .is_some_and(|m| m.contains("ant update")));
3073    }
3074
3075    /// A refusal that does not describe this client is a confused or lying
3076    /// peer, not evidence about this build, and must not count toward the
3077    /// threshold.
3078    #[test]
3079    fn an_incoherent_refusal_is_treated_as_a_bad_peer() {
3080        let peer_id = PeerId::from_bytes([0x64; 32]);
3081
3082        // Claims to be about some other client version.
3083        let wrong_echo = settlement_refusal_error(
3084            &peer_id,
3085            CURRENT_SETTLEMENT_VERSION.saturating_add(7),
3086            CURRENT_SETTLEMENT_VERSION.saturating_add(8),
3087        );
3088        assert!(matches!(wrong_echo, Error::Protocol(_)), "{wrong_echo:?}");
3089
3090        // Claims a minimum that our version already satisfies.
3091        let no_gap = settlement_refusal_error(
3092            &peer_id,
3093            CURRENT_SETTLEMENT_VERSION,
3094            CURRENT_SETTLEMENT_VERSION,
3095        );
3096        assert!(matches!(no_gap, Error::Protocol(_)), "{no_gap:?}");
3097
3098        // A coherent one is believed, and carries the upgrade instruction.
3099        let real = settlement_refusal_error(
3100            &peer_id,
3101            CURRENT_SETTLEMENT_VERSION,
3102            CURRENT_SETTLEMENT_VERSION.saturating_add(1),
3103        );
3104        match real {
3105            Error::ClientUpdateRequired(msg) => assert!(msg.contains("ant update"), "{msg}"),
3106            other => panic!("expected ClientUpdateRequired, got {other:?}"),
3107        }
3108    }
3109}