chio_store_sqlite/

economic_state_cache.rs

use std::sync::{Arc, Mutex, MutexGuard};

use chio_core::canonical::canonical_json_bytes;
use chio_core::economic_continuity::{
    economic_effect_slot_from_head, verify_economic_completed_effect,
    verify_economic_state_batch_commit, EconomicAdmissionHandoffStateV1, EconomicContentV1,
    EconomicContinuityError, EconomicEffectSlotV1, EconomicEffectStateV1, EconomicEffectTerminalV1,
    EconomicResourceHeadV1, EconomicResourceKeyV1, EconomicStateAnchorError,
    EconomicStateAnchorPins, EconomicStateAnchorViewV1, EconomicStateBatchV1,
    EconomicTerminalResultV1, VerifiedEconomicStateBatchAdvance, VerifiedEconomicStateView,
    MAX_ECONOMIC_BATCH_BYTES,
};
use chio_core::{sha256_hex, StoreMutationFence};
use chio_credit::clearing::CLEARING_LIFECYCLE_REPLAY_DESCRIPTOR_KIND;
use chio_kernel::admission_operation::{
    verify_economic_cancellation_terminal_advance, AdmissionOperationState, AdmissionOperationV1,
    AdmissionProjectionRecordKind, AdmissionRecoveryLease, PersistedAdmissionOperationV1,
    SignedAdmissionTerminalProjectionV1, VerifiedAdmissionTerminalProjectionV1,
};
use chio_kernel::ADMISSION_TERMINAL_PROJECTION_DESCRIPTOR_KIND;
use chio_settle::channel::{
    CHANNEL_ESCROW_RESERVATION_RESOURCE_FAMILY, CHANNEL_LIFECYCLE_RESOURCE_FAMILY,
    CHANNEL_SERVICE_DISPATCH_EFFECT_KIND, CHANNEL_TRANSITION_REPLAY_FORMAT,
};
use rusqlite::{params, Connection, OptionalExtension, Row, Transaction, TransactionBehavior};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};

use crate::serving_owner::{SqliteServingOwner, SqliteServingOwnerError};

mod persistence;
pub(crate) use persistence::verify_cache_sql_invariants;
use persistence::*;
pub(crate) use persistence::{append_stage_commit, load_stage_tx, update_stage};

const ECONOMIC_STATE_CACHE_SCHEMA_KEY: &str = "economic_state_cache";
pub(crate) const ECONOMIC_STATE_CACHE_SUPPORTED_SCHEMA_VERSION: i32 = 3;
const ECONOMIC_STATE_CACHE_SCHEMA_ANCHORS: &[&str] =
    &["economic_state_stages", "capability_grant_budgets"];
const MAX_REASON_BYTES: usize = 4 * 1024;
const MAX_VIEW_BYTES: usize = 4 * MAX_ECONOMIC_BATCH_BYTES;
const MAX_DESCRIPTOR_BYTES: usize = 4 * MAX_ECONOMIC_BATCH_BYTES;
const MAX_DESCRIPTOR_KIND_BYTES: usize = 128;
const MAX_DESCRIPTOR_KEY_BYTES: usize = 2_048;
const MAX_TRUSTED_UNIX_MS: u64 = (1_u64 << 53) - 1;
const GENESIS_STAGE_COMMIT_DIGEST: &str =
    "0000000000000000000000000000000000000000000000000000000000000000";
const ADMISSION_TERMINAL_EFFECT_RESULT_SCHEMA: &str = "chio.admission.terminal-effect-result.v1";

const ECONOMIC_STATE_CACHE_SCHEMA: &str = include_str!("economic_state_cache.sql");
const ECONOMIC_STATE_CACHE_DESCRIPTOR_MIGRATION: &str = r#"
ALTER TABLE economic_state_stages ADD COLUMN descriptor_kind TEXT
    CHECK (descriptor_kind IS NULL OR length(descriptor_kind) BETWEEN 1 AND 128);
ALTER TABLE economic_state_stages ADD COLUMN descriptor_key TEXT
    CHECK (descriptor_key IS NULL OR length(descriptor_key) BETWEEN 1 AND 2048);
ALTER TABLE economic_state_stages ADD COLUMN descriptor_digest TEXT
    CHECK (descriptor_digest IS NULL OR (
        length(descriptor_digest) = 64
        AND descriptor_digest NOT GLOB '*[^0-9a-f]*'
    ));
ALTER TABLE economic_state_stages ADD COLUMN descriptor_json BLOB
    CHECK (descriptor_json IS NULL OR length(descriptor_json) BETWEEN 1 AND 4194304);
DROP TRIGGER IF EXISTS economic_state_stage_identity_immutable;
CREATE TRIGGER economic_state_stage_identity_immutable
BEFORE UPDATE OF batch_id, checkpoint_sequence, checkpoint_digest,
    base_view_json, batch_json, operation_binding_json, descriptor_kind,
    descriptor_key, descriptor_digest, descriptor_json, created_at_unix_ms
ON economic_state_stages
BEGIN
    SELECT RAISE(ABORT, 'economic stage identity is immutable');
END;
"#;

#[derive(Debug, thiserror::Error)]
pub enum EconomicStateCacheError {
    #[error("economic state cache is unavailable: {0}")]
    Unavailable(String),
    #[error("economic state cache mutation was fenced")]
    Fenced,
    #[error("economic state stage was not found")]
    NotFound,
    #[error("economic state stage conflicts with retained state")]
    Conflict,
    #[error("economic state stage transition from {from:?} to {to:?} is invalid")]
    InvalidTransition {
        from: EconomicStateStageStatus,
        to: EconomicStateStageStatus,
    },
    #[error("economic state cache invariant failed: {0}")]
    Invariant(String),
    #[error("economic state cache durable outcome is unknown: {0}")]
    OutcomeUnknown(String),
    #[error(transparent)]
    Continuity(#[from] EconomicContinuityError),
    #[error(transparent)]
    Anchor(#[from] EconomicStateAnchorError),
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum EconomicStateStageStatus {
    DbStaged,
    EconomicAnchorAdvanced,
    DbFinalized,
    Discarded,
    Quarantined,
}

impl EconomicStateStageStatus {
    fn as_str(self) -> &'static str {
        match self {
            Self::DbStaged => "db_staged",
            Self::EconomicAnchorAdvanced => "economic_anchor_advanced",
            Self::DbFinalized => "db_finalized",
            Self::Discarded => "discarded",
            Self::Quarantined => "quarantined",
        }
    }

    fn parse(value: &str) -> Result<Self, EconomicStateCacheError> {
        match value {
            "db_staged" => Ok(Self::DbStaged),
            "economic_anchor_advanced" => Ok(Self::EconomicAnchorAdvanced),
            "db_finalized" => Ok(Self::DbFinalized),
            "discarded" => Ok(Self::Discarded),
            "quarantined" => Ok(Self::Quarantined),
            _ => Err(invariant("economic stage status is invalid")),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct EconomicOperationStageBinding {
    operation_id: String,
    operation_state: AdmissionOperationState,
    operation_version: u64,
    coordinator_lease_epoch: u64,
    coordinator_lease_id: String,
    recovery_claimant_id: String,
    recovery_expires_at_unix_ms: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    not_after_unix_ms: Option<u64>,
    store_fence: StoreMutationFence,
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EconomicStateStageDescriptor {
    kind: String,
    key: String,
    digest: String,
    canonical_json: Vec<u8>,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct AdmissionTerminalEffectRecordCommitment {
    kind: AdmissionProjectionRecordKind,
    record_id: String,
    record_digest: String,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct AdmissionTerminalEffectCommitment {
    schema: &'static str,
    descriptor_digest: String,
    operation_id: String,
    request_id: String,
    source_operation_digest: String,
    terminal_operation_digest: String,
    terminal_state: AdmissionOperationState,
    projection_digest: String,
    records: Vec<AdmissionTerminalEffectRecordCommitment>,
}

impl EconomicStateStageDescriptor {
    pub fn new<T: Serialize>(
        kind: impl Into<String>,
        key: impl Into<String>,
        value: &T,
    ) -> Result<Self, EconomicStateCacheError> {
        let descriptor = Self {
            kind: kind.into(),
            key: key.into(),
            digest: String::new(),
            canonical_json: canonical_bounded(value, MAX_DESCRIPTOR_BYTES, "stage descriptor")?,
        };
        descriptor.with_digest()
    }

    fn from_stored(
        kind: String,
        key: String,
        digest: String,
        canonical_json: Vec<u8>,
    ) -> Result<Self, EconomicStateCacheError> {
        let descriptor = Self {
            kind,
            key,
            digest,
            canonical_json,
        };
        descriptor.validate()?;
        Ok(descriptor)
    }

    fn with_digest(mut self) -> Result<Self, EconomicStateCacheError> {
        self.digest = sha256_hex(&self.canonical_json);
        self.validate()?;
        Ok(self)
    }

    fn validate(&self) -> Result<(), EconomicStateCacheError> {
        validate_descriptor_text(
            &self.kind,
            MAX_DESCRIPTOR_KIND_BYTES,
            "stage descriptor kind",
        )?;
        validate_descriptor_text(&self.key, MAX_DESCRIPTOR_KEY_BYTES, "stage descriptor key")?;
        validate_digest(&self.digest, "stage descriptor digest")?;
        if self.canonical_json.is_empty()
            || self.canonical_json.len() > MAX_DESCRIPTOR_BYTES
            || sha256_hex(&self.canonical_json) != self.digest
        {
            return Err(invariant("economic stage descriptor is invalid"));
        }
        decode_exact::<serde_json::Value>(&self.canonical_json, "stage descriptor")?;
        Ok(())
    }

    #[must_use]
    pub fn kind(&self) -> &str {
        &self.kind
    }

    #[must_use]
    pub fn key(&self) -> &str {
        &self.key
    }

    #[must_use]
    pub fn digest(&self) -> &str {
        &self.digest
    }

    pub fn decode<T: DeserializeOwned + Serialize>(&self) -> Result<T, EconomicStateCacheError> {
        self.validate()?;
        decode_exact(&self.canonical_json, "stage descriptor")
    }
}

impl EconomicOperationStageBinding {
    fn validate(&self) -> Result<(), EconomicStateCacheError> {
        if self.operation_id.is_empty()
            || self.coordinator_lease_id.is_empty()
            || self.recovery_claimant_id.is_empty()
            || self.store_fence.store_uuid.is_empty()
            || self.store_fence.lease_id.is_empty()
        {
            return Err(invariant("economic operation binding identity is invalid"));
        }
        if self.operation_version == 0 || self.operation_version > MAX_TRUSTED_UNIX_MS {
            return Err(invariant(
                "economic operation binding operation_version is invalid",
            ));
        }
        if self.coordinator_lease_epoch == 0 || self.coordinator_lease_epoch > MAX_TRUSTED_UNIX_MS {
            return Err(invariant(
                "economic operation binding coordinator_lease_epoch is invalid",
            ));
        }
        if self.recovery_expires_at_unix_ms == 0
            || self.recovery_expires_at_unix_ms > MAX_TRUSTED_UNIX_MS
        {
            return Err(invariant(
                "economic operation binding recovery_expires_at_unix_ms is invalid",
            ));
        }
        if self
            .not_after_unix_ms
            .is_some_and(|value| value == 0 || value > MAX_TRUSTED_UNIX_MS)
        {
            return Err(invariant(
                "economic operation binding not_after_unix_ms is invalid",
            ));
        }
        if self.store_fence.owner_epoch == 0 || self.store_fence.owner_epoch > MAX_TRUSTED_UNIX_MS {
            return Err(invariant(
                "economic operation binding store fence is invalid",
            ));
        }
        Ok(())
    }

    #[must_use]
    pub fn operation_id(&self) -> &str {
        &self.operation_id
    }

    #[must_use]
    pub fn operation_state(&self) -> AdmissionOperationState {
        self.operation_state
    }

    #[must_use]
    pub fn operation_version(&self) -> u64 {
        self.operation_version
    }

    #[must_use]
    pub fn coordinator_lease_epoch(&self) -> u64 {
        self.coordinator_lease_epoch
    }

    #[must_use]
    pub fn coordinator_lease_id(&self) -> &str {
        &self.coordinator_lease_id
    }

    #[must_use]
    pub fn recovery_claimant_id(&self) -> &str {
        &self.recovery_claimant_id
    }

    #[must_use]
    pub const fn recovery_expires_at_unix_ms(&self) -> u64 {
        self.recovery_expires_at_unix_ms
    }

    #[must_use]
    pub const fn not_after_unix_ms(&self) -> Option<u64> {
        self.not_after_unix_ms
    }

    #[must_use]
    pub const fn store_fence(&self) -> &StoreMutationFence {
        &self.store_fence
    }
}

#[derive(Clone, Copy)]
pub struct EconomicOperationStageContext<'a> {
    operation: &'a AdmissionOperationV1,
    recovery_lease: &'a AdmissionRecoveryLease,
    not_after_unix_ms: Option<u64>,
}

pub(crate) struct EconomicStageAdmissionCheckpoint<'a> {
    pub(crate) store_id: &'a str,
    pub(crate) sequence: u64,
    pub(crate) digest: &'a str,
}

struct EconomicStageOptions<'a> {
    operation: Option<EconomicOperationStageContext<'a>>,
    descriptor: Option<EconomicStateStageDescriptor>,
    admission_checkpoint: Option<EconomicStageAdmissionCheckpoint<'a>>,
    consumer_descriptor_kind: Option<&'static str>,
}

impl<'a> EconomicOperationStageContext<'a> {
    #[must_use]
    pub const fn new(
        operation: &'a AdmissionOperationV1,
        recovery_lease: &'a AdmissionRecoveryLease,
    ) -> Self {
        Self {
            operation,
            recovery_lease,
            not_after_unix_ms: None,
        }
    }

    pub fn with_not_after_unix_ms(
        mut self,
        not_after_unix_ms: u64,
    ) -> Result<Self, EconomicStateCacheError> {
        validate_trusted_time(not_after_unix_ms)?;
        self.not_after_unix_ms = Some(not_after_unix_ms);
        Ok(self)
    }

    #[must_use]
    pub const fn operation(&self) -> &'a AdmissionOperationV1 {
        self.operation
    }

    #[must_use]
    pub const fn recovery_lease(&self) -> &'a AdmissionRecoveryLease {
        self.recovery_lease
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EconomicStateStageRecord {
    base_view: EconomicStateAnchorViewV1,
    batch: EconomicStateBatchV1,
    committed_view: Option<EconomicStateAnchorViewV1>,
    operation_binding: Option<EconomicOperationStageBinding>,
    descriptor: Option<EconomicStateStageDescriptor>,
    status: EconomicStateStageStatus,
    reason: Option<String>,
    version: u64,
    created_at_unix_ms: u64,
    updated_at_unix_ms: u64,
    snapshot_digest: String,
}

impl EconomicStateStageRecord {
    #[must_use]
    pub fn base_view(&self) -> &EconomicStateAnchorViewV1 {
        &self.base_view
    }

    #[must_use]
    pub fn batch(&self) -> &EconomicStateBatchV1 {
        &self.batch
    }

    #[must_use]
    pub fn committed_view(&self) -> Option<&EconomicStateAnchorViewV1> {
        self.committed_view.as_ref()
    }

    #[must_use]
    pub fn operation_binding(&self) -> Option<&EconomicOperationStageBinding> {
        self.operation_binding.as_ref()
    }

    #[must_use]
    pub fn not_after_unix_ms(&self) -> Option<u64> {
        self.operation_binding
            .as_ref()
            .and_then(EconomicOperationStageBinding::not_after_unix_ms)
    }

    #[must_use]
    pub fn descriptor(&self) -> Option<&EconomicStateStageDescriptor> {
        self.descriptor.as_ref()
    }

    #[must_use]
    pub fn status(&self) -> EconomicStateStageStatus {
        self.status
    }

    #[must_use]
    pub fn reason(&self) -> Option<&str> {
        self.reason.as_deref()
    }

    #[must_use]
    pub fn version(&self) -> u64 {
        self.version
    }
}

pub fn admission_terminal_projection_effect_result(
    envelope: &SignedAdmissionTerminalProjectionV1,
) -> Result<EconomicEffectTerminalV1, EconomicStateCacheError> {
    let verified = envelope
        .verify()
        .map_err(|_| EconomicStateCacheError::Conflict)?;
    let descriptor = admission_terminal_projection_descriptor(envelope, &verified)?;
    terminal_projection_effect_result(&descriptor, &verified)
}

fn admission_terminal_projection_descriptor(
    envelope: &SignedAdmissionTerminalProjectionV1,
    verified: &VerifiedAdmissionTerminalProjectionV1,
) -> Result<EconomicStateStageDescriptor, EconomicStateCacheError> {
    EconomicStateStageDescriptor::new(
        ADMISSION_TERMINAL_PROJECTION_DESCRIPTOR_KIND,
        verified
            .source_operation()
            .binding()
            .operation_id()
            .as_str(),
        envelope,
    )
}

fn terminal_projection_effect_result(
    descriptor: &EconomicStateStageDescriptor,
    verified: &VerifiedAdmissionTerminalProjectionV1,
) -> Result<EconomicEffectTerminalV1, EconomicStateCacheError> {
    let source = verified.source_operation();
    let terminal = verified.terminal_operation();
    let projection_digest = terminal
        .terminal_replay()
        .ok_or(EconomicStateCacheError::Conflict)?
        .projection_digest()
        .as_str()
        .to_owned();
    let source_operation_digest =
        sha256_hex(&canonical_json_bytes(&source.to_persisted()).map_err(canonical_error)?);
    let terminal_operation_digest =
        sha256_hex(&canonical_json_bytes(&terminal.to_persisted()).map_err(canonical_error)?);
    let commitment = AdmissionTerminalEffectCommitment {
        schema: ADMISSION_TERMINAL_EFFECT_RESULT_SCHEMA,
        descriptor_digest: descriptor.digest().to_owned(),
        operation_id: source.binding().operation_id().as_str().to_owned(),
        request_id: source.replay_key().request_id.as_str().to_owned(),
        source_operation_digest,
        terminal_operation_digest,
        terminal_state: terminal.state(),
        projection_digest: projection_digest.clone(),
        records: verified
            .records()
            .iter()
            .map(|record| AdmissionTerminalEffectRecordCommitment {
                kind: record.kind(),
                record_id: record.record_id().as_str().to_owned(),
                record_digest: record.record_digest().as_str().to_owned(),
            })
            .collect(),
    };
    let result = EconomicContentV1::Inline {
        value: serde_json::to_value(commitment).map_err(canonical_error)?,
    };
    Ok(EconomicEffectTerminalV1::Completed {
        result_id: projection_digest,
        result_digest: result.digest()?,
        result,
    })
}

fn qualify_generic_terminal_projection_effect_slot(
    base_view: &EconomicStateAnchorViewV1,
    batch: &EconomicStateBatchV1,
    descriptor: &EconomicStateStageDescriptor,
    verified: &VerifiedAdmissionTerminalProjectionV1,
) -> Result<EconomicEffectSlotV1, EconomicStateCacheError> {
    if batch.transitions.len() != 1
        || !batch.effect_slots.is_empty()
        || !batch.request_replays.is_empty()
        || batch.transitions[0].prepared_effect.is_some()
    {
        return Err(EconomicStateCacheError::Conflict);
    }
    let transition = &batch.transitions[0];
    if transition.resource_key.resource_family != "effect_slot" {
        return Err(EconomicStateCacheError::Conflict);
    }
    let current_head = base_view
        .head(&transition.resource_key)
        .ok_or(EconomicStateCacheError::Conflict)?;
    let current_slot = economic_effect_slot_from_head(current_head)
        .map_err(|_| EconomicStateCacheError::Conflict)?;
    let completed_slot = economic_effect_slot_from_head(&transition.next_head)
        .map_err(|_| EconomicStateCacheError::Conflict)?;
    current_slot
        .validate_successor(&completed_slot)
        .map_err(|_| EconomicStateCacheError::Conflict)?;
    if completed_slot.state == EconomicEffectStateV1::NoEffect {
        return verify_economic_cancellation_terminal_advance(base_view, batch, verified)
            .map_err(|_| EconomicStateCacheError::Conflict);
    }
    let source = verified.source_operation();
    let binding = source.binding();
    let dispatch_commit = source
        .dispatch_commit()
        .ok_or(EconomicStateCacheError::Conflict)?;
    let expected_terminal = terminal_projection_effect_result(descriptor, verified)?;
    let expected_result = match &expected_terminal {
        EconomicEffectTerminalV1::Completed {
            result_id,
            result_digest,
            result,
        } => EconomicTerminalResultV1 {
            result_id: result_id.clone(),
            result_digest: result_digest.clone(),
            result: result.clone(),
        },
        EconomicEffectTerminalV1::NoEffect { .. } => return Err(EconomicStateCacheError::Conflict),
    };
    if current_slot.state != EconomicEffectStateV1::DispatchCommitted
        && current_slot.state != EconomicEffectStateV1::Unknown
        || completed_slot.state != EconomicEffectStateV1::Completed
        || completed_slot.terminal.as_ref() != Some(&expected_terminal)
        || completed_slot.operation_id != binding.operation_id().as_str()
        || completed_slot.request.request_namespace_digest
            != binding.request_namespace_digest().as_str()
        || completed_slot.request.request_id != binding.request_id().as_str()
        || completed_slot.request.request_binding_digest != binding.request_binding_hash().as_str()
        || completed_slot.parameters_digest != binding.action_parameter_hash().as_str()
        || completed_slot.admission_handoff.state
            != EconomicAdmissionHandoffStateV1::DispatchCommitted
        || completed_slot.admission_handoff.operation_version != dispatch_commit.committed_version
        || completed_slot.admission_handoff.lifecycle_fence
            != dispatch_commit.coordinator_lease_epoch
        || completed_slot.admission_handoff.store_fence != dispatch_commit.store_fence
        || !current_fence_serves_historical(
            &dispatch_commit.store_fence,
            &verified.context().store_fence,
        )
        || batch.operation_id.as_deref() != Some(binding.operation_id().as_str())
        || transition.next_head.lifecycle_state != "completed"
        || transition.next_head.terminal_result.as_ref() != Some(&expected_result)
    {
        return Err(EconomicStateCacheError::Conflict);
    }
    Ok(completed_slot)
}

fn qualify_terminal_projection_advance(
    advance: &VerifiedEconomicStateBatchAdvance,
    descriptor: &EconomicStateStageDescriptor,
    verified: &VerifiedAdmissionTerminalProjectionV1,
) -> Result<EconomicEffectSlotV1, EconomicStateCacheError> {
    match verified.channel_terminal() {
        Some(channel) => channel
            .qualify_anchored_advance(advance)
            .cloned()
            .map_err(|_| EconomicStateCacheError::Conflict),
        None => qualify_generic_terminal_projection_effect_slot(
            advance.current().view(),
            advance.batch(),
            descriptor,
            verified,
        ),
    }
}

fn qualify_retained_terminal_projection_advance(
    base_view: &EconomicStateAnchorViewV1,
    batch: &EconomicStateBatchV1,
    descriptor: &EconomicStateStageDescriptor,
    verified: &VerifiedAdmissionTerminalProjectionV1,
) -> Result<EconomicEffectSlotV1, EconomicStateCacheError> {
    match verified.channel_terminal() {
        Some(channel) => channel
            .qualify_retained_anchored_advance(base_view, batch)
            .cloned()
            .map_err(|_| EconomicStateCacheError::Conflict),
        None => {
            qualify_generic_terminal_projection_effect_slot(base_view, batch, descriptor, verified)
        }
    }
}

fn committed_view_matches_batch(
    committed: &EconomicStateAnchorViewV1,
    batch: &EconomicStateBatchV1,
) -> bool {
    batch
        .transitions
        .iter()
        .all(|transition| committed.head(&transition.resource_key) == Some(&transition.next_head))
}

fn batch_contains_channel_content(batch: &EconomicStateBatchV1) -> bool {
    batch.transitions.iter().any(|transition| {
        matches!(
            transition.resource_key.resource_family.as_str(),
            CHANNEL_LIFECYCLE_RESOURCE_FAMILY | CHANNEL_ESCROW_RESERVATION_RESOURCE_FAMILY
        )
    }) || batch
        .effect_slots
        .iter()
        .any(|effect| effect.effect_kind == CHANNEL_SERVICE_DISPATCH_EFFECT_KIND)
}

fn is_protected_channel_stage(record: &EconomicStateStageRecord) -> bool {
    record
        .descriptor
        .as_ref()
        .is_some_and(|descriptor| descriptor.kind == CHANNEL_TRANSITION_REPLAY_FORMAT)
        || batch_contains_channel_content(&record.batch)
}

fn validate_stage_options(
    batch: &EconomicStateBatchV1,
    options: &EconomicStageOptions<'_>,
) -> Result<(), EconomicStateCacheError> {
    if let Some(descriptor) = &options.descriptor {
        descriptor.validate()?;
        let reserved = matches!(
            descriptor.kind(),
            CLEARING_LIFECYCLE_REPLAY_DESCRIPTOR_KIND
                | ADMISSION_TERMINAL_PROJECTION_DESCRIPTOR_KIND
                | CHANNEL_TRANSITION_REPLAY_FORMAT
        );
        if reserved != (options.consumer_descriptor_kind == Some(descriptor.kind()))
            || options
                .consumer_descriptor_kind
                .is_some_and(|kind| kind != descriptor.kind())
        {
            return Err(EconomicStateCacheError::Conflict);
        }
    } else if options.consumer_descriptor_kind.is_some() {
        return Err(EconomicStateCacheError::Conflict);
    }
    let channel_content = batch_contains_channel_content(batch);
    let channel_consumer = matches!(
        options.consumer_descriptor_kind,
        Some(CHANNEL_TRANSITION_REPLAY_FORMAT | ADMISSION_TERMINAL_PROJECTION_DESCRIPTOR_KIND)
    );
    let reservation_consumer =
        options.consumer_descriptor_kind == Some(CHANNEL_TRANSITION_REPLAY_FORMAT);
    if channel_content && !channel_consumer || !channel_content && reservation_consumer {
        return Err(EconomicStateCacheError::Conflict);
    }
    Ok(())
}

#[allow(clippy::too_many_arguments)]
pub(crate) fn stage_channel_batch_in_transaction(
    transaction: &Transaction<'_>,
    advance: &VerifiedEconomicStateBatchAdvance,
    operation: EconomicOperationStageContext<'_>,
    descriptor: EconomicStateStageDescriptor,
    active_fence: &StoreMutationFence,
    trusted_now_unix_ms: u64,
    serving_owner: &SqliteServingOwner,
) -> Result<EconomicStateStageRecord, EconomicStateCacheError> {
    validate_trusted_time(trusted_now_unix_ms)?;
    let options = EconomicStageOptions {
        operation: Some(operation),
        descriptor: Some(descriptor),
        admission_checkpoint: None,
        consumer_descriptor_kind: Some(CHANNEL_TRANSITION_REPLAY_FORMAT),
    };
    validate_stage_options(advance.batch(), &options)?;
    stage_batch_in_transaction(
        transaction,
        advance,
        options,
        active_fence,
        trusted_now_unix_ms,
        serving_owner,
    )
}

pub(crate) fn record_channel_anchor_advanced_in_transaction(
    transaction: &Transaction<'_>,
    advance: &VerifiedEconomicStateBatchAdvance,
    committed: &VerifiedEconomicStateView,
    trusted_now_unix_ms: u64,
    serving_owner: &SqliteServingOwner,
) -> Result<EconomicStateStageRecord, EconomicStateCacheError> {
    validate_trusted_time(trusted_now_unix_ms)?;
    if !committed_view_matches_batch(committed.view(), advance.batch()) {
        return Err(EconomicStateCacheError::Conflict);
    }
    let committed_bytes = canonical_bounded(committed.view(), MAX_VIEW_BYTES, "committed view")?;
    let mut record = load_stage_tx(transaction, &advance.batch().batch_id)?
        .ok_or(EconomicStateCacheError::NotFound)?;
    if record.base_view != *advance.current().view()
        || record.batch != *advance.batch()
        || record
            .descriptor
            .as_ref()
            .is_none_or(|descriptor| descriptor.kind != CHANNEL_TRANSITION_REPLAY_FORMAT)
        || !is_protected_channel_stage(&record)
    {
        return Err(EconomicStateCacheError::Conflict);
    }
    if matches!(
        record.status,
        EconomicStateStageStatus::EconomicAnchorAdvanced | EconomicStateStageStatus::DbFinalized
    ) {
        return if record.committed_view.as_ref() == Some(committed.view()) {
            Ok(record)
        } else {
            Err(EconomicStateCacheError::Conflict)
        };
    }
    require_transition(
        record.status,
        EconomicStateStageStatus::EconomicAnchorAdvanced,
    )?;
    record.status = EconomicStateStageStatus::EconomicAnchorAdvanced;
    record.committed_view = Some(committed.view().clone());
    record.version = next_version(record.version)?;
    record.updated_at_unix_ms = monotonic_time(&record, trusted_now_unix_ms)?;
    record.snapshot_digest = stage_snapshot_digest(&record, &[])?;
    update_stage(
        transaction,
        &record,
        Some(committed_bytes),
        None,
        EconomicStateStageStatus::DbStaged,
    )?;
    append_stage_commit(
        transaction,
        &record,
        "channel_anchor_advanced",
        serving_owner,
    )?;
    Ok(record)
}

fn stage_batch_in_transaction(
    transaction: &Transaction<'_>,
    advance: &VerifiedEconomicStateBatchAdvance,
    options: EconomicStageOptions<'_>,
    active_fence: &StoreMutationFence,
    trusted_now_unix_ms: u64,
    serving_owner: &SqliteServingOwner,
) -> Result<EconomicStateStageRecord, EconomicStateCacheError> {
    let EconomicStageOptions {
        operation,
        descriptor,
        admission_checkpoint,
        consumer_descriptor_kind: _,
    } = options;
    if let Some(existing) = load_stage_tx(transaction, &advance.batch().batch_id)? {
        if existing.base_view != *advance.current().view() || existing.batch != *advance.batch() {
            return Err(EconomicStateCacheError::Conflict);
        }
        let qualified = qualify_operation_binding(
            transaction,
            advance.batch(),
            operation,
            active_fence,
            trusted_now_unix_ms,
        )?;
        if qualified != existing.operation_binding || descriptor != existing.descriptor {
            return Err(EconomicStateCacheError::Conflict);
        }
        return Ok(existing);
    }
    if let Some(descriptor) = &descriptor {
        let retained_batch_id = transaction
            .query_row(
                r#"
                SELECT batch_id FROM economic_state_stages
                WHERE descriptor_kind = ?1 AND descriptor_key = ?2
                "#,
                params![descriptor.kind(), descriptor.key()],
                |row| row.get::<_, String>(0),
            )
            .optional()
            .map_err(sqlite_error)?;
        if retained_batch_id.is_some() {
            return Err(EconomicStateCacheError::Conflict);
        }
    }
    if let Some(checkpoint) = admission_checkpoint {
        verify_stage_admission_checkpoint(
            transaction,
            &checkpoint,
            active_fence,
            trusted_now_unix_ms,
        )?;
    }
    let operation_binding = qualify_operation_binding(
        transaction,
        advance.batch(),
        operation,
        active_fence,
        trusted_now_unix_ms,
    )?;
    let operation_binding_bytes = operation_binding
        .as_ref()
        .map(canonical_json_bytes)
        .transpose()
        .map_err(canonical_error)?;
    let mut record = EconomicStateStageRecord {
        base_view: advance.current().view().clone(),
        batch: advance.batch().clone(),
        committed_view: None,
        operation_binding,
        descriptor,
        status: EconomicStateStageStatus::DbStaged,
        reason: None,
        version: 1,
        created_at_unix_ms: trusted_now_unix_ms,
        updated_at_unix_ms: trusted_now_unix_ms,
        snapshot_digest: String::new(),
    };
    record.snapshot_digest = stage_snapshot_digest(&record, &[])?;
    let base_view_bytes = canonical_bounded(advance.current().view(), MAX_VIEW_BYTES, "base view")?;
    let batch_bytes = advance.batch().canonical_bytes()?;
    transaction
        .execute(
            r#"
            INSERT INTO economic_state_stages (
                batch_id, checkpoint_sequence, checkpoint_digest,
                base_view_json, batch_json, committed_view_json,
                operation_binding_json, descriptor_kind, descriptor_key,
                descriptor_digest, descriptor_json, status, reason,
                stage_version, snapshot_digest, created_at_unix_ms,
                updated_at_unix_ms
            ) VALUES (
                ?1, ?2, ?3, ?4, ?5, NULL, ?6, ?7, ?8, ?9, ?10,
                ?11, NULL, 1, ?12, ?13, ?13
            )
            "#,
            params![
                &record.batch.batch_id,
                sqlite_i64(record.batch.checkpoint_sequence, "checkpoint_sequence")?,
                &record.batch.checkpoint_digest,
                base_view_bytes,
                batch_bytes,
                operation_binding_bytes,
                record.descriptor.as_ref().map(|value| value.kind.as_str()),
                record.descriptor.as_ref().map(|value| value.key.as_str()),
                record
                    .descriptor
                    .as_ref()
                    .map(|value| value.digest.as_str()),
                record
                    .descriptor
                    .as_ref()
                    .map(|value| value.canonical_json.as_slice()),
                record.status.as_str(),
                &record.snapshot_digest,
                sqlite_i64(trusted_now_unix_ms, "trusted_now_unix_ms")?,
            ],
        )
        .map_err(sqlite_error)?;
    append_stage_commit(transaction, &record, "stage_batch", serving_owner)?;
    Ok(record)
}

#[derive(Clone)]
pub struct SqliteEconomicStateCache {
    connection: Arc<Mutex<Connection>>,
    serving_owner: Arc<SqliteServingOwner>,
}

impl SqliteEconomicStateCache {
    pub(crate) fn open_alongside(
        connection: Arc<Mutex<Connection>>,
        serving_owner: Arc<SqliteServingOwner>,
    ) -> Self {
        Self {
            connection,
            serving_owner,
        }
    }

    fn connection(&self) -> Result<MutexGuard<'_, Connection>, EconomicStateCacheError> {
        self.connection
            .lock()
            .map_err(|_| invariant("economic state cache lock is poisoned"))
    }

    fn begin_read<'a>(
        &self,
        connection: &'a mut Connection,
    ) -> Result<Transaction<'a>, EconomicStateCacheError> {
        let transaction = connection
            .transaction_with_behavior(TransactionBehavior::Deferred)
            .map_err(sqlite_error)?;
        verify_active_owner(&transaction, &self.serving_owner, None)?;
        self.serving_owner
            .verify_authority_anchor(&transaction)
            .map_err(map_owner_error)?;
        verify_cache_sql_invariants(&transaction)?;
        Ok(transaction)
    }

    fn begin_write<'a>(
        &self,
        connection: &'a mut Connection,
        fence: &StoreMutationFence,
    ) -> Result<Transaction<'a>, EconomicStateCacheError> {
        let transaction = connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(sqlite_error)?;
        verify_active_owner(&transaction, &self.serving_owner, Some(fence))?;
        self.serving_owner
            .verify_authority_anchor(&transaction)
            .map_err(map_owner_error)?;
        verify_cache_sql_invariants(&transaction)?;
        Ok(transaction)
    }

    fn commit_write(&self, transaction: Transaction<'_>) -> Result<(), EconomicStateCacheError> {
        transaction.commit().map_err(|error| {
            map_owner_error(self.serving_owner.outcome_unknown(format!(
                "sqlite economic state cache commit outcome is unknown: {error}"
            )))
        })
    }

    fn sync_after_write(&self, connection: &Connection) -> Result<(), EconomicStateCacheError> {
        self.serving_owner
            .sync_authority_anchor(connection)
            .map_err(map_owner_error)
    }

    pub fn stage_batch(
        &self,
        advance: &VerifiedEconomicStateBatchAdvance,
        operation: Option<EconomicOperationStageContext<'_>>,
        active_fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<EconomicStateStageRecord, EconomicStateCacheError> {
        self.stage_batch_with_descriptor(
            advance,
            operation,
            None,
            active_fence,
            trusted_now_unix_ms,
        )
    }

    pub fn stage_batch_with_descriptor(
        &self,
        advance: &VerifiedEconomicStateBatchAdvance,
        operation: Option<EconomicOperationStageContext<'_>>,
        descriptor: Option<EconomicStateStageDescriptor>,
        active_fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<EconomicStateStageRecord, EconomicStateCacheError> {
        self.stage_batch_inner(
            advance,
            EconomicStageOptions {
                operation,
                descriptor,
                admission_checkpoint: None,
                consumer_descriptor_kind: None,
            },
            active_fence,
            trusted_now_unix_ms,
        )
    }

    pub fn stage_admission_terminal_projection(
        &self,
        advance: &VerifiedEconomicStateBatchAdvance,
        operation: EconomicOperationStageContext<'_>,
        envelope: &SignedAdmissionTerminalProjectionV1,
        active_fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<EconomicStateStageRecord, EconomicStateCacheError> {
        validate_trusted_time(trusted_now_unix_ms)?;
        let verified = envelope
            .verify()
            .map_err(|_| EconomicStateCacheError::Conflict)?;
        let descriptor = admission_terminal_projection_descriptor(envelope, &verified)?;
        let context = verified.context();
        let source = verified.source_operation();
        let lease = operation.recovery_lease;
        let expected_claimant = format!("kernel:{}", verified.signer_key().to_hex());
        if operation.operation != source
            || context.operation_id != *source.binding().operation_id()
            || context.request_id != source.replay_key().request_id
            || context.expected_operation_version != source.version()
            || context.coordinator_lease_id != *lease.coordinator_lease_id()
            || context.coordinator_lease_epoch != lease.coordinator_lease_epoch()
            || context.store_fence != *lease.store_fence()
            || lease.claimant_id().as_str() != expected_claimant
            || context.trusted_time_unix_ms > trusted_now_unix_ms
            || context.trusted_time_unix_ms >= lease.expires_at_unix_ms()
        {
            return Err(EconomicStateCacheError::Conflict);
        }
        qualify_terminal_projection_advance(advance, &descriptor, &verified)?;
        self.stage_batch_inner(
            advance,
            EconomicStageOptions {
                operation: Some(operation),
                descriptor: Some(descriptor),
                admission_checkpoint: None,
                consumer_descriptor_kind: Some(ADMISSION_TERMINAL_PROJECTION_DESCRIPTOR_KIND),
            },
            active_fence,
            trusted_now_unix_ms,
        )
    }

    pub(crate) fn stage_clearing_lifecycle_batch(
        &self,
        advance: &VerifiedEconomicStateBatchAdvance,
        descriptor: EconomicStateStageDescriptor,
        checkpoint: Option<EconomicStageAdmissionCheckpoint<'_>>,
        active_fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<EconomicStateStageRecord, EconomicStateCacheError> {
        self.stage_batch_inner(
            advance,
            EconomicStageOptions {
                operation: None,
                descriptor: Some(descriptor),
                admission_checkpoint: checkpoint,
                consumer_descriptor_kind: Some(CLEARING_LIFECYCLE_REPLAY_DESCRIPTOR_KIND),
            },
            active_fence,
            trusted_now_unix_ms,
        )
    }

    fn stage_batch_inner(
        &self,
        advance: &VerifiedEconomicStateBatchAdvance,
        options: EconomicStageOptions<'_>,
        active_fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<EconomicStateStageRecord, EconomicStateCacheError> {
        validate_trusted_time(trusted_now_unix_ms)?;
        validate_stage_options(advance.batch(), &options)?;
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, active_fence)?;
        let record = stage_batch_in_transaction(
            &transaction,
            advance,
            options,
            active_fence,
            trusted_now_unix_ms,
            &self.serving_owner,
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)?;
        Ok(record)
    }

    pub fn record_anchor_advanced(
        &self,
        advance: &VerifiedEconomicStateBatchAdvance,
        committed: &VerifiedEconomicStateView,
        pins: &EconomicStateAnchorPins,
        active_fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<EconomicStateStageRecord, EconomicStateCacheError> {
        validate_trusted_time(trusted_now_unix_ms)?;
        verify_economic_state_batch_commit(advance, committed, pins)?;
        if let Some(descriptor) = self
            .load_stage(&advance.batch().batch_id)?
            .and_then(|stage| stage.descriptor().cloned())
            .filter(|descriptor| descriptor.kind() == ADMISSION_TERMINAL_PROJECTION_DESCRIPTOR_KIND)
        {
            let envelope = descriptor.decode::<SignedAdmissionTerminalProjectionV1>()?;
            let verified = envelope
                .verify()
                .map_err(|_| EconomicStateCacheError::Conflict)?;
            let terminal_slot =
                qualify_terminal_projection_advance(advance, &descriptor, &verified)?;
            if !committed_view_matches_batch(committed.view(), advance.batch()) {
                return Err(EconomicStateCacheError::Conflict);
            }
            if terminal_slot.state == EconomicEffectStateV1::Completed {
                verify_economic_completed_effect(committed, &terminal_slot)?;
            }
        }
        if self
            .load_stage(&advance.batch().batch_id)?
            .is_some_and(|stage| {
                stage
                    .descriptor()
                    .is_some_and(|descriptor| descriptor.kind() == CHANNEL_TRANSITION_REPLAY_FORMAT)
                    || (batch_contains_channel_content(stage.batch())
                        && stage.descriptor().is_none_or(|descriptor| {
                            descriptor.kind() != ADMISSION_TERMINAL_PROJECTION_DESCRIPTOR_KIND
                        }))
            })
        {
            return Err(EconomicStateCacheError::Conflict);
        }
        let committed_bytes =
            canonical_bounded(committed.view(), MAX_VIEW_BYTES, "committed view")?;
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, active_fence)?;
        let mut record = load_stage_tx(&transaction, &advance.batch().batch_id)?
            .ok_or(EconomicStateCacheError::NotFound)?;
        if record.base_view != *advance.current().view() || record.batch != *advance.batch() {
            return Err(EconomicStateCacheError::Conflict);
        }
        if matches!(
            record.status,
            EconomicStateStageStatus::EconomicAnchorAdvanced
                | EconomicStateStageStatus::DbFinalized
        ) {
            if record.committed_view.as_ref() == Some(committed.view()) {
                transaction.commit().map_err(sqlite_error)?;
                return Ok(record);
            }
            return Err(EconomicStateCacheError::Conflict);
        }
        require_transition(
            record.status,
            EconomicStateStageStatus::EconomicAnchorAdvanced,
        )?;
        record.status = EconomicStateStageStatus::EconomicAnchorAdvanced;
        record.committed_view = Some(committed.view().clone());
        record.version = next_version(record.version)?;
        record.updated_at_unix_ms = monotonic_time(&record, trusted_now_unix_ms)?;
        record.snapshot_digest = stage_snapshot_digest(&record, &[])?;
        update_stage(
            &transaction,
            &record,
            Some(committed_bytes),
            None,
            EconomicStateStageStatus::DbStaged,
        )?;
        append_stage_commit(
            &transaction,
            &record,
            "anchor_advanced",
            &self.serving_owner,
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)?;
        Ok(record)
    }

    pub fn finalize_stage(
        &self,
        batch_id: &str,
        active_fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<EconomicStateStageRecord, EconomicStateCacheError> {
        validate_digest(batch_id, "batch_id")?;
        validate_trusted_time(trusted_now_unix_ms)?;
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, active_fence)?;
        if load_stage_tx(&transaction, batch_id)?.is_some_and(|record| {
            is_protected_channel_stage(&record)
                || record.descriptor.as_ref().is_some_and(|descriptor| {
                    descriptor.kind == ADMISSION_TERMINAL_PROJECTION_DESCRIPTOR_KIND
                })
        }) {
            return Err(EconomicStateCacheError::Conflict);
        }
        let record = finalize_stage_in_transaction(
            &transaction,
            batch_id,
            &self.serving_owner,
            trusted_now_unix_ms,
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)?;
        Ok(record)
    }

    pub fn discard_unanchored_stage(
        &self,
        batch_id: &str,
        reason: &str,
        active_fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<EconomicStateStageRecord, EconomicStateCacheError> {
        self.close_stage(
            batch_id,
            reason,
            EconomicStateStageStatus::Discarded,
            active_fence,
            trusted_now_unix_ms,
        )
    }

    pub fn quarantine_stage(
        &self,
        batch_id: &str,
        reason: &str,
        active_fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<EconomicStateStageRecord, EconomicStateCacheError> {
        self.close_stage(
            batch_id,
            reason,
            EconomicStateStageStatus::Quarantined,
            active_fence,
            trusted_now_unix_ms,
        )
    }

    fn close_stage(
        &self,
        batch_id: &str,
        reason: &str,
        target: EconomicStateStageStatus,
        active_fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<EconomicStateStageRecord, EconomicStateCacheError> {
        validate_digest(batch_id, "batch_id")?;
        validate_reason(reason)?;
        validate_trusted_time(trusted_now_unix_ms)?;
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection, active_fence)?;
        let mut record =
            load_stage_tx(&transaction, batch_id)?.ok_or(EconomicStateCacheError::NotFound)?;
        if is_protected_channel_stage(&record)
            || record.descriptor.as_ref().is_some_and(|descriptor| {
                descriptor.kind == ADMISSION_TERMINAL_PROJECTION_DESCRIPTOR_KIND
            })
        {
            return Err(EconomicStateCacheError::Conflict);
        }
        if record.status == target && record.reason.as_deref() == Some(reason) {
            transaction.commit().map_err(sqlite_error)?;
            return Ok(record);
        }
        let previous_status = record.status;
        require_transition(previous_status, target)?;
        record.status = target;
        record.reason = Some(reason.to_owned());
        record.version = next_version(record.version)?;
        record.updated_at_unix_ms = monotonic_time(&record, trusted_now_unix_ms)?;
        record.snapshot_digest = stage_snapshot_digest(&record, &[])?;
        update_stage(&transaction, &record, None, Some(reason), previous_status)?;
        append_stage_commit(
            &transaction,
            &record,
            if target == EconomicStateStageStatus::Discarded {
                "discard_stage"
            } else {
                "quarantine_stage"
            },
            &self.serving_owner,
        )?;
        self.commit_write(transaction)?;
        self.sync_after_write(&connection)?;
        Ok(record)
    }

    pub fn load_stage(
        &self,
        batch_id: &str,
    ) -> Result<Option<EconomicStateStageRecord>, EconomicStateCacheError> {
        validate_digest(batch_id, "batch_id")?;
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let record = load_stage_tx(&transaction, batch_id)?;
        transaction.commit().map_err(sqlite_error)?;
        Ok(record)
    }

    pub(crate) fn load_stage_by_descriptor(
        &self,
        kind: &str,
        key: &str,
    ) -> Result<Option<EconomicStateStageRecord>, EconomicStateCacheError> {
        validate_descriptor_text(kind, MAX_DESCRIPTOR_KIND_BYTES, "stage descriptor kind")?;
        validate_descriptor_text(key, MAX_DESCRIPTOR_KEY_BYTES, "stage descriptor key")?;
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let batch_id = transaction
            .query_row(
                r#"
                SELECT batch_id FROM economic_state_stages
                WHERE descriptor_kind = ?1 AND descriptor_key = ?2
                "#,
                params![kind, key],
                |row| row.get::<_, String>(0),
            )
            .optional()
            .map_err(sqlite_error)?;
        let record = batch_id
            .as_deref()
            .map(|batch_id| load_stage_tx(&transaction, batch_id))
            .transpose()?
            .flatten();
        transaction.commit().map_err(sqlite_error)?;
        Ok(record)
    }

    pub fn list_pending(
        &self,
        limit: usize,
    ) -> Result<Vec<EconomicStateStageRecord>, EconomicStateCacheError> {
        if limit == 0 || limit > 256 {
            return Err(invariant("economic recovery limit must be within 1..=256"));
        }
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let mut statement = transaction
            .prepare(
                r#"
                SELECT batch_id FROM economic_state_stages
                WHERE status IN ('db_staged', 'economic_anchor_advanced')
                ORDER BY checkpoint_sequence, batch_id LIMIT ?1
                "#,
            )
            .map_err(sqlite_error)?;
        let batch_ids = statement
            .query_map(
                [i64::try_from(limit).map_err(|_| invariant("limit overflow"))?],
                |row| row.get::<_, String>(0),
            )
            .map_err(sqlite_error)?
            .collect::<Result<Vec<_>, _>>()
            .map_err(sqlite_error)?;
        drop(statement);
        let records = batch_ids
            .iter()
            .map(|batch_id| {
                load_stage_tx(&transaction, batch_id)?
                    .ok_or_else(|| invariant("pending economic stage disappeared"))
            })
            .collect::<Result<Vec<_>, _>>()?;
        transaction.commit().map_err(sqlite_error)?;
        Ok(records)
    }

    pub fn load_finalized_head(
        &self,
        key: &EconomicResourceKeyV1,
    ) -> Result<Option<EconomicResourceHeadV1>, EconomicStateCacheError> {
        key.validate()?;
        let key_bytes = canonical_json_bytes(key).map_err(canonical_error)?;
        let key_digest = sha256_hex(&key_bytes);
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let stored = transaction
            .query_row(
                r#"
                SELECT current.resource_key_json, current.head_digest,
                       current.head_json, current.source_batch_id,
                       staged.head_digest, staged.head_json
                FROM economic_state_heads AS current
                JOIN economic_state_stage_heads AS staged
                  ON staged.batch_id = current.source_batch_id
                 AND staged.resource_key_digest = current.resource_key_digest
                JOIN economic_state_stages AS stage
                  ON stage.batch_id = current.source_batch_id
                 AND stage.status = 'db_finalized'
                WHERE current.resource_key_digest = ?1
                "#,
                [&key_digest],
                |row| {
                    Ok((
                        row.get::<_, Vec<u8>>(0)?,
                        row.get::<_, String>(1)?,
                        row.get::<_, Vec<u8>>(2)?,
                        row.get::<_, String>(3)?,
                        row.get::<_, String>(4)?,
                        row.get::<_, Vec<u8>>(5)?,
                    ))
                },
            )
            .optional()
            .map_err(sqlite_error)?;
        let result = stored
            .map(
                |(stored_key, current_digest, current_head, _batch, staged_digest, staged_head)| {
                    if stored_key != key_bytes
                        || current_digest != staged_digest
                        || current_head != staged_head
                    {
                        return Err(invariant(
                            "current economic head lost its finalized projection",
                        ));
                    }
                    let head: EconomicResourceHeadV1 =
                        decode_exact(&current_head, "resource head")?;
                    if head.resource_key != *key || head.digest()? != current_digest {
                        return Err(invariant("cached economic resource head is corrupt"));
                    }
                    Ok(head)
                },
            )
            .transpose()?;
        transaction.commit().map_err(sqlite_error)?;
        Ok(result)
    }
}

pub(crate) fn load_anchored_terminal_projection_in_transaction(
    transaction: &Transaction<'_>,
    batch_id: &str,
    active_fence: &StoreMutationFence,
    require_anchored: bool,
) -> Result<
    (
        VerifiedAdmissionTerminalProjectionV1,
        EconomicOperationStageBinding,
    ),
    EconomicStateCacheError,
> {
    validate_digest(batch_id, "batch_id")?;
    verify_cache_sql_invariants(transaction)?;
    let record = load_stage_tx(transaction, batch_id)?.ok_or(EconomicStateCacheError::NotFound)?;
    if matches!(
        record.status,
        EconomicStateStageStatus::Discarded | EconomicStateStageStatus::Quarantined
    ) || require_anchored
        && !matches!(
            record.status,
            EconomicStateStageStatus::EconomicAnchorAdvanced
                | EconomicStateStageStatus::DbFinalized
        )
    {
        return Err(EconomicStateCacheError::Conflict);
    }
    let descriptor = record
        .descriptor
        .as_ref()
        .ok_or(EconomicStateCacheError::Conflict)?;
    let envelope = descriptor.decode::<SignedAdmissionTerminalProjectionV1>()?;
    let verified = envelope
        .verify()
        .map_err(|_| EconomicStateCacheError::Conflict)?;
    let context = verified.context();
    let source = verified.source_operation();
    let operation_id = source.binding().operation_id().as_str();
    let binding = record
        .operation_binding
        .as_ref()
        .ok_or(EconomicStateCacheError::Conflict)?;
    qualify_retained_terminal_projection_advance(
        &record.base_view,
        &record.batch,
        descriptor,
        &verified,
    )?;
    let committed_head_matches = match record.committed_view.as_ref() {
        Some(committed) => committed_view_matches_batch(committed, &record.batch),
        None => record.status == EconomicStateStageStatus::DbStaged,
    };
    let expected_claimant = format!("kernel:{}", verified.signer_key().to_hex());
    if descriptor.kind != ADMISSION_TERMINAL_PROJECTION_DESCRIPTOR_KIND
        || descriptor.key != operation_id
        || record.batch.operation_id.as_deref() != Some(operation_id)
        || binding.operation_id != operation_id
        || binding.operation_state != source.state()
        || binding.operation_version != source.version()
        || binding.coordinator_lease_epoch != source.coordinator_lease_epoch()
        || binding.coordinator_lease_id != context.coordinator_lease_id.as_str()
        || binding.recovery_claimant_id != expected_claimant
        || binding.recovery_expires_at_unix_ms <= context.trusted_time_unix_ms
        || !current_fence_serves_historical(&binding.store_fence, active_fence)
        || context.store_fence != binding.store_fence
        || context.trusted_time_unix_ms > record.created_at_unix_ms
        || !committed_head_matches
    {
        return Err(EconomicStateCacheError::Conflict);
    }
    Ok((verified, binding.clone()))
}

pub(crate) fn has_reserved_terminal_stage(
    transaction: &Transaction<'_>,
    operation_id: &str,
) -> Result<bool, EconomicStateCacheError> {
    transaction
        .query_row(
            r#"
            SELECT EXISTS(
                SELECT 1 FROM economic_state_stages
                WHERE descriptor_kind = ?1
                  AND descriptor_key = ?2
            )
            "#,
            params![ADMISSION_TERMINAL_PROJECTION_DESCRIPTOR_KIND, operation_id],
            |row| row.get(0),
        )
        .map_err(sqlite_error)
}

fn current_fence_serves_historical(
    historical: &StoreMutationFence,
    current: &StoreMutationFence,
) -> bool {
    historical.store_uuid == current.store_uuid
        && (historical == current || current.owner_epoch > historical.owner_epoch)
}

pub(crate) fn finalize_stage_in_transaction(
    transaction: &Transaction<'_>,
    batch_id: &str,
    serving_owner: &SqliteServingOwner,
    trusted_now_unix_ms: u64,
) -> Result<EconomicStateStageRecord, EconomicStateCacheError> {
    validate_digest(batch_id, "batch_id")?;
    validate_trusted_time(trusted_now_unix_ms)?;
    verify_cache_sql_invariants(transaction)?;
    let mut record =
        load_stage_tx(transaction, batch_id)?.ok_or(EconomicStateCacheError::NotFound)?;
    if record.status == EconomicStateStageStatus::DbFinalized {
        return Ok(record);
    }
    require_transition(record.status, EconomicStateStageStatus::DbFinalized)?;
    let committed = record
        .committed_view
        .as_ref()
        .ok_or_else(|| invariant("anchor-advanced stage omitted its committed view"))?;
    let mut head_digests = Vec::with_capacity(record.batch.transitions.len());
    for transition in &record.batch.transitions {
        let head = committed
            .heads
            .iter()
            .find(|head| head.resource_key == transition.resource_key)
            .ok_or_else(|| invariant("committed view omitted a staged resource head"))?;
        if head != &transition.next_head {
            return Err(invariant("committed resource head changed before finalize"));
        }
        let key_bytes = canonical_json_bytes(&head.resource_key).map_err(canonical_error)?;
        let key_digest = sha256_hex(&key_bytes);
        let head_bytes = canonical_json_bytes(head).map_err(canonical_error)?;
        let head_digest = head.digest()?;
        head_digests.push((key_digest.clone(), head_digest.clone()));
        transaction
            .execute(
                r#"
                INSERT INTO economic_state_stage_heads (
                    batch_id, resource_key_digest, resource_key_json,
                    head_digest, head_json
                ) VALUES (?1, ?2, ?3, ?4, ?5)
                "#,
                params![batch_id, &key_digest, &key_bytes, &head_digest, &head_bytes],
            )
            .map_err(sqlite_error)?;
        let checkpoint_sequence =
            sqlite_i64(record.batch.checkpoint_sequence, "checkpoint_sequence")?;
        let current = transaction
            .query_row(
                r#"
                SELECT resource_key_json, head_digest, head_json,
                       checkpoint_sequence, checkpoint_digest, source_batch_id
                FROM economic_state_heads WHERE resource_key_digest = ?1
                "#,
                [&key_digest],
                |row| {
                    Ok((
                        row.get::<_, Vec<u8>>(0)?,
                        row.get::<_, String>(1)?,
                        row.get::<_, Vec<u8>>(2)?,
                        row.get::<_, i64>(3)?,
                        row.get::<_, String>(4)?,
                        row.get::<_, String>(5)?,
                    ))
                },
            )
            .optional()
            .map_err(sqlite_error)?;
        let publish = match current {
            Some((
                current_key,
                current_head_digest,
                current_head,
                current_sequence,
                current_checkpoint,
                current_batch,
            )) if current_sequence == checkpoint_sequence => {
                if current_key != key_bytes
                    || current_head_digest != head_digest
                    || current_head != head_bytes
                    || current_checkpoint != record.batch.checkpoint_digest
                    || current_batch != batch_id
                {
                    return Err(invariant(
                        "equal economic checkpoint has conflicting cached state",
                    ));
                }
                false
            }
            Some((_, _, _, current_sequence, _, _)) if current_sequence > checkpoint_sequence => {
                false
            }
            _ => true,
        };
        if publish {
            transaction
                .execute(
                    r#"
                INSERT INTO economic_state_heads (
                    resource_key_digest, resource_key_json, head_digest,
                    head_json, checkpoint_sequence, checkpoint_digest,
                    source_batch_id
                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
                ON CONFLICT(resource_key_digest) DO UPDATE SET
                    resource_key_json = excluded.resource_key_json,
                    head_digest = excluded.head_digest,
                    head_json = excluded.head_json,
                    checkpoint_sequence = excluded.checkpoint_sequence,
                    checkpoint_digest = excluded.checkpoint_digest,
                    source_batch_id = excluded.source_batch_id
                WHERE economic_state_heads.checkpoint_sequence
                    < excluded.checkpoint_sequence
                "#,
                    params![
                        &key_digest,
                        &key_bytes,
                        &head_digest,
                        &head_bytes,
                        checkpoint_sequence,
                        &record.batch.checkpoint_digest,
                        batch_id,
                    ],
                )
                .map_err(sqlite_error)?;
        }
    }
    head_digests.sort();
    record.status = EconomicStateStageStatus::DbFinalized;
    record.version = next_version(record.version)?;
    record.updated_at_unix_ms = monotonic_time(&record, trusted_now_unix_ms)?;
    record.snapshot_digest = stage_snapshot_digest(&record, &head_digests)?;
    update_stage(
        transaction,
        &record,
        None,
        None,
        EconomicStateStageStatus::EconomicAnchorAdvanced,
    )?;
    append_stage_commit(transaction, &record, "finalize_stage", serving_owner)?;
    Ok(record)
}

pub(crate) fn initialize_economic_state_cache_schema(
    connection: &mut Connection,
) -> Result<(), EconomicStateCacheError> {
    let on_disk = crate::check_schema_version(
        connection,
        ECONOMIC_STATE_CACHE_SCHEMA_KEY,
        ECONOMIC_STATE_CACHE_SUPPORTED_SCHEMA_VERSION,
        ECONOMIC_STATE_CACHE_SCHEMA_ANCHORS,
    )
    .map_err(|error| invariant(error.to_string()))?;
    if on_disk == ECONOMIC_STATE_CACHE_SUPPORTED_SCHEMA_VERSION {
        return verify_cache_sql_invariants(connection);
    }
    let transaction = connection
        .transaction_with_behavior(TransactionBehavior::Immediate)
        .map_err(sqlite_error)?;
    let stage_table_exists = transaction
        .query_row(
            "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE type = 'table' AND name = 'economic_state_stages')",
            [],
            |row| row.get::<_, bool>(0),
        )
        .map_err(sqlite_error)?;
    let descriptor_column_exists = stage_table_exists
        && transaction
            .query_row(
                "SELECT EXISTS(SELECT 1 FROM pragma_table_info('economic_state_stages') WHERE name = 'descriptor_kind')",
                [],
                |row| row.get::<_, bool>(0),
            )
            .map_err(sqlite_error)?;
    if stage_table_exists && !descriptor_column_exists {
        transaction
            .execute_batch(ECONOMIC_STATE_CACHE_DESCRIPTOR_MIGRATION)
            .map_err(sqlite_error)?;
    }
    transaction
        .execute_batch(ECONOMIC_STATE_CACHE_SCHEMA)
        .map_err(sqlite_error)?;
    crate::stamp_schema_version(
        &transaction,
        ECONOMIC_STATE_CACHE_SCHEMA_KEY,
        ECONOMIC_STATE_CACHE_SUPPORTED_SCHEMA_VERSION,
    )
    .map_err(|error| invariant(error.to_string()))?;
    verify_cache_sql_invariants(&transaction)?;
    transaction.commit().map_err(sqlite_error)
}

fn qualify_operation_binding(
    transaction: &Transaction<'_>,
    batch: &EconomicStateBatchV1,
    context: Option<EconomicOperationStageContext<'_>>,
    active_fence: &StoreMutationFence,
    trusted_now_unix_ms: u64,
) -> Result<Option<EconomicOperationStageBinding>, EconomicStateCacheError> {
    let Some(context) = context else {
        if batch.operation_id.is_some() {
            return Err(invariant(
                "operation-bound batch omitted recovery authority",
            ));
        }
        return Ok(None);
    };
    let operation = context.operation;
    let lease = context.recovery_lease;
    operation
        .validate()
        .map_err(|error| invariant(error.to_string()))?;
    if operation.state().is_terminal()
        || batch.operation_id.as_deref() != Some(operation.binding().operation_id().as_str())
        || lease.operation_id() != operation.binding().operation_id()
        || lease.claimed_version() != operation.version()
        || lease.coordinator_lease_epoch() != operation.coordinator_lease_epoch()
        || lease.store_fence() != active_fence
        || trusted_now_unix_ms >= lease.expires_at_unix_ms()
        || context
            .not_after_unix_ms
            .is_some_and(|not_after| trusted_now_unix_ms >= not_after)
    {
        return Err(EconomicStateCacheError::Fenced);
    }
    let persisted = transaction
        .query_row(
            r#"
            SELECT operation_json, recovery_claimant_id,
                   recovery_coordinator_lease_id, recovery_coordinator_lease_epoch,
                   recovery_claimed_version, recovery_expires_at_unix_ms,
                   recovery_store_uuid, recovery_store_lease_id,
                   recovery_store_owner_epoch
            FROM admission_operations WHERE operation_id = ?1 AND terminal = 0
            "#,
            [operation.binding().operation_id().as_str()],
            |row| {
                Ok((
                    row.get::<_, Vec<u8>>(0)?,
                    row.get::<_, Option<String>>(1)?,
                    row.get::<_, Option<String>>(2)?,
                    row.get::<_, Option<i64>>(3)?,
                    row.get::<_, Option<i64>>(4)?,
                    row.get::<_, Option<i64>>(5)?,
                    row.get::<_, Option<String>>(6)?,
                    row.get::<_, Option<String>>(7)?,
                    row.get::<_, Option<i64>>(8)?,
                ))
            },
        )
        .optional()
        .map_err(sqlite_error)?
        .ok_or(EconomicStateCacheError::Fenced)?;
    let stored: PersistedAdmissionOperationV1 = decode_exact(&persisted.0, "admission operation")?;
    let stored = AdmissionOperationV1::from_persisted(stored)
        .map_err(|error| invariant(error.to_string()))?;
    let exact_claim = persisted.1.as_deref() == Some(lease.claimant_id().as_str())
        && persisted.2.as_deref() == Some(lease.coordinator_lease_id().as_str())
        && optional_u64(persisted.3, "recovery_coordinator_lease_epoch")?
            == Some(lease.coordinator_lease_epoch())
        && optional_u64(persisted.4, "recovery_claimed_version")? == Some(lease.claimed_version())
        && optional_u64(persisted.5, "recovery_expires_at_unix_ms")?
            == Some(lease.expires_at_unix_ms())
        && persisted.6.as_deref() == Some(&active_fence.store_uuid)
        && persisted.7.as_deref() == Some(&active_fence.lease_id)
        && optional_u64(persisted.8, "recovery_store_owner_epoch")?
            == Some(active_fence.owner_epoch);
    if stored != *operation || !exact_claim {
        return Err(EconomicStateCacheError::Fenced);
    }
    let binding = EconomicOperationStageBinding {
        operation_id: operation.binding().operation_id().as_str().to_owned(),
        operation_state: operation.state(),
        operation_version: operation.version(),
        coordinator_lease_epoch: operation.coordinator_lease_epoch(),
        coordinator_lease_id: lease.coordinator_lease_id().as_str().to_owned(),
        recovery_claimant_id: lease.claimant_id().as_str().to_owned(),
        recovery_expires_at_unix_ms: lease.expires_at_unix_ms(),
        not_after_unix_ms: context.not_after_unix_ms,
        store_fence: active_fence.clone(),
    };
    binding.validate()?;
    Ok(Some(binding))
}

fn validate_descriptor_text(
    value: &str,
    maximum: usize,
    field: &'static str,
) -> Result<(), EconomicStateCacheError> {
    if value.is_empty()
        || value.len() > maximum
        || value.trim() != value
        || value.chars().any(char::is_control)
    {
        Err(invariant(format!("{field} is invalid")))
    } else {
        Ok(())
    }
}

fn verify_stage_admission_checkpoint(
    transaction: &Transaction<'_>,
    expected: &EconomicStageAdmissionCheckpoint<'_>,
    active_fence: &StoreMutationFence,
    trusted_now_unix_ms: u64,
) -> Result<(), EconomicStateCacheError> {
    validate_descriptor_text(
        expected.store_id,
        MAX_DESCRIPTOR_KEY_BYTES,
        "admission store id",
    )?;
    validate_digest(expected.digest, "admission commit digest")?;
    let current = crate::admission_operation_store::verify_admission_commit_chain(transaction)
        .map_err(|error| invariant(error.to_string()))?;
    if expected.store_id != active_fence.store_uuid
        || expected.sequence != current.head_sequence
        || expected.digest != current.chain_digest
        || trusted_now_unix_ms < current.trusted_time_high_water_unix_ms
    {
        return Err(EconomicStateCacheError::Conflict);
    }
    Ok(())
}

#[cfg(test)]
#[path = "economic_state_cache_tests.rs"]
#[allow(clippy::expect_used, clippy::unwrap_used)]
mod tests;