ant-core 0.4.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
//! On-disk cache for merkle batch payment receipts.
//!
//! Why this exists
//! ---------------
//! A merkle batch upload pays for *all* chunks in one on-chain transaction
//! up-front, then stores each chunk to its close-group. If the store phase
//! fails partway through (network flake, slow close-K, client crash), the
//! on-chain payment is gone but the proofs needed to re-attempt the store
//! are lost too — the user has to pay again from scratch.
//!
//! By persisting the [`MerkleBatchPaymentResult`] to disk **immediately after
//! the on-chain payment lands**, the next invocation can resume the upload
//! using the already-paid proofs instead of re-paying. The cache is keyed by
//! a derivation of the source file path so the same upload, re-issued for
//! the same file, transparently picks up where it left off.
//!
//! Lifecycle
//! ---------
//! * **save** — called once per upload, right after the merkle batch payment
//!   transaction confirms. Writes JSON to
//!   `<data_dir>/payments/<timestamp>_<file_hash>`.
//! * **load_for_file** — called at the top of every merkle upload. If a
//!   non-expired cached receipt exists for the file, it is returned so the
//!   upload can skip the pay phase and go straight to store.
//! * **delete_for_file** — called after a fully successful upload to remove
//!   the receipt so a future re-upload of the same path pays anew.
//! * **cleanup_outdated** — called opportunistically inside `load_for_file`
//!   to garbage-collect receipts past the 7-day expiry window.
//!
//! Filename format
//! ---------------
//! `<timestamp>_<file_hash>` where:
//! * `timestamp` is the merkle payment timestamp (seconds since epoch) used
//!   on-chain. Expiry is computed from this value so we can prune stale
//!   receipts even if their on-disk mtime has been touched.
//! * `file_hash` is the SHA-256 of the source file path string, truncated
//!   to keep filenames short. Same-name uploads from different directories
//!   collide deliberately — the user can name their file uniquely if they
//!   need parallel uploads.
//!
//! Failure-mode tolerance
//! ----------------------
//! All errors in this module are logged and swallowed in the public-facing
//! API (`try_load_for_file`, `try_save`, `try_delete_for_file`): a busted
//! cache directory must never prevent a real upload from running. The
//! tradeoff is that a corrupt cache file is silently treated as "no
//! cache", forcing the user to re-pay — but never causes data loss.

use crate::config;
use crate::data::client::merkle::MerkleBatchPaymentResult;
use crate::error::Result;
use ant_protocol::payment::{deserialize_merkle_proof, serialize_merkle_proof};
use std::fs::{self, DirEntry, File, OpenOptions};
use std::hash::{Hash, Hasher};
use std::io::{BufReader, BufWriter};
use std::path::{Path, PathBuf};
use std::time::{SystemTime, UNIX_EPOCH};
use tracing::{debug, info, warn};

/// Cached merkle receipts older than this are removed from disk.
///
/// Set to match `MERKLE_PAYMENT_EXPIRATION` in `evmlib` (7 days). After
/// the payment ages out on-chain there is no point keeping the cache —
/// the proofs can no longer be verified by storers.
const PAYMENT_EXPIRATION_SECS: u64 = 7 * 24 * 60 * 60;

/// Subdirectory under the platform-appropriate data dir.
const PAYMENTS_SUBDIR: &str = "payments";

/// Returns the directory used for cached payments, creating it if needed.
fn payments_dir() -> Result<PathBuf> {
    let dir = config::data_dir()?.join(PAYMENTS_SUBDIR);
    fs::create_dir_all(&dir)?;
    Ok(dir)
}

/// Short non-cryptographic hash of the source file path string, used as
/// the on-disk cache key.
///
/// Filename collisions are not a correctness problem (the loaded
/// receipt is content-validated against the current encrypted chunk
/// addresses before being trusted) but they would waste a re-pay, so
/// we want low collision probability across a single user's upload
/// history. `std::hash::DefaultHasher` with 16 hex chars of output is
/// far below the collision threshold for that scale.
fn file_hash_key(file_path: &str) -> String {
    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    file_path.hash(&mut hasher);
    format!("{:016x}", hasher.finish())
}

/// Save the merkle batch payment receipt for a given source file path.
///
/// Idempotent: re-saving for the same `(timestamp, file_path)` overwrites
/// the previous file. Different timestamps for the same file produce
/// different filenames, which is fine — `cleanup_outdated` reaps them.
pub fn save(file_path: &str, result: &MerkleBatchPaymentResult) -> Result<PathBuf> {
    let dir = payments_dir()?;
    let ts = if result.merkle_payment_timestamp > 0 {
        result.merkle_payment_timestamp
    } else {
        // Fall back to now() if the result wasn't populated. Should not
        // happen in practice — every constructor stamps this field —
        // but defensively avoid emitting a `0_*` filename that would
        // immediately be treated as expired.
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_secs())
            .unwrap_or(0)
    };
    let path = dir.join(format!("{ts}_{}", file_hash_key(file_path)));
    let handle = File::create(&path)?;
    // msgpack (rmp-serde) rather than JSON because `proofs` is keyed by
    // `[u8; 32]` which JSON cannot represent as a map key.
    rmp_serde::encode::write(&mut BufWriter::new(handle), result)
        .map_err(|e| crate::error::Error::Io(std::io::Error::other(e.to_string())))?;
    debug!(
        "Cached merkle payment receipt for {file_path:?} to {}",
        path.display()
    );
    Ok(path)
}

/// Best-effort save. Logs on failure but never returns an error.
///
/// Intended for the upload path: if we can't cache the receipt we still
/// want to attempt the chunk PUTs.
pub fn try_save(file_path: &str, result: &MerkleBatchPaymentResult) {
    if let Err(e) = save(file_path, result) {
        warn!(
            "Failed to cache merkle payment receipt for {file_path:?}: {e}. \
             Upload will proceed without resume support."
        );
    }
}

/// Load the cached merkle batch receipt for a given source file path.
///
/// Side-effect: opportunistically removes any expired receipts found in
/// the directory while scanning.
///
/// Returns `Ok(None)` if no matching non-expired receipt is found.
pub fn load_for_file(file_path: &str) -> Result<Option<(PathBuf, MerkleBatchPaymentResult)>> {
    cleanup_outdated();
    let dir = payments_dir()?;
    let key = file_hash_key(file_path);

    let read_dir = match fs::read_dir(&dir) {
        Ok(rd) => rd,
        Err(e) => {
            debug!("Could not read payments dir {}: {e}", dir.display());
            return Ok(None);
        }
    };

    for entry in read_dir.flatten() {
        let path = entry.path();
        if !path.is_file() {
            continue;
        }
        let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
            continue;
        };
        if !name.contains(&key) {
            continue;
        }
        if is_expired_filename(name) {
            // Found the file but it has aged out; cleanup will
            // collect it. Keep scanning in case a newer one exists.
            continue;
        }
        match read_receipt(&path) {
            Ok(receipt) => {
                info!(
                    "Found previous merkle upload attempt for {file_path}, \
                     resuming with payment cached at {}",
                    path.display()
                );
                return Ok(Some((path, receipt)));
            }
            Err(e) => {
                warn!(
                    "Cached merkle receipt at {} is unreadable ({e}). \
                     Ignoring and starting a fresh upload.",
                    path.display()
                );
            }
        }
    }
    Ok(None)
}

/// Best-effort load. Logs on failure and returns `None`.
pub fn try_load_for_file(file_path: &str) -> Option<(PathBuf, MerkleBatchPaymentResult)> {
    match load_for_file(file_path) {
        Ok(opt) => opt,
        Err(e) => {
            warn!(
                "Failed to look up cached merkle receipt for {file_path:?}: {e}. \
                 Starting a fresh upload."
            );
            None
        }
    }
}

/// Delete the cached receipt(s) matching the file path. Called on
/// successful upload completion.
pub fn delete_for_file(file_path: &str) -> Result<()> {
    let dir = payments_dir()?;
    let key = file_hash_key(file_path);
    if let Ok(read_dir) = fs::read_dir(&dir) {
        for entry in read_dir.flatten() {
            let path = entry.path();
            if let Some(name) = path.file_name().and_then(|n| n.to_str()) {
                if name.contains(&key) {
                    let _ = fs::remove_file(&path);
                    debug!("Deleted cached merkle receipt {}", path.display());
                }
            }
        }
    }
    Ok(())
}

/// Best-effort delete. Logs on failure but never returns an error.
pub fn try_delete_for_file(file_path: &str) {
    if let Err(e) = delete_for_file(file_path) {
        warn!(
            "Failed to delete cached merkle receipt for {file_path:?}: {e}. \
             Will be cleaned up after expiry."
        );
    }
}

/// Garbage-collect cached receipts past the expiry window.
///
/// Logs each removal at info level so users see what we cleaned up.
/// Best-effort: any IO error is silently ignored.
pub fn cleanup_outdated() {
    let Ok(dir) = payments_dir() else {
        return;
    };
    let Ok(read_dir) = fs::read_dir(&dir) else {
        return;
    };
    for entry in read_dir.flatten() {
        if is_expired_entry(&entry) {
            let path = entry.path();
            info!(
                "Removing expired cached merkle payment file: {}",
                path.display()
            );
            let _ = fs::remove_file(path);
        }
    }
}

fn is_expired_entry(entry: &DirEntry) -> bool {
    let path = entry.path();
    if !path.is_file() {
        return false;
    }
    let Some(name) = path.file_name().and_then(|n| n.to_str()) else {
        return false;
    };
    is_expired_filename(name)
}

fn is_expired_filename(name: &str) -> bool {
    let ts_str = match name.split_once('_') {
        Some((ts, _)) => ts,
        None => return false,
    };
    let Ok(ts) = ts_str.parse::<u64>() else {
        return false;
    };
    let now = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0);
    now > ts.saturating_add(PAYMENT_EXPIRATION_SECS)
}

fn read_receipt(path: &Path) -> Result<MerkleBatchPaymentResult> {
    let handle = File::open(path)?;
    let mut receipt: MerkleBatchPaymentResult =
        rmp_serde::decode::from_read(BufReader::new(handle))
            .map_err(|e| crate::error::Error::Io(std::io::Error::other(e.to_string())))?;

    if strip_commitment_sidecars(&mut receipt) {
        info!(
            "Stripped legacy commitment sidecars from cached merkle receipt at {}",
            path.display()
        );
        // Best-effort write-back so the strip happens once; a failure here
        // only means we re-strip on the next load.
        if let Err(e) = overwrite_receipt(path, &receipt) {
            warn!(
                "Failed to persist slimmed merkle receipt at {}: {e}",
                path.display()
            );
        }
    }

    Ok(receipt)
}

/// Strip ADR-0004 commitment sidecars from every proof in a cached receipt.
///
/// Receipts saved by clients built before sidecars were dropped from the
/// per-chunk merkle proofs carry all 16 winner-pool sidecars in EVERY proof
/// (~214 KB per proof), which pushed the proof past the storer's
/// payment-proof size cap — resuming with them would replay the exact
/// failure the slim proofs fixed. Stripping is always safe: the pool hash
/// and address branch stay exactly as paid on-chain, and storers resolve
/// commitment pins from gossip or a `GetCommitmentByPin` fetch. Returns
/// whether anything was stripped.
fn strip_commitment_sidecars(receipt: &mut MerkleBatchPaymentResult) -> bool {
    let mut stripped = false;
    for proof_bytes in receipt.proofs.values_mut() {
        // Non-merkle or unreadable proof bytes are left untouched; the
        // storer remains the judge of those.
        let Ok(mut proof) = deserialize_merkle_proof(proof_bytes) else {
            continue;
        };
        if proof.commitment_sidecars.is_empty() {
            continue;
        }
        proof.commitment_sidecars.clear();
        match serialize_merkle_proof(&proof) {
            Ok(slim) => {
                *proof_bytes = slim;
                stripped = true;
            }
            Err(e) => warn!("Failed to re-serialize slimmed cached merkle proof: {e}"),
        }
    }
    stripped
}

/// Overwrite a cached receipt via `tmp + fsync + rename` (same canonical
/// path), mirroring `cached_single::write_receipt_atomic`: an interrupted
/// write must never truncate the only copy of a paid receipt — losing it
/// forces the user to re-pay. The tmp name carries pid + nanos and is opened
/// with `create_new`, so concurrent migrations (threads or processes) can
/// never share a tmp inode — a residual name collision fails this best-effort
/// write instead of corrupting it, and the next load simply re-strips. A
/// leftover tmp from a crash is harmless (the canonical stays intact until
/// rename) and ages out with the same `<ts>_` filename prefix.
fn overwrite_receipt(path: &Path, receipt: &MerkleBatchPaymentResult) -> Result<()> {
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    let pid = std::process::id();
    let tmp_path = path.with_extension(format!("{pid}-{nanos}.tmp"));
    {
        let handle = OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(&tmp_path)?;
        let mut writer = BufWriter::new(handle);
        if let Err(e) = rmp_serde::encode::write(&mut writer, receipt) {
            let _ = fs::remove_file(&tmp_path);
            return Err(crate::error::Error::Io(std::io::Error::other(
                e.to_string(),
            )));
        }
        // `into_inner` flushes; a swallowed flush error here would defeat
        // the atomicity, so surface it.
        let handle = match writer.into_inner() {
            Ok(handle) => handle,
            Err(e) => {
                let _ = fs::remove_file(&tmp_path);
                return Err(crate::error::Error::Io(std::io::Error::other(format!(
                    "BufWriter flush failed: {e}"
                ))));
            }
        };
        if let Err(e) = handle.sync_all() {
            let _ = fs::remove_file(&tmp_path);
            return Err(e.into());
        }
    }
    if let Err(e) = fs::rename(&tmp_path, path) {
        let _ = fs::remove_file(&tmp_path);
        return Err(e.into());
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::collections::HashMap;

    fn dummy_receipt(ts: u64) -> MerkleBatchPaymentResult {
        let mut proofs: HashMap<[u8; 32], Vec<u8>> = HashMap::new();
        proofs.insert([0u8; 32], vec![1, 2, 3]);
        MerkleBatchPaymentResult {
            proofs,
            chunk_count: 1,
            storage_cost_atto: "0".to_string(),
            gas_cost_wei: 0,
            merkle_payment_timestamp: ts,
        }
    }

    /// A serialized merkle proof carrying legacy commitment sidecars, as saved
    /// by clients built before sidecars were dropped from per-chunk proofs.
    fn fat_merkle_proof_bytes(ts: u64) -> Vec<u8> {
        use ant_protocol::evm::{
            Amount, MerklePaymentCandidateNode, MerklePaymentCandidatePool, MerklePaymentProof,
            MerkleTree, RewardsAddress, CANDIDATES_PER_POOL,
        };
        use xor_name::XorName;

        let xornames: Vec<XorName> = (0..4u8).map(|i| XorName([i; 32])).collect();
        let tree = MerkleTree::from_xornames(xornames.clone()).unwrap();
        let midpoint = tree.reward_candidates(ts).unwrap().remove(0);
        let candidate_nodes: [MerklePaymentCandidateNode; CANDIDATES_PER_POOL] =
            std::array::from_fn(|i| MerklePaymentCandidateNode {
                pub_key: vec![i as u8; 32],
                price: Amount::from(1024u64),
                reward_address: RewardsAddress::new([i as u8; 20]),
                merkle_payment_timestamp: ts,
                signature: vec![i as u8; 64],
                committed_key_count: 9_000,
                commitment_pin: Some([7u8; 32]),
            });
        let pool = MerklePaymentCandidatePool {
            midpoint_proof: midpoint,
            candidate_nodes,
        };
        let address_proof = tree.generate_address_proof(0, xornames[0]).unwrap();
        let mut proof = MerklePaymentProof::new(xornames[0], address_proof, pool);
        proof.commitment_sidecars = vec![vec![0xAB; 5_000]; CANDIDATES_PER_POOL];
        serialize_merkle_proof(&proof).unwrap()
    }

    /// DEV-01 recovery: a receipt cached by a pre-fix client carries proofs
    /// with all 16 commitment sidecars (~342 KB each on the wire) — resuming
    /// with them would replay the storer's size rejection. Loading must strip
    /// the sidecars while leaving the paid pool and address branch intact.
    #[test]
    fn strip_removes_legacy_sidecars_and_is_idempotent() {
        let ts = 1_000_000;
        let fat = fat_merkle_proof_bytes(ts);
        let mut proofs: HashMap<[u8; 32], Vec<u8>> = HashMap::new();
        proofs.insert([0u8; 32], fat.clone());
        // A non-merkle blob must pass through untouched.
        proofs.insert([1u8; 32], vec![1, 2, 3]);
        let mut receipt = MerkleBatchPaymentResult {
            proofs,
            chunk_count: 2,
            storage_cost_atto: "0".to_string(),
            gas_cost_wei: 0,
            merkle_payment_timestamp: ts,
        };

        assert!(strip_commitment_sidecars(&mut receipt));

        let slim = receipt.proofs.get(&[0u8; 32]).unwrap();
        assert!(slim.len() < fat.len(), "stripped proof must shrink");
        let proof = deserialize_merkle_proof(slim).unwrap();
        assert!(proof.commitment_sidecars.is_empty());
        assert_eq!(receipt.proofs.get(&[1u8; 32]).unwrap(), &vec![1, 2, 3]);

        // Second pass finds nothing left to strip.
        assert!(!strip_commitment_sidecars(&mut receipt));
    }

    /// The strip write-back must replace the canonical receipt atomically:
    /// new content lands, and no `.tmp` sibling survives a successful write.
    #[test]
    fn overwrite_receipt_is_atomic_and_leaves_no_tmp() -> Result<()> {
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_nanos();
        let dir = std::env::temp_dir().join(format!("anselme-merkle-overwrite-test-{nanos}"));
        fs::create_dir_all(&dir)?;
        let path = dir.join("123_abcd");
        fs::write(&path, b"pre-fix receipt bytes")?;

        overwrite_receipt(&path, &dummy_receipt(42))?;

        let reloaded = read_receipt(&path)?;
        assert_eq!(reloaded.merkle_payment_timestamp, 42);
        let leftover_tmps = fs::read_dir(&dir)?
            .flatten()
            .filter(|e| {
                e.path()
                    .extension()
                    .is_some_and(|ext| ext.eq_ignore_ascii_case("tmp"))
            })
            .count();
        assert_eq!(leftover_tmps, 0, "no tmp sibling may survive");

        fs::remove_dir_all(&dir).ok();
        Ok(())
    }

    #[test]
    fn file_hash_key_is_stable() {
        let a = file_hash_key("/tmp/some/file.bin");
        let b = file_hash_key("/tmp/some/file.bin");
        assert_eq!(a, b);
        let c = file_hash_key("/tmp/some/other.bin");
        assert_ne!(a, c);
    }

    #[test]
    fn expired_filename_detected() {
        // Just past the expiry boundary.
        let stale = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs()
            .saturating_sub(PAYMENT_EXPIRATION_SECS + 60);
        let name = format!("{stale}_abcd1234");
        assert!(is_expired_filename(&name));

        // Within the window.
        let fresh = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs()
            .saturating_sub(60);
        let name = format!("{fresh}_abcd1234");
        assert!(!is_expired_filename(&name));
    }

    #[test]
    fn malformed_filename_is_not_expired() {
        // Defensive: garbage in payments dir must not be auto-deleted.
        assert!(!is_expired_filename("nonsense"));
        assert!(!is_expired_filename("not_a_number_abcd1234"));
    }

    #[test]
    fn roundtrip_save_load_delete() -> Result<()> {
        let file_path = format!(
            "/tmp/anselme-resumable-merkle-test-{}",
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        );
        let ts = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs();
        let receipt = dummy_receipt(ts);
        let saved_path = save(&file_path, &receipt)?;
        assert!(saved_path.exists());

        let loaded = load_for_file(&file_path)?;
        let (loaded_path, loaded_receipt) = loaded.expect("receipt should be loadable");
        assert_eq!(loaded_path, saved_path);
        assert_eq!(loaded_receipt.chunk_count, receipt.chunk_count);
        assert_eq!(loaded_receipt.merkle_payment_timestamp, ts);

        delete_for_file(&file_path)?;
        assert!(load_for_file(&file_path)?.is_none());
        Ok(())
    }
}