chio_store_sqlite/budget_store/

snapshot.rs

use super::*;
use chio_core::canonical::canonical_json_bytes;
use chio_core::crypto::{PublicKey, Signature};
use chio_core::sha256_hex;
use hmac::{Hmac, Mac};
use serde::{Deserialize, Serialize};
use sha2::Sha256;
use subtle::ConstantTimeEq;

const BUDGET_ANCHOR_GENESIS_DIGEST: &str =
    "0000000000000000000000000000000000000000000000000000000000000000";

#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct BudgetSnapshotAnchorCommitment {
    pub schema: String,
    pub commit_sequence: u64,
    pub previous_chain_digest: String,
    pub chain_digest: String,
    pub anchor_set_digest: String,
    pub leader_url: String,
    pub election_term: u64,
    pub committed_at: u64,
    pub signer_public_key: String,
}

#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct SignedBudgetSnapshotAnchorCommitment {
    pub body: BudgetSnapshotAnchorCommitment,
    pub signature: Signature,
}

#[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub struct BudgetSnapshotAnchorProvenance {
    pub schema: String,
    pub chain: Vec<SignedBudgetSnapshotAnchorCommitment>,
    pub cluster_authenticator: String,
}

#[derive(Serialize)]
struct AnchorSetDigestBody<'a> {
    schema: &'static str,
    anchors: Vec<AnchorDigestRecord<'a>>,
}

#[derive(Serialize)]
struct AnchorDigestRecord<'a> {
    capability_id: &'a str,
    grant_index: u32,
    invocation_count: u32,
    updated_at: i64,
    seq: u64,
    total_cost_exposed: u64,
    total_cost_realized_spend: u64,
}

#[derive(Serialize)]
struct AnchorChainDigestBody<'a> {
    schema: &'static str,
    commit_sequence: u64,
    previous_chain_digest: &'a str,
    anchor_set_digest: &'a str,
    leader_url: &'a str,
    election_term: u64,
    committed_at: u64,
    signer_public_key: &'a str,
}

#[derive(Serialize)]
struct AnchorAuthenticatorBody<'a> {
    scheme: &'static str,
    chain: &'a [SignedBudgetSnapshotAnchorCommitment],
}

pub fn budget_snapshot_anchor_set_digest(
    anchors: &[BudgetUsageRecord],
) -> Result<String, BudgetStoreError> {
    let mut canonical_anchors = anchors.to_vec();
    canonical_anchors.sort_by(|left, right| {
        (&left.capability_id, left.grant_index).cmp(&(&right.capability_id, right.grant_index))
    });
    let anchors = canonical_anchors
        .iter()
        .map(|anchor| AnchorDigestRecord {
            capability_id: &anchor.capability_id,
            grant_index: anchor.grant_index,
            invocation_count: anchor.invocation_count,
            updated_at: anchor.updated_at,
            seq: anchor.seq,
            total_cost_exposed: anchor.total_cost_exposed,
            total_cost_realized_spend: anchor.total_cost_realized_spend,
        })
        .collect();
    let bytes = canonical_json_bytes(&AnchorSetDigestBody {
        schema: "chio.budget-snapshot-anchor-set.v1",
        anchors,
    })
    .map_err(|error| BudgetStoreError::Invariant(error.to_string()))?;
    Ok(sha256_hex(&bytes))
}

pub fn budget_snapshot_anchor_chain_digest(
    body: &BudgetSnapshotAnchorCommitment,
) -> Result<String, BudgetStoreError> {
    let bytes = canonical_json_bytes(&AnchorChainDigestBody {
        schema: "chio.budget-snapshot-anchor-chain.v1",
        commit_sequence: body.commit_sequence,
        previous_chain_digest: &body.previous_chain_digest,
        anchor_set_digest: &body.anchor_set_digest,
        leader_url: &body.leader_url,
        election_term: body.election_term,
        committed_at: body.committed_at,
        signer_public_key: &body.signer_public_key,
    })
    .map_err(|error| BudgetStoreError::Invariant(error.to_string()))?;
    Ok(sha256_hex(&bytes))
}

pub fn budget_snapshot_anchor_authenticator(
    service_token: &str,
    chain: &[SignedBudgetSnapshotAnchorCommitment],
) -> Result<String, BudgetStoreError> {
    let bytes = canonical_json_bytes(&AnchorAuthenticatorBody {
        scheme: "chio.cluster-budget-anchor-auth.v1",
        chain,
    })
    .map_err(|error| BudgetStoreError::Invariant(error.to_string()))?;
    let mut authenticator = Hmac::<Sha256>::new_from_slice(service_token.as_bytes())
        .map_err(|error| BudgetStoreError::Invariant(error.to_string()))?;
    authenticator.update(&bytes);
    Ok(hex::encode(authenticator.finalize().into_bytes()))
}

#[derive(Clone, Debug, PartialEq, Eq)]
pub struct BudgetStoreSnapshot {
    pub usages: Vec<BudgetUsageRecord>,
    pub usage_history_anchors: Vec<BudgetUsageRecord>,
    pub mutation_events: Vec<BudgetMutationRecord>,
    pub abandoned_seq_ranges: Vec<(u64, u64)>,
    pub covered_head: u64,
    pub origin_ack_heads: Vec<(String, u64)>,
}

struct BudgetImportBatch<'a> {
    usages: &'a [BudgetUsageRecord],
    events: &'a [BudgetMutationRecord],
    anchors: &'a [BudgetUsageRecord],
    abandoned_seq_ranges: &'a [(u64, u64)],
    covered_head: Option<u64>,
    origin_ack_heads: Option<&'a [(String, u64)]>,
    allow_verified_anchor_install: bool,
}

impl SqliteBudgetStore {
    pub fn budget_import_floor(&self, authority_id: &str) -> Result<u64, BudgetStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let floor: i64 = transaction
            .query_row(
                "SELECT floor_seq FROM budget_import_floors WHERE authority_id = ?1",
                rusqlite::params![authority_id],
                |row| row.get(0),
            )
            .optional()?
            .unwrap_or(0);
        transaction.rollback()?;
        Ok(floor.max(0) as u64)
    }

    pub fn record_budget_import_floors(
        &self,
        events: &[BudgetMutationRecord],
    ) -> Result<(), BudgetStoreError> {
        self.require_standalone_mutation("budget import floor")?;
        use std::collections::BTreeMap;
        let mut min_by_origin: BTreeMap<&str, u64> = BTreeMap::new();
        for event in events {
            let Some(authority) = event.authority.as_ref() else {
                continue;
            };
            let entry = min_by_origin
                .entry(authority.authority_id.as_str())
                .or_insert(event.event_seq);
            *entry = (*entry).min(event.event_seq);
        }
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection)?;
        for (origin, min_seq) in min_by_origin {
            let floor = min_seq.saturating_sub(1);
            transaction.execute(
                "INSERT INTO budget_import_floors (authority_id, floor_seq) VALUES (?1, ?2) \
                 ON CONFLICT(authority_id) DO UPDATE SET floor_seq = MAX(floor_seq, excluded.floor_seq)",
                rusqlite::params![origin, budget_u64_to_sqlite(floor, "floor_seq")?],
            )?;
        }
        transaction.commit()?;
        Ok(())
    }

    pub fn upsert_usage(&self, record: &BudgetUsageRecord) -> Result<(), BudgetStoreError> {
        self.require_standalone_mutation("budget usage upsert")?;
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection)?;
        Self::reconcile_imported_usages(&transaction, std::slice::from_ref(record), &[])?;
        transaction.commit()?;
        Ok(())
    }

    pub fn import_snapshot_records(
        &self,
        usages: &[BudgetUsageRecord],
        events: &[BudgetMutationRecord],
    ) -> Result<(), BudgetStoreError> {
        self.import_records(
            &BudgetImportBatch {
                usages,
                events,
                anchors: &[],
                abandoned_seq_ranges: &[],
                covered_head: None,
                origin_ack_heads: None,
                allow_verified_anchor_install: false,
            },
            "budget snapshot import",
        )
    }

    pub fn import_snapshot_records_with_anchors(
        &self,
        usages: &[BudgetUsageRecord],
        events: &[BudgetMutationRecord],
        anchors: &[BudgetUsageRecord],
        abandoned_seq_ranges: &[(u64, u64)],
        covered_head: u64,
    ) -> Result<(), BudgetStoreError> {
        self.import_records(
            &BudgetImportBatch {
                usages,
                events,
                anchors,
                abandoned_seq_ranges,
                covered_head: Some(covered_head),
                origin_ack_heads: None,
                allow_verified_anchor_install: false,
            },
            "budget snapshot import",
        )
    }

    pub fn import_budget_snapshot(
        &self,
        snapshot: &BudgetStoreSnapshot,
    ) -> Result<(), BudgetStoreError> {
        const OPERATION: &str = "budget snapshot import";
        self.require_standalone_mutation(OPERATION)?;

        let validated = Self::validate_budget_snapshot_in_isolation(snapshot)?;
        let batch = BudgetImportBatch {
            usages: &validated.usages,
            events: &validated.mutation_events,
            anchors: &validated.usage_history_anchors,
            abandoned_seq_ranges: &validated.abandoned_seq_ranges,
            covered_head: Some(validated.covered_head),
            origin_ack_heads: Some(&validated.origin_ack_heads),
            allow_verified_anchor_install: false,
        };
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection)?;
        Self::validate_exact_snapshot_usage_anchors(&transaction, batch.anchors)?;
        Self::validate_local_snapshot_subset(&transaction, &validated)?;
        Self::apply_import_batch(&transaction, &batch)?;
        transaction.commit()?;
        Ok(())
    }

    pub fn import_budget_snapshot_with_anchor_provenance(
        &self,
        snapshot: &BudgetStoreSnapshot,
        provenance: &BudgetSnapshotAnchorProvenance,
        expected_leader_url: &str,
        expected_election_term: u64,
        cluster_service_token: &str,
    ) -> Result<(), BudgetStoreError> {
        const OPERATION: &str = "verified budget snapshot import";
        self.require_standalone_mutation(OPERATION)?;
        let head = verify_budget_snapshot_anchor_provenance(
            snapshot,
            provenance,
            expected_leader_url,
            expected_election_term,
            cluster_service_token,
        )?;
        let validated = Self::validate_budget_snapshot_in_isolation(snapshot)?;
        let batch = BudgetImportBatch {
            usages: &validated.usages,
            events: &validated.mutation_events,
            anchors: &validated.usage_history_anchors,
            abandoned_seq_ranges: &validated.abandoned_seq_ranges,
            covered_head: Some(validated.covered_head),
            origin_ack_heads: Some(&validated.origin_ack_heads),
            allow_verified_anchor_install: true,
        };
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection)?;
        verify_local_anchor_provenance_continuity(&transaction, provenance)?;
        Self::validate_local_snapshot_subset(&transaction, &validated)?;
        Self::apply_import_batch(&transaction, &batch)?;
        let local = Self::snapshot_usage_history_anchors(&transaction)?;
        if local != validated.usage_history_anchors {
            return Err(BudgetStoreError::Invariant(
                "verified budget snapshot history anchor set is not exact".to_string(),
            ));
        }
        transaction.execute(
            r#"
            INSERT INTO budget_snapshot_anchor_provenance (
                leader_url, commit_sequence, chain_digest, anchor_set_digest,
                election_term, signer_public_key, committed_at
            ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
            ON CONFLICT(leader_url) DO UPDATE SET
                commit_sequence = excluded.commit_sequence,
                chain_digest = excluded.chain_digest,
                anchor_set_digest = excluded.anchor_set_digest,
                election_term = excluded.election_term,
                signer_public_key = excluded.signer_public_key,
                committed_at = excluded.committed_at
            "#,
            params![
                &head.leader_url,
                budget_u64_to_sqlite(head.commit_sequence, "anchor_commit_sequence")?,
                &head.chain_digest,
                &head.anchor_set_digest,
                budget_u64_to_sqlite(head.election_term, "anchor_election_term")?,
                &head.signer_public_key,
                budget_u64_to_sqlite(head.committed_at, "anchor_committed_at")?,
            ],
        )?;
        transaction.commit()?;
        Ok(())
    }

    pub fn import_delta_records(
        &self,
        usages: &[BudgetUsageRecord],
        events: &[BudgetMutationRecord],
    ) -> Result<(), BudgetStoreError> {
        self.import_records(
            &BudgetImportBatch {
                usages,
                events,
                anchors: &[],
                abandoned_seq_ranges: &[],
                covered_head: None,
                origin_ack_heads: None,
                allow_verified_anchor_install: false,
            },
            "budget delta import",
        )
    }

    fn import_records(
        &self,
        batch: &BudgetImportBatch<'_>,
        operation: &str,
    ) -> Result<(), BudgetStoreError> {
        self.require_standalone_mutation(operation)?;
        Self::validate_import_batch_order(batch.events)?;
        let mut connection = self.connection()?;
        let transaction = self.begin_write(&mut connection)?;
        Self::apply_import_batch(&transaction, batch)?;
        transaction.commit()?;
        Ok(())
    }

    fn apply_import_batch(
        transaction: &rusqlite::Transaction<'_>,
        batch: &BudgetImportBatch<'_>,
    ) -> Result<(), BudgetStoreError> {
        Self::install_snapshot_usage_anchors(
            transaction,
            batch.anchors,
            batch.allow_verified_anchor_install,
        )?;
        for event in batch.events {
            Self::import_mutation_record_in_transaction(transaction, event)?;
        }
        Self::reconcile_imported_usages(transaction, batch.usages, batch.events)?;
        Self::insert_abandoned_event_seq_ranges(transaction, batch.abandoned_seq_ranges)?;
        if let Some(claimed_head) = batch.covered_head {
            let exact_head = Self::contiguous_snapshot_head_from(transaction, 0)?;
            if claimed_head != exact_head {
                return Err(BudgetStoreError::Invariant(format!(
                    "budget snapshot claimed covered head {claimed_head}, but imported records prove {exact_head}"
                )));
            }
            let exact_origin_heads = Self::origin_ack_heads_at(transaction, exact_head)?;
            if let Some(claimed_origin_heads) = batch.origin_ack_heads {
                if claimed_origin_heads != exact_origin_heads {
                    return Err(BudgetStoreError::Invariant(
                        "budget snapshot origin acknowledgement heads are not proved by the retained mutation prefix"
                            .to_string(),
                    ));
                }
            }
            Self::install_snapshot_coverage(transaction, exact_head, &exact_origin_heads)?;
        }
        Ok(())
    }

    fn validate_budget_snapshot_in_isolation(
        snapshot: &BudgetStoreSnapshot,
    ) -> Result<BudgetStoreSnapshot, BudgetStoreError> {
        let validator = Self::open(":memory:")?;
        Self::seed_snapshot_validation_anchors(&validator, &snapshot.usage_history_anchors)?;
        validator.import_records(
            &BudgetImportBatch {
                usages: &snapshot.usages,
                events: &snapshot.mutation_events,
                anchors: &snapshot.usage_history_anchors,
                abandoned_seq_ranges: &snapshot.abandoned_seq_ranges,
                covered_head: Some(snapshot.covered_head),
                origin_ack_heads: Some(&snapshot.origin_ack_heads),
                allow_verified_anchor_install: false,
            },
            "budget snapshot validation",
        )?;
        let validated = validator.export_budget_snapshot()?;
        Self::validate_staged_snapshot_matches_payload(snapshot, &validated)?;
        Ok(validated)
    }

    fn validate_staged_snapshot_matches_payload(
        payload: &BudgetStoreSnapshot,
        staged: &BudgetStoreSnapshot,
    ) -> Result<(), BudgetStoreError> {
        let mut payload_usages = payload.usages.clone();
        payload_usages.sort_by(|left, right| {
            (&left.capability_id, left.grant_index).cmp(&(&right.capability_id, right.grant_index))
        });
        if payload_usages != staged.usages {
            return Err(BudgetStoreError::Invariant(
                "budget snapshot does not carry its complete final usage projection".to_string(),
            ));
        }

        let same_event_history = payload.mutation_events.len() == staged.mutation_events.len()
            && payload
                .mutation_events
                .iter()
                .zip(&staged.mutation_events)
                .all(|(payload, staged)| {
                    payload.event_id == staged.event_id && payload.event_seq == staged.event_seq
                });
        let same_abandoned_history =
            payload.abandoned_seq_ranges.iter().all(|&(start, end)| {
                snapshot_ranges_cover(&staged.abandoned_seq_ranges, start, end)
            }) && staged.abandoned_seq_ranges.iter().all(|&(start, end)| {
                snapshot_ranges_cover(&payload.abandoned_seq_ranges, start, end)
            });
        if !same_event_history || !same_abandoned_history {
            return Err(BudgetStoreError::Invariant(
                "budget snapshot validation changed the supplied durable history".to_string(),
            ));
        }
        Ok(())
    }

    fn seed_snapshot_validation_anchors(
        validator: &Self,
        anchors: &[BudgetUsageRecord],
    ) -> Result<(), BudgetStoreError> {
        let mut connection = validator.connection()?;
        let transaction = validator.begin_write(&mut connection)?;
        transaction.execute(
            "INSERT OR IGNORE INTO budget_usage_anchor_migration_gate(singleton) VALUES (1)",
            [],
        )?;
        for anchor in anchors {
            Self::upsert_usage_in_transaction(&transaction, anchor)?;
            transaction.execute(
                r#"
                INSERT INTO budget_usage_history_anchors (
                    capability_id, grant_index, invocation_count, updated_at, seq,
                    total_cost_exposed, total_cost_realized_spend,
                    anchored_schema_version
                ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 6)
                "#,
                params![
                    &anchor.capability_id,
                    i64::from(anchor.grant_index),
                    i64::from(anchor.invocation_count),
                    anchor.updated_at,
                    budget_u64_to_sqlite(anchor.seq, "anchor_seq")?,
                    budget_u64_to_sqlite(anchor.total_cost_exposed, "anchor_total_cost_exposed")?,
                    budget_u64_to_sqlite(
                        anchor.total_cost_realized_spend,
                        "anchor_total_cost_realized_spend"
                    )?,
                ],
            )?;
        }
        transaction.execute("DELETE FROM budget_usage_anchor_migration_gate", [])?;
        transaction.commit()?;
        Ok(())
    }

    fn validate_exact_snapshot_usage_anchors(
        transaction: &rusqlite::Transaction<'_>,
        anchors: &[BudgetUsageRecord],
    ) -> Result<(), BudgetStoreError> {
        Self::install_snapshot_usage_anchors(transaction, anchors, false)?;
        let local = Self::snapshot_usage_history_anchors(transaction)?;
        let mut incoming = anchors.to_vec();
        incoming.sort_by(|left, right| {
            (&left.capability_id, left.grant_index).cmp(&(&right.capability_id, right.grant_index))
        });
        if local != incoming {
            return Err(BudgetStoreError::Invariant(
                "budget snapshot history anchor set differs from immutable local migration anchors"
                    .to_string(),
            ));
        }
        Ok(())
    }

    fn validate_local_snapshot_subset(
        transaction: &rusqlite::Transaction<'_>,
        snapshot: &BudgetStoreSnapshot,
    ) -> Result<(), BudgetStoreError> {
        let incoming_events = snapshot
            .mutation_events
            .iter()
            .map(|event| (event.event_id.as_str(), event))
            .collect::<std::collections::BTreeMap<_, _>>();
        for local in Self::snapshot_mutation_events(transaction)? {
            if incoming_events.get(local.event_id.as_str()).copied() != Some(&local) {
                return Err(BudgetStoreError::Invariant(format!(
                    "budget snapshot does not retain identical local event `{}`",
                    local.event_id
                )));
            }
        }

        for (start, end) in Self::snapshot_abandoned_seq_ranges(transaction)? {
            if !snapshot_ranges_cover(&snapshot.abandoned_seq_ranges, start, end) {
                return Err(BudgetStoreError::Invariant(format!(
                    "budget snapshot does not retain local abandoned sequence range {start}..={end}"
                )));
            }
        }

        for usage in Self::snapshot_usages(transaction)? {
            if !snapshot_proves_usage(snapshot, &usage) {
                return Err(BudgetStoreError::Invariant(format!(
                    "budget snapshot does not prove local usage `{}` grant {} at sequence {}",
                    usage.capability_id, usage.grant_index, usage.seq
                )));
            }
        }
        Ok(())
    }

    fn install_snapshot_coverage(
        transaction: &rusqlite::Transaction<'_>,
        covered_head: u64,
        origin_ack_heads: &[(String, u64)],
    ) -> Result<(), BudgetStoreError> {
        let covered_head = budget_u64_to_sqlite(covered_head, "covered_head")?;
        transaction.execute(
            "UPDATE budget_snapshot_coverage SET covered_head = ?1 WHERE singleton = 1",
            params![covered_head],
        )?;
        transaction.execute(
            "UPDATE budget_ack_head_watermark SET head_seq = ?1 WHERE singleton = 1",
            params![covered_head],
        )?;
        transaction.execute("DELETE FROM budget_origin_ack_heads", [])?;
        for (authority_id, head_seq) in origin_ack_heads {
            transaction.execute(
                "INSERT INTO budget_origin_ack_heads (authority_id, head_seq) VALUES (?1, ?2)",
                params![
                    authority_id,
                    budget_u64_to_sqlite(*head_seq, "origin_ack_head")?
                ],
            )?;
        }
        raise_budget_replication_seq_floor(transaction, covered_head as u64)?;
        Ok(())
    }

    fn origin_ack_heads_at(
        connection: &Connection,
        covered_head: u64,
    ) -> Result<Vec<(String, u64)>, BudgetStoreError> {
        let mut statement = connection.prepare(
            r#"
            SELECT authority_id, MAX(event_seq)
            FROM budget_mutation_events
            WHERE authority_id IS NOT NULL AND event_seq <= ?1
            GROUP BY authority_id
            ORDER BY authority_id
            "#,
        )?;
        let rows = statement
            .query_map(
                params![budget_u64_to_sqlite(covered_head, "covered_head")?],
                |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        budget_u64_from_row(row, 1, "origin_ack_head")?,
                    ))
                },
            )?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(rows)
    }

    pub fn export_budget_snapshot(&self) -> Result<BudgetStoreSnapshot, BudgetStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let usages = Self::snapshot_usages(&transaction)?;
        let usage_history_anchors = Self::snapshot_usage_history_anchors(&transaction)?;
        let mutation_events = Self::snapshot_mutation_events(&transaction)?;
        let abandoned_seq_ranges = Self::snapshot_abandoned_seq_ranges(&transaction)?;
        let covered_head = Self::contiguous_snapshot_head_in(&transaction)?;
        let origin_ack_heads = Self::origin_ack_heads_at(&transaction, covered_head)?;
        transaction.rollback()?;
        Ok(BudgetStoreSnapshot {
            usages,
            usage_history_anchors,
            mutation_events,
            abandoned_seq_ranges,
            covered_head,
            origin_ack_heads,
        })
    }

    fn snapshot_usages(
        connection: &Connection,
    ) -> Result<Vec<BudgetUsageRecord>, BudgetStoreError> {
        let mut statement = connection.prepare(
            r#"
            SELECT capability_id, grant_index, invocation_count, updated_at, seq,
                   total_cost_exposed, total_cost_realized_spend
            FROM capability_grant_budgets
            ORDER BY capability_id, grant_index
            "#,
        )?;
        let rows = statement
            .query_map([], record_from_row)?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(rows)
    }

    fn snapshot_usage_history_anchors(
        connection: &Connection,
    ) -> Result<Vec<BudgetUsageRecord>, BudgetStoreError> {
        let mut statement = connection.prepare(
            r#"
            SELECT capability_id, grant_index, invocation_count, updated_at, seq,
                   total_cost_exposed, total_cost_realized_spend
            FROM budget_usage_history_anchors
            ORDER BY capability_id, grant_index
            "#,
        )?;
        let rows = statement
            .query_map([], record_from_row)?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(rows)
    }

    fn snapshot_mutation_events(
        transaction: &rusqlite::Transaction<'_>,
    ) -> Result<Vec<BudgetMutationRecord>, BudgetStoreError> {
        let event_ids = {
            let mut statement = transaction
                .prepare("SELECT event_id FROM budget_mutation_events ORDER BY event_seq")?;
            let event_ids = statement
                .query_map([], |row| row.get::<_, String>(0))?
                .collect::<Result<Vec<_>, _>>()?;
            event_ids
        };
        event_ids
            .iter()
            .map(|event_id| {
                Self::load_projected_mutation_event(transaction, event_id)?.ok_or_else(|| {
                    BudgetStoreError::Invariant(format!(
                        "budget mutation event `{event_id}` disappeared during snapshot export"
                    ))
                })
            })
            .collect()
    }

    fn snapshot_abandoned_seq_ranges(
        connection: &Connection,
    ) -> Result<Vec<(u64, u64)>, BudgetStoreError> {
        let mut statement = connection.prepare(
            r#"
            SELECT MIN(seq), MAX(seq)
            FROM (
                SELECT seq, seq - ROW_NUMBER() OVER (ORDER BY seq) AS island
                FROM budget_abandoned_event_seqs
            )
            GROUP BY island
            ORDER BY MIN(seq)
            "#,
        )?;
        let rows = statement
            .query_map([], |row| {
                Ok((
                    budget_u64_from_row(row, 0, "abandoned_range_start")?,
                    budget_u64_from_row(row, 1, "abandoned_range_end")?,
                ))
            })?
            .collect::<Result<Vec<_>, _>>()?;
        Ok(rows)
    }

    pub fn budget_snapshot_covered_head(&self) -> Result<u64, BudgetStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let head = Self::contiguous_snapshot_head_in(&transaction)?;
        transaction.rollback()?;
        Ok(head)
    }

    pub(super) fn contiguous_snapshot_head_in(
        connection: &Connection,
    ) -> Result<u64, BudgetStoreError> {
        Self::contiguous_snapshot_head_from(connection, 0)
    }

    pub(super) fn rebuild_snapshot_proof_caches(
        connection: &mut Connection,
    ) -> Result<(), BudgetStoreError> {
        let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?;
        let covered_head = Self::contiguous_snapshot_head_from(&transaction, 0)?;
        let origin_ack_heads = Self::origin_ack_heads_at(&transaction, covered_head)?;
        Self::install_snapshot_coverage(&transaction, covered_head, &origin_ack_heads)?;
        transaction.commit()?;
        Ok(())
    }

    fn contiguous_snapshot_head_from(
        connection: &Connection,
        floor: u64,
    ) -> Result<u64, BudgetStoreError> {
        let floor_sqlite = budget_u64_to_sqlite(floor, "snapshot_covered_head")?;
        let next_slot = floor_sqlite.checked_add(1).ok_or_else(|| {
            BudgetStoreError::Overflow("budget snapshot head overflowed i64".to_string())
        })?;
        let next_slot_filled: bool = connection.query_row(
            r#"
            SELECT EXISTS(
                SELECT 1 FROM budget_mutation_events WHERE event_seq = ?1
                UNION ALL
                SELECT 1 FROM budget_abandoned_event_seqs WHERE seq = ?1
            )
            "#,
            params![next_slot],
            |row| row.get::<_, i64>(0).map(|value| value != 0),
        )?;
        if !next_slot_filled {
            return Ok(floor);
        }
        let head: i64 = connection.query_row(
            r#"
            WITH filled AS (
                SELECT event_seq AS seq
                FROM budget_mutation_events
                WHERE event_seq IS NOT NULL AND event_seq > ?1
                UNION
                SELECT seq
                FROM budget_abandoned_event_seqs
                WHERE seq > ?1
            ),
            run AS (
                SELECT seq, seq - ROW_NUMBER() OVER (ORDER BY seq) AS island
                FROM filled
            )
            SELECT COALESCE(MAX(seq), ?1)
            FROM run
            WHERE island = ?1
            "#,
            params![floor_sqlite],
            |row| row.get(0),
        )?;
        u64::try_from(head).map_err(|_| {
            BudgetStoreError::Invariant("budget snapshot head has a negative sequence".to_string())
        })
    }

    pub fn list_usage_history_anchors(&self) -> Result<Vec<BudgetUsageRecord>, BudgetStoreError> {
        let mut connection = self.connection()?;
        let transaction = self.begin_read(&mut connection)?;
        let anchors = {
            let mut statement = transaction.prepare(
                r#"
                SELECT capability_id, grant_index, invocation_count, updated_at, seq,
                       total_cost_exposed, total_cost_realized_spend
                FROM budget_usage_history_anchors
                ORDER BY capability_id, grant_index
                "#,
            )?;
            let rows = statement
                .query_map([], record_from_row)?
                .collect::<Result<Vec<_>, _>>()?;
            rows
        };
        transaction.rollback()?;
        Ok(anchors)
    }
}

fn snapshot_ranges_cover(ranges: &[(u64, u64)], start: u64, end: u64) -> bool {
    let mut next = start;
    for &(range_start, range_end) in ranges {
        if range_end < next {
            continue;
        }
        if range_start > next {
            return false;
        }
        if range_end >= end {
            return true;
        }
        let Some(after_range) = range_end.checked_add(1) else {
            return true;
        };
        next = after_range;
    }
    false
}

fn snapshot_proves_usage(snapshot: &BudgetStoreSnapshot, usage: &BudgetUsageRecord) -> bool {
    snapshot
        .usage_history_anchors
        .iter()
        .any(|anchor| anchor == usage)
        || snapshot.mutation_events.iter().any(|event| {
            event.capability_id == usage.capability_id
                && event.grant_index == usage.grant_index
                && event.event_seq == usage.seq
                && event.usage_seq == Some(event.event_seq)
                && event.recorded_at == usage.updated_at
                && event.invocation_count_after == usage.invocation_count
                && event.total_cost_exposed_after == usage.total_cost_exposed
                && event.total_cost_realized_spend_after == usage.total_cost_realized_spend
        })
}

fn verify_budget_snapshot_anchor_provenance<'a>(
    snapshot: &BudgetStoreSnapshot,
    provenance: &'a BudgetSnapshotAnchorProvenance,
    expected_leader_url: &str,
    expected_election_term: u64,
    cluster_service_token: &str,
) -> Result<&'a BudgetSnapshotAnchorCommitment, BudgetStoreError> {
    if expected_leader_url.is_empty()
        || expected_election_term == 0
        || cluster_service_token.is_empty()
        || provenance.chain.is_empty()
        || provenance.schema != "chio.budget-snapshot-anchor-provenance.v1"
    {
        return Err(BudgetStoreError::Invariant(
            "budget snapshot anchor provenance trust context is incomplete".to_string(),
        ));
    }
    let expected_authenticator =
        budget_snapshot_anchor_authenticator(cluster_service_token, &provenance.chain)?;
    if !bool::from(
        provenance
            .cluster_authenticator
            .as_bytes()
            .ct_eq(expected_authenticator.as_bytes()),
    ) {
        return Err(BudgetStoreError::Invariant(
            "budget snapshot anchor provenance cluster authentication failed".to_string(),
        ));
    }
    let expected_anchor_digest =
        budget_snapshot_anchor_set_digest(&snapshot.usage_history_anchors)?;
    let mut previous_digest = BUDGET_ANCHOR_GENESIS_DIGEST;
    for (index, signed) in provenance.chain.iter().enumerate() {
        let expected_sequence = u64::try_from(index)
            .ok()
            .and_then(|value| value.checked_add(1))
            .ok_or_else(|| {
                BudgetStoreError::Invariant(
                    "budget snapshot anchor provenance sequence overflowed".to_string(),
                )
            })?;
        let body = &signed.body;
        if body.schema != "chio.budget-snapshot-anchor-commitment.v1"
            || body.commit_sequence != expected_sequence
            || body.previous_chain_digest != previous_digest
            || body.election_term == 0
            || body.leader_url.is_empty()
            || body.signer_public_key.is_empty()
            || budget_snapshot_anchor_chain_digest(body)? != body.chain_digest
        {
            return Err(BudgetStoreError::Invariant(
                "budget snapshot anchor provenance chain is invalid".to_string(),
            ));
        }
        let signer = PublicKey::from_hex(&body.signer_public_key)
            .map_err(|error| BudgetStoreError::Invariant(error.to_string()))?;
        if !signer
            .verify_canonical(body, &signed.signature)
            .map_err(|error| BudgetStoreError::Invariant(error.to_string()))?
        {
            return Err(BudgetStoreError::Invariant(
                "budget snapshot anchor provenance signature is invalid".to_string(),
            ));
        }
        previous_digest = &body.chain_digest;
    }
    let head = provenance.chain.last().ok_or_else(|| {
        BudgetStoreError::Invariant("budget snapshot anchor provenance is empty".to_string())
    })?;
    if head.body.leader_url != expected_leader_url
        || head.body.election_term != expected_election_term
        || head.body.anchor_set_digest != expected_anchor_digest
    {
        return Err(BudgetStoreError::Invariant(
            "budget snapshot anchor provenance is not bound to the elected leader, term, and exact anchor set"
                .to_string(),
        ));
    }
    Ok(&head.body)
}

fn verify_local_anchor_provenance_continuity(
    transaction: &rusqlite::Transaction<'_>,
    provenance: &BudgetSnapshotAnchorProvenance,
) -> Result<(), BudgetStoreError> {
    let head = provenance.chain.last().ok_or_else(|| {
        BudgetStoreError::Invariant("budget snapshot anchor provenance is empty".to_string())
    })?;
    let local = transaction
        .query_row(
            r#"
            SELECT commit_sequence, chain_digest
            FROM budget_snapshot_anchor_provenance
            WHERE leader_url = ?1
            "#,
            params![&head.body.leader_url],
            |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
        )
        .optional()?;
    let Some((local_sequence, local_digest)) = local else {
        return Ok(());
    };
    let local_sequence = local_sequence.max(0) as u64;
    let index = usize::try_from(local_sequence.saturating_sub(1)).map_err(|_| {
        BudgetStoreError::Invariant(
            "persisted budget snapshot anchor provenance sequence is invalid".to_string(),
        )
    })?;
    let committed = provenance.chain.get(index).ok_or_else(|| {
        BudgetStoreError::Invariant(
            "budget snapshot anchor provenance rewinds the persisted leader chain".to_string(),
        )
    })?;
    if committed.body.commit_sequence != local_sequence
        || committed.body.chain_digest != local_digest
    {
        return Err(BudgetStoreError::Invariant(
            "budget snapshot anchor provenance forks the persisted leader chain".to_string(),
        ));
    }
    Ok(())
}