alkblobs 0.2.2

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
//! The store facade's value types (store-api.md §Public surface): the
//! shapes put/get/stat/sweep return, the pin lifecycle vocabulary, and
//! the registered-liveness vocabulary. Engine-blind — nothing here
//! knows which engines a store runs (ADR-007/008/010); these types are
//! the contract the facade behavior implements and every consumer
//! programs against.
//!
//! **The pin types ([`Pin`], [`Batch`]) carry their registry
//! back-reference.** Each guard holds an `Arc` to the store instance's
//! GC registry core ([`crate::gc`]), so a guard's release reaches the
//! per-key refcounts under the arbitration lock wherever the guard
//! goes — the wave-1 shape carried only the key. The guard's drop
//! decrements its refcount; a [`Batch`] holds its members' keys and
//! releases every one of them in **one** ordered step under the same
//! lock on commit or drop (no early single-pin release). The guard
//! types never expose the registry consumer-side: the Arc is private,
//! minting happens on the store core only.
//!
//! **The registered-liveness vocabulary is consumer-owned shared
//! state, never a snapshot.** [`LivenessSource`] and
//! [`ProtectCallback`] are the two consumer-side halves of ADR-005's
//! three liveness sources (the pins are the third); both are
//! registered through the facade and are consulted *live* — at mark
//! time and at delete-time arbitration — never copied at registration
//! (POC #1 finding 7's install-implies-copy failure is the pinning
//! reason). A [`Registration`] is the deregistration token the
//! correction flows of ADR-012 §6.4 need.

use std::fmt;
use std::sync::Arc;
use std::time::SystemTime;

use crate::backend::ReadCursor;
use crate::gc::RegistryCore;
use crate::hashing::Algorithm;
use crate::keys::Key;
use crate::store::AnyCursor;

/// The `get` seam's public stream (`(len, stream)` — store-api.md §Read
/// path): a bounded, re-entrant **windowed reader** over a stored entry,
/// uniform across both tiers with the crate-internal read cursor
/// (ADR-010 §1) staying behind it. The kv tier's windows slice
/// materialized content (bounded by tier policy); the large tier's read
/// window-by-window over the committed file — streaming without ever
/// materializing the whole value. `read_at` is a pread-style bounded
/// window (over-reads clamp at the entry's end; a read at EOF is empty,
/// never an error); `read_range` is the same window plus the ADR-006
/// slice digest — plain SHA-256 over exactly the returned bytes, fixed
/// by convention.
///
/// Constructed by the store core only (via `get`); engine cursors never
/// leak onto this surface. A `Blob` stays valid across concurrent pool
/// activity because committed entries are immutable. No verification
/// rides the windows themselves: canonical verification fires at get
/// time on the kv tier's materialized reads (and at put time), per the
/// store core's recorded verification-on-get posture (ADR-006).
pub struct Blob {
    cursor: AnyCursor,
}

impl fmt::Debug for Blob {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Blob")
            .field("len", &self.cursor.len())
            .finish()
    }
}

impl Blob {
    /// Store-core constructor: a `Blob` wraps a cursor-dispatched
    /// engine body; consumers obtain one only through the read seam.
    pub(crate) fn new(cursor: AnyCursor) -> Blob {
        Blob { cursor }
    }

    /// The entry's length in bytes — the metadata-only length the
    /// cursor body carries (the same probe `stat` rides).
    pub fn len(&self) -> u64 {
        self.cursor.len()
    }

    /// Whether the entry holds zero bytes.
    pub fn is_empty(&self) -> bool {
        self.cursor.len() == 0
    }

    /// The bytes at `[offset, offset + len)` — clamped at the entry's
    /// end (pread-style short read); `offset >= len()` (EOF) is empty,
    /// never an error.
    pub async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, crate::Error> {
        self.cursor.read_at(offset, len).await
    }

    /// The bytes at `[offset, offset + len)` plus the ADR-006 slice
    /// digest — SHA-256 over exactly the returned bytes, for the caller
    /// to carry out-of-band.
    pub async fn read_range(
        &self,
        offset: u64,
        len: u64,
    ) -> Result<(Vec<u8>, [u8; 32]), crate::Error> {
        self.cursor.read_range(offset, len).await
    }
}

/// The cheap stat probe's result (store-api.md §Read path): the entry's
/// length and its hash algorithm — the "type" at this layer is the hash
/// algorithm, there is no other (ADR-002). `stat` rides the Backend
/// trait's `size` method (ADR-008): an indexed-column probe on kv
/// engines, a file stat on `local`, a companion-row probe on pg-lo —
/// never a value/content load.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct EntryMeta {
    /// The entry's length in bytes.
    pub len: u64,
    /// The key's hash algorithm — ADR-002's two-member policy.
    pub algorithm: Algorithm,
}

/// The RAII put-path pin guard (ADR-005 §Marks 2; store-api.md §Write
/// path).
///
/// **Lifecycle contract (ADR-005; the machinery is wired):**
///
/// - Acquired **before the entry becomes visible** — pin-before-publish:
///   an entry cannot appear in a sweep's `list()` without already
///   carrying its pin (or its converted registration, ADR-005's put-side
///   invariant).
/// - **Back-reference shape (named design point):** the guard holds an
///   `Arc` to the store's GC registry core (`RegistryCore`,
///   crate-internal) alongside
///   its key, so the drop reaches the per-key refcounts under the
///   arbitration lock the sweep consults — wherever the guard travels.
///   The registry core outlives the `Store` instance when guards still
///   reference it; a late release against a detached registry is
///   harmless (the refcount bookkeeping is self-contained).
/// - The guard holds a per-key refcount under the same arbitration lock
///   the sweep consults, until **dropped or converted into a registered
///   consumer reference** (a registered liveness source or a root tag).
///   Conversion goes through the registry's ordered-step primitive
///   (`RegistryCore::convert`, crate-internal): the pin's release and
///   the replacement's registration into a liveness source are one
///   ordered step under the same lock — a consumer may not release a
///   pin before its registered source observes the replacement (the
///   local analog of the ops-layer pin-token contract).
/// - A dedup put's guard protects exactly as a fresh put's does: a
///   re-put acquires no new pool entry but refcounts the existing one
///   under the same arbitration lock (ADR-005).
/// - Fleet correspondence (ADR-010 §4, ADR-012 §2): in fleet mode the
///   joint tx commits the expiry-carrying pin row alongside the entry,
///   and **the guard carries the row's renewal duty while it lives** —
///   the store's TTL/3 schedule (ADR-008 §2; the landed
///   `store-fleet-put` design point: a per-guard tokio task, aborted on
///   the guard's drop/conversion/batch contribution, silent on a failed
///   renewal, ended by the never-resurrect `Ok(false)` — the row's
///   expiry is the once-protection after the guard leaves). A guard
///   that outlives its store instance keeps renewing through the
///   registry-adjacent engine handle it carries (same detachment as the
///   refcount bookkeeping); in non-fleet modes the ops layer's token is
///   this guard's in-process handle — nothing expires outside fleet
///   mode.
pub struct Pin {
    registry: Arc<RegistryCore>,
    key: Key,
    /// False once the guard has been contributed to a [`Batch`] (the
    /// release duty moved to the batch's one-step drop) or consumed by
    /// the ordered-step conversion.
    armed: bool,
    /// The fleet pin row's renewal duty — `Some` exactly on a guard a
    /// fleet put returned (store-fleet-put's fleet form of
    /// pin-before-publish, ADR-008 §2). Aborted wherever the guard's
    /// life or ownership ends: drop, batch contribution (the batch's
    /// one-step release), or the ordered-step conversion. Gated to the
    /// fleet-capable build (feature `postgres`) — fleet state can only
    /// ride the postgres host.
    #[cfg(feature = "postgres")]
    renewal: Option<PinRenewal>,
}

impl Pin {
    /// The key whose entry the guard protects.
    pub fn key(&self) -> &Key {
        &self.key
    }

    /// Store-core constructor: pins are minted by the GC registry
    /// ([`crate::gc`]) when the refcount is acquired under the
    /// arbitration lock — never held by consumers before a put exists.
    pub(crate) fn minted(registry: Arc<RegistryCore>, key: Key) -> Pin {
        Pin {
            registry,
            key,
            armed: true,
            #[cfg(feature = "postgres")]
            renewal: None,
        }
    }

    /// Attach the pin row's TTL/3 renewal duty (store-core call, fleet
    /// puts only): while the guard lives, its task moves the row's
    /// expiry forward — see the struct's fleet-correspondence bullet.
    #[cfg(feature = "postgres")]
    pub(crate) fn attach_renewal(&mut self, renewal: PinRenewal) {
        self.renewal = Some(renewal);
    }

    /// Take the renewal duty out (the batch contribution's move: the
    /// batch's one-step release owns it from here).
    #[cfg(feature = "postgres")]
    pub(crate) fn take_renewal(&mut self) -> Option<PinRenewal> {
        self.renewal.take()
    }

    /// Stop the renewal duty without moving it — the ordered-step
    /// conversion's pre-stop (the pin's own drop is suppressed there,
    /// so the abort must happen before [`RegistryCore::convert`]'s
    /// `ManuallyDrop` takes over).
    #[cfg(feature = "postgres")]
    pub(crate) fn stop_renewal(&mut self) {
        if let Some(renewal) = self.renewal.take() {
            renewal.stop();
        }
    }
}

impl fmt::Debug for Pin {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("Pin")
            .field("key", &self.key.digest())
            .field("armed", &self.armed)
            .finish()
    }
}

impl Drop for Pin {
    fn drop(&mut self) {
        if self.armed {
            self.registry.release_pin(&self.key);
        }
        #[cfg(feature = "postgres")]
        if let Some(renewal) = self.renewal.take() {
            renewal.stop();
        }
    }
}

/// The fleet pin row's renewal duty — the guard's TTL/3 renewal task's
/// handle (store-fleet-put's named design point: a **per-guard tokio
/// task**; the shared-renewal-sweep keyed off a guard registry was the
/// rejected alternative — it needs the same lifecycle notifications an
/// abort handle gives for free). The task moves the row's expiry
/// forward every TTL/3 while it lives; wherever the guard's life or
/// ownership ends (drop, batch contribution, conversion), the handle
/// aborts it — no leaked tasks, no renewal past the guard's life.
/// Crate-internal, minted by the store core's fleet put path only.
#[cfg(feature = "postgres")]
pub(crate) struct PinRenewal {
    handle: tokio::task::JoinHandle<()>,
}

#[cfg(feature = "postgres")]
impl PinRenewal {
    /// Store-core constructor: the duty wraps the renewal task the
    /// fleet put path spawned.
    pub(crate) fn spawned(handle: tokio::task::JoinHandle<()>) -> PinRenewal {
        PinRenewal { handle }
    }
    /// Stop the duty — aborts the task at its next await point. A
    /// renewal caught mid-flight is harmless: the clause's expiry
    /// update is idempotent, and a task killed between its sleep and
    /// the UPDATE leaves the row exactly as it was.
    pub(crate) fn stop(&self) {
        self.handle.abort();
    }
}

/// Batch-scoped pin group (ADR-005 §Marks 2; store-api.md): multi-put
/// writes — manifest writes are exactly this — group under one
/// batch-scoped pin.
///
/// **Lifecycle contract (ADR-005; the machinery is wired):**
///
/// - Created by the store core's batching put seam; each put within the
///   batch contributes its pin (batch-internal puts publish
///   pin-before-publish exactly as lone puts do).
/// - **A batch's pins drop together on batch commit or batch drop**, in
///   one ordered step under the same arbitration lock the sweep
///   consults. No early single-pin release: protection spans the whole
///   batch or none of it. The batch holds its members' keys; the
///   release takes the arbitration lock once and decrements every
///   member refcount inside that one section.
/// - Batch drop is the failure path (stream error, caller abandonment):
///   every member pin releases together; member liveness never outlives
///   the batch.
pub struct Batch {
    registry: Arc<RegistryCore>,
    keys: Vec<Key>,
    /// The members' renewal duties, moved out of each contributed pin
    /// ([`Batch::add`]) and aborted by the one-step release — the batch
    /// group's life ends the members' fleet renewals with it
    /// (store-fleet-put: no renewal past the batch's life). Gated to
    /// the fleet-capable build, feature `postgres` ([`PinRenewal`]'s
    /// gate).
    #[cfg(feature = "postgres")]
    renewals: Vec<PinRenewal>,
}

impl Batch {
    /// Store-core constructor: the batching put seam creates batches
    /// through the GC registry ([`crate::gc`]).
    pub(crate) fn created(registry: Arc<RegistryCore>) -> Batch {
        Batch {
            registry,
            keys: Vec::new(),
            #[cfg(feature = "postgres")]
            renewals: Vec::new(),
        }
    }

    /// Contribute a put's pin to the group (store-core call). The
    /// contributed guard is defused — its own drop releases nothing;
    /// the batch's one-step release owns the protection from here —
    /// and its renewal duty moves into the batch (release stops it;
    /// while the batch lives, the member's row keeps renewing).
    pub(crate) fn add(&mut self, mut pin: Pin) {
        pin.armed = false;
        #[cfg(feature = "postgres")]
        if let Some(renewal) = pin.take_renewal() {
            self.renewals.push(renewal);
        }
        self.keys.push(pin.key().clone());
    }

    /// The batch's GC registry Arc — the batching put seam's
    /// foreign-Batch identity check (review-001 N-5) compares it
    /// against the minting store's.
    pub(crate) fn registry(&self) -> &Arc<RegistryCore> {
        &self.registry
    }

    /// How many pins the group currently protects.
    pub fn len(&self) -> usize {
        self.keys.len()
    }

    /// Whether the group currently protects nothing.
    pub fn is_empty(&self) -> bool {
        self.keys.is_empty()
    }

    /// **Batch commit**: the group's pins drop together in one ordered
    /// step — protection spans the whole batch or none of it, and ends
    /// when the batch ends (ADR-005 §Marks 2; the same release batch
    /// drop drives). A committed batch's members are no longer
    /// batch-protected; durable protection beyond the write group is
    /// the consumer's liveness seam (`Store::register_liveness_source`
    /// or the pin-conversion ordering).
    pub fn commit(mut self) {
        self.release_members();
    }

    /// The one-step release: every member refcount decremented inside
    /// a single arbitration-lock section, sorted into the keys' total
    /// order. Driven by [`Drop::drop`] (batch drop) and available for
    /// the batching put seam's commit step (which consumes the batch
    /// the same way).
    pub(crate) fn release_members(&mut self) {
        #[cfg(feature = "postgres")]
        {
            let renewals = std::mem::take(&mut self.renewals);
            for renewal in &renewals {
                renewal.stop();
            }
        }
        self.registry.release_pins(std::mem::take(&mut self.keys));
    }
}

impl Drop for Batch {
    fn drop(&mut self) {
        self.release_members();
    }
}

/// A sweep's abort cause — report data only, never an error (ADR-012 §1:
/// sweeps are never errors; an aborted sweep returns `Ok(SweepReport)`
/// with `aborted: Some(cause)`, and there is no sweep-abort error
/// variant).
///
/// Exactly three variants, ADR-pinned; new causes are new ADRs.
/// `#[non_exhaustive]` makes the ADR-012 §1 ruling — callers match
/// non-exhaustively — structural for downstream crates.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum GcAbortCause {
    /// A protection source failed to answer (ADR-005's protect-callback
    /// failure): the sweep aborted, nothing deleted.
    ProtectFailed,
    /// The fleet sweep lease was contended (ADR-008; ADR-013 §2's
    /// exclusive-lease semantics): the second sweeper aborts — nothing
    /// deleted; its next run takes the lock.
    SweeperLock,
    /// The safe default: a sweep invoked with zero registered liveness
    /// sources aborts without deleting (store-api.md §Lifecycle).
    NoLivenessSources,
}

/// A key's pin classification (ADR-010 §2): the engine's only vocabulary
/// for pins — the engine never decides deletion, it only classifies, and
/// the store core receives the answer and applies the decision.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PinState {
    /// An un-expired pin row exists for the key — the key is protected.
    Protected,
    /// Nothing pins the key.
    Unprotected,
}

/// A sweep's outcome record — always data, never an error carrier
/// (ADR-012 §1). Counts are exact; the test gate asserts sweep outcomes
/// against these fields (store-api.md invariants 4/7).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SweepReport {
    /// Candidates staged into the delete window at the mark phase.
    pub candidates_staged: u64,
    /// Candidates actually deleted at the commit phase — plus, on the
    /// sweep-as-executor path, the overdue staged-window candidates this
    /// sweep re-arbitrated and executed (ADR-012 §6.7); `0` on any abort.
    pub deleted: u64,
    /// Candidates cancelled at delete time by arbitration — the put side
    /// of the sweep-vs-put race won, and protection composed over the
    /// candidate (ADR-005's delete-window protocol).
    pub cancelled_by_arbitration: u64,
    /// `Some(cause)` when the sweep aborted — nothing was deleted;
    /// `None` on a completed sweep.
    pub aborted: Option<GcAbortCause>,
    /// Wall-clock start of the run.
    pub started_at: SystemTime,
    /// Wall-clock finish — an aborted sweep also finishes; the report is
    /// its outcome record either way.
    pub finished_at: SystemTime,
}

/// A registered liveness source — the live-shared GC seam (ADR-005
/// §Marks 1; store-api.md §Lifecycle, `register_liveness_source`).
///
/// **Named design point — the source shape: two sync verbs, live-shared
/// consults, none blocking.** The source is consumer-owned state shared
/// by handle (`Arc<dyn LivenessSource>`); the store consults the
/// *current* contents every time and never snapshots at registration —
/// POC #1 finding 7's install-implies-copy failure (roots registered
/// after install were invisible to GC) is the pinning reason. The two
/// verbs split by protocol phase:
///
/// - [`live_keys`](Self::live_keys) — mark-time enumeration, consulted
///   outside the arbitration lock (whole live-set computation);
/// - [`contains`](Self::contains) — delete-time arbitration probe,
///   consulted **under the in-process arbitration lock** (the pin/
///   liveness lock), so it must answer from an indexed structure and
///   never block or await; the provided default derives it from
///   [`live_keys`](Self::live_keys) (correct but linear — indexed
///   sources override).
///
/// The verbs are sync (non-async) by the same lock discipline: pin
/// drops are sync RAII, and arbitration consults run under a
/// short-held `parking_lot` lock (store-api.md §Concurrency posture;
/// AGENTS convention 3). A source whose set changes concurrently with
/// a sweep is the consumer's coordination duty — the delete-time
/// re-observation composes the sources' live-shared answers, and the
/// ADR-005 Consequences assign pin/liveness misuse to the consumer
/// contract (store-api.md's invariants document it).
pub trait LivenessSource: Send + Sync + 'static {
    /// The keys currently live per this source's own reference state —
    /// consulted at mark time, every sweep, never a stale snapshot.
    fn live_keys(&self) -> Vec<Key>;

    /// Whether the key is live right now — consulted at delete-time
    /// arbitration (sweep candidates' cancel check and the direct
    /// delete's full protection set). Runs under the arbitration lock:
    /// quick, indexed, never blocking. Default: linear over
    /// [`live_keys`](Self::live_keys).
    fn contains(&self, key: &Key) -> bool {
        self.live_keys().contains(key)
    }
}

/// A registered protect callback — ADR-005 §Marks 3's protection
/// source, consulted pre-sweep and at direct-delete arbitration.
///
/// **Named design point — the callback shape: one sync verb returning
/// the protection set or a failure to answer.** `None` means the
/// protection source could not answer, and every consumer of that
/// answer takes the safe side (iroh's `ProtectOutcome::Abort`
/// conclusion, adopted): a sweep aborts with
/// [`GcAbortCause::ProtectFailed`], nothing deleted (ADR-012 §1:
/// abort is report data); a direct delete refuses conservatively with
/// `Error::GcRefuse` — nothing deleted either way. The verb is sync by
/// the same discipline as [`LivenessSource`] (direct-delete
/// arbitration runs it under the arbitration lock — quick, never
/// blocking); mark-time consults run outside the lock and may be
/// slower.
pub trait ProtectCallback: Send + Sync + 'static {
    /// The externally-known-live keys, or `None` when the protection
    /// source failed to answer (abort/refuse semantics above).
    fn protect(&self) -> Option<Vec<Key>>;
}

/// A registration token — the opaque handle the store's
/// `unregister_liveness_source`/`unregister_protect_callback` verbs
/// consume to remove a registered [`LivenessSource`], a
/// [`ProtectCallback`], or a fleet liveness table registration (the
/// [`LivenessRegistration`] verb's fleet form). ADR-012 §6.4's
/// embedder correction flow ("deregisters/re-registers around the
/// delete") is exactly this: the token makes deregistration
/// expressible without exposing the registry internals.
///
/// Copyable, but each token deregisters at most once (the second call
/// returns `false`).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct Registration {
    pub(crate) id: u64,
    pub(crate) kind: RegistrationKind,
}

/// The registration record's kind — crate-internal bookkeeping behind
/// [`Registration`], routing the deregistration verbs (a source token
/// never deregisters a protect callback, and so on).
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub(crate) enum RegistrationKind {
    /// An in-process [`LivenessSource`] registration.
    Source,
    /// A [`ProtectCallback`] registration.
    Protect,
    /// A fleet liveness table registration — the
    /// [`LivenessRegistration::FleetTable`] form of
    /// `register_liveness_source` (ADR-012 §6.6).
    FleetSource,
}

/// The liveness registration's **argument form** — ADR-012 §6.6's
/// fleet-shaped argument riding the same verb
/// (`register_liveness_source`): the two legal forms of registering a
/// liveness source, single-node and fleet.
///
/// **Named design points (recorded):**
///
/// - **The verb's shape and error surface.** The fleet form is not a
///   sibling method: ADR-012 §6.6's "same verb" wording pins the
///   argument-form reading, so this type IS the verb's argument and
///   the verb returns `Result<Registration, Error>` — the pre-fleet
///   signature's infallibility was an accident of the in-process form
///   alone (registration bookkeeping cannot fail there); the fleet
///   form validates the table reference (malformed references error
///   at registration, never interpolate unquoted) and forwards to the
///   shared engine, so the unified signature carries the fleet form's
///   fallibility. The in-process form never errors in practice.
/// - **In-process sources remain legal under `fleet: true`**
///   (allow-and-document): an instance-local registration's
///   protection is instance-local only — it covers *this* instance's
///   sweeps (mark-time enumeration) and *this* instance's direct
///   deletes (the in-process domain's probes); it is not a *fleet*
///   liveness form (ADR-008 §2: the SQL protocol sees tables, not
///   callbacks), so the embedder's fleet-wide protection must
///   materialize into its liveness table — or pin rows — before the
///   fleet's other instances sweep. The two domains compose: the
///   in-process registrations protect locally, the registered table
///   protects fleet-wide.
pub enum LivenessRegistration {
    /// The single-node form: an in-process [`LivenessSource`] — the
    /// live-shared seam consulted at mark time and at delete-time
    /// arbitration (ADR-005 §Marks 1), exactly the original verb's
    /// argument. Legal under every topology; under `fleet: true` its
    /// protection is instance-local (see the type's design points).
    InProcess(Arc<dyn LivenessSource>),
    /// The fleet form (ADR-008 §2; ADR-012 §6.6): a **named table
    /// reference** in the shared engine — the embedder's own liveness
    /// table (`(key bytea, …)`; schema-qualified name like
    /// `"schema"."table"` — every dot-component rides the
    /// quoted-identifier boundary, an empty component is an error).
    /// The store core reads the table at delete-time arbitration and
    /// **never owns it** — the schema-ownership rule (the engine never
    /// creates, alters, or drops it); registration only records the
    /// reference (idempotent re-registration of the same reference).
    /// Only legal on a fleet store (`fleet: true` — under `fleet: false`
    /// the fleet machinery is dormant and the verb errors).
    FleetTable(String),
}

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

    /// The stat value's shape: length plus the key's algorithm — there
    /// is no other type at this layer (store-api.md §Read path).
    #[test]
    fn entry_meta_carries_len_and_algorithm() {
        let key = Key::from(Algorithm::GitBlobSha256.derive(b"meta", 4));
        let meta = EntryMeta {
            len: 4,
            algorithm: key.algorithm(),
        };
        assert_eq!(meta.len, 4);
        assert_eq!(meta.algorithm, Algorithm::GitBlobSha256);
        assert_eq!(meta.algorithm.digest_len(), 32);
    }

    /// The abort vocabulary's exhaustiveness shape: exactly three
    /// variants, ADR-pinned (ADR-012 §1). The in-crate match below is
    /// exhaustive without a wildcard — adding a variant breaks it,
    /// which is the "new causes are new ADRs" discipline surfacing at
    /// compile time in-crate; `#[non_exhaustive]` enforces the
    /// wildcard discipline on downstream callers.
    #[test]
    fn gc_abort_cause_is_exactly_three_variants() {
        let mut matched: Vec<&'static str> = Vec::new();
        for cause in [
            GcAbortCause::ProtectFailed,
            GcAbortCause::SweeperLock,
            GcAbortCause::NoLivenessSources,
        ] {
            match cause {
                GcAbortCause::ProtectFailed => matched.push("protect"),
                GcAbortCause::SweeperLock => matched.push("lease"),
                GcAbortCause::NoLivenessSources => matched.push("sources"),
            }
        }
        assert_eq!(matched, ["protect", "lease", "sources"]);
    }

    /// Report field semantics: exact counts, `aborted: None` on a
    /// completed run, and the abort shape — nothing deleted, the cause
    /// carried, the report still a finished record.
    #[test]
    fn sweep_report_carries_exact_counts_and_abort_data() {
        let now = SystemTime::now();
        let completed = SweepReport {
            candidates_staged: 3,
            deleted: 2,
            cancelled_by_arbitration: 1,
            aborted: None,
            started_at: now,
            finished_at: now,
        };
        assert!(completed.aborted.is_none());
        assert!(completed.finished_at >= completed.started_at);

        let aborted = SweepReport {
            deleted: 0,
            cancelled_by_arbitration: 0,
            aborted: Some(GcAbortCause::NoLivenessSources),
            ..completed
        };
        assert_eq!(aborted.candidates_staged, 3, "mark-phase count is kept");
        assert_eq!(aborted.deleted, 0, "an aborted sweep deletes nothing");
        assert_eq!(
            aborted.aborted,
            Some(GcAbortCause::NoLivenessSources),
            "the safe default cause is report data"
        );
    }

    /// Pin/Batch shape invariants: the guard holds its key and the
    /// batch groups pins, released in one step on batch drop — the
    /// refcounts land under the registry's arbitration lock (GcHome
    /// drives the minting; the drop behavior is observable through the
    /// refcount probe).
    #[tokio::test]
    async fn pin_guard_holds_its_key_and_batch_groups_pins() {
        let gc = crate::gc::GcHome::default();
        let key = Key::from(Algorithm::GitBlobSha256.derive(b"guarded", 7));
        let pin = gc.acquire(&key).await;
        assert_eq!(pin.key().algorithm(), Algorithm::GitBlobSha256);
        assert_eq!(gc.pin_count(&key), 1, "acquire minted the refcount");
        drop(pin);
        assert_eq!(gc.pin_count(&key), 0, "drop released the refcount");

        let a = Key::from(Algorithm::GitBlobSha256.derive(b"a", 1));
        let b = Key::from(Algorithm::GitBlobSha256.derive(b"b", 1));
        let mut batch = gc.new_batch();
        assert!(batch.is_empty(), "fresh batch protects nothing yet");
        batch.add(gc.acquire(&a).await);
        batch.add(gc.acquire(&b).await);
        assert_eq!(batch.len(), 2, "member pins counted");
        assert_eq!(
            gc.pin_count(&a) + gc.pin_count(&b),
            2,
            "member refcounts held while the batch lives"
        );
        drop(batch);
        assert_eq!(
            gc.pin_count(&a) + gc.pin_count(&b),
            0,
            "batch drop released every member in one step"
        );
    }

    /// The arbitration vocabulary: two states, and the engine's reading
    /// is classification only (ADR-010 §2).
    #[test]
    fn pin_state_is_the_two_variant_classification() {
        assert_ne!(PinState::Protected, PinState::Unprotected);
        for state in [PinState::Protected, PinState::Unprotected] {
            match state {
                PinState::Protected | PinState::Unprotected => {}
            }
        }
    }
}