flint-store 0.1.0

//! The object store behind a trait — L2 step 4 (design review A13,
//! with A6's conditional-publish contract baked into the signatures).
//!
//! Only generation ASSEMBLY is backend-specific; everything above this
//! trait (interval log, flush pipeline, eviction, hydration, epoch
//! state machine) is backend-neutral. Backends:
//!
//! - `memory` — the in-process test double, implementing the FULL
//!   conditional semantics (412/409, stamps, CRC validation, MPU
//!   state, epoch CAS). Every backend-neutral tier test runs here.
//! - `s3` — MPU + UploadPartCopy + conditional PUT/Complete per A6/A8.
//! - Azure Blob (second, per A13): Put Block List with committed-block
//!   reuse and NATIVE leases for the epoch. GCS later.
//!
//! Contract points the signatures enforce:
//!
//! - **No unconditional publish exists.** Every write carries either
//!   `IfMatch` (update of a known generation) or `IfNoneMatchAny`
//!   (first generation of a new key). A caller that wants an unguarded
//!   overwrite has made a design error upstream (A6).
//! - **The multipart ETag is NOT a content hash** and never appears in
//!   any integrity role; content identity is the full-object
//!   CRC-64/NVME, computed by the flusher from LOCAL truth and
//!   validated server-side at publish (a mismatch fails the publish,
//!   it does not warn).
//! - **Assembly failure aborts the assembly** (A9): no backend may
//!   leave a partial MPU behind on an error path it can reach.
//!
//! Extracted from `spdk-csi-driver/src/tier/store/` (2026-08-25) so
//! flint-lean can depend on the store WITHOUT the hub crate's build
//! (13 binaries, the aws/kube/nfs dep trees). The hub crate re-exports
//! this crate at its old path (`crate::tier::store::*`), so every
//! existing import keeps working. The `s3` backend is feature-gated —
//! test loops that only need the memory double never build the AWS
//! SDK.

pub mod layout;
pub mod memory;
pub mod probe;
#[cfg(feature = "s3")]
pub mod s3;
#[cfg(feature = "s3")]
pub mod rawread;

use async_trait::async_trait;
use bytes::Bytes;
use std::collections::HashMap;

// ── CRC-64/NVME ──────────────────────────────────────────────────────
//
// The tier's one content-identity algorithm (A6): linearly combinable
// (so S3 can compute it FULL_OBJECT across multipart, copied parts
// included) and supported end-to-end by both S3 and the local flusher.
// Hand-rolled table CRC — 30 lines beats a dependency; the test pins
// the catalog check value.

/// Reflected form of the CRC-64/NVME polynomial 0xAD93D23594C935A9.
const CRC64_NVME_POLY: u64 = 0x9A6C_9329_AC4B_C9B5;

/// Slice-by-8: eight tables, so `update` consumes a u64 word per round
/// instead of a byte. Table 0 is the classic byte table; table k is
/// table k-1 advanced one more zero byte. Same polynomial, same digest
/// — the catalog check value and the split-update property both pin it.
///
/// This is the checkout resume path's inner loop: a restarted container
/// re-CRCs every present file before it can serve, so the byte-at-a-time
/// form put a full tree re-hash on the agent-blocking path.
fn crc64_tables8() -> &'static [[u64; 256]; 8] {
    static TABLES: std::sync::OnceLock<[[u64; 256]; 8]> = std::sync::OnceLock::new();
    TABLES.get_or_init(|| {
        let mut t = [[0u64; 256]; 8];
        t[0] = *crc64_table();
        let mut i = 0usize;
        while i < 256 {
            let mut k = 1usize;
            while k < 8 {
                let prev = t[k - 1][i];
                t[k][i] = (prev >> 8) ^ t[0][(prev & 0xFF) as usize];
                k += 1;
            }
            i += 1;
        }
        t
    })
}

fn crc64_table() -> &'static [u64; 256] {
    static TABLE: std::sync::OnceLock<[u64; 256]> = std::sync::OnceLock::new();
    TABLE.get_or_init(|| {
        let mut t = [0u64; 256];
        let mut i = 0usize;
        while i < 256 {
            let mut crc = i as u64;
            let mut b = 0;
            while b < 8 {
                crc = if crc & 1 == 1 { (crc >> 1) ^ CRC64_NVME_POLY } else { crc >> 1 };
                b += 1;
            }
            t[i] = crc;
            i += 1;
        }
        t
    })
}

/// Streaming CRC-64/NVME (the flusher feeds file ranges through this).
#[derive(Clone)]
pub struct Crc64Nvme(u64);

impl Default for Crc64Nvme {
    fn default() -> Self {
        Self::new()
    }
}

impl Crc64Nvme {
    pub fn new() -> Self {
        Crc64Nvme(u64::MAX)
    }
    pub fn update(&mut self, data: &[u8]) {
        let t = crc64_tables8();
        let mut crc = self.0;
        let mut rest = data;
        // Eight bytes per round off the aligned body...
        while let Some((head, tail)) = rest.split_first_chunk::<8>() {
            let w = u64::from_le_bytes(*head) ^ crc;
            crc = t[7][(w & 0xFF) as usize]
                ^ t[6][((w >> 8) & 0xFF) as usize]
                ^ t[5][((w >> 16) & 0xFF) as usize]
                ^ t[4][((w >> 24) & 0xFF) as usize]
                ^ t[3][((w >> 32) & 0xFF) as usize]
                ^ t[2][((w >> 40) & 0xFF) as usize]
                ^ t[1][((w >> 48) & 0xFF) as usize]
                ^ t[0][((w >> 56) & 0xFF) as usize];
            rest = tail;
        }
        // ...then the <8-byte tail, byte-at-a-time as before. `update`
        // is a STREAMING call: the tail runs per chunk, so an arbitrary
        // split must give the same digest as one pass. The test pins it.
        for &b in rest {
            crc = t[0][((crc ^ b as u64) & 0xFF) as usize] ^ (crc >> 8);
        }
        self.0 = crc;
    }
    pub fn finalize(self) -> u64 {
        self.0 ^ u64::MAX
    }
}

/// One column of the GF(2) operator applied to a register.
fn gf2_times(mat: &[u64; 64], mut v: u64) -> u64 {
    let mut sum = 0u64;
    let mut i = 0usize;
    while v != 0 {
        if v & 1 == 1 {
            sum ^= mat[i];
        }
        v >>= 1;
        i += 1;
    }
    sum
}

/// `out` becomes the operator that applies `mat` twice.
fn gf2_square(out: &mut [u64; 64], mat: &[u64; 64]) {
    for n in 0..64 {
        out[n] = gf2_times(mat, mat[n]);
    }
}

/// CRC-64/NVME of `A || B`, from the CRCs of `A` and `B` and the LENGTH
/// of `B`. Bytes are never re-read.
///
/// This exists so a full-object checksum can be assembled from parts
/// computed INDEPENDENTLY — and therefore concurrently. Without it a
/// caller that wants the store to compute the checksum has to hand it
/// the parts in order, one at a time, because a CRC taken out of order
/// is a different number; the alternative was a whole extra pass over
/// the file before the upload. Now it is neither.
///
/// The identity it rests on: for a reflected CRC whose init and xorout
/// are both all-ones, appending `n` zero bytes is a LINEAR operator `Z`
/// on the register, and `F(A||B) = Z_n(F(A)) ^ F(B)` — the init and
/// xorout terms cancel because `Z(X) ^ Z(I) = Z(X ^ I) = Z(0) = 0`.
/// Structure follows zlib's `crc32_combine`, widened to 64 bits.
pub fn crc64_combine(crc_a: u64, crc_b: u64, len_b: u64) -> u64 {
    if len_b == 0 {
        return crc_a;
    }
    let mut odd = [0u64; 64];
    let mut even = [0u64; 64];
    // One zero BIT.
    odd[0] = CRC64_NVME_POLY;
    let mut row = 1u64;
    for n in 1..64 {
        odd[n] = row;
        row <<= 1;
    }
    gf2_square(&mut even, &odd); // two bits
    gf2_square(&mut odd, &even); // four bits
    let mut crc = crc_a;
    let mut len = len_b;
    loop {
        gf2_square(&mut even, &odd); // eight bits == one zero BYTE
        if len & 1 == 1 {
            crc = gf2_times(&even, crc);
        }
        len >>= 1;
        if len == 0 {
            break;
        }
        gf2_square(&mut odd, &even);
        if len & 1 == 1 {
            crc = gf2_times(&odd, crc);
        }
        len >>= 1;
        if len == 0 {
            break;
        }
    }
    crc ^ crc_b
}

/// One-shot convenience.
pub fn crc64_nvme(data: &[u8]) -> u64 {
    let mut c = Crc64Nvme::new();
    c.update(data);
    c.finalize()
}

/// The inverse of [`crc64_to_b64`]. `None` for anything that is not a
/// 12-character base64 encoding of exactly 8 bytes — a caller that gets
/// `None` has been handed something other than a CRC-64, and guessing a
/// number for it would validate a publish against a checksum nobody
/// computed.
pub fn crc64_from_b64(s: &str) -> Option<u64> {
    const AL: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
    let s = s.trim();
    if s.len() != 12 || !s.ends_with('=') {
        return None;
    }
    let mut acc: u32 = 0;
    let mut bits = 0u32;
    let mut out: Vec<u8> = Vec::with_capacity(8);
    for c in s.bytes() {
        if c == b'=' {
            break;
        }
        let v = AL.iter().position(|&a| a == c)? as u32;
        acc = (acc << 6) | v;
        bits += 6;
        if bits >= 8 {
            bits -= 8;
            out.push((acc >> bits) as u8);
        }
    }
    if out.len() != 8 {
        return None;
    }
    let mut b = [0u8; 8];
    b.copy_from_slice(&out);
    Some(u64::from_be_bytes(b))
}

/// S3's wire form: base64 of the checksum's 8 big-endian bytes.
pub fn crc64_to_b64(crc: u64) -> String {
    const AL: &[u8; 64] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
    let bytes = crc.to_be_bytes();
    let mut out = String::with_capacity(12);
    for chunk in bytes.chunks(3) {
        let b = [chunk[0], *chunk.get(1).unwrap_or(&0), *chunk.get(2).unwrap_or(&0)];
        let n = u32::from_be_bytes([0, b[0], b[1], b[2]]);
        out.push(AL[(n >> 18) as usize & 63] as char);
        out.push(AL[(n >> 12) as usize & 63] as char);
        out.push(if chunk.len() > 1 { AL[(n >> 6) as usize & 63] as char } else { '=' });
        out.push(if chunk.len() > 2 { AL[n as usize & 63] as char } else { '=' });
    }
    out
}

// ── stamps and metadata ──────────────────────────────────────────────

/// The A6 publish stamps (`x-amz-meta-flint-*` on S3; each backend maps
/// the bare keys to its metadata namespace). HEAD-based 412 arbitration
/// keys off these — they are how a torn OWN flush is told apart from a
/// genuinely foreign object.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct GenerationStamps {
    /// (`gen` is reserved in edition 2024 — spelled out.)
    pub generation: u64,
    pub epoch: u64,
    pub flush_uuid: String,
    /// The coherence provenance of a lean manifest citation
    /// (boundary-verbs plan §2.4.1): `sentinel`, `quiescence`,
    /// `forced-lag-cap`, `forced-backlog-cap`, `cadence`, `drain`,
    /// `recovered`. Stamped on the OBJECT so a downstream consumer —
    /// checkout, a sibling cluster's sync, HITL — can tell a
    /// declared-coherent citation from a forced possibly-torn one from
    /// the bucket alone, for the price of a HEAD rather than a GET of a
    /// manifest that can run to hundreds of MiB. `None` on every
    /// non-manifest object.
    pub boundary_source: Option<String>,
    /// A12: POSIX metadata riding on the object, so a bucket reader —
    /// and the DR import — can restore mode/ownership/mtime without
    /// the manifest. OPTIONAL both ways: absence never makes an object
    /// foreign (steps 4–11 published without it).
    pub posix: Option<PosixStamps>,
}

/// The A12 per-object POSIX stamps (`flint-mode` is octal; times are
/// unix seconds). Enumerated NON-goals of the round-trip: hard-link
/// structure, sockets/FIFOs/devices, and sparseness — see
/// `tier::manifest`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct PosixStamps {
    /// Full st_mode (type bits included; restore applies the 07777).
    pub mode: u32,
    pub uid: u32,
    pub gid: u32,
    pub mtime_unix: i64,
}

impl PosixStamps {
    pub const META_MODE: &'static str = "flint-mode";
    pub const META_UID: &'static str = "flint-uid";
    pub const META_GID: &'static str = "flint-gid";
    pub const META_MTIME: &'static str = "flint-mtime";

    #[cfg(unix)]
    pub fn from_metadata(m: &std::fs::Metadata) -> PosixStamps {
        use std::os::unix::fs::MetadataExt;
        PosixStamps { mode: m.mode(), uid: m.uid(), gid: m.gid(), mtime_unix: m.mtime() }
    }

    fn to_meta(self) -> Vec<(String, String)> {
        vec![
            (Self::META_MODE.into(), format!("{:o}", self.mode)),
            (Self::META_UID.into(), self.uid.to_string()),
            (Self::META_GID.into(), self.gid.to_string()),
            (Self::META_MTIME.into(), self.mtime_unix.to_string()),
        ]
    }

    /// Lenient: any stamp absent/malformed ⇒ None as a set (never a
    /// foreignness signal).
    pub fn from_meta(meta: &HashMap<String, String>) -> Option<PosixStamps> {
        Some(PosixStamps {
            mode: u32::from_str_radix(meta.get(Self::META_MODE)?, 8).ok()?,
            uid: meta.get(Self::META_UID)?.parse().ok()?,
            gid: meta.get(Self::META_GID)?.parse().ok()?,
            mtime_unix: meta.get(Self::META_MTIME)?.parse().ok()?,
        })
    }
}

impl GenerationStamps {
    pub const META_GEN: &'static str = "flint-gen";
    pub const META_EPOCH: &'static str = "flint-epoch";
    pub const META_FLUSH_UUID: &'static str = "flint-flush-uuid";
    pub const META_BOUNDARY_SOURCE: &'static str = "flint-boundary-source";

    pub fn to_meta(&self) -> Vec<(String, String)> {
        let mut v = vec![
            (Self::META_GEN.into(), self.generation.to_string()),
            (Self::META_EPOCH.into(), self.epoch.to_string()),
            (Self::META_FLUSH_UUID.into(), self.flush_uuid.clone()),
        ];
        if let Some(b) = &self.boundary_source {
            v.push((Self::META_BOUNDARY_SOURCE.into(), b.clone()));
        }
        if let Some(p) = self.posix {
            v.extend(p.to_meta());
        }
        v
    }

    /// Parse stamps back out of object metadata; None if any IDENTITY
    /// stamp is absent or malformed (an unstamped object is by
    /// definition foreign). The posix set parses leniently.
    pub fn from_meta(meta: &HashMap<String, String>) -> Option<GenerationStamps> {
        Some(GenerationStamps {
            generation: meta.get(Self::META_GEN)?.parse().ok()?,
            epoch: meta.get(Self::META_EPOCH)?.parse().ok()?,
            flush_uuid: meta.get(Self::META_FLUSH_UUID)?.clone(),
            boundary_source: meta.get(Self::META_BOUNDARY_SOURCE).cloned(),
            posix: PosixStamps::from_meta(meta),
        })
    }
}

/// One version of one key (boundary-verbs plan D8's reclamation and DR
/// surfacing).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ListedVersion {
    pub key: String,
    pub version_id: String,
    pub etag: String,
    pub size: u64,
    pub is_current: bool,
    /// A delete marker carries no bytes; it makes the key read absent
    /// while every prior version stays fetchable by id.
    pub is_delete_marker: bool,
    pub last_modified_unix: Option<u64>,
}

/// What HEAD/GET/publish return about an object.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ObjectMeta {
    pub etag: String,
    pub size: u64,
    /// Full-object CRC-64/NVME in wire (base64) form, when the store
    /// has one (objects written by this tier always do).
    pub crc64_b64: Option<String>,
    /// User metadata with backend prefixes stripped (bare keys, e.g.
    /// `flint-gen`).
    pub meta: HashMap<String, String>,
    pub last_modified_unix: Option<u64>,
    /// Backend storage class (S3: None ≡ STANDARD). The A11 IA
    /// copy-source guard reads this: a non-Standard base pays
    /// retrieval on every CLEAN byte copied, silently inverting the
    /// tier's saving — the flusher refuses BaseCopy from it.
    pub storage_class: Option<String>,
    /// The object VERSION this metadata describes, on a versioned
    /// bucket (boundary-verbs plan D7). S3 already returns
    /// `x-amz-version-id` on PUT and this crate used to discard it;
    /// gated citation needs it and therefore costs no extra request.
    ///
    /// `None` means the backend returned no version id — an
    /// unversioned bucket, or a proxy that STRIPS the header. The
    /// second case is why gated mode probes for it and REFUSES rather
    /// than degrading: falling back to etag semantics on a key whose
    /// current version is uncited is precisely the torn view the mode
    /// exists to prevent.
    pub version_id: Option<String>,
}

impl ObjectMeta {
    /// May this object serve as a server-side copy source without
    /// per-byte retrieval charges? (A11's IA copy-source guard.)
    pub fn copy_source_allowed(&self) -> bool {
        matches!(self.storage_class.as_deref(), None | Some("STANDARD"))
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ListedObject {
    pub key: String,
    pub size: u64,
    pub etag: String,
    pub last_modified_unix: Option<u64>,
}

/// An in-progress multipart assembly (A9's hygiene surface).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingUpload {
    pub key: String,
    pub upload_id: String,
    pub initiated_unix: Option<u64>,
}

// ── publish conditions ───────────────────────────────────────────────

/// A6: every publish is conditional. There is deliberately NO
/// unconditional variant — a caller wanting one has an upstream design
/// error.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PutCondition {
    /// Publish generation g+1 only if the object is still generation
    /// g's bytes (its ETag). The 412 this produces is an internal
    /// fencing event — see `tier::arbitrate`.
    IfMatch(String),
    /// First generation of a new key: fail if ANY object exists
    /// (closes the create race with outside writers).
    IfNoneMatchAny,
    /// No precondition: create, or overwrite whatever is there.
    ///
    /// Deliberately narrow. It exists for CONTENT-ADDRESSED objects,
    /// where the rewrite is byte-identical and the POINT is to refresh
    /// the object's age rather than to change its contents: an adopted
    /// manifest chunk is an old object that a new publish is about to
    /// make live, and a sweep that reads age would otherwise collect it
    /// as the orphan it still looks like (flint-lean chunked-manifest
    /// design §8.1, machine-checked by `LeanChunkGCAdoptSkips`).
    ///
    /// Never reach for this where a concurrent writer's content could
    /// DIFFER from yours — that is what the other two variants are for,
    /// and losing that distinction is how a last-writer-wins overwrite
    /// gets reintroduced.
    Unconditional,
}

// ── generation assembly ──────────────────────────────────────────────

/// One part of a composed generation, in object order.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum PartSource {
    /// Dirty range: bytes [offset, offset+len) read from the local
    /// file and uploaded.
    Local { offset: u64, len: u64 },
    /// Clean range: server-side copy of [offset, offset+len) from the
    /// base generation object at the SAME key, guarded by
    /// copy-source-if-match on the base ETag (detects foreign bucket
    /// overwrite; structurally cannot detect local staleness — that is
    /// capture's job, finding C5).
    BaseCopy { offset: u64, len: u64 },
}

/// The flusher's order for one generation publish.
#[derive(Debug, Clone)]
pub struct ComposeSpec<'a> {
    pub key: &'a str,
    /// Local source of `Local` parts. Clean ranges are byte-identical
    /// local and remote by definition, so `crc64` is computable from
    /// this file alone.
    pub local_path: &'a std::path::Path,
    /// Contiguous from offset 0, sizes within the backend's part
    /// granularity (the A11 part-size grid upstream owns that).
    pub parts: Vec<PartSource>,
    /// Where the base generation LIVES, when it is not `key` — the A7
    /// re-key flush publishes a renamed file under its new key while
    /// clean ranges still copy from the old one. None = same as `key`.
    pub base_key: Option<&'a str>,
    /// Base generation's ETag — required iff any part is `BaseCopy`.
    pub base_etag: Option<String>,
    pub condition: PutCondition,
    pub stamps: GenerationStamps,
    /// Full-object CRC-64/NVME of the composed content, from local
    /// truth. Backends hand it to the store for SERVER-SIDE validation
    /// at publish: a mismatch fails the publish, never warns.
    ///
    /// `None` asks the store to accumulate it from the `Local` parts
    /// as it reads them for upload. Parts go up in order, one at a
    /// time, so a streaming CRC over them in that loop IS the CRC of
    /// the object — the same local truth, without a second pass over
    /// the file (a 40 GiB pack's pre-pass was ~70 s on forge's push
    /// path, 2026-09-05). A `BaseCopy` part is never read locally, so
    /// a spec with one and `None` is refused before any part moves.
    pub crc64: Option<u64>,
    /// Bytes landed so far, for a caller that has to prove LIVENESS
    /// while a long compose runs: forge's lease renewer renews a push
    /// only while this advances, so a wedged upload lets the token go
    /// quiet and a challenger take over — which is the one takeover a
    /// wedged holder can get. Advanced by `len` after every part,
    /// `Local` and `BaseCopy` alike. `None` = nobody is watching.
    pub progress: Option<std::sync::Arc<std::sync::atomic::AtomicU64>>,
}

impl ComposeSpec<'_> {
    /// Record `bytes` more of the object as landed.
    pub fn note_progress(&self, bytes: u64) {
        if let Some(p) = &self.progress {
            p.fetch_add(bytes, std::sync::atomic::Ordering::Relaxed);
        }
    }
}

// ── errors ───────────────────────────────────────────────────────────

#[derive(Debug, thiserror::Error)]
pub enum StoreError {
    /// 412 — the guarded condition failed. Route to arbitration, never
    /// to an operator runbook (A6).
    #[error("precondition failed (412): {0}")]
    PreconditionFailed(String),
    /// 409 — a concurrent conditional writer; retry-after-arbitrate.
    #[error("conditional conflict (409): {0}")]
    Conflict(String),
    #[error("not found: {0}")]
    NotFound(String),
    /// The assembly was fenced (aborted) under us — A8's takeover
    /// sweep produces exactly this at a deposed hub's Complete.
    #[error("no such upload: {0}")]
    NoSuchUpload(String),
    /// Server-side content validation failed — the store computed a
    /// different full-object checksum than local truth claims.
    #[error("checksum mismatch: {0}")]
    ChecksumMismatch(String),
    /// 401/403, or an S3 code that means the same thing: the request
    /// was well-formed and we are not allowed to make it.
    ///
    /// Split out of `Other` because the caller's response is different
    /// in kind. A transient 5xx or a connection reset is "the store is
    /// having a moment, keep trying"; this is "the credentials are
    /// wrong, expired, or the policy changed", and no amount of retrying
    /// fixes it. It reached the epoch heartbeat as an ordinary renew
    /// failure, so a rotated key produced a fence and an exit whose log
    /// line talked about lease windows — pointing an operator at the
    /// bucket, at contention, at anything but the credential that had
    /// actually expired.
    ///
    /// `RequestTimeTooSkewed` belongs here too: S3 answers 403, and the
    /// cause is the node's clock rather than the key, which is
    /// diagnosable only if the error says so.
    #[error("not authorized: {0}")]
    Auth(String),
    #[error("store: {0}")]
    Other(String),
}

pub type StoreResult<T> = Result<T, StoreError>;

// ── the epoch surface (A8; the step-7 state machine drives these) ────

/// Observed epoch state (read side — takeover judgment input).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EpochState {
    pub holder_id: String,
    pub epoch: u64,
    /// Backend token guarding the next transition (S3: the epoch
    /// object's ETag; Azure: the lease id).
    pub token: String,
    /// The store's own clock for the last renewal (S3 Last-Modified) —
    /// A8: takeover is judged against the STORE's clock, not ours.
    pub last_renew_unix: Option<u64>,
    /// An opaque observed-state echo the holder wrote with its last
    /// heartbeat, if any ([`LeaseEcho`]). Read-only to everyone but the
    /// holder: it is the ONE fleet-visible surface that reports what a
    /// syncer is actually *doing* rather than what its spec asked for,
    /// and it costs nothing — the heartbeat CAS was already happening.
    /// `None` = the holder wrote none (an older binary, a clean
    /// release, or a backend that cannot carry it).
    pub echo: Option<String>,
    /// The holder shut down cleanly and will never publish under this
    /// epoch again — a successor supersedes immediately instead of
    /// waiting out the lease. Written by [`ObjectStore::epoch_release`].
    pub released: bool,
}

/// A held lease (write side).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EpochLease {
    pub holder_id: String,
    pub epoch: u64,
    pub token: String,
}

/// What a live lean syncer echoes into its lease-heartbeat cell
/// (boundary-verbs plan §2.6). The schema lives here, beside the cell
/// that carries it, because the WRITER (the syncer) and the READER
/// (the operator) are in different crates and a duplicated mirror
/// struct on either side would drift silently — which is the failure
/// this field exists to detect, not to reproduce.
///
/// It answers one question no spec field can: is the binary in that pod
/// actually running the mode the CR asked for? An old syncer reads a
/// FIXED env list, so `FLINT_SYNC_BOUNDARY_MODE=gated` reaching a
/// pre-boundary binary is silently ignored and the workspace runs fused
/// cadence behind a green condition.
#[derive(Debug, Clone, PartialEq, Eq, Default, serde::Serialize, serde::Deserialize)]
pub struct LeaseEcho {
    /// The syncer binary's own version — the mixed-fleet tell.
    #[serde(default)]
    pub syncer_version: String,
    /// `.flint/capabilities.json`'s protocol number.
    #[serde(default)]
    pub protocol: u32,
    /// The mode the syncer is RUNNING, not the mode it was asked for.
    #[serde(default)]
    pub active_boundary_mode: String,
    /// The last citation this syncer installed.
    #[serde(default)]
    pub last_cited_seq: u64,
    #[serde(default)]
    pub last_cited_unix: u64,
    /// Durable-but-invisible work standing right now — the gated
    /// exposure, per workspace, with no metrics stack in the picture.
    #[serde(default)]
    pub staged_uncited_count: u64,
    /// Whether the boundary verbs are live in this workspace. The
    /// pre-flight that turns them off (an app already owns `.flint/`)
    /// runs INSIDE the pod against the agent's own tree, so this is the
    /// only surface on which an operator can see the answer.
    #[serde(default)]
    pub sentinel_verbs_active: bool,
    /// `None` = exposition not enabled. `Some(false)` = enabled and the
    /// port was TAKEN — the workspace is fully operable and nothing is
    /// scraping it, which is a degradation only this field can report.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub metrics_bound: Option<bool>,
}

// ── the trait ────────────────────────────────────────────────────────

#[async_trait]
pub trait ObjectStore: Send + Sync {
    /// Whole-object publish (below the multipart threshold).
    /// `crc64` is validated server-side; stamps ride as metadata.
    async fn put_whole(
        &self,
        key: &str,
        body: Bytes,
        condition: &PutCondition,
        stamps: &GenerationStamps,
        crc64: u64,
    ) -> StoreResult<ObjectMeta>;

    /// Assemble and publish one generation from local dirty bytes plus
    /// server-side clean copies. MUST abort its partial assembly on
    /// every failure path (A9) — an error return means nothing is left
    /// pending under the key by this call.
    /// Server-side copy of a WHOLE object to a DIFFERENT key.
    ///
    /// Exists because a key IS a path in every layout this crate serves,
    /// so moving a file moves bytes. Without it a rename is a download
    /// and a re-upload, and `PartSource::BaseCopy` cannot stand in: it
    /// copies ranges from the SAME key by construction, so a destination
    /// that has no generation of its own has nothing to copy from.
    ///
    /// Two rules the implementations must agree on, and the memory
    /// double must model or it will pass a test the real store fails:
    ///
    /// - the destination's full-object CRC **is** the source's. A copy
    ///   is byte-identical; a checksum that changed means the bytes did.
    /// - the destination's **stamps are not** the source's. `generation`,
    ///   `epoch` and `flush_uuid` describe a publish, and inheriting them
    ///   files one object's history under another object's key — a
    ///   reader asking "which flush wrote this?" would be told about a
    ///   flush that wrote something else.
    ///
    /// `src_if_match` guards the SOURCE: copying an object that moved
    /// since the caller read it copies something the caller never saw.
    /// `condition` guards the DESTINATION, exactly as on `put_whole`.
    async fn copy_object(
        &self,
        src_key: &str,
        src_if_match: Option<&str>,
        dst_key: &str,
        condition: &PutCondition,
        stamps: &GenerationStamps,
    ) -> StoreResult<ObjectMeta>;

    async fn compose_generation(&self, spec: &ComposeSpec<'_>) -> StoreResult<ObjectMeta>;

    async fn head(&self, key: &str) -> StoreResult<ObjectMeta>;

    /// Whole-object read, optionally guarded (hydration always guards:
    /// a 412 here is S3-WINS — the foreign-overwrite posture, opposite
    /// of the publish path).
    async fn get_whole(&self, key: &str, if_match: Option<&str>) -> StoreResult<(ObjectMeta, Bytes)>;

    /// Ranged read (step 11's streaming restore — a multi-GiB
    /// hydration must never buffer the whole object). ALWAYS guarded:
    /// each chunk's If-Match pins the same object version end-to-end;
    /// a 412 mid-restore is the S3-wins foreign-overwrite signal. The
    /// caller never requests past the object's end.
    async fn get_range(
        &self,
        key: &str,
        offset: u64,
        len: u64,
        if_match: &str,
    ) -> StoreResult<Bytes>;

    /// The same ranged read, WITHOUT flattening the response into one
    /// contiguous buffer.
    ///
    /// `get_range` ends in `AggregatedBytes::into_bytes`, which for any
    /// multi-frame body — i.e. always — takes `SegmentedBuf`'s general
    /// branch: `BytesMut::with_capacity(len); res.put(self.take(len))`.
    /// That is one full-size allocation and one full-size copy per
    /// range, on top of the frames the SDK is already holding. At a 16
    /// MiB chunk the allocation goes through mmap and is returned on
    /// free, so the next range faults in fresh zeroed pages: measured
    /// on i4i.large as system time beating user time 6:1.
    ///
    /// The DEFAULT deliberately wraps `get_range`, so a store with
    /// nothing to stream (the in-memory double) behaves exactly as it
    /// did and no implementation is forced to care.
    async fn get_range_segments(
        &self,
        key: &str,
        offset: u64,
        len: u64,
        if_match: &str,
    ) -> StoreResult<Vec<Bytes>> {
        Ok(vec![self.get_range(key, offset, len, if_match).await?])
    }

    async fn list(&self, prefix: &str) -> StoreResult<Vec<ListedObject>>;

    async fn delete(&self, key: &str) -> StoreResult<()>;

    // ── version-scoped operations (boundary-verbs plan D7/D8) ────────
    //
    // On a versioned bucket an in-place PUT destroys nothing: the cited
    // generation survives as a version, the manifest cites
    // (key, version_id, etag), and readers resolve the cited version
    // exactly. The version id ADDRESSES; the etag ATTESTS.
    //
    // Backends that cannot express these answer `Other` and the caller
    // refuses gated mode — never degrades into it.

    /// HEAD one specific version.
    async fn head_version(&self, key: &str, version_id: &str) -> StoreResult<ObjectMeta> {
        let _ = (key, version_id);
        Err(StoreError::Other("this backend has no version-scoped HEAD".into()))
    }

    /// GET one specific version. This is the read `pinned_reads`
    /// citations resolve through (D13): under a gated citation flint's
    /// readers resolve EXCLUSIVELY by the cited version and never
    /// S3-wins-adopt the current one — without which every gated
    /// checkout would 412 on its own staged bytes and adopt them.
    async fn get_version(&self, key: &str, version_id: &str) -> StoreResult<(ObjectMeta, Bytes)> {
        let _ = (key, version_id);
        Err(StoreError::Other("this backend has no version-scoped GET".into()))
    }

    /// DELETE one specific version — the exact reclamation flint's own
    /// per-citation GC uses. Free, and the ONLY reaper that can tell
    /// cited from uncited; lifecycle cannot, which is why the noncurrent
    /// rule is a long backstop rather than the mechanism (D8).
    async fn delete_version(&self, key: &str, version_id: &str) -> StoreResult<()> {
        let _ = (key, version_id);
        Err(StoreError::Other("this backend has no version-scoped DELETE".into()))
    }

    /// Every version under a prefix, current and noncurrent, oldest
    /// first per key. The claim-time/DR fallback when `orphans.json` is
    /// missing or stale — the expensive path, which is why the durable
    /// summary is written eagerly.
    async fn list_versions(&self, prefix: &str) -> StoreResult<Vec<ListedVersion>> {
        let _ = prefix;
        Err(StoreError::Other("this backend cannot list object versions".into()))
    }

    /// A time-limited URL that GETs `key` with no credential of the
    /// caller's own.
    ///
    /// One consumer, and it is worth naming because it is what makes
    /// the URL's lifetime a design constraint rather than a parameter:
    /// flint forge advertises a clone BUNDLE this way, so that a
    /// thousand agents cloning at once pull the bytes from the object
    /// store instead of through one pod's NIC (forge design §8). The
    /// URL is handed to a git client, which may hold it for the length
    /// of a clone, and S3 caps a SigV4 presigned URL at seven days —
    /// so the caller must re-sign on a schedule shorter than the TTL it
    /// asked for, and this returns the URL rather than storing it
    /// anywhere.
    ///
    /// Defaulted to a refusal: a backend that cannot presign must say
    /// so, and the caller must fall back to serving the bytes itself
    /// rather than advertising a URL that will not resolve.
    async fn presign_get(&self, key: &str, ttl_secs: u64) -> StoreResult<String> {
        let _ = (key, ttl_secs);
        Err(StoreError::Other("this backend cannot presign a URL".into()))
    }

    /// A time-limited URL that PUTs `key` with no credential of the
    /// caller's own.
    ///
    /// The counterpart of [`ObjectStore::presign_get`], and it exists
    /// for the same consumer: flint forge's LFS batch API hands a git
    /// client one of these so a multi-gigabyte checkpoint travels
    /// straight to the object store and never crosses the repository
    /// server's network interface.
    ///
    /// **The URL carries no content check.** Whoever holds it may put
    /// anything at that key until it expires, so the caller must issue
    /// it only for a key whose NAME already constrains the content —
    /// a content-addressed one — and must verify after the fact.
    async fn presign_put(&self, key: &str, ttl_secs: u64) -> StoreResult<String> {
        let _ = (key, ttl_secs);
        Err(StoreError::Other("this backend cannot presign a URL".into()))
    }

    /// Every lifecycle rule on the bucket, reduced to [`LifecycleView`].
    ///
    /// Defaulted to a refusal rather than to an empty list: "no rules"
    /// and "I cannot see the rules" must never be the same answer here.
    /// An empty list is a POSITIVE claim that nothing is reaping
    /// noncurrent versions, and gated mode is accepted on the strength
    /// of it (D8).
    async fn lifecycle_rules(&self) -> StoreResult<Vec<LifecycleView>> {
        Err(StoreError::Other("this backend cannot read lifecycle rules".into()))
    }

    /// Provision the noncurrent-version retention backstop on `prefix`
    /// with read-merge-append, and report what is installed.
    ///
    /// `PutBucketLifecycleConfiguration` is FULL-REPLACE: an
    /// implementation that writes its rule alone silently deletes the
    /// MPU-abort rule and every rule the customer owns. Read, merge,
    /// append — the discipline `bootstrap_lifecycle` already follows.
    async fn ensure_noncurrent_retention(
        &self,
        prefix: &str,
        days: u64,
    ) -> StoreResult<RetentionOutcome> {
        let _ = (prefix, days);
        Err(StoreError::Other("this backend cannot write lifecycle rules".into()))
    }

    /// A9 hygiene: every in-progress assembly under the prefix.
    async fn list_uploads(&self, prefix: &str) -> StoreResult<Vec<PendingUpload>>;

    /// Abort one assembly. Aborting an already-absent upload is Ok —
    /// the sweep and the abort-on-failure paths both race the
    /// lifecycle rule.
    async fn abort_upload(&self, key: &str, upload_id: &str) -> StoreResult<()>;

    /// A9: verify/create the bucket posture (lifecycle abort rule,
    /// versioning, encryption notes, IAM probe). Named prefix so
    /// probes stay under the tier's namespace.
    async fn bootstrap(&self, prefix: &str) -> StoreResult<BootstrapReport>;

    // Epoch (A8). S3 implements CAS re-PUT on the epoch object; Azure
    // will use native blob leases (strictly better — A13).

    async fn epoch_read(&self, key: &str) -> StoreResult<Option<EpochState>>;

    /// Claim: `supersede: None` creates (If-None-Match:*); `Some`
    /// replaces the OBSERVED state via CAS on its token — the caller
    /// (step 7) is responsible for having JUDGED that holder dead
    /// against the store's clock. The new epoch is observed+1.
    async fn epoch_acquire(
        &self,
        key: &str,
        holder_id: &str,
        supersede: Option<&EpochState>,
    ) -> StoreResult<EpochLease>;

    /// Heartbeat CAS. `PreconditionFailed` means deposed: self-fence.
    ///
    /// `echo` rides the same write (see [`EpochState::echo`]) — the
    /// observed-state surface is deliberately a passenger on a request
    /// the holder already pays for every ≤30 s. A caller with nothing
    /// to report passes `None`, which CLEARS any previous echo rather
    /// than preserving it: a stale echo read as live is exactly the
    /// mixed-version hole the field exists to close.
    async fn epoch_renew(
        &self,
        key: &str,
        lease: &EpochLease,
        echo: Option<&str>,
    ) -> StoreResult<EpochLease>;

    /// Mark the cell released: a clean handoff. The epoch NUMBER must
    /// survive (deleting the cell would restart numbering at 1 and
    /// break the monotonicity every publish stamp depends on), and the
    /// write must be guarded on `lease`'s own token so a deposed
    /// holder cannot mark a live successor's cell.
    async fn epoch_release(&self, key: &str, lease: &EpochLease) -> StoreResult<()>;

    /// Backend part granularity for the A11 part-size grid.
    fn min_part_size(&self) -> u64;
    fn max_parts(&self) -> usize;
}

/// One lifecycle rule, reduced to what VERSION RETENTION cares about
/// (lean boundary-verbs plan D8).
///
/// Gated staging makes the CITED version noncurrent the moment a newer
/// generation stages, so a `NoncurrentVersionExpiration` rule covering
/// `<prefix>/files/` runs a clock against live cited data — and never
/// against the newest uncited bytes, which are current and which no
/// lifecycle rule can reach. That inversion is why the rules have to be
/// READABLE and not merely writable: a customer sitting at
/// noncurrent-days=1 arms the destroyer without touching flint at all.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LifecycleView {
    pub id: String,
    pub enabled: bool,
    /// The rule's prefix filter; `""` = the whole bucket.
    pub prefix: String,
    /// `NoncurrentVersionExpiration.NoncurrentDays`, when the rule has
    /// one. `None` = this rule expires no noncurrent versions.
    pub noncurrent_days: Option<u64>,
    /// Whether the rule also expires orphaned delete markers.
    pub expired_delete_marker: bool,
}

/// What [`ObjectStore::ensure_noncurrent_retention`] found or did.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RetentionOutcome {
    pub rule_id: String,
    pub noncurrent_days: u64,
    /// False ⇒ a conforming rule was already installed and nothing was
    /// written (the read-merge-append path never runs blind).
    pub created: bool,
}

/// What bootstrap found/did (A9). `errors` non-empty ⇒ the tier must
/// refuse to start; `warnings` are loud-degrade.
#[derive(Debug, Default, Clone)]
pub struct BootstrapReport {
    pub notes: Vec<String>,
    pub warnings: Vec<String>,
    pub errors: Vec<String>,
}

impl BootstrapReport {
    pub fn ok(&self) -> bool {
        self.errors.is_empty()
    }
}

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

    #[test]
    fn crc64_nvme_catalog_check_value() {
        // CRC-64/NVME check input from the reveng catalog.
        assert_eq!(crc64_nvme(b"123456789"), 0xAE8B_1486_0A79_9888);
        // Streaming across split updates must agree with one-shot.
        let mut c = Crc64Nvme::new();
        c.update(b"1234");
        c.update(b"56789");
        assert_eq!(c.finalize(), 0xAE8B_1486_0A79_9888);
        assert_eq!(crc64_nvme(b""), 0);
    }

    #[test]
    fn crc64_slice_by_8_agrees_with_the_byte_at_a_time_form() {
        // `update` consumes 8-byte words with a byte-at-a-time tail, and
        // it is a STREAMING call — so the word/tail boundary can land
        // anywhere a caller happens to split. That makes split ALIGNMENT
        // load-bearing in a way the 4/5 split above never exercised:
        // every arbitrary split must give the one-shot digest.
        fn byte_at_a_time(data: &[u8]) -> u64 {
            let t = crc64_table();
            let mut crc = u64::MAX;
            for &b in data {
                crc = t[((crc ^ b as u64) & 0xFF) as usize] ^ (crc >> 8);
            }
            crc ^ u64::MAX
        }
        let data: Vec<u8> =
            (0..1000u32).map(|i| (i.wrapping_mul(2654435761) >> 13) as u8).collect();

        // Every length across the word boundary, against the reference.
        for n in 0..40usize {
            assert_eq!(crc64_nvme(&data[..n]), byte_at_a_time(&data[..n]), "len {n}");
        }
        assert_eq!(crc64_nvme(&data), byte_at_a_time(&data));

        // Every single split point of the full buffer.
        let want = crc64_nvme(&data);
        for cut in 0..data.len() {
            let mut c = Crc64Nvme::new();
            c.update(&data[..cut]);
            c.update(&data[cut..]);
            assert_eq!(c.finalize(), want, "split at {cut}");
        }

        // A three-way split at two unaligned points.
        let mut c = Crc64Nvme::new();
        c.update(&data[..13]);
        c.update(&data[13..457]);
        c.update(&data[457..]);
        assert_eq!(c.finalize(), want);
    }

    #[test]
    fn crc64_b64_is_8_bytes_big_endian_base64() {
        // 0xAE8B14860A799888 → bytes AE 8B 14 86 0A 79 98 88.
        assert_eq!(crc64_to_b64(0xAE8B_1486_0A79_9888), "rosUhgp5mIg=");
        assert_eq!(crc64_to_b64(0), "AAAAAAAAAAA=");
    }

    #[test]
    fn stamps_roundtrip_and_reject_partial() {
        let s = GenerationStamps { generation: 7, epoch: 3, flush_uuid: "u-1".into(), boundary_source: None, posix: None };
        let m: HashMap<String, String> = s.to_meta().into_iter().collect();
        assert_eq!(GenerationStamps::from_meta(&m), Some(s));
        let mut partial = m.clone();
        partial.remove(GenerationStamps::META_FLUSH_UUID);
        assert_eq!(
            GenerationStamps::from_meta(&partial),
            None,
            "an unstamped/partially-stamped object is foreign by definition"
        );
    }

    #[test]
    fn posix_stamps_roundtrip_and_stay_lenient() {
        let p = PosixStamps { mode: 0o100644, uid: 501, gid: 20, mtime_unix: 1_723_000_000 };
        let s = GenerationStamps {
            generation: 2,
            epoch: 1,
            flush_uuid: "u-2".into(),
            boundary_source: None,
            posix: Some(p),
        };
        let m: HashMap<String, String> = s.to_meta().into_iter().collect();
        assert_eq!(m.get(PosixStamps::META_MODE).map(String::as_str), Some("100644"));
        assert_eq!(GenerationStamps::from_meta(&m), Some(s));
        // A pre-step-12 object (identity stamps only) parses with
        // posix: None — NEVER foreign.
        let mut old = m.clone();
        old.remove(PosixStamps::META_UID);
        let parsed = GenerationStamps::from_meta(&old).expect("identity stamps intact");
        assert_eq!(parsed.posix, None);
    }
}

#[cfg(test)]
mod combine_tests {
    use super::{crc64_combine, crc64_nvme};

    /// The one-shot CRC over the concatenation is the oracle. Nothing
    /// here asserts a hand-computed constant: the existing
    /// implementation is the ground truth, and `combine` either agrees
    /// with it on every split or it is wrong.
    #[test]
    fn combining_two_parts_equals_hashing_the_whole() {
        // Non-uniform bytes: a run of one value would hide an operator
        // that shifted by the wrong number of positions.
        let whole: Vec<u8> = (0..4096u32).map(|i| (i.wrapping_mul(2654435761) >> 13) as u8).collect();
        for split in [0usize, 1, 2, 3, 7, 8, 9, 255, 256, 257, 1023, 2048, 4095, 4096] {
            let (a, b) = whole.split_at(split);
            assert_eq!(
                crc64_combine(crc64_nvme(a), crc64_nvme(b), b.len() as u64),
                crc64_nvme(&whole),
                "split at {split}"
            );
        }
    }

    #[test]
    fn combining_many_parts_folds_left_to_right() {
        let whole: Vec<u8> = (0..10_000u32).map(|i| (i ^ (i >> 5)) as u8).collect();
        // An 8 MiB-style grid in miniature: unequal parts, last one short.
        let sizes = [1usize, 999, 4096, 1, 4903]; // sums to exactly 10_000
        let mut off = 0usize;
        let mut acc: Option<(u64, u64)> = None; // (crc, len)
        for sz in sizes {
            let part = &whole[off..off + sz];
            let c = crc64_nvme(part);
            acc = Some(match acc {
                None => (c, sz as u64),
                Some((pc, plen)) => (crc64_combine(pc, c, sz as u64), plen + sz as u64),
            });
            off += sz;
        }
        let (crc, len) = acc.unwrap();
        assert_eq!(len, whole.len() as u64, "the grid must cover the object");
        assert_eq!(crc, crc64_nvme(&whole), "folded parts must equal one pass");
    }

    #[test]
    fn an_empty_tail_changes_nothing() {
        let a = b"some bytes";
        assert_eq!(crc64_combine(crc64_nvme(a), crc64_nvme(b""), 0), crc64_nvme(a));
    }
}

#[cfg(test)]
mod fold_order_tests {
    use super::{crc64_combine, crc64_nvme};

    /// The compose path sorts parts by number BEFORE folding, and this
    /// is why: `crc64_combine` composes A-then-B, so a grid folded in
    /// completion order rather than part order yields the checksum of a
    /// DIFFERENT object. Without this test the sort reads as being only
    /// about what CompleteMultipartUpload accepts.
    #[test]
    fn folding_out_of_order_yields_a_different_object() {
        let a = b"first-part-bytes";
        let b = b"second";
        let c = b"third-part!";
        let whole = [&a[..], &b[..], &c[..]].concat();

        let fold = |parts: &[&[u8]]| {
            let mut acc: Option<u64> = None;
            for p in parts {
                let x = crc64_nvme(p);
                acc = Some(match acc {
                    None => x,
                    Some(prev) => crc64_combine(prev, x, p.len() as u64),
                });
            }
            acc.unwrap()
        };

        assert_eq!(fold(&[a, b, c]), crc64_nvme(&whole), "part order IS the object");
        assert_ne!(
            fold(&[b, a, c]),
            crc64_nvme(&whole),
            "a shuffled fold must NOT coincidentally agree, or the sort proves nothing"
        );
    }
}