ant-core 0.10.0

Headless Rust library for the Autonomi network: data storage and retrieval with self-encryption and EVM payments, plus node lifecycle management.
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
1021
1022
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
1077
1078
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
1153
1154
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
1221
1222
1223
1224
1225
1226
1227
1228
1229
1230
1231
1232
1233
1234
1235
1236
1237
1238
1239
1240
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
//! Client operations for the Autonomi network.
//!
//! Provides high-level APIs for storing and retrieving data
//! on the Autonomi decentralized network.

pub mod adaptive {
    pub use crate::client_engine::adaptive::*;
}
pub mod batch;
pub mod cache;
#[cfg(feature = "native")]
pub(crate) mod cached_merkle;
#[cfg(feature = "native")]
pub(crate) mod cached_single;
pub mod chunk;
pub mod data;
#[cfg(feature = "native")]
pub mod diagnostics;
pub mod file;
pub mod merkle;
#[cfg(feature = "native")]
mod native_payment;
pub mod payment;
pub mod quote;
pub mod upload;
pub mod upload_state;

use crate::data::client::adaptive::{AdaptiveConfig, AdaptiveController, ChannelStart, Outcome};
use crate::data::client::cache::ChunkCache;
use crate::data::error::{Error, Result};
use crate::data::network::Network;
#[cfg(feature = "native")]
use crate::data::network::NetworkHealth;
#[cfg(feature = "native")]
use crate::data::peer_cache;
use ant_protocol::evm::Wallet;
#[cfg(feature = "native")]
use ant_protocol::transport::P2PNode;
use ant_protocol::transport::{MultiAddr, PeerId};
use ant_protocol::{XorName, CLOSE_GROUP_SIZE};
use std::collections::HashSet;
use std::path::PathBuf;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::sync::Mutex;
#[cfg(feature = "native")]
use tracing::debug;

/// Width of the chunk PUT-target set (initial writes plus fallback): the
/// closest `PUT_TARGET_WIDTH` peers to the address.
///
/// Mirrors the node-side `K_BUCKET_SIZE` / `PAID_QUOTE_ISSUER_CLOSENESS_WIDTH`
/// (20): a node accepts a reused payment proof only when one of the proof's
/// closest-`CLOSE_GROUP_SIZE` quote issuers is within its own local 20-closest,
/// so trying peers past this width is pointless.
pub(crate) use crate::quote_policy::PUT_TARGET_WIDTH;

/// Ceiling on how long to wait for a peer to answer a settlement-versioned
/// quote request before falling back to the unversioned shape.
///
/// A storer that predates the versioned request cannot decode it and never
/// replies, so the only way to find out is to wait.
///
/// Note what is being waited for. This is **not** a cheap parse check: a peer
/// that understands the request runs the whole quote handler, queueing,
/// storage reads, pricing and signing, so the answer takes as long as any
/// quote takes. Abandoning the wait does not cancel that work either, it just
/// stops listening and adds a duplicate legacy request on top. So the wait has
/// to be sized for a real quote, not for a ping.
///
/// The full timeout is what made this expensive. The merkle E2E suite runs
/// with `quote_timeout_secs = 120`, so every probe against a fleet on the
/// published node cost two minutes and the suite blew the 60-minute CI cap.
///
/// # It must never bind in production
///
/// **Keep this at or above the largest production `quote_timeout_secs`**,
/// currently 10s. That is not a tuning preference, it is the safety property.
///
/// This wait is the only window in which a peer can refuse. Abandoning it
/// early does not merely mislabel a slow peer: it drops the refusal, so it
/// never counts toward corroboration, never sets the latch, and the legacy
/// request it raced can return a perfectly good quote the client then pays
/// against.
///
/// The two fallbacks lose it by different mechanisms. The merkle path sends
/// its legacy request under a *new* request id (`merkle.rs`), so a refusal
/// arriving after the ceiling is answering a request nobody is listening to
/// and is discarded on the id mismatch. The single-node path reuses the same
/// id (`quote.rs`), so a late refusal is seen only if it happens to arrive
/// after the retry has resubscribed; the await that would have matched it is
/// already gone, and anything landing in the gap between the two waits is
/// lost. Neither path observes a late refusal reliably, which is what the
/// ceiling exists to prevent.
///
/// A shorter ceiling was tried, at 5s, to bring the slower CI runner under the
/// cap. It would have turned every legitimate 5-to-10 second refusal in
/// production into exactly that silent downgrade. The never-demote rule does
/// not help: it stops a peer being *cached* as legacy after it has answered
/// once, but it does not stop the request in flight from falling back. And the
/// compile-time guard does not help either, because it binds future builds
/// while the clients at risk are the ones already released.
///
/// So the ceiling exists solely to bound configurations that set a timeout far
/// above any real answer time, which in practice means test harnesses. Above
/// 10s it never binds on a production client, and nothing real is truncated.
///
/// The cost of leaving it here is roughly two minutes per merkle E2E test
/// while the suite's devnet still speaks the pre-versioned dialect. That is a
/// consequence of the temporary fork-branch protocol pin, not of the design:
/// once the fleet under test can answer a versioned request there are no
/// probes to pay for, and the suite returns to its baseline.
pub(crate) const VERSIONED_QUOTE_PROBE_CEILING: std::time::Duration =
    std::time::Duration::from_secs(15);

/// How many distinct peers must refuse this client's settlement version before
/// the refusal is believed and uploads stop.
///
/// Nothing authenticates a refusal, so one peer's word cannot be enough: a
/// single hostile or misconfigured storer answering `ClientUpdateRequired` to
/// everything would otherwise deny every upload, turning an over-query design
/// that tolerates many bad peers into one that tolerates none.
///
/// Two is deliberately low. A genuine incompatibility reaches it instantly,
/// because every peer enforcing the newer rule refuses, and a client queries
/// far more than two. An attacker has to control two of the peers a given
/// request happens to reach, which is a materially different proposition from
/// controlling one.
pub(crate) const SETTLEMENT_REFUSAL_QUORUM: usize = 2;

/// Compile-time cutover guard shared by **every** unversioned quote retry.
///
/// Both quote paths fall back to an unversioned request when a peer stays
/// silent, because a storer built before the settlement version existed cannot
/// decode the versioned one. Silence is not proof of that, though: a dropped
/// response, packet loss, an overloaded peer, or one deliberately discarding
/// versioned requests are indistinguishable from the client side. So the
/// fallback is a downgrade path and can be provoked.
///
/// It is not only literal silence, either. `send_and_await_chunk_response`
/// keeps waiting when a reply decodes but carries a body the mapper does not
/// recognise, so a peer answering with an unexpected variant also lands on the
/// timeout and takes this path. That grants no capability beyond staying
/// silent, which is why it is bounded rather than special-cased, but a comment
/// claiming "any structured response prevents the retry" would be wrong.
///
/// It is safe only while **no refusal of either kind is possible**, and that
/// means bounding both constants, not just the minimum.
///
/// Raising `MIN` is the obvious hazard: a client below it would be refused,
/// and the retry hands it a quote anyway. Raising `CURRENT` is the subtler
/// one. As soon as some node runs a newer `CURRENT` than another, the older
/// node refuses newer clients with `StorerUpdateRequired` precisely because it
/// cannot promise to honour their payment. A client that retries such a peer
/// unversioned gets that unhonourable quote, and for a settlement change that
/// is not a pure increase the payment is then rejected after it has settled.
/// Guarding only `MIN` would leave that route open.
///
/// So the guard requires both to still be at the first declarable version.
///
/// This bounds **future builds** only. A client binary already in the field
/// carries whatever fallback it shipped with, and no source change reaches it;
/// that is inherent to shipping software and is why the storer still verifies
/// every payment it is actually offered.
///
/// Each fallback site references this constant so the guard cannot be orphaned
/// by deleting one path and forgetting the other. Retiring the fallbacks means
/// deleting this constant and every reference to it, which the compiler then
/// points at one by one.
pub(crate) const UNVERSIONED_RETRY_REQUIRES_MIN_V1: () = assert!(
    ant_protocol::MIN_SUPPORTED_SETTLEMENT_VERSION == 1
        && ant_protocol::CURRENT_SETTLEMENT_VERSION == 1,
    "an unversioned quote retry is still compiled in: it is a downgrade path around \
     both the too-old and the node-behind refusals, so delete every fallback site \
     before raising MIN_SUPPORTED_SETTLEMENT_VERSION or CURRENT_SETTLEMENT_VERSION"
);

/// Distinct peers that have refused this client's settlement version, plus the
/// wording of the first refusal.
///
/// A separate type rather than a field so the corroboration rule can be tested
/// on its own: it is the piece that decides whether an upload stops, and it
/// has to hold against both a lone lying peer and a genuine incompatibility.
#[derive(Clone, Default)]
pub(crate) struct SettlementRefusals {
    inner: Arc<Mutex<(HashSet<PeerId>, Option<String>)>>,
}

impl SettlementRefusals {
    /// Record a refusal from `peer_id`, returning the wording once
    /// [`SETTLEMENT_REFUSAL_QUORUM`] distinct peers agree and `None` below it.
    pub(crate) fn note(&self, peer_id: PeerId, message: &str) -> Option<String> {
        let mut guard = self.inner.lock().ok()?;
        let (peers, wording) = &mut *guard;
        peers.insert(peer_id);
        if wording.is_none() {
            *wording = Some(message.to_string());
        }
        (peers.len() >= SETTLEMENT_REFUSAL_QUORUM)
            .then(|| wording.clone())
            .flatten()
    }

    /// The corroborated refusal, if one has been established.
    pub(crate) fn corroborated(&self) -> Option<String> {
        let guard = self.inner.lock().ok()?;
        let (peers, wording) = &*guard;
        (peers.len() >= SETTLEMENT_REFUSAL_QUORUM)
            .then(|| wording.clone())
            .flatten()
    }

    /// The distinct refusing peers recorded so far, rendered for logging,
    /// sorted so the line is deterministic.
    ///
    /// Exists so the terminal abort can NAME its corroborators. Without it the
    /// quorum is unverifiable from outside: "awaiting corroboration" is only
    /// logged below the quorum and the terminal error carries no peer ids, so
    /// a log reader sees one warned peer followed by an abort whether the
    /// quorum counted two distinct peers or misfired on one — which is exactly
    /// the ambiguity the V2-1109 testnet run hit (0/463 aborts showed a second
    /// peer, because the second peer was structurally unloggable).
    pub(crate) fn corroborating_peers(&self) -> Vec<String> {
        let Ok(guard) = self.inner.lock() else {
            return Vec::new();
        };
        let (peers, _) = &*guard;
        let mut ids: Vec<String> = peers.iter().map(|p| format!("{p}")).collect();
        ids.sort();
        ids
    }
}

#[cfg(test)]
mod settlement_refusal_tests {
    use super::*;

    fn peer(seed: u8) -> PeerId {
        PeerId::from_bytes([seed; 32])
    }

    /// The question the V2-1109 run could not answer from logs: does one peer
    /// refusing many times count as many? It must not — the quorum is over
    /// DISTINCT peers, and a lone hostile or misconfigured storer repeating
    /// itself must never become a verdict about this build.
    #[test]
    fn one_peer_refusing_many_times_never_reaches_the_quorum() {
        let refusals = SettlementRefusals::default();
        for _ in 0..50 {
            assert!(
                refusals.note(peer(7), "run ant update").is_none(),
                "a single peer's repeated refusals must stay below the quorum"
            );
        }
        assert!(refusals.corroborated().is_none());
        assert_eq!(refusals.corroborating_peers().len(), 1);
    }

    /// The terminal log line must be able to name both corroborators, so the
    /// distinct-peer property is verifiable from the outside.
    #[test]
    fn corroborating_peers_names_every_distinct_refuser() {
        let refusals = SettlementRefusals::default();
        assert!(refusals.note(peer(1), "run ant update").is_none());
        assert!(refusals.note(peer(2), "run ant update").is_some());

        let ids = refusals.corroborating_peers();
        assert_eq!(ids.len(), 2);
        assert_ne!(ids[0], ids[1]);
        assert_eq!(ids, {
            let mut sorted = ids.clone();
            sorted.sort();
            sorted
        });
    }
}

/// Classify a `data::error::Error` into a controller `Outcome`.
///
/// Capacity signals (Timeout / NetworkError) drive the controller
/// down; application errors do not. The mapping is conservative:
/// anything that COULD be transport-related is treated as a network
/// signal, because under-classifying a real network failure as
/// "application error" makes the controller blind to genuine stress.
///
/// Mapping policy:
/// - `Timeout` -> `Timeout` (per-op deadline elapsed)
/// - `Network`, `InsufficientPeers`, `Io` -> `NetworkError` (transport
///   layer reported failure)
/// - `Protocol`, `Storage` -> `NetworkError` (these wrap remote errors
///   that frequently include peer disconnects mid-stream — under
///   network stress these are how transport failures surface)
/// - `PartialUpload` -> `NetworkError` (literal capacity signal: some
///   chunks could not be stored)
/// - `AlreadyStored`, `Encryption`, `Crypto`, `Payment`,
///   `Serialization`, `InvalidData`, `NotFound`, `SignatureVerification`,
///   `Config`, `InsufficientDiskSpace`, `CostEstimationInconclusive`,
///   `Cancelled` -> `ApplicationError` (would happen on a perfectly
///   healthy link; `Cancelled` is caller-initiated and must not be retried
///   as a transport failure; `NotFound` is a definitively absent record
///   reported over a working link, not a transport symptom)
/// - `RemotePut` -> `ApplicationError` (the remote node responded with a
///   structured rejection — the transport succeeded, so the node declined
///   at the application layer; not a local capacity signal)
/// - `ClientUpdateRequired` -> `ApplicationError` (the storer refused to quote
///   a client that settles under superseded rules — a terminal verdict about
///   this build, not about link capacity, and no retry rate clears it)
/// - `CloseGroupShortfall` -> `ApplicationError` (a quorum shortfall caused
///   by close-group dial/relay churn with no PUT-response timeouts — remote
///   peer churn, not local backpressure; a timeout-bearing shortfall keeps
///   `InsufficientPeers`/`NetworkError` instead, so genuine congestion still
///   cuts the cap — V2-554)
pub(crate) fn classify_error(err: &Error) -> Outcome {
    match err {
        Error::Timeout(_) => crate::transfer_policy::FailureKind::Timeout,
        Error::Network(_)
        | Error::InsufficientPeers(_)
        | Error::Io(_)
        | Error::Protocol(_)
        | Error::Storage(_)
        | Error::PartialUpload { .. } => crate::transfer_policy::FailureKind::Network,
        Error::AlreadyStored
        | Error::Encryption(_)
        | Error::Crypto(_)
        | Error::Payment(_)
        | Error::Serialization(_)
        | Error::InvalidData(_)
        // A definitively absent record, reported over a working link —
        // the peers answered, there was just nothing stored there. Not a
        // transport symptom, so it must not push the limiter down.
        | Error::NotFound(_)
        | Error::SignatureVerification(_)
        | Error::Config(_)
        | Error::InsufficientDiskSpace(_)
        | Error::CostEstimationInconclusive(_)
        | Error::Cancelled(_)
        // The storer parsed our request and refused it on its merits, over a
        // working link. Sending fewer requests would not help, and treating it
        // as congestion would quietly shrink the limiter for the rest of the
        // run on the basis of a fault no retry can clear.
        | Error::ClientUpdateRequired(_)
        | Error::StorerUpdateRequired(_)
        | Error::BadQuoteBinding { .. }
        | Error::BadQuoteCommitment { .. }
        // An external-signer merkle batch larger than one tree can hold —
        // a caller-shape refusal raised before any network work, so it says
        // nothing about link capacity.
        | Error::MerkleBatchTooLarge { .. }
        // A remote node responded with a structured rejection — the
        // transport round-trip succeeded, so the node declined at the
        // application layer (payment/disk/quote/pool). Not a local
        // capacity signal; recorded but must not push the limiter down.
        | Error::RemotePut { .. }
        // A close-group PUT shortfall caused purely by dial/relay churn
        // (dead/stale relayed peer addresses), with no PUT-response
        // timeouts to signal local backpressure. Remote peer churn, not
        // "client sending too fast" — must not push the limiter down
        // (V2-554). A shortfall that DID time out keeps `InsufficientPeers`
        // (`NetworkError`) so real congestion still cuts the cap.
        | Error::CloseGroupShortfall(_) => crate::transfer_policy::FailureKind::Application,
    }
    .outcome()
}

/// Compute XOR distance between a peer's ID bytes and a target address.
///
/// Uses the first 32 bytes of the peer ID (or fewer if shorter) XORed
/// with the target address. The returned byte array sorts
/// lexicographically from closest to furthest.
pub(crate) fn peer_xor_distance(peer_id: &PeerId, target: &[u8; 32]) -> [u8; 32] {
    let peer_bytes = peer_id.as_bytes();
    let mut distance = [0u8; 32];
    for (i, d) in distance.iter_mut().enumerate() {
        let peer_byte = peer_bytes.get(i).copied().unwrap_or(0);
        *d = peer_byte ^ target[i];
    }
    distance
}

/// Default timeout for lightweight network operations (quotes, DHT lookups) in seconds.
const DEFAULT_QUOTE_TIMEOUT_SECS: u64 = 10;

/// Default timeout for the per-peer chunk GET response and any other
/// caller that explicitly reads `store_timeout_secs`, in seconds.
///
/// Note despite the name: this knob does **not** govern the non-merkle
/// chunk PUT response timeout — that path uses the
/// `STORE_RESPONSE_TIMEOUT` constant in `chunk.rs` directly. Nor does
/// it govern the merkle batch PUT timeout — see
/// `DEFAULT_MERKLE_STORE_TIMEOUT_SECS`.
///
/// 10 s matches the pre-existing `main` default and intentionally
/// excludes residential-upload tuning, which is Mick's PR #78
/// territory (splitting GET into its own field).
const DEFAULT_STORE_TIMEOUT_SECS: u64 = 10;

/// Default timeout for **merkle batch** chunk store operations in seconds.
///
/// Separate from `DEFAULT_STORE_TIMEOUT_SECS` because merkle PUTs carry
/// an extra storer-side cost: the payment verifier runs an iterative
/// DHT lookup (`CLOSENESS_LOOKUP_TIMEOUT` in `ant-node`, **240 s**
/// post-PR #89) before accepting the proof.
///
/// This timeout MUST be >= the storer-side `CLOSENESS_LOOKUP_TIMEOUT`
/// plus padding for the store-response round-trip and storer-local
/// I/O. Otherwise the client gives up while the storer is still
/// happily verifying, the storer wastes CPU/bandwidth on a chunk the
/// client has already discarded, and the client re-targets a
/// different close-K member — potentially double-storing the same
/// chunk and polluting routing.
///
/// 270 s = 240 s (storer lookup) + 30 s padding (network RTT + LMDB
/// put + fsync + clock skew tolerance).
///
/// This invariant must be re-validated if either side's timeout
/// changes. Empirically surfaced as "every cross-region merkle chunk
/// times out at 10 s" on a 210-node 7-region testnet run on
/// 2026-05-12; bumping to 270 s flipped that 0/22 -> 9/9 pass rate.
const DEFAULT_MERKLE_STORE_TIMEOUT_SECS: u64 = 270;

/// Default timeout for chunk GET response operations in seconds.
const DEFAULT_CHUNK_GET_TIMEOUT_SECS: u64 = 10;

/// Default quote concurrency: high because quoting is pure network I/O
/// (DHT lookups + small request/response messages) with no CPU-bound work.
const DEFAULT_QUOTE_CONCURRENCY: usize = 32;

/// Default store concurrency: moderate because each chunk PUT sends ~4MB
/// to 7 close-group peers. At 8 concurrent stores, ~225MB of outbound
/// traffic can be in flight. Users on fast connections can increase this
/// with --store-concurrency; users on slow connections can decrease it.
const DEFAULT_STORE_CONCURRENCY: usize = 8;

/// Configuration for the Autonomi client.
#[derive(Debug, Clone)]
pub struct ClientConfig {
    /// Per-op timeout for lightweight network operations (quotes,
    /// DHT lookups), in seconds. The adaptive controller does NOT
    /// currently size timeouts; this remains a static knob.
    pub quote_timeout_secs: u64,
    /// Per-op timeout, in seconds, for the chunk GET response path
    /// (`chunk_get_from_peer`) and any other caller that reads this
    /// field directly.
    ///
    /// Note despite the historical name `store_timeout_secs`: this
    /// knob does **not** govern the non-merkle chunk PUT response
    /// timeout (that path uses the `STORE_RESPONSE_TIMEOUT` constant
    /// in `chunk.rs`) and does **not** govern the merkle batch PUT
    /// timeout (see `merkle_store_timeout_secs`). Rename pending in
    /// Mick's PR #78 which adds a dedicated `chunk_get_timeout_secs`.
    ///
    /// The adaptive controller does NOT currently size timeouts;
    /// this remains a static knob.
    pub store_timeout_secs: u64,
    /// Per-op timeout for **merkle batch** chunk store (PUT)
    /// operations, in seconds. Separate from `store_timeout_secs`
    /// because merkle PUTs incur the storer-side
    /// `CLOSENESS_LOOKUP_TIMEOUT` (240 s post-PR #89) on top of the
    /// usual store path; the client must wait at least that long
    /// plus padding, or the storer wastes work on a chunk the client
    /// has already given up on. Default 270 s.
    pub merkle_store_timeout_secs: u64,
    /// Per-peer response timeout for chunk GET operations, in seconds.
    /// This is intentionally independent from `store_timeout_secs`: PUTs
    /// and GETs have different payload direction and performance profiles.
    pub chunk_get_timeout_secs: u64,
    /// Number of closest peers to consider for routing.
    pub close_group_size: usize,
    /// **Deprecated.** Pre-adaptive ceiling for quote concurrency.
    ///
    /// The adaptive controller now sizes quote fan-out from observed
    /// signals. This field, when non-zero and smaller than the
    /// controller's per-channel default, clamps the **quote channel
    /// only** (it does NOT bleed into store or fetch). Removed in a
    /// future release.
    pub quote_concurrency: usize,
    /// **Deprecated.** Pre-adaptive ceiling for store concurrency.
    ///
    /// The adaptive controller now sizes store fan-out from observed
    /// signals. This field, when non-zero and smaller than the
    /// controller's per-channel default, clamps the **store channel
    /// only** (it does NOT bleed into quote or fetch). Removed in a
    /// future release.
    pub store_concurrency: usize,
    /// Adaptive controller configuration. Defaults are tuned to match
    /// or exceed the prior static behavior — disabling adaptation
    /// (`adaptive.enabled = false`) reverts to the controller's
    /// `initial` values without re-evaluation.
    pub adaptive: AdaptiveConfig,
    /// Allow loopback (`127.0.0.1`) connections in the saorsa-transport
    /// layer. Set to `true` only for devnet / local testing. Production
    /// peers on the public Autonomi network reject the QUIC handshake
    /// variant produced when this is `true`, so the default is `false`.
    ///
    /// This mirrors the `--allow-loopback` flag in `ant-cli`, which already
    /// defaults to `false` and threads through to the same
    /// `CoreNodeConfig::builder().local(...)` call.
    pub allow_loopback: bool,
    /// Bind a dual-stack IPv6 socket (`true`) or an IPv4-only socket
    /// (`false`). Defaults to `true`, matching the CLI default.
    ///
    /// Set to `false` only when running on hosts without a working IPv6
    /// stack, to avoid advertising unreachable v6 addresses to the DHT
    /// (which causes slow connects and junk DHT address records). This
    /// mirrors the `--ipv4-only` flag in `ant-cli`.
    pub ipv6: bool,
    /// Per-batch leaf cap for **external-signer** merkle preparation,
    /// clamped to `3..=MAX_LEAVES` when set (a cap of 2 cannot partition odd
    /// totals — parts of 3 and 2 compose any count, so 3 is the smallest
    /// safe cap). `None` (the default) uses the contract maximum
    /// (`MAX_LEAVES` = 256).
    ///
    /// This is a test seam (ADR-0003): a small cap makes
    /// `file_prepare_upload_with_mode` produce a genuine multi-batch
    /// prepared upload from a kilobyte file, so the N-signature external
    /// flow is exercisable in E2E without a multi-GiB fixture. Production
    /// callers should leave it `None` — a lower cap only means more payment
    /// transactions for the same chunks.
    pub merkle_external_batch_cap: Option<usize>,
}

impl Default for ClientConfig {
    fn default() -> Self {
        Self {
            quote_timeout_secs: DEFAULT_QUOTE_TIMEOUT_SECS,
            store_timeout_secs: DEFAULT_STORE_TIMEOUT_SECS,
            merkle_store_timeout_secs: DEFAULT_MERKLE_STORE_TIMEOUT_SECS,
            chunk_get_timeout_secs: DEFAULT_CHUNK_GET_TIMEOUT_SECS,
            close_group_size: CLOSE_GROUP_SIZE,
            quote_concurrency: DEFAULT_QUOTE_CONCURRENCY,
            store_concurrency: DEFAULT_STORE_CONCURRENCY,
            adaptive: AdaptiveConfig::default(),
            allow_loopback: false,
            ipv6: true,
            merkle_external_batch_cap: None,
        }
    }
}

/// Build the adaptive controller for a `Client`. Loads any persisted
/// snapshot, clamps cold-start values into the deprecated-flag bounds
/// **per channel** (so a pin on `--store-concurrency` does NOT bleed
/// into the fetch / quote channels), and returns the persistence path
/// so callers can save back at shutdown.
fn build_controller(config: &ClientConfig) -> (AdaptiveController, Option<PathBuf>) {
    let mut adaptive_cfg = config.adaptive.clone();

    // Per-channel ceilings: each legacy field is interpreted as a cap
    // for ONLY its matching channel. The fetch channel has no
    // pre-existing legacy field; it always uses the controller's
    // default ceiling.
    //
    // The legacy fields are non-zero by ClientConfig::default(), but
    // we honor them as bounds only when they would actually CONSTRAIN
    // the controller — i.e. when smaller than the per-channel default
    // max. A default ClientConfig must not silently lower the
    // controller's ceilings.
    // A value equal to the historic legacy default is treated as
    // "not pinned by the user" — without this, every default
    // ClientConfig would silently lower the controller's per-channel
    // ceilings to the prior static values (32/8) and the controller
    // could never grow above them.
    let user_quote_max = config.quote_concurrency;
    let user_store_max = config.store_concurrency;
    let quote_pinned = user_quote_max > 0 && user_quote_max != DEFAULT_QUOTE_CONCURRENCY;
    let store_pinned = user_store_max > 0 && user_store_max != DEFAULT_STORE_CONCURRENCY;
    if quote_pinned && user_quote_max < adaptive_cfg.max.quote {
        adaptive_cfg.max.quote = user_quote_max;
    }
    if store_pinned && user_store_max < adaptive_cfg.max.store {
        adaptive_cfg.max.store = user_store_max;
    }

    // Cold-start values: matched to the prior static defaults. If the
    // legacy field caps the channel below the cold-start, lower the
    // start to match — never start above the channel's max.
    let mut start = ChannelStart::default();
    start.quote = start.quote.min(adaptive_cfg.max.quote);
    start.store = start.store.min(adaptive_cfg.max.store);
    start.fetch = start.fetch.min(adaptive_cfg.max.fetch);

    #[cfg(feature = "native")]
    let adaptive_enabled = adaptive_cfg.enabled;
    let controller = AdaptiveController::new(start, adaptive_cfg);
    // Skip disk warm-start entirely when adaptation is disabled —
    // fixed-concurrency mode means the user wants exactly the cold
    // start, no surprises from prior runs. (warm_start is also a
    // no-op when disabled, but skipping the load avoids file I/O
    // and the path-resolution side effects.)
    #[cfg(feature = "native")]
    let persist_path = if adaptive_enabled {
        let p = adaptive::default_persist_path();
        if let Some(ref path) = p {
            if let Some(snap) = adaptive::load_snapshot(path) {
                debug!(path = %path.display(), "adaptive: warm-start from disk");
                controller.warm_start(snap);
            }
        }
        p
    } else {
        // Even with adaptation off, persist_path is computed so
        // explicit save_adaptive_snapshot() calls still work — but
        // the controller currently never moves, so saving the cold
        // start is harmless.
        adaptive::default_persist_path()
    };

    // File downloads choose a stream-decrypt batch size per download
    // from the current fetch cap and usable RAM, then pass it into
    // self_encryption's runtime batch-size API. The adaptive controller
    // still drives fan-out inside each batch by re-reading
    // `controller.fetch.current()` in the decrypt callback.

    #[cfg(not(feature = "native"))]
    let persist_path = None;
    (controller, persist_path)
}

/// Client for the Autonomi decentralized network.
///
/// Provides high-level APIs for storing and retrieving chunks
/// and files on the network.
pub struct Client {
    config: ClientConfig,
    network: Network,
    wallet: Option<Arc<Wallet>>,
    evm_network: Option<ant_protocol::evm::Network>,
    chunk_cache: ChunkCache,
    next_request_id: AtomicU64,
    /// Adaptive concurrency controller: replaces the static
    /// quote/store concurrency knobs. See `adaptive` module.
    controller: AdaptiveController,
    /// Path the controller persists its snapshot to. `None` disables
    /// persistence (useful for tests / non-disk environments).
    #[cfg(feature = "native")]
    persist_path: Option<PathBuf>,
    /// Path for the persistent client peer cache. `None` disables the cache.
    #[cfg(feature = "native")]
    peer_cache_path: Option<PathBuf>,
    /// Peers that did not answer a settlement-versioned quote request, and are
    /// therefore asked in the legacy shape from now on.
    ///
    /// Without this the probe cost is paid on **every** request rather than
    /// roughly once per peer. Measured on the merkle E2E suite against a fleet
    /// that predates the versioned requests, re-probing took the run from ~24
    /// minutes to over 60, because each of the sixteen candidates per pool sat
    /// out a full `quote_timeout_secs` before the fallback.
    ///
    /// Roughly, not exactly: concurrent first contacts are not coalesced, so
    /// several in-flight requests can all miss the cache for the same peer and
    /// each probe it once before any of them records the answer. Observed at
    /// about two probes per peer on a 35-node devnet. Single-flighting them
    /// would remove the duplicates but not the wall-clock cost, which is set
    /// by how many *sequential* quote rounds an upload performs rather than by
    /// how many probes each round contains.
    ///
    /// Process-local and never persisted. A peer that upgrades mid-run keeps
    /// being asked in the legacy shape until the next start, which is
    /// acceptable while the legacy shape still gets a quote, and stops
    /// mattering when the fallback is deleted (see
    /// [`UNVERSIONED_RETRY_REQUIRES_MIN_V1`]).
    ///
    /// Entries are only ever added for a peer that has **never** answered a
    /// versioned request. Without that condition a single lost response would
    /// pin an upgraded peer to the legacy shape for the rest of the session,
    /// turning one dropped packet into a standing downgrade; with it, a peer
    /// that has shown it understands the versioned shape can never be demoted.
    ///
    /// A peer that has never answered can still get itself asked without a
    /// version by staying silent, exactly as it could through the fallback
    /// alone. Remembering the answer makes that cheaper to sustain, so it is
    /// not a new capability but it is a wider one, and the compile-time guard
    /// requires the whole path to be gone before any refusal is possible.
    unversioned_quote_peers: Arc<Mutex<HashSet<PeerId>>>,
    /// Peers observed answering a settlement-versioned request. Never
    /// downgraded, however they behave later.
    versioned_capable_peers: Arc<Mutex<HashSet<PeerId>>>,
    /// Distinct peers that have refused this client on settlement-version
    /// grounds, and the wording of the first such refusal.
    ///
    /// Client-wide and sticky, for two reasons that pull in opposite
    /// directions and are both real.
    ///
    /// It must outlive one operation, because the verdict is about this
    /// **build**, not this upload. Held in a single collector's local state, a
    /// refusal observed by one in-flight upload says nothing to another that
    /// is about to submit a payment, and merkle payments cannot be undone.
    ///
    /// It must not fire on one peer's say-so, because nothing authenticates a
    /// refusal. A single hostile or confused peer answering
    /// `ClientUpdateRequired` to every query would otherwise abort every
    /// upload the client attempts, converting an over-query design that
    /// tolerates many bad peers into one that tolerates none. So a refusal
    /// becomes terminal only once [`SETTLEMENT_REFUSAL_QUORUM`] distinct peers
    /// agree, which a genuine incompatibility reaches immediately (every
    /// upgraded peer refuses) and a lone attacker cannot reach at all.
    settlement_refusals: SettlementRefusals,
}

impl Client {
    /// Create a client using a platform network adapter.
    #[must_use]
    pub fn from_network(network: Network, config: ClientConfig) -> Self {
        let (controller, _persist_path) = build_controller(&config);
        Self {
            config,
            network,
            wallet: None,
            evm_network: None,
            chunk_cache: ChunkCache::default(),
            next_request_id: AtomicU64::new(1),
            controller,
            #[cfg(feature = "native")]
            persist_path: _persist_path,
            #[cfg(feature = "native")]
            peer_cache_path: None,
            unversioned_quote_peers: Arc::new(Mutex::new(HashSet::new())),
            versioned_capable_peers: Arc::new(Mutex::new(HashSet::new())),
            settlement_refusals: SettlementRefusals::default(),
        }
    }

    /// Keep compatibility decisions across browser operations whose network adapter
    /// carries a per-operation payment network identity.
    #[cfg(not(feature = "native"))]
    pub(crate) fn with_shared_quote_state(mut self, session: &Self) -> Self {
        self.unversioned_quote_peers = Arc::clone(&session.unversioned_quote_peers);
        self.versioned_capable_peers = Arc::clone(&session.versioned_capable_peers);
        self.settlement_refusals = session.settlement_refusals.clone();
        self
    }

    /// Create a client connected to the given P2P node.
    #[must_use]
    #[cfg(feature = "native")]
    pub fn from_node(node: Arc<P2PNode>, config: ClientConfig) -> Self {
        Self::from_node_with_peer_cache(node, config, None)
    }

    /// Create a client connected to the given P2P node and attach an optional
    /// persistent peer cache path.
    #[must_use]
    #[cfg(feature = "native")]
    pub fn from_node_with_peer_cache(
        node: Arc<P2PNode>,
        config: ClientConfig,
        #[cfg(feature = "native")] peer_cache_path: Option<PathBuf>,
    ) -> Self {
        let network = Network::from_node(node);
        let (controller, persist_path) = build_controller(&config);
        Self {
            config,
            network,
            wallet: None,
            evm_network: None,
            chunk_cache: ChunkCache::default(),
            next_request_id: AtomicU64::new(1),
            unversioned_quote_peers: Arc::new(Mutex::new(HashSet::new())),
            versioned_capable_peers: Arc::new(Mutex::new(HashSet::new())),
            settlement_refusals: SettlementRefusals::default(),
            controller,
            persist_path,
            peer_cache_path,
        }
    }

    /// Create a client connected to bootstrap peers.
    ///
    /// Threads `config.allow_loopback` and `config.ipv6` through to
    /// `Network::new`, which controls the saorsa-transport `local` and
    /// `ipv6` flags on the underlying `CoreNodeConfig`. See
    /// `ClientConfig::allow_loopback` and `ClientConfig::ipv6` for details.
    ///
    /// # Errors
    ///
    /// Returns an error if the P2P node cannot be created or bootstrapping fails.
    #[cfg(feature = "native")]
    pub async fn connect(
        bootstrap_peers: &[std::net::SocketAddr],
        config: ClientConfig,
    ) -> Result<Self> {
        let seeds: Vec<_> = bootstrap_peers
            .iter()
            .copied()
            .map(ant_protocol::transport::MultiAddr::quic)
            .collect();
        Self::connect_multiaddrs(&seeds, config).await
    }

    /// Connect using native QUIC multiaddresses without discarding peer pins.
    #[cfg(feature = "native")]
    pub async fn connect_multiaddrs(
        bootstrap_peers: &[ant_protocol::transport::MultiAddr],
        config: ClientConfig,
    ) -> Result<Self> {
        debug!(
            "Connecting to Autonomi network with {} bootstrap peers (allow_loopback={}, ipv6={})",
            bootstrap_peers.len(),
            config.allow_loopback,
            config.ipv6,
        );
        let network =
            Network::new_multiaddrs(bootstrap_peers, config.allow_loopback, config.ipv6).await?;
        let (controller, persist_path) = build_controller(&config);
        Ok(Self {
            config,
            network,
            wallet: None,
            evm_network: None,
            chunk_cache: ChunkCache::default(),
            next_request_id: AtomicU64::new(1),
            unversioned_quote_peers: Arc::new(Mutex::new(HashSet::new())),
            versioned_capable_peers: Arc::new(Mutex::new(HashSet::new())),
            settlement_refusals: SettlementRefusals::default(),
            controller,
            persist_path,
            peer_cache_path: None,
        })
    }

    /// Set the wallet for payment operations.
    ///
    /// Also populates the EVM network from the wallet so that
    /// token approvals work without a separate `with_evm_network` call.
    #[must_use]
    pub fn with_wallet(mut self, wallet: Wallet) -> Self {
        self.evm_network = Some(wallet.network().clone());
        self.wallet = Some(Arc::new(wallet));
        self
    }

    /// Set the EVM network without requiring a wallet.
    ///
    /// This enables token approval and contract interactions
    /// for external-signer flows where the private key lives outside Rust.
    #[must_use]
    pub fn with_evm_network(mut self, network: ant_protocol::evm::Network) -> Self {
        self.evm_network = Some(network);
        self
    }

    /// Get the EVM network, falling back to the wallet's network if available.
    ///
    /// # Errors
    ///
    /// Returns an error if neither `with_evm_network` nor `with_wallet` was called.
    pub(crate) fn require_evm_network(&self) -> Result<&ant_protocol::evm::Network> {
        if let Some(ref net) = self.evm_network {
            return Ok(net);
        }
        if let Some(ref wallet) = self.wallet {
            return Ok(wallet.network());
        }
        Err(Error::Payment(
            "EVM network not configured — call with_evm_network() or with_wallet() first"
                .to_string(),
        ))
    }

    /// Get the client configuration.
    #[must_use]
    pub fn config(&self) -> &ClientConfig {
        &self.config
    }

    /// Get a mutable reference to the client configuration.
    pub fn config_mut(&mut self) -> &mut ClientConfig {
        &mut self.config
    }

    /// Get a reference to the network layer.
    #[must_use]
    pub fn network(&self) -> &Network {
        &self.network
    }

    /// Compute the live network-participation snapshot.
    ///
    /// Convenience pass-through to [`Network::health`] — the single
    /// write-readiness implementation shared by all embedded-client
    /// consumers (antd, ant-gui, ant-ffi, ant-tui).
    #[cfg(feature = "native")]
    pub async fn network_health(&self) -> NetworkHealth {
        self.network.health().await
    }

    /// Get the wallet, if configured.
    #[must_use]
    pub fn wallet(&self) -> Option<&Arc<Wallet>> {
        self.wallet.as_ref()
    }

    /// Set the in-memory cache budget for this client.
    #[must_use]
    pub fn with_chunk_cache(mut self, cache: ChunkCache) -> Self {
        self.chunk_cache = cache;
        self
    }

    /// Get a reference to the chunk cache.
    #[must_use]
    pub fn chunk_cache(&self) -> &ChunkCache {
        &self.chunk_cache
    }

    /// Adaptive concurrency controller. Hot loops read
    /// `controller().<channel>.current()` to size their fan-out and
    /// call `.observe(...)` on each completion.
    #[must_use]
    pub fn controller(&self) -> &AdaptiveController {
        &self.controller
    }

    /// Persist the current adaptive snapshot to disk so the next
    /// `Client::connect` warm-starts at the learned values instead of
    /// cold defaults. Best effort — failures log and are discarded.
    /// Idempotent. Safe to call from a Drop impl or an explicit
    /// shutdown hook.
    #[cfg(feature = "native")]
    pub fn save_adaptive_snapshot(&self) {
        if let Some(ref path) = self.persist_path {
            adaptive::save_snapshot(path, self.controller.snapshot());
        }
    }

    /// Persist currently connected peers that have Direct-tagged addresses in
    /// the DHT. Best effort; failures are logged and do not affect the client
    /// operation that just completed.
    #[cfg(feature = "native")]
    pub async fn save_peer_cache(&self) {
        if let Some(ref path) = self.peer_cache_path {
            let node = self.network().node();
            peer_cache::promote_connected_direct_peers(node.as_ref(), path, node.dht().k_value())
                .await;
        }
    }

    /// Get the next request ID for protocol messages.
    pub(crate) fn next_request_id(&self) -> u64 {
        self.next_request_id.fetch_add(1, Ordering::Relaxed)
    }

    /// Handle to the set of peers that cannot answer a settlement-versioned
    /// quote request, shared with the per-peer request futures on both quote
    /// paths.
    ///
    /// Callers read it before choosing a request shape and insert into it when
    /// a peer stays silent. A poisoned lock is treated as "nothing known", so
    /// the worst case is a wasted probe rather than a silently skipped version
    /// declaration.
    pub(crate) fn unversioned_quote_peers(&self) -> Arc<Mutex<HashSet<PeerId>>> {
        Arc::clone(&self.unversioned_quote_peers)
    }

    /// Handle to the set of peers already seen answering a versioned request.
    /// Consulted before demoting a peer, so a lost response cannot strand an
    /// upgraded peer in the legacy shape.
    pub(crate) fn versioned_quote_capable_handle(&self) -> Arc<Mutex<HashSet<PeerId>>> {
        Arc::clone(&self.versioned_capable_peers)
    }

    /// Record that `peer_id` refused this client's settlement version, and
    /// report whether enough distinct peers now agree for it to be believed.
    ///
    /// Returns the refusal wording once [`SETTLEMENT_REFUSAL_QUORUM`] is met,
    /// and `None` below it, so a lone peer is treated as a peer fault rather
    /// than a verdict about this build.
    pub(crate) fn note_settlement_refusal(&self, peer_id: PeerId, message: &str) -> Option<String> {
        self.settlement_refusals.note(peer_id, message)
    }

    /// The corroborated refusal, if this client has already been told by
    /// enough peers that it cannot settle.
    ///
    /// Checked before spending money. The verdict concerns this build rather
    /// than any one upload, so an upload that starts after another has already
    /// established it must not proceed to pay.
    pub(crate) fn corroborated_settlement_refusal(&self) -> Option<String> {
        self.settlement_refusals.corroborated()
    }

    /// Handle to the shared refusal tracker, for collectors that run outside
    /// `&self`.
    pub(crate) fn settlement_refusals(&self) -> SettlementRefusals {
        self.settlement_refusals.clone()
    }

    /// Return the chunk PUT-target set: the closest [`PUT_TARGET_WIDTH`] peers
    /// to the address, each paired with its known network addresses.
    ///
    /// Used by the merkle store path, which — unlike single-node payment — has
    /// no witnessed put-target list to forward, so it fetches the closest-K
    /// neighbourhood locally.
    pub(crate) async fn put_target_peers(
        &self,
        target: &XorName,
    ) -> Result<Vec<(PeerId, Vec<MultiAddr>)>> {
        self.closest_peers(target, PUT_TARGET_WIDTH).await
    }

    /// Return the requested number of closest peers for a target address.
    ///
    /// Queries the DHT for peers by XOR distance. Returns each peer
    /// paired with its known network addresses.
    pub(crate) async fn closest_peers(
        &self,
        target: &XorName,
        count: usize,
    ) -> Result<Vec<(PeerId, Vec<MultiAddr>)>> {
        let peers = self.network().find_closest_peers(target, count).await?;

        if peers.is_empty() {
            return Err(Error::InsufficientPeers(
                "DHT returned no peers for target address".to_string(),
            ));
        }
        Ok(peers)
    }
}

/// Persist the adaptive snapshot when the `Client` is dropped, so any
/// caller — CLI, daemon, library user, integration test — gets
/// warm-start carry-over for free without remembering to call
/// `save_adaptive_snapshot()` explicitly. Best effort, sync `std::fs`,
/// no panic risk on a poisoned mutex (the inner helper handles it).
///
/// We deliberately write SYNCHRONOUSLY (not via `spawn_blocking`)
/// because Drop runs during process shutdown / runtime teardown,
/// when fire-and-forget background tasks can be dropped before they
/// complete and the snapshot is silently lost. A small synchronous
/// stall on a tokio worker (typically <1ms for a local-disk JSON
/// write of ~50 bytes) is the right tradeoff for guaranteed
/// persistence — BOUNDED by `DROP_SAVE_TIMEOUT` so a stalled
/// network-mounted data dir cannot block process shutdown.
#[cfg(feature = "native")]
const DROP_SAVE_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(500);

#[cfg(feature = "native")]
impl Drop for Client {
    fn drop(&mut self) {
        let Some(path) = self.persist_path.clone() else {
            return;
        };
        let snap = self.controller.snapshot();
        adaptive::save_snapshot_with_timeout(path, snap, DROP_SAVE_TIMEOUT);
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use super::*;

    /// Cover EVERY variant of `data::error::Error`. Build an instance of
    /// each, classify it, and assert the resulting `Outcome` matches the
    /// only sensible mapping. If a future commit adds a new error variant
    /// without updating `classify_error`, this test fails to ensure the
    /// adaptive controller always sees correct capacity signals.
    ///
    /// Mapping policy (mirrors `classify_error` doc):
    /// - `Timeout` -> `Outcome::Timeout`
    /// - `Network`, `InsufficientPeers`, `Io`, `Protocol`, `Storage`,
    ///   `PartialUpload` -> `Outcome::NetworkError` (transport-related
    ///   or literal capacity failure)
    /// - everything else -> `Outcome::ApplicationError` (would happen
    ///   on a perfectly healthy network)
    #[test]
    fn classify_error_covers_all_variants() {
        let cases: Vec<(Error, Outcome)> = vec![
            (Error::Timeout("t".to_string()), Outcome::Timeout),
            (Error::Network("n".to_string()), Outcome::NetworkError),
            (
                Error::InsufficientPeers("p".to_string()),
                Outcome::NetworkError,
            ),
            (Error::Storage("s".to_string()), Outcome::NetworkError),
            (Error::Payment("p".to_string()), Outcome::ApplicationError),
            (Error::Protocol("p".to_string()), Outcome::NetworkError),
            (
                Error::InvalidData("d".to_string()),
                Outcome::ApplicationError,
            ),
            // A definitively absent record over a working link — the peers
            // answered, nothing was stored there. Must NOT register as a
            // capacity signal.
            (
                Error::NotFound("missing".to_string()),
                Outcome::ApplicationError,
            ),
            (
                Error::Serialization("s".to_string()),
                Outcome::ApplicationError,
            ),
            (Error::Crypto("c".to_string()), Outcome::ApplicationError),
            (
                Error::Io(std::io::Error::other("io")),
                Outcome::NetworkError,
            ),
            (Error::Config("c".to_string()), Outcome::ApplicationError),
            (
                Error::SignatureVerification("s".to_string()),
                Outcome::ApplicationError,
            ),
            (
                Error::Encryption("e".to_string()),
                Outcome::ApplicationError,
            ),
            (Error::AlreadyStored, Outcome::ApplicationError),
            (
                Error::InsufficientDiskSpace("d".to_string()),
                Outcome::ApplicationError,
            ),
            (
                Error::CostEstimationInconclusive("c".to_string()),
                Outcome::ApplicationError,
            ),
            (
                Error::PartialUpload {
                    stored: vec![],
                    stored_count: 0,
                    failed: vec![],
                    failed_count: 0,
                    total_chunks: 0,
                    spend: Box::new(crate::data::error::PartialUploadSpend {
                        storage_cost_atto: "0".to_string(),
                        gas_cost_wei: 0,
                    }),
                    reason: "r".to_string(),
                },
                Outcome::NetworkError,
            ),
            (
                Error::BadQuoteBinding {
                    peer_id: "peer".to_string(),
                    detail: "mismatch".to_string(),
                },
                Outcome::ApplicationError,
            ),
            // A remote application rejection: the node responded with a
            // structured `ProtocolError`, so the transport succeeded and
            // this must NOT register as a capacity signal (V2-468).
            (
                Error::RemotePut {
                    address: "abcd".to_string(),
                    source: ant_protocol::ProtocolError::PaymentFailed("stale quote".to_string()),
                },
                Outcome::ApplicationError,
            ),
            // A close-group quorum shortfall caused by dial/relay churn with
            // no PUT-response timeouts — remote peer churn, not local
            // backpressure, so it must NOT register as a capacity signal
            // (V2-554). A timeout-bearing shortfall keeps `InsufficientPeers`.
            (
                Error::CloseGroupShortfall("Stored on 3 peers, need 4".to_string()),
                Outcome::ApplicationError,
            ),
            // Refusing an oversized external-signer merkle batch happens
            // before any network work, so it is not a capacity signal.
            (
                Error::MerkleBatchTooLarge {
                    addresses: 257,
                    max_leaves: 256,
                },
                Outcome::ApplicationError,
            ),
        ];
        for (err, expected) in &cases {
            let got = classify_error(err);
            assert_eq!(
                got, *expected,
                "classify_error({err:?}) = {got:?}, expected {expected:?}",
            );
        }
    }

    /// C4 fix guard: pinning the legacy `quote_concurrency` /
    /// `store_concurrency` ClientConfig fields must clamp ONLY the
    /// matching channel's max in the resulting controller. The fetch
    /// (download) channel must keep its full default ceiling.
    #[test]
    fn legacy_concurrency_pin_does_not_bleed_across_channels() {
        let cfg = ClientConfig {
            quote_concurrency: 4,
            store_concurrency: 2,
            ..ClientConfig::default()
        };
        let (controller, _) = build_controller(&cfg);
        // The store/quote caps must be clamped to the user's pin.
        assert_eq!(controller.config.max.quote, 4, "quote pin not respected");
        assert_eq!(controller.config.max.store, 2, "store pin not respected");
        // The fetch cap must NOT have been lowered — that's the
        // regression C4 was about.
        let default_fetch_max = adaptive::ChannelMax::default().fetch;
        assert_eq!(
            controller.config.max.fetch, default_fetch_max,
            "fetch cap was lowered by store/quote pin (C4 regression)"
        );
        // Cold-start values must respect the lowered ceilings.
        assert!(
            controller.quote.current() <= 4,
            "quote start exceeds its cap"
        );
        assert!(
            controller.store.current() <= 2,
            "store start exceeds its cap"
        );
    }

    /// Default ClientConfig must NOT silently lower the controller's
    /// per-channel ceilings — the adaptive defaults give every channel
    /// real headroom to grow. This guards against future commits
    /// re-introducing a global clamp.
    #[test]
    fn default_client_config_does_not_clamp_controller_max() {
        let cfg = ClientConfig::default();
        let (controller, _) = build_controller(&cfg);
        let defaults = adaptive::ChannelMax::default();
        // The legacy fields default to 32/8 (the prior static knobs),
        // both of which are <= the per-channel adaptive defaults
        // (128/64). build_controller must keep the larger, not clobber
        // with the legacy values.
        assert_eq!(controller.config.max.quote, defaults.quote);
        assert_eq!(controller.config.max.store, defaults.store);
        assert_eq!(controller.config.max.fetch, defaults.fetch);
        // Compile-time-ish guard: if a new variant is added to Error,
        // this match forces an update here.
        let _ = |e: &Error| match e {
            Error::Timeout(_)
            | Error::Network(_)
            | Error::InsufficientPeers(_)
            | Error::Storage(_)
            | Error::Payment(_)
            | Error::Protocol(_)
            | Error::InvalidData(_)
            | Error::NotFound(_)
            | Error::Serialization(_)
            | Error::Crypto(_)
            | Error::Io(_)
            | Error::Config(_)
            | Error::SignatureVerification(_)
            | Error::Encryption(_)
            | Error::AlreadyStored
            | Error::InsufficientDiskSpace(_)
            | Error::CostEstimationInconclusive(_)
            | Error::Cancelled(_)
            | Error::PartialUpload { .. }
            | Error::BadQuoteBinding { .. }
            | Error::BadQuoteCommitment { .. }
            | Error::MerkleBatchTooLarge { .. }
            | Error::RemotePut { .. }
            | Error::ClientUpdateRequired(_)
            | Error::StorerUpdateRequired(_)
            | Error::CloseGroupShortfall(_) => (),
        };
    }
}