chio_store_sqlite/

channel_release_publisher_store.rs

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

use chio_core::canonical::canonical_json_bytes;
use chio_core::economic_continuity::{
    economic_effect_slot_from_head, EconomicAdmissionHandoffStateV1, EconomicContentV1,
    EconomicEffectSlotV1, EconomicEffectStateV1, EconomicEffectTerminalV1, EconomicResourceHeadV1,
    EconomicResourceKeyV1, VerifiedEconomicEffectDispatch,
};
use chio_core::{sha256_hex, StoreMutationFence};
use chio_kernel::admission_operation::{
    AdmissionOperationId, AdmissionOperationKind, AdmissionOperationState, AdmissionOperationV1,
    SideEffectClass,
};
use chio_settle::channel::{
    ChannelEscrowReservationStatusV1, ChannelEscrowReservationViewV1, ChannelLifecycleStatusV1,
    ChannelLifecycleViewV1, VerifiedChannelLifecycleSnapshotV1,
    VerifiedSignedChannelReleaseAuthorizationV1, CHANNEL_RELEASE_BROADCAST_EFFECT_KIND,
    CHANNEL_RELEASE_ROOT_PUBLICATION_EFFECT_KIND,
};
use chio_settle::{PreparedAuthorizedChannelMerkleReleaseV1, SettlementChainConfig};
use rusqlite::{
    params, Connection, ErrorCode, OptionalExtension, Row, Transaction, TransactionBehavior,
};
use serde::{Deserialize, Serialize};

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

mod persistence;
mod qualification;

use persistence::*;
pub(crate) use persistence::{
    initialize_channel_release_publisher_schema, verify_channel_release_publisher_invariants,
};
use qualification::*;

const CHANNEL_RELEASE_PUBLISHER_SCHEMA_KEY: &str = "channel_release_publisher";
pub(crate) const CHANNEL_RELEASE_PUBLISHER_SUPPORTED_SCHEMA_VERSION: i32 = 1;
const CHANNEL_RELEASE_PUBLISHER_SCHEMA_ANCHORS: &[&str] = &[
    "chio_channel_release_publisher_meta",
    "chio_channel_release_publications",
    "channel_lifecycle_records",
];
const CHANNEL_RELEASE_PUBLISHER_SCHEMA: &str = include_str!("channel_release_publisher_store.sql");
const CHANNEL_RELEASE_AUTHORIZATION_DIGEST_DOMAIN: &[u8] =
    b"chio.channel.release-authorization.signed-digest.v1\0";
const MAX_PUBLISHER_ARTIFACT_BYTES: usize = 1024 * 1024;
#[cfg(test)]
const MAX_FAILURE_DETAIL_BYTES: usize = 4 * 1024;
const CHANNEL_ROOT_PUBLICATION_RESULT_SCHEMA: &str = "chio.channel.root-publication-result.v1";

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct ChannelRootPublicationResultV1 {
    schema: String,
    publication_root: String,
    call_digest: String,
    transaction_hash: String,
}

#[derive(Debug, thiserror::Error)]
pub enum ChannelReleasePublisherError {
    #[error("channel release publisher store is unavailable: {0}")]
    Unavailable(String),
    #[error("channel release publisher mutation was fenced")]
    Fenced,
    #[error("channel release publisher record conflicts with retained state")]
    Conflict,
    #[error("channel release publisher invariant failed: {0}")]
    Invalid(String),
    #[error("channel release publisher durable outcome is unknown: {0}")]
    OutcomeUnknown(String),
    #[error("channel release publisher broadcast is disabled: {0}")]
    BroadcastDisabled(String),
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ChannelReleasePublicationStatusV1 {
    DispatchCommitted,
    Submitted,
    Unknown,
    Incident,
}

impl ChannelReleasePublicationStatusV1 {
    fn parse(value: &str) -> Result<Self, ChannelReleasePublisherError> {
        match value {
            "dispatch_committed" => Ok(Self::DispatchCommitted),
            "submitted" => Ok(Self::Submitted),
            "unknown" => Ok(Self::Unknown),
            "incident" => Ok(Self::Incident),
            _ => Err(invalid("retained channel release status is invalid")),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ChannelReleasePublicationRecordV1 {
    channel_id: String,
    publication_binding_digest: String,
    authorization_digest: String,
    prepared_call_digest: String,
    publisher_fence: u64,
    status: ChannelReleasePublicationStatusV1,
    transaction_hash: Option<String>,
    failure_detail: Option<String>,
    record_version: u64,
    created_at_unix_ms: u64,
    updated_at_unix_ms: u64,
}

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

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

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

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

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

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

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

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

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

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

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

#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ChannelReleaseSubmissionOutcomeV1 {
    Submitted(ChannelReleasePublicationRecordV1),
    AlreadyClaimed(ChannelReleasePublicationRecordV1),
}

#[derive(Debug)]
pub struct VerifiedChannelReleasePublicationV1 {
    candidate: ChannelReleasePublisherCandidate,
}

impl VerifiedChannelReleasePublicationV1 {
    pub fn verify(
        authorization: &VerifiedSignedChannelReleaseAuthorizationV1,
        prepared: &PreparedAuthorizedChannelMerkleReleaseV1,
        closing: &VerifiedChannelLifecycleSnapshotV1,
    ) -> Result<Self, ChannelReleasePublisherError> {
        Ok(Self {
            candidate: ChannelReleasePublisherCandidate::from_verified(
                authorization,
                prepared,
                closing,
            )?,
        })
    }
}

#[cfg(test)]
#[derive(Debug, Clone)]
struct ChannelReleaseBroadcastPermitV1 {
    channel_id: String,
    publication_binding_digest: String,
    authorization_digest: String,
    prepared_call_digest: String,
    publisher_fence: u64,
    record_version: u64,
}

#[cfg(test)]
#[derive(Debug, Clone)]
enum ChannelReleasePermitClaimV1 {
    Claimed {
        permit: ChannelReleaseBroadcastPermitV1,
    },
    ExactReplay(ChannelReleasePublicationRecordV1),
}

#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
struct ChannelReleasePublisherCandidate {
    channel_id: String,
    chain_id: String,
    escrow_contract: String,
    escrow_id: String,
    open_digest: String,
    close_digest: String,
    close_body_digest: String,
    effective_close_digest: String,
    final_state_digest: String,
    final_state_sequence: u64,
    source_channel_state_version: u64,
    source_escrow_reservation_version: u64,
    source_lifecycle_fence: u64,
    source_checkpoint_sequence: u64,
    source_checkpoint_digest: String,
    source_channel_head_digest: String,
    source_escrow_head_digest: String,
    closing_checkpoint_sequence: u64,
    closing_checkpoint_digest: String,
    closing_channel_head_digest: String,
    closing_escrow_head_digest: String,
    closing_channel_predecessor_digest: String,
    closing_escrow_predecessor_digest: String,
    closing_lifecycle: ChannelLifecycleViewV1,
    closing_escrow: ChannelEscrowReservationViewV1,
    bound_token_base_units: String,
    release_token_base_units: String,
    refund_token_base_units: String,
    original_operator: String,
    original_operator_key_hash: String,
    beneficiary_address: String,
    asset_binding_digest: String,
    original_dispatch_digest: String,
    close_submission_cutoff_unix_ms: u64,
    escrow_deadline_unix_ms: u64,
    publisher_fence: u64,
    authorized_at_unix_ms: u64,
    frost_slot_id: String,
    frost_authorization_id: String,
    frost_action_digest: String,
    frost_envelope_digest: String,
    frost_scope_id: String,
    frost_resource_id: String,
    frost_resource_version: u64,
    frost_resource_fence: u64,
    frost_roster_digest: String,
    frost_key_epoch: u64,
    frost_issued_at_unix_ms: u64,
    publication_root: String,
    root_operation_id: String,
    root_effect_slot_id: String,
    root_effect_scope_id: String,
    root_effect_kind: String,
    root_idempotency_key: String,
    root_call_digest: String,
    root_resource_head_digest: String,
    release_operation_id: String,
    release_effect_slot_id: String,
    release_effect_scope_id: String,
    release_effect_kind: String,
    release_idempotency_key: String,
    release_call_digest: String,
    release_resource_head_digest: String,
    authorization_digest: String,
    authorization_json: Vec<u8>,
    prepared_call_digest: String,
    prepared_call_json: Vec<u8>,
}

impl ChannelReleasePublisherCandidate {
    fn from_verified(
        authorization: &VerifiedSignedChannelReleaseAuthorizationV1,
        prepared: &PreparedAuthorizedChannelMerkleReleaseV1,
        closing: &VerifiedChannelLifecycleSnapshotV1,
    ) -> Result<Self, ChannelReleasePublisherError> {
        let authority = authorization.authority();
        let binding = authority.binding();
        let body = authorization.body();
        if prepared.authorization() != binding
            || prepared.authorization_digest() != authority.authorization_digest()
            || body.authority() != binding
            || body.publication_root() != prepared.publication_root()
        {
            return Err(invalid(
                "prepared channel release lost its verified authorization",
            ));
        }
        let root = body.root_publication();
        let release_binding = body.release_broadcast();
        let source = authority.close().snapshot();
        let authorization_json = encode(
            authorization.artifact(),
            "signed channel release authorization",
        )?;
        let authorization_digest = authorization_digest(&authorization_json);
        if authorization_digest != authorization.digest() {
            return Err(invalid(
                "channel release authorization digest is not canonical",
            ));
        }
        let prepared_call_json = prepared
            .canonical_call()
            .map_err(|error| invalid(error.to_string()))?;
        let prepared_call_digest = sha256_hex(&prepared_call_json);
        if release_binding.call_digest != prepared_call_digest {
            return Err(invalid(
                "signed channel release authorization changed the prepared release",
            ));
        }
        let frost = binding.frost();
        Ok(Self {
            channel_id: binding.channel_id().to_owned(),
            chain_id: binding.escrow_reference().chain_id.clone(),
            escrow_contract: binding.escrow_reference().escrow_contract.clone(),
            escrow_id: binding.escrow_reference().escrow_id.clone(),
            open_digest: binding.open_digest().to_owned(),
            close_digest: binding.close_digest().to_owned(),
            close_body_digest: binding.close_body_digest().to_owned(),
            effective_close_digest: binding.effective_close_digest().to_owned(),
            final_state_digest: binding.final_state_digest().to_owned(),
            final_state_sequence: binding.final_state_sequence(),
            source_channel_state_version: binding.channel_state_version(),
            source_escrow_reservation_version: binding.escrow_reservation_version(),
            source_lifecycle_fence: binding.lifecycle_fence(),
            source_checkpoint_sequence: source.checkpoint_sequence(),
            source_checkpoint_digest: source.checkpoint_digest().to_owned(),
            source_channel_head_digest: source.channel_head_digest().to_owned(),
            source_escrow_head_digest: source.escrow_head_digest().to_owned(),
            closing_checkpoint_sequence: closing.checkpoint_sequence(),
            closing_checkpoint_digest: closing.checkpoint_digest().to_owned(),
            closing_channel_head_digest: closing.channel_head_digest().to_owned(),
            closing_escrow_head_digest: closing.escrow_head_digest().to_owned(),
            closing_channel_predecessor_digest: closing
                .channel_predecessor_digest()
                .ok_or_else(|| invalid("closing channel head has no predecessor"))?
                .to_owned(),
            closing_escrow_predecessor_digest: closing
                .escrow_predecessor_digest()
                .ok_or_else(|| invalid("closing escrow head has no predecessor"))?
                .to_owned(),
            closing_lifecycle: closing.lifecycle().clone(),
            closing_escrow: closing.escrow().clone(),
            bound_token_base_units: binding.bound_token_base_units().to_owned(),
            release_token_base_units: binding.expected_release_token_base_units().to_owned(),
            refund_token_base_units: binding.expected_refund_token_base_units().to_owned(),
            original_operator: binding.original_operator().to_owned(),
            original_operator_key_hash: binding.original_operator_key_hash().to_owned(),
            beneficiary_address: binding.payee_beneficiary_address().to_owned(),
            asset_binding_digest: binding.asset_binding_digest().to_owned(),
            original_dispatch_digest: binding.original_web3_dispatch_digest().to_owned(),
            close_submission_cutoff_unix_ms: binding.close_submission_cutoff_unix_ms(),
            escrow_deadline_unix_ms: binding.escrow_deadline_unix_ms(),
            publisher_fence: binding.publisher_fence(),
            authorized_at_unix_ms: binding.authorized_at_unix_ms(),
            frost_slot_id: frost.authorization_slot_id.clone(),
            frost_authorization_id: frost.authorization_id.clone(),
            frost_action_digest: frost.action_digest.clone(),
            frost_envelope_digest: frost.signed_envelope_digest.clone(),
            frost_scope_id: binding.frost_scope_id().to_owned(),
            frost_resource_id: binding.frost_resource_id().to_owned(),
            frost_resource_version: binding.frost_resource_version(),
            frost_resource_fence: binding.frost_resource_fence(),
            frost_roster_digest: binding.frost_roster_digest().to_owned(),
            frost_key_epoch: binding.frost_key_epoch(),
            frost_issued_at_unix_ms: binding.frost_issued_at_unix_ms(),
            publication_root: body.publication_root().to_owned(),
            root_operation_id: root.operation_id.clone(),
            root_effect_slot_id: root.effect_slot_id.clone(),
            root_effect_scope_id: root.scope_id.clone(),
            root_effect_kind: root.effect_kind.clone(),
            root_idempotency_key: root.idempotency_key.clone(),
            root_call_digest: root.call_digest.clone(),
            root_resource_head_digest: root.resource_head_digest.clone(),
            release_operation_id: release_binding.operation_id.clone(),
            release_effect_slot_id: release_binding.effect_slot_id.clone(),
            release_effect_scope_id: release_binding.scope_id.clone(),
            release_effect_kind: release_binding.effect_kind.clone(),
            release_idempotency_key: release_binding.idempotency_key.clone(),
            release_call_digest: release_binding.call_digest.clone(),
            release_resource_head_digest: release_binding.resource_head_digest.clone(),
            authorization_digest,
            authorization_json,
            prepared_call_digest,
            prepared_call_json,
        })
    }

    fn binding_digest(&self) -> Result<String, ChannelReleasePublisherError> {
        Ok(sha256_hex(&encode(self, "channel release publication")?))
    }

    fn verify_dispatch(
        &self,
        dispatch: &VerifiedEconomicEffectDispatch,
    ) -> Result<(), ChannelReleasePublisherError> {
        self.verify_dispatch_slot(dispatch.slot())
    }

    fn verify_dispatch_slot(
        &self,
        slot: &EconomicEffectSlotV1,
    ) -> Result<(), ChannelReleasePublisherError> {
        let frost = slot.frost.as_ref();
        if slot.slot_id != self.release_effect_slot_id
            || slot.operation_id != self.release_operation_id
            || slot.resource_key.resource_family != "channel_escrow_reservation"
            || slot.resource_key.scope_id != self.release_effect_scope_id
            || slot.resource_key.scope_id != self.frost_scope_id
            || slot.resource_key.resource_id != self.channel_id
            || slot.effect_kind != self.release_effect_kind
            || slot.effect_kind != CHANNEL_RELEASE_BROADCAST_EFFECT_KIND
            || slot.idempotency_key != self.release_idempotency_key
            || slot.parameters_digest != self.release_call_digest
            || slot.parameters_digest != self.prepared_call_digest
            || slot.resource_head_digest != self.release_resource_head_digest
            || slot.action_digest != self.frost_action_digest
            || slot.admission_handoff.state != EconomicAdmissionHandoffStateV1::MutationSubmitted
            || slot.state != EconomicEffectStateV1::DispatchCommitted
            || frost.is_none_or(|binding| {
                binding.authorization_slot_id != self.frost_slot_id
                    || binding.authorization_id != self.frost_authorization_id
                    || binding.action_digest != self.frost_action_digest
                    || binding.signed_envelope_digest != self.frost_envelope_digest
            })
        {
            return Err(ChannelReleasePublisherError::Fenced);
        }
        Ok(())
    }
}

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

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

    #[must_use]
    pub fn mutation_fence(&self) -> StoreMutationFence {
        self.serving_owner.fence.clone()
    }

    pub fn verify_invariants(&self) -> Result<(), ChannelReleasePublisherError> {
        let connection = self.connection()?;
        verify_channel_release_publisher_invariants(&connection).map_err(owner_error)
    }

    pub fn load(
        &self,
        channel_id: &str,
    ) -> Result<Option<ChannelReleasePublicationRecordV1>, ChannelReleasePublisherError> {
        let connection = self.connection()?;
        self.serving_owner
            .verify_authority_anchor(&connection)
            .map_err(owner_error)?;
        load_record(&connection, channel_id)
    }

    pub async fn submit_authorized_channel_release(
        &self,
        config: &SettlementChainConfig,
        publication: &VerifiedChannelReleasePublicationV1,
        dispatch: VerifiedEconomicEffectDispatch,
        fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<ChannelReleaseSubmissionOutcomeV1, ChannelReleasePublisherError> {
        let candidate = &publication.candidate;
        if let Some(record) = self.replay_candidate(candidate, fence, trusted_now_unix_ms)? {
            return Ok(ChannelReleaseSubmissionOutcomeV1::AlreadyClaimed(record));
        }
        candidate.verify_dispatch(&dispatch)?;
        if config.chain_id != candidate.chain_id
            || config.escrow_contract != candidate.escrow_contract
            || config.operator_address != candidate.original_operator
        {
            return Err(invalid(
                "settlement configuration does not match channel release authority",
            ));
        }
        config
            .validate()
            .map_err(|error| invalid(error.to_string()))?;
        self.qualify_new_dispatch(candidate, &dispatch, fence, trusted_now_unix_ms)?;
        Err(ChannelReleasePublisherError::BroadcastDisabled(
            "deterministic raw transaction signing, nonce leasing, recovery, and live FROST/operator key resolution are unavailable".to_owned(),
        ))
    }

    pub fn replay_authorized_channel_release(
        &self,
        publication: &VerifiedChannelReleasePublicationV1,
        fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<Option<ChannelReleaseSubmissionOutcomeV1>, ChannelReleasePublisherError> {
        let candidate = &publication.candidate;
        self.replay_candidate(candidate, fence, trusted_now_unix_ms)
            .map(|record| record.map(ChannelReleaseSubmissionOutcomeV1::AlreadyClaimed))
    }

    fn replay_candidate(
        &self,
        candidate: &ChannelReleasePublisherCandidate,
        fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<Option<ChannelReleasePublicationRecordV1>, ChannelReleasePublisherError> {
        self.replay_candidate_inner(candidate, fence, trusted_now_unix_ms, true)
    }

    #[cfg(test)]
    fn replay_candidate_unqualified_for_test(
        &self,
        candidate: &ChannelReleasePublisherCandidate,
        fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<Option<ChannelReleasePublicationRecordV1>, ChannelReleasePublisherError> {
        self.replay_candidate_inner(candidate, fence, trusted_now_unix_ms, false)
    }

    fn replay_candidate_inner(
        &self,
        candidate: &ChannelReleasePublisherCandidate,
        fence: &StoreMutationFence,
        _trusted_now_unix_ms: u64,
        _require_durable_execution: bool,
    ) -> Result<Option<ChannelReleasePublicationRecordV1>, ChannelReleasePublisherError> {
        if fence != &self.serving_owner.fence {
            return Err(ChannelReleasePublisherError::Fenced);
        }
        let mut connection = self.connection()?;
        self.serving_owner
            .verify_authority_anchor(&connection)
            .map_err(owner_error)?;
        let transaction = connection.transaction().map_err(sqlite_error)?;
        validate_candidate(candidate)?;
        let binding_digest = candidate.binding_digest()?;
        match load_record(&transaction, &candidate.channel_id)? {
            Some(record) if record_matches_candidate(&record, candidate, &binding_digest) => {
                Ok(Some(record))
            }
            Some(_) => Err(ChannelReleasePublisherError::Conflict),
            None => Ok(None),
        }
    }

    #[cfg(test)]
    fn claim_candidate_unqualified_for_test(
        &self,
        candidate: &ChannelReleasePublisherCandidate,
        fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<ChannelReleasePermitClaimV1, ChannelReleasePublisherError> {
        self.claim_candidate_inner(candidate, fence, trusted_now_unix_ms, false)
    }

    #[cfg(test)]
    fn claim_candidate_inner(
        &self,
        candidate: &ChannelReleasePublisherCandidate,
        fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
        require_durable_execution: bool,
    ) -> Result<ChannelReleasePermitClaimV1, ChannelReleasePublisherError> {
        if fence != &self.serving_owner.fence {
            return Err(ChannelReleasePublisherError::Fenced);
        }
        let mut connection = self.connection()?;
        self.serving_owner
            .verify_authority_anchor(&connection)
            .map_err(owner_error)?;
        let transaction = connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(sqlite_error)?;
        validate_candidate(candidate)?;
        let binding_digest = candidate.binding_digest()?;
        if let Some(existing) = load_record(&transaction, &candidate.channel_id)? {
            if record_matches_candidate(&existing, candidate, &binding_digest) {
                return Ok(ChannelReleasePermitClaimV1::ExactReplay(existing));
            }
            return Err(ChannelReleasePublisherError::Conflict);
        }
        verify_trusted_time(&transaction, trusted_now_unix_ms)?;
        validate_candidate_freshness(candidate, trusted_now_unix_ms)?;
        let lifecycle_json = encode(&candidate.closing_lifecycle, "closing channel lifecycle")?;
        let escrow_json = encode(&candidate.closing_escrow, "closing escrow reservation")?;
        qualify_durable_closing(&transaction, candidate, &lifecycle_json, &escrow_json)?;
        if require_durable_execution {
            let _ = qualify_durable_release_execution(&transaction, candidate, fence)?;
        }
        advance_trusted_time(&transaction, trusted_now_unix_ms)?;
        insert_publication(
            &transaction,
            candidate,
            &binding_digest,
            &lifecycle_json,
            &escrow_json,
            fence,
            trusted_now_unix_ms,
        )?;
        self.serving_owner
            .append_global_commit(
                &transaction,
                "channel_release_dispatch_committed",
                "channel_release_publication",
                &candidate.channel_id,
                1,
            )
            .map_err(owner_error)?;
        transaction.commit().map_err(|error| {
            owner_error(self.serving_owner.outcome_unknown(format!(
                "sqlite channel release claim outcome is unknown: {error}"
            )))
        })?;
        self.serving_owner
            .sync_authority_anchor(&connection)
            .map_err(owner_error)?;
        Ok(ChannelReleasePermitClaimV1::Claimed {
            permit: ChannelReleaseBroadcastPermitV1 {
                channel_id: candidate.channel_id.clone(),
                publication_binding_digest: binding_digest,
                authorization_digest: candidate.authorization_digest.clone(),
                prepared_call_digest: candidate.prepared_call_digest.clone(),
                publisher_fence: candidate.publisher_fence,
                record_version: 1,
            },
        })
    }

    fn qualify_new_dispatch(
        &self,
        candidate: &ChannelReleasePublisherCandidate,
        dispatch: &VerifiedEconomicEffectDispatch,
        fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<(), ChannelReleasePublisherError> {
        if fence != &self.serving_owner.fence {
            return Err(ChannelReleasePublisherError::Fenced);
        }
        let mut connection = self.connection()?;
        self.serving_owner
            .verify_authority_anchor(&connection)
            .map_err(owner_error)?;
        let transaction = connection.transaction().map_err(sqlite_error)?;
        verify_trusted_time(&transaction, trusted_now_unix_ms)?;
        validate_candidate(candidate)?;
        validate_candidate_freshness(candidate, trusted_now_unix_ms)?;
        let lifecycle_json = encode(&candidate.closing_lifecycle, "closing channel lifecycle")?;
        let escrow_json = encode(&candidate.closing_escrow, "closing escrow reservation")?;
        qualify_durable_closing(&transaction, candidate, &lifecycle_json, &escrow_json)?;
        let durable_slot = qualify_durable_release_execution(&transaction, candidate, fence)?;
        qualify_exact_dispatch_slot(
            candidate,
            &durable_slot,
            dispatch.slot(),
            dispatch.commit_id(),
        )
    }

    #[cfg(test)]
    fn record_submission_unknown(
        &self,
        permit: &ChannelReleaseBroadcastPermitV1,
        detail: &str,
        fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<ChannelReleasePublicationRecordV1, ChannelReleasePublisherError> {
        self.record_submission(
            permit,
            SubmissionRecord::Unknown(detail),
            fence,
            trusted_now_unix_ms,
        )
    }

    #[cfg(test)]
    fn record_submission(
        &self,
        permit: &ChannelReleaseBroadcastPermitV1,
        submission: SubmissionRecord<'_>,
        fence: &StoreMutationFence,
        trusted_now_unix_ms: u64,
    ) -> Result<ChannelReleasePublicationRecordV1, ChannelReleasePublisherError> {
        if fence != &self.serving_owner.fence || trusted_now_unix_ms == 0 {
            return Err(ChannelReleasePublisherError::Fenced);
        }
        let mut connection = self.connection()?;
        self.serving_owner
            .verify_authority_anchor(&connection)
            .map_err(owner_error)?;
        let transaction = connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(sqlite_error)?;
        verify_trusted_time(&transaction, trusted_now_unix_ms)?;
        let (status, transaction_hash, failure_detail, mutation_kind) = match submission {
            SubmissionRecord::Submitted(transaction_hash)
                if is_evm_transaction_hash(transaction_hash) =>
            {
                (
                    "submitted",
                    Some(transaction_hash),
                    None,
                    "channel_release_submitted",
                )
            }
            SubmissionRecord::Unknown(detail)
                if !detail.is_empty() && detail.len() <= MAX_FAILURE_DETAIL_BYTES =>
            {
                ("unknown", None, Some(detail), "channel_release_unknown")
            }
            _ => return Err(invalid("channel release submission result is empty")),
        };
        let changed = transaction
            .execute(
                r#"
                UPDATE chio_channel_release_publications
                SET status = ?1, transaction_hash = ?2, failure_detail = ?3,
                    record_version = record_version + 1,
                    store_uuid = ?4, store_lease_id = ?5, store_owner_epoch = ?6,
                    updated_at_unix_ms = ?7
                WHERE channel_id = ?8 AND status = 'dispatch_committed'
                  AND publication_binding_digest = ?9
                  AND authorization_digest = ?10 AND prepared_call_digest = ?11
                  AND publisher_fence = ?12 AND record_version = ?13
                  AND updated_at_unix_ms <= ?7
                "#,
                params![
                    status,
                    transaction_hash,
                    failure_detail,
                    &fence.store_uuid,
                    &fence.lease_id,
                    sqlite_i64(fence.owner_epoch, "store_owner_epoch")?,
                    sqlite_i64(trusted_now_unix_ms, "updated_at_unix_ms")?,
                    &permit.channel_id,
                    &permit.publication_binding_digest,
                    &permit.authorization_digest,
                    &permit.prepared_call_digest,
                    sqlite_i64(permit.publisher_fence, "publisher_fence")?,
                    sqlite_i64(permit.record_version, "record_version")?,
                ],
            )
            .map_err(sqlite_error)?;
        if changed != 1 {
            let existing = load_record(&transaction, &permit.channel_id)?
                .ok_or(ChannelReleasePublisherError::Fenced)?;
            let exact = existing.publication_binding_digest == permit.publication_binding_digest
                && existing.authorization_digest == permit.authorization_digest
                && existing.prepared_call_digest == permit.prepared_call_digest
                && existing.publisher_fence == permit.publisher_fence
                && existing.status == submission.status()
                && existing.transaction_hash.as_deref() == submission.transaction_hash()
                && existing.failure_detail.as_deref() == submission.failure_detail();
            if exact {
                return Ok(existing);
            }
            return Err(ChannelReleasePublisherError::Fenced);
        }
        advance_trusted_time(&transaction, trusted_now_unix_ms)?;
        self.serving_owner
            .append_global_commit(
                &transaction,
                mutation_kind,
                "channel_release_publication",
                &permit.channel_id,
                permit
                    .record_version
                    .checked_add(1)
                    .ok_or_else(|| invalid("channel release record version overflowed"))?,
            )
            .map_err(owner_error)?;
        transaction.commit().map_err(|error| {
            owner_error(self.serving_owner.outcome_unknown(format!(
                "sqlite channel release submission record outcome is unknown: {error}"
            )))
        })?;
        self.serving_owner
            .sync_authority_anchor(&connection)
            .map_err(owner_error)?;
        load_record(&connection, &permit.channel_id)?
            .ok_or_else(|| invalid("committed channel release record disappeared"))
    }

    fn connection(&self) -> Result<MutexGuard<'_, Connection>, ChannelReleasePublisherError> {
        self.connection.lock().map_err(|_| {
            ChannelReleasePublisherError::Unavailable(
                "sqlite channel release publisher lock poisoned".to_owned(),
            )
        })
    }
}

pub(crate) fn quarantine_incomplete_dispatches_at_startup(
    connection: &mut Connection,
    serving_owner: &SqliteServingOwner,
) -> Result<(), SqliteServingOwnerError> {
    serving_owner.verify_authority_anchor(connection)?;
    let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
    let retained = {
        let mut statement = transaction.prepare(
            r#"
            SELECT channel_id, record_version, updated_at_unix_ms
            FROM chio_channel_release_publications
            WHERE status = 'dispatch_committed'
            ORDER BY channel_id
            "#,
        )?;
        let rows = statement
            .query_map([], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, i64>(1)?,
                    row.get::<_, i64>(2)?,
                ))
            })?
            .collect::<Result<Vec<_>, _>>()?;
        rows
    };
    if retained.is_empty() {
        return Ok(());
    }
    let trusted_time_high_water = transaction.query_row(
        r#"
        SELECT trusted_time_high_water_unix_ms
        FROM chio_channel_release_publisher_meta
        WHERE singleton = 1
        "#,
        [],
        |row| row.get::<_, i64>(0),
    )?;
    let detail = "deterministic transaction recovery is unavailable for a previously committed channel release dispatch";
    for (channel_id, record_version, updated_at_unix_ms) in retained {
        let next_version = record_version.checked_add(1).ok_or_else(|| {
            SqliteServingOwnerError::Invalid(
                "channel release recovery record version overflowed".to_owned(),
            )
        })?;
        let changed = transaction.execute(
            r#"
            UPDATE chio_channel_release_publications
            SET status = 'unknown', transaction_hash = NULL, failure_detail = ?1,
                record_version = ?2, store_uuid = ?3, store_lease_id = ?4,
                store_owner_epoch = ?5, updated_at_unix_ms = ?6
            WHERE channel_id = ?7 AND status = 'dispatch_committed'
              AND record_version = ?8
            "#,
            params![
                detail,
                next_version,
                &serving_owner.fence.store_uuid,
                &serving_owner.fence.lease_id,
                i64::try_from(serving_owner.fence.owner_epoch).map_err(|_| {
                    SqliteServingOwnerError::Invalid(
                        "channel release recovery owner epoch exceeds SQLite range".to_owned(),
                    )
                })?,
                updated_at_unix_ms.max(trusted_time_high_water),
                &channel_id,
                record_version,
            ],
        )?;
        if changed != 1 {
            return Err(SqliteServingOwnerError::Invalid(
                "channel release startup quarantine lost its retained row".to_owned(),
            ));
        }
        serving_owner.append_global_commit(
            &transaction,
            "channel_release_startup_outcome_unknown",
            "channel_release_publication",
            &channel_id,
            u64::try_from(next_version).map_err(|_| {
                SqliteServingOwnerError::Invalid(
                    "channel release recovery record version is invalid".to_owned(),
                )
            })?,
        )?;
    }
    transaction.commit().map_err(|error| {
        serving_owner.outcome_unknown(format!(
            "sqlite channel release startup quarantine outcome is unknown: {error}"
        ))
    })?;
    serving_owner.sync_authority_anchor(connection)
}

#[cfg(test)]
#[derive(Clone, Copy)]
enum SubmissionRecord<'a> {
    Submitted(&'a str),
    Unknown(&'a str),
}

#[cfg(test)]
impl<'a> SubmissionRecord<'a> {
    const fn status(self) -> ChannelReleasePublicationStatusV1 {
        match self {
            Self::Submitted(_) => ChannelReleasePublicationStatusV1::Submitted,
            Self::Unknown(_) => ChannelReleasePublicationStatusV1::Unknown,
        }
    }

    const fn transaction_hash(self) -> Option<&'a str> {
        match self {
            Self::Submitted(transaction_hash) => Some(transaction_hash),
            Self::Unknown(_) => None,
        }
    }

    fn failure_detail(&self) -> Option<&'a str> {
        match *self {
            Self::Submitted(_) => None,
            Self::Unknown(detail) => Some(detail),
        }
    }
}

#[cfg(test)]
#[path = "channel_release_publisher_store_tests.rs"]
mod tests;