agentplane 0.46.0

//! The embedded journal, on redb.
//!
//! Two of the runtime's three core guarantees are enforced *here*, by the shape
//! of the data rather than by application logic — because application logic can
//! be bypassed by the next caller:
//!
//! * **Exactly-once** — [`EFFECT_ONCE`] is keyed by `(run_id, effect_key)`, so a
//!   second start for one effect is not rejected, it is *inexpressible*. `insert`
//!   returns the prior value, and a prior value means the effect already began.
//!   This is stronger than the partial unique index it replaces: an index is a
//!   declaration a later migration can drop, a key is the table's identity.
//! * **Fencing** — the lease epoch is read and the records written in **one**
//!   write transaction. redb gives multi-table atomicity, so there is no window
//!   between the check and the append for a stale writer to slip into.
//!
//! redb is pure Rust and two crates deep. That is not tidiness: the previous
//! backend linked a cloud sync engine and a vector-search index that nothing
//! here uses, and every downstream consumer paid for them.
//!
//! It is also **synchronous**, so writes go through [`spawn_blocking`]. A commit
//! fsyncs, and an fsync on the async executor stalls every other task in the
//! process.
//!
//! [`spawn_blocking`]: tokio::task::spawn_blocking

use std::path::Path;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use async_trait::async_trait;
use redb::{Database, ReadableDatabase, ReadableTable, TableDefinition};

use crate::core::{Digest, EffectKey, Epoch, RunId, Seq, StoreError};
use crate::journal::{Append, Cancellation, Head, JournalStore, Lease, Record, WaitingRun};

/// `(run_id, seq) -> encoded record`. Ordered by construction, so a run's
/// history is a range scan and its head is the last entry of that range.
const JOURNAL: TableDefinition<(&str, u64), &[u8]> = TableDefinition::new("journal");

/// Every journaled record that belongs to a case, in case order.
///
/// `(tenant, case, run, seq) -> ()`. The primary row stays in [`JOURNAL`]; this
/// is only the ordering, so a record has exactly one home and the two cannot
/// disagree about its bytes.
///
/// Why an index rather than a filter: *show me everything about this matter* is
/// the question a regulated deployment asks, and answering it by listing a
/// case's runs and reading each one is a join whose cost grows with the case's
/// life. It also **misses** any record written by a run the case does not own —
/// which is exactly what a sweep is, since one tick may escalate several cases
/// and belongs to none of them.
///
/// Tenant-first, like every other key here, so a scan that forgets the
/// predicate returns nothing rather than somebody else's matter.
///
/// The write costs one B-tree insert inside the transaction that was already
/// open. Measured against uncased appends it is **below the noise floor**: two
/// runs of 2,000 appends disagreed about the sign, because a durable append is
/// dominated by the fsync and this is not. A number quoted from either run
/// would have been noise with a decimal point.
const JOURNAL_BY_CASE: TableDefinition<(&str, &str, &str, u64), ()> =
    TableDefinition::new("journal_by_case");

/// `(run_id, effect_key) -> seq`. Exactly-once. Only `EffectStarted` is written
/// here — the `WHERE` clause the shared-store backend's partial index spells.
const EFFECT_ONCE: TableDefinition<(&str, &str), u64> = TableDefinition::new("effect_once");

/// `run_id -> (owner, epoch, expires_at)`.
const RUN_LEASE: TableDefinition<&str, (&str, u64, u64)> = TableDefinition::new("run_lease");

/// `run_id -> (outcome, chain_head, sealed_at)`.
const RUN_SEAL: TableDefinition<&str, (&str, &[u8], u64)> = TableDefinition::new("run_seal");

/// `(tenant, outcome, ordinal) -> run_key`: concluded runs by how they ended.
///
/// A **derived** index. The outcome's home is the chain — the executor appends
/// `RunSealed` with the conclusion, so tamper detection covers how a run ended,
/// which it would not if a side table were authoritative. This is a convenience
/// for the one question the chain answers slowly: *what is quarantined right
/// now?*
///
/// Maintained by `append`, in the same transaction as the record it derives
/// from, and **the last conclusion wins**: a run that failed, was resumed and
/// then succeeded moves from `failed` to `succeeded`. Deriving it from the seal
/// instead would freeze the *first* conclusion forever — a resumed run would be
/// listed as failed for the rest of its life, which is worse than no index,
/// because a wrong answer reads exactly like a right one.
///
/// The ordinal is a per-tenant conclusion counter, so a scan is oldest-first
/// without a sort and a bounded reverse scan yields the newest conclusions.
const RUN_BY_OUTCOME: TableDefinition<(&str, &str, u64), &str> =
    TableDefinition::new("run_by_outcome");

/// `(tenant, run_key) -> (outcome, ordinal)`: the reverse of `RUN_BY_OUTCOME`.
///
/// What lets a re-conclusion *replace* its index row rather than accumulate a
/// second one — without it, a failed-then-succeeded run would appear under both
/// outcomes, and the failed listing would never drain.
const RUN_OUTCOME: TableDefinition<(&str, &str), (&str, u64)> = TableDefinition::new("run_outcome");

/// `(tenant, key) -> (run_id, admitted_at)`: the admission keys this tenant has
/// issued.
///
/// A **derived** index, like `RUN_OUTCOME`: the key's home is the `RunAdmitted`
/// record, and `append` maintains this in the same transaction as that record.
/// The key *shape* is the constraint, so a second admission under an issued key
/// is inexpressible rather than refused by a check somebody could reorder
/// around — the property `CORR_OPEN` relies on for "one open case per business
/// key".
///
/// The tenant leads because an admission key is a business value: without the
/// scope one tenant's redelivery would be answered with another's run.
const RUN_ADMISSION: TableDefinition<(&str, &str), (&str, i64)> =
    TableDefinition::new("run_admission");

/// `(tenant, updated_at, run_id) -> ()`, every run by last durable append.
const RUN_ACTIVITY: TableDefinition<(&str, u64, &str), ()> = TableDefinition::new("run_activity");
/// `(tenant, run_id) -> updated_at`, for replacing the activity index row.
const RUN_LAST_ACTIVITY: TableDefinition<(&str, &str), u64> =
    TableDefinition::new("run_last_activity");

/// `(tenant, run_key) -> source`: the producer that admitted a run, from its
/// `RunAdmitted` admission key. Written once, in that record's transaction.
const RUN_SOURCE: TableDefinition<(&str, &str), &str> = TableDefinition::new("run_source");

/// `(tenant, source, updated_at, run_key) -> ()`: [`RUN_ACTIVITY`] narrowed to
/// one producer's runs, moved with it on every append — what lets a served
/// listing read only its caller's runs.
const RUN_BY_SOURCE: TableDefinition<(&str, &str, u64, &str), ()> =
    TableDefinition::new("run_by_source");

/// `(tenant, until, run_key) -> reason`: the runs whose last record is a
/// suspension, soonest-due first.
///
/// **Derived from the chain on every append, never appended to.** A run leaves
/// this table by making any progress at all, which is what keeps an
/// oldest-first page from stopping dead: the head of it is removed by the verb
/// an operator came here to run. Scanning history for `RunSuspended` would be
/// the wrong answer for the opposite reason — every run that ever waited
/// carries one forever.
///
/// Ordered by the instant rather than by insertion, because the actionable
/// question is *which of these should have moved by now*, and a run whose
/// `until` is in the past is the one lying inert.
const RUN_WAITING: TableDefinition<(&str, i64, &str), &str> = TableDefinition::new("run_waiting");
/// `(tenant, run_key) -> until`, for removing the row above without scanning.
///
/// The same reverse-index shape [`RUN_OUTCOME`] uses, and for the same reason: a
/// run that suspends twice under different instants would otherwise accumulate
/// two rows, and the stale one would name a wait that is over.
const RUN_WAITING_AT: TableDefinition<(&str, &str), i64> = TableDefinition::new("run_waiting_at");

/// `log_index -> run_id`, the plane's Merkle log in seal order.
///
/// Positions are never reissued, so this table keeps gaps when a run is removed
/// — deliberately, so a freed position cannot be silently reused by a different
/// run. The *tree* is built by iterating in key order, which yields dense
/// positions; that is what makes the `ROW_NUMBER() OVER (ORDER BY log_index)`
/// the SQL backend needed unnecessary here, and with it the rank-vs-index bug
/// that cost an afternoon.
const SEAL_LOG: TableDefinition<(&str, u64), &str> = TableDefinition::new("seal_log");

/// `run_id -> (actor, reason, requested_at)`. An operator's stop request.
///
/// Keyed by run, so the request is idempotent and a retry cannot overwrite who
/// intervened. Deliberately not fenced: whoever wants a run stopped is not its
/// owner, holds no epoch, and is usually asking because the owner is busy.
const RUN_CANCEL: TableDefinition<&str, (&str, &str, &str, u64)> =
    TableDefinition::new("run_cancel");

/// The next log position to hand out. `MAX + 1` over a table that keeps its gaps
/// would reuse a removed run's slot.
const COUNTERS: TableDefinition<&str, u64> = TableDefinition::new("counters");

const NEXT_LOG_INDEX: &str = "next_log_index";

/// The next conclusion ordinal to hand out, per tenant.
///
/// Separate from `NEXT_LOG_INDEX` because conclusions are not seals: a failed
/// run concludes without entering the Merkle log, and interleaving the two
/// sequences would leave the log with gaps that mean nothing.
const NEXT_CONCLUSION: &str = "next_conclusion";

/// An upper bound for a `&str` range end.
///
/// redb orders by encoded bytes, and `\u{10FFFF}` encodes to the highest
/// sequence UTF-8 admits, so no valid key can sort above it.
pub(super) const MAX_STR: &str = "\u{10FFFF}";

/// A single-node journal store on redb.
#[derive(Debug, Clone)]
pub struct RedbStore {
    db: Arc<Database>,
    /// Attached to the store rather than passed per append, because only the
    /// store knows a record's chain hash — it assigns `seq` and `prev_hash`
    /// inside the same transaction that writes. Signing anywhere else would be
    /// signing a guess about what the hash will be.
    signer: Option<Arc<dyn crate::core::Signer>>,
    /// Names this plane's Merkle log in every checkpoint.
    origin: String,
    /// Which tenant's keyspace this handle addresses.
    ///
    /// Part of the *key*, not a filter applied after reading. A store handle for
    /// tenant A cannot name tenant B's run: the key it builds does not exist, so
    /// a cross-tenant read is a miss rather than a row that some `WHERE` clause
    /// was supposed to remove. A filter that can be forgotten is not isolation,
    /// and the query that forgets it looks exactly like working software.
    tenant: crate::core::TenantId,
    /// What every read lifts a stored record through. The bytes and their
    /// hash are never touched; only the body this build reads is.
    upcaster: Arc<dyn crate::journal::Upcaster>,
}

impl RedbStore {
    /// Open (or create) a database file.
    ///
    /// # Errors
    ///
    /// If the file cannot be opened or the tables cannot be created.
    pub fn open(path: impl AsRef<Path>) -> Result<Self, StoreError> {
        let db = Database::create(path).map_err(|e| be(&e))?;
        Self::init(db)
    }

    /// An ephemeral database — for tests, for the simulator, and for the CLI's
    /// default journal.
    ///
    /// **Actually in memory.** It was not, and the reason is a comment that
    /// aged into a defect: it said *redb has no in-memory backend*, which was
    /// true of redb 1 and false since the dependency moved to 3. So the
    /// implementation created a file under [`std::env::temp_dir`] and unlinked
    /// it — an anonymous file, which behaves like memory right up until there is
    /// nowhere to put it.
    ///
    /// That is not a hypothetical. `agentplane run` journals here by default,
    /// and the container image runs on a **read-only root filesystem** with no
    /// writable temp directory, so the documented first command failed with
    /// `Read-only file system (os error 30)` — a message naming neither the
    /// journal nor the temp directory. The same failure waits in any hardened
    /// deployment, a scratch container, or a sandbox with no `/tmp`.
    ///
    /// A name that says *in memory* should not need a filesystem to be true.
    ///
    /// # Errors
    ///
    /// If the database cannot be initialised.
    pub fn open_in_memory() -> Result<Self, StoreError> {
        let db = Database::builder()
            .create_with_backend(redb::backends::InMemoryBackend::new())
            .map_err(|e| be(&e))?;
        Self::init(db)
    }

    /// Create every table, so a read on a fresh database is not a missing-table
    /// error the read paths would each have to special-case.
    fn init(db: Database) -> Result<Self, StoreError> {
        let w = begin_write(&db)?;
        {
            w.open_table(JOURNAL).map_err(|e| be(&e))?;
            w.open_table(JOURNAL_BY_CASE).map_err(|e| be(&e))?;
            w.open_table(EFFECT_ONCE).map_err(|e| be(&e))?;
            w.open_table(RUN_LEASE).map_err(|e| be(&e))?;
            w.open_table(RUN_SEAL).map_err(|e| be(&e))?;
            w.open_table(RUN_BY_OUTCOME).map_err(|e| be(&e))?;
            w.open_table(RUN_OUTCOME).map_err(|e| be(&e))?;
            w.open_table(RUN_WAITING).map_err(|e| be(&e))?;
            w.open_table(RUN_WAITING_AT).map_err(|e| be(&e))?;
            w.open_table(SEAL_LOG).map_err(|e| be(&e))?;
            w.open_table(RUN_CANCEL).map_err(|e| be(&e))?;
            w.open_table(COUNTERS).map_err(|e| be(&e))?;
            super::redb_cases::create_tables(&w)?;
            super::redb_events::create_tables(&w)?;
            super::redb_tasks::create_tables(&w)?;
            super::redb_timers::create_tables(&w)?;
            super::redb_batches::create_tables(&w)?;
        }
        w.commit().map_err(|e| be(&e))?;
        Ok(Self {
            db: Arc::new(db),
            signer: None,
            origin: "agentplane".to_owned(),
            tenant: crate::core::TenantId::default(),
            upcaster: crate::journal::current_upcaster(),
        })
    }

    /// A handle addressing one tenant's keyspace.
    ///
    /// Every key this store builds is prefixed, so two tenants may share one
    /// physical database and still not reach each other: a read for a run that
    /// belongs to another tenant finds nothing, because the key it looks under
    /// was never written.
    ///
    /// The checkpoint origin carries the tenant too — two tenants sharing a
    /// database would otherwise publish checkpoints under one origin name, and
    /// a witness could not tell whose history it was cosigning. Composed at
    /// read time from the base name and the tenant rather than baked into a
    /// field: baking it in made the builder order-sensitive, so `for_tenant`
    /// called twice double-qualified the origin and `origin()` after
    /// `for_tenant` silently dropped the tenant — either way a checkpoint
    /// under a name no verifier ever saw again.
    #[must_use]
    pub fn for_tenant(mut self, tenant: crate::core::TenantId) -> Self {
        self.tenant = tenant;
        self
    }

    /// The name this tenant's Merkle log publishes under.
    ///
    /// The default tenant keeps the bare base name, so a single-tenant plane's
    /// checkpoints read as they always have.
    fn log_origin(&self) -> String {
        if self.tenant.as_str() == crate::core::TenantId::DEFAULT {
            self.origin.clone()
        } else {
            format!("{}/{}", self.origin, self.tenant)
        }
    }

    /// The storage key for a run, and the only place one is derived.
    ///
    /// One derivation because many places building the same string is how one
    /// of them ends up building a slightly different one.
    pub(super) fn run_key(&self, run: RunId) -> String {
        run_key_in(self.tenant.as_str(), &run.to_string())
    }

    /// This handle's tenant, for a key or a range bound.
    ///
    /// Owned rather than borrowed because every caller moves it into a closure
    /// that outlives the borrow.
    pub(super) fn tenant_name(&self) -> String {
        self.tenant.to_string()
    }

    /// The tenant this handle is scoped to, for the store traits' accessors.
    ///
    /// Borrowed, unlike `tenant_name` above: a trait answering *who do you
    /// serve* is not moving the answer into a closure and should not allocate
    /// to give it.
    pub(super) fn tenant_str(&self) -> &str {
        self.tenant.as_str()
    }

    /// Name this plane's Merkle log.
    ///
    /// Goes into every checkpoint, so two planes' checkpoints cannot be
    /// confused for one another — which matters the moment they are published
    /// to a shared witness.
    #[must_use]
    pub fn origin(mut self, origin: impl Into<String>) -> Self {
        self.origin = origin.into();
        self
    }

    /// Write records signed as this identity.
    ///
    /// Off unless asked, and the default is *unsigned rather than self-signed*:
    /// a plane that quietly minted its own key would produce records that look
    /// attested and prove nothing.
    #[must_use]
    pub fn signing_as(mut self, signer: Arc<dyn crate::core::Signer>) -> Self {
        self.signer = Some(signer);
        self
    }

    /// Read records through this upcaster instead of the one this build ships.
    ///
    /// The default is [`current_upcaster`](crate::journal::current_upcaster),
    /// which is what a deployment wants; another is for a reader that must
    /// lift shapes this build's own does not.
    #[must_use]
    pub fn upcasting_with(mut self, upcaster: Arc<dyn crate::journal::Upcaster>) -> Self {
        self.upcaster = upcaster;
        self
    }

    /// Run a closure against the database on the blocking pool.
    ///
    /// redb is synchronous and a commit fsyncs; doing that on the async
    /// executor stalls every other task in the process.
    pub(super) async fn with_db<T, F>(&self, f: F) -> Result<T, StoreError>
    where
        T: Send + 'static,
        F: FnOnce(&Database) -> Result<T, StoreError> + Send + 'static,
    {
        let db = Arc::clone(&self.db);
        tokio::task::spawn_blocking(move || f(&db))
            .await
            .map_err(|e| StoreError::Backend(format!("blocking pool: {e}")))?
    }

    /// The log's leaves, in seal order — already leaf-hashed, as the type
    /// says.
    async fn log_leaves(&self) -> Result<Vec<crate::core::merkle::LeafHash>, StoreError> {
        let tenant = self.tenant.clone();
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let log = r.open_table(SEAL_LOG).map_err(|e| be(&e))?;
            let seals = r.open_table(RUN_SEAL).map_err(|e| be(&e))?;
            let mut out = Vec::new();
            // This tenant's range only. Iterating the whole table would build a
            // Merkle log covering other tenants' runs, so a checkpoint would
            // commit to history its holder may not see.
            for entry in log
                .range((tenant.as_str(), 0)..=(tenant.as_str(), u64::MAX))
                .map_err(|e| be(&e))?
            {
                let (_, run) = entry.map_err(|e| be(&e))?;
                let Some(seal) = seals.get(run.value()).map_err(|e| be(&e))? else {
                    continue;
                };
                let (_, head, _) = seal.value();
                out.push(crate::core::merkle::leaf_hash(&digest(head)?));
            }
            Ok(out)
        })
        .await
    }

    /// Where each run sits in the log, and its leaf value, from one walk.
    ///
    /// The position is the run's index in seal order, counted while iterating —
    /// dense by construction, because the tree is built from the same walk.
    async fn positions(&self, runs: &[RunId]) -> Result<Vec<Option<(u64, Digest)>>, StoreError> {
        let keys: Vec<String> = runs.iter().map(|&run| self.run_key(run)).collect();
        let tenant = self.tenant.clone();
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let log = r.open_table(SEAL_LOG).map_err(|e| be(&e))?;
            let seals = r.open_table(RUN_SEAL).map_err(|e| be(&e))?;
            let mut wanted: std::collections::HashMap<&str, Option<(u64, Digest)>> =
                keys.iter().map(|k| (k.as_str(), None)).collect();
            let mut rank = 0u64;
            for entry in log
                .range((tenant.as_str(), 0)..=(tenant.as_str(), u64::MAX))
                .map_err(|e| be(&e))?
            {
                let (_, run_id) = entry.map_err(|e| be(&e))?;
                let Some(seal) = seals.get(run_id.value()).map_err(|e| be(&e))? else {
                    continue;
                };
                if let Some(slot) = wanted.get_mut(run_id.value()) {
                    let (_, head, _) = seal.value();
                    *slot = Some((rank, digest(head)?));
                }
                rank += 1;
            }
            Ok(keys.iter().map(|k| wanted[k.as_str()]).collect())
        })
        .await
    }

    /// Overwrite a record's stored bytes while leaving its hash untouched,
    /// simulating an after-the-fact edit.
    ///
    /// Exists so the test suite can prove that tampering is *detected* rather
    /// than assumed impossible. Not part of the supported surface.
    ///
    /// # Errors
    ///
    /// If the write fails.
    #[doc(hidden)]
    #[cfg(feature = "testkit")]
    pub async fn tamper_for_test(
        &self,
        run: RunId,
        seq: Seq,
        body: Vec<u8>,
    ) -> Result<(), StoreError> {
        let key = self.run_key(run);
        self.with_db(move |db| {
            let w = begin_write(db)?;
            {
                let mut t = w.open_table(JOURNAL).map_err(|e| be(&e))?;
                let existing = t
                    .get((key.as_str(), seq))
                    .map_err(|e| be(&e))?
                    .map(|v| v.value().to_vec());
                if let Some(bytes) = existing {
                    let row = Row::decode(&bytes)?;
                    let tampered = Row { body, ..row };
                    t.insert((key.as_str(), seq), tampered.encode()?.as_slice())
                        .map_err(|e| be(&e))?;
                }
            }
            w.commit().map_err(|e| be(&e))?;
            Ok(())
        })
        .await
    }

    /// Remove a run entirely — journal rows and seal alike.
    ///
    /// The one tampering the per-run chain structurally cannot see, so a suite
    /// that could not perform it could not test the mechanism that does. Not
    /// part of the supported surface.
    ///
    /// # Errors
    ///
    /// If the write fails.
    #[doc(hidden)]
    #[cfg(feature = "testkit")]
    pub async fn delete_run_for_test(&self, run: RunId) -> Result<(), StoreError> {
        let key = self.run_key(run);
        self.with_db(move |db| {
            let w = begin_write(db)?;
            {
                let mut j = w.open_table(JOURNAL).map_err(|e| be(&e))?;
                let seqs: Vec<u64> = j
                    .range((key.as_str(), 0u64)..=(key.as_str(), u64::MAX))
                    .map_err(|e| be(&e))?
                    .filter_map(Result::ok)
                    .map(|(k, _)| k.value().1)
                    .collect();
                for s in seqs {
                    j.remove((key.as_str(), s)).map_err(|e| be(&e))?;
                }
                w.open_table(RUN_SEAL)
                    .map_err(|e| be(&e))?
                    .remove(key.as_str())
                    .map_err(|e| be(&e))?;
                // The log position is left in place, holding its gap: a freed
                // slot must never be reissued to a different run.
            }
            w.commit().map_err(|e| be(&e))?;
            Ok(())
        })
        .await
    }
}

/// One journal row.
///
/// Encoded by hand rather than through serde: the body is already bytes, and a
/// JSON envelope would base64 it on every append and every read.
struct Row {
    body: Vec<u8>,
    prev_hash: [u8; 32],
    hash: [u8; 32],
    key_id: Option<String>,
    signature: Option<Vec<u8>>,
}

impl Row {
    fn encode(&self) -> Result<Vec<u8>, StoreError> {
        let mut out = Vec::with_capacity(self.body.len() + 96);
        out.extend_from_slice(&self.prev_hash);
        out.extend_from_slice(&self.hash);
        // A key id without a signature (or the reverse) is a half-written row,
        // not an unsigned record. Written as one optional pair so the two
        // cannot drift apart.
        match (&self.key_id, &self.signature) {
            (Some(k), Some(s)) => {
                out.push(1);
                push_bytes(&mut out, k.as_bytes())?;
                push_bytes(&mut out, s)?;
            }
            _ => out.push(0),
        }
        push_bytes(&mut out, &self.body)?;
        Ok(out)
    }

    fn decode(raw: &[u8]) -> Result<Self, StoreError> {
        let corrupt = |what: &str| StoreError::Corrupt {
            seq: 0,
            detail: format!("journal row truncated in {what}"),
        };
        if raw.len() < 65 {
            return Err(corrupt("header"));
        }
        let prev_hash: [u8; 32] = raw[0..32].try_into().map_err(|_| corrupt("prev_hash"))?;
        let hash: [u8; 32] = raw[32..64].try_into().map_err(|_| corrupt("hash"))?;
        let mut at = 64;
        // A boolean byte, read as one: any other value is damage, and reading
        // it as *unsigned* would strip the signature without a word.
        let attested = match raw[at] {
            0 => false,
            1 => true,
            _ => return Err(corrupt("the signature flag")),
        };
        at += 1;
        let (key_id, signature) = if attested {
            let (k, n) = take_bytes(raw, at).ok_or_else(|| corrupt("key_id"))?;
            at = n;
            let (s, n) = take_bytes(raw, at).ok_or_else(|| corrupt("signature"))?;
            at = n;
            (
                Some(String::from_utf8(k).map_err(|_| corrupt("key_id utf-8"))?),
                Some(s),
            )
        } else {
            (None, None)
        };
        let (body, end) = take_bytes(raw, at).ok_or_else(|| corrupt("body"))?;
        if end != raw.len() {
            return Err(StoreError::Corrupt {
                seq: 0,
                detail: "journal row carries bytes after its body".to_owned(),
            });
        }
        Ok(Self {
            body,
            prev_hash,
            hash,
            key_id,
            signature,
        })
    }

    fn into_record(self, upcaster: &dyn crate::journal::Upcaster) -> Result<Record, StoreError> {
        let signature = self
            .key_id
            .zip(self.signature)
            .map(|(key_id, signature)| crate::core::KeySignature { key_id, signature });
        Record::from_stored_with(
            upcaster,
            self.body,
            Digest::from_bytes(self.prev_hash),
            Digest::from_bytes(self.hash),
            signature,
        )
    }
}

fn push_bytes(out: &mut Vec<u8>, bytes: &[u8]) -> Result<(), StoreError> {
    // A record is refused above `MAX_RECORD_BYTES`, so no field can reach 4 GiB
    // — but the cast is checked anyway rather than assumed, because a length
    // prefix that disagrees with the payload corrupts every record after it.
    let len = u32::try_from(bytes.len()).map_err(|_| {
        StoreError::Backend(format!(
            "a row field of {} bytes has no u32 length",
            bytes.len()
        ))
    })?;
    out.extend_from_slice(&len.to_le_bytes());
    out.extend_from_slice(bytes);
    Ok(())
}

fn take_bytes(raw: &[u8], at: usize) -> Option<(Vec<u8>, usize)> {
    let end = at.checked_add(4)?;
    let len = u32::from_le_bytes(raw.get(at..end)?.try_into().ok()?) as usize;
    let stop = end.checked_add(len)?;
    Some((raw.get(end..stop)?.to_vec(), stop))
}

fn digest(bytes: &[u8]) -> Result<Digest, StoreError> {
    let b: [u8; 32] = bytes.try_into().map_err(|_| StoreError::Corrupt {
        seq: 0,
        detail: "a stored hash is not 32 bytes".into(),
    })?;
    Ok(Digest::from_bytes(b))
}

/// Begin a write transaction with its durability stated rather than inherited.
///
/// redb already defaults to [`Durability::Immediate`] — a commit is persistent
/// by the time it returns. Stating it anyway, in one place every write path
/// goes through, because "a committed record survives the process" is this
/// crate's central promise and a promise resting on a dependency's default can
/// be weakened by an upgrade with nothing here changing.
///
/// What this does *not* buy is a test: no suite here kills a process mid-commit,
/// so a regression would not be caught. It is a belt on a guarantee the tests
/// cannot reach, not a checked invariant.
pub(super) fn begin_write(db: &Database) -> Result<redb::WriteTransaction, StoreError> {
    let mut w = db.begin_write().map_err(|e| be(&e))?;
    w.set_durability(redb::Durability::Immediate)
        .map_err(|e| be(&e))?;
    Ok(w)
}

/// [`RedbStore::run_key`] for a tenant and a run already rendered as
/// strings — the form the wait tables key their rows by.
pub(super) fn run_key_in(tenant: &str, run: &str) -> String {
    format!("{tenant}/{run}")
}

/// Whether `run` is sealed, read inside the caller's write transaction.
///
/// The liveness check the wait tables make before handing a run a timer or an
/// event: a sealed run can record neither.
pub(super) fn is_sealed(
    w: &redb::WriteTransaction,
    tenant: &str,
    run: &str,
) -> Result<bool, StoreError> {
    let seals = w.open_table(RUN_SEAL).map_err(|e| be(&e))?;
    Ok(seals
        .get(run_key_in(tenant, run).as_str())
        .map_err(|e| be(&e))?
        .is_some())
}

pub(super) fn be<E: std::fmt::Display>(e: &E) -> StoreError {
    StoreError::Backend(e.to_string())
}

/// Read a stored column through the type's own vocabulary.
///
/// Every string column this backend keeps is decoded through the core type's
/// `parse` rather than through a match of its own — see [`CaseStatus::parse`]
/// for why nothing else may spell a stored vocabulary. An unrecognised value is
/// damage, and damage is reported rather than defaulted: a decoder that answers
/// with a safe value hands its caller a fact this store invented about a row it
/// could not read.
///
/// [`CaseStatus::parse`]: crate::core::CaseStatus::parse
pub(super) fn decoded<T>(what: &str, raw: &str, parsed: Option<T>) -> Result<T, StoreError> {
    parsed.ok_or_else(|| StoreError::Corrupt {
        seq: 0,
        detail: format!("unknown {what} '{raw}'"),
    })
}

/// Wall-clock read for lease expiry.
///
/// Lease timing is infrastructure, not run logic: it never enters the journal's
/// content and therefore cannot affect replay. Run-visible time goes through
/// `StepCtx::now`, which journals the instant.
#[allow(clippy::disallowed_methods)]
fn now_secs() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map_or(0, |d| d.as_secs())
}

/// A lease expiry instant from a TTL, refusing what whole seconds cannot hold.
///
/// Lease timing here has whole-second granularity, so a TTL below one second is
/// **refused rather than clamped**. A clamp would turn "expire immediately"
/// into "hold for a second" without telling anyone — a contract the runtime
/// relies on, enforced only upstream. Nothing legitimate reaches
/// this refusal: the runtime builder already refuses any TTL below its own
/// two-second minimum at `build()` time. This is the *store's* boundary
/// enforcement, for every other embedder of the trait — they get the same
/// refusal instead of a silently longer lease. A TTL of 1.5s still truncates
/// to one second; the granularity is the contract, and only the value that
/// truncates to *zero* is a lie worth refusing.
///
/// The addition is checked, not assumed: `now + Duration::MAX` would wrap into
/// the past and produce a lease that is born expired — free for anyone to
/// claim while its owner believes it holds forever.
fn lease_expiry(now: u64, ttl: Duration) -> Result<u64, StoreError> {
    let secs = ttl.as_secs();
    if secs == 0 {
        return Err(StoreError::Backend(format!(
            "a lease TTL of {ttl:?} is below this store's whole-second \
             granularity and would round to zero — pass at least one second, \
             or use the runtime's lease_ttl which enforces its own minimum"
        )));
    }
    now.checked_add(secs).ok_or_else(|| {
        StoreError::Backend(format!(
            "a lease TTL of {ttl:?} overflows the expiry instant — the lease \
             would wrap into the past and read as already expired"
        ))
    })
}

/// Refuse to seal a run whose last record is not its conclusion.
///
/// A seal is a leaf in the log: sealing an empty run commits the zero digest,
/// and sealing a run mid-flight freezes a history that never ended.
fn refuse_unconcluded(run: &str, last: Option<&[u8]>) -> Result<(), StoreError> {
    let kind = last
        .and_then(|body| serde_json::from_slice::<serde_json::Value>(body).ok())
        .and_then(|body| body.get("kind").and_then(|k| k.as_str().map(str::to_owned)));
    if kind.as_deref() == Some("RunConcluded") {
        return Ok(());
    }
    Err(StoreError::Backend(format!(
        "run {run} cannot be sealed: its last record is {} rather than its conclusion",
        kind.as_deref().unwrap_or("absent")
    )))
}

/// The last record of a run, or genesis.
fn head_of(
    t: &impl ReadableTable<(&'static str, u64), &'static [u8]>,
    run: &str,
) -> Result<Head, StoreError>
where
{
    let last = t
        .range((run, 0u64)..=(run, u64::MAX))
        .map_err(|e| be(&e))?
        .next_back();
    match last {
        None => Ok(Head::genesis()),
        Some(entry) => {
            let (k, v) = entry.map_err(|e| be(&e))?;
            let row = Row::decode(v.value())?;
            Ok(Head {
                seq: k.value().1,
                hash: Digest::from_bytes(row.hash),
            })
        }
    }
}

/// One past the highest epoch this run's records already carry.
///
/// The fencing token's high-water mark normally lives in the lease table, and
/// the lease table is store-local state an export does not carry. Reading it
/// back from the journal makes the token monotonic across **any** way the lease
/// row can go missing — a restore, a hand-rebuilt store, a truncated table —
/// rather than only the one somebody remembered to seed.
///
/// One read rather than a scan, because epochs never decrease along a run's own
/// chain: a lease is only ever issued above the current mark, and a restore
/// replays bodies in sequence order. An empty journal answers 1.
///
/// **What it cannot see** is an epoch issued and never written under — a lease
/// taken by an owner that died before appending leaves no record. So it bounds
/// every epoch that ever wrote, which is every epoch that can conflict with the
/// rebuilt history, and a pre-disaster instance still holding such a lease is
/// fenced by the runbook rather than by this: the old deployment must not reach
/// the restored store.
fn epoch_after_history(
    w: &redb::WriteTransaction,
    run: &str,
    upcaster: &dyn crate::journal::Upcaster,
) -> Result<Epoch, StoreError> {
    let t = w.open_table(JOURNAL).map_err(|e| be(&e))?;
    let last = t
        .range((run, 0u64)..=(run, u64::MAX))
        .map_err(|e| be(&e))?
        .next_back();
    match last {
        None => Ok(1),
        Some(entry) => {
            let (_, v) = entry.map_err(|e| be(&e))?;
            Ok(Row::decode(v.value())?.into_record(upcaster)?.body.epoch + 1)
        }
    }
}

#[allow(clippy::too_many_lines)]
#[async_trait]
impl JournalStore for RedbStore {
    /// One file, one writer. That single writer is what makes exactly-once a
    /// table key and fencing race-free here — and it is why a plane that needs
    /// two instances needs `PostgreSQL` instead.
    fn is_shared(&self) -> bool {
        false
    }
    fn seals(&self) -> bool {
        false
    }

    fn tenant(&self) -> &str {
        self.tenant.as_str()
    }

    async fn append(&self, epoch: Epoch, batch: Vec<Append>) -> Result<Vec<Record>, StoreError> {
        if batch.is_empty() {
            return Ok(Vec::new());
        }
        let run = batch[0].run;
        if let Some(a) = batch.iter().find(|a| a.run != run) {
            return Err(StoreError::Backend(format!(
                "batch spans runs {run} and {} — a batch is one run's atomic unit",
                a.run
            )));
        }
        let signer = self.signer.clone();
        let key = self.run_key(run);
        let run_id = run.to_string();
        let tenant = self.tenant_name();
        let upcaster = Arc::clone(&self.upcaster);

        self.with_db(move |db| {
            let w = begin_write(db)?;
            let sealed = {
                let mut journal = w.open_table(JOURNAL).map_err(|e| be(&e))?;
                let mut by_case = w.open_table(JOURNAL_BY_CASE).map_err(|e| be(&e))?;
                let mut once = w.open_table(EFFECT_ONCE).map_err(|e| be(&e))?;

                // Fence inside the write transaction: no window between the
                // check and the append for a stale writer to use.
                {
                    let leases = w.open_table(RUN_LEASE).map_err(|e| be(&e))?;
                    if let Some(l) = leases.get(key.as_str()).map_err(|e| be(&e))? {
                        let (_, current, _) = l.value();
                        if epoch != current {
                            return Err(StoreError::Fenced {
                                run: key.clone(),
                                held: epoch,
                                current,
                            });
                        }
                    }
                }

                // Sealed is frozen. The Merkle leaf is the chain head at seal
                // time, so an append past it — even by the epoch's rightful
                // holder — advances the true head past what every checkpoint
                // attests. Checked inside the write transaction like the fence,
                // because the executor's refusal to resume a closed run is
                // logic a future caller can bypass, and this is the constraint
                // that cannot be. The sealing appends themselves precede the
                // seal row, so they never meet this check.
                {
                    let seals = w.open_table(RUN_SEAL).map_err(|e| be(&e))?;
                    if let Some(seal) = seals.get(key.as_str()).map_err(|e| be(&e))? {
                        let (outcome, _, _) = seal.value();
                        return Err(StoreError::RunSealed {
                            run: key.clone(),
                            outcome: outcome.to_owned(),
                        });
                    }
                }

                let mut head = head_of(&journal, &key)?;
                let mut sealed = Vec::with_capacity(batch.len());
                // A run that suspends and then makes any progress has
                // stopped waiting, and a listing an operator re-arms from must
                // not still name it.
                let mut waiting: Option<crate::core::SuspendReason> = None;
                let mut admitted_from: Option<String> = None;

                for append in batch {
                    let (body, written) =
                        append.into_parts(head.seq + 1, epoch, upcaster.as_ref())?;
                    let effect_key = body.effect_key.map(EffectKey::to_hex);
                    let is_start =
                        matches!(body.kind, crate::journal::RecordKind::EffectStarted { .. });
                    let (conclusion, claimed) = match &body.kind {
                        crate::journal::RecordKind::RunConcluded { outcome, .. } => {
                            (Some(outcome.clone()), None)
                        }
                        crate::journal::RecordKind::RunAdmitted {
                            idempotency_key: Some(k),
                            ..
                        } => (None, Some(k.clone())),
                        _ => (None, None),
                    };
                    // Cleared and re-set on every record rather than only on a
                    // suspension: that is what makes the index track the run's
                    // *last* record instead of accumulating.
                    waiting = match &body.kind {
                        crate::journal::RecordKind::RunSuspended { reason } => Some(reason.clone()),
                        _ => None,
                    };
                    let record = Record::seal_at(body, written, head.hash, signer.as_deref())?;
                    let seq = record.seq();

                    // Exactly-once. The key is the constraint: a prior value
                    // means this effect already started in this run.
                    if is_start && let Some(ek) = &effect_key {
                        let prior = once
                            .insert((key.as_str(), ek.as_str()), seq)
                            .map_err(|e| be(&e))?;
                        if prior.is_some() {
                            return Err(record.effect_key().map_or_else(
                                || StoreError::Backend("duplicate effect".into()),
                                StoreError::DuplicateEffect,
                            ));
                        }
                    }

                    let row = Row {
                        body: record.raw().to_vec(),
                        prev_hash: *record.prev_hash.as_bytes(),
                        hash: *record.hash.as_bytes(),
                        key_id: record.signature.as_ref().map(|a| a.key_id.clone()),
                        signature: record.signature.as_ref().map(|a| a.signature.clone()),
                    };
                    journal
                        .insert((key.as_str(), seq), row.encode()?.as_slice())
                        .map_err(|e| be(&e))?;
                    // Written in the same transaction as the row it points at,
                    // so the index cannot outlive or precede its record.
                    if let Some(case) = record.body.case {
                        by_case
                            .insert(
                                (
                                    tenant.as_str(),
                                    case.to_string().as_str(),
                                    key.as_str(),
                                    seq,
                                ),
                                (),
                            )
                            .map_err(|e| be(&e))?;
                    }

                    // Claimed in the same transaction as the record that
                    // carries it, so the claim and the run are one event. An
                    // existing entry is the refusal, never something to
                    // overwrite.
                    if let Some(source) = record.admission_source() {
                        admitted_from = Some(source.to_owned());
                    }
                    if let Some(k) = claimed {
                        let mut admissions = w.open_table(RUN_ADMISSION).map_err(|e| be(&e))?;
                        if let Some(held) = admissions
                            .get((tenant.as_str(), k.as_str()))
                            .map_err(|e| be(&e))?
                            .map(|v| v.value().0.to_owned())
                        {
                            return Err(StoreError::DuplicateAdmission { key: k, run: held });
                        }
                        admissions
                            .insert(
                                (tenant.as_str(), k.as_str()),
                                (run_id.as_str(), now_secs().cast_signed()),
                            )
                            .map_err(|e| be(&e))?;
                    }

                    // The outcome index derives from the chain, here, in the
                    // same transaction as the record it derives from — and the
                    // last conclusion wins. A failed run that is resumed and
                    // succeeds moves between listings; an index fed by the seal
                    // would keep saying "failed" forever, and a wrong answer
                    // reads exactly like a right one.
                    if let Some(outcome) = conclusion {
                        let mut by_outcome = w.open_table(RUN_BY_OUTCOME).map_err(|e| be(&e))?;
                        let mut outcomes = w.open_table(RUN_OUTCOME).map_err(|e| be(&e))?;
                        let mut counters = w.open_table(COUNTERS).map_err(|e| be(&e))?;
                        if let Some(prior) = outcomes
                            .get((tenant.as_str(), key.as_str()))
                            .map_err(|e| be(&e))?
                            .map(|v| {
                                let (o, ord) = v.value();
                                (o.to_owned(), ord)
                            })
                        {
                            by_outcome
                                .remove((tenant.as_str(), prior.0.as_str(), prior.1))
                                .map_err(|e| be(&e))?;
                        }
                        let counter = format!("{NEXT_CONCLUSION}/{tenant}");
                        let next = counters
                            .get(counter.as_str())
                            .map_err(|e| be(&e))?
                            .map_or(0, |v| v.value());
                        by_outcome
                            .insert((tenant.as_str(), outcome.as_str(), next), key.as_str())
                            .map_err(|e| be(&e))?;
                        outcomes
                            .insert((tenant.as_str(), key.as_str()), (outcome.as_str(), next))
                            .map_err(|e| be(&e))?;
                        counters
                            .insert(counter.as_str(), next + 1)
                            .map_err(|e| be(&e))?;
                    }

                    head = Head {
                        seq,
                        hash: record.hash,
                    };
                    sealed.push(record);
                }
                {
                    let mut index = w.open_table(RUN_WAITING).map_err(|e| be(&e))?;
                    let mut at = w.open_table(RUN_WAITING_AT).map_err(|e| be(&e))?;
                    if let Some(prior) = at
                        .remove((tenant.as_str(), key.as_str()))
                        .map_err(|e| be(&e))?
                        .map(|v| v.value())
                    {
                        index
                            .remove((tenant.as_str(), prior, key.as_str()))
                            .map_err(|e| be(&e))?;
                    }
                    if let Some(reason) = &waiting {
                        let until = reason.until().unix_timestamp();
                        let encoded = String::from_utf8(
                            crate::core::canon::to_bytes(reason)
                                .map_err(|e| StoreError::Backend(e.to_string()))?,
                        )
                        .map_err(|e| StoreError::Backend(e.to_string()))?;
                        index
                            .insert((tenant.as_str(), until, key.as_str()), encoded.as_str())
                            .map_err(|e| be(&e))?;
                        at.insert((tenant.as_str(), key.as_str()), until)
                            .map_err(|e| be(&e))?;
                    }
                }

                let updated = now_secs();
                let mut activity = w.open_table(RUN_ACTIVITY).map_err(|e| be(&e))?;
                let mut last = w.open_table(RUN_LAST_ACTIVITY).map_err(|e| be(&e))?;
                let previous = last
                    .get((tenant.as_str(), key.as_str()))
                    .map_err(|e| be(&e))?
                    .map(|value| value.value());
                if let Some(previous) = previous {
                    activity
                        .remove((tenant.as_str(), previous, key.as_str()))
                        .map_err(|e| be(&e))?;
                }
                activity
                    .insert((tenant.as_str(), updated, key.as_str()), ())
                    .map_err(|e| be(&e))?;
                last.insert((tenant.as_str(), key.as_str()), updated)
                    .map_err(|e| be(&e))?;
                // The per-producer view of the same index, moved with it.
                let mut sources = w.open_table(RUN_SOURCE).map_err(|e| be(&e))?;
                if let Some(source) = &admitted_from {
                    sources
                        .insert((tenant.as_str(), key.as_str()), source.as_str())
                        .map_err(|e| be(&e))?;
                }
                let source = sources
                    .get((tenant.as_str(), key.as_str()))
                    .map_err(|e| be(&e))?
                    .map(|value| value.value().to_owned());
                if let Some(source) = source {
                    let mut by_source = w.open_table(RUN_BY_SOURCE).map_err(|e| be(&e))?;
                    if let Some(previous) = previous {
                        by_source
                            .remove((tenant.as_str(), source.as_str(), previous, key.as_str()))
                            .map_err(|e| be(&e))?;
                    }
                    by_source
                        .insert(
                            (tenant.as_str(), source.as_str(), updated, key.as_str()),
                            (),
                        )
                        .map_err(|e| be(&e))?;
                }
                sealed
            };
            w.commit().map_err(|e| be(&e))?;
            Ok(sealed)
        })
        .await
    }

    async fn read(&self, run: RunId, from: Seq) -> Result<Vec<Record>, StoreError> {
        self.read_page(run, from, usize::MAX).await
    }

    /// The bounded read, and the one that decodes — `read` is this with no
    /// ceiling, so the two cannot come to disagree about what a record is.
    async fn read_page(
        &self,
        run: RunId,
        from: Seq,
        limit: usize,
    ) -> Result<Vec<Record>, StoreError> {
        let key = self.run_key(run);
        let upcaster = Arc::clone(&self.upcaster);
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let t = r.open_table(JOURNAL).map_err(|e| be(&e))?;
            let mut out = Vec::new();
            for entry in t
                .range((key.as_str(), from)..=(key.as_str(), u64::MAX))
                .map_err(|e| be(&e))?
                .take(limit)
            {
                let (_, v) = entry.map_err(|e| be(&e))?;
                out.push(Row::decode(v.value())?.into_record(upcaster.as_ref())?);
            }
            Ok(out)
        })
        .await
    }

    async fn case_history(
        &self,
        case: crate::core::CaseId,
        limit: usize,
    ) -> Result<Vec<Record>, StoreError> {
        let tenant = self.tenant_name();
        let case = case.to_string();
        let upcaster = Arc::clone(&self.upcaster);
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let idx = r.open_table(JOURNAL_BY_CASE).map_err(|e| be(&e))?;
            let j = r.open_table(JOURNAL).map_err(|e| be(&e))?;
            let mut out = Vec::new();
            for entry in idx
                .range(
                    (tenant.as_str(), case.as_str(), "", 0)
                        ..=(tenant.as_str(), case.as_str(), MAX_STR, u64::MAX),
                )
                .map_err(|e| be(&e))?
            {
                if out.len() >= limit {
                    break;
                }
                let (k, _) = entry.map_err(|e| be(&e))?;
                let (_, _, run, seq) = k.value();
                // The index carries the ordering; the row carries the bytes. A
                // record has one home, so the two cannot disagree about it.
                if let Some(v) = j.get((run, seq)).map_err(|e| be(&e))? {
                    out.push(Row::decode(v.value())?.into_record(upcaster.as_ref())?);
                }
            }
            Ok(out)
        })
        .await
    }

    async fn head(&self, run: RunId) -> Result<Head, StoreError> {
        let key = self.run_key(run);
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let t = r.open_table(JOURNAL).map_err(|e| be(&e))?;
            head_of(&t, &key)
        })
        .await
    }

    async fn acquire(&self, run: RunId, owner: &str, ttl: Duration) -> Result<Lease, StoreError> {
        let key = self.run_key(run);
        let owner = owner.to_owned();
        let owner_out = owner.clone();
        let upcaster = Arc::clone(&self.upcaster);
        let epoch = self
            .with_db(move |db| {
                let w = begin_write(db)?;
                let epoch = {
                    let mut leases = w.open_table(RUN_LEASE).map_err(|e| be(&e))?;
                    let now = now_secs();
                    let expires = lease_expiry(now, ttl)?;

                    let existing = leases.get(key.as_str()).map_err(|e| be(&e))?.map(|v| {
                        let (o, e, x) = v.value();
                        (o.to_owned(), e, x)
                    });

                    let epoch = match existing {
                        // No lease row. Either a genuinely fresh run — in which
                        // case the journal is empty and this is 1 — or a run
                        // whose lease table did not survive, which is what a
                        // restore leaves behind. The journal is the evidence
                        // that did survive, and every record in it carries the
                        // epoch it was written under.
                        None => epoch_after_history(&w, key.as_str(), upcaster.as_ref())?,
                        // Expired or released: claim and fence whoever held
                        // it, **including this caller**. Checked before any
                        // ownership test on purpose — a lease that has lapsed
                        // is not yours to renew, because you cannot know
                        // whether somebody took over in the gap. Assuming you
                        // were fenced is the only safe reading.
                        Some((_, epoch, expires_at)) if expires_at <= now => epoch + 1,
                        // Still live — held by anyone, this caller included.
                        // `acquire` claims and never renews: a same-owner
                        // "renewal" here is a second entry point on one
                        // instance asking to drive a run the instance is
                        // already executing, and handing it the same epoch is
                        // two executors fencing cannot tell apart. Renewal is
                        // `renew`, which proves the caller still holds the
                        // exact `(owner, epoch)` it claims to.
                        Some((held_by, epoch, expires_at)) => {
                            return Err(StoreError::LeaseHeld {
                                run: key.clone(),
                                owner: held_by,
                                epoch,
                                remaining_secs: expires_at.saturating_sub(now),
                            });
                        }
                    };

                    leases
                        .insert(key.as_str(), (owner.as_str(), epoch, expires))
                        .map_err(|e| be(&e))?;
                    epoch
                };
                w.commit().map_err(|e| be(&e))?;
                Ok(epoch)
            })
            .await?;

        Ok(Lease {
            run,
            owner: owner_out,
            epoch,
        })
    }

    async fn renew(
        &self,
        run: RunId,
        owner: &str,
        epoch: Epoch,
        ttl: Duration,
    ) -> Result<Lease, StoreError> {
        let key = self.run_key(run);
        let owner = owner.to_owned();
        let owner_out = owner.clone();
        self.with_db(move |db| {
            let w = begin_write(db)?;
            {
                let mut leases = w.open_table(RUN_LEASE).map_err(|e| be(&e))?;
                let now = now_secs();
                // Refused before the ownership check, so both backends answer
                // a bad TTL the same way whatever state the lease is in.
                let expires = lease_expiry(now, ttl)?;
                let existing = leases.get(key.as_str()).map_err(|e| be(&e))?.map(|v| {
                    let (o, e, x) = v.value();
                    (o.to_owned(), e, x)
                });
                // Held, unexpired, unreleased, by exactly `(owner, epoch)` —
                // anything else is a lease this caller has lost, and a renewal
                // never claims. A released row blanks the owner, so it fails
                // the ownership comparison; an expired row fails the expiry
                // one, because whoever claims it next takes epoch + 1 and a
                // renewal that got in first would resurrect the fenced past.
                match existing {
                    Some((held_by, held_epoch, expires_at))
                        if held_by == owner && held_epoch == epoch && expires_at > now =>
                    {
                        leases
                            .insert(key.as_str(), (owner.as_str(), epoch, expires))
                            .map_err(|e| be(&e))?;
                    }
                    _ => {
                        return Err(StoreError::LeaseNotHeld {
                            run: key.clone(),
                            epoch,
                        });
                    }
                }
            }
            w.commit().map_err(|e| be(&e))?;
            Ok(())
        })
        .await?;

        Ok(Lease {
            run,
            owner: owner_out,
            epoch,
        })
    }

    async fn release_lease(&self, run: RunId, epoch: Epoch) -> Result<(), StoreError> {
        let key = self.run_key(run);
        self.with_db(move |db| {
            let w = begin_write(db)?;
            {
                let mut leases = w.open_table(RUN_LEASE).map_err(|e| be(&e))?;
                let held = leases
                    .get(key.as_str())
                    .map_err(|e| be(&e))?
                    .map(|v| v.value().1);
                // Only the epoch that holds it may free it. A fenced caller
                // shutting down must not hand the run to a third party while
                // the instance that took over is mid-write.
                //
                // Marked expired rather than **removed**, and that distinction
                // is the whole safety of this operation. The epoch lives in
                // this row: delete it and `append` has nothing to fence
                // against, while the next `acquire` restarts at 1 — so a writer
                // already fenced at 2 outranks the new owner. Releasing must
                // free the lease without forgetting what has happened to it.
                if held == Some(epoch) {
                    leases
                        .insert(key.as_str(), ("", epoch, 0))
                        .map_err(|e| be(&e))?;
                }
            }
            w.commit().map_err(|e| be(&e))?;
            Ok(())
        })
        .await
    }

    async fn waiting_runs(&self, limit: usize) -> Result<Vec<WaitingRun>, StoreError> {
        let tenant = self.tenant.to_string();
        let prefix = format!("{}/", self.tenant);
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let index = r.open_table(RUN_WAITING).map_err(|e| be(&e))?;
            let mut out = Vec::new();
            // A range over this tenant's slice in key order, which is `until`
            // order — no sort, and no read of a row past the page. The index
            // holds one row per *currently* waiting run rather than one per
            // suspension ever recorded, so its size is the backlog an operator
            // is looking at rather than the history.
            for entry in index
                .range((tenant.as_str(), i64::MIN, "")..=(tenant.as_str(), i64::MAX, "\u{10FFFF}"))
                .map_err(|e| be(&e))?
                .take(limit)
            {
                let (k, v) = entry.map_err(|e| be(&e))?;
                let (_, _, key) = k.value();
                // Corruption rather than a skip, for the reason
                // `abandoned_runs` gives: a waiting run dropped from this
                // listing is one nobody re-arms, and this is its one page.
                let id = key.strip_prefix(prefix.as_str()).unwrap_or(key);
                let run = RunId::parse(id).map_err(|e| StoreError::Corrupt {
                    seq: 0,
                    detail: format!("run_waiting holds an unparsable run id '{id}': {e}"),
                })?;
                let reason = serde_json::from_str(v.value()).map_err(|e| StoreError::Corrupt {
                    seq: 0,
                    detail: format!("run_waiting holds an unreadable reason for {id}: {e}"),
                })?;
                out.push(WaitingRun { run, reason });
            }
            Ok(out)
        })
        .await
    }

    async fn abandoned_runs(&self, limit: usize) -> Result<Vec<RunId>, StoreError> {
        let prefix = format!("{}/", self.tenant);
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let leases = r.open_table(RUN_LEASE).map_err(|e| be(&e))?;
            let now = now_secs();
            // A full scan over this tenant's lease rows, released ones
            // included — they keep the epoch, so they cannot be deleted, and
            // an embedded store has no second index to range instead. That is
            // an accepted cost, not an oversight: the table holds one small
            // row per run ever admitted, the sweep reads it once per tick, and
            // redb walks it in memory-mapped pages. The shared-store backend
            // pays for an index because its table is shared by every instance;
            // this one is not.
            let mut expired: Vec<(u64, RunId)> = Vec::new();
            for entry in leases.range(prefix.as_str()..).map_err(|e| be(&e))? {
                let (k, v) = entry.map_err(|e| be(&e))?;
                let key = k.value();
                if !key.starts_with(prefix.as_str()) {
                    break;
                }
                let (owner, _, expires_at) = v.value();
                // Released rows blank the owner; live rows have not lapsed.
                // What remains is the set this method exists for: an owner
                // that stopped renewing without handing the run back.
                if owner.is_empty() || expires_at > now {
                    continue;
                }
                // Corruption, per the contract on `JournalStore::abandoned_runs`
                // and matching the shared-store backend: a stranded run
                // silently dropped from this listing is never recovered, so an
                // unparsable id is refused loudly rather than skipped.
                let id = &key[prefix.len()..];
                let run = RunId::parse(id).map_err(|e| StoreError::Corrupt {
                    seq: 0,
                    detail: format!("run_lease holds an unparsable run id '{id}': {e}"),
                })?;
                expired.push((expires_at, run));
            }
            // Oldest expiry first, so a bounded page cannot starve the run
            // that has been stranded longest behind fresher failures.
            expired.sort_unstable_by_key(|(at, _)| *at);
            Ok(expired.into_iter().take(limit).map(|(_, r)| r).collect())
        })
        .await
    }

    async fn seal(&self, run: RunId, epoch: Epoch, outcome: &str) -> Result<Digest, StoreError> {
        let key = self.run_key(run);
        let tenant = self.tenant.clone();
        let outcome = outcome.to_owned();
        self.with_db(move |db| {
            let w = begin_write(db)?;
            let head_hash = {
                {
                    let leases = w.open_table(RUN_LEASE).map_err(|e| be(&e))?;
                    if let Some(l) = leases.get(key.as_str()).map_err(|e| be(&e))? {
                        let (_, current, _) = l.value();
                        if epoch != current {
                            return Err(StoreError::Fenced {
                                run: key.clone(),
                                held: epoch,
                                current,
                            });
                        }
                    }
                }

                let head = {
                    let journal = w.open_table(JOURNAL).map_err(|e| be(&e))?;
                    let last = journal
                        .range((key.as_str(), 0u64)..=(key.as_str(), u64::MAX))
                        .map_err(|e| be(&e))?
                        .next_back()
                        .transpose()
                        .map_err(|e| be(&e))?
                        .map(|(_, row)| Row::decode(row.value()))
                        .transpose()?;
                    refuse_unconcluded(&key, last.as_ref().map(|row| row.body.as_slice()))?;
                    head_of(&journal, &key)?
                };

                let mut seals = w.open_table(RUN_SEAL).map_err(|e| be(&e))?;
                // First seal wins; a re-seal must not rewrite the outcome.
                if seals.get(key.as_str()).map_err(|e| be(&e))?.is_none() {
                    seals
                        .insert(
                            key.as_str(),
                            (
                                outcome.as_str(),
                                head.hash.as_bytes().as_slice(),
                                now_secs(),
                            ),
                        )
                        .map_err(|e| be(&e))?;

                    // Enter the log in the same transaction that seals: a seal
                    // the checkpoint does not commit to is the hole this closes.
                    // The counter is per tenant too. A shared counter would
                    // interleave two tenants' indices in one sequence, and the
                    // log would no longer be dense for either.
                    let counter = format!("{NEXT_LOG_INDEX}/{tenant}");
                    let mut counters = w.open_table(COUNTERS).map_err(|e| be(&e))?;
                    let next = counters
                        .get(counter.as_str())
                        .map_err(|e| be(&e))?
                        .map_or(0, |v| v.value());
                    w.open_table(SEAL_LOG)
                        .map_err(|e| be(&e))?
                        .insert((tenant.as_str(), next), key.as_str())
                        .map_err(|e| be(&e))?;
                    // The outcome index is *not* written here: it derives from
                    // the `RunSealed` record inside `append`, where the last
                    // conclusion wins. The seal freezes the journal and enters
                    // the log — it is not the outcome's home.
                    counters
                        .insert(counter.as_str(), next + 1)
                        .map_err(|e| be(&e))?;
                }
                head.hash
            };
            w.commit().map_err(|e| be(&e))?;
            Ok(head_hash)
        })
        .await
    }

    async fn admitted_as(&self, key: &str) -> Result<Option<RunId>, StoreError> {
        let tenant = self.tenant_name();
        let key = key.to_owned();
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let admissions = match r.open_table(RUN_ADMISSION) {
                Ok(t) => t,
                // A store that has never admitted under a key has no table
                // yet, and "no table" is the same fact as "no such key".
                Err(redb::TableError::TableDoesNotExist(_)) => return Ok(None),
                Err(e) => return Err(be(&e)),
            };
            let Some(held) = admissions
                .get((tenant.as_str(), key.as_str()))
                .map_err(|e| be(&e))?
                .map(|v| v.value().0.to_owned())
            else {
                return Ok(None);
            };
            RunId::parse(&held)
                .map(Some)
                .map_err(|e| StoreError::Corrupt {
                    seq: 0,
                    detail: format!("run_admission holds an unparseable run id '{held}': {e}"),
                })
        })
        .await
    }

    async fn forget_admissions(
        &self,
        older_than: crate::core::Timestamp,
    ) -> Result<usize, StoreError> {
        let tenant = self.tenant_name();
        let cutoff = older_than.unix_timestamp();
        self.with_db(move |db| {
            let w = begin_write(db)?;
            let removed = {
                let mut admissions = match w.open_table(RUN_ADMISSION) {
                    Ok(t) => t,
                    // Nothing has ever been admitted under a key here, so there
                    // is nothing to retire.
                    Err(redb::TableError::TableDoesNotExist(_)) => return Ok(0),
                    Err(e) => return Err(be(&e)),
                };
                // Collected before removing: redb will not hand out a mutable
                // borrow while a range iterator over the same table is live.
                let stale: Vec<String> = admissions
                    .range((tenant.as_str(), "")..=(tenant.as_str(), MAX_STR))
                    .map_err(|e| be(&e))?
                    .filter_map(|entry| {
                        let (k, v) = entry.ok()?;
                        (v.value().1 < cutoff).then(|| k.value().1.to_owned())
                    })
                    .collect();
                for key in &stale {
                    admissions
                        .remove((tenant.as_str(), key.as_str()))
                        .map_err(|e| be(&e))?;
                }
                stale.len()
            };
            w.commit().map_err(|e| be(&e))?;
            Ok(removed)
        })
        .await
    }

    async fn runs_by_outcome(&self, outcome: &str, limit: usize) -> Result<Vec<RunId>, StoreError> {
        let tenant = self.tenant_name();
        let outcome = outcome.to_owned();
        let prefix = format!("{tenant}/");
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let idx = r.open_table(RUN_BY_OUTCOME).map_err(|e| be(&e))?;
            let mut out = Vec::new();
            // **Newest first**, which is what makes the bound survivable. See
            // `JournalStore::runs_by_outcome`: ascending order plus a page
            // limit means a plane whose backlog already exceeds one page
            // returns the *same* runs forever, and the quarantine that just
            // happened is the one that never appears.
            for entry in idx
                .range(
                    (tenant.as_str(), outcome.as_str(), 0)
                        ..=(tenant.as_str(), outcome.as_str(), u64::MAX),
                )
                .map_err(|e| be(&e))?
                .rev()
            {
                if out.len() >= limit {
                    break;
                }
                let (_, v) = entry.map_err(|e| be(&e))?;
                // Keys are `tenant/run`; the caller asked for runs. An entry
                // that does not carry this tenant's prefix, or whose id does
                // not parse, is corruption — reported rather than skipped,
                // matching the shared-store backend, because a quarantined run
                // silently thinned out of this page is the unreachable-signal
                // failure the method exists to remove.
                let key = v.value();
                let id = key
                    .strip_prefix(prefix.as_str())
                    .ok_or_else(|| StoreError::Corrupt {
                        seq: 0,
                        detail: format!(
                            "run_by_outcome points at '{key}', which is outside tenant '{tenant}'"
                        ),
                    })?;
                let run = RunId::parse(id).map_err(|e| StoreError::Corrupt {
                    seq: 0,
                    detail: format!("run_by_outcome holds an unparsable run id '{id}': {e}"),
                })?;
                out.push(run);
            }
            Ok(out)
        })
        .await
    }

    async fn count_by_outcome(&self, outcome: &str) -> Result<u64, StoreError> {
        let tenant = self.tenant_name();
        let outcome = outcome.to_owned();
        let prefix = format!("{tenant}/");
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let idx = r.open_table(RUN_BY_OUTCOME).map_err(|e| be(&e))?;
            let mut n: u64 = 0;
            // The whole range, never a page of it: a gauge read from a bounded
            // listing flattens at the page size exactly when it becomes a
            // backlog.
            for entry in idx
                .range(
                    (tenant.as_str(), outcome.as_str(), 0)
                        ..=(tenant.as_str(), outcome.as_str(), u64::MAX),
                )
                .map_err(|e| be(&e))?
            {
                let (_, v) = entry.map_err(|e| be(&e))?;
                // Counted only once it is proven to be this tenant's row. An
                // entry the listing would refuse must not be quietly included
                // in the number the listing is supposed to explain.
                let key = v.value();
                if key.strip_prefix(prefix.as_str()).is_none() {
                    return Err(StoreError::Corrupt {
                        seq: 0,
                        detail: format!(
                            "run_by_outcome points at '{key}', which is outside tenant '{tenant}'"
                        ),
                    });
                }
                n += 1;
            }
            Ok(n)
        })
        .await
    }

    async fn recent_runs(
        &self,
        after: Option<(u64, RunId)>,
        limit: usize,
    ) -> Result<Vec<(RunId, u64)>, StoreError> {
        let tenant = self.tenant_name();
        let prefix = format!("{tenant}/");
        // The cursor's own key, so the range can end *below* it. Skipping in the
        // iterator instead would still walk every newer row, which is the cost
        // this page bound exists to remove.
        let end = after.map(|(updated, run)| (updated, format!("{prefix}{run}")));
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let Ok(activity) = r.open_table(RUN_ACTIVITY) else {
                return Ok(Vec::new());
            };
            // Exclusive upper bound at the cursor, so the row the caller last
            // saw is not served twice.
            let rows = match &end {
                Some((updated, key)) => activity
                    .range((tenant.as_str(), 0, "")..(tenant.as_str(), *updated, key.as_str()))
                    .map_err(|e| be(&e))?,
                None => activity
                    .range((tenant.as_str(), 0, "")..=(tenant.as_str(), u64::MAX, MAX_STR))
                    .map_err(|e| be(&e))?,
            };
            rows.rev()
                .filter_map(|entry| match entry {
                    Ok((key, _)) => {
                        let (_, updated, stored) = key.value();
                        stored
                            .strip_prefix(prefix.as_str())
                            .and_then(|id| RunId::parse(id).ok())
                            .map(|run| Ok((run, updated)))
                    }
                    Err(error) => Some(Err(be(&error))),
                })
                .take(limit)
                .collect()
        })
        .await
    }

    async fn runs_by_id(
        &self,
        after: Option<RunId>,
        limit: usize,
    ) -> Result<Vec<RunId>, StoreError> {
        use std::ops::Bound;
        let tenant = self.tenant_name();
        let prefix = format!("{tenant}/");
        let start = after.map(|run| format!("{prefix}{run}"));
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let Ok(last) = r.open_table(RUN_LAST_ACTIVITY) else {
                return Ok(Vec::new());
            };
            // Strictly after the cursor, so the run last served is not served twice.
            let from = match &start {
                Some(key) => Bound::Excluded((tenant.as_str(), key.as_str())),
                None => Bound::Included((tenant.as_str(), "")),
            };
            last.range((from, Bound::Included((tenant.as_str(), MAX_STR))))
                .map_err(|e| be(&e))?
                .filter_map(|entry| match entry {
                    Ok((key, _)) => {
                        let (_, stored) = key.value();
                        stored
                            .strip_prefix(prefix.as_str())
                            .and_then(|id| RunId::parse(id).ok())
                            .map(Ok)
                    }
                    Err(error) => Some(Err(be(&error))),
                })
                .take(limit)
                .collect()
        })
        .await
    }

    async fn recent_runs_from(
        &self,
        source: &str,
        after: Option<(u64, RunId)>,
        limit: usize,
    ) -> Result<Vec<(RunId, u64)>, StoreError> {
        let tenant = self.tenant_name();
        let source = source.to_owned();
        let prefix = format!("{tenant}/");
        let end = after.map(|(updated, run)| (updated, format!("{prefix}{run}")));
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let by_source = match r.open_table(RUN_BY_SOURCE) {
                Ok(t) => t,
                Err(redb::TableError::TableDoesNotExist(_)) => return Ok(Vec::new()),
                Err(e) => return Err(be(&e)),
            };
            let (t, s) = (tenant.as_str(), source.as_str());
            // Exclusive upper bound at the cursor, as `recent_runs` has it.
            let rows = match &end {
                Some((updated, key)) => by_source
                    .range((t, s, 0, "")..(t, s, *updated, key.as_str()))
                    .map_err(|e| be(&e))?,
                None => by_source
                    .range((t, s, 0, "")..=(t, s, u64::MAX, MAX_STR))
                    .map_err(|e| be(&e))?,
            };
            rows.rev()
                .filter_map(|entry| match entry {
                    Ok((key, _)) => {
                        let (_, _, updated, stored) = key.value();
                        stored
                            .strip_prefix(prefix.as_str())
                            .and_then(|id| RunId::parse(id).ok())
                            .map(|run| Ok((run, updated)))
                    }
                    Err(error) => Some(Err(be(&error))),
                })
                .take(limit)
                .collect()
        })
        .await
    }

    async fn checkpoint(&self) -> Result<crate::journal::Checkpoint, StoreError> {
        let leaves = self.log_leaves().await?;
        Ok(crate::journal::Checkpoint {
            origin: self.log_origin(),
            size: leaves.len() as u64,
            root: crate::core::merkle::root(&leaves),
        })
    }

    async fn consistency_proof(&self, old_size: u64) -> Result<Vec<Digest>, StoreError> {
        let leaves = self.log_leaves().await?;
        let old = usize::try_from(old_size).unwrap_or(usize::MAX);
        if old > leaves.len() {
            // Refused rather than answered with an empty proof: an empty proof
            // is what a *consistent* log of unchanged size returns, so handing
            // one back here would let "your checkpoint is from the future" read
            // as "everything is fine".
            return Err(StoreError::Backend(format!(
                "a checkpoint of size {old_size} is larger than this log ({}) — \
                 either it belongs to another plane, or runs were removed",
                leaves.len()
            )));
        }
        Ok(crate::core::merkle::consistency_proof(&leaves, old))
    }

    async fn inclusion_proof(
        &self,
        run: RunId,
    ) -> Result<Option<crate::journal::Inclusion>, StoreError> {
        let leaves = self.log_leaves().await?;
        let Some((index, seal)) = self.positions(&[run]).await?.pop().flatten() else {
            return Ok(None);
        };
        Ok(Some(crate::journal::Inclusion {
            index,
            size: leaves.len() as u64,
            seal,
            proof: crate::core::merkle::inclusion_proof(
                &leaves,
                usize::try_from(index).unwrap_or(usize::MAX),
            ),
        }))
    }

    async fn inclusion_proof_at(
        &self,
        run: RunId,
        size: u64,
    ) -> Result<Option<crate::journal::Inclusion>, StoreError> {
        let leaves = self.log_leaves().await?;
        let prefix = usize::try_from(size)
            .ok()
            .and_then(|size| leaves.get(..size))
            .ok_or_else(|| {
                StoreError::Backend(format!(
                    "asked to prove at size {size} and the log holds {} leaves",
                    leaves.len()
                ))
            })?;
        let Some((index, seal)) = self.positions(&[run]).await?.pop().flatten() else {
            return Ok(None);
        };
        let Some(at) = usize::try_from(index).ok().filter(|&at| at < prefix.len()) else {
            return Ok(None);
        };
        Ok(Some(crate::journal::Inclusion {
            index,
            size,
            seal,
            proof: crate::core::merkle::inclusion_proof(prefix, at),
        }))
    }

    async fn log_positions(
        &self,
        runs: &[RunId],
    ) -> Result<Vec<Option<(u64, Digest)>>, StoreError> {
        self.positions(runs).await
    }

    async fn request_cancel(
        &self,
        run: RunId,
        actor: &crate::core::Operator,
        reason: &str,
    ) -> Result<bool, StoreError> {
        let key = self.run_key(run);
        let (who, basis, reason) = (
            actor.actor().to_owned(),
            actor.basis().as_str(),
            reason.to_owned(),
        );
        self.with_db(move |db| {
            let w = begin_write(db)?;
            let first = {
                let mut t = w.open_table(RUN_CANCEL).map_err(|e| be(&e))?;
                // The first asker stays on the record; a retry must not rewrite
                // who intervened.
                if t.get(key.as_str()).map_err(|e| be(&e))?.is_some() {
                    false
                } else {
                    t.insert(
                        key.as_str(),
                        (who.as_str(), basis, reason.as_str(), now_secs()),
                    )
                    .map_err(|e| be(&e))?;
                    true
                }
            };
            w.commit().map_err(|e| be(&e))?;
            Ok(first)
        })
        .await
    }

    async fn cancellation(&self, run: RunId) -> Result<Option<Cancellation>, StoreError> {
        let key = self.run_key(run);
        self.with_db(move |db| {
            let r = db.begin_read().map_err(|e| be(&e))?;
            let t = r.open_table(RUN_CANCEL).map_err(|e| be(&e))?;
            t.get(key.as_str())
                .map_err(|e| be(&e))?
                .map(|v| {
                    let (who, basis, reason, _) = v.value();
                    Ok(Cancellation {
                        actor: super::decode_operator(who, basis, "run_cancel")?,
                        reason: reason.to_owned(),
                    })
                })
                .transpose()
        })
        .await
    }
}

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

    fn row(signed: bool) -> Vec<u8> {
        Row {
            body: b"{}".to_vec(),
            prev_hash: [1; 32],
            hash: [2; 32],
            key_id: signed.then(|| "k".to_owned()),
            signature: signed.then(|| vec![3; 64]),
        }
        .encode()
        .expect("encodes")
    }

    /// The signature flag is a boolean byte: any other value is a damaged
    /// row, not an unsigned record. Read as unsigned, a damaged flag strips
    /// a signature silently — the record reads back as history written
    /// before signing was configured.
    ///
    /// Flipped on an *unsigned* row, so the rest of the row still decodes
    /// under either reading and only the flag rule can refuse it.
    #[test]
    fn a_signature_flag_other_than_zero_or_one_is_corrupt() {
        let mut row = row(false);
        assert!(Row::decode(&row).is_ok_and(|r| r.signature.is_none()));
        row[64] = 2;
        assert!(
            matches!(Row::decode(&row), Err(StoreError::Corrupt { .. })),
            "a flag byte of 2 decoded as an unsigned row"
        );
    }

    /// Bytes past the body are a damaged row, not padding to ignore.
    #[test]
    fn trailing_bytes_after_a_journal_row_are_corrupt() {
        let mut row = row(true);
        assert!(Row::decode(&row).is_ok_and(|r| r.signature.is_some()));
        row.push(0);
        assert!(
            matches!(Row::decode(&row), Err(StoreError::Corrupt { .. })),
            "a row with bytes after its body decoded as whole"
        );
    }

    /// `origin` and `for_tenant` compose the same name whatever order they are
    /// called in, and calling either twice changes nothing.
    ///
    /// Baking the origin into a field at `for_tenant` time is what breaks it:
    /// `for_tenant` twice double-qualifies the name, and `origin` afterwards
    /// silently drops the tenant — either way a checkpoint published under a
    /// name no verifier would ever see again.
    #[tokio::test]
    async fn checkpoint_origin_is_order_insensitive_and_idempotent() {
        let tenant = crate::core::TenantId::new("acme").expect("tenant");

        let tenant_then_origin = RedbStore::open_in_memory()
            .expect("store")
            .for_tenant(tenant.clone())
            .origin("plane-1");
        let origin_then_tenant = RedbStore::open_in_memory()
            .expect("store")
            .origin("plane-1")
            .for_tenant(tenant.clone());
        let twice = RedbStore::open_in_memory()
            .expect("store")
            .origin("plane-1")
            .for_tenant(tenant.clone())
            .for_tenant(tenant);

        let a = tenant_then_origin.checkpoint().await.expect("checkpoint");
        let b = origin_then_tenant.checkpoint().await.expect("checkpoint");
        let c = twice.checkpoint().await.expect("checkpoint");
        assert_eq!(a.origin, "plane-1/acme");
        assert_eq!(b.origin, a.origin, "builder order changed the log's name");
        assert_eq!(c.origin, a.origin, "for_tenant is not idempotent");
    }

    /// A stored run id that does not parse is reported as corruption, not
    /// silently thinned out of the listing.
    ///
    /// In here rather than in the integration suite because planting the
    /// corrupt row takes the private tables: no public write path produces
    /// one, which is exactly why a skip would hide it forever. The two
    /// listings this pins are the two whose silent thinning is worst — a
    /// stranded run dropped from `abandoned_runs` is never recovered, and a
    /// quarantined run dropped from `runs_by_outcome` is a detected failure
    /// whose signal reaches nobody.
    ///
    /// Each has a positive half first, so the error genuinely comes from the
    /// garbage row rather than from a scan that refuses everything.
    #[tokio::test]
    async fn an_unparsable_run_id_is_reported_as_corruption_not_skipped() {
        let store = RedbStore::open_in_memory().expect("store");
        let stranded = RunId::generate();
        let good_key = store.run_key(stranded);

        // A legitimate expired lease and a legitimate outcome row: the
        // positive halves.
        store
            .with_db({
                let good_key = good_key.clone();
                move |db| {
                    let w = begin_write(db)?;
                    {
                        w.open_table(RUN_LEASE)
                            .map_err(|e| be(&e))?
                            .insert(good_key.as_str(), ("worker", 1u64, 0u64))
                            .map_err(|e| be(&e))?;
                        w.open_table(RUN_BY_OUTCOME)
                            .map_err(|e| be(&e))?
                            .insert(("default", "failed", 0u64), good_key.as_str())
                            .map_err(|e| be(&e))?;
                    }
                    w.commit().map_err(|e| be(&e))?;
                    Ok(())
                }
            })
            .await
            .expect("plant the healthy rows");
        assert_eq!(
            store.abandoned_runs(10).await.expect("a clean scan"),
            vec![stranded]
        );
        assert_eq!(
            store
                .runs_by_outcome("failed", 10)
                .await
                .expect("a clean scan"),
            vec![stranded]
        );

        // The garbage no correct writer produces.
        store
            .with_db(|db| {
                let w = begin_write(db)?;
                {
                    w.open_table(RUN_LEASE)
                        .map_err(|e| be(&e))?
                        .insert("default/not-a-run-id", ("worker", 1u64, 0u64))
                        .map_err(|e| be(&e))?;
                    w.open_table(RUN_BY_OUTCOME)
                        .map_err(|e| be(&e))?
                        .insert(("default", "failed", 1u64), "default/not-a-run-id")
                        .map_err(|e| be(&e))?;
                }
                w.commit().map_err(|e| be(&e))?;
                Ok(())
            })
            .await
            .expect("plant the corrupt rows");

        let err = store
            .abandoned_runs(10)
            .await
            .expect_err("an unparsable lease row must refuse the sweep, not vanish from it");
        assert!(
            matches!(err, StoreError::Corrupt { .. }),
            "the refusal must be Corrupt, so it is promoted past retry logic: {err}"
        );
        let err = store
            .runs_by_outcome("failed", 10)
            .await
            .expect_err("an unparsable outcome row must refuse the listing, not vanish from it");
        assert!(
            matches!(err, StoreError::Corrupt { .. }),
            "the refusal must be Corrupt, so it is promoted past retry logic: {err}"
        );
    }
}