agentplane 0.57.0

//! Content-addressed bytes, kept out of the chain.
//!
//! The journal refuses a record over [`Record::MAX_RECORD_BYTES`], and this is
//! the other half of that refusal: somewhere for the bytes to go. The pattern is
//! the field's — Temporal calls it a claim check, offloading payloads above a
//! threshold and passing a reference through the event history instead — with
//! one difference that matters here.
//!
//! **The reference is the digest.** Temporal's token identifies a payload;
//! a digest *is* the payload's identity. So the hash chain still commits to the
//! exact bytes even though it does not contain them: an auditor who fetches a
//! blob can check it against the digest the chain already signed, and a swapped
//! blob is as detectable as a rewritten record. A reference that merely pointed
//! at mutable storage would move the tamper-evidence boundary without saying so.
//!
//! Three properties follow, and each is a rule rather than a nicety:
//!
//! * **The store computes the digest, never the caller.** A caller who supplied
//!   both bytes and digest could supply a pair that does not match, and every
//!   later verification would compare a blob against a claim rather than a fact.
//! * **Reads verify before returning.** Storage is the least trusted thing here
//!   — it is the part an operator can reach with a text editor.
//! * **Writes are idempotent by construction.** Same bytes, same address; there
//!   is nothing to race and no transaction to need. That is precisely why an
//!   object store is the right shape for this and the wrong shape for the
//!   journal, which needs ordered scans and multi-key atomicity.
//!
//! [`Record::MAX_RECORD_BYTES`]: crate::journal::Record::MAX_RECORD_BYTES

use std::fmt::Debug;

use async_trait::async_trait;

use crate::core::{Digest, Timestamp};

#[cfg(feature = "opendal")]
mod opendal_store;
#[cfg(feature = "opendal")]
pub use opendal_store::{OpenDalBlobs, TOMBSTONE_FORMAT_VERSION};

mod memory;
pub use memory::MemoryBlobs;

mod scoped;
pub use scoped::{ScopedBlobs, unit_address};

/// What can go wrong reaching content-addressed storage.
#[derive(Debug, thiserror::Error)]
pub enum BlobError {
    /// The backing store failed.
    #[error("blob storage: {0}")]
    Backend(String),

    /// Nothing is stored at that address.
    ///
    /// Distinct from a corrupt read on purpose: a missing blob is a retention or
    /// configuration problem, a corrupt one is an integrity problem, and the
    /// second is the one somebody has to be paged about.
    #[error("no blob at {0}")]
    NotFound(String),

    /// The bytes do not hash to the address they were fetched from.
    ///
    /// Someone or something changed them after they were written. Reported
    /// rather than returned-with-a-warning for the same reason a broken hash
    /// chain is: content you cannot trust is worse than content you do not have,
    /// because it is used.
    #[error("blob at {expected} hashes to {actual} — the stored bytes were altered")]
    Corrupt { expected: String, actual: String },

    /// The bytes were deliberately expired, and a tombstone says so.
    ///
    /// The whole reason retention needs a distinct answer rather than reusing
    /// [`NotFound`](Self::NotFound). Three states an operator must be able to
    /// tell apart, and only one of them is an incident:
    ///
    /// | | means |
    /// |---|---|
    /// | `NotFound` | nothing was ever here, or it was lost. Investigate. |
    /// | `Expired` | retention did its job on a stated date, for a stated reason. |
    /// | `Corrupt` | the bytes were altered. Page someone. |
    ///
    /// Collapsing the middle case into the first is what makes an erasure
    /// request indistinguishable from data loss six months later — and the
    /// journal cannot settle it, because the journal deliberately never held
    /// the bytes.
    #[error("blob at {digest} was expired at {at}: {reason}")]
    Expired {
        digest: String,
        at: i64,
        reason: String,
    },

    /// The bytes are gone and the record of *why* cannot be read.
    ///
    /// A fourth state, and the one a lax reader hides. A tombstone is the only
    /// evidence an erasure happened that outlives the bytes it describes, so a
    /// reader that filled in a default — epoch zero, the reason "expired" —
    /// would answer [`Expired`](Self::Expired) with a date and a reason it made
    /// up, and a compliance drill would count a completed erasure it never saw.
    ///
    /// Reported as a finding rather than as a backend fault: the store *was*
    /// reached, and what came back was unreadable.
    #[error("blob at {digest}: the bytes are gone and their tombstone does not read ({detail})")]
    UnreadableTombstone { digest: String, detail: String },

    /// Sealed bytes are stored there and did not open, for a cause that is
    /// neither an erasure nor damage.
    ///
    /// A retired wrapping-key version, an envelope written by a build that
    /// reads another construction, or a header this build cannot parse. The
    /// first two are reversible — lower the key service's version floor, run
    /// the other build — and the third cannot be told apart from damage here,
    /// which the detail says. Folded into [`Corrupt`](Self::Corrupt) it pages
    /// somebody about tampering; folded into [`Backend`](Self::Backend) it
    /// reads as an outage that retrying will clear.
    #[error(
        "blob at {digest}: the sealed bytes did not open, and no erasure explains it: {detail}"
    )]
    Unopened { digest: String, detail: String },
}

/// Why an erasure did not happen.
///
/// Separate from [`BlobError`], whose every variant describes something that
/// already happened to some bytes. An erasure verb is not a blob operation — it
/// walks a case, tombstones each blob and destroys a key scope — and it can
/// refuse outright, which is not a state any reader of a blob can observe.
#[derive(Debug, thiserror::Error)]
pub enum EraseError {
    /// The matter is under a legal hold, so **nothing was destroyed**.
    ///
    /// A refusal that is the control working rather than a fault: no tombstone
    /// was written and no key destroyed, because somebody outside the plane
    /// said this matter must be preserved. The reason travels
    /// with it so whoever reads the report can say on whose instruction the data
    /// is still there.
    #[error(
        "case {case} is under a legal hold placed at {placed_at}, so nothing was erased: {reason}"
    )]
    UnderLegalHold {
        case: String,
        placed_at: String,
        reason: String,
    },

    /// The matter is still open, so **nothing was destroyed**.
    ///
    /// The second refusal, and it protects the plane rather than the data
    /// subject. Destroying a key freezes the runs it covers: their plans and
    /// effect outputs stop opening, so none of them can be replayed, resumed
    /// or unwound again — and a run left mid-flight by an erasure is one an
    /// operator can only abandon. Erasing under live work is therefore a
    /// decision about the *work*, and it is the operator's to make
    /// deliberately: conclude the runs, then erase.
    ///
    /// The rule is not new; where it lived was the defect. A retention pass
    /// selects closed cases only and says why, and everything else about
    /// erasure ran on the honour system of that one caller
    /// — including the path an Article 17 request actually takes, which names
    /// a matter rather than a window.
    #[error(
        "case {case} is {status} rather than closed, so nothing was erased: erasing the \
         data under a live run leaves work that can no longer be replayed or unwound. \
         Conclude the case's runs and close it, then erase"
    )]
    CaseStillOpen { case: String, status: String },

    /// A run of the matter has not concluded, so **nothing was destroyed**.
    ///
    /// A closed case is not a case with no live work: a run suspended on a
    /// wait, quarantined for a person, or never concluded is still a run the
    /// plane may resume, and destroying the key under it leaves work that
    /// can no longer be replayed or unwound — the outage
    /// [`CaseStillOpen`](Self::CaseStillOpen) refuses, reached through a
    /// status that does not describe it. Only a conclusion nothing resumes
    /// from ([`SEALED_OUTCOMES`](crate::runtime::SEALED_OUTCOMES)) counts.
    #[error(
        "run {run} {state}, so nothing was erased: erasing the data under a run that may \
         still resume leaves work that can no longer be replayed or unwound. Conclude the \
         run, then erase"
    )]
    RunNotConcluded { run: String, state: String },

    /// The run belongs to a case, so **nothing was destroyed**.
    ///
    /// A case-bound run's payloads are sealed under its case's scope, so
    /// destroying the run's own scope reaches none of them while answering as
    /// if it had. The erasure unit for that run is its case.
    #[error(
        "run {run} belongs to case {case}, so nothing was erased: its payloads are sealed \
         under the case, and erasing the run's own scope reaches none of them. Erase the case"
    )]
    RunBelongsToCase { run: String, case: String },

    /// A tombstone could not be written.
    #[error(transparent)]
    Blob(#[from] BlobError),

    /// The case layer or the key ring could not be reached.
    #[error(transparent)]
    Store(#[from] crate::core::StoreError),
}

/// Bytes addressed by their own hash.
///
/// # Two roles, and only one of them answers every method
///
/// Most of this trait is the contract every implementation carries: `put`,
/// `get`, `expire`, `has`, and the rule that an expired address stays expired.
///
/// [`put_at`](Self::put_at) and [`get_raw`](Self::get_raw) are the **envelope
/// pair**, and they belong to a *backing* store — one something may seal onto.
/// A sealing decorator refuses both on purpose
/// ([`EncryptedBlobs`](crate::keyring::EncryptedBlobs) does), because exposing
/// them through it is an unsealed side door: a caller reaching `put_at` on a
/// deployment that asked for sealing writes plaintext under a scope whose
/// erasure can never reach it. So a store that refuses the pair is not
/// incomplete, and a battery demanding it of everything would fail the one
/// implementation whose refusal is the guarantee — which is why
/// `testkit::conformance_blob` has two entry points rather than one.
///
/// Stated here rather than only at the decorator that refuses, because the
/// reader who needs it is writing the *next* implementation.
#[async_trait]
pub trait BlobStore: Send + Sync + Debug {
    /// Whose bytes this handle can reach.
    ///
    /// Content addressing makes two tenants writing identical bytes land on one
    /// object, which is a feature within a tenant and a defect across them. The
    /// severe half is not reading — payloads are sealed under a per-tenant data
    /// key — it is **erasure**: tombstoning a shared object destroys the other
    /// tenant's data while discharging one tenant's request, and reports success
    /// for both.
    ///
    /// Reported so a plane can refuse a blob store scoped to a different tenant
    /// than itself, the same way it refuses a mismatched journal.
    fn tenant(&self) -> &str {
        crate::core::TenantId::DEFAULT
    }

    /// Store bytes and return the address they landed at.
    ///
    /// The returned digest MUST be [`Digest::of`] of exactly the bytes given.
    /// That is a contract, not a description: callers that case-link before
    /// writing — governed media,
    /// [`store_blob`](crate::runtime::StepCtx::store_blob) — commit erasure
    /// traversal to the digest they computed, so a store answering with any
    /// other address would file the bytes *outside* that traversal, linked
    /// under one digest and stored under another where no erasure that follows
    /// the link can reach them. The media ingest treats a mismatch as a broken
    /// contract and fails the effect rather than prefer the store's answer.
    /// Envelope encryption keeps the contract true by addressing ciphertext at
    /// the **plaintext** digest through [`put_at`](Self::put_at).
    ///
    /// Writing the same bytes twice is the same write.
    ///
    /// **An expired address stays expired.** A store MUST refuse a write to an
    /// address it holds a tombstone for, with
    /// [`BlobError::Expired`]. Content addressing makes
    /// this the one rule that is not obvious: the address *is* the bytes, so a
    /// later write of the same bytes lands on the erased object and puts the
    /// data back — silently, under a tombstone that still says when and why it
    /// went. An erasure a subsequent write reverses is worse than one that
    /// never ran, because the first was reported as discharged. It is reachable
    /// through the ordinary API: a resumed run re-storing what it stored
    /// before, or a second run of the same matter doing the same work.
    ///
    /// A *different* erasure unit writing the same bytes is unaffected —
    /// [`ScopedBlobs`] gives it another address, which is what that type is
    /// for.
    ///
    /// # Errors
    ///
    /// [`BlobError::Expired`] if the address holds a tombstone, or
    /// [`BlobError::Backend`] if the backing store rejects the write.
    async fn put(&self, bytes: &[u8]) -> Result<Digest, BlobError>;

    /// Store bytes at an address that is **not** their own digest.
    ///
    /// The one legitimate reason to separate the two: envelope encryption, where
    /// a payload is addressed by the digest of the plaintext and stored as
    /// ciphertext. Every digest already written to a journal keeps meaning what
    /// it meant, and the encryption stays invisible to everything that only ever
    /// held an address.
    ///
    /// Callers other than [`EncryptedBlobs`](crate::keyring::EncryptedBlobs)
    /// almost certainly want [`put`](Self::put): an address that does not
    /// describe its contents is a content-addressed store with its defining
    /// property switched off, and [`get`](Self::get) can no longer verify.
    ///
    /// Carries [`put`](Self::put)'s tombstone rule, and needs it more: this is
    /// the write path a sealed deployment takes, so the refusal has to be here
    /// or sealing is the configuration that loses it.
    ///
    /// # Errors
    ///
    /// [`BlobError::Expired`] if the address holds a tombstone, or
    /// [`BlobError::Backend`] if the backing store rejects the write.
    async fn put_at(&self, digest: Digest, bytes: &[u8]) -> Result<(), BlobError>;

    /// Fetch exactly what is stored, without verifying it against the address.
    ///
    /// The counterpart to [`put_at`](Self::put_at): what is stored there is an
    /// envelope, so it does not hash to the address and the ordinary check would
    /// reject it. Verification does not disappear — it moves to after the
    /// envelope is opened, where it is a claim about the payload rather than
    /// about the envelope.
    ///
    /// # Errors
    ///
    /// [`BlobError::NotFound`] if nothing is stored there,
    /// [`BlobError::Expired`] if a tombstone says the bytes were erased, and
    /// [`BlobError::UnreadableTombstone`] if one is there and does not read.
    async fn get_raw(&self, digest: Digest) -> Result<Vec<u8>, BlobError>;

    /// Fetch bytes, verifying them against the address before returning.
    ///
    /// # Errors
    ///
    /// [`BlobError::NotFound`] if nothing is stored there,
    /// [`BlobError::Corrupt`] if what is stored does not hash to `digest`,
    /// [`BlobError::Expired`] if a tombstone says the bytes were erased, and
    /// [`BlobError::UnreadableTombstone`] if one is there and does not read.
    async fn get(&self, digest: Digest) -> Result<Vec<u8>, BlobError>;

    /// Drop a blob's bytes, leaving a tombstone that says it was deliberate.
    ///
    /// This is the erasure half of retention, and it works *because* the chain
    /// only ever committed to a digest: the record still proves what happened
    /// and that it was not altered, while the bytes it described are gone. That
    /// is the property an Article 17 request needs and an Article 12 obligation
    /// must survive — they are only in tension if the payload lives in the
    /// chain, which is why it does not.
    ///
    /// Expiring twice is the same expiry; the first tombstone stands, so a
    /// retry cannot rewrite when or why the data went.
    ///
    /// # Errors
    ///
    /// If the backing store rejects the write.
    async fn expire(&self, digest: Digest, at: Timestamp, reason: &str) -> Result<(), BlobError>;

    /// Whether anything is stored at that address.
    ///
    /// Does not verify: this answers a retention question, and a caller who
    /// needs to trust the bytes must read them.
    ///
    /// # Errors
    ///
    /// If the backing store cannot be reached.
    async fn has(&self, digest: Digest) -> Result<bool, BlobError>;
}

/// Expire every blob a case produced.
///
/// The erasure unit, because a case is what a request actually names — nobody
/// asks to forget a digest. Each blob is tombstoned with the same reason, so a
/// later read says *expired, on this date, for this reason* rather than
/// *missing*, and the journal still proves what happened.
///
/// `tenant` derives the addresses to tombstone: blobs live at
/// [`unit_address`]`(erasure_scope(tenant, case), digest)`, so this erasure
/// reaches exactly this case's copies — [`ScopedBlobs`] carries the argument
/// for why the erasure unit leads the address.
///
/// Returns how many blobs were expired — zero is an ordinary answer — and
/// every disclosure of the case or its runs that `disclosures` holds.
///
/// What this does **not** touch is the journal's records: they are
/// append-only, so personal data written into one cannot be removed — keep it
/// out of records rather than expecting erasure to reach it.
///
/// `blobs` is optional because the erasure unit is the **key scope**: a plane
/// that seals with a key ring and stores no blobs still erases by the key
/// destruction, and a later drill reads any linked bytes as *erased by design*
/// through the key rather than through a tombstone.
///
/// # What is decided before anything is destroyed
///
/// * A held or unclosed case is refused as such before its runs are read:
///   the hold names on whose instruction the matter stays, and closing it is
///   the operator's next step, so either is the more useful answer.
/// * Every run of the case has concluded under an outcome nothing resumes
///   from, read from `journal`. Without a journal a case with runs is refused
///   rather than assumed quiet: *no evidence of live work* is not evidence of
///   none.
/// * [`CaseStore::begin_erasure`](crate::case::CaseStore::begin_erasure)
///   writes the erasure record in the transaction that checks the hold and the
///   status, so a hold placed concurrently either lands first and stops this,
///   or is refused. From the record on, the case cannot be reopened, written
///   or held.
///
/// A retry of an erasure that was begun and not completed runs again under
/// the instant and reason the record holds, and records it complete at the
/// end.
///
/// # Errors
///
/// [`EraseError::Store`] carrying [`StoreError::NotFound`](crate::core::StoreError::NotFound)
/// if the case does not exist; [`EraseError::UnderLegalHold`],
/// [`EraseError::CaseStillOpen`] or [`EraseError::RunNotConcluded`] when
/// nothing was destroyed; and an error if the journal, the register or the
/// case's blob list cannot be read, a blob cannot be expired, or the key
/// cannot be destroyed.
#[allow(clippy::too_many_arguments)]
pub async fn erase_case(
    blobs: Option<&dyn BlobStore>,
    cases: &dyn crate::case::CaseStore,
    journal: Option<&dyn crate::journal::JournalStore>,
    #[cfg(feature = "keyring")] keyring: Option<&dyn crate::keyring::KeyRing>,
    disclosures: Option<&dyn crate::disclosure::DisclosureRegister>,
    tenant: &crate::core::TenantId,
    case: crate::core::CaseId,
    at: crate::core::Timestamp,
    reason: &str,
) -> Result<Erased, EraseError> {
    use crate::case::ErasureStart;

    // A case this plane does not hold is refused: answering `Ok(0)` would read
    // as a matter that stored nothing and was erased, closing a request whose
    // real matter is untouched.
    let Some(found) = cases.case(case).await? else {
        return Err(EraseError::Store(crate::core::StoreError::NotFound(
            case.to_string(),
        )));
    };

    // An erasure already begun had its runs checked before its record landed,
    // and from then on nothing reopens the case; a retry only finishes it.
    if cases.erasure(case).await?.is_none() {
        if let Some(hold) = cases.hold(case).await? {
            return Err(held(case, hold));
        }
        if found.status != crate::core::CaseStatus::Closed {
            return Err(still_open(case, found.status));
        }
        if let Some((run, state)) = unconcluded(journal, &found.runs).await? {
            return Err(EraseError::RunNotConcluded {
                run: run.to_string(),
                state,
            });
        }
    }

    // Read before anything is destroyed, so a register that cannot answer
    // leaves the matter whole rather than erased with its copies unnamed.
    let copies = copies_of(disclosures, &[case], &found.runs).await?;

    // **The decision the destruction rests on.** The hold and status above
    // are a courtesy for a clear answer; this write decides them again with
    // the record, in one transaction a concurrent `place_hold` serializes
    // against.
    let marked = match cases.begin_erasure(case, at, reason).await? {
        ErasureStart::Marked(marked) => marked,
        ErasureStart::Held(hold) => return Err(held(case, hold)),
        ErasureStart::NotClosed(status) => return Err(still_open(case, status)),
    };
    let (at, reason) = (marked.at, marked.reason.as_str());

    let digests = cases.blobs_of(case).await?;
    let scope = crate::core::erasure_scope(tenant, &case.to_string());
    let mut n = 0;
    if let Some(blobs) = blobs {
        for digest in digests {
            blobs
                .expire(unit_address(&scope, digest), at, reason)
                .await?;
            n += 1;
        }
    }

    // The key last, and only once every tombstone is written.
    //
    // Order matters in one direction only. Tombstones first means a crash
    // between the two leaves bytes that are still there and still readable —
    // recoverable by running the erasure again. Key first would leave
    // tombstones unwritten over bytes nobody can read, so a later read reports
    // *corrupt* instead of *expired* and an operator is paged for an integrity
    // fault that is really a completed erasure.
    //
    // This is the step that makes the erasure reach backups: the tombstones
    // above only cover the live store.
    #[cfg(feature = "keyring")]
    if let Some(keys) = keyring {
        keys.destroy(&scope, at, reason)
            .await
            .map_err(|e| EraseError::Blob(BlobError::Backend(e.to_string())))?;
    }
    cases.complete_erasure(case).await?;
    Ok(Erased { blobs: n, copies })
}

fn held(case: crate::core::CaseId, hold: crate::core::LegalHold) -> EraseError {
    EraseError::UnderLegalHold {
        case: case.to_string(),
        placed_at: hold.placed_at.to_string(),
        reason: hold.reason,
    }
}

fn still_open(case: crate::core::CaseId, status: crate::core::CaseStatus) -> EraseError {
    EraseError::CaseStillOpen {
        case: case.to_string(),
        status: format!("{status:?}").to_lowercase(),
    }
}

/// The first of `runs` that may still resume, and what its journal says.
///
/// Read whole rather than from the head: records can follow a conclusion, and
/// the question is the *last conclusion*, which
/// [`audit::has_sealing_conclusion`](crate::audit) already answers for the
/// offline checker — one rule for *which conclusions close*.
async fn unconcluded(
    journal: Option<&dyn crate::journal::JournalStore>,
    runs: &[crate::core::RunId],
) -> Result<Option<(crate::core::RunId, String)>, EraseError> {
    let Some(&first) = runs.first() else {
        return Ok(None);
    };
    let Some(journal) = journal else {
        return Ok(Some((
            first,
            "was not checked: no journal was given to read its conclusion from".to_owned(),
        )));
    };
    for &run in runs {
        let records = journal.read(run, 0).await?;
        if records.is_empty() {
            return Ok(Some((run, "has no record in this journal".to_owned())));
        }
        if !crate::audit::has_sealing_conclusion(&records) {
            return Ok(Some((
                run,
                "has no conclusion it cannot resume from".to_owned(),
            )));
        }
    }
    Ok(None)
}

/// What an erasure did, and the copies outside the plane it did not reach.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Erased {
    /// Blob references tombstoned.
    pub blobs: usize,
    /// One sentence per disclosure of what was erased, each saying what the
    /// erasure does to that copy and that the name comes from the operator's
    /// register — an unchained row, so an empty list does not show that
    /// nothing was disclosed.
    pub copies: Vec<String>,
}

/// The copies `disclosures` names for `cases` or `runs`, as an erasure reports
/// them; with no register, the one sentence saying none was consulted.
async fn copies_of(
    disclosures: Option<&dyn crate::disclosure::DisclosureRegister>,
    cases: &[crate::core::CaseId],
    runs: &[crate::core::RunId],
) -> Result<Vec<String>, EraseError> {
    let Some(register) = disclosures else {
        return Ok(vec![
            "no disclosure register was consulted, so no copy made outside the plane is \
             named here"
                .to_owned(),
        ]);
    };
    Ok(register
        .disclosures(cases, runs)
        .await?
        .iter()
        .map(crate::disclosure::Disclosure::after_erasure)
        .collect())
}

/// Destroy the erasure scope of a run that belongs to no case.
///
/// The counterpart of [`erase_case`], for the unit that call can never reach:
/// a record bound to no case seals its payloads under `tenant/<run>` (see
/// `SealedJournal`). This destroys exactly that scope, and with it every
/// payload sealed under that run — in the live store, every replica, and every
/// backup ever taken, because what is destroyed was never in them.
///
/// **The erasure unit is the run.** There is no blob traversal here because
/// blob writes are scoped to a run's case; a run with no case links no blobs
/// through the case layer, and anything sealed for it lives in its journal
/// payloads. The journal's records — chain, routing fields, the fact the run
/// happened — remain readable and verifiable, which is the whole design: the
/// chain committed to ciphertext.
///
/// Idempotent as [`KeyRing::destroy`](crate::keyring::KeyRing::destroy) is:
/// the first destruction stands, so a retry cannot rewrite when or why the
/// data went.
///
/// **Refused for a run that belongs to a case**, or one that may still
/// resume — both decided from `journal` before anything is destroyed. A
/// case-bound run's payloads are sealed under the case, so this verb would
/// destroy a scope holding none of them and answer as if it had erased the
/// run; and a run that may resume would be left unable to.
///
/// **A case-less run cannot be held.** A legal hold is placed on a *matter*,
/// so nothing here consults one. A deployment that needs a run preserved binds
/// it to a case, which also gives it an obligation trail.
///
/// # Errors
///
/// [`EraseError::RunBelongsToCase`] or [`EraseError::RunNotConcluded`] when
/// nothing was destroyed, or an error if the journal or the key ring cannot be
/// reached.
#[cfg(feature = "keyring")]
pub async fn erase_run(
    keyring: &dyn crate::keyring::KeyRing,
    journal: &dyn crate::journal::JournalStore,
    disclosures: Option<&dyn crate::disclosure::DisclosureRegister>,
    tenant: &crate::core::TenantId,
    run: crate::core::RunId,
    at: crate::core::Timestamp,
    reason: &str,
) -> Result<Vec<String>, EraseError> {
    let records = journal.read(run, 0).await?;
    if let Some(case) = records.iter().find_map(|r| r.body.case) {
        return Err(EraseError::RunBelongsToCase {
            run: run.to_string(),
            case: case.to_string(),
        });
    }
    if let Some((run, state)) = unconcluded(Some(journal), &[run]).await? {
        return Err(EraseError::RunNotConcluded {
            run: run.to_string(),
            state,
        });
    }
    let copies = copies_of(disclosures, &[], &[run]).await?;
    keyring
        .destroy(&crate::keyring::scope(tenant, &run.to_string()), at, reason)
        .await
        .map_err(|e| EraseError::Blob(BlobError::Backend(e.to_string())))?;
    Ok(copies)
}

/// Re-state a blob failure in the vocabulary a step is refused in.
///
/// Only one variant changes shape, and it is the one that is not a fault:
/// [`BlobError::Expired`] on a *write* means the address was erased, which is a
/// rule rather than an outage and must not be classified as one. Everything
/// else keeps its own words inside a backend error, because to a caller holding
/// a step there is nothing else to do about them.
pub(crate) fn refusal(e: BlobError) -> crate::core::StoreError {
    match e {
        BlobError::Expired { digest, at, reason } => {
            crate::core::StoreError::BlobErased { digest, at, reason }
        }
        other => crate::core::StoreError::Backend(other.to_string()),
    }
}

/// Check fetched bytes against the address they came from.
///
/// Shared by every backend so the verification cannot be implemented slightly
/// differently — or omitted — by one of them.
pub(crate) fn verify(digest: Digest, bytes: Vec<u8>) -> Result<Vec<u8>, BlobError> {
    let actual = Digest::of(&bytes);
    if actual == digest {
        Ok(bytes)
    } else {
        Err(BlobError::Corrupt {
            expected: digest.to_hex(),
            actual: actual.to_hex(),
        })
    }
}

#[cfg(test)]
mod tests {
    use super::{BlobError, refusal};
    use crate::core::StoreError;

    /// An erased address refuses a write as a rule, not as an outage: the
    /// classification is the half of the rule a caller holding a step sees.
    #[test]
    fn an_erased_write_is_refused_as_a_rule_rather_than_a_fault() {
        let refused = refusal(BlobError::Expired {
            digest: "ab".to_owned(),
            at: 7,
            reason: "art-17 request".to_owned(),
        });
        assert!(
            matches!(refused, StoreError::BlobErased { ref reason, .. } if reason == "art-17 request"),
            "an erased write must not be classified as a backend fault: {refused:?}"
        );
        assert!(matches!(
            refusal(BlobError::NotFound("ab".to_owned())),
            StoreError::Backend(_)
        ));
    }
}