ant_core/data/client/mod.rs
1//! Client operations for the Autonomi network.
2//!
3//! Provides high-level APIs for storing and retrieving data
4//! on the Autonomi decentralized network.
5
6pub mod adaptive {
7 pub use crate::client_engine::adaptive::*;
8}
9pub mod batch;
10pub mod cache;
11#[cfg(feature = "native")]
12pub(crate) mod cached_merkle;
13#[cfg(feature = "native")]
14pub(crate) mod cached_single;
15pub mod chunk;
16pub mod data;
17#[cfg(feature = "native")]
18pub mod diagnostics;
19pub mod file;
20pub mod merkle;
21#[cfg(feature = "native")]
22mod native_payment;
23pub mod payment;
24pub mod quote;
25pub mod upload;
26pub mod upload_state;
27
28use crate::data::client::adaptive::{AdaptiveConfig, AdaptiveController, ChannelStart, Outcome};
29use crate::data::client::cache::ChunkCache;
30use crate::data::error::{Error, Result};
31use crate::data::network::Network;
32#[cfg(feature = "native")]
33use crate::data::network::NetworkHealth;
34#[cfg(feature = "native")]
35use crate::data::peer_cache;
36use ant_protocol::evm::Wallet;
37#[cfg(feature = "native")]
38use ant_protocol::transport::P2PNode;
39use ant_protocol::transport::{MultiAddr, PeerId};
40use ant_protocol::{XorName, CLOSE_GROUP_SIZE};
41use std::collections::HashSet;
42use std::path::PathBuf;
43use std::sync::atomic::{AtomicU64, Ordering};
44use std::sync::Arc;
45use std::sync::Mutex;
46#[cfg(feature = "native")]
47use tracing::debug;
48
49/// Width of the chunk PUT-target set (initial writes plus fallback): the
50/// closest `PUT_TARGET_WIDTH` peers to the address.
51///
52/// Mirrors the node-side `K_BUCKET_SIZE` / `PAID_QUOTE_ISSUER_CLOSENESS_WIDTH`
53/// (20): a node accepts a reused payment proof only when one of the proof's
54/// closest-`CLOSE_GROUP_SIZE` quote issuers is within its own local 20-closest,
55/// so trying peers past this width is pointless.
56pub(crate) use crate::quote_policy::PUT_TARGET_WIDTH;
57
58/// Ceiling on how long to wait for a peer to answer a settlement-versioned
59/// quote request before falling back to the unversioned shape.
60///
61/// A storer that predates the versioned request cannot decode it and never
62/// replies, so the only way to find out is to wait.
63///
64/// Note what is being waited for. This is **not** a cheap parse check: a peer
65/// that understands the request runs the whole quote handler, queueing,
66/// storage reads, pricing and signing, so the answer takes as long as any
67/// quote takes. Abandoning the wait does not cancel that work either, it just
68/// stops listening and adds a duplicate legacy request on top. So the wait has
69/// to be sized for a real quote, not for a ping.
70///
71/// The full timeout is what made this expensive. The merkle E2E suite runs
72/// with `quote_timeout_secs = 120`, so every probe against a fleet on the
73/// published node cost two minutes and the suite blew the 60-minute CI cap.
74///
75/// # It must never bind in production
76///
77/// **Keep this at or above the largest production `quote_timeout_secs`**,
78/// currently 10s. That is not a tuning preference, it is the safety property.
79///
80/// This wait is the only window in which a peer can refuse. Abandoning it
81/// early does not merely mislabel a slow peer: it drops the refusal, so it
82/// never counts toward corroboration, never sets the latch, and the legacy
83/// request it raced can return a perfectly good quote the client then pays
84/// against.
85///
86/// The two fallbacks lose it by different mechanisms. The merkle path sends
87/// its legacy request under a *new* request id (`merkle.rs`), so a refusal
88/// arriving after the ceiling is answering a request nobody is listening to
89/// and is discarded on the id mismatch. The single-node path reuses the same
90/// id (`quote.rs`), so a late refusal is seen only if it happens to arrive
91/// after the retry has resubscribed; the await that would have matched it is
92/// already gone, and anything landing in the gap between the two waits is
93/// lost. Neither path observes a late refusal reliably, which is what the
94/// ceiling exists to prevent.
95///
96/// A shorter ceiling was tried, at 5s, to bring the slower CI runner under the
97/// cap. It would have turned every legitimate 5-to-10 second refusal in
98/// production into exactly that silent downgrade. The never-demote rule does
99/// not help: it stops a peer being *cached* as legacy after it has answered
100/// once, but it does not stop the request in flight from falling back. And the
101/// compile-time guard does not help either, because it binds future builds
102/// while the clients at risk are the ones already released.
103///
104/// So the ceiling exists solely to bound configurations that set a timeout far
105/// above any real answer time, which in practice means test harnesses. Above
106/// 10s it never binds on a production client, and nothing real is truncated.
107///
108/// The cost of leaving it here is roughly two minutes per merkle E2E test
109/// while the suite's devnet still speaks the pre-versioned dialect. That is a
110/// consequence of the temporary fork-branch protocol pin, not of the design:
111/// once the fleet under test can answer a versioned request there are no
112/// probes to pay for, and the suite returns to its baseline.
113pub(crate) const VERSIONED_QUOTE_PROBE_CEILING: std::time::Duration =
114 std::time::Duration::from_secs(15);
115
116/// How many distinct peers must refuse this client's settlement version before
117/// the refusal is believed and uploads stop.
118///
119/// Nothing authenticates a refusal, so one peer's word cannot be enough: a
120/// single hostile or misconfigured storer answering `ClientUpdateRequired` to
121/// everything would otherwise deny every upload, turning an over-query design
122/// that tolerates many bad peers into one that tolerates none.
123///
124/// Two is deliberately low. A genuine incompatibility reaches it instantly,
125/// because every peer enforcing the newer rule refuses, and a client queries
126/// far more than two. An attacker has to control two of the peers a given
127/// request happens to reach, which is a materially different proposition from
128/// controlling one.
129pub(crate) const SETTLEMENT_REFUSAL_QUORUM: usize = 2;
130
131/// Compile-time cutover guard shared by **every** unversioned quote retry.
132///
133/// Both quote paths fall back to an unversioned request when a peer stays
134/// silent, because a storer built before the settlement version existed cannot
135/// decode the versioned one. Silence is not proof of that, though: a dropped
136/// response, packet loss, an overloaded peer, or one deliberately discarding
137/// versioned requests are indistinguishable from the client side. So the
138/// fallback is a downgrade path and can be provoked.
139///
140/// It is not only literal silence, either. `send_and_await_chunk_response`
141/// keeps waiting when a reply decodes but carries a body the mapper does not
142/// recognise, so a peer answering with an unexpected variant also lands on the
143/// timeout and takes this path. That grants no capability beyond staying
144/// silent, which is why it is bounded rather than special-cased, but a comment
145/// claiming "any structured response prevents the retry" would be wrong.
146///
147/// It is safe only while **no refusal of either kind is possible**, and that
148/// means bounding both constants, not just the minimum.
149///
150/// Raising `MIN` is the obvious hazard: a client below it would be refused,
151/// and the retry hands it a quote anyway. Raising `CURRENT` is the subtler
152/// one. As soon as some node runs a newer `CURRENT` than another, the older
153/// node refuses newer clients with `StorerUpdateRequired` precisely because it
154/// cannot promise to honour their payment. A client that retries such a peer
155/// unversioned gets that unhonourable quote, and for a settlement change that
156/// is not a pure increase the payment is then rejected after it has settled.
157/// Guarding only `MIN` would leave that route open.
158///
159/// So the guard requires both to still be at the first declarable version.
160///
161/// This bounds **future builds** only. A client binary already in the field
162/// carries whatever fallback it shipped with, and no source change reaches it;
163/// that is inherent to shipping software and is why the storer still verifies
164/// every payment it is actually offered.
165///
166/// Each fallback site references this constant so the guard cannot be orphaned
167/// by deleting one path and forgetting the other. Retiring the fallbacks means
168/// deleting this constant and every reference to it, which the compiler then
169/// points at one by one.
170pub(crate) const UNVERSIONED_RETRY_REQUIRES_MIN_V1: () = assert!(
171 ant_protocol::MIN_SUPPORTED_SETTLEMENT_VERSION == 1
172 && ant_protocol::CURRENT_SETTLEMENT_VERSION == 1,
173 "an unversioned quote retry is still compiled in: it is a downgrade path around \
174 both the too-old and the node-behind refusals, so delete every fallback site \
175 before raising MIN_SUPPORTED_SETTLEMENT_VERSION or CURRENT_SETTLEMENT_VERSION"
176);
177
178/// Distinct peers that have refused this client's settlement version, plus the
179/// wording of the first refusal.
180///
181/// A separate type rather than a field so the corroboration rule can be tested
182/// on its own: it is the piece that decides whether an upload stops, and it
183/// has to hold against both a lone lying peer and a genuine incompatibility.
184#[derive(Clone, Default)]
185pub(crate) struct SettlementRefusals {
186 inner: Arc<Mutex<(HashSet<PeerId>, Option<String>)>>,
187}
188
189impl SettlementRefusals {
190 /// Record a refusal from `peer_id`, returning the wording once
191 /// [`SETTLEMENT_REFUSAL_QUORUM`] distinct peers agree and `None` below it.
192 pub(crate) fn note(&self, peer_id: PeerId, message: &str) -> Option<String> {
193 let mut guard = self.inner.lock().ok()?;
194 let (peers, wording) = &mut *guard;
195 peers.insert(peer_id);
196 if wording.is_none() {
197 *wording = Some(message.to_string());
198 }
199 (peers.len() >= SETTLEMENT_REFUSAL_QUORUM)
200 .then(|| wording.clone())
201 .flatten()
202 }
203
204 /// The corroborated refusal, if one has been established.
205 pub(crate) fn corroborated(&self) -> Option<String> {
206 let guard = self.inner.lock().ok()?;
207 let (peers, wording) = &*guard;
208 (peers.len() >= SETTLEMENT_REFUSAL_QUORUM)
209 .then(|| wording.clone())
210 .flatten()
211 }
212
213 /// The distinct refusing peers recorded so far, rendered for logging,
214 /// sorted so the line is deterministic.
215 ///
216 /// Exists so the terminal abort can NAME its corroborators. Without it the
217 /// quorum is unverifiable from outside: "awaiting corroboration" is only
218 /// logged below the quorum and the terminal error carries no peer ids, so
219 /// a log reader sees one warned peer followed by an abort whether the
220 /// quorum counted two distinct peers or misfired on one — which is exactly
221 /// the ambiguity the V2-1109 testnet run hit (0/463 aborts showed a second
222 /// peer, because the second peer was structurally unloggable).
223 pub(crate) fn corroborating_peers(&self) -> Vec<String> {
224 let Ok(guard) = self.inner.lock() else {
225 return Vec::new();
226 };
227 let (peers, _) = &*guard;
228 let mut ids: Vec<String> = peers.iter().map(|p| format!("{p}")).collect();
229 ids.sort();
230 ids
231 }
232}
233
234#[cfg(test)]
235mod settlement_refusal_tests {
236 use super::*;
237
238 fn peer(seed: u8) -> PeerId {
239 PeerId::from_bytes([seed; 32])
240 }
241
242 /// The question the V2-1109 run could not answer from logs: does one peer
243 /// refusing many times count as many? It must not — the quorum is over
244 /// DISTINCT peers, and a lone hostile or misconfigured storer repeating
245 /// itself must never become a verdict about this build.
246 #[test]
247 fn one_peer_refusing_many_times_never_reaches_the_quorum() {
248 let refusals = SettlementRefusals::default();
249 for _ in 0..50 {
250 assert!(
251 refusals.note(peer(7), "run ant update").is_none(),
252 "a single peer's repeated refusals must stay below the quorum"
253 );
254 }
255 assert!(refusals.corroborated().is_none());
256 assert_eq!(refusals.corroborating_peers().len(), 1);
257 }
258
259 /// The terminal log line must be able to name both corroborators, so the
260 /// distinct-peer property is verifiable from the outside.
261 #[test]
262 fn corroborating_peers_names_every_distinct_refuser() {
263 let refusals = SettlementRefusals::default();
264 assert!(refusals.note(peer(1), "run ant update").is_none());
265 assert!(refusals.note(peer(2), "run ant update").is_some());
266
267 let ids = refusals.corroborating_peers();
268 assert_eq!(ids.len(), 2);
269 assert_ne!(ids[0], ids[1]);
270 assert_eq!(ids, {
271 let mut sorted = ids.clone();
272 sorted.sort();
273 sorted
274 });
275 }
276}
277
278/// Classify a `data::error::Error` into a controller `Outcome`.
279///
280/// Capacity signals (Timeout / NetworkError) drive the controller
281/// down; application errors do not. The mapping is conservative:
282/// anything that COULD be transport-related is treated as a network
283/// signal, because under-classifying a real network failure as
284/// "application error" makes the controller blind to genuine stress.
285///
286/// Mapping policy:
287/// - `Timeout` -> `Timeout` (per-op deadline elapsed)
288/// - `Network`, `InsufficientPeers`, `Io` -> `NetworkError` (transport
289/// layer reported failure)
290/// - `Protocol`, `Storage` -> `NetworkError` (these wrap remote errors
291/// that frequently include peer disconnects mid-stream — under
292/// network stress these are how transport failures surface)
293/// - `PartialUpload` -> `NetworkError` (literal capacity signal: some
294/// chunks could not be stored)
295/// - `AlreadyStored`, `Encryption`, `Crypto`, `Payment`,
296/// `Serialization`, `InvalidData`, `NotFound`, `SignatureVerification`,
297/// `Config`, `InsufficientDiskSpace`, `CostEstimationInconclusive`,
298/// `Cancelled` -> `ApplicationError` (would happen on a perfectly
299/// healthy link; `Cancelled` is caller-initiated and must not be retried
300/// as a transport failure; `NotFound` is a definitively absent record
301/// reported over a working link, not a transport symptom)
302/// - `RemotePut` -> `ApplicationError` (the remote node responded with a
303/// structured rejection — the transport succeeded, so the node declined
304/// at the application layer; not a local capacity signal)
305/// - `ClientUpdateRequired` -> `ApplicationError` (the storer refused to quote
306/// a client that settles under superseded rules — a terminal verdict about
307/// this build, not about link capacity, and no retry rate clears it)
308/// - `CloseGroupShortfall` -> `ApplicationError` (a quorum shortfall caused
309/// by close-group dial/relay churn with no PUT-response timeouts — remote
310/// peer churn, not local backpressure; a timeout-bearing shortfall keeps
311/// `InsufficientPeers`/`NetworkError` instead, so genuine congestion still
312/// cuts the cap — V2-554)
313pub(crate) fn classify_error(err: &Error) -> Outcome {
314 match err {
315 Error::Timeout(_) => crate::transfer_policy::FailureKind::Timeout,
316 Error::Network(_)
317 | Error::InsufficientPeers(_)
318 | Error::Io(_)
319 | Error::Protocol(_)
320 | Error::Storage(_)
321 | Error::PartialUpload { .. } => crate::transfer_policy::FailureKind::Network,
322 Error::AlreadyStored
323 | Error::Encryption(_)
324 | Error::Crypto(_)
325 | Error::Payment(_)
326 | Error::Serialization(_)
327 | Error::InvalidData(_)
328 // A definitively absent record, reported over a working link —
329 // the peers answered, there was just nothing stored there. Not a
330 // transport symptom, so it must not push the limiter down.
331 | Error::NotFound(_)
332 | Error::SignatureVerification(_)
333 | Error::Config(_)
334 | Error::InsufficientDiskSpace(_)
335 | Error::CostEstimationInconclusive(_)
336 | Error::Cancelled(_)
337 // The storer parsed our request and refused it on its merits, over a
338 // working link. Sending fewer requests would not help, and treating it
339 // as congestion would quietly shrink the limiter for the rest of the
340 // run on the basis of a fault no retry can clear.
341 | Error::ClientUpdateRequired(_)
342 | Error::StorerUpdateRequired(_)
343 | Error::BadQuoteBinding { .. }
344 | Error::BadQuoteCommitment { .. }
345 // An external-signer merkle batch larger than one tree can hold —
346 // a caller-shape refusal raised before any network work, so it says
347 // nothing about link capacity.
348 | Error::MerkleBatchTooLarge { .. }
349 // A remote node responded with a structured rejection — the
350 // transport round-trip succeeded, so the node declined at the
351 // application layer (payment/disk/quote/pool). Not a local
352 // capacity signal; recorded but must not push the limiter down.
353 | Error::RemotePut { .. }
354 // A close-group PUT shortfall caused purely by dial/relay churn
355 // (dead/stale relayed peer addresses), with no PUT-response
356 // timeouts to signal local backpressure. Remote peer churn, not
357 // "client sending too fast" — must not push the limiter down
358 // (V2-554). A shortfall that DID time out keeps `InsufficientPeers`
359 // (`NetworkError`) so real congestion still cuts the cap.
360 | Error::CloseGroupShortfall(_) => crate::transfer_policy::FailureKind::Application,
361 }
362 .outcome()
363}
364
365/// Compute XOR distance between a peer's ID bytes and a target address.
366///
367/// Uses the first 32 bytes of the peer ID (or fewer if shorter) XORed
368/// with the target address. The returned byte array sorts
369/// lexicographically from closest to furthest.
370pub(crate) fn peer_xor_distance(peer_id: &PeerId, target: &[u8; 32]) -> [u8; 32] {
371 let peer_bytes = peer_id.as_bytes();
372 let mut distance = [0u8; 32];
373 for (i, d) in distance.iter_mut().enumerate() {
374 let peer_byte = peer_bytes.get(i).copied().unwrap_or(0);
375 *d = peer_byte ^ target[i];
376 }
377 distance
378}
379
380/// Default timeout for lightweight network operations (quotes, DHT lookups) in seconds.
381const DEFAULT_QUOTE_TIMEOUT_SECS: u64 = 10;
382
383/// Default timeout for the per-peer chunk GET response and any other
384/// caller that explicitly reads `store_timeout_secs`, in seconds.
385///
386/// Note despite the name: this knob does **not** govern the non-merkle
387/// chunk PUT response timeout — that path uses the
388/// `STORE_RESPONSE_TIMEOUT` constant in `chunk.rs` directly. Nor does
389/// it govern the merkle batch PUT timeout — see
390/// `DEFAULT_MERKLE_STORE_TIMEOUT_SECS`.
391///
392/// 10 s matches the pre-existing `main` default and intentionally
393/// excludes residential-upload tuning, which is Mick's PR #78
394/// territory (splitting GET into its own field).
395const DEFAULT_STORE_TIMEOUT_SECS: u64 = 10;
396
397/// Default timeout for **merkle batch** chunk store operations in seconds.
398///
399/// Separate from `DEFAULT_STORE_TIMEOUT_SECS` because merkle PUTs carry
400/// an extra storer-side cost: the payment verifier runs an iterative
401/// DHT lookup (`CLOSENESS_LOOKUP_TIMEOUT` in `ant-node`, **240 s**
402/// post-PR #89) before accepting the proof.
403///
404/// This timeout MUST be >= the storer-side `CLOSENESS_LOOKUP_TIMEOUT`
405/// plus padding for the store-response round-trip and storer-local
406/// I/O. Otherwise the client gives up while the storer is still
407/// happily verifying, the storer wastes CPU/bandwidth on a chunk the
408/// client has already discarded, and the client re-targets a
409/// different close-K member — potentially double-storing the same
410/// chunk and polluting routing.
411///
412/// 270 s = 240 s (storer lookup) + 30 s padding (network RTT + LMDB
413/// put + fsync + clock skew tolerance).
414///
415/// This invariant must be re-validated if either side's timeout
416/// changes. Empirically surfaced as "every cross-region merkle chunk
417/// times out at 10 s" on a 210-node 7-region testnet run on
418/// 2026-05-12; bumping to 270 s flipped that 0/22 -> 9/9 pass rate.
419const DEFAULT_MERKLE_STORE_TIMEOUT_SECS: u64 = 270;
420
421/// Default timeout for chunk GET response operations in seconds.
422const DEFAULT_CHUNK_GET_TIMEOUT_SECS: u64 = 10;
423
424/// Default quote concurrency: high because quoting is pure network I/O
425/// (DHT lookups + small request/response messages) with no CPU-bound work.
426const DEFAULT_QUOTE_CONCURRENCY: usize = 32;
427
428/// Default store concurrency: moderate because each chunk PUT sends ~4MB
429/// to 7 close-group peers. At 8 concurrent stores, ~225MB of outbound
430/// traffic can be in flight. Users on fast connections can increase this
431/// with --store-concurrency; users on slow connections can decrease it.
432const DEFAULT_STORE_CONCURRENCY: usize = 8;
433
434/// Configuration for the Autonomi client.
435#[derive(Debug, Clone)]
436pub struct ClientConfig {
437 /// Per-op timeout for lightweight network operations (quotes,
438 /// DHT lookups), in seconds. The adaptive controller does NOT
439 /// currently size timeouts; this remains a static knob.
440 pub quote_timeout_secs: u64,
441 /// Per-op timeout, in seconds, for the chunk GET response path
442 /// (`chunk_get_from_peer`) and any other caller that reads this
443 /// field directly.
444 ///
445 /// Note despite the historical name `store_timeout_secs`: this
446 /// knob does **not** govern the non-merkle chunk PUT response
447 /// timeout (that path uses the `STORE_RESPONSE_TIMEOUT` constant
448 /// in `chunk.rs`) and does **not** govern the merkle batch PUT
449 /// timeout (see `merkle_store_timeout_secs`). Rename pending in
450 /// Mick's PR #78 which adds a dedicated `chunk_get_timeout_secs`.
451 ///
452 /// The adaptive controller does NOT currently size timeouts;
453 /// this remains a static knob.
454 pub store_timeout_secs: u64,
455 /// Per-op timeout for **merkle batch** chunk store (PUT)
456 /// operations, in seconds. Separate from `store_timeout_secs`
457 /// because merkle PUTs incur the storer-side
458 /// `CLOSENESS_LOOKUP_TIMEOUT` (240 s post-PR #89) on top of the
459 /// usual store path; the client must wait at least that long
460 /// plus padding, or the storer wastes work on a chunk the client
461 /// has already given up on. Default 270 s.
462 pub merkle_store_timeout_secs: u64,
463 /// Per-peer response timeout for chunk GET operations, in seconds.
464 /// This is intentionally independent from `store_timeout_secs`: PUTs
465 /// and GETs have different payload direction and performance profiles.
466 pub chunk_get_timeout_secs: u64,
467 /// Number of closest peers to consider for routing.
468 pub close_group_size: usize,
469 /// **Deprecated.** Pre-adaptive ceiling for quote concurrency.
470 ///
471 /// The adaptive controller now sizes quote fan-out from observed
472 /// signals. This field, when non-zero and smaller than the
473 /// controller's per-channel default, clamps the **quote channel
474 /// only** (it does NOT bleed into store or fetch). Removed in a
475 /// future release.
476 pub quote_concurrency: usize,
477 /// **Deprecated.** Pre-adaptive ceiling for store concurrency.
478 ///
479 /// The adaptive controller now sizes store fan-out from observed
480 /// signals. This field, when non-zero and smaller than the
481 /// controller's per-channel default, clamps the **store channel
482 /// only** (it does NOT bleed into quote or fetch). Removed in a
483 /// future release.
484 pub store_concurrency: usize,
485 /// Adaptive controller configuration. Defaults are tuned to match
486 /// or exceed the prior static behavior — disabling adaptation
487 /// (`adaptive.enabled = false`) reverts to the controller's
488 /// `initial` values without re-evaluation.
489 pub adaptive: AdaptiveConfig,
490 /// Allow loopback (`127.0.0.1`) connections in the saorsa-transport
491 /// layer. Set to `true` only for devnet / local testing. Production
492 /// peers on the public Autonomi network reject the QUIC handshake
493 /// variant produced when this is `true`, so the default is `false`.
494 ///
495 /// This mirrors the `--allow-loopback` flag in `ant-cli`, which already
496 /// defaults to `false` and threads through to the same
497 /// `CoreNodeConfig::builder().local(...)` call.
498 pub allow_loopback: bool,
499 /// Bind a dual-stack IPv6 socket (`true`) or an IPv4-only socket
500 /// (`false`). Defaults to `true`, matching the CLI default.
501 ///
502 /// Set to `false` only when running on hosts without a working IPv6
503 /// stack, to avoid advertising unreachable v6 addresses to the DHT
504 /// (which causes slow connects and junk DHT address records). This
505 /// mirrors the `--ipv4-only` flag in `ant-cli`.
506 pub ipv6: bool,
507 /// Per-batch leaf cap for **external-signer** merkle preparation,
508 /// clamped to `3..=MAX_LEAVES` when set (a cap of 2 cannot partition odd
509 /// totals — parts of 3 and 2 compose any count, so 3 is the smallest
510 /// safe cap). `None` (the default) uses the contract maximum
511 /// (`MAX_LEAVES` = 256).
512 ///
513 /// This is a test seam (ADR-0003): a small cap makes
514 /// `file_prepare_upload_with_mode` produce a genuine multi-batch
515 /// prepared upload from a kilobyte file, so the N-signature external
516 /// flow is exercisable in E2E without a multi-GiB fixture. Production
517 /// callers should leave it `None` — a lower cap only means more payment
518 /// transactions for the same chunks.
519 pub merkle_external_batch_cap: Option<usize>,
520}
521
522impl Default for ClientConfig {
523 fn default() -> Self {
524 Self {
525 quote_timeout_secs: DEFAULT_QUOTE_TIMEOUT_SECS,
526 store_timeout_secs: DEFAULT_STORE_TIMEOUT_SECS,
527 merkle_store_timeout_secs: DEFAULT_MERKLE_STORE_TIMEOUT_SECS,
528 chunk_get_timeout_secs: DEFAULT_CHUNK_GET_TIMEOUT_SECS,
529 close_group_size: CLOSE_GROUP_SIZE,
530 quote_concurrency: DEFAULT_QUOTE_CONCURRENCY,
531 store_concurrency: DEFAULT_STORE_CONCURRENCY,
532 adaptive: AdaptiveConfig::default(),
533 allow_loopback: false,
534 ipv6: true,
535 merkle_external_batch_cap: None,
536 }
537 }
538}
539
540/// Build the adaptive controller for a `Client`. Loads any persisted
541/// snapshot, clamps cold-start values into the deprecated-flag bounds
542/// **per channel** (so a pin on `--store-concurrency` does NOT bleed
543/// into the fetch / quote channels), and returns the persistence path
544/// so callers can save back at shutdown.
545fn build_controller(config: &ClientConfig) -> (AdaptiveController, Option<PathBuf>) {
546 let mut adaptive_cfg = config.adaptive.clone();
547
548 // Per-channel ceilings: each legacy field is interpreted as a cap
549 // for ONLY its matching channel. The fetch channel has no
550 // pre-existing legacy field; it always uses the controller's
551 // default ceiling.
552 //
553 // The legacy fields are non-zero by ClientConfig::default(), but
554 // we honor them as bounds only when they would actually CONSTRAIN
555 // the controller — i.e. when smaller than the per-channel default
556 // max. A default ClientConfig must not silently lower the
557 // controller's ceilings.
558 // A value equal to the historic legacy default is treated as
559 // "not pinned by the user" — without this, every default
560 // ClientConfig would silently lower the controller's per-channel
561 // ceilings to the prior static values (32/8) and the controller
562 // could never grow above them.
563 let user_quote_max = config.quote_concurrency;
564 let user_store_max = config.store_concurrency;
565 let quote_pinned = user_quote_max > 0 && user_quote_max != DEFAULT_QUOTE_CONCURRENCY;
566 let store_pinned = user_store_max > 0 && user_store_max != DEFAULT_STORE_CONCURRENCY;
567 if quote_pinned && user_quote_max < adaptive_cfg.max.quote {
568 adaptive_cfg.max.quote = user_quote_max;
569 }
570 if store_pinned && user_store_max < adaptive_cfg.max.store {
571 adaptive_cfg.max.store = user_store_max;
572 }
573
574 // Cold-start values: matched to the prior static defaults. If the
575 // legacy field caps the channel below the cold-start, lower the
576 // start to match — never start above the channel's max.
577 let mut start = ChannelStart::default();
578 start.quote = start.quote.min(adaptive_cfg.max.quote);
579 start.store = start.store.min(adaptive_cfg.max.store);
580 start.fetch = start.fetch.min(adaptive_cfg.max.fetch);
581
582 #[cfg(feature = "native")]
583 let adaptive_enabled = adaptive_cfg.enabled;
584 let controller = AdaptiveController::new(start, adaptive_cfg);
585 // Skip disk warm-start entirely when adaptation is disabled —
586 // fixed-concurrency mode means the user wants exactly the cold
587 // start, no surprises from prior runs. (warm_start is also a
588 // no-op when disabled, but skipping the load avoids file I/O
589 // and the path-resolution side effects.)
590 #[cfg(feature = "native")]
591 let persist_path = if adaptive_enabled {
592 let p = adaptive::default_persist_path();
593 if let Some(ref path) = p {
594 if let Some(snap) = adaptive::load_snapshot(path) {
595 debug!(path = %path.display(), "adaptive: warm-start from disk");
596 controller.warm_start(snap);
597 }
598 }
599 p
600 } else {
601 // Even with adaptation off, persist_path is computed so
602 // explicit save_adaptive_snapshot() calls still work — but
603 // the controller currently never moves, so saving the cold
604 // start is harmless.
605 adaptive::default_persist_path()
606 };
607
608 // File downloads choose a stream-decrypt batch size per download
609 // from the current fetch cap and usable RAM, then pass it into
610 // self_encryption's runtime batch-size API. The adaptive controller
611 // still drives fan-out inside each batch by re-reading
612 // `controller.fetch.current()` in the decrypt callback.
613
614 #[cfg(not(feature = "native"))]
615 let persist_path = None;
616 (controller, persist_path)
617}
618
619/// Client for the Autonomi decentralized network.
620///
621/// Provides high-level APIs for storing and retrieving chunks
622/// and files on the network.
623pub struct Client {
624 config: ClientConfig,
625 network: Network,
626 wallet: Option<Arc<Wallet>>,
627 evm_network: Option<ant_protocol::evm::Network>,
628 chunk_cache: ChunkCache,
629 next_request_id: AtomicU64,
630 /// Adaptive concurrency controller: replaces the static
631 /// quote/store concurrency knobs. See `adaptive` module.
632 controller: AdaptiveController,
633 /// Path the controller persists its snapshot to. `None` disables
634 /// persistence (useful for tests / non-disk environments).
635 #[cfg(feature = "native")]
636 persist_path: Option<PathBuf>,
637 /// Path for the persistent client peer cache. `None` disables the cache.
638 #[cfg(feature = "native")]
639 peer_cache_path: Option<PathBuf>,
640 /// Peers that did not answer a settlement-versioned quote request, and are
641 /// therefore asked in the legacy shape from now on.
642 ///
643 /// Without this the probe cost is paid on **every** request rather than
644 /// roughly once per peer. Measured on the merkle E2E suite against a fleet
645 /// that predates the versioned requests, re-probing took the run from ~24
646 /// minutes to over 60, because each of the sixteen candidates per pool sat
647 /// out a full `quote_timeout_secs` before the fallback.
648 ///
649 /// Roughly, not exactly: concurrent first contacts are not coalesced, so
650 /// several in-flight requests can all miss the cache for the same peer and
651 /// each probe it once before any of them records the answer. Observed at
652 /// about two probes per peer on a 35-node devnet. Single-flighting them
653 /// would remove the duplicates but not the wall-clock cost, which is set
654 /// by how many *sequential* quote rounds an upload performs rather than by
655 /// how many probes each round contains.
656 ///
657 /// Process-local and never persisted. A peer that upgrades mid-run keeps
658 /// being asked in the legacy shape until the next start, which is
659 /// acceptable while the legacy shape still gets a quote, and stops
660 /// mattering when the fallback is deleted (see
661 /// [`UNVERSIONED_RETRY_REQUIRES_MIN_V1`]).
662 ///
663 /// Entries are only ever added for a peer that has **never** answered a
664 /// versioned request. Without that condition a single lost response would
665 /// pin an upgraded peer to the legacy shape for the rest of the session,
666 /// turning one dropped packet into a standing downgrade; with it, a peer
667 /// that has shown it understands the versioned shape can never be demoted.
668 ///
669 /// A peer that has never answered can still get itself asked without a
670 /// version by staying silent, exactly as it could through the fallback
671 /// alone. Remembering the answer makes that cheaper to sustain, so it is
672 /// not a new capability but it is a wider one, and the compile-time guard
673 /// requires the whole path to be gone before any refusal is possible.
674 unversioned_quote_peers: Arc<Mutex<HashSet<PeerId>>>,
675 /// Peers observed answering a settlement-versioned request. Never
676 /// downgraded, however they behave later.
677 versioned_capable_peers: Arc<Mutex<HashSet<PeerId>>>,
678 /// Distinct peers that have refused this client on settlement-version
679 /// grounds, and the wording of the first such refusal.
680 ///
681 /// Client-wide and sticky, for two reasons that pull in opposite
682 /// directions and are both real.
683 ///
684 /// It must outlive one operation, because the verdict is about this
685 /// **build**, not this upload. Held in a single collector's local state, a
686 /// refusal observed by one in-flight upload says nothing to another that
687 /// is about to submit a payment, and merkle payments cannot be undone.
688 ///
689 /// It must not fire on one peer's say-so, because nothing authenticates a
690 /// refusal. A single hostile or confused peer answering
691 /// `ClientUpdateRequired` to every query would otherwise abort every
692 /// upload the client attempts, converting an over-query design that
693 /// tolerates many bad peers into one that tolerates none. So a refusal
694 /// becomes terminal only once [`SETTLEMENT_REFUSAL_QUORUM`] distinct peers
695 /// agree, which a genuine incompatibility reaches immediately (every
696 /// upgraded peer refuses) and a lone attacker cannot reach at all.
697 settlement_refusals: SettlementRefusals,
698}
699
700impl Client {
701 /// Create a client using a platform network adapter.
702 #[must_use]
703 pub fn from_network(network: Network, config: ClientConfig) -> Self {
704 let (controller, _persist_path) = build_controller(&config);
705 Self {
706 config,
707 network,
708 wallet: None,
709 evm_network: None,
710 chunk_cache: ChunkCache::default(),
711 next_request_id: AtomicU64::new(1),
712 controller,
713 #[cfg(feature = "native")]
714 persist_path: _persist_path,
715 #[cfg(feature = "native")]
716 peer_cache_path: None,
717 unversioned_quote_peers: Arc::new(Mutex::new(HashSet::new())),
718 versioned_capable_peers: Arc::new(Mutex::new(HashSet::new())),
719 settlement_refusals: SettlementRefusals::default(),
720 }
721 }
722
723 /// Keep compatibility decisions across browser operations whose network adapter
724 /// carries a per-operation payment network identity.
725 #[cfg(not(feature = "native"))]
726 pub(crate) fn with_shared_quote_state(mut self, session: &Self) -> Self {
727 self.unversioned_quote_peers = Arc::clone(&session.unversioned_quote_peers);
728 self.versioned_capable_peers = Arc::clone(&session.versioned_capable_peers);
729 self.settlement_refusals = session.settlement_refusals.clone();
730 self
731 }
732
733 /// Create a client connected to the given P2P node.
734 #[must_use]
735 #[cfg(feature = "native")]
736 pub fn from_node(node: Arc<P2PNode>, config: ClientConfig) -> Self {
737 Self::from_node_with_peer_cache(node, config, None)
738 }
739
740 /// Create a client connected to the given P2P node and attach an optional
741 /// persistent peer cache path.
742 #[must_use]
743 #[cfg(feature = "native")]
744 pub fn from_node_with_peer_cache(
745 node: Arc<P2PNode>,
746 config: ClientConfig,
747 #[cfg(feature = "native")] peer_cache_path: Option<PathBuf>,
748 ) -> Self {
749 let network = Network::from_node(node);
750 let (controller, persist_path) = build_controller(&config);
751 Self {
752 config,
753 network,
754 wallet: None,
755 evm_network: None,
756 chunk_cache: ChunkCache::default(),
757 next_request_id: AtomicU64::new(1),
758 unversioned_quote_peers: Arc::new(Mutex::new(HashSet::new())),
759 versioned_capable_peers: Arc::new(Mutex::new(HashSet::new())),
760 settlement_refusals: SettlementRefusals::default(),
761 controller,
762 persist_path,
763 peer_cache_path,
764 }
765 }
766
767 /// Create a client connected to bootstrap peers.
768 ///
769 /// Threads `config.allow_loopback` and `config.ipv6` through to
770 /// `Network::new`, which controls the saorsa-transport `local` and
771 /// `ipv6` flags on the underlying `CoreNodeConfig`. See
772 /// `ClientConfig::allow_loopback` and `ClientConfig::ipv6` for details.
773 ///
774 /// # Errors
775 ///
776 /// Returns an error if the P2P node cannot be created or bootstrapping fails.
777 #[cfg(feature = "native")]
778 pub async fn connect(
779 bootstrap_peers: &[std::net::SocketAddr],
780 config: ClientConfig,
781 ) -> Result<Self> {
782 let seeds: Vec<_> = bootstrap_peers
783 .iter()
784 .copied()
785 .map(ant_protocol::transport::MultiAddr::quic)
786 .collect();
787 Self::connect_multiaddrs(&seeds, config).await
788 }
789
790 /// Connect using native QUIC multiaddresses without discarding peer pins.
791 #[cfg(feature = "native")]
792 pub async fn connect_multiaddrs(
793 bootstrap_peers: &[ant_protocol::transport::MultiAddr],
794 config: ClientConfig,
795 ) -> Result<Self> {
796 debug!(
797 "Connecting to Autonomi network with {} bootstrap peers (allow_loopback={}, ipv6={})",
798 bootstrap_peers.len(),
799 config.allow_loopback,
800 config.ipv6,
801 );
802 let network =
803 Network::new_multiaddrs(bootstrap_peers, config.allow_loopback, config.ipv6).await?;
804 let (controller, persist_path) = build_controller(&config);
805 Ok(Self {
806 config,
807 network,
808 wallet: None,
809 evm_network: None,
810 chunk_cache: ChunkCache::default(),
811 next_request_id: AtomicU64::new(1),
812 unversioned_quote_peers: Arc::new(Mutex::new(HashSet::new())),
813 versioned_capable_peers: Arc::new(Mutex::new(HashSet::new())),
814 settlement_refusals: SettlementRefusals::default(),
815 controller,
816 persist_path,
817 peer_cache_path: None,
818 })
819 }
820
821 /// Set the wallet for payment operations.
822 ///
823 /// Also populates the EVM network from the wallet so that
824 /// token approvals work without a separate `with_evm_network` call.
825 #[must_use]
826 pub fn with_wallet(mut self, wallet: Wallet) -> Self {
827 self.evm_network = Some(wallet.network().clone());
828 self.wallet = Some(Arc::new(wallet));
829 self
830 }
831
832 /// Set the EVM network without requiring a wallet.
833 ///
834 /// This enables token approval and contract interactions
835 /// for external-signer flows where the private key lives outside Rust.
836 #[must_use]
837 pub fn with_evm_network(mut self, network: ant_protocol::evm::Network) -> Self {
838 self.evm_network = Some(network);
839 self
840 }
841
842 /// Get the EVM network, falling back to the wallet's network if available.
843 ///
844 /// # Errors
845 ///
846 /// Returns an error if neither `with_evm_network` nor `with_wallet` was called.
847 pub(crate) fn require_evm_network(&self) -> Result<&ant_protocol::evm::Network> {
848 if let Some(ref net) = self.evm_network {
849 return Ok(net);
850 }
851 if let Some(ref wallet) = self.wallet {
852 return Ok(wallet.network());
853 }
854 Err(Error::Payment(
855 "EVM network not configured — call with_evm_network() or with_wallet() first"
856 .to_string(),
857 ))
858 }
859
860 /// Get the client configuration.
861 #[must_use]
862 pub fn config(&self) -> &ClientConfig {
863 &self.config
864 }
865
866 /// Get a mutable reference to the client configuration.
867 pub fn config_mut(&mut self) -> &mut ClientConfig {
868 &mut self.config
869 }
870
871 /// Get a reference to the network layer.
872 #[must_use]
873 pub fn network(&self) -> &Network {
874 &self.network
875 }
876
877 /// Compute the live network-participation snapshot.
878 ///
879 /// Convenience pass-through to [`Network::health`] — the single
880 /// write-readiness implementation shared by all embedded-client
881 /// consumers (antd, ant-gui, ant-ffi, ant-tui).
882 #[cfg(feature = "native")]
883 pub async fn network_health(&self) -> NetworkHealth {
884 self.network.health().await
885 }
886
887 /// Get the wallet, if configured.
888 #[must_use]
889 pub fn wallet(&self) -> Option<&Arc<Wallet>> {
890 self.wallet.as_ref()
891 }
892
893 /// Set the in-memory cache budget for this client.
894 #[must_use]
895 pub fn with_chunk_cache(mut self, cache: ChunkCache) -> Self {
896 self.chunk_cache = cache;
897 self
898 }
899
900 /// Get a reference to the chunk cache.
901 #[must_use]
902 pub fn chunk_cache(&self) -> &ChunkCache {
903 &self.chunk_cache
904 }
905
906 /// Adaptive concurrency controller. Hot loops read
907 /// `controller().<channel>.current()` to size their fan-out and
908 /// call `.observe(...)` on each completion.
909 #[must_use]
910 pub fn controller(&self) -> &AdaptiveController {
911 &self.controller
912 }
913
914 /// Persist the current adaptive snapshot to disk so the next
915 /// `Client::connect` warm-starts at the learned values instead of
916 /// cold defaults. Best effort — failures log and are discarded.
917 /// Idempotent. Safe to call from a Drop impl or an explicit
918 /// shutdown hook.
919 #[cfg(feature = "native")]
920 pub fn save_adaptive_snapshot(&self) {
921 if let Some(ref path) = self.persist_path {
922 adaptive::save_snapshot(path, self.controller.snapshot());
923 }
924 }
925
926 /// Persist currently connected peers that have Direct-tagged addresses in
927 /// the DHT. Best effort; failures are logged and do not affect the client
928 /// operation that just completed.
929 #[cfg(feature = "native")]
930 pub async fn save_peer_cache(&self) {
931 if let Some(ref path) = self.peer_cache_path {
932 let node = self.network().node();
933 peer_cache::promote_connected_direct_peers(node.as_ref(), path, node.dht().k_value())
934 .await;
935 }
936 }
937
938 /// Get the next request ID for protocol messages.
939 pub(crate) fn next_request_id(&self) -> u64 {
940 self.next_request_id.fetch_add(1, Ordering::Relaxed)
941 }
942
943 /// Handle to the set of peers that cannot answer a settlement-versioned
944 /// quote request, shared with the per-peer request futures on both quote
945 /// paths.
946 ///
947 /// Callers read it before choosing a request shape and insert into it when
948 /// a peer stays silent. A poisoned lock is treated as "nothing known", so
949 /// the worst case is a wasted probe rather than a silently skipped version
950 /// declaration.
951 pub(crate) fn unversioned_quote_peers(&self) -> Arc<Mutex<HashSet<PeerId>>> {
952 Arc::clone(&self.unversioned_quote_peers)
953 }
954
955 /// Handle to the set of peers already seen answering a versioned request.
956 /// Consulted before demoting a peer, so a lost response cannot strand an
957 /// upgraded peer in the legacy shape.
958 pub(crate) fn versioned_quote_capable_handle(&self) -> Arc<Mutex<HashSet<PeerId>>> {
959 Arc::clone(&self.versioned_capable_peers)
960 }
961
962 /// Record that `peer_id` refused this client's settlement version, and
963 /// report whether enough distinct peers now agree for it to be believed.
964 ///
965 /// Returns the refusal wording once [`SETTLEMENT_REFUSAL_QUORUM`] is met,
966 /// and `None` below it, so a lone peer is treated as a peer fault rather
967 /// than a verdict about this build.
968 pub(crate) fn note_settlement_refusal(&self, peer_id: PeerId, message: &str) -> Option<String> {
969 self.settlement_refusals.note(peer_id, message)
970 }
971
972 /// The corroborated refusal, if this client has already been told by
973 /// enough peers that it cannot settle.
974 ///
975 /// Checked before spending money. The verdict concerns this build rather
976 /// than any one upload, so an upload that starts after another has already
977 /// established it must not proceed to pay.
978 pub(crate) fn corroborated_settlement_refusal(&self) -> Option<String> {
979 self.settlement_refusals.corroborated()
980 }
981
982 /// Handle to the shared refusal tracker, for collectors that run outside
983 /// `&self`.
984 pub(crate) fn settlement_refusals(&self) -> SettlementRefusals {
985 self.settlement_refusals.clone()
986 }
987
988 /// Return the chunk PUT-target set: the closest [`PUT_TARGET_WIDTH`] peers
989 /// to the address, each paired with its known network addresses.
990 ///
991 /// Used by the merkle store path, which — unlike single-node payment — has
992 /// no witnessed put-target list to forward, so it fetches the closest-K
993 /// neighbourhood locally.
994 pub(crate) async fn put_target_peers(
995 &self,
996 target: &XorName,
997 ) -> Result<Vec<(PeerId, Vec<MultiAddr>)>> {
998 self.closest_peers(target, PUT_TARGET_WIDTH).await
999 }
1000
1001 /// Return the requested number of closest peers for a target address.
1002 ///
1003 /// Queries the DHT for peers by XOR distance. Returns each peer
1004 /// paired with its known network addresses.
1005 pub(crate) async fn closest_peers(
1006 &self,
1007 target: &XorName,
1008 count: usize,
1009 ) -> Result<Vec<(PeerId, Vec<MultiAddr>)>> {
1010 let peers = self.network().find_closest_peers(target, count).await?;
1011
1012 if peers.is_empty() {
1013 return Err(Error::InsufficientPeers(
1014 "DHT returned no peers for target address".to_string(),
1015 ));
1016 }
1017 Ok(peers)
1018 }
1019}
1020
1021/// Persist the adaptive snapshot when the `Client` is dropped, so any
1022/// caller — CLI, daemon, library user, integration test — gets
1023/// warm-start carry-over for free without remembering to call
1024/// `save_adaptive_snapshot()` explicitly. Best effort, sync `std::fs`,
1025/// no panic risk on a poisoned mutex (the inner helper handles it).
1026///
1027/// We deliberately write SYNCHRONOUSLY (not via `spawn_blocking`)
1028/// because Drop runs during process shutdown / runtime teardown,
1029/// when fire-and-forget background tasks can be dropped before they
1030/// complete and the snapshot is silently lost. A small synchronous
1031/// stall on a tokio worker (typically <1ms for a local-disk JSON
1032/// write of ~50 bytes) is the right tradeoff for guaranteed
1033/// persistence — BOUNDED by `DROP_SAVE_TIMEOUT` so a stalled
1034/// network-mounted data dir cannot block process shutdown.
1035#[cfg(feature = "native")]
1036const DROP_SAVE_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(500);
1037
1038#[cfg(feature = "native")]
1039impl Drop for Client {
1040 fn drop(&mut self) {
1041 let Some(path) = self.persist_path.clone() else {
1042 return;
1043 };
1044 let snap = self.controller.snapshot();
1045 adaptive::save_snapshot_with_timeout(path, snap, DROP_SAVE_TIMEOUT);
1046 }
1047}
1048
1049#[cfg(test)]
1050#[allow(clippy::unwrap_used)]
1051mod tests {
1052 use super::*;
1053
1054 /// Cover EVERY variant of `data::error::Error`. Build an instance of
1055 /// each, classify it, and assert the resulting `Outcome` matches the
1056 /// only sensible mapping. If a future commit adds a new error variant
1057 /// without updating `classify_error`, this test fails to ensure the
1058 /// adaptive controller always sees correct capacity signals.
1059 ///
1060 /// Mapping policy (mirrors `classify_error` doc):
1061 /// - `Timeout` -> `Outcome::Timeout`
1062 /// - `Network`, `InsufficientPeers`, `Io`, `Protocol`, `Storage`,
1063 /// `PartialUpload` -> `Outcome::NetworkError` (transport-related
1064 /// or literal capacity failure)
1065 /// - everything else -> `Outcome::ApplicationError` (would happen
1066 /// on a perfectly healthy network)
1067 #[test]
1068 fn classify_error_covers_all_variants() {
1069 let cases: Vec<(Error, Outcome)> = vec![
1070 (Error::Timeout("t".to_string()), Outcome::Timeout),
1071 (Error::Network("n".to_string()), Outcome::NetworkError),
1072 (
1073 Error::InsufficientPeers("p".to_string()),
1074 Outcome::NetworkError,
1075 ),
1076 (Error::Storage("s".to_string()), Outcome::NetworkError),
1077 (Error::Payment("p".to_string()), Outcome::ApplicationError),
1078 (Error::Protocol("p".to_string()), Outcome::NetworkError),
1079 (
1080 Error::InvalidData("d".to_string()),
1081 Outcome::ApplicationError,
1082 ),
1083 // A definitively absent record over a working link — the peers
1084 // answered, nothing was stored there. Must NOT register as a
1085 // capacity signal.
1086 (
1087 Error::NotFound("missing".to_string()),
1088 Outcome::ApplicationError,
1089 ),
1090 (
1091 Error::Serialization("s".to_string()),
1092 Outcome::ApplicationError,
1093 ),
1094 (Error::Crypto("c".to_string()), Outcome::ApplicationError),
1095 (
1096 Error::Io(std::io::Error::other("io")),
1097 Outcome::NetworkError,
1098 ),
1099 (Error::Config("c".to_string()), Outcome::ApplicationError),
1100 (
1101 Error::SignatureVerification("s".to_string()),
1102 Outcome::ApplicationError,
1103 ),
1104 (
1105 Error::Encryption("e".to_string()),
1106 Outcome::ApplicationError,
1107 ),
1108 (Error::AlreadyStored, Outcome::ApplicationError),
1109 (
1110 Error::InsufficientDiskSpace("d".to_string()),
1111 Outcome::ApplicationError,
1112 ),
1113 (
1114 Error::CostEstimationInconclusive("c".to_string()),
1115 Outcome::ApplicationError,
1116 ),
1117 (
1118 Error::PartialUpload {
1119 stored: vec![],
1120 stored_count: 0,
1121 failed: vec![],
1122 failed_count: 0,
1123 total_chunks: 0,
1124 spend: Box::new(crate::data::error::PartialUploadSpend {
1125 storage_cost_atto: "0".to_string(),
1126 gas_cost_wei: 0,
1127 }),
1128 reason: "r".to_string(),
1129 },
1130 Outcome::NetworkError,
1131 ),
1132 (
1133 Error::BadQuoteBinding {
1134 peer_id: "peer".to_string(),
1135 detail: "mismatch".to_string(),
1136 },
1137 Outcome::ApplicationError,
1138 ),
1139 // A remote application rejection: the node responded with a
1140 // structured `ProtocolError`, so the transport succeeded and
1141 // this must NOT register as a capacity signal (V2-468).
1142 (
1143 Error::RemotePut {
1144 address: "abcd".to_string(),
1145 source: ant_protocol::ProtocolError::PaymentFailed("stale quote".to_string()),
1146 },
1147 Outcome::ApplicationError,
1148 ),
1149 // A close-group quorum shortfall caused by dial/relay churn with
1150 // no PUT-response timeouts — remote peer churn, not local
1151 // backpressure, so it must NOT register as a capacity signal
1152 // (V2-554). A timeout-bearing shortfall keeps `InsufficientPeers`.
1153 (
1154 Error::CloseGroupShortfall("Stored on 3 peers, need 4".to_string()),
1155 Outcome::ApplicationError,
1156 ),
1157 // Refusing an oversized external-signer merkle batch happens
1158 // before any network work, so it is not a capacity signal.
1159 (
1160 Error::MerkleBatchTooLarge {
1161 addresses: 257,
1162 max_leaves: 256,
1163 },
1164 Outcome::ApplicationError,
1165 ),
1166 ];
1167 for (err, expected) in &cases {
1168 let got = classify_error(err);
1169 assert_eq!(
1170 got, *expected,
1171 "classify_error({err:?}) = {got:?}, expected {expected:?}",
1172 );
1173 }
1174 }
1175
1176 /// C4 fix guard: pinning the legacy `quote_concurrency` /
1177 /// `store_concurrency` ClientConfig fields must clamp ONLY the
1178 /// matching channel's max in the resulting controller. The fetch
1179 /// (download) channel must keep its full default ceiling.
1180 #[test]
1181 fn legacy_concurrency_pin_does_not_bleed_across_channels() {
1182 let cfg = ClientConfig {
1183 quote_concurrency: 4,
1184 store_concurrency: 2,
1185 ..ClientConfig::default()
1186 };
1187 let (controller, _) = build_controller(&cfg);
1188 // The store/quote caps must be clamped to the user's pin.
1189 assert_eq!(controller.config.max.quote, 4, "quote pin not respected");
1190 assert_eq!(controller.config.max.store, 2, "store pin not respected");
1191 // The fetch cap must NOT have been lowered — that's the
1192 // regression C4 was about.
1193 let default_fetch_max = adaptive::ChannelMax::default().fetch;
1194 assert_eq!(
1195 controller.config.max.fetch, default_fetch_max,
1196 "fetch cap was lowered by store/quote pin (C4 regression)"
1197 );
1198 // Cold-start values must respect the lowered ceilings.
1199 assert!(
1200 controller.quote.current() <= 4,
1201 "quote start exceeds its cap"
1202 );
1203 assert!(
1204 controller.store.current() <= 2,
1205 "store start exceeds its cap"
1206 );
1207 }
1208
1209 /// Default ClientConfig must NOT silently lower the controller's
1210 /// per-channel ceilings — the adaptive defaults give every channel
1211 /// real headroom to grow. This guards against future commits
1212 /// re-introducing a global clamp.
1213 #[test]
1214 fn default_client_config_does_not_clamp_controller_max() {
1215 let cfg = ClientConfig::default();
1216 let (controller, _) = build_controller(&cfg);
1217 let defaults = adaptive::ChannelMax::default();
1218 // The legacy fields default to 32/8 (the prior static knobs),
1219 // both of which are <= the per-channel adaptive defaults
1220 // (128/64). build_controller must keep the larger, not clobber
1221 // with the legacy values.
1222 assert_eq!(controller.config.max.quote, defaults.quote);
1223 assert_eq!(controller.config.max.store, defaults.store);
1224 assert_eq!(controller.config.max.fetch, defaults.fetch);
1225 // Compile-time-ish guard: if a new variant is added to Error,
1226 // this match forces an update here.
1227 let _ = |e: &Error| match e {
1228 Error::Timeout(_)
1229 | Error::Network(_)
1230 | Error::InsufficientPeers(_)
1231 | Error::Storage(_)
1232 | Error::Payment(_)
1233 | Error::Protocol(_)
1234 | Error::InvalidData(_)
1235 | Error::NotFound(_)
1236 | Error::Serialization(_)
1237 | Error::Crypto(_)
1238 | Error::Io(_)
1239 | Error::Config(_)
1240 | Error::SignatureVerification(_)
1241 | Error::Encryption(_)
1242 | Error::AlreadyStored
1243 | Error::InsufficientDiskSpace(_)
1244 | Error::CostEstimationInconclusive(_)
1245 | Error::Cancelled(_)
1246 | Error::PartialUpload { .. }
1247 | Error::BadQuoteBinding { .. }
1248 | Error::BadQuoteCommitment { .. }
1249 | Error::MerkleBatchTooLarge { .. }
1250 | Error::RemotePut { .. }
1251 | Error::ClientUpdateRequired(_)
1252 | Error::StorerUpdateRequired(_)
1253 | Error::CloseGroupShortfall(_) => (),
1254 };
1255 }
1256}