flint_store/

memory.rs

//! The in-process ObjectStore test double.
//!
//! Implements the FULL conditional contract — 412 on both put flavors,
//! copy-source-if-match at part-copy time, server-side CRC-64/NVME
//! validation at publish, MPU state with list/abort, epoch CAS — so
//! every backend-neutral tier test (arbitration, flush pipeline,
//! hydration) runs against semantics, not stubs. Failure injection
//! covers the two adversarial publish shapes A6's arbitration must
//! tell apart:
//!
//! - [`inject_torn_complete`]: the Complete LANDS server-side but the
//!   response is lost (network tear) — the object exists at g+1 with
//!   our stamps while the client saw an error.
//! - [`inject_crash_before_complete`]: the assembly dies before
//!   Complete — parts pending, nothing published, base object intact.
//!
//! [`inject_torn_complete`]: MemoryStore::inject_torn_complete
//! [`inject_crash_before_complete`]: MemoryStore::inject_crash_before_complete

use super::*;
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicU8, Ordering};
use std::sync::Mutex;

fn now_unix() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|d| d.as_secs())
        .unwrap_or(0)
}

#[derive(Clone)]
struct StoredObject {
    bytes: Bytes,
    etag: String,
    crc64: u64,
    meta: HashMap<String, String>,
    last_modified_unix: u64,
    /// The version's identity (boundary-verbs plan D7). Minted from a
    /// per-store COUNTER, never a uuid or a clock: a reproducible
    /// battery is the whole reason this double exists, and a
    /// nondeterministic id would make every version-citing assertion
    /// order-dependent in a way that only shows up under load.
    version_id: String,
    /// A delete marker. Under versioning a DELETE never destroys data:
    /// it makes the key read as absent while every prior version stays
    /// fetchable by id.
    deleted: bool,
}

impl StoredObject {
    fn to_meta(&self) -> ObjectMeta {
        ObjectMeta {
            etag: self.etag.clone(),
            size: self.bytes.len() as u64,
            crc64_b64: Some(crc64_to_b64(self.crc64)),
            meta: self.meta.clone(),
            last_modified_unix: Some(self.last_modified_unix),
            // The fake models STANDARD only; the IA guard's unit tests
            // exercise copy_allowed=false at the planner level.
            storage_class: None,
            version_id: Some(self.version_id.clone()),
        }
    }
}

struct Mpu {
    key: String,
    parts: BTreeMap<usize, Bytes>,
    initiated_unix: u64,
}

/// A key's version chain: oldest first, LAST is current. Nothing in
/// this double ever overwrites in place — that is the entire point of
/// modelling versions, since an in-place overwrite is the destructive
/// design versioned staging exists to avoid (§2.4.2).
#[derive(Default, Clone)]
struct VersionChain {
    versions: Vec<StoredObject>,
}

impl VersionChain {
    /// The current version, or None if the key reads as absent (empty
    /// chain, or a delete marker on top).
    fn current(&self) -> Option<&StoredObject> {
        match self.versions.last() {
            Some(o) if !o.deleted => Some(o),
            _ => None,
        }
    }
}

#[derive(Default)]
struct Inner {
    chains: BTreeMap<String, VersionChain>,
    uploads: HashMap<String, Mpu>,
    /// Deterministic version-id source (see `StoredObject::version_id`).
    version_seq: u64,
    /// The bucket's lifecycle rules. Modelled because D8's hazard is a
    /// rule the CUSTOMER already owns — the battery has to be able to
    /// plant one (`plant_lifecycle_rule`) and watch gated mode refuse.
    lifecycle: Vec<LifecycleView>,
}

impl Inner {
    fn next_version_id(&mut self) -> String {
        self.version_seq += 1;
        format!("v{:012}", self.version_seq)
    }

    /// The current version of `key`, or None if absent.
    fn current(&self, key: &str) -> Option<&StoredObject> {
        self.chains.get(key).and_then(|c| c.current())
    }

    /// Append a version and make it current.
    fn push(&mut self, key: &str, mut obj: StoredObject) -> String {
        let vid = self.next_version_id();
        obj.version_id = vid.clone();
        self.chains.entry(key.to_string()).or_default().versions.push(obj);
        vid
    }

    /// A keyed DELETE: a delete marker, never destruction.
    fn push_delete_marker(&mut self, key: &str) {
        if self.chains.get(key).map(|c| c.current().is_none()).unwrap_or(true) {
            return; // already absent: S3 still adds a marker; we keep the chain quiet
        }
        let vid = self.next_version_id();
        self.chains.entry(key.to_string()).or_default().versions.push(StoredObject {
            bytes: Bytes::new(),
            etag: String::new(),
            crc64: 0,
            meta: HashMap::new(),
            last_modified_unix: now_unix(),
            version_id: vid,
            deleted: true,
        });
    }

    fn version(&self, key: &str, version_id: &str) -> Option<&StoredObject> {
        self.chains
            .get(key)?
            .versions
            .iter()
            .find(|o| o.version_id == version_id && !o.deleted)
    }

    /// Version-scoped DELETE: removes exactly that version. Removing
    /// the current version promotes the one beneath it, exactly as S3
    /// does.
    fn remove_version(&mut self, key: &str, version_id: &str) {
        if let Some(chain) = self.chains.get_mut(key) {
            chain.versions.retain(|o| o.version_id != version_id);
            if chain.versions.is_empty() {
                self.chains.remove(key);
            }
        }
    }
}

const INJECT_NONE: u8 = 0;
const INJECT_TORN_COMPLETE: u8 = 1;
const INJECT_CRASH_BEFORE_COMPLETE: u8 = 2;
const INJECT_SWEPT: u8 = 3;

pub struct MemoryStore {
    inner: Mutex<Inner>,
    upload_seq: AtomicU64,
    /// One-shot failure injection for the NEXT compose publish.
    inject: AtomicU8,
    /// Set by the crash injection: the failing compose must LEAVE its
    /// orphan MPU (the A9 sweep's test state) instead of aborting it.
    leave_orphan: AtomicBool,
    /// Step-11 drill injections: counted get_range failures / stall.
    fail_get_range_count: AtomicU64,
    /// Counted HEAD failures — the sweep's transient-throttle drill.
    fail_head_count: AtomicU64,
    /// Counted `epoch_renew` failures that are NOT a 412: the store
    /// unreachable from the holder (forge X13's drill).
    fail_epoch_renew_count: AtomicU64,
    /// The next n whole PUTs LAND and then report 409. See
    /// `inject_put_lands_then_fails`.
    fail_put_after_land_count: AtomicU64,
    fail_put_after_land_key: Mutex<String>,
    stall_next_get_range_ms: AtomicU64,
    /// A latency every get_range pays, in ms: the double's stand-in for
    /// a network round trip, so a test can observe whether fetches
    /// OVERLAP rather than time them.
    get_range_delay_ms: AtomicU64,
    /// A latency every whole PUT pays, in ms: the write side's
    /// `get_range_delay_ms`, so a test can observe whether whole-body
    /// uploads OVERLAP — and how many bytes they hold while they do.
    put_whole_delay_ms: AtomicU64,
    /// A latency EVERY request pays, in ms (`inject_request_delay_ms`): the
    /// double's stand-in for a store's round trip, on every call a real
    /// store would send — GETs, HEADs, LISTs, PUTs, DELETEs, the lease
    /// cell's reads and writes — so a measurement can show what a chain of
    /// sequential requests costs against one that overlaps.
    request_delay_ms: AtomicU64,
    /// Requests inside that delay right now, and the most there have ever
    /// been at once: overlap, observed rather than timed.
    inflight_requests: AtomicU64,
    peak_requests: AtomicU64,
    /// While set, a LIST whose prefix contains this answers 403: a
    /// credential whose policy grants reads but not the listing.
    deny_list_matching: Mutex<Option<String>>,
    /// While set, every whole PUT parks at the door until
    /// `release_held_puts`: a stalled upload a test can run something
    /// BESIDE (a batch beside a fold's upload), with the ordering held
    /// by the gate rather than by a sleep the scheduler may outlast.
    hold_puts: AtomicBool,
    /// When set, only a whole PUT whose key contains this parks: a test
    /// can stall one upload while the lease's own writes go through.
    hold_puts_matching: Mutex<Option<String>>,
    held_puts: Mutex<Vec<tokio::sync::oneshot::Sender<()>>>,
    /// get_range calls in flight, and the most there have ever been at
    /// once: a fan-out's bound, observed from both sides.
    inflight_get_range: AtomicU64,
    peak_get_range: AtomicU64,
    /// Model a project-scoped proxy that STRIPS `x-amz-version-id`:
    /// every write still succeeds and reports no version. This is D8's
    /// silent-degradation hazard, and the only way to test that the
    /// conformance probe refuses instead of degrading.
    strip_version_ids: AtomicBool,
    /// Model a bucket whose lifecycle rules are readable but not
    /// writable (a scoped operator principal): the backstop cannot be
    /// provisioned, which is a DEGRADATION, not a torn view.
    fail_lifecycle_writes: AtomicBool,
    /// How a conditional write answers `If-Match` on a key that does not
    /// exist. Real S3 answers 404 `NoSuchKey`; Ozone 2.2.1 answers 412
    /// (both measured 2026-09-13 with the AWS CLI). The double used to give
    /// only Ozone's answer, so the syncer's "a vanished base is a create"
    /// rule — written for the 412 — was green here and wedged every
    /// barrier on S3 (drill host leg H2). Default: S3's 404.
    if_match_missing_412: AtomicBool,
    /// Per-operation call counts. Cost measurement is otherwise
    /// guesswork: the request SHAPE of a tick is the thing the plan
    /// prices, and it is not derivable from the code by reading.
    ops: Mutex<std::collections::BTreeMap<&'static str, u64>>,
    pub min_part: u64,
    pub max_parts: usize,
    /// Parts of one compose read and staged concurrently — the S3
    /// backend's `part_parallelism`, mirrored so the double runs the
    /// same window shape. 1 = sequential.
    part_parallelism: usize,
    /// The upload byte gate, when `with_upload_inflight_max_bytes` set
    /// one: the same charge at the same points as the S3 backend, so a
    /// test can read the bound off the gate's high-water mark.
    upload_gate: Option<std::sync::Arc<crate::gate::ByteGate>>,
}

/// Leaves the in-flight count on drop, so an early `return` still
/// leaves.
struct InFlight<'a>(&'a AtomicU64);
impl Drop for InFlight<'_> {
    fn drop(&mut self) {
        self.0.fetch_sub(1, Ordering::SeqCst);
    }
}

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

impl MemoryStore {
    /// Age an object's last-modified stamp. Liveness here is judged
    /// against the STORE's clock (A8), so a test about a dead holder
    /// has to be able to move that clock rather than sleep.
    pub fn backdate_epoch(&self, key: &str, secs: u64) {
        let mut inner = self.inner.lock().unwrap();
        if let Some(chain) = inner.chains.get_mut(key) {
            for v in chain.versions.iter_mut() {
                v.last_modified_unix = v.last_modified_unix.saturating_sub(secs);
            }
        }
    }

    /// Age an in-progress multipart upload's initiation stamp, so a
    /// test of an age-gated sweep need not sleep past its threshold.
    pub fn backdate_upload(&self, upload_id: &str, secs: u64) {
        if let Some(m) = self.inner.lock().unwrap().uploads.get_mut(upload_id) {
            m.initiated_unix = m.initiated_unix.saturating_sub(secs);
        }
    }

    /// Plant a lifecycle rule the way a customer's own fleet policy
    /// would. D8's destroyer is a rule flint never wrote.
    pub fn plant_lifecycle_rule(&self, rule: LifecycleView) {
        self.inner.lock().unwrap().lifecycle.push(rule);
    }

    pub fn new() -> Self {
        MemoryStore {
            ops: Mutex::new(Default::default()),
            inner: Mutex::new(Inner::default()),
            upload_seq: AtomicU64::new(1),
            inject: AtomicU8::new(INJECT_NONE),
            leave_orphan: AtomicBool::new(false),
            fail_get_range_count: AtomicU64::new(0),
            fail_head_count: AtomicU64::new(0),
            fail_epoch_renew_count: AtomicU64::new(0),
            fail_put_after_land_count: AtomicU64::new(0),
            fail_put_after_land_key: Mutex::new(String::new()),
            stall_next_get_range_ms: AtomicU64::new(0),
            get_range_delay_ms: AtomicU64::new(0),
            put_whole_delay_ms: AtomicU64::new(0),
            request_delay_ms: AtomicU64::new(0),
            inflight_requests: AtomicU64::new(0),
            peak_requests: AtomicU64::new(0),
            deny_list_matching: Mutex::new(None),
            hold_puts: AtomicBool::new(false),
            hold_puts_matching: Mutex::new(None),
            held_puts: Mutex::new(Vec::new()),
            inflight_get_range: AtomicU64::new(0),
            peak_get_range: AtomicU64::new(0),
            strip_version_ids: AtomicBool::new(false),
            fail_lifecycle_writes: AtomicBool::new(false),
            if_match_missing_412: AtomicBool::new(false),
            // Tiny granularity by default so tests compose small
            // files; S3's real limits live in the S3 backend.
            min_part: 1,
            max_parts: 10_000,
            part_parallelism: 1,
            upload_gate: None,
        }
    }

    /// Answer `If-Match` on a missing key with 412, as Ozone's S3 gateway
    /// does, instead of S3's 404 `NoSuchKey`. See `if_match_missing_412`.
    pub fn with_if_match_missing_as_412(self) -> Self {
        self.if_match_missing_412.store(true, Ordering::SeqCst);
        self
    }

    /// Stage this many parts of one compose concurrently (the S3
    /// backend's `with_part_parallelism`). Clamped to at least 1.
    pub fn with_part_parallelism(mut self, n: usize) -> Self {
        self.part_parallelism = n.max(1);
        self
    }

    /// Bound the bytes the upload path holds at once (the S3 backend's
    /// `with_upload_inflight_max_bytes`; see `gate.rs`). `0` = no gate.
    pub fn with_upload_inflight_max_bytes(mut self, n: u64) -> Self {
        self.upload_gate = (n > 0).then(|| std::sync::Arc::new(crate::gate::ByteGate::new(n)));
        self
    }

    /// Every whole PUT from now on sleeps `ms` first — the write side's
    /// `inject_get_range_delay_ms`: whole bodies that overlap are in
    /// flight together for the whole delay, so the upload byte gate's
    /// high-water mark reads the caller's bound rather than the
    /// scheduler's luck.
    pub fn inject_put_whole_delay_ms(&self, ms: u64) {
        self.put_whole_delay_ms.store(ms, Ordering::SeqCst);
    }

    /// Every request from now on sleeps `ms` first (0 turns it off). The
    /// per-kind delays above stay as they are and add to this one.
    pub fn inject_request_delay_ms(&self, ms: u64) {
        self.request_delay_ms.store(ms, Ordering::SeqCst);
    }

    /// The most requests that were ever inside the injected delay at once.
    /// Only a delay makes overlap observable: with none, a request never
    /// yields, so this stays at 0 or 1.
    pub fn peak_requests_in_flight(&self) -> u64 {
        self.peak_requests.load(Ordering::SeqCst)
    }

    pub fn reset_peak_requests_in_flight(&self) {
        self.peak_requests.store(0, Ordering::SeqCst);
    }

    /// Every LIST whose prefix contains `m` answers 403 (`StoreError::Auth`)
    /// from now on; `None` lets them through again.
    pub fn deny_list_matching(&self, m: Option<&str>) {
        *self.deny_list_matching.lock().unwrap() = m.map(str::to_string);
    }

    /// The injected round trip (`inject_request_delay_ms`).
    async fn pace(&self) {
        let ms = self.request_delay_ms.load(Ordering::SeqCst);
        if ms == 0 {
            return;
        }
        let now = self.inflight_requests.fetch_add(1, Ordering::SeqCst) + 1;
        self.peak_requests.fetch_max(now, Ordering::SeqCst);
        let _left = InFlight(&self.inflight_requests);
        tokio::time::sleep(std::time::Duration::from_millis(ms)).await;
    }

    /// Every write from here on reports no version id — a proxy that
    /// strips the header (D8's refusal arm).
    pub fn strip_version_ids(&self, on: bool) {
        self.strip_version_ids.store(on, Ordering::SeqCst);
    }

    /// Lifecycle writes fail from here on; reads keep working.
    pub fn fail_lifecycle_writes(&self, on: bool) {
        self.fail_lifecycle_writes.store(on, Ordering::SeqCst);
    }

    /// Next compose: Complete LANDS but the response is lost.
    pub fn inject_torn_complete(&self) {
        self.inject.store(INJECT_TORN_COMPLETE, Ordering::SeqCst);
    }

    /// Next compose: dies before Complete — MPU left pending.
    pub fn inject_crash_before_complete(&self) {
        self.inject.store(INJECT_CRASH_BEFORE_COMPLETE, Ordering::SeqCst);
    }

    /// Next compose: the operator's abort sweep takes the assembly
    /// before Complete — `NoSuchUpload`, S3's answer, and nothing lands.
    pub fn inject_compose_swept(&self) {
        self.inject.store(INJECT_SWEPT, Ordering::SeqCst);
    }

    /// Step-11 drills: the NEXT get_range fails with a transport
    /// error — a single failure is absorbed by the chunk-retry loop.
    pub fn inject_get_range_failure(&self) {
        self.inject_get_range_failures(1);
    }

    /// The next `n` get_range calls fail with a transport error —
    /// exceed the chunk-retry budget to force the truncate-back path.
    pub fn inject_get_range_failures(&self, n: u64) {
        self.fail_get_range_count.store(n, Ordering::SeqCst);
    }

    /// Every get_range from now on sleeps `ms` first. Unlike the
    /// one-shot stall this is persistent, and it is what makes a
    /// fan-out OBSERVABLE: calls that overlap are in flight together
    /// for the whole delay, so `peak_get_range_in_flight` reads the
    /// caller's bound rather than the scheduler's luck.
    pub fn inject_get_range_delay_ms(&self, ms: u64) {
        self.get_range_delay_ms.store(ms, Ordering::SeqCst);
    }

    /// Every whole PUT from now on parks until `release_held_puts`
    /// (`false` lets new PUTs through again; a PUT already parked stays
    /// parked). Checked at the start of each PUT, so a test can stall
    /// one task's upload, let the next caller's through at once, and
    /// release the stalled one when it chooses.
    pub fn inject_put_hold(&self, on: bool) {
        *self.hold_puts_matching.lock().unwrap() = None;
        self.hold_puts.store(on, Ordering::SeqCst);
    }

    /// Like [`inject_put_hold`], but only for keys containing
    /// `key_contains`. `inject_put_hold(false)` ends it.
    ///
    /// [`inject_put_hold`]: MemoryStore::inject_put_hold
    pub fn inject_put_hold_matching(&self, key_contains: &str) {
        *self.hold_puts_matching.lock().unwrap() = Some(key_contains.to_string());
        self.hold_puts.store(true, Ordering::SeqCst);
    }

    /// How many whole PUTs are parked right now.
    pub fn held_puts(&self) -> usize {
        self.held_puts.lock().unwrap().len()
    }

    /// Let every parked PUT land.
    pub fn release_held_puts(&self) {
        for tx in self.held_puts.lock().unwrap().drain(..) {
            let _ = tx.send(());
        }
    }

    /// The most get_range calls ever in flight at once since the last
    /// reset. A fan-out bounded at N reads exactly N here when there is
    /// work for N — and exactly 1 when it is not a fan-out at all.
    pub fn peak_get_range_in_flight(&self) -> u64 {
        self.peak_get_range.load(Ordering::SeqCst)
    }
    pub fn reset_peak_get_range_in_flight(&self) {
        self.peak_get_range.store(0, Ordering::SeqCst);
    }

    /// The next `n` HEAD calls fail with a transport error.
    ///
    /// The condition this models is not exotic: the foreign-key sweep
    /// issues one HEAD per unknown object, which on a large bucket is
    /// exactly the shape that earns a 503 SlowDown. The SDK's default
    /// retry absorbs a blip; a sustained throttle outlives it.
    pub fn inject_head_failures(&self, n: u64) {
        self.fail_head_count.store(n, Ordering::SeqCst);
    }

    /// The next `n` renewals fail as a transport error — not a 412, so
    /// the holder cannot tell "deposed" from "cut off". Zero clears it.
    pub fn inject_epoch_renew_failures(&self, n: u64) {
        self.fail_epoch_renew_count.store(n, Ordering::SeqCst);
    }

    /// The next `n` whole PUTs STORE THE OBJECT and then report 409.
    ///
    /// This is the one failure a conditional writer cannot interpret.
    /// S3 answers a contended conditional write with 409
    /// `ConditionalRequestConflict`, and a lost response to a PUT that
    /// landed looks the same from the client: the write is durable and
    /// the caller was told it failed. Every other injector here models
    /// a failure that did NOT happen server-side; this one models the
    /// failure that DID.
    /// `key_contains` narrows it to one object — a batch uploads its
    /// packs BEFORE the snapshot CAS, so an untargeted injection would
    /// break a pack upload and never reach the write under test.
    pub fn inject_put_lands_then_fails(&self, key_contains: &str, n: u64) {
        *self.fail_put_after_land_key.lock().unwrap() = key_contains.to_string();
        self.fail_put_after_land_count.store(n, Ordering::SeqCst);
    }

    /// Step-11 drills: the NEXT get_range stalls `ms` before serving —
    /// a slow hydration to race concurrent I/O and priority against.
    pub fn inject_get_range_stall(&self, ms: u64) {
        self.stall_next_get_range_ms.store(ms, Ordering::SeqCst);
    }

    /// Test surface: the bytes under a key change while its etag and its
    /// stored checksum claim do NOT — bit-rot, a broken gateway, a cache
    /// serving the wrong body. Nothing on the wire says so: If-Match
    /// still matches, and a store that never carried a checksum header
    /// validates nothing. Only a reader that checks the body against
    /// what the MANIFEST cites can see it.
    pub fn inject_corrupt_body(&self, key: &str, mutate: impl FnOnce(&mut Vec<u8>)) {
        let mut inner = self.inner.lock().unwrap();
        let chain = inner.chains.get_mut(key).expect("corrupting a key that exists");
        let obj = chain.versions.last_mut().expect("a version to corrupt");
        let mut b = obj.bytes.to_vec();
        mutate(&mut b);
        obj.bytes = Bytes::from(b);
    }

    /// Test surface: plant an object directly (foreign writers, torn
    /// own states, import fixtures). Stamps go in as given.
    pub fn raw_put(&self, key: &str, bytes: Bytes, meta: Vec<(String, String)>) -> ObjectMeta {
        let obj = StoredObject {
            etag: put_etag(&bytes),
            crc64: crc64_nvme(&bytes),
            meta: meta.into_iter().collect(),
            last_modified_unix: now_unix(),
            bytes,
            version_id: String::new(),
            deleted: false,
        };
        let mut inner = self.inner.lock().unwrap();
        let vid = inner.push(key, obj);
        let mut m = inner.current(key).map(|o| o.to_meta()).expect("just pushed");
        m.version_id = Some(vid);
        m
    }

    /// Test surface: REDEEM a URL from [`ObjectStore::presign_put`] the
    /// way S3 does, so a test can play the client — or the attacker —
    /// who holds it.
    ///
    /// Without this the double's presigned URL was a string nothing
    /// ever used, and "the URL grants only the bytes it was issued for"
    /// could not be asked of it at all. What S3 enforces, and so what
    /// this enforces (and nothing more):
    ///
    /// - every header the URL SIGNED must be presented, with the signed
    ///   value, or the signature does not match (403). An unsigned
    ///   header binds nothing: the holder simply leaves it out.
    /// - a presented `x-amz-checksum-sha256` must be the base64 SHA-256
    ///   of the body (400 BadDigest).
    /// - a presented `If-None-Match: *` refuses a key that exists (412).
    ///
    /// Header names are compared case-insensitively, as HTTP does.
    pub fn redeem_presigned_put(
        &self,
        url: &str,
        headers: &BTreeMap<String, String>,
        body: Bytes,
    ) -> StoreResult<ObjectMeta> {
        let rest = url
            .strip_prefix("memory://presigned-put/")
            .ok_or_else(|| StoreError::Other(format!("not a memory presigned PUT: {url}")))?;
        let (key, query) = rest.split_once('?').unwrap_or((rest, ""));
        let presented: BTreeMap<String, &String> =
            headers.iter().map(|(k, v)| (k.to_ascii_lowercase(), v)).collect();
        for param in query.split('&') {
            if let Some((name, want)) = param.split_once('=').and_then(|(k, v)| Some((k.strip_prefix("h.")?, v))) {
                if presented.get(name).map(|v| v.as_str()) != Some(want) {
                    return Err(StoreError::Auth(format!(
                        "SignatureDoesNotMatch: signed header {name} not presented as signed"
                    )));
                }
            }
        }
        if let Some(sum) = presented.get("x-amz-checksum-sha256") {
            let actual = crate::base64_std(&Sha256::digest(&body));
            if **sum != actual {
                return Err(StoreError::ChecksumMismatch(format!(
                    "BadDigest: {key}: x-amz-checksum-sha256 {sum} but the body hashes to {actual}"
                )));
            }
        }
        if presented.get("if-none-match").map(|v| v.as_str()) == Some("*")
            && self.inner.lock().unwrap().current(key).is_some()
        {
            return Err(StoreError::PreconditionFailed(format!("{key} exists")));
        }
        Ok(self.raw_put(key, body, Vec::new()))
    }

    /// Test surface: the NONCURRENT-VERSION LIFECYCLE BACKSTOP (D8),
    /// modelled as an explicit method because it is a timer in reality
    /// and an untestable one if left implicit.
    ///
    /// The hazard it exists to demonstrate is INVERTED from the
    /// eager-prefix design's: gated staging makes the **cited** version
    /// noncurrent the moment a newer generation is staged, so a
    /// `NoncurrentVersionExpiration` rule on `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.
    /// Reaping a cited noncurrent version dangles the manifest; that is
    /// `Inv_CitedVersionLives`'s required-firing mutation and leg B23's
    /// abandoned-mid-stage endgame.
    ///
    /// Returns the (key, version_id) pairs reaped.
    pub fn expire_noncurrent(&self, older_than_secs: u64) -> Vec<(String, String)> {
        let now = now_unix();
        let mut inner = self.inner.lock().unwrap();
        let mut reaped = vec![];
        let keys: Vec<String> = inner.chains.keys().cloned().collect();
        for k in keys {
            let Some(chain) = inner.chains.get_mut(&k) else { continue };
            let last = chain.versions.len().saturating_sub(1);
            let doomed: Vec<String> = chain
                .versions
                .iter()
                .enumerate()
                .filter(|(i, o)| {
                    *i != last && now.saturating_sub(o.last_modified_unix) >= older_than_secs
                })
                .map(|(_, o)| o.version_id.clone())
                .collect();
            for vid in doomed {
                chain.versions.retain(|o| o.version_id != vid);
                reaped.push((k.clone(), vid));
            }
        }
        reaped
    }

    /// Test surface: how many versions a key currently holds (leg B21's
    /// oracle — steady state must return to ONE live version per key,
    /// reclaimed by flint's exact per-citation GC, NOT by waiting for
    /// the backstop).
    pub fn version_count(&self, key: &str) -> usize {
        self.inner.lock().unwrap().chains.get(key).map(|c| c.versions.len()).unwrap_or(0)
    }

    /// Test surface: a DEPOSED writer's late CompleteMultipartUpload.
    /// Assembles whatever parts exist (unconditionally — the deposed
    /// straggler of the A8 drill carries no working guard); answers
    /// `NoSuchUpload` if the assembly was fenced away, exactly S3's
    /// answer after the takeover abort-sweep.
    pub fn raw_complete_upload(&self, upload_id: &str) -> StoreResult<ObjectMeta> {
        let mut inner = self.inner.lock().unwrap();
        let Some(mpu) = inner.uploads.remove(upload_id) else {
            return Err(StoreError::NoSuchUpload(format!("upload {}", upload_id)));
        };
        let mut bytes = Vec::new();
        let n_parts = mpu.parts.len().max(1);
        for part in mpu.parts.values() {
            bytes.extend_from_slice(part);
        }
        let bytes = Bytes::from(bytes);
        let obj = StoredObject {
            etag: mpu_etag(&bytes, n_parts),
            crc64: crc64_nvme(&bytes),
            meta: HashMap::new(),
            last_modified_unix: now_unix(),
            bytes,
                   version_id: String::new(),
            deleted: false,
        };
        let mkey = mpu.key.clone();
        let vid = inner.push(&mkey, obj);
        let mut m = inner.current(&mkey).expect("just pushed").to_meta();
        m.version_id = Some(vid);
        Ok(m)
    }

    /// Test surface: an orphan MPU (for the A9 sweep tests).
    pub fn raw_begin_upload(&self, key: &str) -> String {
        let id = format!("mpu-{}", self.upload_seq.fetch_add(1, Ordering::SeqCst));
        self.inner.lock().unwrap().uploads.insert(
            id.clone(),
            Mpu { key: key.to_string(), parts: BTreeMap::new(), initiated_unix: now_unix() },
        );
        id
    }

    fn check_condition(
        &self,
        existing: Option<&StoredObject>,
        condition: &PutCondition,
    ) -> StoreResult<()> {
        match (condition, existing) {
            (PutCondition::IfMatch(want), Some(o)) if o.etag == *want => Ok(()),
            (PutCondition::IfMatch(_), Some(_)) => {
                Err(StoreError::PreconditionFailed("If-Match: etag differs".into()))
            }
            (PutCondition::IfMatch(_), None) if self.if_match_missing_412.load(Ordering::SeqCst) => {
                Err(StoreError::PreconditionFailed("If-Match: no object".into()))
            }
            (PutCondition::IfMatch(_), None) => Err(StoreError::NotFound("NoSuchKey: If-Match on a missing key".into())),
            (PutCondition::IfNoneMatchAny, None) => Ok(()),
            (PutCondition::IfNoneMatchAny, Some(_)) => {
                Err(StoreError::PreconditionFailed("If-None-Match: object exists".into()))
            }
            (PutCondition::Unconditional, _) => Ok(()),
        }
    }
}

fn put_etag(bytes: &[u8]) -> String {
    let mut h = Sha256::new();
    h.update(bytes);
    format!("\"{:x}\"", h.finalize())
}

/// Multipart etag: content hash + part count — deliberately a DIFFERENT
/// string than a whole-put of the same bytes, mirroring S3's "the
/// multipart ETag is not a content hash".
fn mpu_etag(bytes: &[u8], parts: usize) -> String {
    let mut h = Sha256::new();
    h.update(bytes);
    format!("\"{:x}-{}\"", h.finalize(), parts)
}

#[derive(serde::Serialize, serde::Deserialize)]
struct EpochBody {
    holder_id: String,
    epoch: u64,
    /// Clean-shutdown mark — see `ObjectStore::epoch_release`.
    #[serde(default)]
    released: bool,
    /// The holder's observed-state echo (`LeaseEcho`), opaque here.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    echo: Option<String>,
    /// The claim queue and the handoff reservation (see the S3 twin).
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    waiters: Vec<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    handoff: Option<String>,
}

impl MemoryStore {
    /// One epoch transition: CAS on `condition`, then store `body`. The
    /// etag is SALTED with a sequence so two writes of the same body
    /// (a renew, a re-enqueue) still rotate the CAS token — a stale
    /// holder's heartbeat must fail even when nothing in it changed.
    fn epoch_store(&self, key: &str, body: EpochBody, condition: &PutCondition) -> StoreResult<String> {
        let bytes = Bytes::from(serde_json::to_vec(&body).unwrap());
        let mut inner = self.inner.lock().unwrap();
        self.check_condition(inner.current(key), condition)?;
        let mut salted = bytes.to_vec();
        salted.extend_from_slice(&self.upload_seq.fetch_add(1, Ordering::SeqCst).to_be_bytes());
        let obj = StoredObject {
            etag: put_etag(&salted),
            crc64: crc64_nvme(&bytes),
            meta: HashMap::new(),
            last_modified_unix: now_unix(),
            bytes,
            version_id: String::new(),
            deleted: false,
        };
        let token = obj.etag.clone();
        inner.push(key, obj);
        Ok(token)
    }

    fn bump(&self, op: &'static str) {
        *self.ops.lock().unwrap().entry(op).or_insert(0) += 1;
    }
    /// Call counts since construction or the last reset.
    pub fn op_counts(&self) -> std::collections::BTreeMap<&'static str, u64> {
        self.ops.lock().unwrap().clone()
    }
    pub fn reset_op_counts(&self) {
        self.ops.lock().unwrap().clear();
    }
    pub fn total_ops(&self) -> u64 {
        self.ops.lock().unwrap().values().sum()
    }
}

#[async_trait]
impl ObjectStore for MemoryStore {
    fn request_counts(&self) -> Option<crate::RequestCounts> {
        // The double counts calls by method; fold them into the kinds a
        // real store would have sent (the cell is a GET or a PUT there).
        let mut c = crate::RequestCounts::default();
        for (op, n) in self.op_counts() {
            match op {
                "get_whole" | "get_range" | "get_version" | "epoch_read" | "presign_get" => c.get += n,
                "head" | "head_version" => c.head += n,
                "put_whole" | "epoch_acquire" | "epoch_renew" | "epoch_release" | "epoch_handoff"
                | "epoch_enqueue" | "presign_put" => c.put += n,
                "copy_object" => c.copy += n,
                "delete" | "delete_version" => c.delete += n,
                "list" | "list_versions" | "list_uploads" => c.list += n,
                "compose_generation" | "compose_inner" | "abort_upload" => c.multipart += n,
                _ => {}
            }
        }
        Some(c)
    }

    async fn put_whole(
        &self,
        key: &str,
        body: Bytes,
        condition: &PutCondition,
        stamps: &GenerationStamps,
        crc64: u64,
    ) -> StoreResult<ObjectMeta> {
        self.bump("put_whole");
        self.pace().await;
        let held = self.hold_puts.load(Ordering::SeqCst)
            && match self.hold_puts_matching.lock().unwrap().as_deref() {
                None => true,
                Some(m) => key.contains(m),
            };
        if held {
            let (tx, rx) = tokio::sync::oneshot::channel();
            self.held_puts.lock().unwrap().push(tx);
            // A dropped sender (a release that drained the list) lets
            // the PUT through as well.
            let _ = rx.await;
        }
        let delay = self.put_whole_delay_ms.load(Ordering::SeqCst);
        if delay > 0 {
            tokio::time::sleep(std::time::Duration::from_millis(delay)).await;
        }
        let actual = crc64_nvme(&body);
        if actual != crc64 {
            return Err(StoreError::ChecksumMismatch(format!(
                "put {}: claimed {:#x}, content is {:#x}",
                key, crc64, actual
            )));
        }
        let mut inner = self.inner.lock().unwrap();
        self.check_condition(inner.current(key), condition)?;
        let obj = StoredObject {
            etag: put_etag(&body),
            crc64,
            meta: stamps.to_meta().into_iter().collect(),
            last_modified_unix: now_unix(),
            bytes: body,
            version_id: String::new(),
            deleted: false,
        };
        // The version id is minted by `push`; reporting a meta built
        // BEFORE it would hand the caller `Some("")` — and a manifest
        // that cites version "" makes the citation GC match nothing and
        // reap every real version of the key.
        let vid = inner.push(key, obj);
        let mut m = inner.current(key).expect("just pushed").to_meta();
        m.version_id = Some(vid);
        if self.strip_version_ids.load(Ordering::SeqCst) {
            // The version still EXISTS — the proxy only hid its name.
            // That asymmetry is the hazard: everything keeps working
            // until a citation needs to name a version.
            m.version_id = None;
        }
        // The object is already stored. Reporting the failure HERE is
        // the whole point: the caller cannot tell this from a write
        // that never happened.
        if self.fail_put_after_land_count.load(Ordering::SeqCst) > 0 {
            let pat = self.fail_put_after_land_key.lock().unwrap().clone();
            if (pat.is_empty() || key.contains(pat.as_str()))
                && self
                    .fail_put_after_land_count
                    .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |v| v.checked_sub(1))
                    .is_ok()
            {
                return Err(StoreError::Conflict(format!(
                    "injected: the PUT of {key} landed and the response did not"
                )));
            }
        }
        Ok(m)
    }

    async fn copy_object(
        &self,
        src_key: &str,
        src_if_match: Option<&str>,
        dst_key: &str,
        condition: &PutCondition,
        stamps: &GenerationStamps,
    ) -> StoreResult<ObjectMeta> {
        self.bump("copy_object");
        self.pace().await;
        let mut inner = self.inner.lock().unwrap();
        let src = inner
            .current(src_key)
            .ok_or_else(|| StoreError::NotFound(src_key.to_string()))?;
        // The SOURCE guard. A 412 here is not the destination's: the
        // object the caller read has been overwritten, and copying the
        // replacement would move bytes nobody asked for.
        if let Some(want) = src_if_match {
            if src.etag != want {
                return Err(StoreError::PreconditionFailed(format!(
                    "copy source {src_key} is at {} , not {want}",
                    src.etag
                )));
            }
        }
        let bytes = src.bytes.clone();
        // The CRC travels with the BYTES — they are the same bytes. The
        // stamps do NOT: a copy is a new publish under a new key, and
        // wearing the source's generation would file one object's
        // history under another's name. A double that copied the source
        // meta wholesale would pass a test the real store fails.
        let crc64 = src.crc64;
        self.check_condition(inner.current(dst_key), condition)?;
        let obj = StoredObject {
            etag: put_etag(&bytes),
            crc64,
            meta: stamps.to_meta().into_iter().collect(),
            last_modified_unix: now_unix(),
            bytes,
            version_id: String::new(),
            deleted: false,
        };
        let vid = inner.push(dst_key, obj);
        let mut m = inner.current(dst_key).expect("just pushed").to_meta();
        m.version_id = Some(vid);
        if self.strip_version_ids.load(Ordering::SeqCst) {
            m.version_id = None;
        }
        Ok(m)
    }

    async fn compose_generation(&self, spec: &ComposeSpec<'_>) -> StoreResult<ObjectMeta> {
        self.bump("compose_generation");
        self.pace().await;
        if spec.parts.is_empty() {
            return Err(StoreError::Other("compose: no parts".into()));
        }
        if spec.parts.len() > self.max_parts {
            return Err(StoreError::Other(format!(
                "compose: {} parts exceeds the backend maximum {}",
                spec.parts.len(),
                self.max_parts
            )));
        }
        let upload_id = self.raw_begin_upload(spec.key);
        // Everything below aborts the MPU on error (A9): no failure
        // path may leave the assembly pending — except the injected
        // crash, whose whole point is the orphan.
        let result = self.compose_inner(spec, &upload_id).await;
        if result.is_err() && !self.leave_orphan.swap(false, Ordering::SeqCst) {
            let mut inner = self.inner.lock().unwrap();
            inner.uploads.remove(&upload_id);
        }
        result
    }

    async fn head(&self, key: &str) -> StoreResult<ObjectMeta> {
        self.bump("head");
        self.pace().await;
        if self
            .fail_head_count
            .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |c| c.checked_sub(1))
            .is_ok()
        {
            return Err(StoreError::Other("injected head failure".into()));
        }
        self.inner
            .lock()
            .unwrap()
            .current(key)
            .map(|o| o.to_meta())
            .ok_or_else(|| StoreError::NotFound(key.into()))
    }

    /// A stand-in URL: deterministic, unmistakably not real, and
    /// refused for a key that does not exist. A test can assert that
    /// the right key was signed and that a missing object is not
    /// advertised; whether a git client can actually fetch it is a
    /// question only a real store answers, and forge's storm leg is
    /// where that is asked.
    async fn presign_get(&self, key: &str, ttl_secs: u64) -> StoreResult<String> {
        self.bump("presign_get");
        let inner = self.inner.lock().unwrap();
        if inner.current(key).is_none() {
            return Err(StoreError::NotFound(key.into()));
        }
        Ok(format!("memory://presigned/{key}?ttl={ttl_secs}"))
    }

    /// A stand-in URL for a write. Unlike the read side it does NOT
    /// require the object to exist — that is the whole point of an
    /// upload. The headers it binds are written INTO the URL as `h.`
    /// parameters, which is how `redeem_presigned_put` plays S3's
    /// signature: a signed header must be presented as signed.
    async fn presign_put(
        &self,
        key: &str,
        ttl_secs: u64,
        sha256: &[u8; 32],
    ) -> StoreResult<PresignedPut> {
        self.bump("presign_put");
        let headers: BTreeMap<String, String> = [
            ("x-amz-checksum-sha256".to_string(), crate::base64_std(sha256)),
            ("if-none-match".to_string(), "*".to_string()),
        ]
        .into_iter()
        .collect();
        let mut url = format!("memory://presigned-put/{key}?ttl={ttl_secs}");
        for (k, v) in &headers {
            url.push_str(&format!("&h.{k}={v}"));
        }
        Ok(PresignedPut { url, headers })
    }

    async fn get_whole(
        &self,
        key: &str,
        if_match: Option<&str>,
    ) -> StoreResult<(ObjectMeta, Bytes)> {
        self.bump("get_whole");
        self.pace().await;
        let inner = self.inner.lock().unwrap();
        let o = inner
            .current(key)
            .ok_or_else(|| StoreError::NotFound(key.into()))?;
        if let Some(want) = if_match {
            if o.etag != want {
                return Err(StoreError::PreconditionFailed(format!(
                    "get {}: etag {} != {}",
                    key, o.etag, want
                )));
            }
        }
        Ok((o.to_meta(), o.bytes.clone()))
    }

    /// Real S3 always delivers a ranged body as MANY frames. This
    /// double handed back exactly one, so a caller that walks segments
    /// and advances a running offset was never actually exercised
    /// against it — a double missing a property its subject always has,
    /// which passes for the wrong reason. Built on `get_range` so every
    /// injected failure, stall and in-flight count still applies.
    async fn get_range_segments(
        &self,
        key: &str,
        offset: u64,
        len: u64,
        if_match: &str,
    ) -> StoreResult<Vec<Bytes>> {
        let whole = self.get_range(key, offset, len, if_match).await?;
        // Deliberately UNEVEN and never aligned to the caller's chunk
        // size: equal segments let an off-by-one in the running offset
        // cancel itself out and land the bytes correctly anyway.
        let mut segs: Vec<Bytes> = Vec::new();
        let (mut at, mut step) = (0usize, 1usize);
        while at < whole.len() {
            let n = step.min(whole.len() - at);
            segs.push(whole.slice(at..at + n)); // zero-copy
            at += n;
            step = (step.saturating_mul(3)).min(64 * 1024);
        }
        if segs.is_empty() {
            segs.push(whole);
        }
        Ok(segs)
    }

    async fn get_range(
        &self,
        key: &str,
        offset: u64,
        len: u64,
        if_match: &str,
    ) -> StoreResult<Bytes> {
        self.bump("get_range");
        self.pace().await;
        let _in_flight = {
            let now = self.inflight_get_range.fetch_add(1, Ordering::SeqCst) + 1;
            self.peak_get_range.fetch_max(now, Ordering::SeqCst);
            InFlight(&self.inflight_get_range)
        };
        // Counted injections for the step-11 drills.
        if self
            .fail_get_range_count
            .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |c| c.checked_sub(1))
            .is_ok()
        {
            return Err(StoreError::Other("injected get_range failure".into()));
        }
        let stall = self.stall_next_get_range_ms.swap(0, Ordering::SeqCst);
        if stall > 0 {
            tokio::time::sleep(std::time::Duration::from_millis(stall)).await;
        }
        let delay = self.get_range_delay_ms.load(Ordering::SeqCst);
        if delay > 0 {
            tokio::time::sleep(std::time::Duration::from_millis(delay)).await;
        }
        let inner = self.inner.lock().unwrap();
        let o = inner
            .current(key)
            .ok_or_else(|| StoreError::NotFound(key.into()))?;
        if o.etag != if_match {
            return Err(StoreError::PreconditionFailed(format!(
                "get_range {}: etag {} != {}",
                key, o.etag, if_match
            )));
        }
        let start = offset.min(o.bytes.len() as u64) as usize;
        let end = (offset + len).min(o.bytes.len() as u64) as usize;
        Ok(o.bytes.slice(start..end))
    }

    async fn list(&self, prefix: &str) -> StoreResult<Vec<ListedObject>> {
        self.bump("list");
        self.pace().await;
        let denied = self.deny_list_matching.lock().unwrap().as_deref().is_some_and(|m| prefix.contains(m));
        if denied {
            return Err(StoreError::Auth(format!("injected: no listing of {prefix}")));
        }
        Ok(self
            .inner
            .lock()
            .unwrap()
            .chains
            .range(prefix.to_string()..)
            .take_while(|(k, _)| k.starts_with(prefix))
            // Current versions only, delete markers skipped — a LIST
            // must never report a key a GET would 404 on.
            .filter_map(|(k, c)| {
                c.current().map(|o| ListedObject {
                    key: k.clone(),
                    size: o.bytes.len() as u64,
                    etag: o.etag.clone(),
                    last_modified_unix: Some(o.last_modified_unix),
                })
            })
            .collect())
    }

    async fn delete(&self, key: &str) -> StoreResult<()> {
        self.bump("delete");
        self.pace().await;
        self.inner.lock().unwrap().push_delete_marker(key);
        Ok(())
    }

    async fn delete_if_match(&self, key: &str, etag: &str) -> StoreResult<()> {
        // Counted as the one DELETE request it is on the wire.
        self.bump("delete");
        self.pace().await;
        let mut inner = self.inner.lock().unwrap();
        let refused = match inner.current(key) {
            None => Some(StoreError::NotFound(key.to_string())),
            Some(o) if o.etag != etag => Some(StoreError::PreconditionFailed(format!(
                "{key}: If-Match {etag} but the current etag is {}",
                o.etag
            ))),
            Some(_) => None,
        };
        if let Some(e) = refused {
            return Err(e);
        }
        inner.push_delete_marker(key);
        Ok(())
    }

    /// One request per thousand keys, as S3's `DeleteObjects` costs — so
    /// a request census over the double reads what the wire would.
    async fn delete_many(&self, keys: &[String]) -> StoreResult<DeleteManyReport> {
        let mut report = DeleteManyReport::default();
        for chunk in keys.chunks(1000) {
            self.bump("delete");
            self.pace().await;
            let mut inner = self.inner.lock().unwrap();
            for key in chunk {
                inner.push_delete_marker(key);
                report.deleted += 1;
            }
        }
        Ok(report)
    }

    async fn head_version(&self, key: &str, version_id: &str) -> StoreResult<ObjectMeta> {
        self.bump("head_version");
        self.pace().await;
        self.inner
            .lock()
            .unwrap()
            .version(key, version_id)
            .map(|o| o.to_meta())
            .ok_or_else(|| StoreError::NotFound(format!("{key}?versionId={version_id}")))
    }

    async fn get_version(&self, key: &str, version_id: &str) -> StoreResult<(ObjectMeta, Bytes)> {
        self.bump("get_version");
        self.pace().await;
        let inner = self.inner.lock().unwrap();
        let o = inner
            .version(key, version_id)
            .ok_or_else(|| StoreError::NotFound(format!("{key}?versionId={version_id}")))?;
        Ok((o.to_meta(), o.bytes.clone()))
    }

    async fn delete_version(&self, key: &str, version_id: &str) -> StoreResult<()> {
        self.bump("delete_version");
        self.pace().await;
        // Idempotent, exactly as S3: deleting an absent version is Ok —
        // the GC pass and the operator sweep race each other by design.
        self.inner.lock().unwrap().remove_version(key, version_id);
        Ok(())
    }

    async fn list_versions(&self, prefix: &str) -> StoreResult<Vec<ListedVersion>> {
        self.bump("list_versions");
        self.pace().await;
        let inner = self.inner.lock().unwrap();
        let mut out = vec![];
        for (k, chain) in inner.chains.range(prefix.to_string()..) {
            if !k.starts_with(prefix) {
                break;
            }
            let last = chain.versions.len().saturating_sub(1);
            for (i, o) in chain.versions.iter().enumerate() {
                out.push(ListedVersion {
                    key: k.clone(),
                    version_id: o.version_id.clone(),
                    etag: o.etag.clone(),
                    size: o.bytes.len() as u64,
                    is_current: i == last,
                    is_delete_marker: o.deleted,
                    last_modified_unix: Some(o.last_modified_unix),
                });
            }
        }
        Ok(out)
    }

    async fn list_uploads(&self, prefix: &str) -> StoreResult<Vec<PendingUpload>> {
        self.bump("list_uploads");
        self.pace().await;
        let inner = self.inner.lock().unwrap();
        let mut v: Vec<PendingUpload> = inner
            .uploads
            .iter()
            .filter(|(_, m)| m.key.starts_with(prefix))
            .map(|(id, m)| PendingUpload {
                key: m.key.clone(),
                upload_id: id.clone(),
                initiated_unix: Some(m.initiated_unix),
            })
            .collect();
        v.sort_by(|a, b| a.upload_id.cmp(&b.upload_id));
        Ok(v)
    }

    async fn abort_upload(&self, _key: &str, upload_id: &str) -> StoreResult<()> {
        self.bump("abort_upload");
        self.pace().await;
        // Absent is Ok: abort races the lifecycle rule by design.
        self.inner.lock().unwrap().uploads.remove(upload_id);
        Ok(())
    }

    async fn bootstrap(&self, _prefix: &str) -> StoreResult<BootstrapReport> {
        self.bump("bootstrap");
        self.pace().await;
        Ok(BootstrapReport {
            notes: vec!["memory store: no bucket posture to verify".into()],
            ..Default::default()
        })
    }

    async fn lifecycle_rules(&self) -> StoreResult<Vec<LifecycleView>> {
        self.bump("lifecycle_rules");
        self.pace().await;
        Ok(self.inner.lock().unwrap().lifecycle.clone())
    }

    async fn ensure_noncurrent_retention(
        &self,
        prefix: &str,
        days: u64,
    ) -> StoreResult<RetentionOutcome> {
        self.bump("ensure_noncurrent_retention");
        self.pace().await;
        if self.fail_lifecycle_writes.load(Ordering::SeqCst) {
            return Err(StoreError::Other("lifecycle writes are denied".into()));
        }
        let rule_id = format!("flint-lean-noncurrent-{}", prefix.trim_end_matches('/').replace('/', "-"));
        let mut inner = self.inner.lock().unwrap();
        // Read-merge-append, exactly as S3 must: an existing conforming
        // rule is left alone and NOTHING is rewritten.
        if let Some(r) = inner
            .lifecycle
            .iter()
            .find(|r| r.enabled && r.prefix == prefix && r.noncurrent_days == Some(days))
        {
            return Ok(RetentionOutcome {
                rule_id: r.id.clone(),
                noncurrent_days: days,
                created: false,
            });
        }
        inner.lifecycle.retain(|r| r.id != rule_id);
        inner.lifecycle.push(LifecycleView {
            id: rule_id.clone(),
            enabled: true,
            prefix: prefix.to_string(),
            noncurrent_days: Some(days),
            expired_delete_marker: true,
        });
        Ok(RetentionOutcome { rule_id, noncurrent_days: days, created: true })
    }

    async fn epoch_read(&self, key: &str) -> StoreResult<Option<EpochState>> {
        self.bump("epoch_read");
        self.pace().await;
        let inner = self.inner.lock().unwrap();
        let Some(o) = inner.current(key) else {
            return Ok(None);
        };
        let body: EpochBody = serde_json::from_slice(&o.bytes)
            .map_err(|e| StoreError::Other(format!("epoch body: {}", e)))?;
        Ok(Some(EpochState {
            holder_id: body.holder_id,
            epoch: body.epoch,
            token: o.etag.clone(),
            last_renew_unix: Some(o.last_modified_unix),
            released: body.released,
            echo: body.echo,
            waiters: body.waiters,
            handoff: body.handoff,
        }))
    }

    async fn epoch_acquire(
        &self,
        key: &str,
        holder_id: &str,
        supersede: Option<&EpochState>,
    ) -> StoreResult<EpochLease> {
        self.bump("epoch_acquire");
        self.pace().await;
        let epoch = supersede.map_or(1, |s| s.epoch + 1);
        let waiters: Vec<String> = supersede
            .map(|s| s.waiters.iter().filter(|w| w.as_str() != holder_id).cloned().collect())
            .unwrap_or_default();
        let condition = match supersede {
            None => PutCondition::IfNoneMatchAny,
            Some(s) => PutCondition::IfMatch(s.token.clone()),
        };
        let token = self.epoch_store(
            key,
            EpochBody {
                holder_id: holder_id.into(),
                epoch,
                released: false,
                echo: None,
                waiters: waiters.clone(),
                handoff: None,
            },
            &condition,
        )?;
        Ok(EpochLease { holder_id: holder_id.into(), epoch, token, waiters })
    }

    async fn epoch_renew(
        &self,
        key: &str,
        lease: &EpochLease,
        echo: Option<&str>,
    ) -> StoreResult<EpochLease> {
        self.bump("epoch_renew");
        self.pace().await;
        if self
            .fail_epoch_renew_count
            .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |c| c.checked_sub(1))
            .is_ok()
        {
            return Err(StoreError::Other("injected renew failure: store unreachable".into()));
        }
        let token = self.epoch_store(
            key,
            EpochBody {
                holder_id: lease.holder_id.clone(),
                epoch: lease.epoch,
                released: false,
                echo: echo.map(|e| e.to_string()),
                waiters: lease.waiters.clone(),
                handoff: None,
            },
            &PutCondition::IfMatch(lease.token.clone()),
        )?;
        Ok(EpochLease {
            holder_id: lease.holder_id.clone(),
            epoch: lease.epoch,
            token,
            waiters: lease.waiters.clone(),
        })
    }

    async fn epoch_release(&self, key: &str, lease: &EpochLease) -> StoreResult<()> {
        self.bump("epoch_release");
        self.pace().await;
        // Mark, never delete — deleting restarts epoch numbering at 1.
        // A released cell reports no live syncer: clearing the echo is
        // the point, not an omission (the tier hub's rule; lean's
        // per-barrier `epoch_handoff` keeps it).
        self.epoch_store(
            key,
            EpochBody {
                holder_id: lease.holder_id.clone(),
                epoch: lease.epoch,
                released: true,
                echo: None,
                waiters: lease.waiters.clone(),
                handoff: None,
            },
            &PutCondition::IfMatch(lease.token.clone()),
        )?;
        Ok(())
    }

    async fn epoch_handoff(
        &self,
        key: &str,
        lease: &EpochLease,
        echo: Option<&str>,
    ) -> StoreResult<()> {
        self.bump("epoch_handoff");
        self.pace().await;
        let mut rest = lease.waiters.clone();
        let head = if rest.is_empty() { None } else { Some(rest.remove(0)) };
        self.epoch_store(
            key,
            EpochBody {
                holder_id: lease.holder_id.clone(),
                epoch: lease.epoch,
                released: true,
                echo: echo.map(|e| e.to_string()),
                waiters: rest,
                handoff: head,
            },
            &PutCondition::IfMatch(lease.token.clone()),
        )?;
        Ok(())
    }

    async fn epoch_enqueue(
        &self,
        key: &str,
        observed: &EpochState,
        holder_id: &str,
    ) -> StoreResult<EpochState> {
        self.bump("epoch_enqueue");
        self.pace().await;
        let mut waiters = observed.waiters.clone();
        if !waiters.iter().any(|w| w == holder_id) {
            waiters.push(holder_id.to_string());
        }
        let token = self.epoch_store(
            key,
            EpochBody {
                holder_id: observed.holder_id.clone(),
                epoch: observed.epoch,
                released: observed.released,
                echo: observed.echo.clone(),
                waiters: waiters.clone(),
                handoff: observed.handoff.clone(),
            },
            &PutCondition::IfMatch(observed.token.clone()),
        )?;
        Ok(EpochState {
            holder_id: observed.holder_id.clone(),
            epoch: observed.epoch,
            token,
            last_renew_unix: Some(now_unix()),
            echo: observed.echo.clone(),
            released: observed.released,
            waiters,
            handoff: observed.handoff.clone(),
        })
    }

    fn min_part_size(&self) -> u64 {
        self.min_part
    }

    fn max_parts(&self) -> usize {
        self.max_parts
    }

    fn upload_gate(&self) -> Option<std::sync::Arc<crate::gate::ByteGate>> {
        self.upload_gate.clone()
    }
}

impl MemoryStore {
    /// One part of a compose, staged into the upload's part map: a
    /// local part read through the upload byte gate, a copied part
    /// taken under its guard. Independent of every other part, which
    /// is what lets `compose_inner` run these `part_parallelism` wide —
    /// the S3 backend's `compose_one_part`, in the double.
    async fn stage_one_part(
        &self,
        spec: &ComposeSpec<'_>,
        upload_id: &str,
        i: usize,
        p: &PartSource,
    ) -> StoreResult<()> {
        match p {
            PartSource::Local { offset, len } => {
                // Charged from BEFORE the read — the read is the
                // allocation — until the part has landed (the permit
                // lives to the end of this arm), exactly as the S3
                // backend holds it until `UploadPart` returns. The read
                // runs on the blocking pool as the backend's does, so
                // the parts of one object genuinely overlap and a test
                // reading the gate's high-water mark sees the window,
                // not the scheduler.
                let _permit = match &self.upload_gate {
                    Some(g) => Some(g.acquire(*len).await),
                    None => None,
                };
                let file = spec.local.clone().ok_or_else(|| {
                    StoreError::Other(format!(
                        "compose of {}: a Local part with no open descriptor",
                        spec.key
                    ))
                })?;
                let (offset, len) = (*offset, *len);
                let bytes = tokio::task::spawn_blocking(move || read_local(&file, offset, len))
                    .await
                    .map_err(|e| StoreError::Other(format!("local read join: {e}")))??;
                // The `UploadPart` round trip, under the permit: with a
                // request delay injected, the parts of one object are in
                // flight together for it, however fast the reads finish
                // on a busy machine (`tests_upload_gate`).
                self.pace().await;
                self.land_part(spec, upload_id, i, len, bytes)
            }
            PartSource::BaseCopy { offset, len } => {
                let want = spec.base_etag.as_deref().ok_or_else(|| {
                    StoreError::Other("compose: BaseCopy without base_etag".into())
                })?;
                let bytes = {
                    let inner = self.inner.lock().unwrap();
                    let base_key = spec.base_key.unwrap_or(spec.key);
                    let base = inner.current(base_key).ok_or_else(|| {
                        StoreError::PreconditionFailed("copy-source: no base object".into())
                    })?;
                    if base.etag != want {
                        return Err(StoreError::PreconditionFailed(
                            "copy-source-if-match: base etag differs".into(),
                        ));
                    }
                    let s = *offset as usize;
                    let e = (*offset + *len) as usize;
                    if e > base.bytes.len() {
                        return Err(StoreError::Other(format!(
                            "copy-source range [{}, {}) beyond base size {}",
                            s,
                            e,
                            base.bytes.len()
                        )));
                    }
                    base.bytes.slice(s..e)
                };
                self.land_part(spec, upload_id, i, *len, bytes)
            }
        }
    }

    /// A staged part lands in the upload's part map — the double's
    /// "UploadPart returned".
    fn land_part(
        &self,
        spec: &ComposeSpec<'_>,
        upload_id: &str,
        i: usize,
        want_len: u64,
        bytes: Bytes,
    ) -> StoreResult<()> {
        if bytes.len() as u64 != want_len {
            return Err(StoreError::Other(format!("compose: part {} short read", i)));
        }
        spec.note_progress(bytes.len() as u64);
        self.inner
            .lock()
            .unwrap()
            .uploads
            .get_mut(upload_id)
            .ok_or_else(|| StoreError::NoSuchUpload(upload_id.into()))?
            .parts
            .insert(i, bytes);
        Ok(())
    }

    async fn compose_inner(
        &self,
        spec: &ComposeSpec<'_>,
        upload_id: &str,
    ) -> StoreResult<ObjectMeta> {
        self.bump("compose_inner");
        // The real store cannot hash a copied part it never reads; the
        // double refuses the same spec so a caller learns it here.
        if spec.crc64.is_none()
            && spec.parts.iter().any(|p| matches!(p, PartSource::BaseCopy { .. }))
        {
            return Err(StoreError::Other(
                "compose: the checksum cannot be accumulated over a copied part; supply crc64".into(),
            ));
        }
        // Parts must be contiguous from 0 and respect granularity —
        // catching a flusher part-grid bug here beats catching it on
        // real S3.
        let mut expect = 0u64;
        for (i, p) in spec.parts.iter().enumerate() {
            let (off, len) = match p {
                PartSource::Local { offset, len } | PartSource::BaseCopy { offset, len } => {
                    (*offset, *len)
                }
            };
            if off != expect {
                return Err(StoreError::Other(format!(
                    "compose: part {} at offset {} but object is contiguous to {}",
                    i, off, expect
                )));
            }
            if len < self.min_part && i + 1 != spec.parts.len() {
                return Err(StoreError::Other(format!(
                    "compose: part {} len {} under the backend minimum {}",
                    i, len, self.min_part
                )));
            }
            expect = off + len;
        }

        // Stage: local reads (through the upload byte gate) + guarded
        // base copies (the copy guard is evaluated per part against the
        // CURRENT object, exactly like x-amz-copy-source-if-match),
        // `part_parallelism` at a time — the S3 backend's window shape.
        // The futures are built by a loop for the reason `s3.rs` gives:
        // a closure here is not general enough over the borrow.
        {
            use futures::stream::StreamExt;
            let par = self.part_parallelism.max(1);
            let mut futs = Vec::with_capacity(spec.parts.len());
            for (i, p) in spec.parts.iter().enumerate() {
                futs.push(self.stage_one_part(spec, upload_id, i, p));
            }
            let staged: Vec<StoreResult<()>> =
                futures::stream::iter(futs).buffer_unordered(par).collect().await;
            staged.into_iter().collect::<StoreResult<Vec<()>>>()?;
        }
        // Assemble in PART order — the parts landed in completion order.
        let mut assembled = Vec::with_capacity(expect as usize);
        {
            let inner = self.inner.lock().unwrap();
            let up = inner
                .uploads
                .get(upload_id)
                .ok_or_else(|| StoreError::NoSuchUpload(upload_id.into()))?;
            for i in 0..spec.parts.len() {
                let bytes = up.parts.get(&i).ok_or_else(|| {
                    StoreError::Other(format!("compose: part {} was staged but did not land", i))
                })?;
                assembled.extend_from_slice(bytes);
            }
        }

        // S3 validates the full-object CRC64-NVME at
        // CompleteMultipartUpload, so a composed upload whose checksum
        // does not match its bytes fails there and not on the next
        // read. Without this the double was WEAKER than the backend on
        // the one path that carries the largest objects: every
        // multipart upload in the tree — forge's packs above the
        // whole-put ceiling, lean's chunks — had its checksum accepted
        // untested, and a wrong CRC would first be seen by a real
        // bucket.
        let actual = crc64_nvme(&assembled);
        if let Some(claimed) = spec.crc64 {
            if actual != claimed {
                return Err(StoreError::ChecksumMismatch(format!(
                    "compose {}: claimed {:#x}, content is {:#x}",
                    spec.key, claimed, actual
                )));
            }
        }

        match self.inject.swap(INJECT_NONE, Ordering::SeqCst) {
            INJECT_CRASH_BEFORE_COMPLETE => {
                // Died before Complete: the MPU stays pending — the
                // one error path that deliberately leaves the orphan
                // (the A9 sweep exists for it).
                self.leave_orphan.store(true, Ordering::SeqCst);
                return Err(StoreError::Other("injected: crash before Complete".into()));
            }
            INJECT_TORN_COMPLETE => {
                // Complete lands server-side; the response is lost.
                self.finish_complete(spec, upload_id, assembled)?;
                return Err(StoreError::Other("injected: Complete response lost".into()));
            }
            INJECT_SWEPT => {
                return Err(StoreError::NoSuchUpload(format!("upload {upload_id} (injected: swept)")));
            }
            _ => {}
        }

        let meta = self.finish_complete(spec, upload_id, assembled)?;
        Ok(meta)
    }

    fn finish_complete(
        &self,
        spec: &ComposeSpec<'_>,
        upload_id: &str,
        assembled: Vec<u8>,
    ) -> StoreResult<ObjectMeta> {
        let actual = crc64_nvme(&assembled);
        if let Some(claimed) = spec.crc64 {
            if actual != claimed {
                return Err(StoreError::ChecksumMismatch(format!(
                    "compose {}: claimed {:#x}, assembled is {:#x}",
                    spec.key, claimed, actual
                )));
            }
        }
        let mut inner = self.inner.lock().unwrap();
        if inner.uploads.remove(upload_id).is_none() {
            // Fenced by a takeover sweep or lifecycle abort.
            return Err(StoreError::NoSuchUpload(upload_id.into()));
        }
        self.check_condition(inner.current(spec.key), &spec.condition)?;
        let parts = spec.parts.len();
        let bytes = Bytes::from(assembled);
        let obj = StoredObject {
            etag: mpu_etag(&bytes, parts),
            crc64: actual,
            meta: spec.stamps.to_meta().into_iter().collect(),
            last_modified_unix: now_unix(),
            bytes,
                   version_id: String::new(),
            deleted: false,
        };
        let vid = inner.push(spec.key, obj);
        let mut m = inner.current(spec.key).expect("just pushed").to_meta();
        m.version_id = Some(vid);
        Ok(m)
    }
}

/// One part, from the descriptor the caller opened — the S3 backend's
/// rule in the double, so a test that plants a swap in the window sees
/// what the real backend does (review 2026-09-18, H5).
fn read_local(file: &std::sync::Arc<std::fs::File>, offset: u64, len: u64) -> StoreResult<Bytes> {
    use std::os::unix::fs::FileExt;
    let mut buf = vec![0u8; len as usize];
    file.read_exact_at(&mut buf, offset)
        .map_err(|e| StoreError::Other(format!("local read: {}", e)))?;
    Ok(Bytes::from(buf))
}

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

    fn stamps(generation: u64) -> GenerationStamps {
        GenerationStamps { generation, epoch: 1, flush_uuid: format!("u-{}", generation), boundary_source: None, posix: None }
    }

    /// The handles collector's delete: every named key goes, an absent
    /// key is not an error, nothing else is touched, and the request
    /// census reads one DELETE per thousand keys — what `DeleteObjects`
    /// costs on the wire, and what a request-count test over this double
    /// must therefore see.
    #[tokio::test]
    async fn delete_many_takes_every_key_tolerates_absent_ones_and_costs_one_request_per_batch() {
        let s = MemoryStore::new();
        let body = Bytes::from_static(b"bytes");
        let crc = crc64_nvme(&body);
        for k in ["h/a@1", "h/a@2", "h/b@1"] {
            s.put_whole(k, body.clone(), &PutCondition::IfNoneMatchAny, &stamps(1), crc).await.unwrap();
        }
        s.reset_op_counts();
        let keys: Vec<String> = ["h/a@1", "h/b@1", "h/never-existed@9"].iter().map(|k| k.to_string()).collect();
        let report = s.delete_many(&keys).await.unwrap();
        assert_eq!(report, DeleteManyReport { deleted: 3, failed: vec![] });
        assert!(matches!(s.head("h/a@1").await, Err(StoreError::NotFound(_))));
        assert!(matches!(s.head("h/b@1").await, Err(StoreError::NotFound(_))));
        assert!(s.head("h/a@2").await.is_ok(), "a key the batch did not name must stay");
        assert_eq!(s.op_counts().get("delete").copied(), Some(1), "one request for a batch under a thousand");

        let big: Vec<String> = (0..2001).map(|i| format!("h/x@{i}")).collect();
        s.reset_op_counts();
        let report = s.delete_many(&big).await.unwrap();
        assert_eq!(report.deleted, 2001);
        assert_eq!(s.op_counts().get("delete").copied(), Some(3), "a thousand keys per request");
    }

    #[tokio::test]
    async fn conditional_puts_and_gets_enforce_the_contract() {
        let s = MemoryStore::new();
        let body = Bytes::from_static(b"gen-1");
        let crc = crc64_nvme(&body);

        // A wrong claimed checksum must fail the publish, not warn.
        let err = s
            .put_whole("k", body.clone(), &PutCondition::IfNoneMatchAny, &stamps(1), crc ^ 1)
            .await
            .unwrap_err();
        assert!(matches!(err, StoreError::ChecksumMismatch(_)));

        let m1 = s
            .put_whole("k", body.clone(), &PutCondition::IfNoneMatchAny, &stamps(1), crc)
            .await
            .unwrap();
        // Create-new against an existing object: 412.
        let err = s
            .put_whole("k", body.clone(), &PutCondition::IfNoneMatchAny, &stamps(1), crc)
            .await
            .unwrap_err();
        assert!(matches!(err, StoreError::PreconditionFailed(_)));

        // Guarded update with the right etag works; the stale etag 412s.
        let b2 = Bytes::from_static(b"gen-2");
        let m2 = s
            .put_whole("k", b2.clone(), &PutCondition::IfMatch(m1.etag.clone()), &stamps(2), crc64_nvme(&b2))
            .await
            .unwrap();
        let err = s
            .put_whole("k", b2.clone(), &PutCondition::IfMatch(m1.etag.clone()), &stamps(3), crc64_nvme(&b2))
            .await
            .unwrap_err();
        assert!(matches!(err, StoreError::PreconditionFailed(_)));

        // Guarded GET: the hydration posture's input — stale etag 412s.
        let (gm, gb) = s.get_whole("k", Some(&m2.etag)).await.unwrap();
        assert_eq!(gb, b2);
        assert_eq!(GenerationStamps::from_meta(&gm.meta).unwrap().generation, 2);
        let err = s.get_whole("k", Some(&m1.etag)).await.unwrap_err();
        assert!(matches!(err, StoreError::PreconditionFailed(_)));
    }

    #[tokio::test]
    async fn epoch_cas_lease_lifecycle() {
        let s = MemoryStore::new();
        const K: &str = "vol/.flint-epoch";

        // First claim creates; a second blind claim 412s.
        let l1 = s.epoch_acquire(K, "hub-a", None).await.unwrap();
        assert_eq!(l1.epoch, 1);
        let err = s.epoch_acquire(K, "hub-b", None).await.unwrap_err();
        assert!(matches!(err, StoreError::PreconditionFailed(_)));

        // Renew with the live token; the token rotates so the OLD one
        // is dead afterwards (a stale holder's heartbeat must fail).
        let l1b = s.epoch_renew(K, &l1, None).await.unwrap();
        assert_ne!(l1b.token, l1.token, "renew must rotate the CAS token");
        let err = s.epoch_renew(K, &l1, None).await.unwrap_err();
        assert!(matches!(err, StoreError::PreconditionFailed(_)), "stale token must be dead");

        // Takeover: supersede the OBSERVED state (step 7 judges the
        // holder dead first); epoch increments; the deposed lease is
        // fully fenced.
        let observed = s.epoch_read(K).await.unwrap().unwrap();
        assert_eq!(observed.holder_id, "hub-a");
        let l2 = s.epoch_acquire(K, "hub-b", Some(&observed)).await.unwrap();
        assert_eq!(l2.epoch, 2);
        let err = s.epoch_renew(K, &l1b, None).await.unwrap_err();
        assert!(matches!(err, StoreError::PreconditionFailed(_)), "deposed renew must fail");
        let err = s.epoch_release(K, &l1b).await.unwrap_err();
        assert!(matches!(err, StoreError::PreconditionFailed(_)), "deposed release must fail");

        // The holder releases: the cell is MARKED, not deleted. Losing
        // it would restart epoch numbering at 1 for the next claimant,
        // and every publish stamp on the volume is already past that.
        s.epoch_release(K, &l2).await.unwrap();
        let after = s.epoch_read(K).await.unwrap().expect("the cell must survive a release");
        assert!(after.released, "a clean shutdown marks the cell released");
        assert_eq!(after.epoch, 2, "and the epoch number is preserved");
        assert_eq!(after.holder_id, "hub-b");

        // The successor supersedes the released cell at epoch+1 — the
        // fast path `claim` takes with no quiet wait.
        let l3 = s.epoch_acquire(K, "hub-c", Some(&after)).await.unwrap();
        assert_eq!(l3.epoch, 3, "numbering continues, it does not reset");
        assert!(
            !s.epoch_read(K).await.unwrap().unwrap().released,
            "a fresh claim clears the released mark"
        );
    }

    /// The per-barrier lease's queue (lean, design 2026-09-13 §4.2):
    /// a waiter appends itself ONCE, the handoff pops the head into the
    /// reservation and keeps the echo, only the reserved holder's
    /// acquire consumes the reservation, and every transition rotates
    /// the token so a heartbeat issued against the pre-enqueue token
    /// is dead — which is what forces the holder to re-read and carry
    /// the longer queue forward.
    #[tokio::test]
    async fn epoch_queue_hands_the_cell_to_the_head_and_keeps_the_echo() {
        const K: &str = "t/.flint/lean/epoch";
        let s = MemoryStore::new();
        let a = s.epoch_acquire(K, "a", None).await.unwrap();
        let seen = s.epoch_read(K).await.unwrap().unwrap();
        let seen = s.epoch_enqueue(K, &seen, "b").await.unwrap();
        let seen = s.epoch_enqueue(K, &seen, "c").await.unwrap();
        let seen = s.epoch_enqueue(K, &seen, "b").await.unwrap();
        assert_eq!(seen.waiters, vec!["b", "c"], "a waiter is appended once, in order");
        // The holder's renew with its claim-time token is dead: the
        // queue moved the token.
        let err = s.epoch_renew(K, &a, None).await.unwrap_err();
        assert!(matches!(err, StoreError::PreconditionFailed(_)));
        // The holder adopts the observed token + queue and renews; the
        // renew CARRIES the queue.
        let adopted = EpochLease {
            holder_id: "a".into(),
            epoch: seen.epoch,
            token: seen.token.clone(),
            waiters: seen.waiters.clone(),
        };
        let a = s.epoch_renew(K, &adopted, Some("echo-a")).await.unwrap();
        assert_eq!(s.epoch_read(K).await.unwrap().unwrap().waiters, vec!["b", "c"]);
        // Handoff: released, reserved for b, c still queued, echo KEPT.
        s.epoch_handoff(K, &a, Some("echo-a")).await.unwrap();
        let cell = s.epoch_read(K).await.unwrap().unwrap();
        assert!(cell.released);
        assert_eq!(cell.handoff.as_deref(), Some("b"));
        assert_eq!(cell.waiters, vec!["c"]);
        assert_eq!(cell.echo.as_deref(), Some("echo-a"), "the last barrier's echo survives the handoff");
        // b acquires: reservation consumed, c stays queued, epoch+1.
        let b = s.epoch_acquire(K, "b", Some(&cell)).await.unwrap();
        assert_eq!(b.epoch, cell.epoch + 1);
        assert_eq!(b.waiters, vec!["c"]);
        let cell = s.epoch_read(K).await.unwrap().unwrap();
        assert!(!cell.released && cell.handoff.is_none());
        assert_eq!(cell.holder_id, "b");
        assert_eq!(cell.waiters, vec!["c"]);
        // b's handoff reserves c with an empty queue behind it; c
        // acquires with nobody left queued.
        s.epoch_handoff(K, &b, None).await.unwrap();
        let cell = s.epoch_read(K).await.unwrap().unwrap();
        assert_eq!(cell.handoff.as_deref(), Some("c"));
        assert!(cell.waiters.is_empty());
        let c = s.epoch_acquire(K, "c", Some(&cell)).await.unwrap();
        assert!(c.waiters.is_empty());
        s.epoch_handoff(K, &c, None).await.unwrap();
        let cell = s.epoch_read(K).await.unwrap().unwrap();
        assert!(cell.released && cell.handoff.is_none() && cell.waiters.is_empty());
    }

    #[tokio::test]
    async fn compose_mixes_local_and_guarded_base_copy() {
        let s = MemoryStore::new();
        // Base generation: 8 bytes.
        let base_body = Bytes::from_static(b"AAAABBBB");
        let base = s
            .put_whole("f", base_body.clone(), &PutCondition::IfNoneMatchAny, &stamps(1), crc64_nvme(&base_body))
            .await
            .unwrap();
        // Local truth: first 4 bytes rewritten, tail clean.
        let dir = tempfile::TempDir::new().unwrap();
        let local = dir.path().join("f.local");
        std::fs::write(&local, b"XXXXBBBB").unwrap();

        let spec = ComposeSpec {
            progress: None,
            key: "f",
            local: Some(std::sync::Arc::new(std::fs::File::open(&local).unwrap())),
            parts: vec![
                PartSource::Local { offset: 0, len: 4 },
                PartSource::BaseCopy { offset: 4, len: 4 },
            ],
            base_key: None,
            base_etag: Some(base.etag.clone()),
            condition: PutCondition::IfMatch(base.etag.clone()),
            stamps: stamps(2),
            crc64: Some(crc64_nvme(b"XXXXBBBB")),
        };
        let m2 = s.compose_generation(&spec).await.unwrap();
        let (_, bytes) = s.get_whole("f", Some(&m2.etag)).await.unwrap();
        assert_eq!(bytes, Bytes::from_static(b"XXXXBBBB"));
        assert!(m2.etag.ends_with("-2\""), "multipart etag must not look like a content hash");

        // A foreign overwrite between generations fences the NEXT
        // compose at the copy-source guard.
        s.raw_put("f", Bytes::from_static(b"foreignXX"), vec![]);
        let err = s.compose_generation(&spec).await.unwrap_err();
        assert!(matches!(err, StoreError::PreconditionFailed(_)));
        assert!(
            s.list_uploads("f").await.unwrap().is_empty(),
            "a failed compose must abort its MPU (A9)"
        );
    }
    /// `crc64: None` — the checksum comes from the parts the store
    /// reads, and the published metadata carries the object's true
    /// CRC. A copied part cannot be hashed that way and is refused
    /// before anything moves, with no upload left pending.
    #[tokio::test]
    async fn compose_without_a_claimed_checksum_publishes_the_content_crc() {
        let s = MemoryStore::new();
        let dir = tempfile::TempDir::new().unwrap();
        let local = dir.path().join("g.local");
        let content: Vec<u8> = (0..24u8).collect();
        std::fs::write(&local, &content).unwrap();
        let spec = ComposeSpec {
            progress: None,
            key: "g",
            local: Some(std::sync::Arc::new(std::fs::File::open(&local).unwrap())),
            parts: vec![
                PartSource::Local { offset: 0, len: 8 },
                PartSource::Local { offset: 8, len: 8 },
                PartSource::Local { offset: 16, len: 8 },
            ],
            base_key: None,
            base_etag: None,
            condition: PutCondition::IfNoneMatchAny,
            stamps: stamps(1),
            crc64: None,
        };
        let m = s.compose_generation(&spec).await.unwrap();
        assert_eq!(m.crc64_b64.as_deref(), Some(crc64_to_b64(crc64_nvme(&content)).as_str()));
        let (h, bytes) = s.get_whole("g", None).await.unwrap();
        assert_eq!(&bytes[..], &content[..]);
        assert_eq!(h.crc64_b64, m.crc64_b64);

        let base_body = Bytes::from_static(b"AAAABBBB");
        let base = s
            .put_whole("h", base_body.clone(), &PutCondition::IfNoneMatchAny, &stamps(1), crc64_nvme(&base_body))
            .await
            .unwrap();
        std::fs::write(&local, b"XXXXBBBB").unwrap();
        let err = s
            .compose_generation(&ComposeSpec {
                progress: None,
                key: "h",
                local: Some(std::sync::Arc::new(std::fs::File::open(&local).unwrap())),
                parts: vec![
                    PartSource::Local { offset: 0, len: 4 },
                    PartSource::BaseCopy { offset: 4, len: 4 },
                ],
                base_key: None,
                base_etag: Some(base.etag.clone()),
                condition: PutCondition::IfMatch(base.etag.clone()),
                stamps: stamps(2),
                crc64: None,
            })
            .await
            .unwrap_err();
        assert!(matches!(err, StoreError::Other(ref m) if m.contains("copied part")), "{err:?}");
        assert!(s.list_uploads("h").await.unwrap().is_empty(), "nothing pending after a refusal");
    }
}

#[cfg(test)]
mod copy_tests {
    use super::*;
    use crate::{crc64_from_b64, crc64_to_b64};

    fn stamps(g: u64) -> GenerationStamps {
        GenerationStamps {
            generation: g,
            epoch: 7,
            flush_uuid: format!("flush-{g}"),
            boundary_source: None,
            posix: None,
        }
    }

    #[tokio::test]
    async fn a_copy_carries_the_bytes_and_the_checksum_but_not_the_stamps() {
        let s = MemoryStore::new();
        let body = Bytes::from_static(b"the bytes that move");
        let src = s
            .put_whole("a.txt", body.clone(), &PutCondition::IfNoneMatchAny, &stamps(3), crc64_nvme(&body))
            .await
            .unwrap();

        let dst = s
            .copy_object("a.txt", Some(&src.etag), "b.txt", &PutCondition::IfNoneMatchAny, &stamps(1))
            .await
            .unwrap();

        let (meta, got) = s.get_whole("b.txt", None).await.unwrap();
        assert_eq!(got, body, "a copy is byte-identical or it is not a copy");
        assert_eq!(
            dst.crc64_b64, src.crc64_b64,
            "identical bytes have an identical checksum — a copy cannot change one"
        );
        // THE trap this test exists for: the destination must not wear
        // the source's publish history.
        assert_eq!(
            meta.meta.get("flint-gen").map(String::as_str),
            Some("1"),
            "the destination's generation is its OWN, not the source's 3: {:?}",
            meta.meta
        );
        assert_eq!(
            meta.meta.get(GenerationStamps::META_FLUSH_UUID).map(String::as_str),
            Some("flush-1"),
            "and its flush_uuid names the publish that WROTE IT, not flush-3: {:?}",
            meta.meta
        );
        // The source is untouched — copy, not move.
        assert!(s.head("a.txt").await.is_ok());
    }

    #[tokio::test]
    async fn a_copy_refuses_when_the_source_moved_under_it() {
        let s = MemoryStore::new();
        let v1 = Bytes::from_static(b"version one");
        let src = s
            .put_whole("a.txt", v1, &PutCondition::IfNoneMatchAny, &stamps(1), crc64_nvme(b"version one"))
            .await
            .unwrap();
        let v2 = Bytes::from_static(b"version two");
        s.put_whole("a.txt", v2, &PutCondition::IfMatch(src.etag.clone()), &stamps(2), crc64_nvme(b"version two"))
            .await
            .unwrap();

        let err = s
            .copy_object("a.txt", Some(&src.etag), "b.txt", &PutCondition::IfNoneMatchAny, &stamps(1))
            .await
            .expect_err("copying an object that moved copies something the caller never read");
        assert!(matches!(err, StoreError::PreconditionFailed(_)), "{err:?}");
        assert!(matches!(s.head("b.txt").await, Err(StoreError::NotFound(_))), "and nothing landed");
    }

    #[tokio::test]
    async fn a_copy_honours_the_destination_precondition() {
        let s = MemoryStore::new();
        let a = Bytes::from_static(b"aaaa");
        s.put_whole("a.txt", a, &PutCondition::IfNoneMatchAny, &stamps(1), crc64_nvme(b"aaaa"))
            .await
            .unwrap();
        let b = Bytes::from_static(b"bbbb");
        let existing = s
            .put_whole("b.txt", b.clone(), &PutCondition::IfNoneMatchAny, &stamps(1), crc64_nvme(b"bbbb"))
            .await
            .unwrap();

        let err = s
            .copy_object("a.txt", None, "b.txt", &PutCondition::IfNoneMatchAny, &stamps(2))
            .await
            .expect_err("a rename must not silently clobber its destination");
        assert!(matches!(err, StoreError::PreconditionFailed(_)), "{err:?}");
        let (_, still) = s.get_whole("b.txt", Some(&existing.etag)).await.unwrap();
        assert_eq!(still, b, "the destination is untouched by a refused copy");
    }

    /// `ComposeSpec::base_key` has existed for the A7 re-key flush and
    /// `base_key: Some(..)` appears NOWHERE in this repository — not in
    /// production, not in a test. `copy_object`'s MPU arm is its first
    /// caller, so this is the branch's first coverage.
    #[tokio::test]
    async fn a_compose_can_copy_ranges_from_a_different_key() {
        let s = MemoryStore::new();
        let body = Bytes::from_static(b"AAAABBBB");
        let src = s
            .put_whole("old/name", body.clone(), &PutCondition::IfNoneMatchAny, &stamps(1), crc64_nvme(&body))
            .await
            .unwrap();

        let spec = ComposeSpec {
            progress: None,
            key: "new/name",
            // Never read: every part is a BaseCopy.
            local: None,
            parts: vec![PartSource::BaseCopy { offset: 0, len: 8 }],
            base_key: Some("old/name"),
            base_etag: Some(src.etag.clone()),
            condition: PutCondition::IfNoneMatchAny,
            stamps: stamps(1),
            crc64: Some(crc64_nvme(&body)),
        };
        let m = s.compose_generation(&spec).await.unwrap();
        let (_, got) = s.get_whole("new/name", Some(&m.etag)).await.unwrap();
        assert_eq!(got, body, "the ranges came from the OTHER key");
        assert!(s.head("old/name").await.is_ok(), "and the source survives");
    }

    #[test]
    fn crc64_b64_round_trips_and_refuses_what_is_not_one() {
        for v in [0u64, 1, u64::MAX, 0x0123_4567_89ab_cdef] {
            assert_eq!(crc64_from_b64(&crc64_to_b64(v)), Some(v), "round trip {v:#x}");
        }
        // A caller handed `None` has something other than a CRC-64, and
        // guessing a number would validate a publish against a checksum
        // nobody computed.
        // "AAAAAAAAAAA" is the discriminating case: 11 significant
        // characters decode to exactly 8 bytes, so the byte-count check
        // alone ACCEPTS it. Only the canonical-form check rejects it —
        // without that case the padding guard is a line no control can
        // move, which is how dead defensive code survives.
        for bad in ["", "AAAA", "not-base64!!", "AAAAAAAAAAAA", "AAAAAAAAAAA", "AAAAAAAAAAAAAAA="] {
            assert_eq!(crc64_from_b64(bad), None, "{bad:?} is not a canonical CRC-64");
        }
        assert_eq!(crc64_from_b64("AAAAAAAAAAA="), Some(0), "the canonical form IS accepted");
    }
}