faucet_core/

state.rs

//! Pluggable state store for incremental replication bookmarks.
//!
//! Sources that support incremental replication need to remember where they
//! left off across runs. This module defines the [`StateStore`] trait that
//! the pipeline orchestrator uses to read and persist that progress, plus two
//! ready-to-use implementations:
//!
//! - [`MemoryStateStore`] — in-process map. Useful for tests and for runs
//!   that intentionally start fresh each time.
//! - [`FileStateStore`] — one JSON file per key, written via atomic rename
//!   so a crash mid-update never leaves the bookmark torn.
//!
//! Heavier backends (Redis, PostgreSQL) live in their own crates so
//! `faucet-core` stays dependency-light. They implement the same trait.

use crate::error::FaucetError;
use async_trait::async_trait;
use serde_json::Value;
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use tokio::io::AsyncWriteExt;
use tokio::sync::Mutex;

/// Persistent key/value store for replication bookmarks and pipeline checkpoints.
///
/// Implementations must be safe to call from multiple tasks at once. The
/// trait is intentionally minimal — three operations cover every bookmark
/// flow the pipeline orchestrator needs.
#[async_trait]
pub trait StateStore: Send + Sync {
    /// Return the value stored under `key`, or `None` if no entry exists.
    async fn get(&self, key: &str) -> Result<Option<Value>, FaucetError>;

    /// Store `value` under `key`, replacing any previous entry.
    ///
    /// Implementations should make the update durable before returning so a
    /// crash immediately after `put` does not lose the bookmark.
    async fn put(&self, key: &str, value: &Value) -> Result<(), FaucetError>;

    /// Remove the entry for `key`. A missing key is not an error.
    async fn delete(&self, key: &str) -> Result<(), FaucetError>;

    /// Run a fast, non-mutating preflight probe (used by `faucet doctor`).
    ///
    /// The default returns
    /// [`CheckReport::not_implemented`](crate::check::CheckReport::not_implemented).
    /// Built-in stores override this with a reachability + sentinel
    /// get/put/delete probe that leaves no residue.
    async fn check(
        &self,
        _ctx: &crate::check::CheckContext,
    ) -> Result<crate::check::CheckReport, FaucetError> {
        Ok(crate::check::CheckReport::not_implemented())
    }

    /// Whether [`list`](Self::list) can enumerate keys. Every built-in store
    /// can; the default is `false` so a third-party store keeps compiling.
    fn supports_list(&self) -> bool {
        false
    }

    /// Every stored key that starts with `prefix`, sorted ascending.
    ///
    /// Used to enumerate a pipeline's namespace (`{name}::…`) for
    /// `faucet state show|export`. The default returns a typed error.
    async fn list(&self, _prefix: &str) -> Result<Vec<String>, FaucetError> {
        Err(FaucetError::State(
            "this state store cannot enumerate its keys".into(),
        ))
    }

    /// Whether [`compare_and_put`](Self::compare_and_put) is atomic.
    fn supports_compare_and_put(&self) -> bool {
        false
    }

    /// Store `value` under `key` only when the entry still equals `expected`
    /// (`None` = no entry). Returns whether it was written. Two holders of one
    /// value (a rotated credential, say) use it so the second writer does not
    /// clobber the first.
    ///
    /// The default reads, compares and writes — not atomic across processes;
    /// stores that can do better override it and
    /// [`supports_compare_and_put`](Self::supports_compare_and_put).
    async fn compare_and_put(
        &self,
        key: &str,
        expected: Option<&Value>,
        value: &Value,
    ) -> Result<bool, FaucetError> {
        if self.get(key).await?.as_ref() != expected {
            return Ok(false);
        }
        self.put(key, value).await?;
        Ok(true)
    }

    /// Whether [`put_batch`](Self::put_batch) commits all-or-nothing.
    fn supports_atomic_batch(&self) -> bool {
        false
    }

    /// Store every entry. When [`supports_atomic_batch`](Self::supports_atomic_batch)
    /// is `true` either all entries land or none do; the default writes them
    /// one by one, so an error can leave a prefix of `entries` written.
    async fn put_batch(&self, entries: &[(String, Value)]) -> Result<(), FaucetError> {
        for (key, value) in entries {
            self.put(key, value).await?;
        }
        Ok(())
    }
}

/// Version of the [`StateExport`] document this build writes and reads.
pub const STATE_EXPORT_VERSION: u32 = 1;

/// The key prefix every state key of pipeline `name` starts with.
pub fn namespace_prefix(pipeline: &str) -> String {
    format!("{pipeline}::")
}

/// A portable snapshot of one pipeline's durable state — every key under its
/// namespace (`{pipeline}::…`) with its stored value, exactly as the store
/// holds it (exactly-once envelopes included).
///
/// Serialized as `{ "version": 1, "pipeline": …, "exported_at": …, "keys": {…} }`.
#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
pub struct StateExport {
    /// Document version — [`STATE_EXPORT_VERSION`] for documents this build writes.
    pub version: u32,
    /// The pipeline whose namespace the keys belong to.
    pub pipeline: String,
    /// When the snapshot was taken (RFC 3339), informational.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub exported_at: Option<String>,
    /// Every key and its stored value.
    #[serde(default)]
    pub keys: std::collections::BTreeMap<String, Value>,
}

impl StateExport {
    /// A version-[`STATE_EXPORT_VERSION`] snapshot of `pipeline`.
    pub fn new(pipeline: impl Into<String>) -> Self {
        Self {
            version: STATE_EXPORT_VERSION,
            pipeline: pipeline.into(),
            exported_at: None,
            keys: Default::default(),
        }
    }

    /// Parse and [`validate`](Self::validate) a document. A version newer than
    /// this build understands is refused rather than half-read.
    pub fn from_value(value: Value) -> Result<Self, FaucetError> {
        let version = match value.get("version") {
            None => {
                return Err(FaucetError::State(
                    "state export has no `version` field".into(),
                ));
            }
            Some(v) => v.as_u64().ok_or_else(|| {
                FaucetError::State(format!(
                    "state export `version` must be an integer, got {v}"
                ))
            })?,
        };
        if version == 0 || version > u64::from(STATE_EXPORT_VERSION) {
            return Err(FaucetError::State(format!(
                "state export version {version} is not supported by this build (supported: \
                 {STATE_EXPORT_VERSION}) — import it with the faucet release that wrote it or newer"
            )));
        }
        let export: Self = serde_json::from_value(value)
            .map_err(|e| FaucetError::State(format!("malformed state export: {e}")))?;
        export.validate()?;
        Ok(export)
    }

    /// Every key must be a valid state key inside the pipeline's namespace.
    pub fn validate(&self) -> Result<(), FaucetError> {
        validate_state_key(&self.pipeline)
            .map_err(|e| FaucetError::State(format!("state export pipeline name: {e}")))?;
        let ns = namespace_prefix(&self.pipeline);
        for key in self.keys.keys() {
            validate_state_key(key)?;
            if !key.starts_with(&ns) {
                return Err(FaucetError::State(format!(
                    "state export key '{key}' is outside the namespace '{ns}' of pipeline '{}'",
                    self.pipeline
                )));
            }
        }
        Ok(())
    }
}

/// Read every key under `pipeline`'s namespace into a [`StateExport`].
pub async fn export_namespace(
    store: &dyn StateStore,
    pipeline: &str,
) -> Result<StateExport, FaucetError> {
    let mut export = StateExport::new(pipeline);
    for key in store.list(&namespace_prefix(pipeline)).await? {
        if let Some(value) = store.get(&key).await? {
            export.keys.insert(key, value);
        }
    }
    Ok(export)
}

/// What [`import_namespace`] did.
#[derive(Debug, Clone, Default, PartialEq, serde::Serialize)]
pub struct ImportReport {
    /// Keys written, in write order.
    pub written: Vec<String>,
    /// Keys that existed in the namespace but not in the export and were
    /// removed (only with `replace`).
    pub deleted: Vec<String>,
    /// Whether the writes went through one all-or-nothing batch.
    pub atomic: bool,
    /// The first failure, if any; with a non-atomic store `written` then lists
    /// exactly what landed before it.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
}

/// Write `export` into `store`. With `replace`, keys already in the namespace
/// that the export does not hold are deleted afterwards, so the namespace ends
/// up exactly as exported. Validates the document first.
pub async fn import_namespace(
    store: &dyn StateStore,
    export: &StateExport,
    replace: bool,
) -> Result<ImportReport, FaucetError> {
    export.validate()?;
    let entries: Vec<(String, Value)> = export
        .keys
        .iter()
        .map(|(k, v)| (k.clone(), v.clone()))
        .collect();
    let stale: Vec<String> = if replace {
        store
            .list(&namespace_prefix(&export.pipeline))
            .await?
            .into_iter()
            .filter(|k| !export.keys.contains_key(k))
            .collect()
    } else {
        Vec::new()
    };
    let mut report = ImportReport {
        atomic: store.supports_atomic_batch(),
        ..Default::default()
    };
    if report.atomic {
        if let Err(e) = store.put_batch(&entries).await {
            report.error = Some(e.to_string());
            return Ok(report);
        }
        report.written = entries.into_iter().map(|(k, _)| k).collect();
    } else {
        for (key, value) in &entries {
            if let Err(e) = store.put(key, value).await {
                report.error = Some(format!("writing '{key}': {e}"));
                return Ok(report);
            }
            report.written.push(key.clone());
        }
    }
    for key in stale {
        if let Err(e) = store.delete(&key).await {
            report.error = Some(format!("deleting stale '{key}': {e}"));
            return Ok(report);
        }
        report.deleted.push(key);
    }
    Ok(report)
}

/// Prefix of the sentinel key used by state-store `check()` probes. Valid per
/// [`validate_state_key`] and deleted after the probe so it leaves no residue.
/// Probes use [`doctor_sentinel_key`], which makes each one unique.
pub const DOCTOR_SENTINEL_KEY: &str = "faucet_doctor_probe";

/// A sentinel key unique to one `check()` probe, so concurrent probes against
/// a shared store never delete each other's sentinel between its put and get
/// (CORE-57).
pub fn doctor_sentinel_key() -> String {
    format!("{DOCTOR_SENTINEL_KEY}.{}", uuid::Uuid::new_v4().simple())
}

/// Reject keys that could escape the storage namespace or break filename
/// rules on common filesystems. Allowed: ASCII letters, digits, `_`, `-`,
/// `:`, `.`. Empty keys are rejected.
pub fn validate_state_key(key: &str) -> Result<(), FaucetError> {
    if key.is_empty() {
        return Err(FaucetError::State("state key must not be empty".into()));
    }
    if key.len() > 256 {
        return Err(FaucetError::State(format!(
            "state key '{key}' exceeds 256 characters"
        )));
    }
    for (i, c) in key.char_indices() {
        let ok = c.is_ascii_alphanumeric() || matches!(c, '_' | '-' | ':' | '.' | '/');
        if !ok {
            return Err(FaucetError::State(format!(
                "state key '{key}' contains illegal character {c:?} at byte {i}"
            )));
        }
    }
    if key == "." || key == ".." || key.starts_with('.') {
        return Err(FaucetError::State(format!(
            "state key '{key}' must not begin with a dot"
        )));
    }
    // `/` separates a hub owner namespace from a name (`acme/erp::…`);
    // it must never read as a path: no empty, `.` or `..` segments, no
    // leading or trailing separator.
    if key.contains('/')
        && key
            .split('/')
            .any(|seg| seg.is_empty() || seg == "." || seg == ".." || seg.starts_with('.'))
    {
        return Err(FaucetError::State(format!(
            "state key '{key}' has a path-like segment — `/` may only separate non-empty, non-dot segments"
        )));
    }
    Ok(())
}

/// A lock file older than this belongs to a writer that died holding it.
const STALE_LOCK: std::time::Duration = std::time::Duration::from_secs(30);
/// How long a compare-and-put waits for another writer's lock.
const LOCK_WAIT: std::time::Duration = std::time::Duration::from_secs(10);

/// Removes the lock file when dropped.
struct LockFile(PathBuf);

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

async fn acquire_lock_file(path: &Path) -> Result<LockFile, FaucetError> {
    let started = tokio::time::Instant::now();
    loop {
        match tokio::fs::OpenOptions::new()
            .write(true)
            .create_new(true)
            .open(path)
            .await
        {
            Ok(_) => return Ok(LockFile(path.to_path_buf())),
            Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => {
                let stale = tokio::fs::metadata(path)
                    .await
                    .ok()
                    .and_then(|m| m.modified().ok())
                    .and_then(|t| t.elapsed().ok())
                    .is_some_and(|age| age > STALE_LOCK);
                if stale {
                    let _ = tokio::fs::remove_file(path).await;
                    continue;
                }
                if started.elapsed() > LOCK_WAIT {
                    return Err(FaucetError::State(format!(
                        "state lock {} is held by another writer",
                        path.display()
                    )));
                }
                tokio::time::sleep(std::time::Duration::from_millis(10)).await;
            }
            Err(e) => {
                return Err(FaucetError::State(format!(
                    "failed to create state lock {}: {e}",
                    path.display()
                )));
            }
        }
    }
}

// ── MemoryStateStore ────────────────────────────────────────────────────────

/// In-memory `StateStore` for tests and ephemeral pipelines.
#[derive(Default)]
pub struct MemoryStateStore {
    inner: Mutex<HashMap<String, Value>>,
}

impl MemoryStateStore {
    /// Create an empty in-memory store.
    pub fn new() -> Self {
        Self::default()
    }
}

#[async_trait]
impl StateStore for MemoryStateStore {
    async fn get(&self, key: &str) -> Result<Option<Value>, FaucetError> {
        validate_state_key(key)?;
        Ok(self.inner.lock().await.get(key).cloned())
    }

    async fn put(&self, key: &str, value: &Value) -> Result<(), FaucetError> {
        validate_state_key(key)?;
        self.inner
            .lock()
            .await
            .insert(key.to_owned(), value.clone());
        Ok(())
    }

    async fn delete(&self, key: &str) -> Result<(), FaucetError> {
        validate_state_key(key)?;
        self.inner.lock().await.remove(key);
        Ok(())
    }

    fn supports_list(&self) -> bool {
        true
    }

    async fn list(&self, prefix: &str) -> Result<Vec<String>, FaucetError> {
        let mut keys: Vec<String> = self
            .inner
            .lock()
            .await
            .keys()
            .filter(|k| k.starts_with(prefix))
            .cloned()
            .collect();
        keys.sort();
        Ok(keys)
    }

    fn supports_compare_and_put(&self) -> bool {
        true
    }

    async fn compare_and_put(
        &self,
        key: &str,
        expected: Option<&Value>,
        value: &Value,
    ) -> Result<bool, FaucetError> {
        validate_state_key(key)?;
        let mut map = self.inner.lock().await;
        if map.get(key) != expected {
            return Ok(false);
        }
        map.insert(key.to_owned(), value.clone());
        Ok(true)
    }

    fn supports_atomic_batch(&self) -> bool {
        true
    }

    async fn put_batch(&self, entries: &[(String, Value)]) -> Result<(), FaucetError> {
        for (key, _) in entries {
            validate_state_key(key)?;
        }
        let mut map = self.inner.lock().await;
        for (key, value) in entries {
            map.insert(key.clone(), value.clone());
        }
        Ok(())
    }

    async fn check(
        &self,
        _ctx: &crate::check::CheckContext,
    ) -> Result<crate::check::CheckReport, FaucetError> {
        // In-process map — always reachable.
        Ok(crate::check::CheckReport::single(
            crate::check::Probe::pass("sentinel", std::time::Duration::ZERO),
        ))
    }
}

// ── FileStateStore ──────────────────────────────────────────────────────────

/// Longest filename stem [`safe_filename`] emits before it hashes the key.
/// The temp name adds `.{32-hex}.{seq}.json.tmp` (at most 63 bytes), so a
/// stem of this length keeps every file the store writes under the 255-byte
/// name limit of common filesystems (CORE-29).
const MAX_STEM: usize = 160;

/// Map a logical state key to a filesystem-safe filename stem.
///
/// - `:` and `/` are percent-encoded as `%3A` / `%2F` (`:` is illegal in an
///   NTFS filename, `/` is a path separator).
/// - Every uppercase ASCII letter is percent-encoded (`A` → `%41`), so the
///   stem holds no uppercase letters outside escapes and two keys that differ
///   only in case can never share a file on a case-insensitive filesystem
///   (APFS, NTFS) (CORE-28).
/// - A stem longer than [`MAX_STEM`] keeps a readable prefix and ends in
///   `~<sha256 of the key>`; `~` never appears in a key or an escape, so the
///   two shapes cannot collide. The full key is kept in a `<stem>.key`
///   sidecar so [`StateStore::list`] can still report it.
///
/// Collision-free because `%` and `~` can never appear in a valid key (see
/// [`validate_state_key`]).
fn safe_filename(key: &str) -> String {
    let mut out = String::with_capacity(key.len() + 8);
    for c in key.chars() {
        match c {
            ':' => out.push_str("%3A"),
            '/' => out.push_str("%2F"),
            'A'..='Z' => out.push_str(&format!("%{:02X}", c as u32)),
            other => out.push(other),
        }
    }
    if out.len() <= MAX_STEM {
        return out;
    }
    use sha2::Digest;
    let digest = sha2::Sha256::digest(key.as_bytes());
    let hex: String = digest.iter().map(|b| format!("{b:02x}")).collect();
    let mut cut = MAX_STEM - hex.len() - 1;
    // Never split a `%XX` escape.
    while cut > 0
        && out.as_bytes()[..cut]
            .iter()
            .rev()
            .take(2)
            .any(|b| *b == b'%')
    {
        cut -= 1;
    }
    format!("{}~{hex}", &out[..cut])
}

/// Inverse of [`safe_filename`] for un-hashed stems (both the current and the
/// legacy shape decode); `None` when the stem is hashed or not a valid key.
fn key_from_filename(stem: &str) -> Option<String> {
    if stem.contains('~') {
        return None;
    }
    let bytes = stem.as_bytes();
    let mut key = String::with_capacity(stem.len());
    let mut i = 0;
    while i < bytes.len() {
        if bytes[i] == b'%' {
            let hex = stem.get(i + 1..i + 3)?;
            let code = u8::from_str_radix(hex, 16).ok()?;
            key.push(char::from(code));
            i += 3;
        } else {
            key.push(char::from(bytes[i]));
            i += 1;
        }
    }
    validate_state_key(&key).ok().map(|()| key)
}

/// File-backed `StateStore`. Each key maps to a JSON file at
/// `{root}/{safe_filename(key)}.json`, written via atomic rename. The
/// filename stem percent-encodes `:`, `/` and uppercase letters, so keys are
/// valid on Windows and keys differing only in case never share a file on a
/// case-insensitive filesystem; over-long keys are hashed.
///
/// Each `put` writes a temp file, fsyncs it, renames it over the final path,
/// and (on Unix) fsyncs the parent directory — so once `put` returns the
/// bookmark is durable across a crash or power loss, not merely sitting in the
/// page cache.
///
/// The store creates its root directory on first use. All writes are
/// serialized through a per-store mutex so two concurrent `put`s for the
/// same key cannot race on the temp filename. Different processes pointing
/// at the same directory rely on the filesystem's atomic rename guarantee.
pub struct FileStateStore {
    root: PathBuf,
    write_lock: Mutex<()>,
    migrated: tokio::sync::OnceCell<()>,
    /// Optional at-rest encryption (#207). When set, `put` seals the JSON
    /// bytes before the atomic write and `get` unseals; plaintext files
    /// remain readable (and are sealed on their next write).
    #[cfg(feature = "encryption")]
    encryption: Option<crate::encryption::CompiledEncryption>,
}

impl FileStateStore {
    /// Open or create a file-backed state store rooted at `root`.
    pub fn new(root: impl Into<PathBuf>) -> Self {
        Self {
            root: root.into(),
            write_lock: Mutex::new(()),
            migrated: tokio::sync::OnceCell::new(),
            #[cfg(feature = "encryption")]
            encryption: None,
        }
    }

    /// Seal bookmark files at rest with the given policy (#207). Reads accept
    /// both sealed and (legacy) plaintext files; every write seals.
    #[cfg(feature = "encryption")]
    pub fn with_encryption(mut self, encryption: crate::encryption::CompiledEncryption) -> Self {
        self.encryption = Some(encryption);
        self
    }

    fn entry_path(&self, key: &str) -> PathBuf {
        self.root.join(format!("{}.json", safe_filename(key)))
    }

    fn temp_path(&self, key: &str) -> PathBuf {
        // Unique per write: a per-process random token + a monotonic counter, so
        // two writers (different processes sharing the directory, or two store
        // instances in one process) never share a temp file. A *fixed* temp name
        // let a second writer `File::create`-truncate the first's half-written
        // temp, which the first could then `rename` over the final path —
        // yielding torn/truncated state JSON that breaks resume (audit #146 H10).
        //
        // The token MUST NOT be derived from the process id: containers on a
        // shared NFS/EFS/host volume routinely reuse the same PID (each starts at
        // PID 1), so a PID + per-process counter collides across processes and
        // reintroduces exactly that torn-bookmark corruption (F50). A random v4
        // UUID minted once per process is unguessable and collision-free across
        // processes regardless of PID; the counter keeps writers within one
        // process distinct. The per-store `write_lock` only serializes writers
        // within one process. Orphaned `.tmp` files from an interrupted write are
        // harmless: `get` only ever reads the final `.json` path.
        use std::sync::OnceLock;
        use std::sync::atomic::{AtomicU64, Ordering};
        static PROC_TOKEN: OnceLock<String> = OnceLock::new();
        static SEQ: AtomicU64 = AtomicU64::new(0);
        let token = PROC_TOKEN.get_or_init(|| uuid::Uuid::new_v4().simple().to_string());
        let seq = SEQ.fetch_add(1, Ordering::Relaxed);
        self.root
            .join(format!("{}.{}.{}.json.tmp", safe_filename(key), token, seq))
    }

    async fn ensure_root(&self) -> Result<(), FaucetError> {
        tokio::fs::create_dir_all(&self.root).await.map_err(|e| {
            FaucetError::State(format!(
                "failed to create state dir {}: {e}",
                self.root.display()
            ))
        })
    }

    async fn read_entry(&self, key: &str, path: &Path) -> Result<Option<Value>, FaucetError> {
        #[cfg(not(feature = "encryption"))]
        let _ = key;
        match tokio::fs::read(path).await {
            Ok(bytes) => {
                #[cfg(feature = "encryption")]
                let bytes: Vec<u8> = if crate::encryption::is_encrypted(&bytes) {
                    match &self.encryption {
                        Some(enc) => enc.decrypt_bound(&bytes, key.as_bytes()).map_err(|e| {
                            // Wrong/rotated key must be a loud, typed error —
                            // treating it as "no bookmark" would silently
                            // trigger a full re-sync.
                            FaucetError::State(format!(
                                "state file {} could not be decrypted: {e}",
                                path.display()
                            ))
                        })?,
                        None => {
                            return Err(FaucetError::State(format!(
                                "state file {} is encrypted but no `encryption` block is \
                                 configured on the file state store — add \
                                 `state.config.encryption` with the original key",
                                path.display()
                            )));
                        }
                    }
                } else {
                    bytes
                };
                #[cfg(not(feature = "encryption"))]
                let bytes = {
                    // Built without the `encryption` feature: refuse to parse a
                    // sealed file as JSON garbage.
                    if bytes.starts_with(b"FCT1") {
                        return Err(FaucetError::State(format!(
                            "state file {} is encrypted but this build of faucet has no \
                             `encryption` feature",
                            path.display()
                        )));
                    }
                    bytes
                };
                let value: Value = serde_json::from_slice(&bytes).map_err(|e| {
                    FaucetError::State(format!(
                        "failed to parse state file {}: {e}",
                        path.display()
                    ))
                })?;
                Ok(Some(value))
            }
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
            Err(e) => Err(FaucetError::State(format!(
                "failed to read state file {}: {e}",
                path.display()
            ))),
        }
    }

    /// Rename every file written under an older naming scheme (uppercase
    /// letters unescaped, long keys unhashed) to its current name, once per
    /// store. Done eagerly rather than as a read fallback because on a
    /// case-insensitive filesystem the current name of `users:x` would open the
    /// legacy file of `Users:x`.
    async fn migrate_legacy_names(&self) -> Result<(), FaucetError> {
        self.migrated
            .get_or_try_init(|| async {
                let mut dir = match tokio::fs::read_dir(&self.root).await {
                    Ok(d) => d,
                    Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(()),
                    Err(e) => {
                        return Err(FaucetError::State(format!(
                            "failed to list state dir {}: {e}",
                            self.root.display()
                        )));
                    }
                };
                let mut renames = Vec::new();
                while let Some(entry) = dir.next_entry().await.map_err(|e| {
                    FaucetError::State(format!(
                        "failed to list state dir {}: {e}",
                        self.root.display()
                    ))
                })? {
                    let name = entry.file_name();
                    let Some(stem) = name.to_str().and_then(|n| n.strip_suffix(".json")) else {
                        continue;
                    };
                    if let Some(key) = key_from_filename(stem)
                        && safe_filename(&key) != stem
                    {
                        renames.push((key, entry.path()));
                    }
                }
                for (key, from) in renames {
                    if let Some(sidecar) = self.key_sidecar(&key) {
                        tokio::fs::write(&sidecar, key.as_bytes())
                            .await
                            .map_err(|e| {
                                FaucetError::State(format!(
                                    "failed to write state key file {}: {e}",
                                    sidecar.display()
                                ))
                            })?;
                    }
                    let to = self.entry_path(&key);
                    if tokio::fs::try_exists(&to).await.unwrap_or(false) {
                        let _ = tokio::fs::remove_file(&from).await;
                        continue;
                    }
                    match tokio::fs::rename(&from, &to).await {
                        Ok(()) => {}
                        Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                        Err(e) => {
                            return Err(FaucetError::State(format!(
                                "failed to rename legacy state file {} to {}: {e}",
                                from.display(),
                                to.display()
                            )));
                        }
                    }
                }
                Ok(())
            })
            .await
            .map(|_| ())
    }

    fn key_sidecar(&self, key: &str) -> Option<PathBuf> {
        let stem = safe_filename(key);
        stem.contains('~')
            .then(|| self.root.join(format!("{stem}.key")))
    }

    /// Returns the root directory this store writes into.
    pub fn root(&self) -> &Path {
        &self.root
    }
}

#[async_trait]
impl StateStore for FileStateStore {
    async fn get(&self, key: &str) -> Result<Option<Value>, FaucetError> {
        validate_state_key(key)?;
        self.migrate_legacy_names().await?;
        self.read_entry(key, &self.entry_path(key)).await
    }

    async fn put(&self, key: &str, value: &Value) -> Result<(), FaucetError> {
        validate_state_key(key)?;
        let _guard = self.write_lock.lock().await;
        self.ensure_root().await?;
        let bytes = serde_json::to_vec(value).map_err(|e| {
            FaucetError::State(format!("failed to serialize state for key '{key}': {e}"))
        })?;
        #[cfg(feature = "encryption")]
        let bytes = match &self.encryption {
            Some(enc) => enc.encrypt_bound(&bytes, key.as_bytes()),
            None => bytes,
        };
        self.migrate_legacy_names().await?;
        let final_path = self.entry_path(key);
        let tmp_path = self.temp_path(key);

        if let Some(sidecar) = self.key_sidecar(key) {
            tokio::fs::write(&sidecar, key.as_bytes())
                .await
                .map_err(|e| {
                    FaucetError::State(format!(
                        "failed to write state key file {}: {e}",
                        sidecar.display()
                    ))
                })?;
        }

        // Write the temp file and fsync it BEFORE the rename. `fs::write`
        // alone leaves the bytes in the page cache: a crash after `put`
        // returns could surface a zero-length or stale file, breaking the
        // durability guarantee this store documents (#78/#8). `sync_all`
        // flushes the file's data and metadata to disk. Any failure removes
        // the temp file so retries do not accumulate orphans (CORE-96).
        let written: Result<(), FaucetError> = async {
            let mut file = tokio::fs::File::create(&tmp_path).await.map_err(|e| {
                FaucetError::State(format!(
                    "failed to create temp state file {}: {e}",
                    tmp_path.display()
                ))
            })?;
            file.write_all(&bytes).await.map_err(|e| {
                FaucetError::State(format!(
                    "failed to write temp state file {}: {e}",
                    tmp_path.display()
                ))
            })?;
            file.sync_all().await.map_err(|e| {
                FaucetError::State(format!(
                    "failed to fsync temp state file {}: {e}",
                    tmp_path.display()
                ))
            })?;
            drop(file);
            tokio::fs::rename(&tmp_path, &final_path)
                .await
                .map_err(|e| {
                    FaucetError::State(format!(
                        "failed to commit state file {}: {e}",
                        final_path.display()
                    ))
                })
        }
        .await;
        if let Err(e) = written {
            let _ = tokio::fs::remove_file(&tmp_path).await;
            return Err(e);
        }

        // fsync the parent directory so the rename itself is durable — an
        // atomic rename can still be lost on crash if the directory entry was
        // never flushed. Directory fsync is a POSIX concept; on platforms that
        // don't allow opening a directory as a file (e.g. Windows) it is
        // skipped.
        #[cfg(unix)]
        {
            let dir = tokio::fs::File::open(&self.root).await.map_err(|e| {
                FaucetError::State(format!(
                    "failed to open state dir {} for fsync: {e}",
                    self.root.display()
                ))
            })?;
            dir.sync_all().await.map_err(|e| {
                FaucetError::State(format!(
                    "failed to fsync state dir {}: {e}",
                    self.root.display()
                ))
            })?;
        }

        tracing::debug!(
            key,
            path = %final_path.display(),
            "state file written"
        );
        Ok(())
    }

    async fn delete(&self, key: &str) -> Result<(), FaucetError> {
        validate_state_key(key)?;
        self.migrate_legacy_names().await?;
        let mut paths = vec![self.entry_path(key)];
        paths.extend(self.key_sidecar(key));
        for path in paths {
            match tokio::fs::remove_file(&path).await {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                Err(e) => {
                    return Err(FaucetError::State(format!(
                        "failed to delete state file {}: {e}",
                        path.display()
                    )));
                }
            }
        }
        Ok(())
    }

    fn supports_compare_and_put(&self) -> bool {
        true
    }

    /// Atomic across processes sharing the directory: the read-compare-write
    /// runs while holding `<key>.json.lock`, created with `O_EXCL`. A lock
    /// left by a crashed writer is broken after 30 seconds.
    async fn compare_and_put(
        &self,
        key: &str,
        expected: Option<&Value>,
        value: &Value,
    ) -> Result<bool, FaucetError> {
        validate_state_key(key)?;
        self.ensure_root().await?;
        let lock = self.root.join(format!("{}.json.lock", safe_filename(key)));
        let _held = acquire_lock_file(&lock).await?;
        if self.get(key).await?.as_ref() != expected {
            return Ok(false);
        }
        self.put(key, value).await?;
        Ok(true)
    }

    fn supports_list(&self) -> bool {
        true
    }

    async fn list(&self, prefix: &str) -> Result<Vec<String>, FaucetError> {
        self.migrate_legacy_names().await?;
        let mut dir = match tokio::fs::read_dir(&self.root).await {
            Ok(d) => d,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
            Err(e) => {
                return Err(FaucetError::State(format!(
                    "failed to list state dir {}: {e}",
                    self.root.display()
                )));
            }
        };
        let mut keys = Vec::new();
        loop {
            let entry = dir.next_entry().await.map_err(|e| {
                FaucetError::State(format!(
                    "failed to list state dir {}: {e}",
                    self.root.display()
                ))
            })?;
            let Some(entry) = entry else { break };
            let name = entry.file_name();
            let Some(stem) = name.to_str().and_then(|n| n.strip_suffix(".json")) else {
                continue;
            };
            let key = if stem.contains('~') {
                tokio::fs::read_to_string(self.root.join(format!("{stem}.key")))
                    .await
                    .ok()
                    .filter(|k| validate_state_key(k).is_ok())
            } else {
                key_from_filename(stem)
            };
            if let Some(key) = key
                && key.starts_with(prefix)
            {
                keys.push(key);
            }
        }
        keys.sort();
        keys.dedup();
        Ok(keys)
    }

    async fn check(
        &self,
        _ctx: &crate::check::CheckContext,
    ) -> Result<crate::check::CheckReport, FaucetError> {
        use crate::check::{CheckReport, Probe};
        // Exercise the real put → get → delete cycle on a sentinel key. `put`
        // creates the root dir if needed, so this validates "dir exists +
        // writable" via the actual code path and leaves no residue.
        let start = std::time::Instant::now();
        let probe = match self.sentinel_roundtrip().await {
            Ok(()) => Probe::pass("sentinel", start.elapsed()),
            Err(e) => Probe::fail_hint(
                "sentinel",
                start.elapsed(),
                e.to_string(),
                format!("ensure {} exists and is writable", self.root.display()),
            ),
        };
        Ok(CheckReport::single(probe))
    }
}

impl FileStateStore {
    /// Write, read back, and delete a sentinel key — the body of the `check()`
    /// probe, factored out so the happy path stays linear.
    async fn sentinel_roundtrip(&self) -> Result<(), FaucetError> {
        let probe = serde_json::json!({ "faucet_doctor": true });
        let key = doctor_sentinel_key();
        self.put(&key, &probe).await?;
        let got = self.get(&key).await?;
        // Best-effort cleanup regardless of the read result.
        let _ = self.delete(&key).await;
        match got {
            Some(v) if v == probe => Ok(()),
            _ => Err(FaucetError::State(
                "sentinel readback did not match what was written".into(),
            )),
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;
    use std::sync::Arc;
    use tempfile::TempDir;

    // ── key validation ─────────────────────────────────────────────────────

    #[tokio::test]
    async fn a_lock_that_cannot_be_created_is_a_state_error() {
        let dir = TempDir::new().unwrap();
        let path = dir.path().join("missing-dir").join("k.lock");
        let err = acquire_lock_file(&path).await.err().unwrap().to_string();
        assert!(err.contains("failed to create state lock"), "{err}");
    }

    #[test]
    fn rejects_empty_key() {
        let err = validate_state_key("").unwrap_err();
        assert!(matches!(err, FaucetError::State(_)));
    }

    #[test]
    fn rejects_path_traversal_segments() {
        for k in [
            "../etc/passwd",
            "a/../b",
            "a/./b",
            "a//b",
            "/a",
            "a/",
            "a/.x",
            "a\\b",
            "..",
            ".",
        ] {
            assert!(validate_state_key(k).is_err(), "expected reject for {k:?}");
        }
        // A `/` that separates a hub owner from a name is fine (#682).
        assert!(validate_state_key("acme/erp::invoices").is_ok());
    }

    #[test]
    fn rejects_leading_dot() {
        assert!(validate_state_key(".hidden").is_err());
    }

    #[test]
    fn rejects_over_long_key() {
        let k = "a".repeat(257);
        assert!(validate_state_key(&k).is_err());
    }

    #[test]
    fn accepts_typical_keys() {
        for k in [
            "github_issues",
            "pipeline:rest:issues",
            "with.dot",
            "with-dash_and_underscore",
            "lower-Case_99",
        ] {
            validate_state_key(k).unwrap_or_else(|e| panic!("expected ok for {k:?}: {e}"));
        }
    }

    // ── MemoryStateStore ────────────────────────────────────────────────────

    #[tokio::test]
    async fn compare_and_put_writes_only_over_the_expected_value() {
        let mem = MemoryStateStore::new();
        assert!(mem.supports_compare_and_put());
        let a = serde_json::json!({"v": 1});
        let b = serde_json::json!({"v": 2});
        assert!(mem.compare_and_put("k", None, &a).await.unwrap());
        assert!(!mem.compare_and_put("k", None, &b).await.unwrap());
        assert!(!mem.compare_and_put("k", Some(&b), &b).await.unwrap());
        assert!(mem.compare_and_put("k", Some(&a), &b).await.unwrap());
        assert_eq!(mem.get("k").await.unwrap(), Some(b.clone()));
        assert!(mem.compare_and_put("bad key!", None, &a).await.is_err());

        let dir = tempfile::tempdir().unwrap();
        let file = FileStateStore::new(dir.path());
        assert!(file.supports_compare_and_put());
        assert!(file.compare_and_put("k", None, &a).await.unwrap());
        assert!(!file.compare_and_put("k", Some(&b), &b).await.unwrap());
        assert!(file.compare_and_put("k", Some(&a), &b).await.unwrap());
        assert_eq!(file.get("k").await.unwrap(), Some(b));
    }

    #[tokio::test]
    async fn file_compare_and_put_is_exclusive_and_breaks_stale_locks() {
        let dir = tempfile::tempdir().unwrap();
        let stores: Vec<std::sync::Arc<FileStateStore>> = (0..8)
            .map(|_| std::sync::Arc::new(FileStateStore::new(dir.path())))
            .collect();
        let tasks: Vec<_> = stores
            .iter()
            .enumerate()
            .map(|(i, s)| {
                let s = std::sync::Arc::clone(s);
                tokio::spawn(async move {
                    s.compare_and_put("lease", None, &json!({"by": i}))
                        .await
                        .unwrap()
                })
            })
            .collect();
        let mut wins = 0;
        for t in tasks {
            wins += usize::from(t.await.unwrap());
        }
        assert_eq!(wins, 1, "exactly one writer takes an absent key");
        let lock = dir.path().join("lease.json.lock");
        assert!(!lock.exists(), "the lock is released");

        let f = std::fs::File::create(&lock).unwrap();
        f.set_modified(std::time::SystemTime::now() - STALE_LOCK * 2)
            .unwrap();
        let held = stores[0].get("lease").await.unwrap();
        assert!(
            stores[0]
                .compare_and_put("lease", held.as_ref(), &json!({"by": "late"}))
                .await
                .unwrap()
        );
        assert_eq!(stores[0].list("").await.unwrap(), vec!["lease".to_string()]);
    }

    #[tokio::test(start_paused = true)]
    async fn file_compare_and_put_gives_up_on_a_live_lock() {
        let dir = tempfile::tempdir().unwrap();
        let store = FileStateStore::new(dir.path());
        std::fs::write(dir.path().join("k.json.lock"), b"").unwrap();
        let err = store
            .compare_and_put("k", None, &json!(1))
            .await
            .unwrap_err()
            .to_string();
        assert!(err.contains("held by another writer"), "{err}");
    }

    #[tokio::test]
    async fn memory_get_returns_none_for_missing_key() {
        let s = MemoryStateStore::new();
        assert!(s.get("nope").await.unwrap().is_none());
    }

    #[tokio::test]
    async fn memory_put_then_get_round_trips() {
        let s = MemoryStateStore::new();
        s.put("k", &json!({"cursor": "abc", "n": 7})).await.unwrap();
        let got = s.get("k").await.unwrap().unwrap();
        assert_eq!(got["cursor"], "abc");
        assert_eq!(got["n"], 7);
    }

    #[tokio::test]
    async fn memory_put_overwrites_previous_value() {
        let s = MemoryStateStore::new();
        s.put("k", &json!(1)).await.unwrap();
        s.put("k", &json!(2)).await.unwrap();
        assert_eq!(s.get("k").await.unwrap().unwrap(), json!(2));
    }

    #[tokio::test]
    async fn memory_delete_makes_get_return_none() {
        let s = MemoryStateStore::new();
        s.put("k", &json!("v")).await.unwrap();
        s.delete("k").await.unwrap();
        assert!(s.get("k").await.unwrap().is_none());
    }

    #[tokio::test]
    async fn memory_delete_missing_key_is_ok() {
        let s = MemoryStateStore::new();
        s.delete("absent").await.unwrap();
    }

    #[tokio::test]
    async fn memory_rejects_invalid_keys() {
        let s = MemoryStateStore::new();
        assert!(s.get("a b").await.is_err());
        assert!(s.put("a b", &json!(1)).await.is_err());
        assert!(s.delete("a b").await.is_err());
    }

    // ── FileStateStore ──────────────────────────────────────────────────────

    #[tokio::test]
    async fn file_get_returns_none_for_missing_key() {
        let dir = TempDir::new().unwrap();
        let s = FileStateStore::new(dir.path());
        assert!(s.get("nope").await.unwrap().is_none());
    }

    #[tokio::test]
    async fn file_put_creates_root_directory_lazily() {
        let dir = TempDir::new().unwrap();
        let root = dir.path().join("nested/state");
        let s = FileStateStore::new(&root);
        s.put("k", &json!("v")).await.unwrap();
        assert!(root.is_dir(), "root dir should be created on first put");
    }

    #[tokio::test]
    async fn file_put_then_get_round_trips() {
        let dir = TempDir::new().unwrap();
        let s = FileStateStore::new(dir.path());
        let value = json!({"cursor": "abc", "n": 42, "nested": {"flag": true}});
        s.put("github_issues", &value).await.unwrap();
        let got = s.get("github_issues").await.unwrap().unwrap();
        assert_eq!(got, value);
    }

    #[test]
    fn temp_path_is_unique_and_not_pid_derived() {
        // Regression for F50: the temp filename must not embed the process id.
        // Containers on a shared volume reuse PIDs (each starts at PID 1), so a
        // PID + per-process counter collides across processes and reintroduces
        // the torn-bookmark corruption the unique-temp scheme prevents. The
        // token is a per-process random UUID instead.
        let dir = TempDir::new().unwrap();
        let s = FileStateStore::new(dir.path());

        let a = s.temp_path("k");
        let b = s.temp_path("k");
        // Distinct temp paths within one process (the monotonic counter differs).
        assert_ne!(a, b);

        let name_a = a.file_name().unwrap().to_str().unwrap();
        let pid = std::process::id().to_string();
        // The PID must NOT appear as a dotted segment of the temp name.
        assert!(
            !name_a.split('.').any(|seg| seg == pid),
            "temp filename {name_a} must not embed the process id ({pid})"
        );
        assert!(name_a.ends_with(".json.tmp"));
    }

    #[test]
    fn safe_filename_percent_encodes_colon() {
        assert_eq!(
            safe_filename("pipeline:rest:issues"),
            "pipeline%3Arest%3Aissues"
        );
        assert_eq!(safe_filename("plain_key-1.v2"), "plain_key-1.v2");
        // Owner-scoped hub ids (`acme/erp::invoices`) must not become
        // nested directories.
        assert_eq!(
            safe_filename("acme/erp::invoices"),
            "acme%2Ferp%3A%3Ainvoices"
        );
        assert!(validate_state_key("acme/erp::invoices").is_ok());
        assert!(validate_state_key("a b").is_err());
    }

    #[tokio::test]
    async fn file_round_trips_colon_keys_with_safe_filename() {
        // Regression for #78 LOW: the documented `pipeline:rest:issues` key
        // convention must round-trip and produce a Windows-legal filename
        // (no `:` on disk).
        let dir = TempDir::new().unwrap();
        let s = FileStateStore::new(dir.path());
        let value = json!({"cursor": "z"});
        s.put("pipeline:rest:issues", &value).await.unwrap();
        assert_eq!(s.get("pipeline:rest:issues").await.unwrap().unwrap(), value);
        // On-disk filename must not contain a colon.
        assert!(dir.path().join("pipeline%3Arest%3Aissues.json").exists());
        let mut has_colon = false;
        for entry in std::fs::read_dir(dir.path()).unwrap() {
            if entry.unwrap().file_name().to_string_lossy().contains(':') {
                has_colon = true;
            }
        }
        assert!(!has_colon, "no state filename may contain ':'");
    }

    #[test]
    fn safe_filename_escapes_uppercase_and_round_trips() {
        assert_eq!(safe_filename("Users::x"), "%55sers%3A%3Ax");
        assert_ne!(
            safe_filename("Users::x").to_lowercase(),
            safe_filename("users::x").to_lowercase()
        );
        for k in ["Users::x", "a/B:c", "plain"] {
            assert_eq!(key_from_filename(&safe_filename(k)).as_deref(), Some(k));
        }
        // Legacy (unescaped) stems still decode.
        assert_eq!(key_from_filename("Users%3Ax").as_deref(), Some("Users:x"));
        assert_eq!(key_from_filename("bad%Z"), None);
        assert_eq!(key_from_filename("a~b"), None);
    }

    #[test]
    fn long_keys_hash_into_a_bounded_stem() {
        let key = format!("p::{}", "A:".repeat(120));
        let stem = safe_filename(&key);
        assert!(stem.len() <= MAX_STEM, "{}", stem.len());
        assert!(stem.contains('~'));
        let other = format!("p::{}B", "A:".repeat(119));
        assert_ne!(safe_filename(&other), stem);
    }

    #[tokio::test]
    async fn keys_differing_only_in_case_keep_separate_files() {
        let dir = TempDir::new().unwrap();
        let s = FileStateStore::new(dir.path());
        s.put("Users::x", &json!(1)).await.unwrap();
        s.put("users::x", &json!(2)).await.unwrap();
        assert_eq!(s.get("Users::x").await.unwrap(), Some(json!(1)));
        assert_eq!(s.get("users::x").await.unwrap(), Some(json!(2)));
        assert_eq!(s.list("").await.unwrap(), vec!["Users::x", "users::x"]);
    }

    #[tokio::test]
    async fn a_long_key_round_trips_lists_and_deletes() {
        let dir = TempDir::new().unwrap();
        let s = FileStateStore::new(dir.path());
        let key = format!("pipe::row::{}", "parent:".repeat(34));
        assert!(key.len() <= 256);
        s.put(&key, &json!({"c": 1})).await.unwrap();
        assert_eq!(s.get(&key).await.unwrap(), Some(json!({"c": 1})));
        assert_eq!(s.list("pipe").await.unwrap(), vec![key.clone()]);
        s.delete(&key).await.unwrap();
        assert_eq!(s.get(&key).await.unwrap(), None);
        assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 0);
    }

    #[tokio::test]
    async fn legacy_named_files_are_read_and_migrated() {
        let dir = TempDir::new().unwrap();
        std::fs::write(dir.path().join("Users%3Ax.json"), b"{\"c\":7}").unwrap();
        let s = FileStateStore::new(dir.path());
        // A lowercase key must not read the uppercase key's legacy file.
        assert_eq!(s.get("users:x").await.unwrap(), None);
        assert_eq!(s.get("Users:x").await.unwrap(), Some(json!({"c": 7})));
        s.put("Users:x", &json!({"c": 8})).await.unwrap();
        let names: Vec<String> = std::fs::read_dir(dir.path())
            .unwrap()
            .map(|e| e.unwrap().file_name().to_string_lossy().into_owned())
            .collect();
        assert_eq!(names, vec!["%55sers%3Ax.json".to_string()]);
        assert_eq!(s.get("Users:x").await.unwrap(), Some(json!({"c": 8})));
        s.delete("Users:x").await.unwrap();
        assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 0);

        // A long legacy name is migrated to its hashed name.
        let dir = TempDir::new().unwrap();
        let key = format!("p::{}", "q".repeat(170));
        std::fs::write(
            dir.path().join(format!("p%3A%3A{}.json", "q".repeat(170))),
            b"3",
        )
        .unwrap();
        let s = FileStateStore::new(dir.path());
        assert_eq!(s.list("p").await.unwrap(), vec![key.clone()]);
        assert_eq!(s.get(&key).await.unwrap(), Some(json!(3)));
    }

    #[tokio::test]
    async fn a_failed_put_leaves_no_temp_file() {
        let dir = TempDir::new().unwrap();
        let s = FileStateStore::new(dir.path());
        // A directory where the final file should be makes the rename fail.
        std::fs::create_dir(dir.path().join("k.json")).unwrap();
        std::fs::write(dir.path().join("k.json").join("x"), b"x").unwrap();
        assert!(s.put("k", &json!(1)).await.is_err());
        assert!(!has_tmp_residue(dir.path()));
    }

    #[test]
    fn doctor_sentinel_keys_are_unique_and_valid() {
        let a = doctor_sentinel_key();
        assert_ne!(a, doctor_sentinel_key());
        assert!(a.starts_with(DOCTOR_SENTINEL_KEY));
        validate_state_key(&a).unwrap();
    }

    /// True if any `.json.tmp` residue remains in `dir`. Temp names are unique
    /// per write since #146 H10, so we glob for the suffix rather than check a
    /// fixed name (which would never exist now and pass vacuously).
    fn has_tmp_residue(dir: &std::path::Path) -> bool {
        std::fs::read_dir(dir)
            .unwrap()
            .filter_map(|e| e.ok())
            .any(|e| e.file_name().to_string_lossy().ends_with(".json.tmp"))
    }

    #[tokio::test]
    async fn file_put_overwrites_previous_value_atomically() {
        let dir = TempDir::new().unwrap();
        let s = FileStateStore::new(dir.path());
        s.put("k", &json!({"v": 1})).await.unwrap();
        s.put("k", &json!({"v": 2})).await.unwrap();
        assert_eq!(s.get("k").await.unwrap().unwrap(), json!({"v": 2}));
        // No temp file left behind.
        assert!(!has_tmp_residue(dir.path()), "no temp residue after put");
    }

    #[test]
    fn file_temp_paths_are_unique_per_write() {
        // H10 (audit #146): the temp path must be unique per write so two
        // writers (different processes, or two store instances in one process
        // with independent write_locks) never `File::create`-truncate a shared
        // temp that the other then renames over the final file (torn state).
        let dir = TempDir::new().unwrap();
        let s = FileStateStore::new(dir.path());
        let a = s.temp_path("k");
        let b = s.temp_path("k");
        assert_ne!(a, b, "each write must get a distinct temp path");
        // The committed (final) path stays stable across writes.
        assert_eq!(s.entry_path("k"), s.entry_path("k"));
    }

    #[tokio::test]
    async fn file_put_writes_complete_durable_file_with_no_temp_residue() {
        // Regression for #78/#8. `put` must produce a fully-written, parseable
        // file and leave no temp file behind. (The fsync that makes this
        // durable across a power loss can't be observed on a healthy
        // filesystem, but a regression in the write/rename path — truncation,
        // a leftover .tmp, or an unwritten file — is caught here.) A large
        // payload makes a partial/unflushed write detectable on read-back.
        let dir = TempDir::new().unwrap();
        let s = FileStateStore::new(dir.path());
        let big: Vec<Value> = (0..1_000)
            .map(|i| json!({"i": i, "s": "x".repeat(20)}))
            .collect();
        let value = json!({"cursor": "abc", "rows": big});

        s.put("github_issues", &value).await.unwrap();

        // Read the raw file directly (bypassing get) to confirm it is complete.
        let raw = tokio::fs::read(dir.path().join("github_issues.json"))
            .await
            .expect("state file must exist after put");
        assert!(!raw.is_empty(), "state file must not be zero-length");
        let parsed: Value = serde_json::from_slice(&raw).expect("state file must be valid JSON");
        assert_eq!(parsed, value);

        // No temp file left behind.
        assert!(!has_tmp_residue(dir.path()), "no temp residue after put");
    }

    #[tokio::test]
    async fn file_delete_removes_file() {
        let dir = TempDir::new().unwrap();
        let s = FileStateStore::new(dir.path());
        s.put("k", &json!("v")).await.unwrap();
        s.delete("k").await.unwrap();
        assert!(s.get("k").await.unwrap().is_none());
        assert!(!dir.path().join("k.json").exists());
    }

    #[tokio::test]
    async fn file_delete_missing_key_is_ok() {
        let dir = TempDir::new().unwrap();
        let s = FileStateStore::new(dir.path());
        s.delete("absent").await.unwrap();
    }

    #[tokio::test]
    async fn file_get_returns_error_for_corrupt_json() {
        let dir = TempDir::new().unwrap();
        let s = FileStateStore::new(dir.path());
        tokio::fs::create_dir_all(dir.path()).await.unwrap();
        tokio::fs::write(dir.path().join("bad.json"), b"not json")
            .await
            .unwrap();
        let err = s.get("bad").await.unwrap_err();
        match err {
            FaucetError::State(msg) => assert!(msg.contains("bad.json")),
            other => panic!("expected State error, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn file_concurrent_puts_do_not_corrupt_or_leak_temp() {
        let dir = TempDir::new().unwrap();
        let s = Arc::new(FileStateStore::new(dir.path()));
        let mut handles = vec![];
        for i in 0..50 {
            let s = Arc::clone(&s);
            handles.push(tokio::spawn(async move {
                s.put("k", &json!({"i": i})).await.unwrap();
            }));
        }
        for h in handles {
            h.await.unwrap();
        }
        // The final value must be one of the 50 we wrote.
        let got = s.get("k").await.unwrap().unwrap();
        let i = got["i"].as_i64().unwrap();
        assert!((0..50).contains(&i));
        // And no temp file should remain.
        assert!(
            !has_tmp_residue(dir.path()),
            "no temp residue after concurrent puts"
        );
    }

    #[tokio::test]
    async fn file_store_works_through_trait_object() {
        let dir = TempDir::new().unwrap();
        let s: Box<dyn StateStore> = Box::new(FileStateStore::new(dir.path()));
        s.put("k", &json!(1)).await.unwrap();
        assert_eq!(s.get("k").await.unwrap().unwrap(), json!(1));
    }

    // ── check() ──────────────────────────────────────────────────────────────

    #[tokio::test]
    async fn memory_check_passes() {
        let s = MemoryStateStore::new();
        let report = s
            .check(&crate::check::CheckContext::default())
            .await
            .unwrap();
        assert_eq!(report.failed_count(), 0);
        assert!(
            report
                .probes
                .iter()
                .all(|p| matches!(p.status, crate::check::ProbeStatus::Pass))
        );
    }

    #[tokio::test]
    async fn file_check_passes_for_writable_root() {
        let dir = TempDir::new().unwrap();
        let s = FileStateStore::new(dir.path());
        let report = s
            .check(&crate::check::CheckContext::default())
            .await
            .unwrap();
        assert_eq!(report.failed_count(), 0, "writable root should pass");
        // The sentinel probe must leave no residue.
        let leftovers: Vec<_> = std::fs::read_dir(dir.path()).unwrap().collect();
        assert!(leftovers.is_empty(), "check() must not leave files behind");
    }

    #[tokio::test]
    async fn file_store_root_returns_configured_directory() {
        let dir = TempDir::new().unwrap();
        let s = FileStateStore::new(dir.path());
        assert_eq!(s.root(), dir.path());
    }

    #[tokio::test]
    async fn default_check_reports_not_implemented() {
        // A StateStore that does not override `check()` falls back to the
        // trait default, which yields a single not-implemented (skipped) probe.
        struct BareStore;
        #[async_trait]
        impl StateStore for BareStore {
            async fn get(&self, _key: &str) -> Result<Option<Value>, FaucetError> {
                Ok(None)
            }
            async fn put(&self, _key: &str, _value: &Value) -> Result<(), FaucetError> {
                Ok(())
            }
            async fn delete(&self, _key: &str) -> Result<(), FaucetError> {
                Ok(())
            }
        }
        let s = BareStore;
        let report = s
            .check(&crate::check::CheckContext::default())
            .await
            .unwrap();
        // not_implemented() reports no failures and is not a pass.
        assert_eq!(report.failed_count(), 0);
        assert!(
            report
                .probes
                .iter()
                .any(|p| matches!(p.status, crate::check::ProbeStatus::Skip { .. })),
            "default check must surface a skipped (not-implemented) probe"
        );
    }

    #[tokio::test]
    async fn file_check_fails_when_root_unusable() {
        // Root whose parent is a regular file → create_dir_all fails.
        let dir = TempDir::new().unwrap();
        let file = dir.path().join("not_a_dir");
        std::fs::write(&file, b"x").unwrap();
        let s = FileStateStore::new(file.join("state"));
        let report = s
            .check(&crate::check::CheckContext::default())
            .await
            .unwrap();
        assert_eq!(report.failed_count(), 1, "unusable root should fail");
    }

    #[cfg(feature = "encryption")]
    mod encryption_at_rest {
        use super::*;
        use crate::encryption::{CompiledEncryption, EncryptionSpec, is_encrypted};
        use serde_json::json;

        fn enc(key: &str) -> CompiledEncryption {
            CompiledEncryption::compile(&EncryptionSpec {
                key: key.into(),
                previous_keys: vec![],
                algorithm: Default::default(),
            })
            .unwrap()
        }

        #[tokio::test]
        async fn encrypted_round_trip_and_ciphertext_on_disk() {
            let dir = tempfile::tempdir().unwrap();
            let store = FileStateStore::new(dir.path()).with_encryption(enc("k1"));
            store.put("bk", &json!({"lsn": 42})).await.unwrap();
            assert_eq!(store.get("bk").await.unwrap(), Some(json!({"lsn": 42})));

            // On disk it is ciphertext, not JSON.
            let raw = std::fs::read(dir.path().join("bk.json")).unwrap();
            assert!(is_encrypted(&raw));
            assert!(serde_json::from_slice::<Value>(&raw).is_err());

            store.delete("bk").await.unwrap();
            assert_eq!(store.get("bk").await.unwrap(), None);
        }

        #[tokio::test]
        async fn plaintext_file_stays_readable_and_is_sealed_on_next_write() {
            let dir = tempfile::tempdir().unwrap();
            // A pre-encryption bookmark written by an older config.
            let plain = FileStateStore::new(dir.path());
            plain.put("bk", &json!("legacy")).await.unwrap();
            let before = std::fs::read(dir.path().join("bk.json")).unwrap();
            assert!(!is_encrypted(&before));

            let sealed = FileStateStore::new(dir.path()).with_encryption(enc("k1"));
            assert_eq!(sealed.get("bk").await.unwrap(), Some(json!("legacy")));
            sealed.put("bk", &json!("updated")).await.unwrap();
            let after = std::fs::read(dir.path().join("bk.json")).unwrap();
            assert!(is_encrypted(&after), "next write must seal the file");
            assert_eq!(sealed.get("bk").await.unwrap(), Some(json!("updated")));
        }

        #[tokio::test]
        async fn wrong_key_is_a_typed_error_not_a_missing_bookmark() {
            let dir = tempfile::tempdir().unwrap();
            let a = FileStateStore::new(dir.path()).with_encryption(enc("right"));
            a.put("bk", &json!(1)).await.unwrap();

            let b = FileStateStore::new(dir.path()).with_encryption(enc("wrong"));
            let err = b.get("bk").await.unwrap_err();
            assert!(matches!(err, FaucetError::State(_)));
            assert!(err.to_string().contains("could not be decrypted"), "{err}");
        }

        #[tokio::test]
        async fn encrypted_file_with_unconfigured_store_is_a_typed_error() {
            let dir = tempfile::tempdir().unwrap();
            let sealed = FileStateStore::new(dir.path()).with_encryption(enc("k1"));
            sealed.put("bk", &json!(1)).await.unwrap();

            let plain = FileStateStore::new(dir.path());
            let err = plain.get("bk").await.unwrap_err();
            assert!(err.to_string().contains("no `encryption` block"), "{err}");
        }

        #[tokio::test]
        async fn rotation_reads_old_key_files() {
            let dir = tempfile::tempdir().unwrap();
            let old = FileStateStore::new(dir.path()).with_encryption(enc("old"));
            old.put("bk", &json!("v1")).await.unwrap();

            let rotated = FileStateStore::new(dir.path()).with_encryption(
                CompiledEncryption::compile(&EncryptionSpec {
                    key: "new".into(),
                    previous_keys: vec!["old".into()],
                    algorithm: Default::default(),
                })
                .unwrap(),
            );
            assert_eq!(rotated.get("bk").await.unwrap(), Some(json!("v1")));
            // Re-seal under the new key; a store knowing only "new" can read it.
            rotated.put("bk", &json!("v2")).await.unwrap();
            let new_only = FileStateStore::new(dir.path()).with_encryption(enc("new"));
            assert_eq!(new_only.get("bk").await.unwrap(), Some(json!("v2")));
        }

        #[tokio::test]
        async fn no_temp_files_left_behind() {
            let dir = tempfile::tempdir().unwrap();
            let store = FileStateStore::new(dir.path()).with_encryption(enc("k1"));
            store.put("bk", &json!(1)).await.unwrap();
            let leftovers: Vec<_> = std::fs::read_dir(dir.path())
                .unwrap()
                .filter_map(Result::ok)
                .filter(|e| e.path().to_string_lossy().ends_with(".tmp"))
                .collect();
            assert!(
                leftovers.is_empty(),
                "atomic write must leave no temp files"
            );
        }
    }
}

#[cfg(test)]
mod list_export_tests {
    use super::*;
    use serde_json::json;

    struct Bare(MemoryStateStore);

    #[async_trait]
    impl StateStore for Bare {
        async fn get(&self, key: &str) -> Result<Option<Value>, FaucetError> {
            self.0.get(key).await
        }
        async fn put(&self, key: &str, value: &Value) -> Result<(), FaucetError> {
            if key.ends_with("::boom") {
                return Err(FaucetError::State("disk full".into()));
            }
            self.0.put(key, value).await
        }
        async fn delete(&self, key: &str) -> Result<(), FaucetError> {
            if key.ends_with("::stuck") {
                return Err(FaucetError::State("locked".into()));
            }
            self.0.delete(key).await
        }
    }

    #[tokio::test]
    async fn defaults_cannot_list_and_batch_sequentially() {
        let s = Bare(MemoryStateStore::new());
        assert!(!s.supports_list());
        assert!(!s.supports_atomic_batch());
        assert!(s.list("p::").await.is_err());
        s.put_batch(&[("p::a".into(), json!(1)), ("p::b".into(), json!(2))])
            .await
            .unwrap();
        assert_eq!(s.get("p::b").await.unwrap(), Some(json!(2)));
        let err = s
            .put_batch(&[("p::c".into(), json!(1)), ("p::boom".into(), json!(2))])
            .await
            .unwrap_err();
        assert!(err.to_string().contains("disk full"));
        assert_eq!(
            s.get("p::c").await.unwrap(),
            Some(json!(1)),
            "prefix landed"
        );
    }

    #[tokio::test]
    async fn memory_lists_by_prefix_sorted_and_batches_atomically() {
        let s = MemoryStateStore::new();
        for k in ["p::b", "p::a", "q::a", "p::a::__sla__"] {
            s.put(k, &json!(k)).await.unwrap();
        }
        assert!(s.supports_list());
        assert_eq!(
            s.list("p::").await.unwrap(),
            vec!["p::a", "p::a::__sla__", "p::b"]
        );
        assert!(s.supports_atomic_batch());
        let err = s
            .put_batch(&[("p::x".into(), json!(1)), ("../bad".into(), json!(2))])
            .await
            .unwrap_err();
        assert!(matches!(err, FaucetError::State(_)));
        assert!(s.get("p::x").await.unwrap().is_none(), "nothing written");
        s.put_batch(&[("p::x".into(), json!(1))]).await.unwrap();
        assert_eq!(s.get("p::x").await.unwrap(), Some(json!(1)));
    }

    #[tokio::test]
    async fn file_lists_decoded_keys_ignoring_temp_and_foreign_files() {
        let dir = tempfile::tempdir().unwrap();
        let s = FileStateStore::new(dir.path());
        assert!(s.list("p::").await.unwrap().is_empty(), "missing root");
        for k in ["p::a", "p::a::__sla__", "acme/p::x", "q::a"] {
            s.put(k, &json!(1)).await.unwrap();
        }
        std::fs::write(dir.path().join("p%3A%3Az.json.tmp"), b"{}").unwrap();
        std::fs::write(dir.path().join("notes.txt"), b"x").unwrap();
        std::fs::write(dir.path().join(".hidden.json"), b"{}").unwrap();
        assert!(s.supports_list());
        assert_eq!(s.list("p::").await.unwrap(), vec!["p::a", "p::a::__sla__"]);
        assert_eq!(s.list("acme/").await.unwrap(), vec!["acme/p::x"]);
        assert!(!s.supports_atomic_batch());
    }

    #[cfg(unix)]
    #[tokio::test]
    async fn file_list_reports_unreadable_root() {
        let dir = tempfile::tempdir().unwrap();
        let file = dir.path().join("not-a-dir");
        std::fs::write(&file, b"x").unwrap();
        let s = FileStateStore::new(&file);
        let err = s.list("p::").await.unwrap_err();
        assert!(
            err.to_string().contains("failed to list state dir"),
            "{err}"
        );
    }

    #[test]
    fn export_document_parses_and_validates() {
        let ok = json!({
            "version": 1,
            "pipeline": "orders",
            "exported_at": "2026-09-26T00:00:00Z",
            "keys": {"orders::a": {"id": 3}, "orders::a::__sla__": {}}
        });
        let e = StateExport::from_value(ok).unwrap();
        assert_eq!(e.keys.len(), 2);
        assert_eq!(namespace_prefix("orders"), "orders::");

        let cases = [
            (json!({"pipeline": "o", "keys": {}}), "no `version`"),
            (
                json!({"version": "1", "pipeline": "o"}),
                "must be an integer",
            ),
            (json!({"version": 2, "pipeline": "o"}), "not supported"),
            (json!({"version": 0, "pipeline": "o"}), "not supported"),
            (json!({"version": 1}), "malformed"),
            (json!({"version": 1, "pipeline": "../o"}), "pipeline name"),
            (
                json!({"version": 1, "pipeline": "o", "keys": {"other::a": 1}}),
                "outside the namespace",
            ),
            (
                json!({"version": 1, "pipeline": "o", "keys": {"o::a b": 1}}),
                "illegal character",
            ),
        ];
        for (doc, needle) in cases {
            let err = StateExport::from_value(doc.clone())
                .unwrap_err()
                .to_string();
            assert!(err.contains(needle), "{doc} → {err}");
        }
    }

    #[tokio::test]
    async fn export_then_import_round_trips_and_replace_drops_stale_keys() {
        let from = MemoryStateStore::new();
        from.put("o::a", &json!({"c": 1})).await.unwrap();
        from.put("o::b", &json!(2)).await.unwrap();
        from.put("other::a", &json!(9)).await.unwrap();
        let export = export_namespace(&from, "o").await.unwrap();
        assert_eq!(export.keys.len(), 2);

        let to = MemoryStateStore::new();
        to.put("o::stale", &json!(0)).await.unwrap();
        let r = import_namespace(&to, &export, false).await.unwrap();
        assert!(r.atomic && r.error.is_none());
        assert_eq!(r.written, vec!["o::a", "o::b"]);
        assert!(
            to.get("o::stale").await.unwrap().is_some(),
            "kept without replace"
        );

        let r = import_namespace(&to, &export, true).await.unwrap();
        assert_eq!(r.deleted, vec!["o::stale"]);
        assert!(to.get("o::stale").await.unwrap().is_none());
        assert_eq!(to.get("o::a").await.unwrap(), Some(json!({"c": 1})));
    }

    #[tokio::test]
    async fn import_reports_partial_progress_on_a_non_atomic_store() {
        let mut export = StateExport::new("o");
        export.keys.insert("o::a".into(), json!(1));
        export.keys.insert("o::boom".into(), json!(2));
        let s = Bare(MemoryStateStore::new());
        let r = import_namespace(&s, &export, false).await.unwrap();
        assert!(!r.atomic);
        assert_eq!(r.written, vec!["o::a"]);
        assert!(r.error.unwrap().contains("o::boom"));

        let mut bad = StateExport::new("o");
        bad.keys.insert("x::a".into(), json!(1));
        assert!(import_namespace(&s, &bad, false).await.is_err());
    }

    #[tokio::test]
    async fn import_reports_batch_and_delete_failures() {
        struct Failing;
        #[async_trait]
        impl StateStore for Failing {
            async fn get(&self, _: &str) -> Result<Option<Value>, FaucetError> {
                Ok(None)
            }
            async fn put(&self, _: &str, _: &Value) -> Result<(), FaucetError> {
                Ok(())
            }
            async fn delete(&self, _: &str) -> Result<(), FaucetError> {
                Ok(())
            }
            fn supports_atomic_batch(&self) -> bool {
                true
            }
            async fn put_batch(&self, _: &[(String, Value)]) -> Result<(), FaucetError> {
                Err(FaucetError::State("tx aborted".into()))
            }
        }
        assert_eq!(Failing.get("o::a").await.unwrap(), None);
        Failing.put("o::a", &json!(1)).await.unwrap();
        Failing.delete("o::a").await.unwrap();
        let mut export = StateExport::new("o");
        export.keys.insert("o::a".into(), json!(1));
        let r = import_namespace(&Failing, &export, false).await.unwrap();
        assert!(r.written.is_empty());
        assert!(r.error.unwrap().contains("tx aborted"));

        // A replace needs to list the namespace first.
        assert!(import_namespace(&Failing, &export, true).await.is_err());

        let s = Bare(MemoryStateStore::new());
        struct Listing(Bare);
        #[async_trait]
        impl StateStore for Listing {
            async fn get(&self, k: &str) -> Result<Option<Value>, FaucetError> {
                self.0.get(k).await
            }
            async fn put(&self, k: &str, v: &Value) -> Result<(), FaucetError> {
                self.0.put(k, v).await
            }
            async fn delete(&self, k: &str) -> Result<(), FaucetError> {
                self.0.delete(k).await
            }
            async fn list(&self, p: &str) -> Result<Vec<String>, FaucetError> {
                self.0.0.list(p).await
            }
        }
        s.put("o::stuck", &json!(1)).await.unwrap();
        let listing = Listing(s);
        assert_eq!(listing.get("o::stuck").await.unwrap(), Some(json!(1)));
        let r = import_namespace(&listing, &export, true).await.unwrap();
        assert_eq!(r.written, vec!["o::a"]);
        assert!(r.error.unwrap().contains("deleting stale 'o::stuck'"));
    }
}