feltdb 0.6.0

//! Lifecycle state as FeltDB records.
//!
//! The server's lifecycle stores (applications and their revisions, environment
//! pointers and promotion history, runtime intent, releases, tenancy) used to
//! persist themselves as JSON files beside the data file. This module makes
//! FeltDB their durable authority instead: every record a store holds is one row
//! in a protected system collection, `_feltdb.lifecycle.<store>.<kind>`, and
//! every change a store makes is **one atomic FeltDB transaction** carrying all
//! of its record changes, fenced on the row versions the store last committed.
//!
//! A store keeps an in-memory projection of its records for its read paths.
//! The projection is derived: it is loaded from FeltDB when the store opens, and
//! it is replaced only after a transaction has committed. A change whose fence
//! fails (a row differs from what the projection believed) applies nothing, and
//! the store reloads its projection from FeltDB before answering again.
//!
//! The JSON file a store used to write is accepted once, as migration input:
//! [`LifecycleRecords::open`] imports it in the same single transaction that
//! writes the migration-complete marker, so a migration is either wholly in
//! FeltDB or not at all, and is retried from the file after an interruption.
//! Once the marker exists the file is never read as state again; it is renamed
//! `<file>.migrated` and kept for recovery.
//!
//! A store with neither a marker nor a file starts empty and writes nothing:
//! its marker is committed in the same transaction as its first change. A
//! server that has changed nothing therefore holds no rows, which is what lets
//! a new replica bootstrap from a peer's snapshot (`FeltDb::install_snapshot`
//! accepts only a pristine database).

use crate::{AtomicMutation, AtomicPrecondition, FeltDb, FlowError, StoredRow};
use serde::{de::DeserializeOwned, Serialize};
use serde_json::{json, Value};
use sha2::{Digest, Sha256};
use std::collections::BTreeMap;
use std::path::{Path, PathBuf};

/// The prefix of every lifecycle system collection. It starts with `_`, so no
/// capability reads it, and the server refuses raw writes to it.
pub const LIFECYCLE_PREFIX: &str = "_feltdb.lifecycle.";
/// The collection of per-store migration markers.
pub const LIFECYCLE_META: &str = "_feltdb.lifecycle.meta";
/// The format of a lifecycle store's records; a marker records it.
pub const LIFECYCLE_FORMAT_VERSION: u32 = 1;

/// True for a lifecycle system collection.
pub fn is_lifecycle_collection(collection: &str) -> bool {
    collection.starts_with(LIFECYCLE_PREFIX) || collection == LIFECYCLE_META
}

/// True when FeltDB holds `store` (its migration-complete marker exists).
pub fn is_migrated(db: &FeltDb, store: &str) -> Result<bool, String> {
    Ok(db
        .get_collection_record(LIFECYCLE_META, &row_key(LIFECYCLE_META, store))
        .map_err(|error| error.to_string())?
        .is_some())
}

/// One record of a lifecycle store: its kind (the collection suffix), its
/// stable key, and its value.
pub type LifecycleRecord = (String, String, Value);

/// A lifecycle store's state, as the records it is made of.
///
/// `records` must give every record a stable, unique `(kind, key)`;
/// `from_records(records())` must reproduce the state. The kinds a store uses
/// are listed in `KINDS`, so that loading reads exactly those collections.
pub trait LifecycleState: Default + Clone + Serialize + DeserializeOwned {
    const STORE: &'static str;
    const KINDS: &'static [&'static str];
    /// Kinds whose records never change once committed. A change that would
    /// rewrite one is refused; removing one (retiring an application) is not.
    const IMMUTABLE: &'static [&'static str] = &[];
    fn records(&self) -> Vec<LifecycleRecord>;
    fn from_records(records: Vec<LifecycleRecord>) -> Result<Self, String>;
}

fn collection(store: &str, kind: &str) -> String {
    format!("{LIFECYCLE_PREFIX}{store}.{kind}")
}

/// A row's engine key: `<capability>:<record key>`, the convention every engine
/// path (snapshots, replication, exports) relies on.
fn row_key(capability: &str, key: &str) -> String {
    format!("{capability}:{key}")
}

/// The record key of an engine row key.
fn record_key<'a>(capability: &str, key: &'a str) -> &'a str {
    key.strip_prefix(capability).and_then(|rest| rest.strip_prefix(':')).unwrap_or(key)
}

fn sequence(row: &StoredRow) -> Option<u64> {
    row.version()
}

fn digest(value: &impl Serialize) -> Result<String, String> {
    let bytes = serde_json::to_vec(value).map_err(|error| error.to_string())?;
    Ok(format!("{:x}", Sha256::digest(bytes)))
}

/// What a store knows about its committed records: each value, and the row
/// sequence it was committed at (the fence for its next change).
#[derive(Clone, Default)]
pub struct Committed {
    rows: BTreeMap<(String, String), (Value, Option<u64>)>,
    /// The write versions of the store's collections when this view was read.
    versions: Vec<u64>,
}

/// How a store opened: from FeltDB, fresh, or by migrating its JSON file.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LifecycleOpen {
    /// The migration marker existed; FeltDB held the state.
    FeltDb { records: usize },
    /// No marker and no file: an empty store. Nothing is written until its
    /// first change, which commits the marker with it.
    Fresh,
    /// The file was imported, verified and renamed `<file>.migrated`.
    Migrated { records: usize, source: PathBuf },
}

/// The durable side of one lifecycle store.
#[derive(Clone)]
pub struct LifecycleRecords {
    db: FeltDb,
    store: &'static str,
}

impl LifecycleRecords {
    /// Open a store's records, migrating its legacy JSON file if FeltDB has
    /// never held them.
    ///
    /// Refuses, rather than choosing a source, when the marker says migration
    /// is complete but a legacy file that differs from FeltDB is present (it was
    /// written after the migration by something that still writes files).
    pub fn open<S: LifecycleState>(
        db: &FeltDb,
        legacy: &Path,
    ) -> Result<(Self, S, Committed, LifecycleOpen), String> {
        let records = Self { db: db.clone(), store: S::STORE };
        let marker = records.marker()?;
        let legacy_state = if legacy.exists() {
            let bytes = std::fs::read(legacy).map_err(|error| format!("read {}: {error}", legacy.display()))?;
            Some(serde_json::from_slice::<S>(&bytes).map_err(|error| {
                format!("LIFECYCLE_MIGRATION_SOURCE_INVALID: {} is not a valid {} store: {error}. Nothing was imported.", legacy.display(), S::STORE)
            })?)
        } else {
            None
        };
        match (marker, legacy_state) {
            (Some(_), legacy_state) => {
                let (state, committed) = records.load::<S>()?;
                if let Some(legacy_state) = legacy_state {
                    if digest(&canonical(&legacy_state.records()))? != digest(&canonical(&state.records()))? {
                        return Err(format!(
                            "LIFECYCLE_SPLIT_AUTHORITY: {} is already migrated into FeltDB, but {} exists and differs from it. Refusing to choose one: remove or rename the file if FeltDB is correct, or restore the store from backup.",
                            S::STORE, legacy.display()
                        ));
                    }
                    retire(legacy)?;
                }
                let count = committed.rows.len();
                Ok((records, state, committed, LifecycleOpen::FeltDb { records: count }))
            }
            (None, None) => {
                // Rows may already be here without a marker only by replication
                // or a snapshot still arriving; load reads whatever FeltDB holds.
                let (state, committed) = records.load::<S>()?;
                Ok((records, state, committed, LifecycleOpen::Fresh))
            }
            (None, Some(source)) => {
                // Validate by round-tripping through the record form: a record
                // the store cannot represent is refused before anything commits.
                let imported = source.records();
                let rebuilt = S::from_records(imported.clone())?;
                if digest(&canonical(&rebuilt.records()))? != digest(&canonical(&imported))? {
                    return Err(format!("LIFECYCLE_MIGRATION_SOURCE_INVALID: {} does not round-trip; nothing was imported", legacy.display()));
                }
                ensure_unique(&imported)?;
                records.import::<S>(&source, &digest(&canonical(&imported))?)?;
                let (state, committed) = records.load::<S>()?;
                if digest(&canonical(&state.records()))? != digest(&canonical(&imported))? {
                    return Err(format!("LIFECYCLE_MIGRATION_VERIFY_FAILED: {} in FeltDB does not equal {} after import", S::STORE, legacy.display()));
                }
                retire(legacy)?;
                Ok((records, state, committed, LifecycleOpen::Migrated { records: imported.len(), source: legacy.to_path_buf() }))
            }
        }
    }

    fn marker(&self) -> Result<Option<Value>, String> {
        Ok(self
            .db
            .get_collection_record(LIFECYCLE_META, &row_key(LIFECYCLE_META, self.store))
            .map_err(|error| error.to_string())?
            .map(|row| row.value))
    }

    /// Every record of the store, read from FeltDB.
    fn load<S: LifecycleState>(&self) -> Result<(S, Committed), String> {
        let mut committed = Committed { versions: self.versions::<S>(), ..Committed::default() };
        let mut records = Vec::new();
        for kind in S::KINDS {
            let capability = collection(self.store, kind);
            let rows = self
                .db
                .query_collection(&capability, None, |row| !row.deleted)
                .map_err(|error| error.to_string())?;
            for mut row in rows {
                row.key = record_key(&capability, &row.key).to_string();
                let version = sequence(&row);
                // `__version` is the engine's, added to a record it creates; the
                // store's own value is everything else.
                let mut value = row.value;
                if let Value::Object(fields) = &mut value {
                    fields.remove("__version");
                }
                committed
                    .rows
                    .insert(((*kind).to_string(), row.key.clone()), (value.clone(), version));
                records.push(((*kind).to_string(), row.key, value));
            }
        }
        Ok((S::from_records(records)?, committed))
    }

    /// Reload after a fence failure: what FeltDB holds now.
    pub fn reload<S: LifecycleState>(&self) -> Result<(S, Committed), String> {
        self.load::<S>()
    }

    fn versions<S: LifecycleState>(&self) -> Vec<u64> {
        let collections: Vec<String> = S::KINDS.iter().map(|kind| collection(self.store, kind)).collect();
        self.db.collection_write_versions(&collections)
    }

    /// True when a store collection changed since `committed` was read: by
    /// another handle, an offline tool, replication or a restore.
    pub fn changed_since<S: LifecycleState>(&self, committed: &Committed) -> bool {
        self.versions::<S>() != committed.versions
    }

    /// Write every record of `state` and the migration marker in one
    /// transaction. Each record must not exist yet, and neither may the marker:
    /// a second import of the same store is refused by the engine, not by a
    /// check that could race it.
    fn import<S: LifecycleState>(&self, state: &S, source: &str) -> Result<(), String> {
        let records = state.records();
        let mut preconditions = Vec::with_capacity(records.len() + 1);
        let mut mutations = Vec::with_capacity(records.len() + 1);
        for (kind, key, value) in &records {
            let capability = collection(self.store, kind);
            preconditions.push(AtomicPrecondition { capability: capability.clone(), key: row_key(&capability, key), expected_version: None });
            mutations.push(AtomicMutation { key: row_key(&capability, key), capability, value: Some(value.clone()) });
        }
        preconditions.push(AtomicPrecondition { capability: LIFECYCLE_META.into(), key: row_key(LIFECYCLE_META, self.store), expected_version: None });
        mutations.push(AtomicMutation {
            capability: LIFECYCLE_META.into(),
            key: row_key(LIFECYCLE_META, self.store),
            value: Some(json!({
                "store": self.store,
                "format_version": LIFECYCLE_FORMAT_VERSION,
                "status": "complete",
                "records": records.len(),
                "source": source,
            })),
        });
        self.db
            .apply_atomic_transaction(
                &format!("lifecycle-migrate-{}-{}", self.store, source),
                None,
                &preconditions,
                &mutations,
                Some(json!({ "kind": "lifecycle_migration", "store": self.store, "records": records.len() })),
            )
            .map(|_| ())
            .map_err(|error| format!("LIFECYCLE_MIGRATION_FAILED: {} could not be imported: {error}", self.store))
    }

    /// Commit the change from `before` to `next` as one transaction.
    ///
    /// Every changed record is fenced on the row sequence `before` was
    /// committed at (or required absent if it was new). The engine evaluates
    /// all the fences and applies all the writes under one lock and one durable
    /// log record, so the change is wholly committed or not at all. Returns the
    /// committed view of `next`.
    pub fn commit<S: LifecycleState>(
        &self,
        before: &Committed,
        next: &S,
        operation: &str,
    ) -> Result<Committed, LifecycleCommitError> {
        let next_records = next.records();
        ensure_unique(&next_records).map_err(LifecycleCommitError::Invalid)?;
        let mut after: BTreeMap<(String, String), Value> = BTreeMap::new();
        for (kind, key, value) in next_records {
            after.insert((kind, key), value);
        }
        let mut preconditions = Vec::new();
        let mut mutations = Vec::new();
        let mut changed = Vec::new();
        for (identity, (old, version)) in &before.rows {
            // Immutable records can be neither changed nor removed: history
            // keyed by position would otherwise be rewritten by a removal.
            if S::IMMUTABLE.contains(&identity.0.as_str()) && after.get(identity) != Some(old) {
                return Err(LifecycleCommitError::Invalid(format!(
                    "{}/{} is immutable once committed and cannot be changed or removed",
                    identity.0, identity.1
                )));
            }
            if after.get(identity) != Some(old) {
                let capability = collection(self.store, &identity.0);
                preconditions.push(AtomicPrecondition { capability: capability.clone(), key: row_key(&capability, &identity.1), expected_version: *version });
                mutations.push(AtomicMutation { key: row_key(&capability, &identity.1), capability, value: after.get(identity).cloned() });
                changed.push(identity.clone());
            }
        }
        for (identity, value) in &after {
            if !before.rows.contains_key(identity) {
                let capability = collection(self.store, &identity.0);
                preconditions.push(AtomicPrecondition { capability: capability.clone(), key: row_key(&capability, &identity.1), expected_version: None });
                mutations.push(AtomicMutation { key: row_key(&capability, &identity.1), capability, value: Some(value.clone()) });
                changed.push(identity.clone());
            }
        }
        if mutations.is_empty() {
            return Ok(before.clone());
        }
        // A fresh store's first change also marks it as held by FeltDB, in the
        // same transaction. Two handles racing to be first: the marker's
        // absence is fenced, so one commits and the other reloads.
        if self.marker().map_err(|error| LifecycleCommitError::Storage(FlowError::CapabilityError(error)))?.is_none() {
            preconditions.push(AtomicPrecondition { capability: LIFECYCLE_META.into(), key: row_key(LIFECYCLE_META, self.store), expected_version: None });
            mutations.push(AtomicMutation {
                capability: LIFECYCLE_META.into(),
                key: row_key(LIFECYCLE_META, self.store),
                value: Some(json!({
                    "store": self.store,
                    "format_version": LIFECYCLE_FORMAT_VERSION,
                    "status": "complete",
                    "records": 0,
                    "source": "fresh",
                })),
            });
        }
        let transaction_id = format!("lifecycle-{}-{:032x}", self.store, rand::random::<u128>());
        match self.db.apply_atomic_transaction(
            &transaction_id,
            None,
            &preconditions,
            &mutations,
            Some(json!({ "kind": "lifecycle", "store": self.store, "operation": operation, "records": mutations.len() })),
        ) {
            Ok(_) => {}
            // A row-sequence precondition is reported as `PRECONDITION_FAILED:<key>:...`.
            Err(FlowError::CapabilityError(detail)) if detail.starts_with("PRECONDITION_FAILED") => {
                return Err(LifecycleCommitError::Stale(format!(
                    "{} changed in FeltDB since this store last read it ({detail}); nothing was applied",
                    self.store
                )))
            }
            Err(FlowError::PreconditionFailed(detail)) => {
                return Err(LifecycleCommitError::Stale(format!(
                    "{} changed in FeltDB since this store last read it ({detail:?}); nothing was applied",
                    self.store
                )))
            }
            Err(error) => return Err(LifecycleCommitError::Storage(error)),
        }
        // Read back the committed sequences of exactly the rows written.
        let mut committed = before.clone();
        for identity in changed {
            let capability = collection(self.store, &identity.0);
            let row = self
                .db
                .get_collection_record(&capability, &row_key(&capability, &identity.1))
                .map_err(LifecycleCommitError::Storage)?;
            match (row, after.get(&identity)) {
                (Some(row), Some(value)) => {
                    committed.rows.insert(identity, (value.clone(), sequence(&row)));
                }
                _ => {
                    committed.rows.remove(&identity);
                }
            }
        }
        // The collections' write versions now include this commit and, possibly,
        // a concurrent writer's. They cannot be told apart here, so the view is
        // marked unknown and the next read reloads from FeltDB.
        committed.versions = Vec::new();
        Ok(committed)
    }

    pub fn store(&self) -> &'static str {
        self.store
    }
}

/// Why a lifecycle change did not commit. In every case nothing was applied,
/// except `Storage`, whose engine error says whether the outcome is known.
#[derive(Debug)]
pub enum LifecycleCommitError {
    /// The store's records are not representable (a duplicate identity).
    Invalid(String),
    /// A fenced row changed since the store read it.
    Stale(String),
    /// The engine refused or could not commit.
    Storage(FlowError),
}

impl std::fmt::Display for LifecycleCommitError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Invalid(message) | Self::Stale(message) => formatter.write_str(message),
            Self::Storage(error) => write!(formatter, "lifecycle commit failed: {error}"),
        }
    }
}

fn ensure_unique(records: &[LifecycleRecord]) -> Result<(), String> {
    let mut seen = std::collections::BTreeSet::new();
    for (kind, key, _) in records {
        if !seen.insert((kind.as_str(), key.as_str())) {
            return Err(format!("duplicate lifecycle record identity {kind}/{key}"));
        }
    }
    Ok(())
}

fn canonical(records: &[LifecycleRecord]) -> Vec<LifecycleRecord> {
    let mut sorted = records.to_vec();
    sorted.sort_by(|left, right| (&left.0, &left.1).cmp(&(&right.0, &right.1)));
    sorted
}

/// Keep the migrated file for recovery, under a name nothing reads as state.
fn retire(legacy: &Path) -> Result<(), String> {
    let mut retired = legacy.as_os_str().to_owned();
    retired.push(".migrated");
    let retired = PathBuf::from(retired);
    if retired.exists() {
        // An earlier migration already kept a copy; keep the oldest.
        std::fs::remove_file(legacy).map_err(|error| format!("remove {}: {error}", legacy.display()))
    } else {
        std::fs::rename(legacy, &retired).map_err(|error| format!("retire {}: {error}", legacy.display()))
    }
}

/// The field that holds a record's position among the records of its kind, so
/// that an ordered list (revisions, promotion history, audit) loads in the order
/// it was written. Records keep stable keys; only the position orders them.
const POSITION: &str = "_lifecycle_position";

/// Records of one kind, keyed by a stable identity, in list order.
pub fn keyed<T: Serialize>(kind: &str, values: &[T], key: impl Fn(&T) -> String) -> Vec<LifecycleRecord> {
    values
        .iter()
        .enumerate()
        .map(|(position, typed)| {
            let mut value = serde_json::to_value(typed).expect("lifecycle records serialize");
            if let Value::Object(fields) = &mut value {
                fields.insert(POSITION.into(), json!(position));
            }
            (kind.to_string(), key(typed), value)
        })
        .collect()
}

/// Records of one kind keyed by a stable identity, stored exactly as they are
/// (no position): for immutable records, whose stored value must never change.
pub fn keyed_as_is<T: Serialize>(kind: &str, values: &[T], key: impl Fn(&T) -> String) -> Vec<LifecycleRecord> {
    values
        .iter()
        .map(|typed| (kind.to_string(), key(typed), serde_json::to_value(typed).expect("lifecycle records serialize")))
        .collect()
}

/// The values of one kind, in the order they were written.
pub fn values_of<T: DeserializeOwned>(records: &[LifecycleRecord], kind: &str) -> Result<Vec<T>, String> {
    let mut ordered: Vec<(u64, T)> = records
        .iter()
        .filter(|(record_kind, _, _)| record_kind == kind)
        .map(|(_, key, value)| {
            let mut value = value.clone();
            let position = match &mut value {
                Value::Object(fields) => fields.remove(POSITION).and_then(|value| value.as_u64()).unwrap_or(u64::MAX),
                _ => u64::MAX,
            };
            serde_json::from_value(value).map(|typed| (position, typed)).map_err(|error| format!("{kind}/{key}: {error}"))
        })
        .collect::<Result<_, _>>()?;
    ordered.sort_by_key(|(position, _)| *position);
    Ok(ordered.into_iter().map(|(_, value)| value).collect())
}

/// Where a lifecycle store keeps its records, and what it last committed.
///
/// A store holds its records in an `RwLock<S>` and this binding beside them.
/// Lock order is always the records first, then the binding's committed view.
#[derive(Clone)]
pub struct LifecycleBinding {
    durability: BindingDurability,
}

#[derive(Clone)]
enum BindingDurability {
    /// A standalone JSON file: tools and tests. The server never uses it.
    File(PathBuf),
    /// FeltDB system collections: the server's authority.
    FeltDb(LifecycleRecords, std::sync::Arc<std::sync::Mutex<Committed>>),
}

impl LifecycleBinding {
    /// A store backed by a JSON file, loading it if it exists.
    pub fn file<S: LifecycleState>(path: impl Into<PathBuf>) -> Result<(Self, S), String> {
        let path = path.into();
        let state = if path.exists() {
            let bytes = std::fs::read(&path).map_err(|error| error.to_string())?;
            serde_json::from_slice(&bytes).map_err(|error| error.to_string())?
        } else {
            S::default()
        };
        Ok((Self { durability: BindingDurability::File(path) }, state))
    }

    /// A store on FeltDB, migrating `legacy` once if FeltDB never held it.
    pub fn feltdb<S: LifecycleState>(db: &FeltDb, legacy: &Path) -> Result<(Self, S, LifecycleOpen), String> {
        let (records, state, committed, opened) = LifecycleRecords::open::<S>(db, legacy)?;
        Ok((
            Self { durability: BindingDurability::FeltDb(records, std::sync::Arc::new(std::sync::Mutex::new(committed))) },
            state,
            opened,
        ))
    }

    /// True when the store is backed by FeltDB.
    pub fn is_feltdb(&self) -> bool {
        matches!(self.durability, BindingDurability::FeltDb(..))
    }

    /// Bring `records` (held under its write lock) up to date with FeltDB.
    pub fn refresh<S: LifecycleState>(&self, records: &mut S) -> Result<(), String> {
        if let BindingDurability::FeltDb(durable, committed) = &self.durability {
            let mut committed = committed.lock().map_err(|_| format!("{} store lock poisoned", S::STORE))?;
            if durable.changed_since::<S>(&committed) {
                let (state, reloaded) = durable.reload::<S>()?;
                *records = state;
                *committed = reloaded;
            }
        }
        Ok(())
    }

    /// The store's write lock, with its records current.
    pub fn write<'a, S: LifecycleState>(
        &self,
        records: &'a std::sync::RwLock<S>,
    ) -> Result<std::sync::RwLockWriteGuard<'a, S>, String> {
        let mut guard = records.write().map_err(|_| format!("{} store lock poisoned", S::STORE))?;
        self.refresh(&mut *guard)?;
        Ok(guard)
    }

    /// The store's read lock, with its records current. A FeltDB read failure
    /// keeps the last committed view rather than inventing one.
    pub fn read<'a, S: LifecycleState>(
        &self,
        records: &'a std::sync::RwLock<S>,
    ) -> std::sync::LockResult<std::sync::RwLockReadGuard<'a, S>> {
        if let BindingDurability::FeltDb(durable, committed) = &self.durability {
            let stale = committed.lock().map(|committed| durable.changed_since::<S>(&committed)).unwrap_or(false);
            if stale {
                if let Ok(mut guard) = records.write() {
                    let _ = self.refresh(&mut *guard);
                }
            }
        }
        records.read()
    }

    /// Make `next` durable and swap it into `guard`, or leave both as they were.
    /// On FeltDB, `next` is committed as one transaction fenced on what `guard`
    /// was committed at; a fence failure reloads `guard` from FeltDB and refuses.
    pub fn persist<S: LifecycleState>(&self, guard: &mut S, next: S, operation: &str) -> Result<(), String> {
        match &self.durability {
            BindingDurability::File(path) => {
                crate::application::write_file_durably(path, &serde_json::to_vec_pretty(&next).map_err(|error| error.to_string())?)?;
            }
            BindingDurability::FeltDb(durable, committed) => {
                let mut committed = committed.lock().map_err(|_| format!("{} store lock poisoned", S::STORE))?;
                match durable.commit(&committed, &next, operation) {
                    Ok(after) => *committed = after,
                    Err(LifecycleCommitError::Stale(message)) => {
                        let (state, reloaded) = durable.reload::<S>()?;
                        *guard = state;
                        *committed = reloaded;
                        return Err(format!("lifecycle_state_changed: {message}"));
                    }
                    Err(error) => return Err(error.to_string()),
                }
            }
        }
        *guard = next;
        Ok(())
    }

    /// Apply `change` to a current copy of the records and make it durable.
    pub fn mutate<S: LifecycleState, T>(
        &self,
        records: &std::sync::RwLock<S>,
        operation: &str,
        change: impl FnOnce(&mut S) -> Result<T, String>,
    ) -> Result<T, String> {
        let mut guard = self.write(records)?;
        let mut next = guard.clone();
        let result = change(&mut next)?;
        self.persist(&mut *guard, next, operation)?;
        Ok(result)
    }
}