fix(payment): require issuer-aware SNP quote quorum

ant-core/src/data/client/batch.rs

@@ -7,15 +7,17 @@
 use crate::data::client::adaptive::observe_op;
 use crate::data::client::classify_error;
 use crate::data::client::file::UploadEvent;
-use crate::data::client::payment::peer_id_to_encoded;
+use crate::data::client::payment::{
+    paid_quote_payment_from_store_quotes_with_target_and_views, peer_id_to_encoded,
+};
 use crate::data::client::Client;
 use crate::data::error::{Error, PartialUploadSpend, Result};
 use ant_protocol::evm::{
     Amount, EncodedPeerId, PayForQuotesError, PaymentQuote, ProofOfPayment, QuoteHash,
     RewardsAddress, TxHash,
 };
 use ant_protocol::payment::{
-    deserialize_proof, serialize_single_node_proof, PaymentProof, SingleNodePayment,
+    deserialize_proof, serialize_single_node_proof, PaymentProof, QuotePaymentInfo,
 };
 use ant_protocol::transport::{MultiAddr, PeerId};
 use ant_protocol::{compute_address, XorName, DATA_TYPE_CHUNK};
@@ -36,6 +38,13 @@ const PAYMENT_WAVE_SIZE: usize = 64;
 /// / adaptive limits instead and are unaffected.
 const STORE_INFLIGHT_BYTE_BUDGET: usize = 64 * 1024 * 1024;
 
+/// Payment entries for a prepared chunk.
+#[derive(Debug, Clone)]
+pub struct PreparedChunkPayment {
+    /// Quote payment entries that must be paid before storing.
+    pub quotes: Vec<QuotePaymentInfo>,
+}
+
 /// Chunk quoted but not yet paid. Produced by [`Client::prepare_chunk_payment`].
 #[derive(Debug)]
 pub struct PreparedChunk {
@@ -45,8 +54,8 @@ pub struct PreparedChunk {
     pub address: XorName,
     /// Closest peers from quote collection — PUT targets for close-group replication.
     pub quoted_peers: Vec<(PeerId, Vec<MultiAddr>)>,
-    /// Payment structure (quotes sorted, median selected, not yet paid on-chain).
-    pub payment: SingleNodePayment,
+    /// Payment entries selected for the proof, not yet paid on-chain.
+    pub payment: PreparedChunkPayment,
     /// Peer quotes for building `ProofOfPayment`.
     pub peer_quotes: Vec<(EncodedPeerId, PaymentQuote)>,
 }
@@ -259,19 +268,16 @@ impl Client {
         // Capture all quoted peers for close-group replication.
         let quoted_peers = quote_plan.put_peers;
 
-        // Build peer_quotes for ProofOfPayment + quotes for SingleNodePayment.
-        // Use node-reported prices directly — no contract price fetch needed.
-        let mut peer_quotes = Vec::with_capacity(quotes_with_peers.len());
-        let mut quotes_for_payment = Vec::with_capacity(quotes_with_peers.len());
-
-        for (peer_id, _addrs, quote, price) in quotes_with_peers {
-            let encoded = peer_id_to_encoded(&peer_id)?;
-            peer_quotes.push((encoded, quote.clone()));
-            quotes_for_payment.push((quote, price));
-        }
-
-        let payment = SingleNodePayment::from_quotes(quotes_for_payment)
-            .map_err(|e| Error::Payment(format!("Failed to create payment: {e}")))?;
+        let (paid_peer_id, paid_quote, paid_quote_info) =
+            paid_quote_payment_from_store_quotes_with_target_and_views(
+                &quotes_with_peers,
+                quote_plan.paid_quote_acceptance_target,
+                &quote_plan.witness_views,
+            )?;
+        let peer_quotes = vec![(peer_id_to_encoded(&paid_peer_id)?, paid_quote)];
+        let payment = PreparedChunkPayment {
+            quotes: vec![paid_quote_info],
+        };
 
         Ok(Some(PreparedChunk {
             content,
@@ -307,17 +313,19 @@ impl Client {
         let intent = PaymentIntent::from_prepared_chunks(&prepared);
         let storage_cost_atto = intent.total_amount.to_string();
 
-        // Flatten all quote payments from all chunks into a single batch.
-        let total_quotes: usize = prepared.iter().map(|c| c.payment.quotes.len()).sum();
-        let mut all_payments = Vec::with_capacity(total_quotes);
+        // Flatten all paid quote entries from all chunks into a single batch.
+        let total_payments: usize = intent.payments.len();
+        let mut all_payments = Vec::with_capacity(total_payments);
         for chunk in &prepared {
             for info in &chunk.payment.quotes {
-                all_payments.push((info.quote_hash, info.rewards_address, info.amount));
+                if !info.amount.is_zero() {
+                    all_payments.push((info.quote_hash, info.rewards_address, info.amount));
+                }
             }
         }
 
         debug!(
-            "Batch payment for {} chunks ({} quote entries)",
+            "Batch payment for {} chunks ({} paid quote entries)",
             prepared.len(),
             all_payments.len()
         );
@@ -549,18 +557,37 @@ impl Client {
             // freshly-quoted peers, bypassing the EVM transaction.
             let mut needs_pay: Vec<PreparedChunk> = Vec::with_capacity(prepared_chunks.len());
             let mut cached_paid: Vec<PaidChunk> = Vec::new();
+            let mut stale_cached: Vec<([u8; 32], Vec<u8>)> = Vec::new();
             for prep in prepared_chunks {
                 if let Some(proof_bytes) = cached_proofs.get(&prep.address).cloned() {
-                    cached_paid.push(PaidChunk {
-                        content: prep.content,
-                        address: prep.address,
-                        quoted_peers: prep.quoted_peers,
-                        proof_bytes,
-                    });
+                    if cached_proof_matches_current_quote_floor(&proof_bytes, &prep) {
+                        cached_paid.push(PaidChunk {
+                            content: prep.content,
+                            address: prep.address,
+                            quoted_peers: prep.quoted_peers,
+                            proof_bytes,
+                        });
+                    } else {
+                        stale_cached.push((prep.address, proof_bytes));
+                        needs_pay.push(prep);
+                    }
                 } else {
                     needs_pay.push(prep);
                 }
             }
+            if let Some(key) = resume_key {
+                if !stale_cached.is_empty() {
+                    info!(
+                        "Wave {wave_num}/{wave_count}: discarding {} cached payment proofs \
+                         below current quote floor",
+                        stale_cached.len()
+                    );
+                    crate::data::client::cached_single::try_drop_proofs_for_file(
+                        key,
+                        &stale_cached,
+                    );
+                }
+            }
             if !cached_paid.is_empty() {
                 info!(
                     "Wave {wave_num}/{wave_count}: reusing {} cached payment proofs",
@@ -901,6 +928,38 @@ fn log_wave_summary(result: &WaveResult) {
     );
 }
 
+/// Returns true when a cached proof's embedded quote prices still meet
+/// the current fresh quote floor selected for this chunk.
+fn cached_proof_matches_current_quote_floor(proof_bytes: &[u8], prepared: &PreparedChunk) -> bool {
+    let Some(current_floor) = prepared
+        .payment
+        .quotes
+        .iter()
+        .map(|quote| quote.price)
+        .max()
+    else {
+        return false;
+    };
+
+    match deserialize_proof(proof_bytes) {
+        Ok((proof, _tx_hashes)) => proof_matches_current_quote_floor(&proof, current_floor),
+        Err(_) => false,
+    }
+}
+
+/// A cached proof can only be replayed if every embedded quote is at least
+/// as expensive as the fresh quote floor. With one paid quote this is the
+/// exact check; with older multi-quote proofs it is deliberately
+/// conservative because the storer may derive its payable quote from the
+/// full embedded set.
+fn proof_matches_current_quote_floor(proof: &ProofOfPayment, current_floor: Amount) -> bool {
+    !proof.peer_quotes.is_empty()
+        && proof
+            .peer_quotes
+            .iter()
+            .all(|(_, quote)| quote.price >= current_floor)
+}
+
 /// Safety margin subtracted from the storer's `QUOTE_MAX_AGE_SECS` (24 h)
 /// when deciding to trust a cached proof.
 ///
@@ -1047,32 +1106,37 @@ mod send_assertions {
 #[allow(clippy::unwrap_used)]
 mod tests {
     use super::*;
-    use ant_protocol::payment::QuotePaymentInfo;
-    use ant_protocol::CLOSE_GROUP_SIZE;
-
-    /// Median index in the quotes array.
-    const MEDIAN_INDEX: usize = CLOSE_GROUP_SIZE / 2;
-
-    /// Helper: build a `PreparedChunk` with `median_amount` at the median
-    /// quote index and zero for all other quotes. Adapts automatically to
-    /// `CLOSE_GROUP_SIZE` changes.
-    fn make_prepared_chunk(median_amount: u64) -> PreparedChunk {
-        let quotes: [QuotePaymentInfo; CLOSE_GROUP_SIZE] = std::array::from_fn(|i| {
-            let amount = if i == MEDIAN_INDEX { median_amount } else { 0 };
-            QuotePaymentInfo {
-                quote_hash: QuoteHash::from([i as u8 + 1; 32]),
-                rewards_address: RewardsAddress::new([i as u8 + 10; 20]),
-                amount: Amount::from(amount),
-                price: Amount::from(amount),
-            }
-        });
-
+    use std::time::SystemTime;
+    use xor_name::XorName as QuoteXorName;
+
+    const TEST_QUOTE_HASH_SEED: u8 = 1;
+    const TEST_PEER_SEED: u8 = 7;
+    const TEST_REWARDS_SEED: u8 = 10;
+
+    /// Helper: build a `PreparedChunk` with one selected paid quote.
+    fn make_prepared_chunk(payment_amount: u64) -> PreparedChunk {
+        let quote = QuotePaymentInfo {
+            quote_hash: QuoteHash::from([TEST_QUOTE_HASH_SEED; 32]),
+            rewards_address: RewardsAddress::new([TEST_REWARDS_SEED; 20]),
+            amount: Amount::from(payment_amount),
+            price: Amount::from(payment_amount),
+        };
+        let peer_quote = PaymentQuote {
+            content: QuoteXorName([0u8; 32]),
+            timestamp: SystemTime::UNIX_EPOCH,
+            price: Amount::from(payment_amount),
+            rewards_address: RewardsAddress::new([TEST_REWARDS_SEED; 20]),
+            pub_key: Vec::new(),
+            signature: Vec::new(),
+        };
         PreparedChunk {
             content: Bytes::from(vec![0xAA; 32]),
             address: [0u8; 32],
             quoted_peers: Vec::new(),
-            payment: SingleNodePayment { quotes },
-            peer_quotes: Vec::new(),
+            payment: PreparedChunkPayment {
+                quotes: vec![quote],
+            },
+            peer_quotes: vec![(EncodedPeerId::from([TEST_PEER_SEED; 32]), peer_quote)],
         }
     }
 
@@ -1085,8 +1149,8 @@ mod tests {
         assert_eq!(intent.total_amount, Amount::from(300));
 
         let (hash, addr, amt) = &intent.payments[0];
-        assert_eq!(*hash, QuoteHash::from([MEDIAN_INDEX as u8 + 1; 32]));
-        assert_eq!(*addr, RewardsAddress::new([MEDIAN_INDEX as u8 + 10; 20]));
+        assert_eq!(*hash, QuoteHash::from([TEST_QUOTE_HASH_SEED; 32]));
+        assert_eq!(*addr, RewardsAddress::new([TEST_REWARDS_SEED; 20]));
         assert_eq!(*amt, Amount::from(300));
     }
 
@@ -1119,7 +1183,7 @@ mod tests {
     #[test]
     fn finalize_batch_payment_builds_proofs() {
         let chunk = make_prepared_chunk(500);
-        let quote_hash = chunk.payment.quotes[MEDIAN_INDEX].quote_hash;
+        let quote_hash = chunk.payment.quotes[0].quote_hash;
 
         let mut tx_map = HashMap::new();
         tx_map.insert(quote_hash, TxHash::from([0xBB; 32]));
@@ -1129,6 +1193,14 @@ mod tests {
         assert_eq!(paid.len(), 1);
         assert!(!paid[0].proof_bytes.is_empty());
         assert_eq!(paid[0].address, [0u8; 32]);
+
+        let (proof, tx_hashes) = deserialize_proof(&paid[0].proof_bytes).unwrap();
+        assert_eq!(proof.peer_quotes.len(), 1);
+        assert_eq!(
+            proof.peer_quotes[0].0,
+            EncodedPeerId::from([TEST_PEER_SEED; 32])
+        );
+        assert_eq!(tx_hashes, vec![TxHash::from([0xBB; 32])]);
     }
 
     #[test]
@@ -1153,7 +1225,7 @@ mod tests {
     fn finalize_batch_payment_multiple_chunks() {
         let c1 = make_prepared_chunk(100);
         let c2 = make_prepared_chunk(200);
-        let q1 = c1.payment.quotes[MEDIAN_INDEX].quote_hash;
+        let q1 = c1.payment.quotes[0].quote_hash;
         let mut tx_map = HashMap::new();
         // Both chunks have the same quote_hash (same index/byte pattern)
         // so one tx_hash covers both
@@ -1191,6 +1263,62 @@ mod tests {
         ProofOfPayment { peer_quotes }
     }
 
+    fn make_proof_with_prices(prices: &[u64]) -> ProofOfPayment {
+        let peer_quotes = prices
+            .iter()
+            .enumerate()
+            .map(|(i, price)| {
+                let quote = PaymentQuote {
+                    content: xor_name::XorName([0u8; 32]),
+                    timestamp: SystemTime::UNIX_EPOCH,
+                    price: Amount::from(*price),
+                    rewards_address: RewardsAddress::new([1u8; 20]),
+                    pub_key: vec![],
+                    signature: vec![],
+                };
+                (EncodedPeerId::from([i as u8; 32]), quote)
+            })
+            .collect();
+        ProofOfPayment { peer_quotes }
+    }
+
+    #[test]
+    fn proof_matches_current_quote_floor_accepts_current_or_higher_price() {
+        let proof = make_proof_with_prices(&[120]);
+
+        assert!(proof_matches_current_quote_floor(&proof, Amount::from(100)));
+    }
+
+    #[test]
+    fn proof_matches_current_quote_floor_rejects_stale_lower_price() {
+        let proof = make_proof_with_prices(&[80]);
+
+        assert!(!proof_matches_current_quote_floor(
+            &proof,
+            Amount::from(100)
+        ));
+    }
+
+    #[test]
+    fn proof_matches_current_quote_floor_rejects_any_lower_multi_quote_price() {
+        let proof = make_proof_with_prices(&[100, 80, 130]);
+
+        assert!(!proof_matches_current_quote_floor(
+            &proof,
+            Amount::from(100)
+        ));
+    }
+
+    #[test]
+    fn proof_matches_current_quote_floor_rejects_empty_proof() {
+        let proof = make_proof_with_prices(&[]);
+
+        assert!(!proof_matches_current_quote_floor(
+            &proof,
+            Amount::from(100)
+        ));
+    }
+
     fn default_max_future_skew() -> Duration {
         Duration::from_secs(CACHED_PROOF_FUTURE_SKEW_TOLERANCE_SECS)
     }

ant-core/src/data/client/file.rs

@@ -20,6 +20,7 @@ use crate::data::client::merkle::{
     merkle_store_with_retry, should_use_merkle, MerkleBatchPaymentResult, PaymentMode,
     PreparedMerkleBatch, DEFERRED_ROUND_DELAYS_SECS,
 };
+use crate::data::client::payment::paid_quote_payment_from_store_quotes;
 use crate::data::client::Client;
 use crate::data::error::{Error, PartialUploadSpend, Result};
 use ant_protocol::evm::{Amount, PaymentQuote, QuoteHash, TxHash, MAX_LEAVES};
@@ -1154,15 +1155,10 @@ impl Client {
             }
         };
 
-        // Use the median price × 3 (matches SingleNodePayment::from_quotes
-        // which pays 3x the median to incentivize reliable storage).
-        let mut prices: Vec<Amount> = quotes.iter().map(|(_, _, _, price)| *price).collect();
-        prices.sort();
-        let median_price = prices
-            .get(prices.len() / 2)
-            .copied()
-            .unwrap_or(Amount::ZERO);
-        let per_chunk_cost = median_price * Amount::from(3u64);
+        // Match the live single-node payment path: one selected quote is paid
+        // at the storer-required multiplier.
+        let (_, _, paid_quote_info) = paid_quote_payment_from_store_quotes(&quotes)?;
+        let per_chunk_cost = paid_quote_info.amount;
 
         let chunk_count_u64 = u64::try_from(chunk_count).unwrap_or(u64::MAX);
         let total_storage = per_chunk_cost * Amount::from(chunk_count_u64);