alkblobs/src/local.rs

//! The `local` large-tier engine (ADR-003/008): flat
//! `{hex-prefix}/{hex-prefix}/{hash}` sharding, stage-then-commit-rename
//! staged puts (ADR-010 §1's named staged-put contract), and pread-based
//! read cursors — the ADR-003 backend as specced, ungated
//! (dependency-free, always present; ADR-013 §4).
//!
//! **Filename encoding (named design point, recorded for the review
//! gate):** committed filenames are the lowercase hex of the full key
//! bytes (`Key::as_bytes` — the alg-tagged, self-describing encoding
//! ADR-002's persistence-boundary ruling requires for anything with no
//! schema-migration story). Shard prefixes are taken from the hex
//! string itself (`hex[..2]`/`hex[2..4]`), so the layout is fully
//! decodable from the filesystem tree alone; `list` decodes filenames
//! back to key bytes and they round-trip through
//! `Key::from_bytes` exactly. A filename that does not decode — as
//! would a foreign or corrupted entry — is a **fatal list error**
//! (`Error::Io` naming the path): a malformed list is as deadly as an
//! incomplete one (ADR-003), and it is silent until a sweep deletes
//! the wrong things. The walk's exclusion is by root-entry *name*:
//! the engine's own staging directory (`staging_dir`) is the known
//! non-shard entry, and a stray root-level *file* named `staging` is
//! likewise excluded (name equals, type unchecked).
//!
//! **Concurrency posture:** every file operation — constructor,
//! stat/open, pread, append, commit-rename, the CAS whole-value put's
//! write-then-rename, delete, the list walk, and discard — runs in
//! `tokio::task::spawn_blocking` (store-api.md §Concurrency posture;
//! the store never blocks the executor). The committed file is
//! immutable once published, so the read cursor preads against an open
//! handle without synchronization; the staging cursor is the sole
//! writer of its stage file and tracks the staged length in an atomic.
//!
//! **Durability posture (ADR-015):** the local tier publishes only
//! fsync-durable content. The publish is no-replace,
//! hardlink-into-place: the publishing file is fsynced (`sync_all`)
//! before `link(stage, committed)` — which atomically fails when the
//! destination exists, so no committed path is ever replaced (the
//! module's pinned invariant held by the filesystem rather than by
//! check-then-act discipline) — then the stage file is removed and the
//! parent directory is fsynced, the entry's name durable before
//! publish returns. A losing publish race (or a dedup re-put) keeps
//! CAS first-writer-wins: the loser's link fails and its stage
//! discards, never overwriting — the *primitive* fails loudly, and the
//! engine's CAS contract maps the loss to the dedup form every re-put
//! shares (`Ok`, stage discarded, first writer's bytes stand). The
//! fsync scope is the publish paths only (the correctness boundary):
//! staging intermediates, delete paths, and discard removals are not
//! fsynced — a delete lost to a crash re-surfaces valid committed
//! content (harmless under CAS); a stage file lost or half-lost is not
//! a committed entry. Operator-facing contract: **commits are
//! crash-durable and never corrupt; crashes cost recent in-flight
//! puts, restored by re-put.** (Platform note: std offers no primitive
//! to open a directory on non-unix, so there the directory-fsync leg
//! is reduced to a no-op; the file-fsync leg is unconditional on
//! both.)
//!
//! **The committed-tree doctrine (review-001 S-6):** committed paths
//! hold regular files; any non-regular entry at one — a planted
//! symlink, a directory — is malformed state **rejected loudly** by
//! every path that touches it (`walk_committed`'s fatal-malformed
//! shape was always the doctrine's other half; the stat/get paths now
//! share it): the portable `symlink_metadata` pre-check
//! (`open_committed`) rejects before any open, the unix open rides
//! `O_NOFOLLOW` (closing the placement race), and the publish's
//! losing-link arm refuses to dedup-map a non-regular destination —
//! one behavior, one doctrine, never silently served and never
//! silently dedup-mapped.
//!
//! **Stage hygiene (store-api.md invariant 3):** publish is the
//! commit-rename; every failure path converges on discard with zero
//! residue — a failed commit removes the stage file itself (there is
//! no post-failure discard to drive), an explicit
//! `StagingCursor::discard` removes it, and a staging cursor dropped
//! without either is best-effort-cleaned by [`Drop`] (the engine's own
//! cancellation
//! story, per the wave-1 review's recorded observation; the hygiene
//! tests below cover all three forms exactly). A mid-staging write
//! failure discards the stage too (F-3.3 — the staged read window can
//! never serve hole-plus-shifted content). CAS is first-writer-wins
//! everywhere a publish meets an existing entry: the committed file for
//! a key is never overwritten — a losing race or a re-put is dedup,
//! landing as `Ok` with the stage discarded.
//!
//! The engine is built by the store facade's local large-tier arms —
//! `dual_tier_mem_local`, `dual_tier_sqlite_local`,
//! `dual_tier_postgres_local` (+ the fleet twin's shared-media
//! declaration) — which give every lib body an in-tree caller; the
//! `#![allow(dead_code)]` below covers only the test-instrument
//! remainder: the cfg(test) staged-commit and staged-append failure
//! injections plus the publish fsync-instrumentation probe (the
//! testable seam behind the ADR-015 publish tests), whose consumers
//! are the store-core tests behind feature `mem` or these in-module
//! tests, so the instruments do not compile into default builds.

#![allow(dead_code)]

use std::fs::{self, File, OpenOptions};
use std::io::{ErrorKind, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
#[cfg(test)]
use std::sync::Mutex;
use std::time::{SystemTime, UNIX_EPOCH};

use crate::backend::{clamped, Backend, ReadCursor, StagingCursor};
use crate::keys::Key;
use crate::Error;

/// The publish-instrumentation probe: production-side it is a
/// zero-sized default marker; under `cfg(test)` it can carry a log the
/// publish path records its fsync/link steps into ([`SyncEvent`]) —
/// the testable seam behind the ADR-015 "fsync calls' presence on the
/// publish paths" tests, recorded as the task's design point (an
/// in-instance log, not a global counter: no cross-test pollution, no
/// lock held across an awaited publish).
#[derive(Clone, Default)]
pub(crate) struct SyncProbe(#[cfg(test)] Option<Arc<Mutex<Vec<SyncEvent>>>>);

/// The publish-path steps a [`SyncProbe`] records, in call order.
#[derive(Debug, PartialEq, Eq, Clone, Copy)]
pub(crate) enum SyncEvent {
    FileSync,
    Link,
    StageUnlinked,
    DirSync,
}

impl SyncProbe {
    fn record(&self, _event: SyncEvent) {
        #[cfg(test)]
        {
            if let Some(log) = &self.0 {
                log.lock().unwrap_or_else(|e| e.into_inner()).push(_event);
            }
        }
    }
}

/// Process-unique counter for stage and put-temp filenames, combined
/// with a wall-clock component so names stay unique across restarts of
/// the same root.
static STAGE_COUNTER: AtomicU64 = AtomicU64::new(0);

pub(crate) struct LocalBackend {
    root: Arc<PathBuf>,
    /// Test-only commit-failure injection (the media failure the local
    /// tier cannot reliably produce in tests); mirrors the mem engine's
    /// knob so the stage-hygiene failure form is exercised against the
    /// real engine.
    #[cfg(test)]
    fail_commit: bool,
    /// Test-only staged-append injection: the number of successful
    /// appends a handed-out staging cursor allows before its next
    /// append write fails ([`u64::MAX`] = never). Disarms per cursor
    /// once fired.
    #[cfg(test)]
    fail_append_after: u64,
    /// Test-only delete-failure injection: every `delete` fails as if
    /// the media refused the removal (the store-core partial-failure
    /// report's driver — review-001 N-7).
    #[cfg(test)]
    fail_delete: bool,
    /// Publish instrumentation ([`SyncEvent`] log); production-side a
    /// marker that records nothing.
    sync_probe: SyncProbe,
}

impl LocalBackend {
    /// Construct the engine over a fresh or existing root directory,
    /// creating the root and its staging area (both blocking work, in
    /// [`tokio::task::spawn_blocking`]).
    pub(crate) async fn new(root: impl Into<PathBuf>) -> Result<Self, Error> {
        let root: Arc<PathBuf> = Arc::new(root.into());
        let staging = staging_dir(&root);
        let path = Arc::clone(&root);
        blocking(move || {
            fs::create_dir_all(&*path).map_err(io_err)?;
            fs::create_dir_all(staging).map_err(io_err)?;
            Ok(())
        })
        .await?;
        Ok(Self {
            root,
            #[cfg(test)]
            fail_commit: false,
            #[cfg(test)]
            fail_append_after: u64::MAX,
            #[cfg(test)]
            fail_delete: false,
            sync_probe: SyncProbe::default(),
        })
    }

    /// Test-only injection arm: force the staged put's commit-rename
    /// failure form so the store-core driver's pin-first/publish-second
    /// ordering (store-fleet-put, ADR-012 §4) is exercised against the
    /// real engine — the pin row lands, the publish fails, the orphan
    /// pin is the expiry machinery's. Mirrors the other engines'
    /// cfg(test) knobs.
    #[cfg(test)]
    pub(crate) fn inject_fail_commit(&mut self) {
        self.fail_commit = true;
    }

    /// Test-only injection arm: hand out staging cursors whose append
    /// after `after` successful writes fails and discards the stage
    /// (F-3.3's mid-write failure, the form real media cannot reliably
    /// produce in tests).
    #[cfg(test)]
    pub(crate) fn inject_fail_append(&mut self, after: u64) {
        self.fail_append_after = after;
    }

    /// Test-only injection arm: every `delete` fails as if the media
    /// refused the removal — the driver for the store-core
    /// partial-failure report test (review-001 N-7), where the large
    /// home's delete must fail after the kv home's succeeded.
    #[cfg(test)]
    pub(crate) fn inject_fail_delete(&mut self) {
        self.fail_delete = true;
    }

    /// Test-only publish instrumentation: from here on, this engine's
    /// publish paths record their fsync/link steps into the returned
    /// log (the ADR-015 publish tests' seam). The live log rides the
    /// engine instance, so concurrent tests do not observe each other.
    #[cfg(test)]
    pub(crate) fn instrument_publish(&mut self) -> Arc<Mutex<Vec<SyncEvent>>> {
        let log = Arc::new(Mutex::new(Vec::new()));
        self.sync_probe = SyncProbe(Some(Arc::clone(&log)));
        log
    }
}

fn io_err(e: std::io::Error) -> Error {
    Error::Io(e.to_string())
}

async fn blocking<F, T>(body: F) -> Result<T, Error>
where
    F: FnOnce() -> Result<T, Error> + Send + 'static,
    T: Send + 'static,
{
    match tokio::task::spawn_blocking(body).await {
        Ok(result) => result,
        Err(e) => Err(Error::Io(format!("blocking task joint failure: {e}"))),
    }
}

/// The staging area under the root — the one directory of the layout
/// that holds in-flight stage and put-temp files, and the one entry the
/// committed-tree walk skips.
fn staging_dir(root: &Path) -> PathBuf {
    root.join("staging")
}

/// A wall-clock-plus-counter unique id for stage and put temp filenames.
fn stage_id() -> String {
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos())
        .unwrap_or(0);
    format!(
        "{:020}-{:x}",
        STAGE_COUNTER.fetch_add(1, Ordering::Relaxed),
        nanos
    )
}

/// The committed path for a key: `{root}/{hex-prefix}/{hex-prefix}/{hex}`
/// — prefixes taken from the hex filename string itself. Requires at
/// least two key bytes; real encodings are 21 or 33.
fn committed_path(root: &Path, key: &[u8]) -> Result<PathBuf, Error> {
    let name = hex::encode(key);
    if name.len() < 4 {
        return Err(Error::Io(format!(
            "key too short for the sharded layout (hex {name})"
        )));
    }
    Ok(root.join(&name[..2]).join(&name[2..4]).join(&name))
}

/// A two-character lowercase-hex shard directory name — the only level-1
/// and level-2 names the committed tree recognizes (besides the staging
/// area).
fn is_shard_name(name: &str) -> bool {
    name.len() == 2 && name.bytes().all(|b| matches!(b, b'0'..=b'9' | b'a'..=b'f'))
}

/// The committed-tree doctrine's loud rejection (review-001 S-6): a
/// non-regular entry at a committed path — a planted symlink, a
/// directory — is malformed state, and it is **rejected loudly**
/// ([`Error::Io`] naming the path), the same fatal-malformed doctrine
/// [`walk_committed`] applies, never silently served (the stat/get
/// paths would otherwise follow the symlink and serve its *target*)
/// and never silently dedup-mapped (the publish arm would otherwise
/// land `Ok` without storing content). No new error variant: the
/// failure family is ADR-pinned (the F-4 refusal's precedent).
fn malformed_committed_entry(committed: &Path, found: &str) -> Error {
    Error::Io(format!(
        "malformed committed entry: {committed:?} holds a {found}, \
         not a regular file — rejected loudly, never served or \
         dedup-mapped (the committed-tree doctrine)"
    ))
}

fn malformed_kind(meta: &fs::Metadata) -> &'static str {
    if meta.is_symlink() {
        "symlink"
    } else if meta.is_dir() {
        "directory"
    } else {
        "non-regular entry"
    }
}

/// Open a committed path for reading under the committed-tree doctrine
/// (review-001 S-6): a portable `symlink_metadata` pre-check rejects
/// any non-regular entry at the path loudly
/// ([`malformed_committed_entry`]) — `symlink_metadata` never follows
/// the link, so a planted symlink is rejected here and the target is
/// never served — then the open itself, then the opened handle's own
/// metadata re-checks the type (the non-unix shape of the race window;
/// on unix the open additionally carries `O_NOFOLLOW`, so a symlink
/// slipped in after the pre-check fails the open with the final
/// component loop error, mapped to the same doctrine rejection — the
/// TOCTOU close). A `NotFound` anywhere is the absent entry
/// ([`None`]) — absence stays a value, never the doctrine's
/// rejection.
fn open_committed(committed: &Path) -> Result<Option<(File, u64)>, Error> {
    match fs::symlink_metadata(committed) {
        Ok(meta) if !meta.is_file() => {
            return Err(malformed_committed_entry(committed, malformed_kind(&meta)));
        }
        Err(e) if e.kind() == ErrorKind::NotFound => return Ok(None),
        Err(e) => return Err(io_err(e)),
        Ok(_) => {}
    }
    let opened = {
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            OpenOptions::new()
                .read(true)
                .custom_flags(libc::O_NOFOLLOW)
                .open(committed)
        }
        #[cfg(not(unix))]
        {
            File::open(committed)
        }
    };
    match opened {
        Ok(file) => {
            let meta = file.metadata().map_err(io_err)?;
            if !meta.is_file() {
                return Err(malformed_committed_entry(committed, malformed_kind(&meta)));
            }
            Ok(Some((file, meta.len())))
        }
        Err(e) if e.kind() == ErrorKind::NotFound => Ok(None),
        Err(e) => {
            #[cfg(unix)]
            if e.raw_os_error() == Some(libc::ELOOP) {
                return Err(malformed_committed_entry(
                    committed,
                    "symlink (the hardened open refused to follow it)",
                ));
            }
            Err(io_err(e))
        }
    }
}

/// The committed-tree walk behind [`Backend::list`]: enumerate every
/// shard directory, decode each filename from hex back to key bytes and
/// validate the decode (`Key::from_bytes`). Complete *and* well-formed
/// by contract (ADR-003): anything in the tree that is not a valid
/// committed entry — a non-shard directory, a stray file, an undecodable
/// filename — is `Err(Error::Io)` naming the offending path, never a
/// silent skip (a malformed list is as deadly as an incomplete one; it
/// is silent until a sweep deletes the wrong things).
fn walk_committed(root: &Path) -> Result<Vec<Vec<u8>>, Error> {
    let mut keys = Vec::new();
    for shard1 in fs::read_dir(root).map_err(io_err)? {
        let shard1 = shard1.map_err(io_err)?;
        let name1 = shard1.file_name().to_string_lossy().into_owned();
        if name1 == "staging" {
            continue;
        }
        if !shard1.file_type().map_err(io_err)?.is_dir() || !is_shard_name(&name1) {
            return Err(Error::Io(format!(
                "malformed pool entry at shard level 1: {name1}"
            )));
        }
        for shard2 in fs::read_dir(shard1.path()).map_err(io_err)? {
            let shard2 = shard2.map_err(io_err)?;
            let name2 = shard2.file_name().to_string_lossy().into_owned();
            if !shard2.file_type().map_err(io_err)?.is_dir() || !is_shard_name(&name2) {
                return Err(Error::Io(format!(
                    "malformed pool entry at shard level 2: {name1}/{name2}"
                )));
            }
            for entry in fs::read_dir(shard2.path()).map_err(io_err)? {
                let entry = entry.map_err(io_err)?;
                let file_name = entry.file_name().to_string_lossy().into_owned();
                let name = format!("{name1}/{name2}/{file_name}");
                if !entry.file_type().map_err(io_err)?.is_file() {
                    return Err(Error::Io(format!(
                        "malformed pool entry at blob level: {name}"
                    )));
                }
                let bytes = hex::decode(&file_name).map_err(|_| {
                    Error::Io(format!("malformed committed filename (not hex): {name}"))
                })?;
                Key::from_bytes(&bytes).map_err(|_| {
                    Error::Io(format!("malformed committed filename (not a key): {name}"))
                })?;
                keys.push(bytes);
            }
        }
    }
    Ok(keys)
}

/// Fill `buf` entirely from `file` at `offset` — the pread loop of the
/// large body family (POC #3 finding A2's sound read window). Callers
/// clamp against the body's length first, so EOF is never reached on a
/// healthy entry; a short pread — the file ending before the clamped
/// span is filled — is a **short-pread error**, agreeing with the
/// non-unix `read_exact` fallback (ADR-015; F-3.1: a zero-filled tail
/// would be silent, uncaught corruption under ADR-006's
/// no-reverify-on-read posture). [`ReadCursor::read_at`]'s
/// EOF-clamping happens in the caller's clamp against the recorded
/// length, never here.
#[cfg(unix)]
fn pread_all(file: &File, mut buf: &mut [u8], mut offset: u64) -> std::io::Result<()> {
    use std::os::unix::fs::FileExt;
    while !buf.is_empty() {
        match file.read_at(buf, offset) {
            Ok(0) => return Err(std::io::Error::new(
                ErrorKind::UnexpectedEof,
                "short pread: the file ended before the clamped span was filled (truncated entry)",
            )),
            Ok(n) => {
                buf = &mut buf[n..];
                offset += n as u64;
            }
            Err(e) if e.kind() == ErrorKind::Interrupted => {}
            Err(e) => return Err(e),
        }
    }
    Ok(())
}

#[cfg(not(unix))]
fn pread_all(file: &File, buf: &mut [u8], offset: u64) -> std::io::Result<()> {
    use std::io::{Read, Seek, SeekFrom};
    let mut handle = file.try_clone()?;
    handle.seek(SeekFrom::Start(offset))?;
    handle.read_exact(buf)?;
    Ok(())
}

/// Flush a directory's entries to storage (the directory fsync after a
/// publish's directory-entry mutation; ADR-015). Unix opens the
/// directory and `sync_all`s it; std offers no primitive for opening a
/// directory on non-unix, so there this is a recorded no-op (the
/// file-fsync leg is unconditional).
#[cfg(unix)]
fn sync_dir(dir: &Path) -> std::io::Result<()> {
    File::open(dir)?.sync_all()
}

#[cfg(not(unix))]
fn sync_dir(_dir: &Path) -> std::io::Result<()> {
    Ok(())
}

/// The fsync-durable no-replace publish (ADR-015): fsync the publishing
/// file, then `link` it into the committed path — atomically failing
/// with [`ErrorKind::AlreadyExists`] when an entry is already there (a
/// publish race's loser, or a dedup re-put), which discards the stage
/// copy — never a replacement. The losing link's arm honors the
/// committed-tree doctrine (review-001 S-6): the destination is
/// re-statted with `symlink_metadata` (which never follows a link) — a
/// regular file is the dedup arm (CAS first-writer-wins, the stage copy
/// discarded and the publish maps the loss to the dedup form every
/// re-put shares, `Ok`); a **symlink** — or any non-regular entry — at
/// the committed path is malformed state, rejected loudly
/// ([`malformed_committed_entry`]) — never silently dedup-mapped, the
/// stage copy discarded with zero residue and nothing written through
/// the link's target (the failed link never touches the destination,
/// ADR-015; the silent-dedup shape the review named is closed).
/// A destination vanished between the failed link and the re-stat (the
/// GC-reaped race) keeps the CAS racing-publisher shape, `Ok` with the
/// stage copy discarded (`NotFound`, mapped alongside the dedup form —
/// the same lie the plain dedup arm shares, where a sweep reaps the
/// first writer between a loser's link and its return). A winning link
/// removes the stage file (a discard-path unlink, unfsynced per the
/// ADR's scope) and fsyncs the committed entry's parent directory.
fn fsync_publish(
    stage: &Path,
    stage_file: &File,
    committed: &Path,
    sync_probe: &SyncProbe,
) -> Result<(), Error> {
    sync_probe.record(SyncEvent::FileSync);
    stage_file.sync_all().map_err(io_err)?;
    let parent = committed
        .parent()
        .ok_or_else(|| Error::Io("committed path has no parent".to_owned()))?;
    fs::create_dir_all(parent).map_err(io_err)?;
    sync_probe.record(SyncEvent::Link);
    match fs::hard_link(stage, committed) {
        Ok(()) => {
            sync_probe.record(SyncEvent::StageUnlinked);
            fs::remove_file(stage).map_err(io_err)?;
            sync_probe.record(SyncEvent::DirSync);
            sync_dir(parent).map_err(io_err)?;
            Ok(())
        }
        Err(e) if e.kind() == ErrorKind::AlreadyExists => {
            sync_probe.record(SyncEvent::StageUnlinked);
            fs::remove_file(stage).map_err(io_err)?;
            match fs::symlink_metadata(committed) {
                Ok(meta) if meta.is_file() => Ok(()),
                Ok(meta) => Err(malformed_committed_entry(committed, malformed_kind(&meta))),
                Err(err) if err.kind() == ErrorKind::NotFound => Ok(()),
                Err(err) => Err(io_err(err)),
            }
        }
        Err(e) => Err(io_err(e)),
    }
}

/// The pread read cursor over a committed entry (ADR-010 §1's streaming
/// large body family): the file is immutable once committed, so reads
/// pread against an open handle at arbitrary offsets — never a content
/// load at open time (the handle and its length stat are all `get`
/// does).
#[derive(Debug)]
pub(crate) struct LocalCursor {
    file: File,
    len: u64,
}

impl ReadCursor for LocalCursor {
    fn len(&self) -> u64 {
        self.len
    }

    async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
        let (start, end) = clamped(offset, len, self.len as usize);
        let wanted = end - start;
        if wanted == 0 {
            return Ok(Vec::new());
        }
        let file = self.file.try_clone().map_err(io_err)?;
        blocking(move || {
            let mut buf = vec![0u8; wanted];
            pread_all(&file, &mut buf, start as u64).map_err(io_err)?;
            Ok(buf)
        })
        .await
    }
}

/// The staged put's staging cursor (ADR-010 §1): backed by a stage file
/// in the staging area, unique per stage — appends write through, the
/// staged length tracks completed appends (the sole-writer protocol),
/// and `commit(key)` publishes the stage file into its sharded
/// committed path under the store-core-resolved canonical key (the
/// staged-put seam's key-at-commit design point: the derivation
/// completes at stream end, so until commit the stage's identity is the
/// engine-internal stage file alone) — publish is the ADR-015
/// no-replace hardlink-into-place (fsync-durable). Every failure path
/// converges on discard with zero residue: a commit that cannot
/// complete (media failure, CAS dedup, parent creation) removes the
/// stage file itself; an append write that fails mid-staging discards
/// the stage and fails the put (F-3.3 — the read window never serves
/// hole-plus-shifted content); [`StagingCursor::discard`] removes it;
/// and a cursor dropped without either is best-effort removed by the
/// [`Drop`] body — the engine's cancellation story, covered by the
/// in-module hygiene tests.
pub(crate) struct LocalStaging {
    root: Arc<PathBuf>,
    stage_path: Option<PathBuf>,
    file: File,
    staged_len: AtomicU64,
    #[cfg(test)]
    fail_commit: bool,
    /// Test-only append-failure countdown ([`u64::MAX`] = never);
    /// disarms once fired.
    #[cfg(test)]
    fail_append_after: u64,
    /// Publish instrumentation ([`SyncEvent`] log); production-side a
    /// marker that records nothing.
    sync_probe: SyncProbe,
}

impl LocalStaging {
    /// Remove the staging file the failed path leaves behind; the
    /// stage_path bookkeeping is already cleared so the [`Drop`] body
    /// is a no-op on this state.
    async fn discard_stage_file(&mut self) {
        if let Some(stage) = self.stage_path.take() {
            let _ = blocking(move || {
                if stage.exists() {
                    fs::remove_file(&stage).map_err(io_err)?;
                }
                Ok(())
            })
            .await;
        }
    }
}

impl ReadCursor for LocalStaging {
    fn len(&self) -> u64 {
        self.staged_len.load(Ordering::Acquire)
    }

    async fn read_at(&self, offset: u64, len: u64) -> Result<Vec<u8>, Error> {
        let (start, end) = clamped(
            offset,
            len,
            self.staged_len.load(Ordering::Acquire) as usize,
        );
        let wanted = end - start;
        if wanted == 0 {
            return Ok(Vec::new());
        }
        let file = self.file.try_clone().map_err(io_err)?;
        blocking(move || {
            let mut buf = vec![0u8; wanted];
            pread_all(&file, &mut buf, start as u64).map_err(io_err)?;
            Ok(buf)
        })
        .await
    }
}

impl StagingCursor for LocalStaging {
    async fn append(&mut self, chunk: &[u8]) -> Result<(), Error> {
        if chunk.is_empty() {
            return Ok(());
        }
        #[cfg(test)]
        if self.fail_append_after == 0 {
            self.fail_append_after = u64::MAX;
            self.discard_stage_file().await;
            return Err(Error::Io(
                "staging append failure (local, injected)".to_owned(),
            ));
        }
        let mut file = self.file.try_clone().map_err(io_err)?;
        let chunk = chunk.to_vec();
        let count = chunk.len() as u64;
        match blocking(move || file.write_all(&chunk).map_err(io_err)).await {
            Ok(()) => {
                #[cfg(test)]
                if self.fail_append_after != u64::MAX {
                    self.fail_append_after -= 1;
                }
                self.staged_len.fetch_add(count, Ordering::AcqRel);
                Ok(())
            }
            Err(e) => {
                self.discard_stage_file().await;
                Err(e)
            }
        }
    }

    async fn commit(mut self, key: &[u8]) -> Result<(), Error> {
        let stage = match self.stage_path.take() {
            Some(stage) => stage,
            None => {
                return Err(Error::Io("staging cursor already consumed".to_owned()));
            }
        };
        let root = Arc::clone(&self.root);
        let key = key.to_vec();
        let cleanup = stage.clone();
        let sync_probe = self.sync_probe.clone();
        let result = blocking(move || {
            #[cfg(test)]
            if self.fail_commit {
                return Err(Error::Io(
                    "engine commit failure (local, injected)".to_owned(),
                ));
            }
            let committed = committed_path(&root, &key)?;
            fsync_publish(&stage, &self.file, &committed, &sync_probe)
        })
        .await;
        if result.is_err() {
            let _ = fs::remove_file(cleanup);
        }
        result
    }

    async fn discard(mut self) {
        self.discard_stage_file().await;
    }
}

impl Drop for LocalStaging {
    /// Best-effort residue cleanup for the cancellation story: a
    /// staging cursor dropped without reaching `commit` or `discard`
    /// removes its stage file (synchronous remove — Drop is sync by
    /// nature; the staging area's contents are the residue this engine
    /// cleans, and the hygiene tests assert it). A cursor reaching
    /// `commit`/`discard` has already cleared the path, so this is a
    /// no-op on those paths.
    fn drop(&mut self) {
        if let Some(stage) = self.stage_path.take() {
            let _ = fs::remove_file(stage);
        }
    }
}

impl Backend for LocalBackend {
    type Cursor = LocalCursor;
    type Staging = LocalStaging;

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

    async fn has(&self, key: &[u8]) -> Result<bool, Error> {
        Ok(self.size(key).await?.is_some())
    }

    async fn get(&self, key: &[u8]) -> Result<Option<Self::Cursor>, Error> {
        let root = Arc::clone(&self.root);
        let key = key.to_vec();
        blocking(move || {
            let committed = committed_path(&root, &key)?;
            match open_committed(&committed)? {
                Some((file, len)) => Ok(Some(LocalCursor { file, len })),
                None => Ok(None),
            }
        })
        .await
    }

    async fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> {
        let root = Arc::clone(&self.root);
        let key = key.to_vec();
        let value = value.to_vec();
        let temp_name = format!("put-{}", stage_id());
        let sync_probe = self.sync_probe.clone();
        blocking(move || {
            let committed = committed_path(&root, &key)?;
            let staging = staging_dir(&root);
            fs::create_dir_all(&staging).map_err(io_err)?;
            let temp = staging.join(temp_name);
            let publish_result = (|| {
                let mut file = OpenOptions::new()
                    .create(true)
                    .truncate(true)
                    .write(true)
                    .open(&temp)
                    .map_err(io_err)?;
                file.write_all(&value).map_err(io_err)?;
                fsync_publish(&temp, &file, &committed, &sync_probe)
            })();
            match publish_result {
                Ok(()) => Ok(()),
                Err(e) => {
                    let _ = fs::remove_file(&temp);
                    Err(e)
                }
            }
        })
        .await
    }

    async fn put_staged(&self) -> Result<Self::Staging, Error> {
        let inner_root = Arc::clone(&self.root);
        let name = format!("stage-{}", stage_id());
        #[cfg(test)]
        let fail_commit = self.fail_commit;
        #[cfg(test)]
        let fail_append_after = self.fail_append_after;
        let sync_probe = self.sync_probe.clone();
        blocking(move || {
            let staging = staging_dir(&inner_root);
            fs::create_dir_all(&staging).map_err(io_err)?;
            let stage_path = staging.join(name);
            let file = OpenOptions::new()
                .read(true)
                .create(true)
                .append(true)
                .open(&stage_path)
                .map_err(io_err)?;
            Ok(LocalStaging {
                root: inner_root,
                stage_path: Some(stage_path),
                file,
                staged_len: AtomicU64::new(0),
                #[cfg(test)]
                fail_commit,
                #[cfg(test)]
                fail_append_after,
                sync_probe,
            })
        })
        .await
    }

    async fn delete(&self, key: &[u8]) -> Result<(), Error> {
        #[cfg(test)]
        if self.fail_delete {
            return Err(Error::Io(
                "engine delete failure (local, injected)".to_owned(),
            ));
        }
        let root = Arc::clone(&self.root);
        let key = key.to_vec();
        blocking(move || {
            let committed = committed_path(&root, &key)?;
            match fs::remove_file(&committed) {
                Ok(()) => Ok(()),
                Err(e) if e.kind() == ErrorKind::NotFound => Err(Error::Missing),
                Err(e) => Err(io_err(e)),
            }
        })
        .await
    }

    async fn list(&self) -> Result<Vec<Vec<u8>>, Error> {
        let root = Arc::clone(&self.root);
        blocking(move || walk_committed(&root)).await
    }

    async fn size(&self, key: &[u8]) -> Result<Option<u64>, Error> {
        let root = Arc::clone(&self.root);
        let key = key.to_vec();
        blocking(move || {
            let committed = committed_path(&root, &key)?;
            match open_committed(&committed)? {
                Some((_, len)) => Ok(Some(len)),
                None => Ok(None),
            }
        })
        .await
    }
}
#[cfg(test)]
mod tests {
    use super::*;
    use crate::hashing::Algorithm;
    use std::time::{SystemTime, UNIX_EPOCH};

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

    /// A fresh temp root per test (the factory's isolation contract in
    /// miniature); Drop removes the whole tree idempotently.
    struct TempRoot(PathBuf);

    fn temp_root(tag: &'static str) -> TempRoot {
        static SEQ: AtomicU64 = AtomicU64::new(0);
        let seq = SEQ.fetch_add(1, Ordering::Relaxed);
        let nanos = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .map(|d| d.as_nanos())
            .unwrap_or(0);
        let dir = std::env::temp_dir().join(format!(
            "alkblobs-local-{tag}-{}-{:x}-{seq}",
            std::process::id(),
            nanos
        ));
        std::fs::create_dir_all(&dir).expect("temp root created");
        TempRoot(dir)
    }

    impl TempRoot {
        fn path(&self) -> &Path {
            &self.0
        }
    }

    impl Drop for TempRoot {
        fn drop(&mut self) {
            let _ = fs::remove_dir_all(&self.0);
        }
    }

    async fn new_backend() -> (TempRoot, LocalBackend) {
        let root = temp_root("engine");
        let backend = LocalBackend::new(root.path()).await.unwrap();
        (root, backend)
    }

    fn key_for(content: &[u8]) -> Vec<u8> {
        Key::from(Algorithm::GitBlobSha256.derive(content, content.len() as u64)).as_bytes()
    }

    fn sha1_key_for(content: &[u8]) -> Vec<u8> {
        let digest = crate::hashing::Algorithm::GitBlobSha1.derive(content, content.len() as u64);
        Key::from(digest).as_bytes()
    }

    /// The real stored value, observed through the read seam.
    async fn stored_value(backend: &LocalBackend, key: &[u8]) -> Option<Vec<u8>> {
        let cursor = backend.get(key).await.expect("get succeeds")?;
        let (bytes, _) = cursor.read_range(0, cursor.len()).await.expect("read");
        Some(bytes)
    }

    /// Files currently in the staging area — the residue counter of the
    /// exact-count hygiene tests.
    fn stage_file_count(root: &Path) -> usize {
        let staging = staging_dir(root);
        match fs::read_dir(&staging) {
            Ok(entries) => entries.count(),
            Err(e) if e.kind() == ErrorKind::NotFound => 0,
            Err(e) => panic!("staging dir unreadable: {e}"),
        }
    }

    async fn staged(backend: &LocalBackend, key: &[u8], content: &[u8]) -> Result<(), Error> {
        let mut staging = backend.put_staged().await?;
        staging.append(content).await?;
        staging.commit(key).await
    }

    /// The Backend method contract: the name, the seven methods'
    /// absence/value/error disciplines, CAS first-writer-wins, and
    /// virgin-store reads as no-ops — mirroring the mem engine's suite.
    #[tokio::test]
    async fn backend_method_contract() {
        let (_root, backend) = new_backend().await;
        let key = key_for(b"method-contract");
        assert_eq!(backend.name(), "local", "reporting identity");

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

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

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

    /// The list contract is complete AND well-formed: both algorithms'
    /// keys land, the walk returns every key byte-exact, and filenames
    /// decode back through `Key::from_bytes` (the named design point's
    /// round-trip).
    #[tokio::test]
    async fn list_is_complete_and_well_formed() {
        let (_root, backend) = new_backend().await;
        let keys = [
            key_for(b"a"),
            sha1_key_for(b"a"),
            key_for(b"bb"),
            sha1_key_for(b"ccc"),
        ];
        for key in &keys {
            backend.put(key, b"value").await.unwrap();
        }

        let mut listed = backend.list().await.unwrap();
        let mut expected = keys.to_vec();
        listed.sort();
        expected.sort();
        assert_eq!(listed, expected, "list is complete and well-formed");

        for raw in &listed {
            let key = Key::from_bytes(raw).expect("listed bytes decode as keys");
            assert_eq!(key.as_bytes(), *raw, "decode round-trips to the same bytes");
            assert!(
                backend.has(raw).await.unwrap(),
                "no spurious or mangled keys"
            );
        }
    }

    /// The filename-encoding design point: the committed file lives at
    /// `{root}/{hex[..2]}/{hex[2..4]}/{hex}` — lowercase hex of the full
    /// key bytes, self-describing through the leading algorithm byte.
    #[tokio::test]
    async fn sharded_layout_and_filename_encoding() {
        let (root, backend) = new_backend().await;
        let key = key_for(CONTENT);
        backend.put(&key, CONTENT).await.unwrap();

        let name = hex::encode(&key);
        assert_eq!(name, name.to_lowercase(), "filename is lowercase hex");
        let path = root.path().join(&name[..2]).join(&name[2..4]).join(&name);
        assert!(
            path.is_file(),
            "committed file at the two-level sharded path"
        );

        let listed = backend.list().await.unwrap();
        assert_eq!(listed, vec![key], "the walk decodes the filename back");
    }

    /// The streaming pread cursor body's read contract: `len()` exact,
    /// EOF-clamped reads (over-read, tail, at-EOF empty, zero-length),
    /// and the ADR-006 range seam's digest over exactly the returned
    /// bytes.
    #[tokio::test]
    async fn cursor_contract_streaming_body() {
        let (_root, backend) = new_backend().await;
        let key = key_for(CONTENT);
        backend.put(&key, CONTENT).await.unwrap();

        let cursor = backend
            .get(&key)
            .await
            .expect("get succeeds")
            .expect("entry present");
        let len = CONTENT.len() as u64;
        assert_eq!(cursor.len(), len, "len is the entry length");

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

        let (slice, digest) = cursor.read_range(mid, 7).await.unwrap();
        assert_eq!(slice, &CONTENT[mid as usize..(mid as usize + 7)]);
        assert_eq!(digest, crate::backend::slice_digest(&slice));
    }

    /// The degenerate case: a zero-byte entry round-trips through put,
    /// the staged form, and the committed cursor (size exact-zero, every
    /// read empty).
    #[tokio::test]
    async fn empty_entry_round_trips() {
        let (_root, backend) = new_backend().await;
        let key = key_for(b"");
        backend.put(&key, b"").await.unwrap();
        assert!(backend.has(&key).await.unwrap(), "empty value still lands");
        assert_eq!(backend.size(&key).await.unwrap(), Some(0), "exact zero");
        let cursor = backend.get(&key).await.unwrap().expect("entry present");
        assert_eq!(cursor.len(), 0);
        assert_eq!(cursor.read_at(0, 8).await.unwrap(), Vec::<u8>::new());

        let staged_key = key_for(b"empty-staged");
        let mut staging = backend.put_staged().await.unwrap();
        staging.append(b"").await.unwrap();
        assert_eq!(staging.len(), 0, "empty append stays empty");
        staging
            .commit(&staged_key)
            .await
            .expect("empty commit publishes");
        assert!(
            backend.has(&staged_key).await.unwrap(),
            "empty entry landed"
        );
        let committed = backend
            .get(&staged_key)
            .await
            .expect("get succeeds")
            .expect("entry present");
        assert_eq!(
            committed.read_at(0, 8).await.unwrap(),
            Vec::<u8>::new(),
            "committed empty body reads empty"
        );
    }

    /// The staged put's happy path: handout, appends, the verification
    /// pass reading the staged artifact back as a whole through the read
    /// seam, commit-rename publishing, and the committed entry
    /// round-tripping through get's cursor — with both re-put forms
    /// (whole-value and staged) dedup no-ops against the live entry.
    #[tokio::test]
    async fn staged_put_round_trips_through_the_read_seam() {
        let (_root, backend) = new_backend().await;
        let key = key_for(b"staged-round-trip");

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

        assert_eq!(
            staging.len() as usize,
            CONTENT.len(),
            "staged length is the appended length"
        );
        let (bytes, digest) = staging
            .read_range(0, staging.len())
            .await
            .expect("staged bytes readable");
        assert_eq!(bytes, CONTENT, "staged artifact readable as a whole");
        assert_eq!(digest, crate::backend::slice_digest(CONTENT));
        let tail = staging.read_at(CONTENT.len() as u64, 32).await.unwrap();
        assert_eq!(tail, Vec::<u8>::new(), "staged read clamps at staged EOF");

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

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

        backend.put(&key, b"overwritten").await.unwrap();
        assert_eq!(
            stored_value(&backend, &key).await.as_deref(),
            Some(CONTENT),
            "whole-value re-put does not overwrite"
        );
        staged(&backend, &key, b"overwritten-by-stage")
            .await
            .expect("staged re-commit is dedup, not failure");
        assert_eq!(
            stored_value(&backend, &key).await.as_deref(),
            Some(CONTENT),
            "staged re-commit does not overwrite"
        );
    }

    /// Stage hygiene, discard form: every non-commit path converges on
    /// discard with zero residue — no stage file, no committed entry.
    #[tokio::test]
    async fn staged_discard_leaves_zero_residue() {
        let (root, backend) = new_backend().await;
        let key = key_for(b"discard-key");

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

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

    /// Stage hygiene, failure form: a failed commit is already
    /// self-discarded — zero residue, entry unpublished, no post-failure
    /// discard left to drive (the consuming call took the cursor). The
    /// injection stands in for the media failure.
    #[tokio::test]
    async fn commit_failure_self_discards() {
        let root = temp_root("commit-failure");
        let backend = LocalBackend {
            root: Arc::new(root.path().to_path_buf()),
            #[cfg(test)]
            fail_commit: true,
            #[cfg(test)]
            fail_append_after: u64::MAX,
            #[cfg(test)]
            fail_delete: false,
            sync_probe: SyncProbe::default(),
        };
        let key = key_for(b"failure-key");

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

        assert!(
            !backend.has(&key).await.unwrap(),
            "failed commit published nothing"
        );
        assert_eq!(backend.size(&key).await.unwrap(), None, "zero residue");
        assert!(
            backend.list().await.unwrap().is_empty(),
            "zero residue in the pool itself"
        );
        assert_eq!(stage_file_count(root.path()), 0, "zero stage files");
    }

    /// The engine's own cancellation story: a staging cursor dropped
    /// without commit or discard leaves no residue (the Drop body's
    /// best-effort cleanup, per the wave-1 review's observation).
    #[tokio::test]
    async fn drop_cancellation_leaves_zero_residue() {
        let (root, backend) = new_backend().await;
        let key = key_for(b"cancelled-key");

        {
            let mut staging = backend.put_staged().await.unwrap();
            staging.append(b"never published").await.unwrap();
            // Dropped without commit or discard.
        }

        assert!(
            !backend.has(&key).await.unwrap(),
            "cancelled stage published nothing"
        );
        assert_eq!(backend.size(&key).await.unwrap(), None, "no entry landed");
        assert!(
            backend.list().await.unwrap().is_empty(),
            "the pool holds exactly zero entries after cancellation"
        );
        assert_eq!(
            stage_file_count(root.path()),
            0,
            "Drop cleanup left zero stage files"
        );
    }

    /// Exact-count stage hygiene (POC #3 A5's shape; store-api.md
    /// invariant 3): N failed puts — over all three forms, explicit
    /// discard, failed commit, and Drop cancellation — leave zero stage
    /// files and zero committed entries, and N successful puts leave
    /// exactly N committed entries and zero stage files.
    #[tokio::test]
    async fn exact_count_hygiene_failure_forms_leave_zero() {
        const N: usize = 5;
        let (root, backend) = new_backend().await;
        let root = root.path().to_path_buf();
        let failure_backend = LocalBackend {
            root: Arc::new(root.clone()),
            #[cfg(test)]
            fail_commit: true,
            #[cfg(test)]
            fail_append_after: u64::MAX,
            #[cfg(test)]
            fail_delete: false,
            sync_probe: SyncProbe::default(),
        };

        for i in 0..N {
            let content = format!("discard-{i}");
            let key = key_for(content.as_bytes());
            let _ = &key;
            let mut staging = backend.put_staged().await.unwrap();
            staging.append(content.as_bytes()).await.unwrap();
            staging.discard().await;
        }
        for i in 0..N {
            let content = format!("commit-failure-{i}");
            let key = key_for(content.as_bytes());
            let mut staging = failure_backend.put_staged().await.unwrap();
            staging.append(content.as_bytes()).await.unwrap();
            staging.commit(&key).await.unwrap_err();
        }
        for i in 0..N {
            let content = format!("cancelled-{i}");
            let key = key_for(content.as_bytes());
            let _ = &key;
            let mut staging = backend.put_staged().await.unwrap();
            staging.append(content.as_bytes()).await.unwrap();
            drop(staging);
        }

        assert!(
            backend.list().await.unwrap().is_empty(),
            "3N failed puts left zero committed entries"
        );
        assert_eq!(
            stage_file_count(&root),
            0,
            "3N failed puts left zero stage files"
        );
    }

    #[tokio::test]
    async fn exact_count_hygiene_success_form_leaves_exactly_n() {
        const N: usize = 5;
        let (root, backend) = new_backend().await;

        for i in 0..N {
            let content = format!("success-{i}");
            let key = key_for(content.as_bytes());
            staged(&backend, &key, content.as_bytes())
                .await
                .expect("commit publishes");
        }

        let listed = backend.list().await.unwrap();
        assert_eq!(
            listed.len(),
            N,
            "N successful puts leave exactly N committed entries"
        );
        for key in &listed {
            assert!(backend.has(key).await.unwrap(), "every entry is real");
        }
        assert_eq!(
            stage_file_count(root.path()),
            0,
            "success leaves zero stage files"
        );
    }

    /// The publish is fsync-durable on both forms (ADR-015): the file
    /// syncs before the directory-entry mutation (link), the stage file
    /// unlinks, and the parent directory syncs — in that order, and
    /// only there (appends and other staging intermediates record
    /// nothing; the ADR's scope held).
    #[tokio::test]
    async fn publish_is_fsync_durable_on_both_forms() {
        let (_root, mut backend) = new_backend().await;
        let log = backend.instrument_publish();

        let key = key_for(b"durable-publish");
        backend.put(&key, CONTENT).await.unwrap();
        assert_eq!(
            *log.lock().unwrap_or_else(|e| e.into_inner()),
            vec![
                SyncEvent::FileSync,
                SyncEvent::Link,
                SyncEvent::StageUnlinked,
                SyncEvent::DirSync
            ],
            "known-length put: file fsync before the link, dir fsync after"
        );

        log.lock().unwrap_or_else(|e| e.into_inner()).clear();
        let staged_key = key_for(b"durable-staged-publish");
        staged(&backend, &staged_key, b"staged-durable")
            .await
            .expect("staged commit publishes");
        assert_eq!(
            *log.lock().unwrap_or_else(|e| e.into_inner()),
            vec![
                SyncEvent::FileSync,
                SyncEvent::Link,
                SyncEvent::StageUnlinked,
                SyncEvent::DirSync
            ],
            "staged commit: same fsync-durable publish sequence"
        );

        log.lock().unwrap_or_else(|e| e.into_inner()).clear();
        let mut staging = backend.put_staged().await.unwrap();
        staging
            .append(b"intermediate, never fsynced")
            .await
            .unwrap();
        assert!(
            log.lock().unwrap_or_else(|e| e.into_inner()).is_empty(),
            "staging intermediates are not fsynced (the ADR's scope)"
        );
        staging.discard().await;
        assert!(
            log.lock().unwrap_or_else(|e| e.into_inner()).is_empty(),
            "discard removals are not fsynced (the ADR's scope)"
        );
    }

    /// The publish is no-replace (ADR-015): a losing race — here the
    /// deterministic same-key form the engine's CAS contract serves —
    /// goes through the losing link, its stage discards, and the first
    /// writer's committed bytes stand: no committed path is ever
    /// replaced. Dedup publishes record the losing link but no
    /// directory fsync (the parent's entries did not change).
    #[tokio::test]
    async fn publish_race_never_replaces_committed_content() {
        let (root, mut backend) = new_backend().await;
        let log = backend.instrument_publish();
        let key = key_for(b"race-key");
        let first = b"the first writer's content";

        backend.put(&key, first).await.unwrap();
        let published = stored_value(&backend, &key).await.expect("entry present");
        assert_eq!(published, first, "the winning publish's bytes landed");

        let second = b"a losing re-put's bytes";

        log.lock().unwrap_or_else(|e| e.into_inner()).clear();
        backend.put(&key, second).await.unwrap();
        let after_put = stored_value(&backend, &key).await.expect("entry present");
        assert_eq!(
            after_put, first,
            "whole-value re-put: the losing link never replaced the entry"
        );
        assert_eq!(
            *log.lock().unwrap_or_else(|e| e.into_inner()),
            vec![
                SyncEvent::FileSync,
                SyncEvent::Link,
                SyncEvent::StageUnlinked
            ],
            "dedup: the link lost, the stage copy discarded, no dir fsync"
        );

        staged(&backend, &key, b"a losing stage's bytes")
            .await
            .expect("staged re-commit is dedup, not failure");
        let after_stage = stored_value(&backend, &key).await.expect("entry present");
        assert_eq!(
            after_stage, first,
            "staged re-commit: the losing link never replaced the entry"
        );
        assert_eq!(
            stage_file_count(root.path()),
            0,
            "the loser's stage file discarded, zero residue"
        );
    }

    /// A truncated committed entry is a short-pread read error,
    /// agreeing with the non-unix `read_exact` fallback (ADR-015's
    /// unconditional platform agreement; F-3.1 — a zero-filled tail
    /// would be the silent corruption shape).
    #[tokio::test]
    async fn truncated_committed_entry_is_a_short_pread_error() {
        let (root, backend) = new_backend().await;
        let key = key_for(CONTENT);
        backend.put(&key, CONTENT).await.unwrap();

        let cursor = backend
            .get(&key)
            .await
            .expect("get succeeds")
            .expect("entry present");
        assert_eq!(cursor.len(), CONTENT.len() as u64);

        let committed = committed_path(root.path(), &key).unwrap();
        {
            let file = OpenOptions::new().write(true).open(&committed).unwrap();
            file.set_len(4).unwrap();
        }

        let err = cursor.read_at(0, CONTENT.len() as u64).await.unwrap_err();
        assert!(
            matches!(err, Error::Io(_)),
            "a short pread against the known length is an error, never a zero-filled tail"
        );
        let tail_err = cursor.read_at(0, CONTENT.len() as u64).await.unwrap_err();
        assert!(
            matches!(tail_err, Error::Io(_)),
            "still an error on re-read"
        );
    }

    /// A mid-staging append failure discards the stage and fails the
    /// put (F-3.3): the staged read window stays the clean prefix (no
    /// hole-plus-shifted content), the cursor is consumed by the
    /// discard (a later commit errors instead of publishing), and the
    /// exact-count ledger is exact — N failed stages leave zero stage
    /// files and zero committed entries.
    #[tokio::test]
    async fn staged_append_failure_discards_the_stage_exact_count() {
        const N: usize = 5;
        let (root, mut backend) = new_backend().await;
        backend.instrument_publish();
        backend.inject_fail_append(1);

        for i in 0..N {
            let mut staging = backend.put_staged().await.unwrap();
            staging
                .append(format!("prefix-{i}").as_bytes())
                .await
                .unwrap();
            staging
                .append(format!("fails-mid-stage-{i}").as_bytes())
                .await
                .unwrap_err();
            assert_eq!(
                staging.len() as usize,
                format!("prefix-{i}").len(),
                "the window stayed the clean written prefix"
            );
            let (bytes, _) = staging.read_range(0, staging.len()).await.unwrap();
            assert_eq!(
                bytes,
                format!("prefix-{i}").into_bytes(),
                "the window serves the clean prefix, never hole-plus-shifted content"
            );
            staging
                .commit(&key_for(format!("never-{i}").as_bytes()))
                .await
                .unwrap_err();
        }

        assert!(
            backend.list().await.unwrap().is_empty(),
            "N mid-write-failed stages left zero committed entries"
        );
        assert_eq!(
            stage_file_count(root.path()),
            0,
            "N mid-write failures left zero stage files"
        );

        backend.inject_fail_append(0);
        for i in 0..N {
            let mut staging = backend.put_staged().await.unwrap();
            staging
                .append(format!("fails-at-first-write-{i}").as_bytes())
                .await
                .unwrap_err();
            drop(staging);
        }
        assert!(
            backend.list().await.unwrap().is_empty(),
            "N first-write-failed stages left zero committed entries"
        );
        assert_eq!(
            stage_file_count(root.path()),
            0,
            "the first-write-failure form's ledger is exact too"
        );
    }

    #[cfg(unix)]
    fn chmod_000(path: &Path) {
        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(path, fs::Permissions::from_mode(0o000)).unwrap();
    }

    #[cfg(unix)]
    fn chmod_755(path: &Path) {
        use std::os::unix::fs::PermissionsExt;
        fs::set_permissions(path, fs::Permissions::from_mode(0o755)).unwrap();
    }

    /// The blocking helper's joint-failure form: a panicking blocking
    /// body surfaces as the named `Io` error, never a panic through
    /// the caller (the executor never dies of an engine bug).
    #[tokio::test]
    async fn blocking_joint_failure_surfaces_as_the_named_io_error() {
        let err = blocking(move || -> Result<(), Error> {
            panic!("the blocking body's injected panic");
        })
        .await
        .unwrap_err();
        match err {
            Error::Io(message) => {
                assert!(
                    message.contains("blocking task joint failure"),
                    "the joint failure names itself: {message}"
                );
            }
            other => panic!("expected the joint-failure Io form, got {other:?}"),
        }
    }

    /// `committed_path` rejects a key too short for the sharded layout
    /// outright — both forms that resolve the path (whole-value put and
    /// staged commit) carry the rejection, never a truncated shard.
    #[tokio::test]
    async fn keys_too_short_for_the_sharded_layout_reject() {
        let (_root, backend) = new_backend().await;
        let short = [0xAAu8];
        let err = backend
            .put(&short, b"the too-short key's bytes")
            .await
            .unwrap_err();
        assert!(
            matches!(err, Error::Io(ref m) if m.contains("key too short")),
            "the whole-value put rejects: {err}"
        );
        let mut staging = backend.put_staged().await.unwrap();
        staging.append(b"bytes").await.unwrap();
        let err = staging.commit(&short).await.unwrap_err();
        assert!(
            matches!(err, Error::Io(ref m) if m.contains("key too short")),
            "the staged commit rejects: {err}"
        );
        assert!(
            backend.list().await.unwrap().is_empty(),
            "the too-short keys never published"
        );
    }

    /// `walk_committed`'s fatal-malformed doctrine, every shape: a
    /// stray file or non-shard directory at shard level 1, a bad level-2
    /// shard name, a directory at blob level, a non-hex filename, and a
    /// hex filename that decodes to bytes that are not a key — each is
    /// a fatal list error naming the offending path, never a silent
    /// skip (ADR-003: a malformed list is as deadly as an incomplete
    /// one — it is silent until a sweep deletes the wrong things).
    #[tokio::test]
    async fn walk_committed_fails_loudly_on_every_malformed_shape() {
        let (root, backend) = new_backend().await;
        let good = key_for(b"the healthy entry the walk must reject around");
        let good_hex = hex::encode(&good);
        let shard1 = root.path().join(&good_hex[..2]);
        fs::create_dir_all(&shard1).unwrap();
        let shard2 = shard1.join(&good_hex[2..4]);
        fs::create_dir_all(&shard2).unwrap();
        fs::write(shard2.join(&good_hex), b"the healthy entry's bytes").unwrap();
        assert_eq!(
            backend.list().await.unwrap(),
            vec![good.clone()],
            "the crafted-but-healthy entry walks clean first"
        );

        for (malformed, expected) in [
            ("not-shard-name-file", "shard level 1"),
            ("zz-not-a-hex-shard", "shard level 1"),
        ] {
            let path = root.path().join(malformed);
            if malformed.ends_with("name-file") {
                fs::write(&path, b"").unwrap();
            } else {
                fs::create_dir_all(&path).unwrap();
            }
            let err = backend.list().await.unwrap_err();
            assert!(
                matches!(&err, Error::Io(m) if m.contains(expected)),
                "malformed {malformed}: {err}"
            );
            fs::remove_file(&path).unwrap_or_else(|_| fs::remove_dir_all(&path).unwrap());
        }

        let bad_l2 = shard1.join("zz");
        fs::create_dir_all(&bad_l2).unwrap();
        let err = backend.list().await.unwrap_err();
        assert!(
            matches!(&err, Error::Io(m) if m.contains("shard level 2")),
            "a non-shard level-2 name fails loudly: {err}"
        );
        fs::remove_dir_all(&bad_l2).unwrap();

        let dir_entry = shard2.join("deadbeef");
        fs::create_dir_all(&dir_entry).unwrap();
        let err = backend.list().await.unwrap_err();
        assert!(
            matches!(&err, Error::Io(m) if m.contains("blob level")),
            "a directory at blob level fails loudly: {err}"
        );
        fs::remove_dir_all(&dir_entry).unwrap();

        for (name, expected) in [
            ("zz-not-hex", "not hex"),
            (&hex::encode([0xFFu8])[..2], "not a key"),
        ] {
            fs::write(shard2.join(name), b"").unwrap();
            let err = backend.list().await.unwrap_err();
            assert!(
                matches!(&err, Error::Io(m) if m.contains(expected)),
                "malformed filename {name}: {err}"
            );
            fs::remove_file(shard2.join(name)).unwrap();
        }

        assert_eq!(
            backend.list().await.unwrap(),
            vec![good.clone()],
            "the doctrine is per-entry loud, and the clean tree walks again"
        );
    }

    /// The committed file's open failing in a way that is not
    /// `NotFound` (the permission-denied form real media produces) is
    /// an error, never a `None` get: the engine refuses rather than
    /// guessing the entry absent (ADR-003's rejects-never-guesses
    /// posture at the read seam).
    #[cfg(unix)]
    #[tokio::test]
    async fn a_get_open_error_other_than_not_found_surfaced() {
        let (root, backend) = new_backend().await;
        let key = key_for(CONTENT);
        backend.put(&key, CONTENT).await.unwrap();
        let committed = committed_path(root.path(), &key).unwrap();
        chmod_000(&committed);

        let err = match backend.get(&key).await {
            Ok(_) => panic!("get succeeded through an unreadable committed file"),
            Err(err) => err,
        };
        assert!(
            matches!(err, Error::Io(_)),
            "the open's non-NotFound failure surfaces: {err}"
        );
        chmod_755(&committed);
        let _ = backend.get(&key).await.unwrap().expect("readable again");
    }

    /// The publish's hard-link failing in a way that is not
    /// `AlreadyExists` (the permission-denied form real media produces)
    /// fails the put, and both failure converging forms clean the stage
    /// copy: a staged commit's own-cleanup and a whole-value put's
    /// temp-file removal leave zero residue in the staging area.
    #[cfg(unix)]
    #[tokio::test]
    async fn publish_failures_other_than_already_exists_leave_zero_residue() {
        let (root, backend) = new_backend().await;
        let key = key_for(CONTENT);
        let hex = hex::encode(&key);
        let shard2 = root.path().join(&hex[..2]).join(&hex[2..4]);
        fs::create_dir_all(&shard2).unwrap();
        chmod_000(&shard2);
        let mut staging = backend.put_staged().await.unwrap();
        staging.append(CONTENT).await.unwrap();
        let staged_err = staging.commit(&key).await.unwrap_err();
        assert!(
            matches!(staged_err, Error::Io(_)),
            "the link's other-failure form surfaces through the staged commit: {staged_err}"
        );
        let put_err = backend.put(&key, CONTENT).await.unwrap_err();
        assert!(
            matches!(put_err, Error::Io(_)),
            "the whole-value put's publish failure surfaces too: {put_err}"
        );
        let list_err = match backend.list().await {
            Ok(list) => {
                assert!(
                    list.is_empty(),
                    "nothing published through either failing form"
                );
                None
            }
            Err(err) => {
                assert!(
                    matches!(err, Error::Io(_)),
                    "the unreadable shard is a fatal walk error, never a silent skip: {err}"
                );
                Some(err)
            }
        };
        let _ = list_err;
        assert_eq!(
            stage_file_count(root.path()),
            0,
            "zero stage files: both failure forms cleaned the stage copy"
        );
        chmod_755(&shard2);
        backend.put(&key, CONTENT).await.unwrap();
        assert!(backend.has(&key).await.unwrap(), "the tree walks again");
    }

    /// A `delete` failing in a way that is not `NotFound` (the path-is-
    /// a-directory corruption form) is an `Io` error, not a `Missing`:
    /// the engine reports the failure, never guesses the entry gone.
    #[tokio::test]
    async fn a_delete_error_other_than_not_found_surfaced() {
        let (root, backend) = new_backend().await;
        let key = key_for(CONTENT);
        let committed = committed_path(root.path(), &key).unwrap();
        fs::create_dir_all(&committed).unwrap();

        let err = backend.delete(&key).await.unwrap_err();
        assert!(
            matches!(err, Error::Io(_)),
            "the removal's non-NotFound failure surfaces: {err}"
        );
        assert!(
            !matches!(err, Error::Missing),
            "the corruption form is never the Missing form"
        );
    }

    /// A `size` failing in a way that is not `NotFound` (the
    /// permission-denied traversal form) is an error, never a `None`:
    /// the stat probe refuses rather than reporting the entry absent.
    #[cfg(unix)]
    #[tokio::test]
    async fn a_size_error_other_than_not_found_surfaced() {
        let (root, backend) = new_backend().await;
        let key = key_for(CONTENT);
        backend.put(&key, CONTENT).await.unwrap();
        let shard2 = committed_path(root.path(), &key)
            .unwrap()
            .parent()
            .unwrap()
            .to_path_buf();
        chmod_000(&shard2);

        let err = backend.size(&key).await.unwrap_err();
        assert!(
            matches!(err, Error::Io(_)),
            "the metadata's non-NotFound failure surfaces: {err}"
        );
        chmod_755(&shard2);
        assert_eq!(
            backend.size(&key).await.unwrap(),
            Some(CONTENT.len() as u64),
            "the tree answers again"
        );
    }

    /// A symlink planted at a committed path is malformed state
    /// rejected loudly (review-001 S-6 — the committed-tree doctrine,
    /// one behavior with [`walk_committed`]'s fatal-malformed
    /// rejection): `get`, `size`, and `has` (which rides `size`) all
    /// reject with the doctrine-naming `Error::Io` — never serving the
    /// link's target — whether the planted path *replaced* a real
    /// committed entry or sits at a never-published key, and the
    /// walk's loud rejection names the planted entry too. A
    /// `symlink_metadata` check never follows the link, so the decoy
    /// the link points at is untouched by the rejections.
    #[cfg(unix)]
    #[tokio::test]
    async fn planted_symlink_at_committed_path_rejects_loudly() {
        let (root, backend) = new_backend().await;
        let decoy = root.path().with_extension("decoy");
        let decoy_bytes = b"the decoy's bytes".to_vec();
        fs::write(&decoy, &decoy_bytes).unwrap();

        struct Cleanup(PathBuf);
        impl Drop for Cleanup {
            fn drop(&mut self) {
                let _ = fs::remove_file(&self.0);
            }
        }
        let _decoy_cleanup = Cleanup(decoy.clone());

        let replaced_key = key_for(b"symlink-replaced-a-real-entry");
        backend.put(&replaced_key, CONTENT).await.unwrap();
        let replaced_path = committed_path(root.path(), &replaced_key).unwrap();
        fs::remove_file(&replaced_path).unwrap();
        std::os::unix::fs::symlink(&decoy, &replaced_path).unwrap();

        let virgin_key = key_for(b"symlink-at-a-never-published-key");
        let virgin_path = committed_path(root.path(), &virgin_key).unwrap();
        fs::create_dir_all(virgin_path.parent().unwrap()).unwrap();
        std::os::unix::fs::symlink(&decoy, &virgin_path).unwrap();

        for (name, key) in [("replaced", &replaced_key), ("virgin", &virgin_key)] {
            let get_err = backend.get(key).await.unwrap_err();
            assert!(
                matches!(&get_err, Error::Io(m) if m.contains("malformed committed entry") && m.contains("symlink")),
                "get rejects the {name} planted symlink loudly: {get_err}"
            );
            let size_err = backend.size(key).await.unwrap_err();
            assert!(
                matches!(&size_err, Error::Io(m) if m.contains("malformed committed entry") && m.contains("symlink")),
                "size rejects the {name} planted symlink loudly: {size_err}"
            );
            let has_err = backend.has(key).await.unwrap_err();
            assert!(
                matches!(&has_err, Error::Io(m) if m.contains("malformed committed entry") && m.contains("symlink")),
                "has (through size) rejects the {name} planted symlink loudly: {has_err}"
            );
        }

        assert_eq!(
            fs::read(&decoy).unwrap(),
            decoy_bytes,
            "the decoy the link points at was never read, removed, or disturbed"
        );

        let walk_err = backend.list().await.unwrap_err();
        assert!(
            matches!(&walk_err, Error::Io(m) if m.contains("malformed pool entry at blob level")),
            "walk_committed's loud rejection holds, unmodified, for the same planted entry: {walk_err}"
        );
    }

    /// A directory at a committed path carries the same doctrine (the
    /// portable arm of the non-regular check — no symlink API needed):
    /// `get`/`size`/`has` reject loudly, and the publish's losing-link
    /// arm refuses to dedup-map the non-regular destination — put and
    /// staged commit reject too, with zero stage residue.
    #[tokio::test]
    async fn planted_directory_at_committed_path_rejects_loudly() {
        let (root, backend) = new_backend().await;
        let key = key_for(b"directory-at-a-committed-path");
        let committed = committed_path(root.path(), &key).unwrap();
        fs::create_dir_all(&committed).unwrap();

        let get_err = backend.get(&key).await.unwrap_err();
        assert!(
            matches!(&get_err, Error::Io(m) if m.contains("malformed committed entry") && m.contains("directory")),
            "get rejects the planted directory loudly: {get_err}"
        );
        let size_err = backend.size(&key).await.unwrap_err();
        assert!(
            matches!(&size_err, Error::Io(m) if m.contains("malformed committed entry") && m.contains("directory")),
            "size rejects the planted directory loudly: {size_err}"
        );
        let has_err = backend.has(&key).await.unwrap_err();
        assert!(
            matches!(&has_err, Error::Io(m) if m.contains("malformed committed entry") && m.contains("directory")),
            "has rejects the planted directory loudly: {has_err}"
        );

        let put_err = backend.put(&key, CONTENT).await.unwrap_err();
        assert!(
            matches!(&put_err, Error::Io(m) if m.contains("malformed committed entry") && m.contains("directory")),
            "the publish rejects the non-regular destination, never dedup-maps it: {put_err}"
        );
        let staged_err = staged(&backend, &key, CONTENT).await.unwrap_err();
        assert!(
            matches!(&staged_err, Error::Io(m) if m.contains("malformed committed entry") && m.contains("directory")),
            "the staged publish rejects the non-regular destination too: {staged_err}"
        );
        assert_eq!(
            stage_file_count(root.path()),
            0,
            "both rejecting publish forms discarded their stage copies"
        );
    }

    /// A symlink planted at a committed path is rejected loudly on the
    /// publish paths (review-001 S-6's post-F-3 arm — the no-replace
    /// hardlink publish's AlreadyExists destination is re-statted and
    /// only a regular file dedup-maps): `put` and the staged `commit`
    /// reject with the doctrine-naming `Error::Io`, nothing is written
    /// through the link (the failed link never touches the destination,
    /// ADR-015) — the decoy's bytes stand and the link itself stands —
    /// and the stage copies discard with zero residue.
    #[cfg(unix)]
    #[tokio::test]
    async fn planted_symlink_rejected_on_the_publish_paths() {
        let (root, backend) = new_backend().await;
        let decoy = root.path().with_extension("decoy");
        let decoy_bytes = b"the decoy's bytes".to_vec();
        fs::write(&decoy, &decoy_bytes).unwrap();

        struct Cleanup(PathBuf);
        impl Drop for Cleanup {
            fn drop(&mut self) {
                let _ = fs::remove_file(&self.0);
            }
        }
        let _decoy_cleanup = Cleanup(decoy.clone());

        let key = key_for(b"publish-meets-a-planted-symlink");
        let committed = committed_path(root.path(), &key).unwrap();
        fs::create_dir_all(committed.parent().unwrap()).unwrap();
        std::os::unix::fs::symlink(&decoy, &committed).unwrap();

        let put_err = backend.put(&key, CONTENT).await.unwrap_err();
        assert!(
            matches!(&put_err, Error::Io(m) if m.contains("malformed committed entry") && m.contains("symlink")),
            "the whole-value put rejects the planted destination loudly: {put_err}"
        );
        let staged_err = staged(&backend, &key, CONTENT).await.unwrap_err();
        assert!(
            matches!(&staged_err, Error::Io(m) if m.contains("malformed committed entry") && m.contains("symlink")),
            "the staged commit rejects the planted destination loudly: {staged_err}"
        );
        assert_eq!(
            stage_file_count(root.path()),
            0,
            "both rejecting publish forms discarded their stage copies"
        );

        assert_eq!(
            fs::read(&decoy).unwrap(),
            decoy_bytes,
            "nothing was written through the link — the decoy's bytes stand"
        );
        assert!(
            fs::symlink_metadata(&committed).unwrap().is_symlink(),
            "the planted link itself stands (the link never replaced it)"
        );
    }
}