alkblobs 0.2.1

Content-addressed blob storage: multi-hash put/get/verify store with pluggable backends, transport- and protocol-agnostic
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
//! The Backend trait contract (ADR-003): opaque byte keys, byte-valued
//! entries, complete `list()`, and the I/O seams every engine implements —
//! the read cursor and the staged put (ADR-010 §1).
//!
//! **The crate's second stable seam, crate-internal (ADR-003; the visibility
//! design point).** The trait contract is stable — engine implementations
//! never guess it — but it does not leave the crate: every engine ships
//! in-crate behind a feature flag and no consumer ever implements a
//! Backend (ADR-004's boundary; backends-and-dispatch.md §What the Backend
//! trait is). Exporting it later is an additive change with no persistence
//! or wire consequence, so `pub(crate)` costs nothing now and keeps the
//! public surface the facade's alone. The module itself is public so this
//! contract is documented.
//!
//! **One trait, two I/O abstractions.** There is no per-tier trait split
//! (ADR-010 §1): every engine implements the same method set, and the two
//! tiers differ only in the bodies behind the crate-internal abstractions —
//! a **read cursor** whose kv bodies materialize the whole value (bounded
//! by tier policy) and whose large bodies stream (`local` pread loop,
//! pg-lo `lo_get` windows), and a **staging cursor** for the staged put.
//! The store core programs the abstractions, never the engine shape. Both
//! body families serve the ADR-006 range contract — a `read_range` returns
//! the slice plus a SHA-256 slice digest, fixed by convention (the
//! `slice_digest` function below) — at one seam.
//!
//! **Engine posture:** namespace-blind (ADR-005) and virgin-store reads are
//! no-ops (missing table = absent/empty; POC #1 finding 2). Verification
//! never rides the trait: engines store opaque bytes under opaque keys;
//! the canonical derivation and its hash-check are store-core concerns
//! (ADR-002/006). GC state never rides this trait either (ADR-010 §2) —
//! the crate-internal engine-state companion seam lives in [`crate::gc`].
//!
//! The `#![allow(dead_code)]` below still has real scope: `Backend`
//! trait items with no in-tree lib caller (`Backend::name` — the
//! engines' reporting identity, consumed today by the engines' and
//! the seam tests) plus the wave-1 seam tests' doubles. The kv
//! engines' whole-value `put` gained its in-tree callers in wave 3
//! (sqlite is the default build's kv engine and the facade's put seam
//! drives the kv tier through it); the allow shrinks as the remaining
//! callers land.

#![allow(dead_code)]

use sha2::{Digest as _, Sha256};

use crate::Error;

/// The slice-digest convention (ADR-006): plain SHA-256 over the slice
/// bytes, fixed by convention so callers and ops handlers agree without
/// negotiation. Per-range verification against the canonical digest is
/// impossible under the preamble (the length is inside the hashed input,
/// ADR-002); transport-integrity is all a slice digest carries.
///
/// Crate-internal: this function is the convention's one seam-side
/// definition — the read cursor's range reads and the staged put's
/// verification pass both digest through it, and the facade's
/// `read_range` re-exports the result behind it. Callers verify
/// out-of-band by SHA-256-hashing the slice bytes themselves; there is
/// nothing to negotiate.
pub(crate) fn slice_digest(bytes: &[u8]) -> [u8; 32] {
    Sha256::digest(bytes).into()
}

/// The read cursor — the abstraction [`Backend::get`] returns, driven by
/// the store core (ADR-010 §1). One abstraction, two body families: kv
/// bodies materialize the whole value (bounded by tier policy; `len` reads
/// the row's size column); large bodies stream (`local`: a pread loop over
/// the committed file — immutable once committed; pg-lo: descriptorless
/// `lo_get(oid, off, len)` windows over companion rows). Both serve the
/// ADR-006 range contract at one seam: [`ReadCursor::read_range`] is the
/// slice + slice-digest return the facade's `read_range` sits behind.
///
/// Crate-internal — not public API: store consumers see the facade's
/// `(len, stream)` shapes with this cursor behind them (store-api.md;
/// ADR-010 §1). The store's stream utility adapts engine cursor bodies; it
/// never exposes them.
pub(crate) trait ReadCursor: Send {
    /// The entry's length in bytes — a metadata-only probe, never a
    /// content load (ADR-008's size discipline): the row's size column on
    /// kv bodies, a file stat or companion-row read on large bodies. For
    /// a staging cursor ([`StagingCursor`]), the staged length so far —
    /// staging is append-only, so it grows with each append.
    fn len(&self) -> u64;

    /// The body's byte-level read: the bytes at `[offset, offset + len)`
    /// clamped to the entry's end — a pread-style short read at EOF.
    /// `offset >= len()` (EOF) yields an empty vec, never an error;
    /// callers wanting strictness check against [`ReadCursor::len`]. The
    /// only error family member this raises is media failure
    /// ([`Error::Io`]).
    async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error>;

    /// The range-read seam both body families serve (ADR-006): the slice
    /// plus its slice digest — [`slice_digest`]'s SHA-256 over exactly
    /// the returned bytes, fixed by convention. The provided body
    /// composes `read_at` with the convention; a body override is a
    /// per-engine optimization and must keep the digest over exactly the
    /// returned bytes.
    async fn read_range(&self, offset: u64, len: u64) -> Result<(Vec<u8>, [u8; 32]), Error> {
        let bytes = self.read_at(offset, len).await?;
        let digest = slice_digest(&bytes);
        Ok((bytes, digest))
    }
}

/// The staged put's staging cursor (ADR-010 §1) — the named per-engine
/// contract of the large tier: the engine hands one out from
/// [`Backend::put_staged`]; content is appended; the store's verification
/// pass reads the eventual stored artifact back as a whole through the
/// same read seam ([`ReadCursor`] is the supertrait — ADR-003's
/// buffered-put ruling: the second derivation pass runs over the complete
/// staged bytes); commit publishes the entry. Every failure path converges
/// on discard with zero residue (the stage-hygiene invariant,
/// store-api.md §Write path; exact-count tested per engine).
///
/// **The key resolves at commit, not at handout (named design point,
/// the wave-2 amendment of the wave-1 handout shape).** The canonical
/// key is the derivation over preamble + content, and it completes at
/// stream end — after the staging cursor already exists (put paths
/// stream content once into the cursor while hashing; ADR-003's put
/// paths). So `commit` carries the store-core-resolved key and the
/// stage's identity before that point is only the engine-internal
/// stage file/tx (unique per `put_staged` handout). Publish is the
/// CAS first-writer-wins entry under that committed key exactly as
/// [`Backend::put`].
///
/// Every engine implements the staged form: kv-tier bodies are the
/// degenerate in-memory stage committed through the engine's whole-value
/// row write (the store core selects the form by tier policy, ADR-010
/// §1 — but the conformance suite runs stage hygiene on every engine,
/// store-api.md invariant 7).
pub(crate) trait StagingCursor: ReadCursor {
    /// Append a chunk to the stage. Unknown-length put streams append as
    /// content flows; the unknown-length flow's buffered prefix flushes
    /// into a fresh stage on threshold overflow, and the stream continues
    /// appending into it (ADR-003's buffered-put semantics — there is no
    /// "restart the stream" requirement anywhere).
    async fn append(&mut self, chunk: &[u8]) -> Result<(), Error>;

    /// Publish the entry: the stage's bytes become the CAS entry under
    /// `key` — the store-core-resolved canonical key (the derivation
    /// completes at stream end; see the type's design point) — in the
    /// engine's put shape (commit-rename on `local`, the in-tx `lowrite`
    /// batch on pg-lo, the row write on kv bodies). CAS first-writer-wins:
    /// a losing race against an existing entry is dedup (`Ok`), never a
    /// failure. This is the staged form's terminal step — the store core
    /// never re-uses the cursor after the call. On failure the stage is
    /// **already discarded** — zero residue by the stage-hygiene
    /// invariant; there is no post-failure discard to drive (the
    /// consuming call took the cursor).
    async fn commit(self, key: &[u8]) -> Result<(), Error>;

    /// Discard the stage: every non-commit path converges here, zero
    /// residue. Driven explicitly by the store core (stream error,
    /// verification failure, dispatch abandonment). The store core never
    /// drops a staging cursor without driving it to commit or discard;
    /// engine Drop bodies are best-effort residue cleanup, an engine-task
    /// concern (waves 4-5 own their blocking discipline there).
    async fn discard(self);
}

/// The engine contract — the crate's second stable seam (ADR-003),
/// crate-internal per the module docs' visibility design point.
///
/// **Opaque byte keys, opaque byte values.** Typed `Key` converts at the
/// store boundary only (ADR-002); backends must survive digest-layer
/// evolution because nothing here knows what a digest is — a kv row, a
/// `local` filename, or an LO companion row has no schema-migration
/// story, so self-describing bytes are the durable choice (POC #1
/// finding 1's generic-over-`K` alternative is rejected).
///
/// **One trait, every engine, same method set** (ADR-010 §1 — no per-tier
/// trait split). The tiers differ only in the bodies behind
/// [`Self::Cursor`] and [`Self::Staging`]. `list` is complete *and*
/// well-formed by contract (ADR-003): a malformed list is as deadly as an
/// incomplete one and is silent until a sweep deletes the wrong things,
/// so list correctness is gate-tested per engine through exact-count
/// sweep outcomes (store-api.md invariants 5/7). Namespace-blind, and
/// virgin-store reads are no-ops. GC state never rides this trait
/// (ADR-010 §2) — the store core reaches the fleet machinery through the
/// crate-internal engine-state companion seam in [`crate::gc`].
pub(crate) trait Backend: Send + Sync {
    /// This engine's read-cursor body (kv: materializing; large: streaming).
    type Cursor: ReadCursor;

    /// This engine's staging-cursor body.
    type Staging: StagingCursor;

    /// The engine's name — the dispatch/conformance reporting identity.
    fn name(&self) -> &'static str;

    /// The pure existence probe (ADR-008's `has` discipline). Virgin
    /// store: `false` — reads are no-ops, never "table does not exist"
    /// errors.
    async fn has(&self, key: &[u8]) -> Result<bool, Error>;

    /// The entry's read cursor, or `None` when absent — absence is a
    /// value on this path, never `Error::Missing` (ADR-012 §6.5; the
    /// facade's `get` sits directly behind). Virgin store: `None`.
    async fn get(&self, key: &[u8]) -> Result<Option<Self::Cursor>, Error>;

    /// The whole-value put — the kv tier's settled form (ADR-010 §1): a
    /// **CAS write, first writer wins** (`INSERT OR IGNORE` /
    /// `ON CONFLICT DO NOTHING` / `entry().or_insert()`). An already
    /// present key is a no-op — content is immutable, so a re-put is
    /// dedup, never overwrite. Hash verification is the store core's;
    /// this method never sees a digest.
    async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error>;

    /// The staged put's cursor handout (ADR-010 §1). The store core
    /// selects the form before content flows (the cursor's identity
    /// before commit is the engine-internal stage only — the canonical
    /// key resolves at [`StagingCursor::commit`], the type's design
    /// point). See [`StagingCursor`] for the contract every body
    /// converges on.
    async fn put_staged(&self) -> Result<Self::Staging, Error>;

    /// Unconditional delete — arbitration has already happened above the
    /// trait (the facade decides and owns the refusal: `GcRefuse` never
    /// fires below the facade, the error module's altitude discipline).
    /// `Err(Error::Missing)` when the key is absent (the facade's
    /// `delete` carries `Missing`, ADR-012 §6.5); virgin store:
    /// `Err(Error::Missing)`.
    async fn delete(&self, key: &[u8]) -> Result<(), Error>;

    /// Whole-engine key enumeration — complete and well-formed by
    /// contract (ADR-003; a non-enumerating engine must not ship, the
    /// rudolfs anti-lesson). Ordering unspecified; the store core sorts
    /// for its live-set diffs. Virgin store: empty.
    async fn list(&self) -> Result<Vec<Vec<u8>>, Error>;

    /// The metadata-only length probe (ADR-008): the indexed size column
    /// on kv engines, a file stat on `local`, a companion-row probe on
    /// pg-lo — never a value/content load. Store-level `stat` and pool
    /// accounting ride this method. `None` when absent; virgin store:
    /// `None`.
    async fn size(&self, key: &[u8]) -> Result<Option<u64>, Error>;
}

/// Clamp a `[offset, offset + len)` request by `u64` read parameters into
/// `usize` slice bounds against a body of `total` bytes — the shared
/// EOF-clamping arithmetic of the cursor contract. Crate-internal: the
/// engine cursor bodies behind [`Backend::Cursor`] compose this, so every
/// engine's materializing and staging bodies clamp identically.
///
/// The arithmetic runs in `u64` — the offset/len domain — and narrows to
/// `usize` only after clamping, so a truncated `as usize` cast can never
/// wrap a huge offset back onto a small slice position on narrow-pointer
/// targets.
pub(crate) fn clamped(offset: u64, len: u64, total: usize) -> (usize, usize) {
    let total = total as u64;
    let start = offset.min(total);
    let end = start.saturating_add(len).min(total);
    (start as usize, end as usize)
}

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

    use parking_lot::Mutex;

    const CONTENT: &[u8] = b"the quick brown fox jumps over the lazy dog";

    /// Clamp a range request to a cursor body's whole length — the test
    /// doubles' shared read arithmetic.
    fn body_slice(whole: &[u8], offset: u64, len: u64) -> Vec<u8> {
        let (start, end) = clamped(offset, len, whole.len());
        whole[start..end].to_vec()
    }

    /// kv body family: the whole value materialized (the row read's shape).
    struct MaterializedCursor {
        value: Vec<u8>,
    }

    impl ReadCursor for MaterializedCursor {
        fn len(&self) -> u64 {
            self.value.len() as u64
        }

        async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
            Ok(body_slice(&self.value, offset, len))
        }
    }

    /// Large body family: a bounded pread-style assembly from segments —
    /// the value is never handed to the caller as one materialized unit,
    /// reads gather through the loop.
    struct StreamingCursor {
        whole: Vec<u8>,
        segment: usize,
    }

    impl ReadCursor for StreamingCursor {
        fn len(&self) -> u64 {
            self.whole.len() as u64
        }

        async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
            let (start, end) = clamped(offset, len, self.whole.len());
            let mut out = Vec::with_capacity(end - start);
            let mut pos = start;
            while pos < end {
                let next = (((pos / self.segment) + 1) * self.segment).min(end);
                out.extend_from_slice(&self.whole[pos..next]);
                pos = next;
            }
            Ok(out)
        }
    }

    /// The full cursor contract every body family must satisfy — run
    /// against both body-family doubles.
    async fn assert_cursor_contract<C: ReadCursor>(cursor: &C, whole: &[u8]) {
        assert_eq!(
            cursor.len() as usize,
            whole.len(),
            "len is the entry length, metadata-only"
        );
        let len = whole.len() as u64;

        assert_eq!(cursor.read_at(0, len).await.unwrap(), whole, "full read");
        assert_eq!(
            cursor.read_at(0, len + 17).await.unwrap(),
            whole,
            "over-read clamps at EOF, pread-style"
        );
        let mid = (whole.len() / 2) as u64;
        assert_eq!(
            cursor.read_at(mid, len).await.unwrap(),
            &whole[mid as usize..],
            "tail read clamps"
        );
        assert_eq!(
            cursor.read_at(len, 5).await.unwrap(),
            Vec::<u8>::new(),
            "read at EOF is empty, never an error"
        );
        assert_eq!(
            cursor.read_at(u64::MAX, 5).await.unwrap(),
            Vec::<u8>::new(),
            "an overflowing u64 offset clamps to EOF in u64 arithmetic — never wraps"
        );
        assert_eq!(
            cursor.read_at(0, u64::MAX).await.unwrap(),
            whole,
            "an overflowing u64 length clamps to the whole body"
        );
        assert_eq!(
            cursor.read_at(0, 0).await.unwrap(),
            Vec::<u8>::new(),
            "zero-length read"
        );

        let (start, end) = clamped(mid, 7.min(len.max(1)), whole.len());
        let (slice, digest) = cursor.read_range(mid, 7.min(len.max(1))).await.unwrap();
        assert_eq!(
            slice,
            &whole[start..end],
            "read_range slice == read_at bytes"
        );
        let expected: [u8; 32] = Sha256::digest(&slice).into();
        assert_eq!(
            digest, expected,
            "slice digest is SHA-256 over exactly the returned bytes"
        );

        let (full, full_digest) = cursor.read_range(0, len).await.unwrap();
        assert_eq!(full, whole);
        assert_eq!(full_digest, slice_digest(whole));
    }

    #[tokio::test]
    async fn cursor_contract_materialized_body() {
        assert_cursor_contract(
            &MaterializedCursor {
                value: CONTENT.to_vec(),
            },
            CONTENT,
        )
        .await;
    }

    #[tokio::test]
    async fn cursor_contract_streaming_body() {
        assert_cursor_contract(
            &StreamingCursor {
                whole: CONTENT.to_vec(),
                segment: 8,
            },
            CONTENT,
        )
        .await;
    }

    #[tokio::test]
    async fn cursor_contract_empty_entry() {
        assert_cursor_contract(&MaterializedCursor { value: Vec::new() }, b"").await;
        assert_cursor_contract(
            &StreamingCursor {
                whole: Vec::new(),
                segment: 8,
            },
            b"",
        )
        .await;
    }

    /// Both body families serve the ADR-006 range contract at one seam:
    /// identical slices and digests for identical ranges.
    #[tokio::test]
    async fn both_body_families_serve_the_range_seam_identically() {
        let materialized = MaterializedCursor {
            value: CONTENT.to_vec(),
        };
        let streaming = StreamingCursor {
            whole: CONTENT.to_vec(),
            segment: 9,
        };
        for (offset, len) in [(0u64, 0u64), (0, 5), (11, 9), (43, 4), (43, 400)] {
            assert_eq!(
                materialized.read_range(offset, len).await.unwrap(),
                streaming.read_range(offset, len).await.unwrap(),
                "range {offset}+{len} identical across body families"
            );
        }
    }

    /// The slice-digest convention is plain SHA-256 over the slice bytes
    /// (ADR-006), asserted against published SHA-256 vectors independent
    /// of this crate's own hashing (the POC lesson: known-answer tests
    /// are only as trustworthy as their source of truth — NIST's vectors
    /// here, git's for the canonical derivation).
    #[test]
    fn slice_digest_convention_matches_published_vectors() {
        assert_eq!(
            hex::encode(slice_digest(b"abc")),
            "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad",
            "sha256 'abc' (NIST)"
        );
        assert_eq!(
            hex::encode(slice_digest(b"")),
            "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855",
            "sha256 empty (NIST)"
        );
    }

    /// The test-double backend: a BTreeMap CAS table whose bodies are the
    /// shapes the trait shapes must compose — a materializing cursor body,
    /// a degenerate in-memory staging body, first-writer-wins puts.
    #[derive(Clone, Default)]
    struct TestBackend {
        rows: Arc<Mutex<BTreeMap<Vec<u8>, Vec<u8>>>>,
        fail_commit: bool,
    }

    impl TestBackend {
        fn row(&self, key: &[u8]) -> Option<Vec<u8>> {
            self.rows.lock().get(key).cloned()
        }
    }

    struct TestCursor {
        value: Vec<u8>,
    }

    impl ReadCursor for TestCursor {
        fn len(&self) -> u64 {
            self.value.len() as u64
        }

        async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
            Ok(body_slice(&self.value, offset, len))
        }
    }

    struct TestStaging {
        staged: Vec<u8>,
        rows: Arc<Mutex<BTreeMap<Vec<u8>, Vec<u8>>>>,
        fail_commit: bool,
    }

    impl ReadCursor for TestStaging {
        fn len(&self) -> u64 {
            self.staged.len() as u64
        }

        async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
            Ok(body_slice(&self.staged, offset, len))
        }
    }

    impl StagingCursor for TestStaging {
        async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> {
            self.staged.extend_from_slice(chunk);
            Ok(())
        }

        async fn commit(mut self, key: &[u8]) -> Result<(), Error> {
            if self.fail_commit {
                self.staged.clear();
                return Err(Error::Io("engine commit failure (test double)".to_owned()));
            }
            let value = std::mem::take(&mut self.staged);
            self.rows.lock().entry(key.to_vec()).or_insert(value);
            Ok(())
        }

        async fn discard(mut self) {
            self.staged.clear();
        }
    }

    impl Backend for TestBackend {
        type Cursor = TestCursor;
        type Staging = TestStaging;

        fn name(&self) -> &'static str {
            "test-double"
        }

        async fn has(&self, key: &[u8]) -> Result<bool, Error> {
            Ok(self.rows.lock().contains_key(key))
        }

        async fn get(&self, key: &[u8]) -> Result<Option<Self::Cursor>, Error> {
            Ok(self
                .rows
                .lock()
                .get(key)
                .cloned()
                .map(|value| TestCursor { value }))
        }

        async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> {
            self.rows
                .lock()
                .entry(key.to_vec())
                .or_insert_with(|| value.to_vec());
            Ok(())
        }

        async fn put_staged(&self) -> Result<Self::Staging, Error> {
            Ok(TestStaging {
                staged: Vec::new(),
                rows: self.rows.clone(),
                fail_commit: self.fail_commit,
            })
        }

        async fn delete(&self, key: &[u8]) -> Result<(), Error> {
            self.rows
                .lock()
                .remove(key)
                .map(|_| ())
                .ok_or(Error::Missing)
        }

        async fn list(&self) -> Result<Vec<Vec<u8>>, Error> {
            Ok(self.rows.lock().keys().cloned().collect())
        }

        async fn size(&self, key: &[u8]) -> Result<Option<u64>, Error> {
            Ok(self.rows.lock().get(key).map(|value| value.len() as u64))
        }
    }

    /// The Backend trait's method contract against the double: the seven
    /// methods' absence/value/error disciplines, CAS first-writer-wins,
    /// and virgin-store reads as no-ops.
    #[tokio::test]
    async fn backend_method_contract() {
        let backend = TestBackend::default();
        let key = b"\x01opaque-key-bytes".to_vec();
        assert_eq!(backend.name(), "test-double", "reporting identity");

        assert!(
            !backend.has(&key).await.unwrap(),
            "virgin store: has is a no-op read"
        );
        assert!(
            backend.get(&key).await.unwrap().is_none(),
            "virgin store: get is a no-op read"
        );
        assert_eq!(
            backend.size(&key).await.unwrap(),
            None,
            "virgin store: size is a no-op read"
        );
        assert!(
            backend.list().await.unwrap().is_empty(),
            "virgin store: list is empty"
        );
        assert_eq!(
            backend.delete(&key).await.unwrap_err(),
            Error::Missing,
            "virgin store: delete reports Missing"
        );

        backend.put(&key, CONTENT).await.unwrap();
        assert!(backend.has(&key).await.unwrap(), "put lands");
        assert_eq!(
            backend.size(&key).await.unwrap(),
            Some(CONTENT.len() as u64),
            "size is metadata-only and exact"
        );
        backend.put(&key, b"overwritten").await.unwrap();
        assert_eq!(
            backend.row(&key).as_deref(),
            Some(CONTENT),
            "CAS first-writer wins: re-put is a no-op"
        );
        assert_eq!(
            backend.list().await.unwrap(),
            vec![key.clone()],
            "list is complete, keys not values"
        );

        backend.delete(&key).await.unwrap();
        assert!(!backend.has(&key).await.unwrap(), "delete lands");
        assert_eq!(
            backend.delete(&key).await.unwrap_err(),
            Error::Missing,
            "delete reports Missing when absent"
        );
    }

    /// The staged put's happy path: handout, appends, the verification
    /// pass reading the staged artifact back as a whole through the read
    /// seam, commit publishing, and the committed entry round-tripping
    /// through get's cursor at the range seam.
    #[tokio::test]
    async fn staged_put_round_trips_through_the_read_seam() {
        let backend = TestBackend::default();
        let key = b"\x01staged-key".to_vec();

        let mut staging = backend.put_staged().await.unwrap();
        staging.append(&CONTENT[..8]).await.unwrap();
        staging.append(&CONTENT[8..16]).await.unwrap();
        staging.append(&CONTENT[16..]).await.unwrap();

        assert_eq!(
            staging.len() as usize,
            CONTENT.len(),
            "staged length is the appended length"
        );
        let (bytes, digest) = staging
            .read_range(0, staging.len())
            .await
            .expect("staged bytes readable");
        assert_eq!(bytes, CONTENT, "staged artifact readable as a whole");
        assert_eq!(digest, slice_digest(CONTENT));

        staging.commit(&key).await.expect("commit publishes");
        assert!(backend.has(&key).await.unwrap(), "entry published");

        let cursor = backend
            .get(&key)
            .await
            .expect("get succeeds")
            .expect("entry present");
        let (bytes, digest) = cursor
            .read_range(0, cursor.len())
            .await
            .expect("committed read");
        assert_eq!(bytes, CONTENT);
        assert_eq!(digest, slice_digest(CONTENT));

        backend.put(&key, b"overwritten").await.unwrap();
        assert_eq!(
            backend.row(&key).as_deref(),
            Some(CONTENT),
            "staged commit also lands CAS-first-writer"
        );
    }

    /// Stage hygiene, discard form: every non-commit path converges on
    /// discard with zero residue — nothing was staged into the pool.
    #[tokio::test]
    async fn staged_discard_leaves_zero_residue() {
        let backend = TestBackend::default();
        let key = b"\x01discard-key".to_vec();

        let mut staging = backend.put_staged().await.unwrap();
        staging.append(b"residue").await.unwrap();
        staging.discard().await;

        assert!(
            !backend.has(&key).await.unwrap(),
            "discarded stage published nothing"
        );
        assert_eq!(backend.size(&key).await.unwrap(), None, "no entry landed");
        assert!(
            backend.list().await.unwrap().is_empty(),
            "the pool holds exactly zero entries after discard"
        );
    }

    /// Stage hygiene, failure form: a failed commit is already
    /// self-discarded — zero residue, entry unpublished, no post-failure
    /// discard left to drive (the consuming call took the cursor).
    #[tokio::test]
    async fn commit_failure_self_discards() {
        let backend = TestBackend {
            fail_commit: true,
            ..Default::default()
        };
        let key = b"\x01failure-key".to_vec();

        let mut staging = backend.put_staged().await.unwrap();
        staging.append(CONTENT).await.unwrap();
        let err = staging.commit(&key).await.unwrap_err();
        assert!(matches!(err, Error::Io(_)), "engine media failure");

        assert!(
            !backend.has(&key).await.unwrap(),
            "failed commit published nothing"
        );
        assert_eq!(backend.size(&key).await.unwrap(), None, "zero residue");
    }
}