feat(payment): settle merkle sub-batches in batched payForMerkleTrees groups

CHANGELOG.md

@@ -7,6 +7,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Changed (wallet path — batched merkle payments, V2-990)
+- Wallet-path merkle uploads larger than one tree (`MAX_LEAVES` = 256 chunks ≈ 1 GiB)
+  now settle in batched `payForMerkleTrees` transactions: sub-batches are paid in
+  groups of `MERKLE_TREES_PER_PAYMENT` (4) trees per on-chain transaction instead of
+  one transaction per tree. Partial-payment semantics are preserved at group
+  granularity — a failed group still returns the proofs of previously-paid groups,
+  and the failed group itself pays nothing (the batched entry point is atomic).
+  Requires a payment vault deployment carrying the batched entry point (V2-992);
+  the cap is re-exported as `ant_core::data::MERKLE_TREES_PER_PAYMENT` so
+  consumers don't hardcode it.
+
 ### Changed (breaking — external-signer merkle API, ADR-0003)
 - External-signer merkle uploads are no longer capped at one payment batch (`MAX_LEAVES` = 256 chunks ≈ 1 GiB): `file_prepare_upload*` now partitions the to-upload set into `MerkleTree`-sized sub-batches (`ExternalPaymentInfo::Merkle` carries `prepared_batches: Vec<PreparedMerkleBatch>`), the signer pays one transaction per batch, and the new `Client::finalize_upload_merkle_multi` takes one winner hash per batch. `finalize_upload_merkle` remains as the single-batch special case. A batch the signer never paid (`None` hash) no longer aborts the upload: paid batches store and the unpaid chunks surface via `Error::PartialUpload`.
 - External-signer merkle prepared uploads no longer hold the encrypted file in memory: chunk bodies stay in the on-disk encryption spill (opaque `ExternalChunkStore` inside `ExternalPaymentInfo::Merkle`, replacing the resident `chunk_contents: Vec<Bytes>`), and finalize stores them via the wallet path's bounded spill fan-out — peak RAM ~256 MB regardless of file size, plus deferred-retry rounds the external path previously lacked.

Cargo.toml

@@ -1,3 +1,12 @@
 [workspace]
 members = ["ant-core", "ant-cli"]
 resolver = "2"
+
+# DRAFT-ONLY: compile against the unreleased batched-payment surface --
+# evmlib `payForMerkleTrees` (WithAutonomi/evmlib#15) re-exported by
+# ant-protocol (WithAutonomi/ant-protocol#26). Before this PR leaves draft:
+# drop both pins and bump ant-core's `ant-protocol` dependency to the
+# released version carrying them.
+[patch.crates-io]
+evmlib = { git = "https://github.com/WithAutonomi/evmlib", branch = "feat/pay-for-merkle-trees" }
+ant-protocol = { git = "https://github.com/WithAutonomi/ant-protocol", branch = "feat/batched-merkle-reexports" }

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

@@ -11,7 +11,7 @@ use crate::data::client::Client;
 use crate::data::error::{Error, Result};
 use ant_protocol::evm::{
     Amount, MerklePaymentCandidateNode, MerklePaymentCandidatePool, MerklePaymentProof, MerkleTree,
-    MidpointProof, PoolCommitment, CANDIDATES_PER_POOL, MAX_LEAVES,
+    MerkleTreePayment, MidpointProof, PoolCommitment, CANDIDATES_PER_POOL, MAX_LEAVES,
 };
 use ant_protocol::payment::commitment::{
     commitment_hash, verify_commitment_signature, StorageCommitment, MAX_COMMITMENT_KEY_COUNT,
@@ -37,6 +37,16 @@ use xor_name::XorName;
 /// Default threshold: use merkle payments when chunk count >= this value.
 pub const DEFAULT_MERKLE_THRESHOLD: usize = 64;
 
+/// Number of merkle trees the wallet path packs into a single on-chain
+/// `payForMerkleTrees` transaction. Re-exported from `evmlib` (via
+/// `ant-protocol`) so consumers (desktop app, FFI) size their batches from
+/// the shared constant instead of hardcoding it.
+pub use ant_protocol::evm::MERKLE_TREES_PER_PAYMENT;
+
+// `chunks()` panics on zero; the evmlib constant is static-asserted against
+// the on-chain upper bound but not against zero, so pin that here.
+const _: () = assert!(MERKLE_TREES_PER_PAYMENT > 0);
+
 /// Payment multiplier applied to a quoted price before settlement.
 ///
 /// Deliberately the **same constant** the single-node path uses rather than a
@@ -874,6 +884,14 @@ impl Client {
     }
 
     /// Handle batches larger than `MAX_LEAVES` by splitting into sub-batches.
+    ///
+    /// Sub-batches settle in groups of [`MERKLE_TREES_PER_PAYMENT`] trees:
+    /// each group is ONE atomic `payForMerkleTrees` transaction, so a large
+    /// upload needs one signature per group instead of one per tree. Partial
+    /// semantics are preserved at group granularity: a failing group returns
+    /// the proofs of prior groups so the caller can still store already-paid
+    /// chunks, and the group's own trees are untouched on-chain (the batched
+    /// entry point is all-or-nothing).
     async fn pay_for_merkle_multi_batch(
         &self,
         addresses: &[[u8; 32]],
@@ -886,62 +904,102 @@ impl Client {
         // upload into a partial failure.
         let sub_batches = merkle_batch_partitions(addresses);
         let total_sub_batches = sub_batches.len();
-        let mut all_proofs = HashMap::with_capacity(addresses.len());
-        let mut total_storage = Amount::ZERO;
-        let mut total_gas: u128 = 0;
-        // Track the oldest sub-batch timestamp so the overall receipt
-        // expires when the *first* sub-batch's on-chain payment ages
-        // out (worst case for resume).
-        let mut oldest_ts: u64 = 0;
-
-        for (i, chunk) in sub_batches.into_iter().enumerate() {
+        let group_count = total_sub_batches.div_ceil(MERKLE_TREES_PER_PAYMENT);
+        info!(
+            "Paying {total_sub_batches} merkle sub-batches in {group_count} batched \
+             transaction(s) of up to {MERKLE_TREES_PER_PAYMENT} trees"
+        );
+
+        let mut group_results: Vec<MerkleBatchPaymentResult> = Vec::with_capacity(group_count);
+        for (i, group) in sub_batches.chunks(MERKLE_TREES_PER_PAYMENT).enumerate() {
             match self
-                .pay_for_merkle_single_batch(chunk, data_type, data_size)
+                .pay_for_merkle_tree_group(group, data_type, data_size)
                 .await
             {
-                Ok(sub_result) => {
-                    if let Ok(cost) = sub_result.storage_cost_atto.parse::<Amount>() {
-                        total_storage += cost;
-                    }
-                    total_gas = total_gas.saturating_add(sub_result.gas_cost_wei);
-                    if oldest_ts == 0
-                        || (sub_result.merkle_payment_timestamp > 0
-                            && sub_result.merkle_payment_timestamp < oldest_ts)
-                    {
-                        oldest_ts = sub_result.merkle_payment_timestamp;
-                    }
-                    all_proofs.extend(sub_result.proofs);
-                }
+                Ok(group_result) => group_results.push(group_result),
                 Err(e) => {
-                    if all_proofs.is_empty() {
-                        // First sub-batch failed, nothing paid yet -- propagate directly.
+                    if group_results.is_empty() {
+                        // First group failed, nothing paid yet -- propagate directly.
                         return Err(e);
                     }
                     // Return partial result so caller can still store already-paid chunks.
                     warn!(
-                        "Merkle sub-batch {}/{total_sub_batches} failed: {e}. \
-                         Returning {} proofs from prior sub-batches",
+                        "Merkle payment group {}/{group_count} ({} of {total_sub_batches} \
+                         sub-batches) failed: {e}. Returning proofs from prior groups",
                         i + 1,
-                        all_proofs.len()
+                        group.len()
                     );
-                    return Ok(MerkleBatchPaymentResult {
-                        chunk_count: all_proofs.len(),
-                        proofs: all_proofs,
-                        storage_cost_atto: total_storage.to_string(),
-                        gas_cost_wei: total_gas,
-                        merkle_payment_timestamp: oldest_ts,
-                    });
+                    break;
                 }
             }
         }
 
-        Ok(MerkleBatchPaymentResult {
-            chunk_count: addresses.len(),
-            proofs: all_proofs,
-            storage_cost_atto: total_storage.to_string(),
-            gas_cost_wei: total_gas,
-            merkle_payment_timestamp: oldest_ts,
-        })
+        Ok(merge_merkle_batch_results(group_results))
+    }
+
+    /// Pay one group of up to [`MERKLE_TREES_PER_PAYMENT`] sub-batches
+    /// atomically: prepare every tree, submit a single `payForMerkleTrees`
+    /// transaction, then generate proofs per tree from the per-tree winner
+    /// pools the contract returned (aligned to input order).
+    async fn pay_for_merkle_tree_group(
+        &self,
+        group: &[&[[u8; 32]]],
+        data_type: u32,
+        data_size: u64,
+    ) -> Result<MerkleBatchPaymentResult> {
+        let wallet = self.require_wallet()?;
+
+        let mut prepared_group = Vec::with_capacity(group.len());
+        for chunk in group {
+            prepared_group.push(
+                self.prepare_merkle_batch_external(chunk, data_type, data_size)
+                    .await?,
+            );
+        }
+
+        let trees: Vec<MerkleTreePayment> = prepared_group
+            .iter()
+            .map(|prepared| MerkleTreePayment {
+                depth: prepared.depth,
+                merkle_payment_timestamp: prepared.merkle_payment_timestamp,
+                pool_commitments: prepared.pool_commitments.clone(),
+            })
+            .collect();
+
+        info!(
+            "Submitting batched merkle payment on-chain ({} tree(s), depths {:?})",
+            trees.len(),
+            trees.iter().map(|t| t.depth).collect::<Vec<_>>()
+        );
+        let (payments, gas_info) = wallet
+            .pay_for_merkle_trees(trees)
+            .await
+            .map_err(|e| Error::Payment(format!("Batched merkle payment failed: {e}")))?;
+
+        if payments.len() != prepared_group.len() {
+            return Err(Error::Payment(format!(
+                "Batched merkle payment returned {} result(s) for {} trees",
+                payments.len(),
+                prepared_group.len()
+            )));
+        }
+
+        let mut per_tree = Vec::with_capacity(prepared_group.len());
+        for (prepared, (winner_pool_hash, amount)) in prepared_group.into_iter().zip(payments) {
+            info!(
+                "Merkle payment succeeded: winner pool {}",
+                hex::encode(winner_pool_hash)
+            );
+            let mut result = finalize_merkle_batch(prepared, winner_pool_hash)?;
+            result.storage_cost_atto = amount.to_string();
+            per_tree.push(result);
+        }
+
+        // One transaction paid for the whole group: fold the per-tree results
+        // and attach the group's single gas cost.
+        let mut merged = merge_merkle_batch_results(per_tree);
+        merged.gas_cost_wei = gas_info.gas_cost_wei;
+        Ok(merged)
     }
 
     /// Build candidate pools by querying the network for each midpoint (concurrently).

ant-core/src/data/mod.rs

@@ -34,7 +34,7 @@ pub use client::file::{
 };
 pub use client::merkle::{
     finalize_merkle_batch, MerkleBatchPaymentResult, PaymentMode, PreparedMerkleBatch,
-    DEFAULT_MERKLE_THRESHOLD,
+    DEFAULT_MERKLE_THRESHOLD, MERKLE_TREES_PER_PAYMENT,
 };
 
 // Re-export self-encryption types

ant-core/tests/e2e_merkle.rs

@@ -294,9 +294,16 @@ async fn test_attack_merkle_proof_swap_within_batch() {
 /// out of self-encryption needs a ~1 GB file, while `pay_for_merkle_batch`
 /// takes the address set straight.
 ///
+/// 1025 stretches the multi-batch path across the `MERKLE_TREES_PER_PAYMENT`
+/// group boundary: it partitions as five trees (`[256, 256, 256, 255, 2]`),
+/// which the wallet path settles as TWO batched `payForMerkleTrees`
+/// transactions — a full group of 4 and a singleton group — so both the
+/// grouped and the single-tree shape of the batched entry point settle
+/// against the real contract.
+///
 /// Settlement is checked against the same padded-leaf model the estimator
-/// bills with — 65 addresses settle 128 leaves, 257 settle 256 + 2 — so the
-/// two counts must cost in that ratio.
+/// bills with — 65 addresses settle 128 leaves, 257 settle 256 + 2, 1025
+/// settles 4×256 + 2 — so consecutive counts must cost in that ratio.
 #[tokio::test(flavor = "multi_thread")]
 #[serial]
 async fn test_merkle_payment_across_batch_boundary() {
@@ -317,7 +324,7 @@ async fn test_merkle_payment_across_batch_boundary() {
 
     let mut paid: Vec<(usize, u128)> = Vec::new();
 
-    for (count, tag) in [(65usize, 0xA1u8), (257usize, 0xB2u8)] {
+    for (count, tag) in [(65usize, 0xA1u8), (257usize, 0xB2u8), (1025usize, 0xC3u8)] {
         let addrs = addresses(count, tag);
 
         // A 65/257-address tree collects far more candidate pools than the
@@ -387,19 +394,21 @@ async fn test_merkle_payment_across_batch_boundary() {
     }
 
     // Prices are uniform across a freshly started local testnet and this test
-    // stores nothing, so the only thing separating the two settlements is the
+    // stores nothing, so the only thing separating the settlements is the
     // padded leaf count each partition pays for.
-    let [(small, small_atto), (large, large_atto)] = paid[..] else {
-        panic!("expected two payments");
-    };
-    let expected =
-        merkle_billable_leaves(large as u64) as f64 / merkle_billable_leaves(small as u64) as f64;
-    let observed = large_atto as f64 / small_atto as f64;
-    assert!(
-        (observed - expected).abs() / expected < 0.15,
-        "settlement should scale with padded leaves: expected ~{expected:.3}x \
-         ({small} -> {large} addresses), observed {observed:.3}x"
-    );
+    for pair in paid.windows(2) {
+        let [(small, small_atto), (large, large_atto)] = pair else {
+            panic!("expected consecutive payments");
+        };
+        let expected = merkle_billable_leaves(*large as u64) as f64
+            / merkle_billable_leaves(*small as u64) as f64;
+        let observed = *large_atto as f64 / *small_atto as f64;
+        assert!(
+            (observed - expected).abs() / expected < 0.15,
+            "settlement should scale with padded leaves: expected ~{expected:.3}x \
+             ({small} -> {large} addresses), observed {observed:.3}x"
+        );
+    }
 
     drop(client);
     testnet.teardown().await;