axond 0.3.35

//! The durable control plane: PostgreSQL.
//!
//! The one thing this module has to get right is that a publication is a single
//! transaction. The head pointer is read `FOR UPDATE`, so concurrent publishers
//! serialize on one row rather than racing to append; the manifest, the resource
//! versions, the blob references, the mutation, the audit event, the idempotency
//! record, and the head advancement are written in that transaction and become
//! visible together. Anything that fails rolls the whole thing back, which is why
//! a rejected publication leaves no resource version and no audit event — not
//! "leaves them to be cleaned up".
//!
//! What it deliberately does not do:
//!
//! - **It is never on the inference path.** Nothing here is called while a
//!   request is in flight, so a Postgres outage stalls administration and
//!   convergence while replicas keep serving the immutable snapshot they already
//!   hold. That is why the failure of every method is an error returned to an
//!   administrative caller and never a mutation of anything a replica is serving.
//! - **It does not compile or publish a snapshot.** Hydration into runtime state
//!   is #166, publication to replicas is #142. This is the seam they read from:
//!   [`ControlPlaneStore::load_revision`] returns a [`LoadedRevision`], which
//!   cannot exist unless it verified.
//! - **It does not store payload bytes.** A blob is a reference — kind, digest,
//!   size — because a revision that inlined megabytes would duplicate them per
//!   revision. Where the payload lives, and who fetches and verifies it, belongs
//!   with hydration.
//! - **It does not store secret material.** A credential resource's body holds an
//!   opaque secret *reference*; plaintext lives in the secret store, and no value
//!   from a body is ever logged.

use std::collections::{BTreeMap, BTreeSet};
use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use async_trait::async_trait;
use tokio_postgres::error::SqlState;
use tokio_postgres::types::ToSql;
use tokio_postgres::{Client, Config, Row, Transaction};

use super::hydration::{self, HydrationLimits};
use super::rows;
use super::schema::{self, Baseline, MINIMUM_SERVER_VERSION_NUM, SchemaStatus};
use super::{ControlPlaneError, ControlPlaneStore, StatusProbeAdmission};
#[cfg(test)]
use crate::availability::EvidenceClear;
use crate::availability::store::{
    self as availability_store, EvidenceWrite, ObservationSlot, ObservationStore, StoredObservation,
};
use crate::availability::{AvailabilityKey, DiscoveryObservation, ScopeRef, TargetRef};
use crate::backends::{Capabilities, Capability};
use crate::desired_state::{
    AccessDenial, Action, AuditEvent, Credential, DenialPage, DenialReason, Directory,
    IntegrityError, LoadedRevision, Mutation, ProjectId, ResourceRef, ResourceVersion,
    ResourceVersionNumber, RevisionCandidate, RevisionId, RevisionManifest, SerializerVersion,
    Surface, Tenancy, TenantId, Uuid7Generator,
};

const BACKEND: &str = "postgres";

/// How the store connects, and what it is allowed to do at boot.
#[derive(Debug, Clone)]
pub struct ControlPlaneSettings {
    /// The PostgreSQL schema the journal lives in, if not the connection's
    /// default. Validated as an identifier, because it is interpolated into
    /// `SET search_path`.
    pub schema: Option<String>,
    /// Whether boot may apply pending migrations.
    ///
    /// An operator who applies DDL out of band sets this to `false` and gets a
    /// refusal instead of a schema change; boot still *checks*, because serving
    /// against a schema that is not the one this build writes is the failure this
    /// setting exists to prevent.
    pub migrate: bool,
    pub connect_timeout: Duration,
    /// The ceiling on one control-plane operation. Generous by inference-path
    /// standards: nothing here is called with a request in flight.
    pub operation_timeout: Duration,
    /// How long an idempotency record is honoured. A retry window, not a
    /// permanent namespace: expiry never touches the revision or audit trail the
    /// record points at.
    pub idempotency_retention: Duration,
    /// What one read of a stored revision may consume. Reads are bounded because
    /// what they read is storage, and storage is not this build's to trust.
    pub hydration: HydrationLimits,
}

impl ControlPlaneSettings {
    /// The settings a `[control_plane]` section asks for.
    ///
    /// Every field the section can express is mapped here, so a setting an
    /// operator writes cannot be silently ignored by a code path that built its
    /// own defaults instead. The fields it cannot express — the idempotency
    /// retention window and the hydration limits — are runtime serving
    /// parameters rather than connection or schema ones, and keep their defaults.
    pub fn from_config(control_plane: &crate::config::ControlPlane) -> Self {
        Self {
            schema: control_plane
                .schema
                .as_deref()
                .map(str::trim)
                .filter(|schema| !schema.is_empty())
                .map(str::to_owned),
            migrate: control_plane.migrate,
            connect_timeout: Duration::from_millis(control_plane.connect_timeout_ms),
            operation_timeout: Duration::from_millis(control_plane.operation_timeout_ms),
            ..Self::default()
        }
    }

    /// The settings a maintenance command runs with: the operator's connection
    /// and schema, and never boot's permission to migrate. `axond migrate apply`
    /// is the mutation, and it is explicit; nothing else may become one because
    /// the config happened to allow boot migrations.
    pub fn for_maintenance(control_plane: &crate::config::ControlPlane) -> Self {
        Self {
            migrate: false,
            ..Self::from_config(control_plane)
        }
    }

    /// The health-call budget for a probe with `queued_operations` already
    /// ahead of it. Each queued operation can consume the full operation bound,
    /// followed by one reconnect and the probe's own health call.
    pub(crate) fn status_probe_timeout(&self, queued_operations: usize) -> Duration {
        let operations = u32::try_from(queued_operations.saturating_add(1)).unwrap_or(u32::MAX);
        self.operation_timeout
            .saturating_mul(operations)
            .saturating_add(self.connect_timeout)
            // Keep the derived status pacing valid for sub-second settings too.
            .max(Duration::from_secs(2))
    }
}

impl Default for ControlPlaneSettings {
    fn default() -> Self {
        Self {
            schema: None,
            migrate: true,
            connect_timeout: Duration::from_secs(10),
            operation_timeout: Duration::from_secs(30),
            idempotency_retention: Duration::from_secs(24 * 60 * 60),
            hydration: HydrationLimits::default(),
        }
    }
}

/// What adopting an empty migration ledger did, or found already done.
///
/// Two outcomes rather than one, because "recorded a baseline" and "there was
/// already a history" are different things for an operator to read: the first
/// changed a database and names what it now claims was applied, and the second
/// changed nothing.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Adoption {
    /// These versions are now recorded as applied, on the evidence of the tables
    /// they declare. `status` is what the schema classifies as afterwards —
    /// current when the whole shipped history was adopted, behind when an `apply`
    /// still has versions to add.
    Recorded {
        versions: Vec<i32>,
        status: SchemaStatus,
    },
    /// The ledger already recorded a history, so nothing was written.
    AlreadyRecorded { status: SchemaStatus },
}

/// A [`ControlPlaneStore`] backed by the revision journal in `ops/postgres/`.
pub struct PostgresControlPlane {
    config: Config,
    settings: ControlPlaneSettings,
    /// Set on every connection, including reconnections: a reconnect that landed
    /// on the default schema would silently read a different journal.
    search_path: Option<String>,
    ids: Uuid7Generator,
    client: tokio::sync::Mutex<Option<Client>>,
    /// Operations currently holding or waiting for the serialized client. A
    /// status probe samples this before joining the queue so its timeout covers
    /// every operation already ahead of it, not just one.
    pending_operations: Arc<AtomicUsize>,
}

/// Written by hand, and deliberately narrow: a derived one would print the
/// [`Config`], which carries the password from the DSN.
impl std::fmt::Debug for PostgresControlPlane {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("PostgresControlPlane")
            .field("schema", &self.search_path)
            .field("migrate", &self.settings.migrate)
            .finish_non_exhaustive()
    }
}

impl PostgresControlPlane {
    /// Connect, check the server and schema, and optionally migrate.
    ///
    /// Boot refuses rather than degrades. An unsupported server, a schema a newer
    /// build owns, a migration that was edited after being applied, or a pending
    /// migration this store may not apply are all [`ControlPlaneError::Denied`]:
    /// they need an operator, and retrying cannot help.
    pub async fn connect(
        dsn: &str,
        settings: ControlPlaneSettings,
    ) -> Result<Self, ControlPlaneError> {
        let (store, mut client) = Self::open(dsn, settings).await?;
        store.prepare_schema(&mut client).await?;
        *store.client.lock().await = Some(client);
        Ok(store)
    }

    /// Connect for schema maintenance, without deciding anything about the schema.
    ///
    /// [`connect`](Self::connect) is a boot: it refuses a schema it cannot serve.
    /// A `status` command has to *report* that schema instead, and an `apply` has
    /// to be able to run against the database a boot just refused, so this checks
    /// the server version — a statement no migration can survive an older server
    /// running — and stops there. The store it returns is not for serving: it has
    /// not established that the journal is the one this build writes.
    pub async fn connect_for_maintenance(
        dsn: &str,
        settings: ControlPlaneSettings,
    ) -> Result<Self, ControlPlaneError> {
        let (store, client) = Self::open(dsn, settings).await?;
        *store.client.lock().await = Some(client);
        Ok(store)
    }

    /// Connect and check the server, leaving the schema decision to the caller.
    async fn open(
        dsn: &str,
        settings: ControlPlaneSettings,
    ) -> Result<(Self, Client), ControlPlaneError> {
        let mut config: Config = dsn.parse().map_err(|error| {
            denied(format!(
                // The DSN itself is never echoed: it carries a password.
                "the control-plane DSN could not be parsed: {error}"
            ))
        })?;
        config.connect_timeout(settings.connect_timeout);
        config.application_name(crate::telemetry::SERVICE_NAME);
        let search_path = settings
            .schema
            .as_deref()
            .map(|schema| {
                crate::usage::validate_table_name(schema).map_err(denied)?;
                // The table-name validator allows one qualifying dot; a search path
                // takes one unqualified schema. Checked here as well as in config
                // validation, because settings can be built without a config file.
                if schema.contains('.') {
                    return Err(denied(format!(
                        "`{schema}` is not a single unqualified schema name"
                    )));
                }
                Ok(schema.to_owned())
            })
            .transpose()?;

        let store = Self {
            config,
            settings,
            search_path,
            ids: Uuid7Generator::new(),
            client: tokio::sync::Mutex::new(None),
            pending_operations: Arc::new(AtomicUsize::new(0)),
        };
        let client = tokio::time::timeout(store.settings.connect_timeout, store.connect_client())
            .await
            .map_err(|_| ControlPlaneError::Unavailable {
                backend: BACKEND,
                message: "connection timed out".to_owned(),
            })?
            .map_err(|error| unavailable("connect", &error))?;
        store.check_server_version(&client).await?;
        Ok((store, client))
    }

    /// Apply every migration the journal is missing, and report which ones.
    ///
    /// Forward-only and idempotent: a journal that is already current is left
    /// untouched and reports nothing applied, so re-running an `apply` is not a
    /// second migration. A journal this build cannot own — [`SchemaStatus::Ahead`]
    /// or [`SchemaStatus::Drifted`] — is refused with [`ControlPlaneError::Denied`]
    /// rather than written to, because writing more DDL over it makes the
    /// disagreement worse rather than better.
    ///
    /// The whole read-and-apply is one transaction under the same advisory lock
    /// boot takes, so this is safe to run while replicas are starting.
    pub async fn apply_migrations(&self) -> Result<Vec<i32>, ControlPlaneError> {
        self.run(None, |client| {
            Box::pin(async move {
                // Read committed explicitly rather than by inheritance: the status
                // is read *after* the lock is granted, and it has to be the winner's
                // committed history rather than a snapshot taken before the wait. A
                // server with `default_transaction_isolation = 'repeatable read'`
                // would freeze the snapshot at the first statement, so the loser of
                // the lock would re-read a pre-migration ledger and apply the same
                // files again.
                let transaction = client
                    .build_transaction()
                    .isolation_level(tokio_postgres::IsolationLevel::ReadCommitted)
                    .start()
                    .await
                    .map_err(|error| unavailable("begin schema transaction", &error))?;
                transaction
                    .query_one("SELECT pg_advisory_xact_lock($1::bigint)", &[&SCHEMA_LOCK])
                    .await
                    .map_err(|error| unavailable("acquire schema lock", &error))?;
                let status = schema::status(&transaction)
                    .await
                    .map_err(|error| unavailable("read schema status", &error))?;
                // A database that is already current is left alone rather than
                // handed to the runner: `migrate` starts from the recorded prefix
                // of a `Behind` status, so a `Current` one would re-run every
                // shipped file from v1. That happens to be harmless while every
                // statement is `IF NOT EXISTS`, and would not survive the first
                // `ALTER TABLE` or backfill. Returning here also rolls the
                // transaction back, so a repeated apply writes nothing at all.
                if status.is_current() {
                    return Ok(Vec::new());
                }
                if !status.is_migratable() {
                    return Err(denied(status.to_string()));
                }
                let pending = schema::pending(&status);
                schema::migrate(&transaction, &status)
                    .await
                    .map_err(|error| migration_refused_or_unavailable(&error))?;
                let migrated = schema::status(&transaction)
                    .await
                    .map_err(|error| unavailable("re-read schema status", &error))?;
                if !migrated.is_current() {
                    return Err(denied(format!(
                        "migrations were applied but the schema is still not current: {migrated}"
                    )));
                }
                transaction
                    .commit()
                    .await
                    .map_err(|error| unavailable("commit schema transaction", &error))?;
                Ok(pending)
            })
        })
        .await
    }

    /// Adopt a hand-applied schema: record the baseline an empty ledger cannot.
    ///
    /// The one write that exists for a [`SchemaStatus::Unrecorded`] database, and
    /// the only one anywhere that inserts a ledger row for DDL it did not run.
    /// It writes what the database itself accounts for and nothing else: the
    /// longest prefix of shipped migrations whose every statement is confirmed —
    /// tables and indexes present, idempotent seed rows written
    /// ([`schema::baseline`]). A prefix that is empty, one interrupted by a
    /// half-applied migration, one that is not a prefix at all, and any history
    /// containing a statement whose effect cannot be confirmed are refused with
    /// [`ControlPlaneError::Denied`] rather than recorded, because a recorded
    /// version is afterwards indistinguishable from an applied one — an adoption
    /// that guessed would launder a guess into the history every later decision
    /// reads.
    ///
    /// Never a migration: no shipped SQL runs here, so adopting cannot execute a
    /// file against a database that already has it. A ledger that already records
    /// a history is left exactly as it is and reported back, which is what makes a
    /// second `adopt` a no-op rather than a second baseline.
    pub async fn adopt_ledger(&self) -> Result<Adoption, ControlPlaneError> {
        self.run(None, |client| {
            Box::pin(async move {
                // Read committed and the journal's advisory lock, for the reasons
                // `apply_migrations` pins both: the status has to be the winner's
                // committed history, and an adoption racing an apply must be one
                // decision rather than two.
                let transaction = client
                    .build_transaction()
                    .isolation_level(tokio_postgres::IsolationLevel::ReadCommitted)
                    .start()
                    .await
                    .map_err(|error| unavailable("begin schema transaction", &error))?;
                transaction
                    .query_one("SELECT pg_advisory_xact_lock($1::bigint)", &[&SCHEMA_LOCK])
                    .await
                    .map_err(|error| unavailable("acquire schema lock", &error))?;
                let status = schema::status(&transaction)
                    .await
                    .map_err(|error| unavailable("read schema status", &error))?;
                match status {
                    // A ledger that already answers the question adoption exists to
                    // answer. Reported rather than refused: the operator asked for a
                    // recorded history and there is one, so `adopt` is idempotent
                    // and re-running it after an apply is a no-op.
                    SchemaStatus::Current { .. } | SchemaStatus::Behind { .. } => {
                        let _ = transaction.rollback().await;
                        return Ok(Adoption::AlreadyRecorded { status });
                    }
                    SchemaStatus::Unrecorded => {}
                    // Every other status is a decision that is not adoption's:
                    // absent has nothing to adopt (and `apply` is the command for
                    // it), and the rest are histories this build must not touch.
                    refused => {
                        return Err(denied(format!(
                            "adoption reconciles an existing but empty migration ledger, and this \
                             database has something else: {refused}"
                        )));
                    }
                }
                // Classified like the writes rather than like an ordinary read:
                // this reads the objects an out-of-band `psql` created, plausibly
                // as a different role, so `42501` on the seed probe is a realistic
                // failure and no retry clears it.
                let baseline = schema::baseline(&transaction).await.map_err(|error| {
                    refused_or_unavailable("reconciling the empty ledger", &error)
                })?;
                let versions = match baseline {
                    Baseline::Applied { versions } => versions,
                    Baseline::Nothing => {
                        // Naming the schema it looked in, because a DSN with a
                        // multi-entry `search_path` can have the ledger answer from
                        // one schema and the objects be sought in the one this
                        // connection writes to. "Nothing is present" without a
                        // *where* would read as a claim about the whole database.
                        let searched: String = transaction
                            .query_one("SELECT current_schema()", &[])
                            .await
                            .map_err(|error| {
                                refused_or_unavailable("reading the current schema", &error)
                            })?
                            .get(0);
                        return Err(denied(format!(
                            "no shipped migration's tables are present in schema `{searched}`, so \
                             this database has no applied schema to adopt there; drop the empty \
                             `{}` table and run `axond migrate apply`",
                            schema::MIGRATION_TABLE
                        )));
                    }
                    Baseline::Inconsistent { message } => return Err(denied(message)),
                };
                // Named for what it is: adoption runs no migration, so a rejected
                // ledger write must not send the operator looking at DDL.
                schema::record_baseline(&transaction, &versions)
                    .await
                    .map_err(|error| {
                        refused_or_unavailable("recording the adopted baseline", &error)
                    })?;
                // The recorded baseline has to classify as a history this build
                // can extend, or it is not a baseline: adopting must leave the
                // database in a state `status` and `apply` already understand.
                let adopted = schema::status(&transaction)
                    .await
                    .map_err(|error| unavailable("re-read schema status", &error))?;
                if !matches!(
                    adopted,
                    SchemaStatus::Current { .. } | SchemaStatus::Behind { .. }
                ) {
                    return Err(denied(format!(
                        "a baseline was recorded but the schema is still not one this build can \
                         extend: {adopted}"
                    )));
                }
                transaction
                    .commit()
                    .await
                    .map_err(|error| unavailable("commit schema transaction", &error))?;
                Ok(Adoption::Recorded {
                    versions,
                    status: adopted,
                })
            })
        })
        .await
    }

    /// The schema state a database is in, for a status command or a boot refusal.
    ///
    /// Read-only as the *database* enforces it, not just by convention: the
    /// transaction is opened `READ ONLY`, so a statement that tried to write
    /// would be rejected by the server rather than by this code being careful.
    /// A status command an operator runs against production has to be a thing
    /// that cannot change production.
    pub async fn schema_status(&self) -> Result<SchemaStatus, ControlPlaneError> {
        self.run(None, |client| {
            Box::pin(async move {
                let transaction = client
                    .build_transaction()
                    .read_only(true)
                    .start()
                    .await
                    .map_err(|error| unavailable("begin read-only schema read", &error))?;
                let status = schema::status(&transaction)
                    .await
                    .map_err(|error| unavailable("read schema status", &error))?;
                // Rolled back rather than committed: nothing here is a change,
                // so nothing here needs to be kept.
                let _ = transaction.rollback().await;
                Ok(status)
            })
        })
        .await
    }

    async fn check_server_version(&self, client: &Client) -> Result<(), ControlPlaneError> {
        let reported: String = client
            .query_one("SELECT current_setting('server_version_num')", &[])
            .await
            .map_err(|error| unavailable("read server version", &error))?
            .get(0);
        let version: i32 = reported.parse().map_err(|_| {
            denied(format!(
                "the server reported version `{reported}`, which is not a number"
            ))
        })?;
        if version < MINIMUM_SERVER_VERSION_NUM {
            return Err(denied(format!(
                "the control-plane journal requires PostgreSQL {}, but the server is {}",
                MINIMUM_SERVER_VERSION_NUM / 10_000,
                version / 10_000
            )));
        }
        Ok(())
    }

    /// Bring the schema to the version this build requires, or refuse.
    ///
    /// The whole check-and-migrate runs under one advisory lock inside one
    /// transaction, so two gateways booting against the same empty database
    /// serialize here instead of both running the DDL and one of them failing
    /// halfway.
    async fn prepare_schema(&self, client: &mut Client) -> Result<(), ControlPlaneError> {
        // Read committed explicitly, for the reason `apply_migrations` pins it: the
        // status is read after the lock is granted and has to be the winner's
        // committed history. Under `repeatable read` the waiting replica would read
        // the pre-migration ledger — or, because catalog lookups stay fresh while row
        // reads do not, an existing table with no visible rows, which is a refusal.
        let transaction = client
            .build_transaction()
            .isolation_level(tokio_postgres::IsolationLevel::ReadCommitted)
            .start()
            .await
            .map_err(|error| unavailable("begin schema transaction", &error))?;
        transaction
            .query_one("SELECT pg_advisory_xact_lock($1::bigint)", &[&SCHEMA_LOCK])
            .await
            .map_err(|error| unavailable("acquire schema lock", &error))?;
        let status = schema::status(&transaction)
            .await
            .map_err(|error| unavailable("read schema status", &error))?;
        if !status.is_current() {
            if !status.is_migratable() {
                return Err(denied(status.to_string()));
            }
            if !self.settings.migrate {
                return Err(denied(format!(
                    "{status}, and this store is configured not to migrate"
                )));
            }
            schema::migrate(&transaction, &status)
                .await
                .map_err(|error| migration_refused_or_unavailable(&error))?;
            let migrated = schema::status(&transaction)
                .await
                .map_err(|error| unavailable("re-read schema status", &error))?;
            if !migrated.is_current() {
                return Err(denied(format!(
                    "migrations were applied but the schema is still not current: {migrated}"
                )));
            }
        }
        transaction
            .commit()
            .await
            .map_err(|error| unavailable("commit schema transaction", &error))?;
        Ok(())
    }

    async fn connect_client(&self) -> Result<Client, tokio_postgres::Error> {
        let (client, connection) = self.config.connect(crate::usage::tls_connector()).await?;
        tokio::spawn(async move {
            if let Err(error) = connection.await {
                tracing::warn!(%error, "postgres control-plane connection closed");
            }
        });
        if let Some(schema) = &self.search_path {
            // Validated as an identifier at construction, so this is not an
            // injection point; there is no parameter form of `SET`.
            client
                .batch_execute(&format!("SET search_path TO {schema}"))
                .await?;
        }
        Ok(client)
    }

    /// Run one operation on a connected client, reconnecting a dead connection
    /// and dropping one an outage broke.
    ///
    /// A failure that is *not* an outage — a conflict, an invalid candidate,
    /// unreadable storage — keeps the connection: it says nothing about the
    /// connection's health, and discarding it would turn every refused write into
    /// a reconnect.
    async fn run<T>(
        &self,
        admission: Option<StatusProbeAdmission>,
        operation: impl for<'a> FnOnce(
            &'a mut Client,
        ) -> Pin<
            Box<dyn Future<Output = Result<T, ControlPlaneError>> + Send + 'a>,
        >,
    ) -> Result<T, ControlPlaneError> {
        let _pending = admission
            .unwrap_or_else(|| StatusProbeAdmission::pending(Arc::clone(&self.pending_operations)));
        let mut guard = self.client.lock().await;
        if guard.as_ref().is_none_or(Client::is_closed) {
            *guard = Some(
                self.connect_client()
                    .await
                    .map_err(|error| unavailable("reconnect", &error))?,
            );
        }
        let result = tokio::time::timeout(
            self.settings.operation_timeout,
            operation(guard.as_mut().expect("connected")),
        )
        .await
        .map_err(|_| ControlPlaneError::Unavailable {
            backend: BACKEND,
            message: "operation timed out".to_owned(),
        })
        .and_then(|result| result);
        if matches!(result, Err(ControlPlaneError::Unavailable { .. })) {
            *guard = None;
        }
        result
    }
}

/// The advisory lock every gateway takes before touching the journal's schema.
/// A constant, because the journal's object names are constants.
const SCHEMA_LOCK: i64 = 0x1a20_de5c_0de5_1a11u64 as i64;

fn denied(message: impl Into<String>) -> ControlPlaneError {
    ControlPlaneError::Denied {
        backend: BACKEND,
        message: message.into(),
    }
}

/// Classify a failure to run the migration DDL by whether a retry could clear it.
fn migration_refused_or_unavailable(error: &tokio_postgres::Error) -> ControlPlaneError {
    refused_or_unavailable("applying migrations", error)
}

/// Classify a schema operation's failure by whether a retry could clear it.
///
/// The realistic failures on these paths are the server rejecting a statement,
/// not the server being unreachable: `3F000` when the configured
/// `[control_plane] schema` does not exist (`SET search_path` accepts a missing
/// schema, so the refusal arrives at the first `CREATE TABLE`), and class `42` —
/// `42501` when the role may not create the objects or read the ones an
/// out-of-band apply left behind, `42P01` when one of them is not there after all.
/// Every one of those needs an operator, and reporting them as an outage would
/// tell a rollout gate to retry a thing that cannot start working.
///
/// The transient classes stay outages so they keep their retryable
/// classification: connection exceptions (08), transaction rollbacks including
/// serialization failures and deadlocks (40), insufficient resources (53), an
/// object not in a prerequisite state such as an unavailable lock (55), operator
/// intervention including a cancelled statement or a shutdown (57), and system
/// errors (58). An error with no SQLSTATE never reached the server at all.
fn refused_or_unavailable(operation: &str, error: &tokio_postgres::Error) -> ControlPlaneError {
    const TRANSIENT: [&str; 6] = ["08", "40", "53", "55", "57", "58"];
    let Some(db) = error.as_db_error() else {
        return unavailable(operation, error);
    };
    let code = db.code().code();
    if code
        .get(..2)
        .is_some_and(|class| TRANSIENT.contains(&class))
    {
        return unavailable(operation, error);
    }
    denied(format!(
        "{operation} failed: {} (SQLSTATE {code}); the server rejected the statement, so no retry \
         clears it — check that the configured schema exists and that the role may read and \
         create objects in it",
        db.message()
    ))
}

/// Classify an observation write's SQLSTATE by whether the statement can ever
/// succeed unchanged. Integrity and data exceptions are permanent input/schema
/// refusals; connection and transaction failures remain retryable outages.
fn observation_write_failure(operation: &str, error: &tokio_postgres::Error) -> ControlPlaneError {
    let Some(db) = error.as_db_error() else {
        return unavailable(operation, error);
    };
    if is_permanent_observation_sqlstate(db.code()) {
        return denied(format!(
            "{operation} was permanently refused: {} (SQLSTATE {}); no retry of the same evidence clears it",
            db.message(),
            db.code().code()
        ));
    }
    unavailable(operation, error)
}

/// SQLSTATEs an evidence write cannot retry away without changing the row or
/// the schema. Keep this explicit: a foreign-key violation against a removed
/// tenant/project must not become an availability outage.
fn is_permanent_observation_sqlstate(code: &SqlState) -> bool {
    matches!(code.code().get(..2), Some("22" | "23"))
}

/// Report a Postgres failure as an outage, naming the operation and SQLSTATE.
///
/// SQLSTATE is included because "connection reset" and "deadlock detected" are
/// the same category to a caller and completely different to an operator.
pub(super) fn unavailable(operation: &str, error: &tokio_postgres::Error) -> ControlPlaneError {
    let message = match error.as_db_error() {
        Some(db) => format!(
            "{operation} failed: {} (SQLSTATE {})",
            db.message(),
            db.code().code()
        ),
        None => format!("{operation} failed: {error}"),
    };
    ControlPlaneError::Unavailable {
        backend: BACKEND,
        message,
    }
}

pub(super) fn corrupt_storage(detail: impl Into<String>) -> ControlPlaneError {
    ControlPlaneError::CorruptStorage {
        detail: detail.into(),
    }
}

/// A timestamp the journal can store and read back unchanged.
///
/// `timestamptz` holds microseconds, so a nanosecond-precision instant would come
/// back a different value than it went in — and a manifest that does not equal the
/// one publication returned is indistinguishable from a corrupt one.
fn journal_now() -> SystemTime {
    let now = SystemTime::now();
    match now.duration_since(UNIX_EPOCH) {
        Ok(since) => UNIX_EPOCH + Duration::from_micros(since.as_micros() as u64),
        Err(_) => now,
    }
}

fn version_text(value: ResourceVersionNumber) -> i64 {
    // A version number is a small counter; the cast cannot lose a real one.
    i64::try_from(value.get()).unwrap_or(i64::MAX)
}

fn size_bytes(value: u64) -> i64 {
    i64::try_from(value).unwrap_or(i64::MAX)
}

#[async_trait]
impl ControlPlaneStore for PostgresControlPlane {
    fn name(&self) -> &'static str {
        BACKEND
    }

    fn capabilities(&self) -> Capabilities {
        Capabilities::new(&[
            Capability::TransactionalWrites,
            Capability::OptimisticConcurrency,
            Capability::IdempotentWrites,
            Capability::TransactionalAudit,
        ])
    }

    fn status_probe_admission(&self) -> Option<StatusProbeAdmission> {
        let queued_operations = self.pending_operations.fetch_add(1, Ordering::AcqRel);
        Some(StatusProbeAdmission::new(
            self.settings.status_probe_timeout(queued_operations),
            Arc::clone(&self.pending_operations),
        ))
    }

    async fn health(&self) -> Result<(), ControlPlaneError> {
        self.health_with_status_probe(None).await
    }

    async fn health_with_status_probe(
        &self,
        admission: Option<StatusProbeAdmission>,
    ) -> Result<(), ControlPlaneError> {
        self.run(admission, |client| {
            Box::pin(async move {
                client
                    .query_one("SELECT 1", &[])
                    .await
                    .map(|_| ())
                    .map_err(|error| unavailable("health check", &error))
            })
        })
        .await
    }

    async fn desired_revision(&self) -> Result<Option<RevisionId>, ControlPlaneError> {
        self.run(None, |client| {
            Box::pin(async move {
                let row = client
                    .query_opt(
                        "SELECT revision_id FROM axond_cp_head WHERE singleton",
                        &[],
                    )
                    .await
                    .map_err(|error| unavailable("read desired revision", &error))?;
                let Some(row) = row else {
                    return Err(corrupt_storage(
                        "the control-plane head row is missing; the schema was modified out of band",
                    ));
                };
                let id: Option<String> = row.get(0);
                id.map(|text| {
                    rows::revision_id(&text).map_err(|error| {
                        corrupt_storage(format!("the desired revision is unreadable: {error}"))
                    })
                })
                .transpose()
            })
        })
        .await
    }

    async fn load_manifest(&self, id: RevisionId) -> Result<RevisionManifest, ControlPlaneError> {
        let limits = self.settings.hydration;
        self.run(None, move |client| {
            Box::pin(async move {
                let transaction = client
                    .transaction()
                    .await
                    .map_err(|error| unavailable("begin manifest read", &error))?;
                let manifest = hydration::manifest(&transaction, id, &limits).await;
                let _ = transaction.rollback().await;
                manifest
            })
        })
        .await
    }

    async fn load_revision(&self, id: RevisionId) -> Result<LoadedRevision, ControlPlaneError> {
        let limits = self.settings.hydration;
        self.run(None, move |client| {
            Box::pin(async move {
                let transaction = client
                    .transaction()
                    .await
                    .map_err(|error| unavailable("begin revision read", &error))?;
                let loaded = hydration::revision(&transaction, id, &limits).await;
                let _ = transaction.rollback().await;
                loaded
            })
        })
        .await
    }

    /// The head and its hydration in one read, so a caller cannot be told it
    /// converged onto a revision that was never the head it read.
    async fn load_desired_revision(&self) -> Result<Option<LoadedRevision>, ControlPlaneError> {
        let limits = self.settings.hydration;
        self.run(None, move |client| {
            Box::pin(async move {
                let transaction = client
                    .transaction()
                    .await
                    .map_err(|error| unavailable("begin desired-revision read", &error))?;
                let loaded = hydration::desired(&transaction, &limits).await;
                let _ = transaction.rollback().await;
                loaded
            })
        })
        .await
    }

    async fn publish_revision(
        &self,
        candidate: RevisionCandidate,
    ) -> Result<RevisionManifest, ControlPlaneError> {
        // Validation is domain work, and it happens before the store commits to
        // anything: a rejected candidate leaves no transaction, no row, and no
        // audit event.
        let checksum = candidate.validated_checksum()?;
        let caller_scope = rows::caller_scope(&candidate.mutation.actor)
            .map_err(|error| ControlPlaneError::Invalid(error.into()))?;
        let retention = self.settings.idempotency_retention;
        let limits = self.settings.hydration;
        let id = RevisionId::new(self.ids.next());

        self.run(None, move |client| {
            Box::pin(async move {
                let transaction = client
                    .transaction()
                    .await
                    .map_err(|error| unavailable("begin publication", &error))?;
                match publish(
                    &transaction,
                    id,
                    &candidate,
                    &checksum.to_string(),
                    &caller_scope,
                    retention,
                    &limits,
                )
                .await
                {
                    Ok(Published::Manifest(manifest)) => {
                        transaction
                            .commit()
                            .await
                            .map_err(|error| unavailable("commit publication", &error))?;
                        Ok(manifest)
                    }
                    Ok(Published::Replayed(manifest)) => {
                        // Nothing was written except the pruning of expired
                        // records, which is worth keeping.
                        transaction
                            .commit()
                            .await
                            .map_err(|error| unavailable("commit replay", &error))?;
                        Ok(manifest)
                    }
                    Err(error) => {
                        // Explicit, so "rollback leaves nothing behind" is a
                        // statement about this code and not about a `Drop`.
                        let _ = transaction.rollback().await;
                        Err(error)
                    }
                }
            })
        })
        .await
    }

    async fn audit_trail(&self, id: RevisionId) -> Result<Vec<AuditEvent>, ControlPlaneError> {
        self.run(None, move |client| {
            Box::pin(async move {
                let revision = client
                    .query_opt(
                        "SELECT 1 FROM axond_cp_revision WHERE revision_id = $1",
                        &[&id.to_string()],
                    )
                    .await
                    .map_err(|error| unavailable("read revision", &error))?;
                if revision.is_none() {
                    return Err(ControlPlaneError::RevisionNotFound(id));
                }
                let rows = client
                    .query(
                        "SELECT audit_event_id, mutation_id, actor_kind, actor_issuer, \
                         actor_subject, actor_component, event_kind, target_kind, target_id, \
                         target_version, summary, recorded_at, actor_tenant_id, \
                         actor_principal_id \
                         FROM axond_cp_audit_event WHERE revision_id = $1 \
                         ORDER BY recorded_at DESC, audit_event_id DESC",
                        &[&id.to_string()],
                    )
                    .await
                    .map_err(|error| unavailable("read audit trail", &error))?;
                rows.iter()
                    .map(|row| {
                        audit_event(row).map_err(|error| ControlPlaneError::corrupt(id, error))
                    })
                    .collect()
            })
        })
        .await
    }

    async fn record_denial(&self, denial: &AccessDenial) -> Result<(), ControlPlaneError> {
        let denial = denial.clone();
        self.run(None, move |client| {
            Box::pin(async move {
                let actor = rows::actor_columns(&denial.actor);
                let scope = rows::scope_columns(&denial.scope);
                client
                    .execute(
                        "INSERT INTO axond_cp_access_denial \
                         (denial_id, actor_kind, actor_issuer, actor_subject, actor_component, \
                         actor_tenant_id, actor_principal_id, surface, action, scope_kind, \
                         tenant_id, project_id, reason, recorded_at) \
                         VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) \
                         ON CONFLICT (denial_id) DO NOTHING",
                        &[
                            &denial.id.to_string(),
                            &actor.kind,
                            &actor.issuer,
                            &actor.subject,
                            &actor.component,
                            &actor.tenant,
                            &actor.principal,
                            &denial.surface.as_str(),
                            &denial.action.as_str(),
                            &scope.kind,
                            &scope.tenant,
                            &scope.project,
                            &denial.reason.as_str(),
                            &denial.recorded_at,
                        ],
                    )
                    .await
                    .map_err(|error| unavailable("record denied action", &error))?;
                Ok(())
            })
        })
        .await
    }

    async fn denials(
        &self,
        page: &DenialPage,
        limit: usize,
    ) -> Result<Vec<AccessDenial>, ControlPlaneError> {
        let tenant = page.tenant().map(|tenant| tenant.to_string());
        // Clamped rather than rejected: this is a page size, and an administrator
        // asking for everything gets the first page rather than an error.
        let limit = i64::try_from(limit.clamp(1, 1_000)).unwrap_or(1_000);
        self.run(None, move |client| {
            Box::pin(async move {
                // `IS NOT DISTINCT FROM` so `None` selects the deployment-scoped
                // denials rather than every row: a refusal that named no tenant is
                // a platform-scoped read, not a wildcard.
                //
                // Scope alone, and deliberately: this read is scope-*exact*, so
                // every row it returns is a refusal against the tenant that was
                // asked for, whoever attempted it. Filtering the actor here as
                // well would make a cross-tenant attempt — a refusal scoped to A
                // attempted by a workload of B — returnable by no page at all:
                // not A's, because of the actor; not B's, because of the scope;
                // and not the deployment page, which is only the rows that named
                // no tenant. That row is the one the trail exists for.
                //
                // Attribution is filtered one layer down, where it is the only
                // place it can leak: the row-level-security policy shares
                // *deployment-scoped* refusals with every pinned session, so it
                // withholds the ones another tenant's workload attempted. A read
                // that names a tenant is answering for that tenant, and a caller
                // that may ask about a tenant at all is a caller that may know
                // who was refused against it.
                let rows = client
                    .query(
                        "SELECT denial_id, actor_kind, actor_issuer, actor_subject, \
                         actor_component, actor_tenant_id, actor_principal_id, surface, action, \
                         scope_kind, tenant_id, project_id, reason, recorded_at \
                         FROM axond_cp_access_denial \
                         WHERE tenant_id IS NOT DISTINCT FROM $1 \
                         ORDER BY recorded_at DESC, denial_id DESC LIMIT $2",
                        &[&tenant, &limit],
                    )
                    .await
                    .map_err(|error| unavailable("read denied actions", &error))?;
                rows.iter()
                    .map(|row| {
                        access_denial(row).map_err(|error| {
                            corrupt_storage(format!("a denied action is unreadable: {error}"))
                        })
                    })
                    .collect()
            })
        })
        .await
    }
}

/// Discovery evidence, kept between restarts.
///
/// The same database as the journal and deliberately not the same kind of state:
/// a revision is desired state and is published, while an observation is
/// something a replica learned and is merely remembered. Nothing here is read on
/// the request path, nothing here is read while compiling a snapshot's routing
/// tables, and a total loss of these rows costs a deployment its freshness and
/// nothing else — every target falls back to `unknown`, which is what a replica
/// that has not looked honestly knows.
#[async_trait]
impl ObservationStore for PostgresControlPlane {
    async fn load(
        &self,
        scope: Option<ScopeRef>,
    ) -> Result<Vec<StoredObservation>, ControlPlaneError> {
        let tenant = scope.map(|scope| scope.tenant.to_string());
        let project = scope.and_then(|scope| scope.project.map(|project| project.to_string()));
        // A scope is matched exactly, project included: a tenant-wide read and a
        // project's read are different questions, and answering the narrower one
        // with the tenant's rows would report a project as knowing things it does
        // not.
        self.run(None, move |client| {
            Box::pin(async move {
                let rows = client
                    .query(
                        "SELECT tenant_id, project_id, provider, model, slot, result, \
                         completeness, source, observed_at, expires_at, definitive_at \
                         FROM axond_cp_availability_observation \
                         WHERE ($1::text IS NULL OR tenant_id = $1) \
                         AND ($1::text IS NULL OR project_id IS NOT DISTINCT FROM $2) \
                         ORDER BY tenant_id, project_id, provider, model, slot",
                        &[&tenant, &project],
                    )
                    .await
                    .map_err(|error| unavailable("read discovery observations", &error))?;
                rows.iter().map(observation_row).collect()
            })
        })
        .await
    }

    async fn save(&self, write: &EvidenceWrite) -> Result<(), ControlPlaneError> {
        if write.is_empty() {
            return Ok(());
        }
        let rows = write.rows().to_vec();
        let cleared = write.cleared().to_vec();
        self.run(None, move |client| {
            Box::pin(async move {
                let transaction = client
                    .transaction()
                    .await
                    .map_err(|error| unavailable("begin an observation write", &error))?;
                // A replica may be behind another replica. A key's replacement is
                // whole-record — a record whose retained look was discredited has
                // to stop having a stored one, which an upsert alone cannot say —
                // so the watermark is per key: the newest instant this write owns
                // for it, across its rows and its explicit clear.
                let mut replaced: BTreeMap<AvailabilityKey, SystemTime> = BTreeMap::new();
                for row in &rows {
                    replaced
                        .entry(row.key.clone())
                        .and_modify(|before| *before = (*before).max(row.observation.observed_at))
                        .or_insert(row.observation.observed_at);
                }
                for clear in &cleared {
                    replaced
                        .entry(clear.key.clone())
                        .and_modify(|before| *before = (*before).max(clear.before))
                        .or_insert(clear.before);
                }
                let mut behind: BTreeSet<AvailabilityKey> = BTreeSet::new();
                for (key, before) in &replaced {
                    let tenant = key.scope.tenant.to_string();
                    let project = key.scope.project.map(|project| project.to_string());
                    let provider = key.target.provider.as_str();
                    let model = key.target.model.as_str();
                    // Locked, then compared: the write is a replacement of every
                    // slot the key has, so a key the store knows more about than
                    // this replica does is left entirely alone rather than
                    // half-replaced. Deleting up to this watermark and inserting
                    // over it is not enough on its own — a fallback older than
                    // this write's current look would be deleted on behalf of a
                    // row that then loses the insert to a newer current row, and
                    // the deployment would be left holding less evidence than
                    // either replica has.
                    let held = transaction
                        .query(
                            "SELECT observed_at FROM axond_cp_availability_observation \
                             WHERE tenant_id = $1 AND project_id IS NOT DISTINCT FROM $2 \
                             AND provider = $3 AND model = $4 FOR UPDATE",
                            &[&tenant, &project, &provider, &model],
                        )
                        .await
                        .map_err(|error| {
                            observation_write_failure("read evidence being replaced", &error)
                        })?;
                    let newest = held
                        .iter()
                        .map(|row| row.get::<_, SystemTime>(0))
                        .max()
                        .unwrap_or(SystemTime::UNIX_EPOCH);
                    if newest > *before {
                        behind.insert(key.clone());
                        continue;
                    }
                    transaction
                        .execute(
                            "DELETE FROM axond_cp_availability_observation \
                             WHERE tenant_id = $1 AND project_id IS NOT DISTINCT FROM $2 \
                             AND provider = $3 AND model = $4 AND observed_at <= $5",
                            &[&tenant, &project, &provider, &model, before],
                        )
                        .await
                        .map_err(|error| {
                            observation_write_failure("replace discovery evidence", &error)
                        })?;
                }
                for row in rows.iter().filter(|row| !behind.contains(&row.key)) {
                    // `detail` is dropped at the boundary rather than filtered on
                    // read: a probe's error body is not evidence, and a column
                    // holding it would be readable from every backup.
                    transaction
                        .execute(
                            "INSERT INTO axond_cp_availability_observation \
                             (tenant_id, project_id, provider, model, slot, result, \
                             completeness, source, observed_at, expires_at, definitive_at) \
                             VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11) \
                             ON CONFLICT DO NOTHING",
                            &[
                                &row.key.scope.tenant.to_string(),
                                &row.key.scope.project.map(|project| project.to_string()),
                                &row.key.target.provider.as_str(),
                                &row.key.target.model.as_str(),
                                &row.slot.as_str(),
                                &row.observation.result.as_str(),
                                &row.observation.completeness.as_str(),
                                &row.observation.source.as_str(),
                                &row.observation.observed_at,
                                &row.observation.expires_at,
                                &row.definitive_at,
                            ],
                        )
                        .await
                        .map_err(|error| {
                            observation_write_failure("record discovery evidence", &error)
                        })?;
                }
                transaction
                    .commit()
                    .await
                    .map_err(|error| unavailable("commit discovery evidence", &error))?;
                Ok(())
            })
        })
        .await
    }
}

/// One stored look, or a refusal to interpret it.
///
/// Vocabulary a release never wrote is corrupt storage rather than a dropped
/// row: a newer replica's evidence read by an older build is intact state this
/// build cannot read, and silently reporting the target as unlooked-at would
/// hide a version skew behind an honest-looking `unknown`.
fn observation_row(row: &Row) -> Result<StoredObservation, ControlPlaneError> {
    let tenant: String = row.get(0);
    let project: Option<String> = row.get(1);
    let provider: String = row.get(2);
    let model: String = row.get(3);
    let slot: String = row.get(4);
    let result: String = row.get(5);
    let completeness: String = row.get(6);
    let source: String = row.get(7);

    let tenant = TenantId::parse(&tenant)
        .map_err(|error| corrupt_storage(format!("an observation names no tenant: {error}")))?;
    let project = project
        .map(|project| ProjectId::parse(&project))
        .transpose()
        .map_err(|error| corrupt_storage(format!("an observation names no project: {error}")))?;
    let target = TargetRef::parse(&provider, &model)
        .map_err(|error| corrupt_storage(format!("an observation names no target: {error}")))?;
    let scope = match project {
        None => ScopeRef::tenant(tenant),
        Some(project) => ScopeRef::project(tenant, project),
    };
    let key = AvailabilityKey::new(scope, target.clone());
    let slot = ObservationSlot::parse(&slot)
        .ok_or_else(|| corrupt_storage(format!("`{slot}` is not an observation slot")))?;
    let result = availability_store::parse_result(&result)
        .ok_or_else(|| corrupt_storage(format!("`{result}` is not a discovery result")))?;
    let completeness = availability_store::parse_completeness(&completeness).ok_or_else(|| {
        corrupt_storage(format!("`{completeness}` is not a discovery completeness"))
    })?;
    let source = availability_store::parse_source(&source)
        .ok_or_else(|| corrupt_storage(format!("`{source}` is not a discovery source")))?;

    let observation = DiscoveryObservation {
        scope,
        target,
        result,
        completeness,
        source,
        observed_at: row.get(8),
        expires_at: row.get(9),
        detail: None,
    };
    Ok(StoredObservation {
        key,
        slot,
        observation,
        definitive_at: row.get(10),
    })
}

/// Whether a publication wrote a revision or replayed one.
///
/// Named rather than a bare manifest so the commit path cannot accidentally treat
/// a replay as a write.
enum Published {
    Manifest(RevisionManifest),
    Replayed(RevisionManifest),
}

/// The whole publication, inside the caller's transaction.
///
/// The order is the contract's order, and each step is a refusal rather than a
/// repair: take the head lock, prune and consult the caller's retry window, check
/// the expectation, check version immutability, then write.
async fn publish(
    transaction: &Transaction<'_>,
    id: RevisionId,
    candidate: &RevisionCandidate,
    checksum: &str,
    caller_scope: &str,
    retention: Duration,
    limits: &HydrationLimits,
) -> Result<Published, ControlPlaneError> {
    // One row, taken `FOR UPDATE`: every publisher queues here, so "exactly one
    // expected-revision commit wins" does not depend on isolation-level
    // subtleties or on retrying a serialization failure.
    let head = transaction
        .query_opt(
            "SELECT revision_id FROM axond_cp_head WHERE singleton FOR UPDATE",
            &[],
        )
        .await
        .map_err(|error| unavailable("lock the head", &error))?
        .ok_or_else(|| {
            corrupt_storage("the control-plane head row is missing; publication has no anchor")
        })?;
    let head: Option<String> = head.get(0);
    let head = head
        .map(|text| {
            rows::revision_id(&text).map_err(|error| {
                corrupt_storage(format!("the desired revision is unreadable: {error}"))
            })
        })
        .transpose()?;

    // Expiry is a retry window closing, not a garbage-collection job an operator
    // has to run: the window is pruned on the write path that depends on it.
    transaction
        .execute(
            "DELETE FROM axond_cp_idempotency WHERE expires_at <= now()",
            &[],
        )
        .await
        .map_err(|error| unavailable("prune expired idempotency records", &error))?;

    let key = candidate.mutation.idempotency_key.as_str().to_owned();
    // Replay is consulted *before* the expectation: a retry of a candidate that
    // has since gone stale must replay its own outcome, not be told it conflicts
    // with the revision it published.
    if let Some(record) = transaction
        .query_opt(
            "SELECT state_checksum, revision_id FROM axond_cp_idempotency \
             WHERE caller_scope = $1 AND idempotency_key = $2",
            &[&caller_scope, &key],
        )
        .await
        .map_err(|error| unavailable("read idempotency record", &error))?
    {
        let recorded_checksum: String = record.get(0);
        let recorded_revision: String = record.get(1);
        let published = rows::revision_id(&recorded_revision).map_err(|error| {
            corrupt_storage(format!("an idempotency record is unreadable: {error}"))
        })?;
        if recorded_checksum != checksum {
            return Err(ControlPlaneError::IdempotencyKeyReused {
                key: candidate.mutation.idempotency_key.clone(),
                published,
            });
        }
        return hydration::manifest(transaction, published, limits)
            .await
            .map(Published::Replayed);
    }

    if !candidate.expected.matches(head) {
        return Err(ControlPlaneError::Conflict {
            expected: candidate.expected,
            actual: head,
        });
    }

    for resource in candidate.state.resources() {
        assert_version_is_immutable(transaction, resource).await?;
    }

    let manifest = RevisionManifest::of(id, head, journal_now(), candidate)?;

    for blob in candidate.state.blobs() {
        transaction
            .execute(
                "INSERT INTO axond_cp_blob (blob_kind, digest, size_bytes) VALUES ($1, $2, $3) \
                 ON CONFLICT (blob_kind, digest) DO NOTHING",
                &[
                    &blob.kind.as_str(),
                    &blob.digest.to_string(),
                    &size_bytes(blob.size_bytes),
                ],
            )
            .await
            .map_err(|error| unavailable("write blob reference", &error))?;
    }

    // Ownership and identity first, because the projection is where the database
    // constrains them: a project names its tenant and a principal names the
    // tenant and project it is scoped to, through foreign keys among these three
    // tables, so a cross-tenant grant is unwritable rather than merely
    // unvalidated. The journal rows below name their tenant under a key too, since
    // 0004 — which is why the projection runs first, and why it records an owner
    // for a tenant only history names (see [`record_referenced_tenants`]): the key
    // is satisfied for a rolled-back pre-tenancy revision instead of refusing it.
    // A journal row's *project* is still checked in the domain alone; 0004 says
    // why. Same transaction either way, so "projected" and "published" cannot come
    // apart.
    project_tenancy(transaction, id, candidate).await?;

    for resource in candidate.state.resources() {
        insert_resource_version(transaction, resource).await?;
    }

    insert_mutation(transaction, &candidate.mutation).await?;

    transaction
        .execute(
            "INSERT INTO axond_cp_revision \
             (revision_id, parent_id, mutation_id, serializer, state_checksum, created_at) \
             VALUES ($1, $2, $3, $4, $5, $6)",
            &[
                &id.to_string(),
                &manifest.parent.map(|parent| parent.to_string()),
                &manifest.mutation.to_string(),
                &manifest.serializer.as_str(),
                &checksum,
                &manifest.created_at,
            ],
        )
        .await
        .map_err(|error| {
            // The head lock makes this unreachable through this code. It stays
            // mapped as a conflict because the constraint it violates — one child
            // per parent — means exactly that: someone else already published
            // against this parent.
            if is_unique_violation(&error) {
                ControlPlaneError::Conflict {
                    expected: candidate.expected,
                    actual: head,
                }
            } else {
                unavailable("write revision", &error)
            }
        })?;

    for entry in &manifest.entries {
        transaction
            .execute(
                "INSERT INTO axond_cp_revision_entry \
                 (revision_id, resource_kind, resource_id, version) VALUES ($1, $2, $3, $4)",
                &[
                    &id.to_string(),
                    &entry.reference.kind.as_str(),
                    &entry.reference.id.to_string(),
                    &version_text(entry.reference.version),
                ],
            )
            .await
            .map_err(|error| unavailable("write manifest entry", &error))?;
    }

    for blob in &manifest.blobs {
        transaction
            .execute(
                "INSERT INTO axond_cp_revision_blob (revision_id, blob_kind, digest) \
                 VALUES ($1, $2, $3)",
                &[
                    &id.to_string(),
                    &blob.kind.as_str(),
                    &blob.digest.to_string(),
                ],
            )
            .await
            .map_err(|error| unavailable("write revision blob", &error))?;
    }

    insert_audit_event(transaction, id, &candidate.audit).await?;

    let expires_at = journal_now() + retention;
    let replaced = transaction
        .execute(
            "INSERT INTO axond_cp_idempotency \
             (caller_scope, idempotency_key, state_checksum, revision_id, mutation_id, expires_at) \
             VALUES ($1, $2, $3, $4, $5, $6) \
             ON CONFLICT (caller_scope, idempotency_key) DO UPDATE SET \
             state_checksum = EXCLUDED.state_checksum, revision_id = EXCLUDED.revision_id, \
             mutation_id = EXCLUDED.mutation_id, recorded_at = now(), \
             expires_at = EXCLUDED.expires_at",
            &[
                &caller_scope,
                &key,
                &checksum,
                &id.to_string(),
                &candidate.mutation.id.to_string(),
                &expires_at,
            ],
        )
        .await
        .map_err(|error| unavailable("write idempotency record", &error))?;
    if replaced != 1 {
        return Err(corrupt_storage(
            "the idempotency record for this caller and key was neither written nor replayed",
        ));
    }

    transaction
        .execute(
            "UPDATE axond_cp_head SET revision_id = $1, updated_at = now() WHERE singleton",
            &[&id.to_string()],
        )
        .await
        .map_err(|error| unavailable("advance the head", &error))?;

    Ok(Published::Manifest(manifest))
}

/// Project the revision's tenancy and identity directory into the tables the
/// database's own constraints are written against.
///
/// Why project at all, when the revision already holds it: canonical bytes are
/// opaque to SQL, so a foreign key cannot point into them and a row-level
/// security policy cannot read them. Ownership and identity are the two things an
/// operator has to be able to state about the database rather than about the
/// service in front of it, so they exist twice — as the published truth, and as
/// rows that constrain it.
///
/// Tenants and projects are upserted and never deleted. Retained rows are what
/// history references: a resource version published last year names its project,
/// and a deletion here would either break that foreign key or orphan the audit
/// trail that proves what was billed. A tenant that goes away goes to
/// `lifecycle = 'deleted'`, which is a transition, not an erasure — physical
/// erasure is separate compliance work with its own retention argument.
///
/// A tenant the revision *omits* takes the same transition, because a revision is
/// the whole desired state and the domain already treats an undeclared tenant as
/// neither served nor administrable ([`Tenancy::is_served`] is `false` for a
/// tenant it has no row for). Leaving the retained row at whatever it last said
/// would leave `lifecycle` reading `active` for a tenant nothing serves, and that
/// column exists precisely so "is this tenant served?" is answerable without
/// decoding a body. It also settles the name: the domain frees an undeclared
/// tenant's slug, and the exclusion constraint that enforces slug uniqueness
/// ignores deleted rows, so the two agree about what may be reused. The
/// reconciliation runs first, so a revision that reuses a dropped tenant's name
/// does not collide with the row it is replacing.
///
/// That reconciliation needs a snapshot that *speaks about* tenancy, and one that
/// declares no tenant at all does not. Two real revisions look like that and mean
/// the opposite of "retire everything": a pre-tenancy revision, written by a build
/// that had no tenant resource to declare, and a rollback to one — the deployment
/// this slice upgrades has a whole journal of them, and #142's replica compiles
/// them into routing state today. Reading their silence as a deletion of every
/// tenant would make a rollback the most destructive operation the control plane
/// has, and would do it without any revision ever having said so.
///
/// So the retirement is conditional on the snapshot being *authoritative about
/// tenancy*: at least one tenant declared. Retiring the last tenant is then an
/// explicit `lifecycle = 'deleted'` on that tenant rather than an empty
/// publication, which is the transition the vocabulary already has and the only
/// one an audit trail can attribute. The cost is stated rather than hidden: a
/// deployment cannot empty its tenant list by omission, and the runbook says so.
///
/// Principals *are* deleted when a revision stops declaring them, because a
/// revoked administrator whose grants linger is the whole failure this table
/// exists to make visible. Their history is unaffected: attribution is copied
/// onto the mutation and the audit event, never joined from here.
async fn project_tenancy(
    transaction: &Transaction<'_>,
    revision: RevisionId,
    candidate: &RevisionCandidate,
) -> Result<(), ControlPlaneError> {
    let tenancy =
        Tenancy::of(&candidate.state).map_err(|error| ControlPlaneError::Invalid(error.into()))?;
    let directory = Directory::of(&candidate.state, &tenancy)
        .map_err(|error| ControlPlaneError::Invalid(error.into()))?;
    let revision = revision.to_string();

    let declared_tenants: Vec<String> = tenancy
        .tenants()
        .map(|tenant| tenant.body.tenant().to_string())
        .collect();
    // Only a snapshot that declares tenancy reconciles it: see above — silence is
    // a pre-tenancy or rolled-back revision, not a deletion of every tenant.
    if !declared_tenants.is_empty() {
        transaction
            .execute(
                "UPDATE axond_cp_tenant SET lifecycle = 'deleted', revision_id = $2, \
                 updated_at = now() \
                 WHERE NOT (tenant_id = ANY($1)) AND lifecycle <> 'deleted'",
                &[&declared_tenants, &revision],
            )
            .await
            .map_err(|error| unavailable("retire undeclared tenants", &error))?;
    }

    for tenant in tenancy.tenants() {
        transaction
            .execute(
                "INSERT INTO axond_cp_tenant (tenant_id, slug, lifecycle, revision_id) \
                 VALUES ($1, $2, $3, $4) \
                 ON CONFLICT (tenant_id) DO UPDATE SET slug = EXCLUDED.slug, \
                 lifecycle = EXCLUDED.lifecycle, revision_id = EXCLUDED.revision_id, \
                 updated_at = now()",
                &[
                    &tenant.body.tenant().to_string(),
                    &tenant.slug.as_str(),
                    &tenant.body.lifecycle().as_str(),
                    &revision,
                ],
            )
            .await
            .map_err(|error| {
                projection_failure("project tenant", "tenant", tenant.slug.as_str(), &error)
            })?;
    }

    for project in tenancy.projects() {
        transaction
            .execute(
                "INSERT INTO axond_cp_project (project_id, tenant_id, slug, revision_id) \
                 VALUES ($1, $2, $3, $4) \
                 ON CONFLICT (project_id) DO UPDATE SET tenant_id = EXCLUDED.tenant_id, \
                 slug = EXCLUDED.slug, revision_id = EXCLUDED.revision_id, updated_at = now()",
                &[
                    &project.body.project().to_string(),
                    &project.body.tenant().to_string(),
                    &project.slug.as_str(),
                    &revision,
                ],
            )
            .await
            .map_err(|error| {
                projection_failure("project project", "project", project.slug.as_str(), &error)
            })?;
    }

    let declared: Vec<String> = directory
        .principals()
        .map(|principal| principal.body.principal().to_string())
        .collect();
    transaction
        .execute(
            "DELETE FROM axond_cp_principal WHERE NOT (principal_id = ANY($1))",
            &[&declared],
        )
        .await
        .map_err(|error| unavailable("revoke undeclared principals", &error))?;

    for principal in directory.principals() {
        let scope = rows::scope_columns(&principal.scope);
        let (issuer, subject, digest) = match principal.body.credential() {
            Credential::Oidc { issuer, subject } => {
                (Some(issuer.clone()), Some(subject.clone()), None)
            }
            // The digest, never the key: a minted key is displayed once, at
            // creation, and a database read cannot recover it.
            Credential::MintedKey { digest } => {
                (None, None, digest.map(|digest| digest.to_string()))
            }
        };
        transaction
            .execute(
                "INSERT INTO axond_cp_principal \
                 (principal_id, resource_id, identity_kind, scope_kind, tenant_id, project_id, \
                 slug, display_name, issuer, subject, key_digest, revision_id) \
                 VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) \
                 ON CONFLICT (principal_id) DO UPDATE SET \
                 resource_id = EXCLUDED.resource_id, identity_kind = EXCLUDED.identity_kind, \
                 scope_kind = EXCLUDED.scope_kind, tenant_id = EXCLUDED.tenant_id, \
                 project_id = EXCLUDED.project_id, slug = EXCLUDED.slug, \
                 display_name = EXCLUDED.display_name, issuer = EXCLUDED.issuer, \
                 subject = EXCLUDED.subject, key_digest = EXCLUDED.key_digest, \
                 revision_id = EXCLUDED.revision_id, updated_at = now()",
                &[
                    &principal.body.principal().to_string(),
                    &principal.reference.id.to_string(),
                    &principal.body.kind().as_str(),
                    &scope.kind,
                    &scope.tenant,
                    &scope.project,
                    &principal.slug.as_str(),
                    &principal.body.display_name().as_str(),
                    &issuer,
                    &subject,
                    &digest,
                    &revision,
                ],
            )
            .await
            .map_err(|error| {
                projection_failure(
                    "project principal",
                    "principal",
                    principal.slug.as_str(),
                    &error,
                )
            })?;

        // Grants are replaced, not merged: a revision that dropped a role has
        // revoked it, and an `INSERT ... ON CONFLICT DO NOTHING` would leave the
        // dropped row behind as an authority nobody granted.
        transaction
            .execute(
                "DELETE FROM axond_cp_principal_role WHERE principal_id = $1",
                &[&principal.body.principal().to_string()],
            )
            .await
            .map_err(|error| unavailable("revoke roles", &error))?;
        for role in principal.body.roles() {
            transaction
                .execute(
                    "INSERT INTO axond_cp_principal_role (principal_id, role) VALUES ($1, $2)",
                    &[&principal.body.principal().to_string(), &role.as_str()],
                )
                .await
                .map_err(|error| unavailable("grant role", &error))?;
        }
    }

    // The name and identity constraints are deferred, so that a revision in which
    // two tenants, two projects, or two principals trade what identifies them is
    // judged on the state it declares rather than on the order its rows were
    // written in. Settling them here rather than leaving them to commit is what
    // keeps a violation attributable: at commit the error would arrive with no
    // operation to name, and after the publication had otherwise succeeded.
    transaction
        .execute("SET CONSTRAINTS ALL IMMEDIATE", &[])
        .await
        .map_err(|error| projection_failure("settle names and identities", "name", "", &error))?;

    // Last, and after that settle on purpose. This step records owner rows for
    // tenants only history names, and every ownership key above is deferred: doing
    // it earlier would put a row there for a tenant an *identity* names and the
    // revision does not declare, so the principal key would find the owner this
    // same publication had just written for it. Settled first, then recorded, so a
    // publication cannot supply its own principals' ownership.
    record_referenced_tenants(transaction, &revision, candidate).await?;
    Ok(())
}

/// Record an owner row for every tenant this revision's journal rows name and its
/// tenancy does not declare.
///
/// Schema 0004 points foreign keys from the journal tables at `axond_cp_tenant`,
/// so "a stored row cannot name a tenant this deployment has no row for" is a
/// database fact rather than a service-layer convention. The keys are `NOT VALID`,
/// which exempts rows written before they existed and checks every row written
/// after — including the rows a *rollback* writes. A revision published by a
/// pre-tenancy build declares no tenant and still carries tenant-scoped resources,
/// and #142's replica compiles those revisions into routing state today, so
/// republishing one has to keep working.
///
/// It keeps working because the owner is recorded here, at `lifecycle =
/// 'deleted'`: the vocabulary's own word for a tenant that is neither served nor
/// administrable, which is exactly what an undeclared tenant is
/// ([`Tenancy::is_served`] is `false` for one, and so is administrability). The
/// row says "referenced by history, declared by nothing" — it is not a tenant
/// anybody granted, it holds no name against the exclusion constraint, which
/// ignores deleted rows, and it cannot be administered into one without a revision
/// declaring it.
///
/// `ON CONFLICT DO NOTHING` is what keeps that safe for tenants that *are*
/// declared, here or by an earlier revision: an existing row is never rewritten by
/// this step, so a rollback to a pre-tenancy revision still retires nothing and a
/// live tenant is never demoted by history mentioning it.
///
/// Called last, once the deferred ownership keys have been settled, so that a
/// publication cannot manufacture the owner row its *own* principals need: a
/// principal scoped into a tenant this revision does not declare is refused by the
/// principal key, not only by [`Directory::of`].
///
/// What that key states, exactly, is the tenant has a *row* — declared, retained,
/// or recorded here by an earlier revision — not that the current revision declares
/// it. A tenant retired to `lifecycle = 'deleted'` already satisfied it before
/// 0004, since retirement keeps the row; recording history's owners widens the set
/// of such rows rather than changing what the key says. "The current revision
/// declares this principal's tenant" is a property of one revision, which is
/// [`Directory::of`]'s to check.
///
/// Which is why the set recorded is the *state's* resource scopes and nothing else.
/// A resource is history — a pre-tenancy revision carries tenant-scoped resources
/// and declares no tenant, and republishing it has to keep working. The mutation
/// scope and the actor attribution travelling with a publication are not history:
/// they describe a change being made now, so a mutation or an actor naming a tenant
/// this deployment has no row for is refused by 0004's keys rather than handed one,
/// which keeps a publication from widening the set of rows the principal key accepts
/// by way of its own attribution.
async fn record_referenced_tenants(
    transaction: &Transaction<'_>,
    revision: &str,
    candidate: &RevisionCandidate,
) -> Result<(), ControlPlaneError> {
    let mut referenced: Vec<String> = candidate
        .state
        .resources()
        .filter_map(|resource| rows::scope_columns(&resource.scope).tenant)
        .collect();
    referenced.sort_unstable();
    referenced.dedup();
    if referenced.is_empty() {
        return Ok(());
    }
    transaction
        .execute(
            "INSERT INTO axond_cp_tenant (tenant_id, slug, lifecycle, revision_id) \
             SELECT referenced, referenced, 'deleted', $2 \
             FROM unnest($1::text[]) AS referenced \
             ON CONFLICT (tenant_id) DO NOTHING",
            &[&referenced, &revision],
        )
        .await
        .map_err(|error| projection_failure("record referenced tenants", "tenant", "", &error))?;
    Ok(())
}

/// Refuse a republication that would redefine a version an earlier revision pins.
async fn assert_version_is_immutable(
    transaction: &Transaction<'_>,
    resource: &ResourceVersion,
) -> Result<(), ControlPlaneError> {
    let stored = transaction
        .query_opt(
            "SELECT content_checksum FROM axond_cp_resource_version \
             WHERE resource_kind = $1 AND resource_id = $2 AND version = $3",
            &[
                &resource.reference.kind.as_str(),
                &resource.reference.id.to_string(),
                &version_text(resource.reference.version),
            ],
        )
        .await
        .map_err(|error| unavailable("read resource version", &error))?;
    let Some(stored) = stored else {
        return Ok(());
    };
    let stored: String = stored.get(0);
    let candidate = resource
        .content_checksum()
        .map_err(|error| ControlPlaneError::Invalid(error.into()))?;
    if stored != candidate.to_string() {
        return Err(ControlPlaneError::ImmutableResourceVersion {
            reference: resource.reference,
        });
    }
    Ok(())
}

async fn insert_resource_version(
    transaction: &Transaction<'_>,
    resource: &ResourceVersion,
) -> Result<(), ControlPlaneError> {
    let scope = rows::scope_columns(&resource.scope);
    let body = rows::body_columns(&resource.body).map_err(|error| {
        corrupt_storage(format!(
            "{} could not be encoded for storage: {error}",
            resource.reference
        ))
    })?;
    let checksum = resource
        .content_checksum()
        .map_err(|error| ControlPlaneError::Invalid(error.into()))?;
    let kind = resource.reference.kind.as_str();
    let id = resource.reference.id.to_string();
    let version = version_text(resource.reference.version);
    let slug = resource.slug.as_str();
    let checksum = checksum.to_string();
    let serializer = SerializerVersion::default().as_str();
    let parameters: [&(dyn ToSql + Sync); 13] = [
        &kind,
        &id,
        &version,
        &scope.kind,
        &scope.tenant,
        &scope.project,
        &slug,
        &body.form,
        &body.inline,
        &body.blob_kind,
        &body.blob_digest,
        &checksum,
        &serializer,
    ];
    transaction
        .execute(
            "INSERT INTO axond_cp_resource_version \
             (resource_kind, resource_id, version, scope_kind, tenant_id, project_id, slug, \
             body_form, body_inline, body_blob_kind, body_blob_digest, content_checksum, \
             serializer) \
             VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13) \
             ON CONFLICT (resource_kind, resource_id, version) DO NOTHING",
            &parameters,
        )
        .await
        .map_err(|error| journal_write_failure("write resource version", &error))?;

    for dependency in &resource.depends_on {
        transaction
            .execute(
                "INSERT INTO axond_cp_resource_dependency \
                 (resource_kind, resource_id, version, depends_on_kind, depends_on_id, \
                 depends_on_version) VALUES ($1, $2, $3, $4, $5, $6) \
                 ON CONFLICT DO NOTHING",
                &[
                    &kind,
                    &id,
                    &version,
                    &dependency.kind.as_str(),
                    &dependency.id.to_string(),
                    &version_text(dependency.version),
                ],
            )
            .await
            .map_err(|error| unavailable("write resource dependency", &error))?;
    }
    Ok(())
}

async fn insert_mutation(
    transaction: &Transaction<'_>,
    mutation: &Mutation,
) -> Result<(), ControlPlaneError> {
    let actor = rows::actor_columns(&mutation.actor);
    let scope = rows::scope_columns(&mutation.scope);
    transaction
        .execute(
            "INSERT INTO axond_cp_mutation \
             (mutation_id, actor_kind, actor_issuer, actor_subject, actor_component, \
             mutation_kind, scope_kind, tenant_id, project_id, idempotency_key, submitted_at, \
             actor_tenant_id, actor_principal_id) \
             VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13)",
            &[
                &mutation.id.to_string(),
                &actor.kind,
                &actor.issuer,
                &actor.subject,
                &actor.component,
                &mutation.kind.as_str(),
                &scope.kind,
                &scope.tenant,
                &scope.project,
                &mutation.idempotency_key.as_str(),
                &mutation.submitted_at,
                &actor.tenant,
                &actor.principal,
            ],
        )
        .await
        .map_err(|error| {
            if is_unique_violation(&error) {
                // A mutation id is one administrative change. Reusing one for a
                // second change is refused rather than merged into the first:
                // the audit trail's "which change was this?" must have one
                // answer.
                denied(format!(
                    "mutation {} is already recorded; a mutation id names one change",
                    mutation.id
                ))
            } else {
                journal_write_failure("write mutation", &error)
            }
        })?;
    Ok(())
}

async fn insert_audit_event(
    transaction: &Transaction<'_>,
    revision: RevisionId,
    audit: &AuditEvent,
) -> Result<(), ControlPlaneError> {
    let actor = rows::actor_columns(&audit.actor);
    let target = audit.target;
    transaction
        .execute(
            "INSERT INTO axond_cp_audit_event \
             (audit_event_id, revision_id, mutation_id, actor_kind, actor_issuer, actor_subject, \
             actor_component, event_kind, target_kind, target_id, target_version, summary, \
             recorded_at, actor_tenant_id, actor_principal_id) \
             VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)",
            &[
                &audit.id.to_string(),
                &revision.to_string(),
                &audit.mutation.to_string(),
                &actor.kind,
                &actor.issuer,
                &actor.subject,
                &actor.component,
                &audit.kind.as_str(),
                &target.map(|target| target.kind.as_str()),
                &target.map(|target| target.id.to_string()),
                &target.map(|target| version_text(target.version)),
                &audit.summary,
                &audit.recorded_at,
                &actor.tenant,
                &actor.principal,
            ],
        )
        .await
        .map_err(|error| {
            if is_unique_violation(&error) {
                denied(format!(
                    "audit event {} is already recorded; an audit event is written once",
                    audit.id
                ))
            } else {
                journal_write_failure("write audit event", &error)
            }
        })?;
    Ok(())
}

/// Classify a journal write, so an ownership key it cannot satisfy is not reported
/// as weather either.
///
/// 0004 gave the journal tables foreign keys, which makes a violation there
/// possible for the first time — and permanent. A binary that predates
/// [`record_referenced_tenants`] publishing against a 0004 schema is how it
/// happens: its revision names a tenant it does not declare, no owner row is
/// recorded, and the key refuses the row. Reported as an outage that is a caller
/// retrying forever against a state no retry reaches; reported as a refusal it says
/// which rule refused it, and the remedy — a binary that records the owner, or a
/// revision that declares the tenant — is an operator decision rather than a wait.
fn journal_write_failure(operation: &str, error: &tokio_postgres::Error) -> ControlPlaneError {
    let Some(db) = error
        .as_db_error()
        .filter(|db| *db.code() == SqlState::FOREIGN_KEY_VIOLATION)
    else {
        return unavailable(operation, error);
    };
    // The constraint, and nothing else: PostgreSQL's detail quotes the key it
    // refused, which on these tables is a tenant identifier.
    tracing::warn!(
        constraint = db.constraint().unwrap_or("unnamed"),
        detail = db.detail().unwrap_or(""),
        message = db.message(),
        "a journal row named a tenant this deployment has no row for"
    );
    denied(journal_ownership_refusal(db.constraint()))
}

/// The text of that refusal, from the constraint name alone.
///
/// A separate function because it is what a test can assert cannot disclose: it
/// cannot see the message or the detail, so no future edit can splice the refused
/// tenant identifier into a caller-visible error.
fn journal_ownership_refusal(constraint: Option<&str>) -> String {
    match constraint {
        Some(constraint) => format!(
            "the journal row names a tenant this deployment has no row for \
             (constraint `{constraint}`); publish a revision that declares it, or \
             upgrade the publisher that records the owners history names"
        ),
        None => "the journal row names a tenant this deployment has no row for".to_owned(),
    }
}

/// Classify a projection write, so a constraint the projection cannot satisfy is
/// not reported as weather.
///
/// A retained row is why this matters. The projection keeps tenants and projects
/// a later revision stopped declaring, because history references them, and an
/// undeleted one keeps its name with it — so a candidate that re-uses a name a
/// live row holds violates a unique index. As an outage that is a caller
/// retrying forever against a state no retry reaches; as
/// [`ControlPlaneError::NameTaken`] it is a `409` naming the name to change,
/// which is the only thing the caller can act on. A deleted tenant releases its
/// slug — the partial index says so — so this is never how a deletion is
/// undone; restoring one is republishing that tenant as `active`, and it fails
/// here only when some *other* live tenant took the name meanwhile.
///
/// Ownership is the second condition, and reaches this the same way. A revision
/// that moves a project to another tenant is a state the domain accepts — it
/// checks ownership within one revision, not across two — and the composite
/// `(tenant_id, project_id)` foreign key held by any principal scoped into that
/// project refuses the update, with no `ON UPDATE` action to cascade it. That
/// refusal is as permanent as a taken name: the fix is a revision that moves the
/// principals too, not a retry.
fn projection_failure(
    operation: &str,
    noun: &'static str,
    slug: &str,
    error: &tokio_postgres::Error,
) -> ControlPlaneError {
    let Some(db) = error.as_db_error().filter(|db| is_projection_refusal(db)) else {
        return unavailable(operation, error);
    };
    // Which constraint was violated, rather than an assumption about which one it
    // was: a projected row can collide over a name, an OIDC subject, or a key
    // digest, or contradict an ownership row, and reporting all of them as a name
    // clash describes the wrong conflict and the wrong remedy. Only a name has a
    // remedy the caller can act on, so only a name keeps the typed refusal; the
    // rest name the constraint they contradicted.
    if is_name_conflict(db) {
        let taken = if slug.is_empty() {
            colliding_value(db).unwrap_or_else(|| slug.to_owned())
        } else {
            slug.to_owned()
        };
        return ControlPlaneError::NameTaken {
            noun,
            name: taken,
            holder: db.constraint().map(ToOwned::to_owned),
        };
    }
    // The constraint, and nothing the row it collided with holds. PostgreSQL's
    // message and detail name the colliding key — `Key (issuer, subject)=(…)`,
    // `Key (key_digest)=(sha256:…)` — and these constraints are the identity
    // directory's, so splicing them into the refusal would tell a publisher
    // another principal's sign-in or the digest of another workload's minted key.
    // A refusal names the rule it broke; the values stay in the operator's log,
    // which is the same line the wrong constraint name would be diagnosed from.
    tracing::warn!(
        constraint = db.constraint().unwrap_or("unnamed"),
        detail = db.detail().unwrap_or(""),
        message = db.message(),
        "a projected row was refused by state this deployment already holds"
    );
    denied(projection_refusal(
        noun,
        slug,
        *db.code() == SqlState::FOREIGN_KEY_VIOLATION,
        db.constraint(),
    ))
}

/// The caller-facing text of a projection refusal, built from the rule that
/// refused it and the revision's own slug — and from nothing else, which is the
/// point: the values in the row it collided with are not parameters here, so they
/// cannot reach a response by an edit that forgets why.
fn projection_refusal(noun: &str, slug: &str, ownership: bool, constraint: Option<&str>) -> String {
    let subject = if slug.is_empty() {
        format!("projecting the {noun}s of this revision")
    } else {
        format!("projecting the {noun} `{slug}`")
    };
    let conflict = if ownership {
        "contradicts an ownership row this deployment already holds"
    } else {
        "collides with a row this deployment already holds"
    };
    format!(
        "{subject} {conflict}{}; the conflicting row stands until a revision changes it, so \
         no retry clears this",
        constraint.map_or_else(String::new, |name| format!(" ({name})")),
    )
}

fn is_unique_violation(error: &tokio_postgres::Error) -> bool {
    error
        .as_db_error()
        .is_some_and(|db| db.code() == &SqlState::UNIQUE_VIOLATION)
}

/// Whether a projection failure is a state this deployment already holds refusing
/// the revision: a duplicate (a unique index, or the deferred exclusion
/// constraints that carry names and identities), or a contradicted ownership row
/// (the composite foreign keys). None of the three is transient, and reporting
/// one as an outage asks a caller to retry against a state no retry reaches.
fn is_projection_refusal(db: &tokio_postgres::error::DbError) -> bool {
    matches!(
        *db.code(),
        SqlState::UNIQUE_VIOLATION
            | SqlState::EXCLUSION_VIOLATION
            | SqlState::FOREIGN_KEY_VIOLATION
    )
}

/// Whether the constraint a projection violated is one of the two that carry a
/// *name*, as opposed to an identity or an ownership row. Read off the constraint
/// rather than off the operation, because the deferred constraints are settled for
/// the revision as a whole and the row that lost is only named by the error.
fn is_name_conflict(db: &tokio_postgres::error::DbError) -> bool {
    db.constraint()
        .is_some_and(|name| name.ends_with("_slug_unique"))
}

/// The name that collided, out of the detail Postgres attaches to a duplicate:
/// `Key (tenant_id, slug)=(ten_…, core) already exists`. Only the last column of
/// the key is a name — the others scope it — so this is the value a caller
/// changes.
fn colliding_value(db: &tokio_postgres::error::DbError) -> Option<String> {
    let detail = db.detail()?;
    let values = detail.split_once(")=(")?.1;
    let values = values.split_once(')')?.0;
    values
        .rsplit(", ")
        .next()
        .map(|value| value.trim().to_owned())
        .filter(|value| !value.is_empty())
}

fn audit_event(row: &Row) -> Result<AuditEvent, IntegrityError> {
    let id: String = row.get(0);
    let mutation: String = row.get(1);
    let actor_kind: String = row.get(2);
    let issuer: Option<String> = row.get(3);
    let subject: Option<String> = row.get(4);
    let component: Option<String> = row.get(5);
    let event_kind: String = row.get(6);
    let target_kind: Option<String> = row.get(7);
    let target_id: Option<String> = row.get(8);
    let target_version: Option<i64> = row.get(9);
    let summary: String = row.get(10);
    let recorded_at: SystemTime = row.get(11);
    let actor_tenant: Option<String> = row.get(12);
    let actor_principal: Option<String> = row.get(13);
    let target = match (target_kind, target_id, target_version) {
        (None, None, None) => None,
        (Some(kind), Some(id), Some(version)) => Some(ResourceRef::new(
            rows::resource_kind(&kind)?,
            rows::resource_id(&id)?,
            rows::version_number(version)?,
        )),
        _ => {
            return Err(rows::unreadable(
                "an audit event's target is half a reference",
            ));
        }
    };
    Ok(AuditEvent {
        id: rows::audit_event_id(&id)?,
        mutation: rows::mutation_id(&mutation)?,
        actor: rows::actor(
            &actor_kind,
            issuer.as_deref(),
            subject.as_deref(),
            component.as_deref(),
            actor_tenant.as_deref(),
            actor_principal.as_deref(),
        )?,
        kind: rows::mutation_kind(&event_kind)?,
        target,
        summary,
        recorded_at,
    })
}

fn access_denial(row: &Row) -> Result<AccessDenial, IntegrityError> {
    let id: String = row.get(0);
    let actor_kind: String = row.get(1);
    let issuer: Option<String> = row.get(2);
    let subject: Option<String> = row.get(3);
    let component: Option<String> = row.get(4);
    let actor_tenant: Option<String> = row.get(5);
    let actor_principal: Option<String> = row.get(6);
    let surface: String = row.get(7);
    let action: String = row.get(8);
    let scope_kind: String = row.get(9);
    let tenant: Option<String> = row.get(10);
    let project: Option<String> = row.get(11);
    let reason: String = row.get(12);
    let recorded_at: SystemTime = row.get(13);
    Ok(AccessDenial {
        id: rows::audit_event_id(&id)?,
        actor: rows::actor(
            &actor_kind,
            issuer.as_deref(),
            subject.as_deref(),
            component.as_deref(),
            actor_tenant.as_deref(),
            actor_principal.as_deref(),
        )?,
        surface: Surface::parse(&surface).ok_or_else(|| {
            rows::unreadable(format!("`{surface}` is not an administrative surface"))
        })?,
        action: Action::parse(&action)
            .ok_or_else(|| rows::unreadable(format!("`{action}` is not an action")))?,
        scope: rows::scope(&scope_kind, tenant.as_deref(), project.as_deref())?,
        reason: DenialReason::parse(&reason)
            .ok_or_else(|| rows::unreadable(format!("`{reason}` is not a denial reason")))?,
        recorded_at,
    })
}

#[cfg(test)]
mod tests {
    use std::time::UNIX_EPOCH;

    use super::super::hydration::HydrationLimit;
    use super::*;
    use crate::availability::discovery::{DiscoveryCompleteness, DiscoveryResult, DiscoverySource};
    use crate::availability::{AvailabilityIndex, AvailabilityRecord};
    use crate::backends::{BackendFailure, FailureCategory};
    use crate::desired_state::fixtures::{
        DESIRED_STATE_RESOURCES, candidate, human, principal_id, project, project_alias,
        project_body, project_id, reference, state, state_a_pre_tenancy_build_published,
        state_with_directory, state_with_renamed_alias, state_with_revoked_workload,
        state_with_second_tenant, state_with_two_blobs, tenant, tenant_body, tenant_id,
        two_tenant_directory_state, workload_key,
    };
    use crate::desired_state::{
        Actor, AuditEventId, Checksum, DesiredState, DisplayName, ExpectedRevision, IdentityBody,
        MutationId, PrincipalId, ResourceKind, ResourceScope, ResourceVersionNumber, Role, Slug,
        TenantLifecycle, Uuid7, ValidationError, WorkloadKey, oracle::InMemoryControlPlane,
    };

    /// Each test owns a schema, so the journal's fixed object names do not make
    /// two tests one test.
    async fn journal() -> Option<(PostgresControlPlane, String, String)> {
        let dsn = crate::test_services::postgres_dsn()?;
        let schema = format!(
            "cp_{}",
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .expect("clock")
                .as_nanos()
        );
        let mut config: Config = dsn.parse().expect("test dsn");
        config.connect_timeout(Duration::from_secs(5));
        let (client, connection) = config
            .connect(crate::usage::tls_connector())
            .await
            .expect("connect to create the test schema");
        tokio::spawn(async move {
            let _ = connection.await;
        });
        client
            .batch_execute(&format!("CREATE SCHEMA {schema}"))
            .await
            .expect("create the test schema");
        let store = PostgresControlPlane::connect(&dsn, settings(&schema))
            .await
            .expect("boot against a fresh schema");
        Some((store, dsn, schema))
    }

    fn settings(schema: &str) -> ControlPlaneSettings {
        ControlPlaneSettings {
            schema: Some(schema.to_owned()),
            operation_timeout: Duration::from_secs(10),
            connect_timeout: Duration::from_secs(5),
            ..ControlPlaneSettings::default()
        }
    }

    #[test]
    fn pending_operations_count_waiters_and_release_on_drop() {
        let count = Arc::new(AtomicUsize::new(0));
        let first = StatusProbeAdmission::pending(Arc::clone(&count));
        assert_eq!(count.load(Ordering::Acquire), 1);

        let second = StatusProbeAdmission::pending(Arc::clone(&count));
        assert_eq!(count.load(Ordering::Acquire), 2);
        drop(second);
        assert_eq!(count.load(Ordering::Acquire), 1);

        drop(first);
        assert_eq!(count.load(Ordering::Acquire), 0);
    }

    #[test]
    fn status_probe_reservation_is_visible_to_a_concurrent_operation() {
        let count = Arc::new(AtomicUsize::new(0));
        count.fetch_add(1, Ordering::AcqRel);
        let probe = StatusProbeAdmission::new(Duration::from_secs(65), Arc::clone(&count));
        let start = Arc::new(std::sync::Barrier::new(2));
        let observed = Arc::new(AtomicUsize::new(0));
        let worker_count = Arc::clone(&count);
        let worker_start = Arc::clone(&start);
        let worker_observed = Arc::clone(&observed);
        let worker = std::thread::spawn(move || {
            worker_start.wait();
            let _admin = StatusProbeAdmission::pending(worker_count.clone());
            worker_observed.store(worker_count.load(Ordering::Acquire), Ordering::Release);
        });

        start.wait();
        worker
            .join()
            .expect("the concurrent operation does not panic");
        assert_eq!(count.load(Ordering::Acquire), 1);
        assert_eq!(observed.load(Ordering::Acquire), 2);
        drop(probe);
        assert_eq!(count.load(Ordering::Acquire), 0);
    }

    /// A second store on the same journal, as a second replica's administrator is.
    async fn second_store(dsn: &str, schema: &str) -> PostgresControlPlane {
        PostgresControlPlane::connect(
            dsn,
            ControlPlaneSettings {
                // The schema is already current; a second booter must not need to
                // be allowed to migrate.
                migrate: false,
                ..settings(schema)
            },
        )
        .await
        .expect("boot against a current schema")
    }

    /// A store on the same journal whose hydration bounds are the caller's.
    ///
    /// The bounds are settings rather than constants precisely so a test can
    /// state one against a five-resource fixture instead of building a revision
    /// large enough to reach a production ceiling.
    async fn bounded_store(
        dsn: &str,
        schema: &str,
        hydration: HydrationLimits,
    ) -> PostgresControlPlane {
        PostgresControlPlane::connect(
            dsn,
            ControlPlaneSettings {
                migrate: false,
                hydration,
                ..settings(schema)
            },
        )
        .await
        .expect("boot against a current schema")
    }

    impl PostgresControlPlane {
        /// Corrupt the journal the way a restored backup or a manual `UPDATE`
        /// would: out of band, behind the store's back. Tests only — this is how
        /// storage the writer would never produce gets produced.
        async fn corrupt_with(&self, statement: &str) {
            let sql = statement.to_owned();
            self.run(None, move |client| {
                let sql = sql.clone();
                Box::pin(async move {
                    client
                        .batch_execute(&sql)
                        .await
                        .map_err(|error| unavailable("corrupt the journal", &error))
                })
            })
            .await
            .expect("the corrupting statement itself must succeed");
        }

        async fn count(&self, table: &str) -> i64 {
            let sql = format!("SELECT count(*) FROM {table}");
            self.run(None, move |client| {
                Box::pin(async move {
                    client
                        .query_one(&sql, &[])
                        .await
                        .map(|row| row.get(0))
                        .map_err(|error| unavailable("count", &error))
                })
            })
            .await
            .expect("count")
        }
    }

    fn uuid(seed: u64) -> Uuid7 {
        Uuid7::from_parts(seed, 0, seed).expect("seed in range")
    }

    /// A candidate carrying an explicit mutation identity, for the cases that
    /// reuse a key the fixture derives its ids from: a mutation and an audit
    /// event are each written once, so a second change needs its own.
    fn candidate_with_mutation(
        expected: ExpectedRevision,
        key: &str,
        state: crate::desired_state::DesiredState,
        seed: u64,
    ) -> RevisionCandidate {
        let mut candidate = candidate(expected, key, state);
        let mutation = MutationId::new(uuid(seed));
        candidate.mutation.id = mutation;
        candidate.audit.mutation = mutation;
        candidate.audit.id = AuditEventId::new(uuid(seed + 1));
        candidate
    }

    #[tokio::test]
    async fn boot_migrates_a_fresh_database_and_reports_the_schema_current() {
        let Some((store, dsn, schema)) = journal().await else {
            return;
        };
        assert_eq!(
            store.schema_status().await.expect("status"),
            SchemaStatus::Current {
                version: schema::required_version()
            }
        );
        assert_eq!(store.name(), "postgres");
        assert!(store.health().await.is_ok());
        assert_eq!(store.desired_revision().await.expect("head"), None);

        // A second boot is a no-op, and needs no permission to migrate.
        let second = second_store(&dsn, &schema).await;
        assert!(second.schema_status().await.expect("status").is_current());

        // An unmigrated database with migration withheld is a refusal, not a
        // gateway that serves against a schema it did not write.
        let bare = format!("{schema}_bare");
        store
            .run(None, move |client| {
                let sql = format!("CREATE SCHEMA {bare}");
                Box::pin(async move {
                    client
                        .batch_execute(&sql)
                        .await
                        .map_err(|error| unavailable("create schema", &error))
                })
            })
            .await
            .expect("create the bare schema");
        let refusal = PostgresControlPlane::connect(
            &dsn,
            ControlPlaneSettings {
                migrate: false,
                ..settings(&format!("{schema}_bare"))
            },
        )
        .await
        .expect_err("an unmigrated schema must not be served");
        assert_eq!(refusal.category(), FailureCategory::Denied);
        assert!(refusal.to_string().contains("not present"), "{refusal}");
    }

    #[tokio::test]
    async fn a_database_a_newer_build_owns_is_refused_rather_than_migrated_backwards() {
        let Some((store, dsn, schema)) = journal().await else {
            return;
        };
        store
            .run(None, |client| {
                Box::pin(async move {
                    client
                        .execute(
                            "INSERT INTO axond_cp_schema_migration (version, name, checksum) \
                             VALUES (99, 'control_plane_0099_future', $1)",
                            &[&crate::desired_state::Checksum::of(b"future").to_string()],
                        )
                        .await
                        .map(|_| ())
                        .map_err(|error| unavailable("record a future migration", &error))
                })
            })
            .await
            .expect("record a future migration");

        let status = store.schema_status().await.expect("status");
        assert!(
            matches!(status, SchemaStatus::Ahead { applied: 99, .. }),
            "{status:?}"
        );
        let refusal = PostgresControlPlane::connect(&dsn, settings(&schema))
            .await
            .expect_err("a newer schema must not be served");
        assert_eq!(refusal.category(), FailureCategory::Denied);
        assert!(refusal.to_string().contains("newer gateway"), "{refusal}");
    }

    #[tokio::test]
    async fn publication_is_a_chain_whose_history_stays_loadable() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let first = store
            .publish_revision(candidate(ExpectedRevision::Empty, "first", state()))
            .await
            .expect("first publication");
        assert_eq!(first.parent, None);
        assert_eq!(
            store.desired_revision().await.expect("head"),
            Some(first.id)
        );

        let second = store
            .publish_revision(candidate(
                ExpectedRevision::Exactly(first.id),
                "second",
                state_with_renamed_alias(),
            ))
            .await
            .expect("second publication");
        assert_eq!(second.parent, Some(first.id));
        assert_ne!(second.checksum, first.checksum);
        assert_eq!(
            store.desired_revision().await.expect("head"),
            Some(second.id)
        );

        // A manifest read back is the manifest publication returned: identity,
        // parentage, entries, blobs, and checksum, not an approximation of them.
        assert_eq!(
            store.load_manifest(first.id).await.expect("manifest"),
            first
        );
        assert_eq!(
            store.load_manifest(second.id).await.expect("manifest"),
            second
        );

        // And the earlier revision still hydrates, unchanged by the later one.
        let loaded = store.load_revision(first.id).await.expect("hydrate");
        assert_eq!(loaded.manifest(), &first);
        assert_eq!(loaded.state().len(), DESIRED_STATE_RESOURCES);
        assert_eq!(loaded.state(), &state());

        let trail = store.audit_trail(first.id).await.expect("audit trail");
        assert_eq!(trail.len(), 1);
        assert_eq!(trail[0].summary, "applied first");

        // Two revisions sharing four of five versions store nine, not ten: a
        // manifest is a reference structure in storage too.
        assert_eq!(store.count("axond_cp_resource_version").await, 6);
        assert_eq!(store.count("axond_cp_blob").await, 1);
        assert_eq!(store.count("axond_cp_revision").await, 2);

        let missing = RevisionId::new(uuid(9_999));
        assert!(matches!(
            store.load_manifest(missing).await,
            Err(ControlPlaneError::RevisionNotFound(_))
        ));
        assert!(matches!(
            store.audit_trail(missing).await,
            Err(ControlPlaneError::RevisionNotFound(_))
        ));
    }

    #[tokio::test]
    async fn concurrent_writers_agree_that_exactly_one_commit_wins() {
        let Some((store, dsn, schema)) = journal().await else {
            return;
        };
        let first = store
            .publish_revision(candidate(ExpectedRevision::Empty, "first", state()))
            .await
            .expect("first publication");
        let other = second_store(&dsn, &schema).await;

        // Two administrators, two connections, one expectation.
        let expected = ExpectedRevision::Exactly(first.id);
        let (left, right) = tokio::join!(
            store.publish_revision(candidate(expected, "race-left", state_with_renamed_alias())),
            other.publish_revision(candidate(expected, "race-right", state()))
        );
        let (winner, loser) = match (left, right) {
            (Ok(winner), Err(loser)) | (Err(loser), Ok(winner)) => (winner, loser),
            (left, right) => panic!(
                "exactly one writer must win an expected-revision race, got {left:?} and {right:?}"
            ),
        };
        assert!(matches!(
            loser,
            ControlPlaneError::Conflict {
                expected: ExpectedRevision::Exactly(_),
                actual: Some(_)
            }
        ));
        assert_eq!(loser.category(), FailureCategory::Conflict);
        assert!(loser.retryable() || !loser.retryable());

        assert_eq!(winner.parent, Some(first.id));
        assert_eq!(
            store.desired_revision().await.expect("head"),
            Some(winner.id)
        );
        // The loser left nothing: two revisions exist, not three, and the loser's
        // mutation and audit event were never recorded.
        assert_eq!(store.count("axond_cp_revision").await, 2);
        assert_eq!(store.count("axond_cp_mutation").await, 2);
        assert_eq!(store.count("axond_cp_audit_event").await, 2);
    }

    #[tokio::test]
    async fn a_failed_publication_rolls_back_every_row_it_had_written() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let first = store
            .publish_revision(candidate(ExpectedRevision::Empty, "first", state()))
            .await
            .expect("first publication");
        let recorded = store.audit_trail(first.id).await.expect("audit trail");

        // A candidate that fails *after* its resources and blobs are written: a
        // fresh resource version, but a mutation id that is already recorded.
        let mut doomed_state = state();
        doomed_state
            .insert(tenant(9, "rolled-back"))
            .expect("a second tenant is valid desired state");
        let mut doomed = candidate(
            ExpectedRevision::Exactly(first.id),
            "rollback",
            doomed_state,
        );
        doomed.mutation.id = first.mutation;
        doomed.audit.mutation = first.mutation;
        let error = store
            .publish_revision(doomed)
            .await
            .expect_err("a duplicate mutation id must not be merged into the first change");
        assert_eq!(error.category(), FailureCategory::Denied);

        // Nothing partial survives: not the new resource version, not a revision,
        // not an audit event, and not the head.
        assert_eq!(store.count("axond_cp_revision").await, 1);
        assert_eq!(store.count("axond_cp_audit_event").await, 1);
        assert_eq!(
            store.count("axond_cp_resource_version").await,
            DESIRED_STATE_RESOURCES as i64,
            "the rolled-back publication's resource version must not remain"
        );
        assert_eq!(
            store.desired_revision().await.expect("head"),
            Some(first.id)
        );
        assert_eq!(store.audit_trail(first.id).await.expect("trail"), recorded);
    }

    #[tokio::test]
    async fn an_immutable_version_cannot_be_redefined_and_leaves_nothing_behind() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let first = store
            .publish_revision(candidate(ExpectedRevision::Empty, "first", state()))
            .await
            .expect("first publication");

        // The same reference, different content: a caller redefining state an
        // earlier revision still pins.
        let mut redefined = DesiredState::new();
        for resource in state().resources() {
            let resource = if resource.reference.kind == ResourceKind::Tenant {
                let mut renamed = resource.clone();
                renamed.slug = Slug::parse("renamed").expect("slug");
                renamed
            } else {
                resource.clone()
            };
            redefined.insert(resource).expect("valid state");
        }
        for blob in state().blobs() {
            redefined.declare_blob(*blob);
        }
        let error = store
            .publish_revision(candidate(
                ExpectedRevision::Exactly(first.id),
                "redefine",
                redefined,
            ))
            .await
            .expect_err("an immutable version must not be redefined");
        assert!(
            matches!(error, ControlPlaneError::ImmutableResourceVersion { .. }),
            "{error:?}"
        );
        assert_eq!(store.count("axond_cp_revision").await, 1);
        assert_eq!(store.count("axond_cp_audit_event").await, 1);
        assert_eq!(
            store.desired_revision().await.expect("head"),
            Some(first.id)
        );
    }

    #[tokio::test]
    async fn a_repeated_key_replays_its_outcome_and_a_reused_one_is_refused() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let first = store
            .publish_revision(candidate(ExpectedRevision::Empty, "retry-1", state()))
            .await
            .expect("first publication");

        // The retry arrives with a stale expectation, which is exactly the case a
        // retry is: replay identity is the state checksum, so it replays rather
        // than conflicting.
        let replayed = store
            .publish_revision(candidate(ExpectedRevision::Empty, "retry-1", state()))
            .await
            .expect("a retry of the same state replays");
        assert_eq!(replayed, first);
        assert_eq!(store.count("axond_cp_revision").await, 1);
        assert_eq!(store.count("axond_cp_mutation").await, 1);
        assert_eq!(store.count("axond_cp_audit_event").await, 1);
        assert_eq!(store.audit_trail(first.id).await.expect("trail").len(), 1);

        // The same key describing different state is refused, never replayed as
        // an outcome the caller did not ask for.
        let error = store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(first.id),
                "retry-1",
                state_with_renamed_alias(),
                4_242,
            ))
            .await
            .expect_err("a reused key must be refused");
        let ControlPlaneError::IdempotencyKeyReused { published, .. } = error else {
            panic!("expected a reused-key refusal, got {error:?}");
        };
        assert_eq!(published, first.id);
        assert_eq!(store.count("axond_cp_revision").await, 1);
    }

    #[tokio::test]
    async fn one_callers_key_neither_replays_nor_blocks_anothers() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let first = store
            .publish_revision(candidate(ExpectedRevision::Empty, "retry-1", state()))
            .await
            .expect("first publication");

        // A different authenticated caller, the same key, the same state: a
        // second write, because deduplication is scoped per caller.
        let mut other = candidate_with_mutation(
            ExpectedRevision::Exactly(first.id),
            "retry-1",
            state(),
            7_777,
        );
        let system = Actor::System {
            component: "catalog-refresh".to_owned(),
        };
        other.mutation.actor = system.clone();
        other.audit.actor = system;
        let second = store
            .publish_revision(other)
            .await
            .expect("another caller's identical key is another write");
        assert_ne!(second.id, first.id);
        assert_eq!(second.parent, Some(first.id));
        assert_eq!(store.count("axond_cp_idempotency").await, 2);
    }

    #[tokio::test]
    async fn an_expired_retry_window_closes_without_touching_the_revision() {
        let Some((store, dsn, schema)) = journal().await else {
            return;
        };
        // A store whose retry window is closed as soon as it is opened, so the
        // retry below arrives after expiry rather than after a wait.
        let expiring = PostgresControlPlane::connect(
            &dsn,
            ControlPlaneSettings {
                migrate: false,
                idempotency_retention: Duration::ZERO,
                ..settings(&schema)
            },
        )
        .await
        .expect("boot");
        let first = expiring
            .publish_revision(candidate(ExpectedRevision::Empty, "retry-1", state()))
            .await
            .expect("first publication");

        let second = store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(first.id),
                "retry-1",
                state(),
                8_888,
            ))
            .await
            .expect("an expired record is not a replay");
        assert_ne!(second.id, first.id);

        // Expiry is a window closing, not a deletion: the revision the record
        // pointed at, and its audit trail, are untouched.
        assert_eq!(
            store.load_manifest(first.id).await.expect("manifest"),
            first
        );
        assert_eq!(store.audit_trail(first.id).await.expect("trail").len(), 1);
        assert_eq!(store.count("axond_cp_revision").await, 2);
    }

    #[tokio::test]
    async fn an_outage_is_an_outage_and_changes_nothing_a_replica_holds() {
        let Some((store, dsn, schema)) = journal().await else {
            return;
        };
        let first = store
            .publish_revision(candidate(ExpectedRevision::Empty, "first", state()))
            .await
            .expect("first publication");
        // What a replica is serving: an immutable value it already holds.
        let held = store.load_revision(first.id).await.expect("hydrate");

        let stalled = PostgresControlPlane::connect(
            &dsn,
            ControlPlaneSettings {
                migrate: false,
                operation_timeout: Duration::from_nanos(1),
                ..settings(&schema)
            },
        )
        .await
        .expect("boot");
        let error = stalled
            .publish_revision(candidate(
                ExpectedRevision::Exactly(first.id),
                "stalled",
                state_with_renamed_alias(),
            ))
            .await
            .expect_err("a publication that cannot finish must not report success");
        assert_eq!(error.category(), FailureCategory::Unavailable);
        assert!(error.retryable());

        // The outage moved nothing: not the head, not the journal, and not the
        // snapshot the replica holds — which is a value, not a query.
        assert_eq!(
            store.desired_revision().await.expect("head"),
            Some(first.id)
        );
        assert_eq!(store.count("axond_cp_revision").await, 1);
        assert_eq!(store.count("axond_cp_audit_event").await, 1);
        assert_eq!(held.manifest(), &first);
        assert_eq!(held.state(), &state());

        // And the store recovers: an outage discards the connection rather than
        // poisoning the store.
        assert!(store.health().await.is_ok());
    }

    #[tokio::test]
    async fn the_journal_answers_the_contract_the_way_the_oracle_does() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let oracle = InMemoryControlPlane::new();
        let stores: [&dyn ControlPlaneStore; 2] = [&store, &oracle];
        let mut heads = Vec::new();
        for store in stores {
            assert_eq!(store.desired_revision().await.expect("head"), None);
            let first = store
                .publish_revision(candidate(ExpectedRevision::Empty, "first", state()))
                .await
                .expect("first publication");
            let stale = store
                .publish_revision(candidate(
                    ExpectedRevision::Empty,
                    "stale",
                    state_with_renamed_alias(),
                ))
                .await
                .expect_err("a stale expectation conflicts");
            assert_eq!(stale.category(), FailureCategory::Conflict);
            let replay = store
                .publish_revision(candidate(ExpectedRevision::Empty, "first", state()))
                .await
                .expect("a retry replays");
            assert_eq!(replay, first);
            let loaded = store.load_revision(first.id).await.expect("hydrate");
            assert_eq!(loaded.state(), &state());
            assert_eq!(
                store.audit_trail(first.id).await.expect("trail").len(),
                1,
                "one mutation is one audit event"
            );
            heads.push(store.desired_revision().await.expect("head").is_some());
        }
        assert_eq!(heads, vec![true, true]);
    }

    /// The boot path classifies a rejected migration the way the operator command
    /// does: a schema nothing created is an operator's to create, so a replica
    /// says so rather than reporting an outage a supervisor would retry forever.
    #[tokio::test]
    async fn a_boot_migration_the_server_rejects_is_denied_rather_than_an_outage() {
        let Some(dsn) = crate::test_services::postgres_dsn() else {
            return;
        };
        let missing = format!(
            "cp_absent_{}",
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .expect("clock")
                .as_nanos()
        );
        let error = PostgresControlPlane::connect(
            &dsn,
            ControlPlaneSettings {
                migrate: true,
                ..settings(&missing)
            },
        )
        .await
        .expect_err("a schema that does not exist cannot be migrated into");
        let ControlPlaneError::Denied { message, .. } = &error else {
            panic!("a rejected statement is not an outage: {error:?}");
        };
        assert!(
            message.contains("schema exists"),
            "the refusal names what to fix: {message}"
        );
    }

    /// Two replicas booting together against one empty database on a server whose
    /// default isolation is `repeatable read`: the one that waits for the advisory
    /// lock must read the winner's committed ledger, not the snapshot it took
    /// before the wait. Reading the stale one either re-runs the DDL or — catalog
    /// lookups being fresh while row reads are not — finds a ledger table with no
    /// visible rows, which is a refused boot.
    #[tokio::test]
    async fn replicas_booting_together_migrate_once_whatever_the_server_default_is() {
        let Some(dsn) = crate::test_services::postgres_dsn() else {
            return;
        };
        let schema = format!(
            "cp_rr_{}",
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .expect("clock")
                .as_nanos()
        );
        let mut config: Config = dsn.parse().expect("test dsn");
        config.connect_timeout(Duration::from_secs(5));
        let (client, connection) = config
            .connect(crate::usage::tls_connector())
            .await
            .expect("connect to create the test schema");
        tokio::spawn(async move {
            let _ = connection.await;
        });
        client
            .batch_execute(&format!("CREATE SCHEMA {schema}"))
            .await
            .expect("create the test schema");

        // The stricter default is asked for per connection rather than set on the
        // server, so the rest of the suite is unaffected by this test.
        let separator = if dsn.contains('?') { "&" } else { "?" };
        // A space inside a startup option's value is escaped, because the server
        // splits `options` on unescaped whitespace.
        let strict = format!(
            "{dsn}{separator}options=-c%20default_transaction_isolation%3Drepeatable%5C%20read"
        );
        let booting = || async {
            PostgresControlPlane::connect(
                &strict,
                ControlPlaneSettings {
                    migrate: true,
                    ..settings(&schema)
                },
            )
            .await
        };
        let (first, second) = tokio::join!(booting(), booting());
        let first = first.expect("a replica migrating an empty database boots");
        second.expect("the replica that waited for the lock boots too");

        assert!(
            first.schema_status().await.expect("status").is_current(),
            "one migration, not two"
        );
    }

    #[tokio::test]
    async fn requested_tls_is_not_silently_downgraded() {
        let Some(dsn) = crate::test_services::postgres_dsn() else {
            return;
        };
        // The CI server speaks no TLS. `sslmode=require` must therefore fail to
        // connect rather than continue in the clear.
        let separator = if dsn.contains('?') { "&" } else { "?" };
        let error = PostgresControlPlane::connect(
            &format!("{dsn}{separator}sslmode=require"),
            ControlPlaneSettings {
                migrate: false,
                ..ControlPlaneSettings::default()
            },
        )
        .await
        .expect_err("TLS that cannot be established is a refusal, not a plaintext session");
        assert!(
            matches!(
                error,
                ControlPlaneError::Unavailable { .. } | ControlPlaneError::Denied { .. }
            ),
            "{error:?}"
        );
    }

    /// The three revisions the hydration tests share: the fixture state, a
    /// rename of one alias, and a second tenant's resources alongside the first's.
    ///
    /// Published in order, so the first two are *historical* by the time anything
    /// loads them — which is the case #166 exists to keep answerable.
    async fn three_revisions(
        store: &PostgresControlPlane,
    ) -> (RevisionManifest, RevisionManifest, RevisionManifest) {
        let first = store
            .publish_revision(candidate(ExpectedRevision::Empty, "first", state()))
            .await
            .expect("first publication");
        let second = store
            .publish_revision(candidate(
                ExpectedRevision::Exactly(first.id),
                "second",
                state_with_renamed_alias(),
            ))
            .await
            .expect("second publication");
        let third = store
            .publish_revision(candidate(
                ExpectedRevision::Exactly(second.id),
                "third",
                state_with_second_tenant(),
            ))
            .await
            .expect("third publication");
        (first, second, third)
    }

    #[tokio::test]
    async fn a_historical_revision_hydrates_as_the_state_it_published() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let (first, second, third) = three_revisions(&store).await;

        // Two newer commits later, the oldest revision is still the value it was
        // published as — not a diff to be applied, and not a state the newer
        // revisions edited.
        for (manifest, published) in [
            (&first, state()),
            (&second, state_with_renamed_alias()),
            (&third, state_with_second_tenant()),
        ] {
            let loaded = store.load_revision(manifest.id).await.expect("hydrate");
            assert_eq!(loaded.state(), &published);
            assert_eq!(loaded.manifest(), manifest);
            // The checksum invariant, stated as the hydration itself: the state
            // that came back hashes to what the manifest recorded.
            assert_eq!(
                loaded.state().checksum().expect("canonical"),
                manifest.checksum
            );
            assert_eq!(loaded.manifest().entries.len(), loaded.state().len());
        }

        // Hydration is repeatable, because a revision is immutable: the same read
        // twice is the same value, byte for byte.
        let once = store.load_revision(first.id).await.expect("hydrate");
        let twice = store.load_revision(first.id).await.expect("hydrate");
        assert_eq!(once.state(), twice.state());
        assert_eq!(once.manifest(), twice.manifest());
    }

    /// The exemption that keeps already-published journals loadable, against a
    /// journal rather than against a hand-built state.
    ///
    /// Typed tenancy made hydration read bodies, so what an older build could
    /// publish has to keep hydrating: tenant-scoped resources with no tenant row,
    /// and a project scope naming a project no row declares — unroutable, not a
    /// contradiction. What is refused is a contradiction, and it is refused as
    /// damage, because no release ever wrote a resource into a project owned by a
    /// tenant other than the one its scope names.
    #[tokio::test]
    async fn a_revision_an_older_build_published_still_hydrates() {
        let Some((store, _, _)) = journal().await else {
            return;
        };

        let published = state_a_pre_tenancy_build_published();
        let manifest = store
            .publish_revision(candidate(
                ExpectedRevision::Empty,
                "pre-tenancy",
                published.clone(),
            ))
            .await
            .expect("a revision without owner rows publishes");
        let loaded = store
            .load_revision(manifest.id)
            .await
            .expect("and hydrates on a build that reads typed bodies");
        assert_eq!(loaded.state(), &published);

        // The same project scope, now with the project declared and owned by the
        // tenant the scope names: the rule applies, and holds.
        let owner = tenant_id(1);
        let mut owned = state();
        owned
            .insert(project_alias(&owner, &project_id(2), 6, "inner"))
            .expect("a project's own alias is valid");
        let second = store
            .publish_revision(candidate(
                ExpectedRevision::Exactly(manifest.id),
                "owned",
                owned.clone(),
            ))
            .await
            .expect("a declared owner is not a new requirement");
        assert_eq!(
            store
                .load_revision(second.id)
                .await
                .expect("hydrate")
                .state(),
            &owned
        );

        // And a contradiction of it, as a restored backup or a manual `UPDATE`
        // produces one: the row now sits in a project whose tenant is not the one
        // it names. The stranger is given a tenant row first, because 0004's key
        // refuses a journal row naming a tenant that has none — which is the point
        // of that key, and is also why the corruption this test is about has to be
        // the one the database does not answer: whether the *project* belongs to
        // the tenant beside it.
        let stranger = tenant_id(11);
        store
            .corrupt_with(&format!(
                "INSERT INTO axond_cp_tenant (tenant_id, slug, lifecycle, revision_id) \
                 VALUES ('{stranger}', 'stranger', 'active', '{}') \
                 ON CONFLICT (tenant_id) DO NOTHING; \
                 UPDATE axond_cp_resource_version SET tenant_id = '{stranger}' \
                 WHERE scope_kind = 'project'",
                second.id
            ))
            .await;
        let error = store
            .load_revision(second.id)
            .await
            .expect_err("a resource in another tenant's project is not hydratable");
        assert_eq!(error.category(), FailureCategory::Corrupt);
    }

    #[tokio::test]
    async fn two_revisions_share_immutable_resources_without_sharing_a_value() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let (first, second, _) = three_revisions(&store).await;

        // The tenant, project, credential, and catalogue are one row each,
        // pinned by every revision that names them; only the alias was rewritten,
        // and the second tenant's three resources are new.
        assert_eq!(
            store.count("axond_cp_resource_version").await,
            i64::try_from(DESIRED_STATE_RESOURCES + 1 + 3).expect("small"),
        );
        // One catalogue payload, three revisions: a blob is referenced, never
        // copied per revision.
        assert_eq!(store.count("axond_cp_blob").await, 1);
        assert_eq!(store.count("axond_cp_revision_blob").await, 3);

        let older = store.load_revision(first.id).await.expect("hydrate");
        let newer = store.load_revision(second.id).await.expect("hydrate");

        // The shared credential hydrates as an equal value in both revisions,
        // and each revision holds its own alias version. Sharing storage is not
        // sharing state: these are two independent values, so neither can be
        // mutated through the other.
        let credential = reference(ResourceKind::ProviderCredential, 3);
        assert_eq!(
            older.state().get(&credential),
            newer.state().get(&credential)
        );
        let alias_v1 = reference(ResourceKind::Alias, 4);
        let alias_v2 = alias_v1.at(ResourceVersionNumber::FIRST.next());
        assert!(older.state().get(&alias_v1).is_some());
        assert!(older.state().get(&alias_v2).is_none());
        assert!(newer.state().get(&alias_v2).is_some());
        assert!(newer.state().get(&alias_v1).is_none());
        assert_ne!(older.state(), newer.state());

        // And the shared blob is the same reference in both, by digest and size.
        let blobs: Vec<_> = older.state().blobs().collect();
        assert_eq!(blobs, newer.state().blobs().collect::<Vec<_>>());
        assert_eq!(blobs.len(), 1);
    }

    #[tokio::test]
    async fn the_desired_revision_and_its_hydration_are_one_read() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        // No publication is an empty answer, not an error: a cold deployment has
        // nothing desired yet.
        assert!(
            store
                .load_desired_revision()
                .await
                .expect("an unpublished journal is not a failure")
                .is_none()
        );

        let (_, _, third) = three_revisions(&store).await;
        let desired = store
            .load_desired_revision()
            .await
            .expect("hydrate the head")
            .expect("a head exists");
        assert_eq!(desired.manifest(), &third);
        assert_eq!(desired.state(), &state_with_second_tenant());

        // The oracle answers the same question the same way through the default
        // implementation, so #142 reads one seam rather than two.
        let oracle = InMemoryControlPlane::new();
        assert!(
            oracle
                .load_desired_revision()
                .await
                .expect("empty")
                .is_none()
        );
        let published = oracle
            .publish_revision(candidate(ExpectedRevision::Empty, "first", state()))
            .await
            .expect("publication");
        let loaded = oracle
            .load_desired_revision()
            .await
            .expect("hydrate the head")
            .expect("a head exists");
        assert_eq!(loaded.manifest(), &published);
        assert_eq!(loaded.state(), &state());
    }

    #[tokio::test]
    async fn a_manifest_reference_whose_version_is_gone_is_named_not_dropped() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let (first, _, _) = three_revisions(&store).await;

        // A restored backup that lost a row, reproduced exactly: the referential
        // integrity the writer relies on is dropped, and then the row is.
        let project = reference(ResourceKind::Project, 2);
        store
            .corrupt_with(
                "DO $$ DECLARE name text; BEGIN \
                 SELECT conname INTO name FROM pg_constraint \
                 WHERE conrelid = 'axond_cp_revision_entry'::regclass AND contype = 'f' \
                 AND conname LIKE '%resource%'; \
                 EXECUTE format('ALTER TABLE axond_cp_revision_entry DROP CONSTRAINT %I', name); \
                 END $$;",
            )
            .await;
        store
            .corrupt_with(&format!(
                "DELETE FROM axond_cp_resource_version WHERE resource_id = '{}'",
                project.id
            ))
            .await;

        // A shorter manifest would hydrate, and then fail its checksum for
        // reasons that name no row. The reference is named instead.
        for error in [
            store
                .load_manifest(first.id)
                .await
                .expect_err("a manifest missing a version is not a manifest"),
            store
                .load_revision(first.id)
                .await
                .expect_err("a revision missing a version is not hydratable"),
        ] {
            assert_eq!(error.category(), FailureCategory::Corrupt);
            let ControlPlaneError::Corrupt { source, .. } = &error else {
                panic!("expected corruption, got {error:?}");
            };
            assert_eq!(
                **source,
                IntegrityError::MissingResource { reference: project },
                "{error:?}"
            );
        }
    }

    #[tokio::test]
    async fn a_declared_blob_whose_record_is_gone_is_named_not_dropped() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let (first, _, _) = three_revisions(&store).await;

        // The blob record deleted from under the revisions that declare it. The
        // declarations remain, so this is the blob-side twin of a missing
        // resource version: the manifest must not come back one blob shorter and
        // well-formed.
        store
            .corrupt_with(
                "DO $$ DECLARE name text; BEGIN \
                 SELECT conname INTO name FROM pg_constraint \
                 WHERE conrelid = 'axond_cp_revision_blob'::regclass AND contype = 'f' \
                 AND conname LIKE '%blob_kind%'; \
                 EXECUTE format('ALTER TABLE axond_cp_revision_blob DROP CONSTRAINT %I', name); \
                 END $$;",
            )
            .await;
        store
            .corrupt_with(
                "ALTER TABLE axond_cp_resource_version \
                 DROP CONSTRAINT axond_cp_resource_version_body_blob_kind_body_blob_digest_fkey; \
                 DELETE FROM axond_cp_blob",
            )
            .await;

        for error in [
            store
                .load_manifest(first.id)
                .await
                .expect_err("a manifest declaring a blob that is gone is not a manifest"),
            store
                .load_revision(first.id)
                .await
                .expect_err("a revision declaring a blob that is gone is not hydratable"),
        ] {
            assert_eq!(error.category(), FailureCategory::Corrupt);
            let ControlPlaneError::Corrupt { source, .. } = &error else {
                panic!("expected corruption, got {error:?}");
            };
            let IntegrityError::Unreadable { detail } = &**source else {
                panic!("expected an unreadable row, got {error:?}");
            };
            assert!(detail.contains("no blob record"), "{detail}");
        }
    }

    #[tokio::test]
    async fn a_dependency_edge_that_leaves_the_revision_is_refused() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let (first, _, _) = three_revisions(&store).await;

        // An edge from the first revision's alias to the *second* revision's
        // alias version: both rows exist, so referential integrity is satisfied,
        // and the reference still leaves the revision that names it.
        let alias = reference(ResourceKind::Alias, 4);
        store
            .corrupt_with(&format!(
                "INSERT INTO axond_cp_resource_dependency \
                 (resource_kind, resource_id, version, depends_on_kind, depends_on_id, \
                 depends_on_version) VALUES ('alias', '{id}', 1, 'alias', '{id}', 2)",
                id = alias.id
            ))
            .await;

        let error = store
            .load_revision(first.id)
            .await
            .expect_err("a reference that leaves the revision is not hydratable");
        let ControlPlaneError::Corrupt { source, .. } = &error else {
            panic!("expected corruption, got {error:?}");
        };
        assert!(
            matches!(
                **source,
                IntegrityError::Invalid(ValidationError::DanglingResourceReference { from, to })
                    if from == alias && to == alias.at(ResourceVersionNumber::FIRST.next())
            ),
            "{error:?}"
        );
    }

    #[tokio::test]
    async fn an_edge_across_a_tenant_boundary_is_refused_by_the_reference_layer() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let (_, _, third) = three_revisions(&store).await;

        // One tenant's alias made to depend on another tenant's credential.
        // Both versions are in this revision, so nothing but the tenant
        // comparison distinguishes this from a legitimate edge.
        let alias = reference(ResourceKind::Alias, 14);
        let credential = reference(ResourceKind::ProviderCredential, 3);
        store
            .corrupt_with(&format!(
                "INSERT INTO axond_cp_resource_dependency \
                 (resource_kind, resource_id, version, depends_on_kind, depends_on_id, \
                 depends_on_version) VALUES ('alias', '{alias}', 1, 'provider-credential', \
                 '{credential}', 1)",
                alias = alias.id,
                credential = credential.id
            ))
            .await;

        let error = store
            .load_revision(third.id)
            .await
            .expect_err("one tenant's state must never hydrate into another's");
        assert_eq!(error.category(), FailureCategory::Corrupt);
        let ControlPlaneError::Corrupt { source, .. } = &error else {
            panic!("expected corruption, got {error:?}");
        };
        assert!(
            matches!(
                **source,
                IntegrityError::Invalid(ValidationError::CrossTenantReference { from, to })
                    if from == alias && to == credential
            ),
            "{error:?}"
        );
    }

    #[tokio::test]
    async fn a_deployment_scoped_version_may_not_depend_on_one_tenants_state() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let (first, _, _) = three_revisions(&store).await;

        // The catalogue is deployment-scoped: shared by every tenant. An edge
        // from it into one tenant's credential would make deployment-wide state
        // depend on one tenant's, which is the isolation failure that scales.
        let catalog = reference(ResourceKind::CatalogModel, 5);
        let credential = reference(ResourceKind::ProviderCredential, 3);
        store
            .corrupt_with(&format!(
                "INSERT INTO axond_cp_resource_dependency \
                 (resource_kind, resource_id, version, depends_on_kind, depends_on_id, \
                 depends_on_version) VALUES ('catalog-model', '{catalog}', 1, \
                 'provider-credential', '{credential}', 1)",
                catalog = catalog.id,
                credential = credential.id
            ))
            .await;

        let error = store
            .load_revision(first.id)
            .await
            .expect_err("deployment-wide state must not depend on one tenant's");
        let ControlPlaneError::Corrupt { source, .. } = &error else {
            panic!("expected corruption, got {error:?}");
        };
        assert!(
            matches!(
                **source,
                IntegrityError::Invalid(ValidationError::TenantScopedDependency { from, to })
                    if from == catalog && to == credential
            ),
            "{error:?}"
        );
    }

    #[tokio::test]
    async fn a_state_checksum_that_no_longer_matches_refuses_the_revision() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let (first, _, _) = three_revisions(&store).await;

        // A revision row made to claim a different state's checksum: the rows
        // are individually well-formed, and the whole no longer adds up.
        let foreign = state_with_renamed_alias().checksum().expect("canonical");
        store
            .corrupt_with(&format!(
                "UPDATE axond_cp_revision SET state_checksum = '{foreign}' \
                 WHERE revision_id = '{}'",
                first.id
            ))
            .await;

        let error = store
            .load_revision(first.id)
            .await
            .expect_err("a state that hashes to something else is not the state");
        let ControlPlaneError::Corrupt { source, .. } = &error else {
            panic!("expected corruption, got {error:?}");
        };
        assert!(
            matches!(
                **source,
                IntegrityError::ChecksumMismatch { expected, .. } if expected == foreign
            ),
            "{error:?}"
        );
    }

    #[tokio::test]
    async fn a_resource_whose_content_checksum_no_longer_matches_is_refused() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let (first, _, _) = three_revisions(&store).await;

        // A slug edited in place. Renaming is a new version, so the row's own
        // content checksum is now the checksum of state this row never held.
        let project = reference(ResourceKind::Project, 2);
        store
            .corrupt_with(&format!(
                "UPDATE axond_cp_resource_version SET slug = 'edited' \
                 WHERE resource_id = '{}'",
                project.id
            ))
            .await;

        let error = store
            .load_revision(first.id)
            .await
            .expect_err("a resource that is not its own content checksum is not hydratable");
        let ControlPlaneError::Corrupt { source, .. } = &error else {
            panic!("expected corruption, got {error:?}");
        };
        assert!(
            matches!(
                **source,
                IntegrityError::ContentMismatch { reference, .. } if reference == project
            ),
            "{error:?}"
        );
    }

    #[tokio::test]
    async fn a_body_that_cannot_be_decoded_is_corruption_and_not_an_outage() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let (first, _, _) = three_revisions(&store).await;

        let project = reference(ResourceKind::Project, 2);
        store
            .corrupt_with(&format!(
                "UPDATE axond_cp_resource_version SET body_inline = '\\xdeadbeef'::bytea \
                 WHERE resource_id = '{}'",
                project.id
            ))
            .await;

        // The manifest does not read bodies, so it still loads: corruption is
        // reported by the read that actually touches the corrupt column, and
        // hydration is that read.
        store
            .load_manifest(first.id)
            .await
            .expect("a manifest hydrates no body");
        let error = store
            .load_revision(first.id)
            .await
            .expect_err("a body this build cannot decode is not hydratable");
        assert_eq!(error.category(), FailureCategory::Corrupt);
        assert!(!error.retryable(), "corruption is not cleared by retrying");
        let ControlPlaneError::Corrupt { source, .. } = &error else {
            panic!("expected corruption, got {error:?}");
        };
        assert!(
            matches!(**source, IntegrityError::Unreadable { .. }),
            "{error:?}"
        );
    }

    #[tokio::test]
    async fn a_row_written_by_a_serializer_this_build_does_not_read_is_refused() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let (first, _, _) = three_revisions(&store).await;

        store
            .corrupt_with(
                "UPDATE axond_cp_resource_version SET serializer = 'axond.desired-state.v99'",
            )
            .await;

        // A version of this encoding this build has not learned: the rows may be
        // entirely intact, and the action is a deployment rather than a restore.
        let error = store
            .load_revision(first.id)
            .await
            .expect_err("a row this build cannot read is not hydratable");
        assert_ne!(error.category(), FailureCategory::Corrupt);
        let ControlPlaneError::Incompatible { source, .. } = &error else {
            panic!("expected an incompatibility, got {error:?}");
        };
        assert!(
            matches!(**source, IntegrityError::UnknownSerializer { .. }),
            "{error:?}"
        );

        // Text naming no version of this encoding at all is the other thing: no
        // release wrote it, so it points at storage.
        store
            .corrupt_with("UPDATE axond_cp_resource_version SET serializer = 'json'")
            .await;
        let error = store
            .load_revision(first.id)
            .await
            .expect_err("a column no build ever wrote is not hydratable");
        assert_eq!(error.category(), FailureCategory::Corrupt);
        let ControlPlaneError::Corrupt { source, .. } = &error else {
            panic!("expected corruption, got {error:?}");
        };
        assert!(
            matches!(**source, IntegrityError::Unreadable { .. }),
            "{error:?}"
        );
    }

    #[tokio::test]
    async fn a_version_number_the_domain_cannot_hold_is_refused() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let (first, _, _) = three_revisions(&store).await;

        // Version numbers start at one, and the DDL says so. This is what a
        // journal looks like after someone removed that guarantee and wrote a
        // zero: a manifest entry naming a version the domain cannot represent.
        let alias = reference(ResourceKind::Alias, 4);
        store
            .corrupt_with(
                "DO $$ DECLARE name text; BEGIN \
                 SELECT conname INTO name FROM pg_constraint \
                 WHERE conrelid = 'axond_cp_revision_entry'::regclass AND contype = 'f' \
                 AND conname LIKE '%resource%'; \
                 EXECUTE format('ALTER TABLE axond_cp_revision_entry DROP CONSTRAINT %I', name); \
                 END $$;",
            )
            .await;
        store
            .corrupt_with(&format!(
                "UPDATE axond_cp_revision_entry SET version = 0 \
                 WHERE resource_id = '{}' AND revision_id = '{revision}'",
                alias.id,
                revision = first.id
            ))
            .await;

        let error = store
            .load_revision(first.id)
            .await
            .expect_err("a version number that is not one is not a version number");
        assert_eq!(error.category(), FailureCategory::Corrupt);
        let ControlPlaneError::Corrupt { source, .. } = &error else {
            panic!("expected corruption, got {error:?}");
        };
        let IntegrityError::Unreadable { detail } = &**source else {
            panic!("expected an unreadable row, got {error:?}");
        };
        assert!(detail.contains("resource version 0"), "{detail}");
    }

    #[tokio::test]
    async fn a_revision_larger_than_the_bound_is_refused_whole() {
        let Some((store, dsn, schema)) = journal().await else {
            return;
        };
        let (first, _, third) = three_revisions(&store).await;

        // Each bound, stated against the same five-resource revision. Every one
        // of them is a refusal: not a truncated manifest, not the resources that
        // fit, and not a candidate missing its blob.
        let cases: [(HydrationLimits, &str); 5] = [
            (
                HydrationLimits {
                    max_entries: 2,
                    ..HydrationLimits::default()
                },
                "resource versions",
            ),
            (
                HydrationLimits {
                    max_blobs: 0,
                    ..HydrationLimits::default()
                },
                "blobs",
            ),
            (
                HydrationLimits {
                    max_blob_bytes: 8,
                    ..HydrationLimits::default()
                },
                "bytes",
            ),
            (
                HydrationLimits {
                    max_dependency_edges: 1,
                    ..HydrationLimits::default()
                },
                "dependency edges",
            ),
            (
                HydrationLimits {
                    max_inline_body_bytes: 4,
                    ..HydrationLimits::default()
                },
                "inline body",
            ),
        ];
        for (limits, expected) in cases {
            let bounded = bounded_store(&dsn, &schema, limits).await;
            let error = bounded
                .load_revision(first.id)
                .await
                .expect_err("a revision past a bound must not hydrate");
            assert_eq!(error.category(), FailureCategory::Denied);
            assert!(!error.retryable(), "a bound is not cleared by retrying");
            assert!(
                matches!(error, ControlPlaneError::TooLarge { revision, .. } if revision == first.id),
                "{error:?}"
            );
            assert!(error.to_string().contains(expected), "{error}");
        }

        // The declared-bytes figure is the revision's real total, not however
        // much of it was counted before the ceiling was crossed: an operator
        // deciding where to put the bound is told the size they have to clear.
        //
        // Stated against a revision declaring *two* blobs, because with one the
        // partial sum and the total are the same number and the assertion would
        // hold under the short-circuiting this pins.
        let two_blobs = state_with_two_blobs();
        let fourth = store
            .publish_revision(candidate(
                ExpectedRevision::Exactly(third.id),
                "fourth",
                two_blobs.clone(),
            ))
            .await
            .expect("fourth publication");
        let sizes: Vec<u64> = two_blobs.blobs().map(|blob| blob.size_bytes).collect();
        assert_eq!(sizes.len(), 2, "the fixture must declare two blobs");
        let total = sizes.iter().sum::<u64>();
        let bounded = bounded_store(
            &dsn,
            &schema,
            HydrationLimits {
                max_blob_bytes: 8,
                ..HydrationLimits::default()
            },
        )
        .await;
        let error = bounded
            .load_manifest(fourth.id)
            .await
            .expect_err("a revision past a bound must not hydrate");
        assert!(error.to_string().contains(&total.to_string()), "{error}");
        // And the figure is not either blob on its own, which is what a sum that
        // stopped at the first row over the ceiling would have reported.
        for size in sizes {
            assert!(
                !error.to_string().contains(&size.to_string()),
                "reported a partial sum: {error}"
            );
        }

        // The candidate bound is the last line: a state that hydrated is still
        // refused if the value itself is larger than this build returns.
        let bounded = bounded_store(
            &dsn,
            &schema,
            HydrationLimits {
                max_state_bytes: 64,
                ..HydrationLimits::default()
            },
        )
        .await;
        let error = bounded
            .load_revision(first.id)
            .await
            .expect_err("a candidate past the bound must not be returned");
        assert!(
            matches!(
                error,
                ControlPlaneError::TooLarge {
                    limit: HydrationLimit::StateBytes { .. },
                    ..
                }
            ),
            "{error:?}"
        );

        // Nothing above changed the journal, and the store whose bounds fit
        // still hydrates the same revision it always did.
        assert_eq!(
            store
                .load_revision(first.id)
                .await
                .expect("hydrate")
                .state(),
            &state()
        );
    }

    /// Test-only reads and writes against the projection, which the store
    /// deliberately has no read API for: nothing in the gateway authorizes
    /// against these rows — the directory of the current revision does that — so
    /// they exist to be constrained, not to be queried.
    impl PostgresControlPlane {
        async fn column(&self, sql: &str) -> Vec<String> {
            let sql = sql.to_owned();
            self.run(None, move |client| {
                let sql = sql.clone();
                Box::pin(async move {
                    client
                        .query(&sql, &[])
                        .await
                        .map(|rows| {
                            rows.iter()
                                .map(|row| {
                                    row.try_get::<_, Option<String>>(0)
                                        .expect("a text column")
                                        .unwrap_or_else(|| "<null>".to_owned())
                                })
                                .collect()
                        })
                        .map_err(|error| unavailable("read a projected column", &error))
                })
            })
            .await
            .expect("the read itself must succeed")
        }

        /// A statement whose *failure* is the assertion: what a service-layer bug
        /// would try to write, and what the database refuses on its own.
        async fn attempt(&self, statement: &str) -> Result<(), ControlPlaneError> {
            let sql = statement.to_owned();
            self.run(None, move |client| {
                let sql = sql.clone();
                Box::pin(async move {
                    client
                        .batch_execute(&sql)
                        .await
                        .map_err(|error| unavailable("attempt a write", &error))
                })
            })
            .await
        }
    }

    /// A digest of the shape the projection writes, for the constraint tests that
    /// need a *valid* digest so the constraint under test is the one that fires.
    const DIGEST: &str =
        "'sha256:0000000000000000000000000000000000000000000000000000000000000000'";

    fn denial(seed: u64, scope: ResourceScope, reason: DenialReason) -> AccessDenial {
        AccessDenial {
            id: AuditEventId::new(uuid(seed)),
            actor: Actor::Human {
                issuer: "https://idp.example".to_owned(),
                subject: "dev".to_owned(),
            },
            surface: Surface::Credential,
            action: Action::Rotate,
            scope,
            reason,
            recorded_at: journal_now() + Duration::from_secs(seed),
        }
    }

    /// The projection is the database's own copy of who owns what and who may do
    /// what, written in the transaction that published it.
    #[tokio::test]
    async fn publishing_a_revision_projects_the_owners_and_the_directory_it_declares() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let first = store
            .publish_revision(candidate(
                ExpectedRevision::Empty,
                "directory",
                state_with_directory(),
            ))
            .await
            .expect("a revision declaring a directory publishes");

        assert_eq!(
            store
                .column("SELECT slug FROM axond_cp_tenant ORDER BY slug")
                .await,
            vec!["acme"]
        );
        assert_eq!(
            store.column("SELECT lifecycle FROM axond_cp_tenant").await,
            vec!["active"]
        );
        assert_eq!(
            store
                .column("SELECT slug FROM axond_cp_project ORDER BY slug")
                .await,
            vec!["core"]
        );
        assert_eq!(
            store
                .column("SELECT slug FROM axond_cp_principal ORDER BY slug")
                .await,
            vec!["admin", "deployer", "dev", "root"]
        );
        assert_eq!(
            store
                .column("SELECT role FROM axond_cp_principal_role ORDER BY role")
                .await,
            vec!["developer", "operator", "platform-admin", "tenant-admin"]
        );
        // The revision that projected each row is recorded, so an operator can ask
        // which publication a grant came from.
        assert_eq!(
            store
                .column("SELECT DISTINCT revision_id FROM axond_cp_principal")
                .await,
            vec![first.id.to_string()]
        );

        // A minted key is stored as a digest and nothing else: the material never
        // reaches the database, and a human never has a digest column at all.
        let digests = store
            .column("SELECT key_digest FROM axond_cp_principal ORDER BY slug")
            .await;
        assert_eq!(
            digests.iter().filter(|digest| *digest != "<null>").count(),
            1
        );
        for digest in &digests {
            assert!(
                !digest.contains(WorkloadKey::PREFIX),
                "a key reached the database: {digest}"
            );
        }
        assert!(
            digests
                .iter()
                .any(|digest| *digest == Checksum::of(workload_key(0xd0).as_bytes()).to_string()),
            "{digests:?}"
        );

        // A revision that stops declaring a principal revokes it: the row and its
        // grants are gone, because a grant nobody published is authority nobody
        // granted. The tenant it belonged to stays, because history points at it.
        store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(first.id),
                "revoked",
                state_with_revoked_workload(),
                71,
            ))
            .await
            .expect("a revision that drops a principal publishes");
        assert_eq!(
            store
                .column("SELECT slug FROM axond_cp_principal ORDER BY slug")
                .await,
            vec!["admin", "dev", "root"]
        );
        assert_eq!(
            store
                .column("SELECT count(*)::text FROM axond_cp_principal_role")
                .await,
            vec!["3"]
        );
        assert_eq!(
            store.column("SELECT slug FROM axond_cp_tenant").await,
            vec!["acme"]
        );
    }

    /// A revision is the whole directory, so a revision declaring no identity
    /// declares that nobody is a principal — and the projection carries that out.
    ///
    /// Asserted rather than guarded against: the alternative reading, that an
    /// empty directory is a mistake to refuse, would make the *only* way to revoke
    /// the last principal impossible, and a deployment whose administrators are
    /// all OIDC-issued has no reason to declare one. What must not happen is a
    /// silent partial revocation, and that is what the `DELETE` above cannot do:
    /// it removes exactly the principals this revision stopped declaring.
    #[tokio::test]
    async fn a_revision_declaring_no_identity_revokes_every_principal_and_keeps_its_tenants() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let first = store
            .publish_revision(candidate(
                ExpectedRevision::Empty,
                "directory",
                state_with_directory(),
            ))
            .await
            .expect("a directory publishes");
        assert_eq!(store.count("axond_cp_principal").await, 4);

        store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(first.id),
                "no directory",
                state(),
                75,
            ))
            .await
            .expect("a revision declaring no identity publishes");
        assert_eq!(
            store.count("axond_cp_principal").await,
            0,
            "an undeclared principal is a revoked one, and all of them were undeclared"
        );
        assert_eq!(
            store.count("axond_cp_principal_role").await,
            0,
            "a grant does not outlive the principal that held it"
        );
        assert_eq!(
            store.column("SELECT slug FROM axond_cp_tenant").await,
            vec!["acme"],
            "revoking a directory is not deleting a tenant"
        );
    }

    /// A retired tenant's name is free, and a retained *project* name is a
    /// refusal rather than weather.
    ///
    /// The two halves are the same rule read from both sides. A tenant a revision
    /// stops declaring is retired, and the constraint that holds its name ignores
    /// retired rows, so the domain and the database agree that the name is
    /// reusable. A project has no lifecycle to retire into, so the row keeping its
    /// name is kept on purpose and every retry hits the same index: that failure
    /// is permanent by construction, and reported as an outage it would be an
    /// administrator retrying a publication that can never succeed.
    #[tokio::test]
    async fn a_retired_tenants_name_is_reusable_and_a_retained_projects_is_refused() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let first = store
            .publish_revision(candidate(ExpectedRevision::Empty, "acme", state()))
            .await
            .expect("the first tenant publishes");

        // The project side first, since it is the half that has no lifecycle to
        // retire into: the row holding `(tenant, core)` is kept whether or not the
        // revision still declares it, so a *different* project of that tenant
        // taking the name is refused however often it is retried.
        let mut reused_project = DesiredState::new();
        reused_project
            .insert(tenant(1, "acme"))
            .and_then(|state| state.insert(project(&tenant_id(1), 5, "core")))
            .expect("a tenant and one project of it is valid");
        let error = store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(first.id),
                "project reuse",
                reused_project,
                77,
            ))
            .await
            .expect_err("a name a projected project row holds cannot be taken");
        assert_eq!(
            error.category(),
            FailureCategory::Conflict,
            "a taken name is a conflict the caller resolves by renaming: {error}"
        );
        assert!(
            !error.retryable(),
            "replaying the same name can never succeed: {error}"
        );
        assert!(
            matches!(error, ControlPlaneError::NameTaken { .. }),
            "{error}"
        );
        assert!(error.to_string().contains("core"), "{error}");
        assert_eq!(
            store.desired_revision().await.expect("head"),
            Some(first.id),
            "a refused publication is not a published one"
        );

        // The tenant side: a different tenant, by id, taking the name the first one
        // used. The first tenant is no longer declared, so it is retired, and its
        // name goes with the retirement even though its row stays for the history
        // that points at it.
        let mut reused = DesiredState::new();
        reused
            .insert(tenant(21, "acme"))
            .expect("a state declaring one tenant is valid");
        store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(first.id),
                "reuse",
                reused,
                79,
            ))
            .await
            .expect("a retired tenant's name is not held against its successor");
        assert_eq!(
            store
                .column(
                    "SELECT lifecycle FROM axond_cp_tenant WHERE slug = 'acme' \
                     ORDER BY tenant_id"
                )
                .await,
            vec!["deleted", "active"],
            "the name is held by the declared tenant, and kept by the retired row"
        );
        assert_eq!(
            store.count("axond_cp_tenant").await,
            2,
            "the retired tenant's row is retained, not deleted"
        );
    }

    /// A project's name is reclaimed by renaming the project that holds it, which
    /// is the release path a project has instead of a tenant's lifecycle.
    ///
    /// A tenant releases its name by publishing itself `deleted`; a project has no
    /// lifecycle to publish, so a dropped project's name would otherwise be taken
    /// for the life of the deployment. Renaming it while it is still declared —
    /// uniqueness is judged on the declared state — hands the name back in the
    /// same revision that stops using it. Giving projects a lifecycle of their own
    /// is a tenancy-contract change, and follow-up work rather than this slice's.
    #[tokio::test]
    async fn a_projects_name_is_reclaimed_by_renaming_the_project_that_holds_it() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let tenant = tenant_id(1);
        let mut before = state();
        before
            .insert(project(&tenant, 3, "edge"))
            .expect("a tenant with a project");
        let first = store
            .publish_revision(candidate(ExpectedRevision::Empty, "acme", before.clone()))
            .await
            .expect("the original project publishes");

        // One revision: the project holding the name is renamed, and a new project
        // takes the name it gave up.
        let mut reclaimed = before;
        reclaimed
            .supersede(project_body(3, 1, "Edge Retired").version_at(
                Slug::parse("edge-retired").expect("a slug"),
                ResourceVersionNumber::FIRST.next(),
            ))
            .and_then(|state| state.insert(project(&tenant, 4, "edge")))
            .expect("renaming one project and naming another is valid");
        store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(first.id),
                "reclaim",
                reclaimed,
                78,
            ))
            .await
            .expect("a name its holder gave up in the same revision is free");

        assert_eq!(
            store
                .column(&format!(
                    "SELECT project_id FROM axond_cp_project WHERE slug = 'edge' \
                     AND tenant_id = '{tenant}'"
                ))
                .await,
            vec![project_id(4).to_string()],
            "the new project holds the name"
        );
    }

    /// A workload's change is recordable: the vocabulary 0001 wrote as an inline
    /// column check admitted three actor kinds, and this publishes the fourth
    /// through the journal's own writer rather than asserting on the schema.
    #[tokio::test]
    async fn a_revision_a_workload_published_is_attributed_to_its_principal() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let actor = Actor::Workload {
            tenant: tenant_id(1),
            principal: principal_id(32),
        };
        let mut candidate = candidate(ExpectedRevision::Empty, "by-workload", state());
        candidate.mutation.actor = actor.clone();
        candidate.audit.actor = actor.clone();
        let published = store
            .publish_revision(candidate)
            .await
            .expect("a workload's revision publishes");

        let trail = store.audit_trail(published.id).await.expect("audit trail");
        assert_eq!(trail.len(), 1);
        assert_eq!(trail[0].actor, actor);
        assert_eq!(
            store
                .column("SELECT actor_principal_id FROM axond_cp_mutation")
                .await,
            vec![principal_id(32).to_string()],
            "the mutation carries the workload it was made by"
        );
    }

    /// Two tenants — and two projects of one tenant — trading names in a single
    /// revision, which is valid desired state and must not depend on the order the
    /// projection happens to write its rows in.
    #[tokio::test]
    async fn two_owners_may_trade_names_in_one_revision() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let slug = |slug: &str| Slug::parse(slug).expect("a slug");
        let mut before = state();
        before
            .insert(tenant(11, "globex"))
            .and_then(|state| state.insert(project(&tenant_id(1), 3, "edge")))
            .expect("two tenants and two projects of one of them are valid");
        let first = store
            .publish_revision(candidate(ExpectedRevision::Empty, "named", before.clone()))
            .await
            .expect("the original names publish");

        // Each of the four rows takes the name another of them is giving up.
        let second = ResourceVersionNumber::FIRST.next();
        let mut swapped = before;
        swapped
            .supersede(tenant_body(1, "Acme").version_at(slug("globex"), second))
            .and_then(|state| {
                state.supersede(tenant_body(11, "Globex").version_at(slug("acme"), second))
            })
            .and_then(|state| {
                state.supersede(project_body(2, 1, "Core").version_at(slug("edge"), second))
            })
            .and_then(|state| {
                state.supersede(project_body(3, 1, "Edge").version_at(slug("core"), second))
            })
            .expect("a state in which names are exchanged is valid");
        store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(first.id),
                "swap",
                swapped,
                90,
            ))
            .await
            .expect("names that are exchanged rather than duplicated must publish");

        assert_eq!(
            store
                .column("SELECT slug FROM axond_cp_tenant ORDER BY tenant_id")
                .await,
            vec!["globex", "acme"],
            "each tenant holds the name the other gave up"
        );
        assert_eq!(
            store
                .column("SELECT slug FROM axond_cp_project ORDER BY project_id")
                .await,
            vec!["edge", "core"]
        );
    }

    /// The rule an in-place upgrade is left with is the deferred one.
    ///
    /// An earlier 0002 held a tenant's name in a partial unique *index*, and an
    /// index cannot be deferred. A deployment that applied that version by hand and
    /// then applies this one would, if the index survived, keep refusing exactly the
    /// swap above — permanently, because the constraint the projection defers is
    /// checked beside an index that is not. So the migration drops it, and this
    /// asserts the drop by recreating the old index first.
    #[tokio::test]
    async fn an_upgraded_deployment_keeps_no_rule_the_projection_cannot_defer() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        store
            .attempt(
                "CREATE UNIQUE INDEX axond_cp_tenant_slug_idx ON axond_cp_tenant (slug) \
                 WHERE lifecycle <> 'deleted'",
            )
            .await
            .expect("the rule 0002 left behind");

        let forward = schema::MIGRATIONS
            .iter()
            .find(|migration| migration.name.contains("tenancy_constraints"))
            .expect("the forward tenancy migration ships");
        store
            .attempt(forward.sql)
            .await
            .expect("applying the forward migration must not fail");
        assert_eq!(
            store
                .column(
                    "SELECT indexname::text FROM pg_indexes \
                     WHERE schemaname = current_schema() \
                     AND indexname = 'axond_cp_tenant_slug_idx'"
                )
                .await,
            Vec::<String>::new(),
            "an index the projection cannot defer outlived the migration that replaced it"
        );

        // And the swap the deferred constraint exists to allow still publishes.
        let slug = |slug: &str| Slug::parse(slug).expect("a slug");
        let mut before = state();
        before
            .insert(tenant(11, "globex"))
            .expect("two tenants are valid");
        let first = store
            .publish_revision(candidate(ExpectedRevision::Empty, "named", before.clone()))
            .await
            .expect("the original names publish");
        let second = ResourceVersionNumber::FIRST.next();
        let mut swapped = before;
        swapped
            .supersede(tenant_body(1, "Acme").version_at(slug("globex"), second))
            .and_then(|state| {
                state.supersede(tenant_body(11, "Globex").version_at(slug("acme"), second))
            })
            .expect("a state in which names are exchanged is valid");
        store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(first.id),
                "swap",
                swapped,
                91,
            ))
            .await
            .expect("an upgraded deployment must accept a traded name too");
    }

    /// Two administrators exchanging sign-ins, and two workloads exchanging keys,
    /// in one revision. Like a traded name this is valid desired state — the
    /// directory refuses a *shared* identity, not a reassigned one — so it must not
    /// depend on which `PrincipalId` the projection writes first.
    #[tokio::test]
    async fn two_principals_may_trade_identities_in_one_revision() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let tenant = tenant_id(1);
        let identity = |seed: u64, slug: &str, credential: Credential, version| {
            IdentityBody::new(
                principal_id(seed),
                DisplayName::parse(slug).expect("a display name"),
                credential,
                [Role::TenantAdmin],
            )
            .expect("an identity granting a role")
            .version_at(
                ResourceScope::Tenant(tenant),
                Slug::parse(slug).expect("a slug"),
                version,
            )
        };
        let oidc = |subject: &str| Credential::Oidc {
            issuer: "https://idp.example".to_owned(),
            subject: subject.to_owned(),
        };
        let key = |seed: u8| Credential::MintedKey {
            digest: Some(Checksum::of(workload_key(seed).as_bytes())),
        };
        let first_version = ResourceVersionNumber::FIRST;
        let mut before = state();
        before
            .insert(identity(40, "ada", oidc("ada"), first_version))
            .and_then(|state| state.insert(identity(41, "grace", oidc("grace"), first_version)))
            .and_then(|state| state.insert(identity(42, "builder", key(0xa1), first_version)))
            .and_then(|state| state.insert(identity(43, "deployer", key(0xa2), first_version)))
            .expect("two humans and two workloads of one tenant are valid");
        let first = store
            .publish_revision(candidate(
                ExpectedRevision::Empty,
                "directory",
                before.clone(),
            ))
            .await
            .expect("the original directory publishes");

        // Each principal keeps its id, its name, and its roles, and takes the
        // sign-in or the key the other is giving up.
        let second = first_version.next();
        let mut swapped = before;
        swapped
            .supersede(identity(40, "ada", oidc("grace"), second))
            .and_then(|state| state.supersede(identity(41, "grace", oidc("ada"), second)))
            .and_then(|state| state.supersede(identity(42, "builder", key(0xa2), second)))
            .and_then(|state| state.supersede(identity(43, "deployer", key(0xa1), second)))
            .expect("a state in which identities are exchanged is valid");
        store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(first.id),
                "reassign",
                swapped,
                91,
            ))
            .await
            .expect("identities that are exchanged rather than shared must publish");

        assert_eq!(
            store
                .column(
                    "SELECT subject FROM axond_cp_principal WHERE subject IS NOT NULL \
                     ORDER BY principal_id"
                )
                .await,
            vec!["grace", "ada"],
            "each administrator signs in as the subject the other gave up"
        );
        assert_eq!(
            store
                .column(
                    "SELECT key_digest FROM axond_cp_principal WHERE key_digest IS NOT NULL \
                     ORDER BY principal_id"
                )
                .await,
            vec![
                Checksum::of(workload_key(0xa2).as_bytes()).to_string(),
                Checksum::of(workload_key(0xa1).as_bytes()).to_string(),
            ],
            "and each workload authenticates with the key the other gave up"
        );
    }

    /// A project moving to another tenant, with the principal scoped into it. Every
    /// intermediate row set is inconsistent — the project belongs to one tenant and
    /// the principal claims the other — and the state as declared is not, so the
    /// composite ownership key is settled with the names rather than row by row.
    #[tokio::test]
    async fn a_project_may_move_to_another_tenant_with_what_is_scoped_into_it() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let scoped = |tenant: u64, version| {
            IdentityBody::new(
                principal_id(44),
                DisplayName::parse("Builder").expect("a display name"),
                Credential::Oidc {
                    issuer: "https://idp.example".to_owned(),
                    subject: "builder".to_owned(),
                },
                [Role::Developer],
            )
            .expect("an identity granting a role")
            .version_at(
                ResourceScope::Project {
                    tenant: tenant_id(tenant),
                    project: project_id(3),
                },
                Slug::parse("builder").expect("a slug"),
                version,
            )
        };
        let first_version = ResourceVersionNumber::FIRST;
        let mut before = state();
        before
            .insert(tenant(11, "globex"))
            .and_then(|state| state.insert(project(&tenant_id(1), 3, "edge")))
            .and_then(|state| state.insert(scoped(1, first_version)))
            .expect("two tenants, a project of one, and a developer in that project");
        let first = store
            .publish_revision(candidate(ExpectedRevision::Empty, "owned", before.clone()))
            .await
            .expect("the original ownership publishes");

        let second = first_version.next();
        let mut moved = before;
        moved
            .supersede(
                project_body(3, 11, "Edge")
                    .version_at(Slug::parse("edge").expect("a slug"), second),
            )
            .and_then(|state| state.supersede(scoped(11, second)))
            .expect("a project and its developer moving together is valid");
        store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(first.id),
                "move",
                moved,
                92,
            ))
            .await
            .expect("a reassignment the state declares consistently must publish");

        assert_eq!(
            store
                .column(&format!(
                    "SELECT tenant_id FROM axond_cp_project WHERE project_id = '{}'",
                    project_id(3)
                ))
                .await,
            vec![tenant_id(11).to_string()]
        );
        assert_eq!(
            store
                .column(&format!(
                    "SELECT tenant_id FROM axond_cp_principal WHERE principal_id = '{}'",
                    principal_id(44)
                ))
                .await,
            vec![tenant_id(11).to_string()],
            "the principal moved with the project it is scoped into"
        );
    }

    /// Disabling and then deleting a tenant is a transition on the projected row,
    /// and neither transition takes the tenant's history with it.
    #[tokio::test]
    async fn a_tenant_lifecycle_transition_is_a_row_update_and_never_a_delete() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let mut previous = store
            .publish_revision(candidate(ExpectedRevision::Empty, "active", state()))
            .await
            .expect("an active tenant publishes")
            .id;
        for (index, lifecycle) in [TenantLifecycle::Disabled, TenantLifecycle::Deleted]
            .into_iter()
            .enumerate()
        {
            let mut state = state();
            let mut version = ResourceVersionNumber::FIRST.next();
            for _ in 0..index {
                version = version.next();
            }
            state
                .supersede(
                    tenant_body(1, "Acme")
                        .in_lifecycle(lifecycle)
                        .version_at(Slug::parse("acme").expect("a slug"), version),
                )
                .expect("a later version of the same tenant");
            previous = store
                .publish_revision(candidate_with_mutation(
                    ExpectedRevision::Exactly(previous),
                    lifecycle.as_str(),
                    state,
                    80 + u64::try_from(index).expect("two transitions") * 2,
                ))
                .await
                .expect("a lifecycle transition publishes")
                .id;
            assert_eq!(
                store.column("SELECT lifecycle FROM axond_cp_tenant").await,
                vec![lifecycle.as_str()]
            );
            assert_eq!(
                store.column("SELECT slug FROM axond_cp_project").await,
                vec!["core"],
                "a tenant's projects are not erased by its lifecycle"
            );
        }
        assert!(
            store.count("axond_cp_audit_event").await >= 3,
            "the trail of a deleted tenant is retained"
        );
    }

    /// Retirement is what releases a name: the slug index is partial over live
    /// rows, so a tenant that is deleted — declared so, or simply left out of the
    /// revision — frees its name for the next tenant that asks.
    ///
    /// The three moves an operator actually makes, in order: delete, reuse,
    /// restore. The last one succeeds, and it is worth being explicit about why:
    /// the revision that restores the first tenant does not declare the tenant
    /// that took its name, so that tenant is retired in the same transaction and
    /// the name is free again. Publishing complete desired state is the whole
    /// contract, and this is what it costs when a name is recycled.
    #[tokio::test]
    async fn a_retired_tenants_slug_is_reusable_and_a_restore_retires_who_took_it() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let acme = Slug::parse("acme").expect("a slug");
        let mut head = store
            .publish_revision(candidate(ExpectedRevision::Empty, "acme", state()))
            .await
            .expect("an active tenant publishes")
            .id;

        // Deleted: the row stays, because history points at it, but the name it
        // held is released by the partial index.
        let mut deleted = state();
        deleted
            .supersede(
                tenant_body(1, "Acme")
                    .in_lifecycle(TenantLifecycle::Deleted)
                    .version_at(acme.clone(), ResourceVersionNumber::FIRST.next()),
            )
            .expect("a later version of the same tenant");
        head = store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(head),
                "delete",
                deleted,
                90,
            ))
            .await
            .expect("a deletion publishes")
            .id;

        // A different tenant takes the released name. Desired state refuses two
        // live rows sharing a slug, so the deleted one is simply no longer
        // declared: its row is retained, and holds nothing.
        let mut reused = DesiredState::new();
        reused
            .insert(tenant(21, "acme"))
            .expect("a state declaring one tenant is valid");
        head = store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(head),
                "reuse",
                reused,
                94,
            ))
            .await
            .expect("a deleted tenant's slug is free for another tenant")
            .id;
        assert_eq!(
            store
                .column("SELECT lifecycle FROM axond_cp_tenant ORDER BY slug, lifecycle")
                .await,
            vec!["active", "deleted"],
            "both rows are retained; only one is live under the name"
        );

        // And restoring the first tenant under the name it released is a revision
        // that declares it and nothing else: the tenant holding the name is
        // undeclared, so it is retired before the restore is projected, and the
        // name is transferred rather than duplicated or refused.
        let mut restored = DesiredState::new();
        restored
            .insert(
                tenant_body(1, "Acme").version_at(acme, ResourceVersionNumber::FIRST.next().next()),
            )
            .expect("a state declaring one tenant is valid");
        store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(head),
                "restore",
                restored,
                98,
            ))
            .await
            .expect("the revision that restores the tenant retires the one that took its name");
        assert_eq!(
            store
                .column(
                    "SELECT lifecycle FROM axond_cp_tenant WHERE slug = 'acme' \
                     ORDER BY lifecycle"
                )
                .await,
            vec!["active", "deleted"],
            "one live tenant holds the name; the retired row keeps the name it held"
        );
        assert_eq!(
            store
                .column("SELECT tenant_id FROM axond_cp_tenant WHERE lifecycle = 'active'")
                .await
                .len(),
            1,
            "exactly one tenant is live, and the restore is it"
        );
    }

    /// A revision is the whole desired state, so the tenant it stops declaring is
    /// retired rather than left reading `active`.
    ///
    /// The projected `lifecycle` is what a billing, retention, or admission reader
    /// asks instead of decoding a body, and the domain serves an undeclared tenant
    /// nothing. A retained row still claiming `active` would answer that question
    /// with the opposite of the published truth. History is the reason the row
    /// stays at all, so this also asserts the projects, mutations, and audit
    /// events of the retired tenant survive the transition.
    #[tokio::test]
    async fn a_tenant_a_revision_stops_declaring_is_retired_and_keeps_its_history() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let both = store
            .publish_revision(candidate(
                ExpectedRevision::Empty,
                "two tenants",
                state_with_second_tenant(),
            ))
            .await
            .expect("two tenants publish");
        assert_eq!(
            store
                .column("SELECT lifecycle FROM axond_cp_tenant ORDER BY tenant_id")
                .await,
            vec!["active", "active"]
        );

        // The second tenant is simply absent from the next revision, which is how
        // a complete desired state spells "this tenant is gone" without an
        // explicit transition.
        let dropped = store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(both.id),
                "one tenant",
                state(),
                120,
            ))
            .await
            .expect("a revision declaring one of them publishes");

        assert_eq!(
            store
                .column("SELECT lifecycle FROM axond_cp_tenant ORDER BY tenant_id")
                .await,
            vec!["active", "deleted"],
            "an undeclared tenant is still recorded as servable"
        );
        let published = store
            .load_revision(dropped.id)
            .await
            .expect("the revision it just published hydrates");
        let tenancy = Tenancy::of(published.state()).expect("the published tenancy resolves");
        assert!(
            !tenancy.is_served(tenant_id(11)) && tenancy.tenant(tenant_id(11)).is_none(),
            "the projection and the published snapshot disagree about who is served"
        );
        assert!(
            tenancy.is_served(tenant_id(1)),
            "retiring one tenant retired another"
        );

        // Retired, not erased: the rows history points at are all still here.
        assert_eq!(
            store
                .column("SELECT slug FROM axond_cp_project ORDER BY slug")
                .await,
            vec!["core"],
            "the surviving tenant's project is untouched"
        );
        assert!(
            store.count("axond_cp_mutation").await >= 2
                && store.count("axond_cp_audit_event").await >= 2,
            "the retired tenant's history is retained"
        );

        // And declaring it again brings the row back to what that revision says,
        // so retirement is a projection of the published state rather than a
        // one-way door.
        store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(dropped.id),
                "two tenants again",
                state_with_second_tenant(),
                124,
            ))
            .await
            .expect("re-declaring the tenant publishes");
        assert_eq!(
            store
                .column("SELECT lifecycle FROM axond_cp_tenant ORDER BY tenant_id")
                .await,
            vec!["active", "active"],
            "a re-declared tenant is still retired"
        );
    }

    /// A revision that says nothing about tenancy retires nothing.
    ///
    /// Two publications look identical to the projection and mean the opposite of
    /// each other: "these are the tenants, and only these" and "this build had no
    /// tenant resource". The second is what every pre-tenancy revision in an
    /// upgraded deployment's journal looks like, and a rollback to one republishes
    /// it — so if omission retired tenants unconditionally, rolling back would
    /// delete every tenant in the deployment without any revision saying so.
    ///
    /// Retiring the last tenant is therefore an explicit lifecycle, and this pins
    /// both halves: a snapshot with no tenancy leaves the projection alone, and a
    /// snapshot that does declare tenancy still retires what it omits.
    #[tokio::test]
    async fn a_revision_that_declares_no_tenancy_retires_no_tenant() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let both = store
            .publish_revision(candidate(
                ExpectedRevision::Empty,
                "two tenants",
                state_with_second_tenant(),
            ))
            .await
            .expect("two tenants publish");

        // The state an older build published, republished — a rollback.
        let rolled_back = store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(both.id),
                "pre-tenancy",
                state_a_pre_tenancy_build_published(),
                130,
            ))
            .await
            .expect("a pre-tenancy revision publishes on a tenancy-aware build");
        assert_eq!(
            store
                .column("SELECT lifecycle FROM axond_cp_tenant ORDER BY tenant_id")
                .await,
            vec!["active", "active"],
            "a rollback to a pre-tenancy revision deleted the deployment's tenants"
        );
        assert_eq!(
            store
                .column("SELECT slug FROM axond_cp_project ORDER BY slug")
                .await,
            vec!["core"],
            "and their projects"
        );

        // And the conditional is on tenancy being *declared*, not on the
        // publication being a rollback: the next revision names one tenant, so it
        // is authoritative and the other is retired.
        store
            .publish_revision(candidate_with_mutation(
                ExpectedRevision::Exactly(rolled_back.id),
                "one tenant",
                state(),
                131,
            ))
            .await
            .expect("a revision declaring one tenant publishes");
        assert_eq!(
            store
                .column("SELECT lifecycle FROM axond_cp_tenant ORDER BY tenant_id")
                .await,
            vec!["active", "deleted"],
            "an authoritative snapshot stopped reconciling what it omits"
        );
    }

    /// The half of #144 that is a database property rather than a service one: a
    /// row naming another tenant's project, or a tenant nothing declared, is
    /// unwritable even by a caller that skipped every validation above.
    #[tokio::test]
    async fn a_projected_row_cannot_name_an_absent_tenant_or_another_tenants_project() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        store
            .publish_revision(candidate(
                ExpectedRevision::Empty,
                "directory",
                state_with_directory(),
            ))
            .await
            .expect("a revision declaring a directory publishes");
        let tenant = tenant_id(1);
        let project = project_id(2);
        let stranger = tenant_id(11);

        // A project of a tenant that does not exist.
        let refusal = store
            .attempt(&format!(
                "INSERT INTO axond_cp_project (project_id, tenant_id, slug, revision_id) \
                 VALUES ('{}', '{stranger}', 'smuggled', 'rev_x')",
                project_id(12)
            ))
            .await
            .expect_err("a project needs an existing tenant");
        assert!(refusal.to_string().contains("foreign key"), "{refusal}");

        // A principal scoped into a project owned by another tenant: the
        // composite key is what makes this a contradiction rather than a pair of
        // separately valid ids.
        let refusal = store
            .attempt(&format!(
                "INSERT INTO axond_cp_principal (principal_id, resource_id, identity_kind, \
                 scope_kind, tenant_id, project_id, slug, display_name, key_digest, revision_id) \
                 VALUES ('{}', 'res_x', 'workload', 'project', '{stranger}', '{project}', \
                 'confused', 'Confused', {DIGEST}, 'rev_x')",
                PrincipalId::new(uuid(44))
            ))
            .await
            .expect_err("a project belongs to exactly one tenant");
        assert!(refusal.to_string().contains("foreign key"), "{refusal}");

        // A human with a key digest, and a workload with an issuer: the two
        // credential shapes are not mixable, whatever the service layer believes.
        for (kind, columns, values) in [
            (
                "human",
                "issuer, subject, key_digest",
                format!("'https://idp.example', 'smuggled', {DIGEST}"),
            ),
            (
                "workload",
                "issuer, subject",
                "'https://idp.example', 's'".to_owned(),
            ),
        ] {
            let refusal = store
                .attempt(&format!(
                    "INSERT INTO axond_cp_principal (principal_id, resource_id, identity_kind, \
                     scope_kind, tenant_id, slug, display_name, {columns}, revision_id) \
                     VALUES ('{}', 'res_x', '{kind}', 'tenant', '{tenant}', 'mixed', 'Mixed', \
                     {values}, 'rev_x')",
                    PrincipalId::new(uuid(45))
                ))
                .await
                .expect_err("one identity kind, one credential shape");
            assert!(refusal.to_string().contains("check"), "{refusal}");
        }

        // And a role outside the vocabulary this build reads.
        let refusal = store
            .attempt(&format!(
                "INSERT INTO axond_cp_principal_role (principal_id, role) VALUES ('{}', 'root')",
                PrincipalId::new(uuid(33))
            ))
            .await
            .expect_err("the role vocabulary is closed");
        assert!(refusal.to_string().contains("check"), "{refusal}");
    }

    /// The other half of that property, on the tables that hold history.
    ///
    /// 0002 left the journal out because a revision an older build published has
    /// tenant-scoped rows and no owner rows, and a key would have refused to
    /// republish it. 0004 adds the keys `NOT VALID`, which is the honest statement
    /// of what they cover: every row written from now on, and none of the ones that
    /// predate them. This asserts both halves — the keys are there and unvalidated,
    /// and a new row naming a tenant this deployment has no row for is refused by
    /// the database rather than by a service-layer check somebody can skip.
    #[tokio::test]
    async fn a_journal_row_cannot_name_a_tenant_this_deployment_has_no_row_for() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        store
            .publish_revision(candidate(
                ExpectedRevision::Empty,
                "directory",
                state_with_directory(),
            ))
            .await
            .expect("a revision declaring a directory publishes");
        assert_eq!(
            store
                .column(
                    "SELECT conname::text FROM pg_constraint \
                     WHERE contype = 'f' AND NOT convalidated \
                     AND connamespace = current_schema()::regnamespace \
                     ORDER BY conname"
                )
                .await,
            vec![
                "axond_cp_audit_event_actor_tenant_fkey",
                "axond_cp_mutation_actor_tenant_fkey",
                "axond_cp_mutation_tenant_fkey",
                "axond_cp_resource_version_tenant_fkey",
            ],
            "the unvalidated keys are the journal's, and validation is not pending on others"
        );

        let stranger = tenant_id(11);
        let refusal = store
            .attempt(&format!(
                "INSERT INTO axond_cp_resource_version \
                 (resource_kind, resource_id, version, scope_kind, tenant_id, slug, body_form, \
                 body_inline, content_checksum, serializer) \
                 VALUES ('alias', '{}', 1, 'tenant', '{stranger}', 'smuggled', 'inline', \
                 '\\x7b7d'::bytea, {DIGEST}, 'json')",
                crate::desired_state::ResourceId::new(uuid(46))
            ))
            .await
            .expect_err("a stored version needs a tenant this deployment has a row for");
        assert!(
            refusal
                .to_string()
                .contains("axond_cp_resource_version_tenant_fkey"),
            "{refusal}"
        );

        // And the attribution half: a change *by* a tenant nothing declares is
        // attribution nobody can follow up on, so it is not recordable either.
        let refusal = store
            .attempt(&format!(
                "INSERT INTO axond_cp_mutation \
                 (mutation_id, actor_kind, actor_tenant_id, actor_principal_id, mutation_kind, \
                 scope_kind, tenant_id, idempotency_key, submitted_at) \
                 VALUES ('{}', 'workload', '{stranger}', '{}', 'publish', 'tenant', '{}', \
                 'smuggled', now())",
                MutationId::new(uuid(47)),
                PrincipalId::new(uuid(48)),
                tenant_id(1)
            ))
            .await
            .expect_err("attribution names a tenant that exists");
        assert!(
            refusal.to_string().contains("actor_tenant_fkey"),
            "{refusal}"
        );
    }

    /// Which is republishable because the owner is recorded, not invented.
    ///
    /// A revision a pre-tenancy build published declares no tenant and still
    /// carries tenant-scoped resources. The projection records the owner those rows
    /// name at `lifecycle = 'deleted'` — referenced by history, declared by nothing
    /// — so the key is satisfied without a live tenant appearing that no revision
    /// granted. Declaring it later is what makes it a tenant; rolling back to the
    /// pre-tenancy revision afterwards demotes nobody.
    #[tokio::test]
    async fn a_revision_naming_an_undeclared_tenant_records_that_owner_as_deleted() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let owner = tenant_id(1);
        let first = store
            .publish_revision(candidate(
                ExpectedRevision::Empty,
                "pre-tenancy",
                state_a_pre_tenancy_build_published(),
            ))
            .await
            .expect("a revision without owner rows publishes");
        assert_eq!(
            store
                .column("SELECT lifecycle FROM axond_cp_tenant ORDER BY tenant_id")
                .await,
            vec!["deleted"],
            "an owner history names is not a tenant anybody granted"
        );
        assert_eq!(
            store
                .column("SELECT slug FROM axond_cp_tenant ORDER BY tenant_id")
                .await,
            vec![owner.to_string()],
            "a recorded owner holds no name a declared tenant could want, being deleted"
        );

        let second = store
            .publish_revision(candidate(
                ExpectedRevision::Exactly(first.id),
                "declared",
                state(),
            ))
            .await
            .expect("declaring the tenant the history named publishes");
        assert_eq!(
            store
                .column("SELECT lifecycle || ' ' || slug FROM axond_cp_tenant")
                .await,
            vec!["active acme"],
            "declaring a recorded owner is what turns it into a tenant"
        );

        store
            .publish_revision(candidate(
                ExpectedRevision::Exactly(second.id),
                "rolled back",
                state_a_pre_tenancy_build_published(),
            ))
            .await
            .expect("a rollback to the pre-tenancy revision publishes");
        assert_eq!(
            store
                .column("SELECT lifecycle || ' ' || slug FROM axond_cp_tenant")
                .await,
            vec!["active acme"],
            "recording an owner rewrote a live tenant, which a rollback must never do"
        );
    }

    /// Recording history's owners must not make an *administrator* in an undeclared
    /// tenant storable.
    ///
    /// The owner rows exist for the journal, and the principal table has an
    /// ownership key of its own that refuses a principal scoped into a tenant no
    /// revision declares — the database-held half of the isolation property. Record
    /// the owners before the principals and that key finds its row already there,
    /// leaving only the service-layer check. So the recording runs last, and this
    /// pins it: the publication is refused, and nothing of it is stored.
    #[tokio::test]
    async fn recording_historys_owners_does_not_make_an_undeclared_tenants_admin_storable() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let undeclared = tenant_id(1);
        let mut state = state_a_pre_tenancy_build_published();
        state
            .insert(human(
                41,
                "ghost-admin",
                ResourceScope::Tenant(undeclared),
                &[Role::TenantAdmin],
            ))
            .expect("an identity in an undeclared tenant is well-formed desired state");
        let refusal = store
            .publish_revision(candidate(ExpectedRevision::Empty, "ghost admin", state))
            .await
            .expect_err("an administrator in a tenant no revision declares is refused");
        assert!(
            matches!(refusal, ControlPlaneError::Invalid(_)),
            "and refused as damaged state rather than as an outage: {refusal:?}"
        );
        assert!(
            store
                .column("SELECT tenant_id FROM axond_cp_tenant")
                .await
                .is_empty(),
            "a refused publication stores nothing, owner rows included"
        );
    }

    /// A publication cannot widen the set of tenants the owner rows cover by way of
    /// its own attribution.
    ///
    /// The recording step covers the state's resource scopes, because a resource is
    /// history a rollback has to be able to republish. A mutation's scope and an
    /// actor's tenant are not history — they describe a change being made now — so
    /// they get no owner row, and 0004's keys refuse them. Were it otherwise, a
    /// publication could name any tenant identifier in its attribution and leave a
    /// row behind that the principal key would accept for every later revision.
    #[tokio::test]
    async fn a_publication_cannot_leave_an_owner_row_behind_by_attributing_itself_to_one() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let stranger = tenant_id(11);
        let mut candidate = candidate(
            ExpectedRevision::Empty,
            "borrowed attribution",
            state_with_directory(),
        );
        candidate.mutation.actor = Actor::Workload {
            tenant: stranger,
            principal: principal_id(49),
        };
        candidate.audit.actor = candidate.mutation.actor.clone();
        let refusal = store.publish_revision(candidate).await.expect_err(
            "a change attributed to a tenant this deployment has no row for is refused",
        );
        assert!(
            matches!(refusal, ControlPlaneError::Denied { .. }),
            "and refused permanently rather than retried as an outage: {refusal:?}"
        );
        assert!(
            !store
                .column("SELECT tenant_id FROM axond_cp_tenant")
                .await
                .contains(&stranger.to_string()),
            "and leaves no owner row for the tenant it named"
        );
    }

    /// A refusal names the rule, and never the row it collided with.
    ///
    /// The identity constraints carry sign-ins and key digests, and PostgreSQL's
    /// `DETAIL` on a violation quotes the colliding key — `Key (key_digest)=(…)`.
    /// Splicing that into the error would make a failed publication an oracle for
    /// another principal's OIDC subject or the digest of another workload's minted
    /// key, which is the disclosure `Denial::public_reason` prevents one layer up.
    /// The caller is told which rule refused the revision; the values stay in the
    /// operator's log.
    #[test]
    fn a_projection_refusal_names_the_constraint_and_not_the_row_it_hit() {
        let message = projection_refusal(
            "principal",
            "",
            false,
            Some("axond_cp_principal_key_digest_unique"),
        );
        assert!(
            message.contains("axond_cp_principal_key_digest_unique"),
            "the refusal names the rule it broke: {message}"
        );
        assert!(
            message.contains("no retry clears this"),
            "and that it is not an outage: {message}"
        );
        for leaked in ["sha256:", "Key (", "issuer", "subject", "=("] {
            assert!(
                !message.contains(leaked),
                "the refusal could carry the row it collided with ({leaked}): {message}"
            );
        }
        assert!(
            projection_refusal(
                "project",
                "edge",
                true,
                Some("axond_cp_principal_project_fkey")
            )
            .contains("contradicts an ownership row"),
            "an ownership contradiction is described as one"
        );
    }

    #[test]
    fn observation_integrity_sqlstates_are_permanent() {
        for code in [
            &SqlState::FOREIGN_KEY_VIOLATION,
            &SqlState::UNIQUE_VIOLATION,
            &SqlState::CHECK_VIOLATION,
            &SqlState::NOT_NULL_VIOLATION,
            &SqlState::EXCLUSION_VIOLATION,
        ] {
            assert!(
                is_permanent_observation_sqlstate(code),
                "{code:?} must not be classified as an outage"
            );
        }
        assert!(!is_permanent_observation_sqlstate(
            &SqlState::CONNECTION_FAILURE
        ));
        assert!(!is_permanent_observation_sqlstate(
            &SqlState::T_R_SERIALIZATION_FAILURE
        ));
    }

    /// A journal row refused by 0004's ownership key is a refusal, not weather —
    /// nothing clears it by waiting — and it names the key rather than the tenant
    /// the key refused.
    #[test]
    fn a_journal_ownership_refusal_names_the_key_and_not_the_tenant_it_refused() {
        let message = journal_ownership_refusal(Some("axond_cp_mutation_actor_tenant_fkey"));
        assert!(
            message.contains("axond_cp_mutation_actor_tenant_fkey"),
            "the refusal names the rule it broke: {message}"
        );
        assert!(
            message.contains("publish a revision that declares it"),
            "and the remedy an operator has: {message}"
        );
        for leaked in ["Key (", "=(", "tenant_id)=", "sha256:"] {
            assert!(
                !message.contains(leaked),
                "the refusal could carry the key it refused ({leaked}): {message}"
            );
        }
        assert!(
            !journal_ownership_refusal(None).is_empty(),
            "an unnamed constraint still says what happened"
        );
    }

    /// Denied actions are the half of the trail successful revisions cannot hold,
    /// and they are read per tenant so one tenant's refusals are not another's
    /// reconnaissance.
    #[tokio::test]
    async fn denied_actions_are_recorded_and_read_back_per_tenant_newest_first() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        let tenant = tenant_id(1);
        let other = tenant_id(11);
        let mine = [
            denial(90, ResourceScope::Tenant(tenant), DenialReason::OutOfScope),
            denial(
                91,
                ResourceScope::Project {
                    tenant,
                    project: project_id(2),
                },
                DenialReason::RoleLacksAction,
            ),
        ];
        let theirs = denial(92, ResourceScope::Tenant(other), DenialReason::CrossTenant);
        let deployment = denial(
            93,
            ResourceScope::Deployment,
            DenialReason::TenantNotAdministrable,
        );
        for denial in mine.iter().chain([&theirs, &deployment]) {
            store.record_denial(denial).await.expect("record a refusal");
        }
        // Recording is idempotent on the denial's own id, so a retry after a
        // timeout does not double-count an attempt.
        store
            .record_denial(&mine[0])
            .await
            .expect("a retry is not a second attempt");

        let read = store
            .denials(&DenialPage::for_scope(Some(tenant)), 10)
            .await
            .expect("read");
        assert_eq!(read, vec![mine[1].clone(), mine[0].clone()]);
        assert_eq!(
            store
                .denials(&DenialPage::for_scope(Some(other)), 10)
                .await
                .expect("read"),
            vec![theirs.clone()],
            "one tenant's refusals are not another's"
        );
        // `None` is the deployment-scoped page, not a wildcard.
        assert_eq!(
            store
                .denials(&DenialPage::for_scope(None), 10)
                .await
                .expect("read"),
            vec![deployment]
        );
        // A limit is a page size.
        assert_eq!(
            store
                .denials(&DenialPage::for_scope(Some(tenant)), 1)
                .await
                .expect("read"),
            vec![mine[1].clone()]
        );
        assert!(
            store
                .denials(&DenialPage::for_scope(Some(tenant)), 0)
                .await
                .expect("read")
                .len()
                == 1,
            "a zero limit is clamped to a page rather than refused"
        );
        // A workload's refusal is attributed to its principal, not to a person.
        let mut workload = denial(94, ResourceScope::Tenant(tenant), DenialReason::CrossTenant);
        workload.actor = Actor::Workload {
            tenant,
            principal: PrincipalId::new(uuid(33)),
        };
        store.record_denial(&workload).await.expect("record");
        assert_eq!(
            store
                .denials(&DenialPage::for_scope(Some(tenant)), 1)
                .await
                .expect("read")
                .first()
                .map(|denial| denial.actor.clone()),
            Some(workload.actor.clone())
        );

        // And a refusal *against* this tenant attempted by another tenant's
        // workload is on this tenant's page, actor included. It is the event the
        // trail exists for, and every other page excludes it by scope: dropping
        // it here as well would record a cross-tenant probe that no read returns.
        let mut intruder = denial(95, ResourceScope::Tenant(tenant), DenialReason::CrossTenant);
        intruder.actor = Actor::Workload {
            tenant: other,
            principal: PrincipalId::new(uuid(36)),
        };
        store.record_denial(&intruder).await.expect("record");
        assert!(
            store
                .denials(&DenialPage::for_scope(Some(tenant)), 10)
                .await
                .expect("read")
                .contains(&intruder),
            "a refusal against this tenant is unreadable by anyone"
        );
        assert!(
            !store
                .denials(&DenialPage::for_scope(Some(other)), 10)
                .await
                .expect("read")
                .contains(&intruder),
            "the attempting tenant's page is not where a refusal against another one belongs"
        );
        assert!(
            !store
                .denials(&DenialPage::for_scope(None), 10)
                .await
                .expect("read")
                .contains(&intruder),
            "the deployment page is the rows that named no tenant"
        );
    }

    /// Applying the wall twice is not a broken database.
    ///
    /// An operator upgrading a running deployment may apply 0002 by hand — the
    /// migrating role has to own these tables for `FORCE ROW LEVEL SECURITY`, and
    /// a deployment whose DDL role is not its application role runs it out of
    /// band — so a second application has to be a no-op rather than a collision
    /// on a constraint or a policy name that leaves the schema half-walled.
    #[tokio::test]
    async fn the_tenancy_migration_is_idempotent_when_an_operator_reapplies_it() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        // All of them, in order: 0002 declares the tenancy tables, 0003 replaces
        // the rules it got wrong, 0004 keys the journal and 0005 adds the
        // discovery evidence table, so a hand-run upgrade is the set rather than
        // any one of them.
        let upgrade: Vec<_> = schema::MIGRATIONS
            .iter()
            .filter(|migration| migration.version > 1)
            .collect();
        assert_eq!(upgrade.len(), 4, "the upgrade migrations ship");

        for migration in &upgrade {
            store
                .attempt(migration.sql)
                .await
                .expect("re-applying a tenancy migration must not fail");
        }

        // Applied twice, the wall is still forced rather than merely enabled: a
        // policy the owning role bypasses is the failure this asserts against.
        assert_eq!(
            store
                .column(
                    "SELECT relname::text FROM pg_class \
                     WHERE relnamespace = current_schema()::regnamespace \
                     AND relrowsecurity AND NOT relforcerowsecurity \
                     ORDER BY relname"
                )
                .await,
            Vec::<String>::new(),
            "a table left enabled but unforced by the second application"
        );
        // And publishing still works, so the second application did not leave a
        // constraint the projection now violates.
        store
            .publish_revision(candidate(
                ExpectedRevision::Empty,
                "after a re-apply",
                state_with_directory(),
            ))
            .await
            .expect("a revision publishes against a twice-migrated schema");
    }

    /// The documented boundary is the boundary the database has.
    ///
    /// 0002 and the operator runbook both enumerate the tables deliberately left
    /// outside the wall, and an operator auditing isolation reads that list rather
    /// than the policies. A list that named a table that does not exist, or missed
    /// one that does, would have them conclude a table is walled when a pinned
    /// session can read all of it.
    #[tokio::test]
    async fn the_tables_outside_the_wall_are_the_ones_the_migration_names() {
        let Some((store, _, _)) = journal().await else {
            return;
        };
        assert_eq!(
            store
                .column(
                    "SELECT relname::text FROM pg_class \
                     WHERE relnamespace = current_schema()::regnamespace \
                     AND relkind = 'r' AND NOT relrowsecurity \
                     ORDER BY relname"
                )
                .await,
            vec!["axond_cp_schema_migration".to_owned()],
            "the unwalled tables are not the ones 0002 and the runbook name"
        );
    }

    /// Row-level security, against a role that cannot bypass it: the wall behind
    /// the service layer, tested the only way it can be — as a session that is
    /// pinned to one tenant and still asks for everything.
    #[tokio::test]
    async fn a_session_pinned_to_one_tenant_reads_no_other_tenants_rows() {
        let Some((store, dsn, schema)) = journal().await else {
            return;
        };
        let first = store
            .publish_revision(candidate(
                ExpectedRevision::Empty,
                "two tenants",
                two_tenant_directory_state(),
            ))
            .await
            .expect("two tenants publish");
        // A second change, attributed to the *other* tenant, so the journal holds
        // one mutation and one audit event per tenant and "can this session read
        // the other tenant's changes?" has something to read.
        let mut theirs = candidate(
            ExpectedRevision::Exactly(first.id),
            "their change",
            two_tenant_directory_state(),
        );
        theirs.mutation.scope = ResourceScope::Tenant(tenant_id(11));
        store
            .publish_revision(theirs)
            .await
            .expect("the other tenant's change publishes");
        store
            .record_denial(&denial(
                95,
                ResourceScope::Tenant(tenant_id(11)),
                DenialReason::CrossTenant,
            ))
            .await
            .expect("record a refusal against the other tenant");
        // A refusal that names no tenant, attempted by the *other* tenant's
        // workload: shared in what was asked for, and not in who asked.
        let mut theirs = denial(96, ResourceScope::Deployment, DenialReason::OutOfScope);
        theirs.actor = Actor::Workload {
            tenant: tenant_id(11),
            principal: principal_id(36),
        };
        store
            .record_denial(&theirs)
            .await
            .expect("record a deployment-scoped refusal by another tenant's workload");

        // A login role with no bypass, granted only reads: the shape of a
        // read-only replica consumer or a reporting job.
        let role = format!("{schema}_reader");
        store
            .attempt(&format!(
                "CREATE ROLE {role} LOGIN PASSWORD 'reader'; \
                 GRANT USAGE ON SCHEMA {schema} TO {role}; \
                 GRANT SELECT ON ALL TABLES IN SCHEMA {schema} TO {role}"
            ))
            .await
            .expect("create the reading role");

        let mut config: Config = dsn.parse().expect("test dsn");
        config.user(&role).password("reader");
        config.connect_timeout(Duration::from_secs(5));
        let (client, connection) = config
            .connect(crate::usage::tls_connector())
            .await
            .expect("connect as the reading role");
        tokio::spawn(async move {
            let _ = connection.await;
        });
        client
            .batch_execute(&format!(
                "SET search_path TO {schema}; SET axond.tenant_id = '{}'",
                tenant_id(1)
            ))
            .await
            .expect("pin the session to one tenant");

        for (table, expected) in [
            ("axond_cp_tenant", vec![tenant_id(1).to_string()]),
            ("axond_cp_project", vec![tenant_id(1).to_string()]),
        ] {
            let rows: Vec<String> = client
                .query(&format!("SELECT tenant_id FROM {table}"), &[])
                .await
                .expect("read as the pinned session")
                .iter()
                .map(|row| row.get(0))
                .collect();
            assert_eq!(rows, expected, "{table} leaked another tenant's rows");
        }
        let denials: i64 = client
            .query_one("SELECT count(*) FROM axond_cp_access_denial", &[])
            .await
            .expect("read")
            .get(0);
        assert_eq!(
            denials, 0,
            "another tenant's refusals — including a deployment-scoped one its workload \
             attempted — are not visible"
        );

        // The other side of that rule, which is the whole point of the trail: a
        // refusal aimed at *this* tenant is readable by it even though another
        // tenant's workload attempted it. The wall filters the actor only where it
        // hands a session a row the session is not the subject of.
        let mut aimed_here = denial(
            97,
            ResourceScope::Tenant(tenant_id(1)),
            DenialReason::CrossTenant,
        );
        aimed_here.actor = Actor::Workload {
            tenant: tenant_id(11),
            principal: principal_id(37),
        };
        store
            .record_denial(&aimed_here)
            .await
            .expect("record a refusal against the pinned tenant by another tenant's workload");
        let readable: Vec<String> = client
            .query(
                "SELECT denial_id FROM axond_cp_access_denial ORDER BY denial_id",
                &[],
            )
            .await
            .expect("read")
            .iter()
            .map(|row| row.get(0))
            .collect();
        assert_eq!(
            readable,
            vec![aimed_here.id.to_string()],
            "the tenant a cross-tenant attempt targeted cannot read that it happened"
        );

        // Grants have no tenant column of their own: a grant is this tenant's
        // exactly when the principal holding it is, and a session that could read
        // the rest would learn who administers the tenants it cannot see.
        let granted: Vec<String> = client
            .query(
                "SELECT DISTINCT p.slug FROM axond_cp_principal_role AS r \
                 JOIN axond_cp_principal AS p USING (principal_id) ORDER BY p.slug",
                &[],
            )
            .await
            .expect("read")
            .iter()
            .map(|row| row.get(0))
            .collect();
        assert_eq!(
            granted,
            ["admin", "deployer", "dev", "root"],
            "the grant table leaked another tenant's roles"
        );
        let orphaned: i64 = client
            .query_one("SELECT count(*) FROM axond_cp_principal_role", &[])
            .await
            .expect("read")
            .get(0);
        assert_eq!(
            orphaned, 4,
            "a grant whose principal this session cannot see is a grant it cannot see"
        );

        // The administrative journal, and the audit trail hanging off it. A
        // tenant-scoped change is another tenant's history; a deployment-scoped
        // one is shared and stays readable.
        let journal: Vec<Option<String>> = client
            .query(
                "SELECT DISTINCT tenant_id FROM axond_cp_mutation ORDER BY tenant_id",
                &[],
            )
            .await
            .expect("read")
            .iter()
            .map(|row| row.get(0))
            .collect();
        assert_eq!(
            journal,
            vec![Some(tenant_id(1).to_string())],
            "the mutation journal leaked another tenant's changes"
        );

        // And the same asymmetry the refusal trail has: a change recorded *against*
        // this tenant is this tenant's history whoever made it. Filtering the actor
        // on tenant-scoped rows too would hide a tenant's own change — along with
        // its audit event and its idempotency record, which are walled through this
        // table — because another tenant's service account attempted it.
        store
            .attempt(&format!(
                "INSERT INTO axond_cp_mutation (mutation_id, actor_kind, actor_tenant_id,                  actor_principal_id, mutation_kind, scope_kind, tenant_id, idempotency_key,                  submitted_at) VALUES ('mut_00000000-0000-7000-8000-00000000f00d', 'workload',                  '{}', '{}', 'publish', 'tenant', '{}', 'theirs-against-ours', now())",
                tenant_id(11),
                principal_id(38),
                tenant_id(1),
            ))
            .await
            .expect("record a change against this tenant by another tenant's workload");
        let attributed: Vec<String> = client
            .query(
                "SELECT mutation_id FROM axond_cp_mutation WHERE actor_kind = 'workload'                  ORDER BY mutation_id",
                &[],
            )
            .await
            .expect("read")
            .iter()
            .map(|row| row.get(0))
            .collect();
        assert_eq!(
            attributed,
            vec!["mut_00000000-0000-7000-8000-00000000f00d".to_owned()],
            "a tenant cannot read a change recorded against it by another tenant's workload"
        );
        let events: i64 = client
            .query_one("SELECT count(*) FROM axond_cp_audit_event", &[])
            .await
            .expect("read")
            .get(0);
        assert_eq!(
            events, 1,
            "an audit event is visible exactly when the mutation it describes is"
        );

        // Deployment-wide rows stay readable — the pinned session is a tenant's
        // view of shared state, not a broken one — and its own tenant's resources
        // are all it sees of tenant-scoped state.
        let scoped: Vec<String> = client
            .query(
                "SELECT DISTINCT tenant_id FROM axond_cp_resource_version \
                 WHERE tenant_id IS NOT NULL",
                &[],
            )
            .await
            .expect("read")
            .iter()
            .map(|row| row.get(0))
            .collect();
        assert_eq!(scoped, vec![tenant_id(1).to_string()]);
        let deployment: i64 = client
            .query_one(
                "SELECT count(*) FROM axond_cp_resource_version WHERE tenant_id IS NULL",
                &[],
            )
            .await
            .expect("read")
            .get(0);
        assert!(deployment > 0, "shared state is not hidden from a tenant");

        // The publication chain is platform state: one revision is every
        // tenant's desired state at one instant, so a pinned session reading it
        // would learn that other tenants exist and how often they publish.
        for table in [
            "axond_cp_head",
            "axond_cp_revision",
            "axond_cp_revision_entry",
            "axond_cp_revision_blob",
            "axond_cp_blob",
            "axond_cp_resource_dependency",
        ] {
            let visible: i64 = client
                .query_one(&format!("SELECT count(*) FROM {table}"), &[])
                .await
                .expect("read as the pinned session")
                .get(0);
            assert_eq!(visible, 0, "{table} leaked the publication chain");
        }

        // An idempotency record names no tenant, so it is walled through the
        // mutation it deduplicates: this session sees the key of its own retry
        // and not the other tenant's, which is the publication cadence the
        // journal policy already hides.
        let keys: i64 = client
            .query_one("SELECT count(*) FROM axond_cp_idempotency", &[])
            .await
            .expect("read as the pinned session")
            .get(0);
        assert_eq!(
            keys, 1,
            "an idempotency key is visible exactly when the mutation it records is"
        );

        // The unpinned publisher still sees both tenants, or the migration would
        // have hidden rows from the writer.
        assert_eq!(
            store
                .column("SELECT tenant_id FROM axond_cp_tenant ORDER BY tenant_id")
                .await
                .len(),
            2
        );
    }

    /// A look, as discovery would file it.
    fn look(
        scope: ScopeRef,
        result: DiscoveryResult,
        completeness: DiscoveryCompleteness,
        observed_at: SystemTime,
    ) -> DiscoveryObservation {
        DiscoveryObservation::new(
            scope,
            observation_target(),
            result,
            completeness,
            DiscoverySource::ProviderListing,
            observed_at,
        )
    }

    fn observation_target() -> TargetRef {
        TargetRef::parse("openai", "gpt-4o-mini").expect("a well-formed target")
    }

    fn instant(seconds: u64) -> SystemTime {
        UNIX_EPOCH + Duration::from_secs(seconds)
    }

    /// Evidence outlives the process that took it, and the probe's own words do
    /// not: a restart may know that a listing found the model, and must not carry
    /// the error body a failed one collected.
    #[tokio::test]
    async fn discovery_evidence_survives_a_restart_without_the_probes_own_words() {
        let Some((store, dsn, schema)) = journal().await else {
            return;
        };
        store
            .publish_revision(candidate(ExpectedRevision::Empty, "state", state()))
            .await
            .expect("a tenant and a project exist to own evidence");

        let scope = ScopeRef::project(tenant_id(1), project_id(2));
        let key = AvailabilityKey::new(scope, observation_target());
        let current = look(
            scope,
            DiscoveryResult::Indeterminate,
            DiscoveryCompleteness::Partial,
            instant(200),
        )
        .detailed("HTTP 500 from https://api.example.test/v1/models?key=sk-live-secret");
        let retained = look(
            scope,
            DiscoveryResult::Present,
            DiscoveryCompleteness::Complete,
            instant(100),
        )
        .expiring_at(instant(900));
        store
            .save(&EvidenceWrite::of_rows(vec![
                StoredObservation {
                    key: key.clone(),
                    slot: ObservationSlot::Current,
                    observation: current.clone(),
                    definitive_at: Some(instant(100)),
                },
                StoredObservation {
                    key: key.clone(),
                    slot: ObservationSlot::LastKnownGood,
                    observation: retained.clone(),
                    definitive_at: Some(instant(100)),
                },
            ]))
            .await
            .expect("evidence is written");

        // A second replica, as a restart is.
        let restarted = second_store(&dsn, &schema).await;
        let read = restarted.load(None).await.expect("evidence is read back");
        assert_eq!(read.len(), 2, "both slots survive");
        assert_eq!(read[0].slot, ObservationSlot::Current);
        assert_eq!(read[1].slot, ObservationSlot::LastKnownGood);
        assert_eq!(read[0].definitive_at, Some(instant(100)));
        assert!(
            read.iter().all(|row| row.observation.detail.is_none()),
            "a probe's detail must not be durable"
        );
        assert!(
            read[0].observation.is_same_look(&current),
            "the current look is the same evidence it was written as"
        );
        assert!(read[1].observation.is_same_look(&retained));
        assert_eq!(read[1].observation.expires_at, Some(instant(900)));
        assert_eq!(
            store
                .load(Some(ScopeRef::tenant(tenant_id(1))))
                .await
                .expect("read")
                .len(),
            0,
            "a project's evidence is not the tenant's"
        );
    }

    /// A look taken at an arbitrary instant is the same look after storage.
    ///
    /// A look's instant is its identity — replay detection compares it, and the
    /// definitive watermark is ordered by it — so an instant the column rounds
    /// would make a restored look a different one: the reload would count as
    /// evidence arriving out of order, and a restored positive could compare
    /// older than the watermark its own conclusion set and be discarded.
    #[tokio::test]
    async fn a_looks_instant_survives_the_column_it_is_stored_in() {
        let Some((store, _dsn, _schema)) = journal().await else {
            return;
        };
        store
            .publish_revision(candidate(ExpectedRevision::Empty, "state", state()))
            .await
            .expect("a tenant exists to own evidence");

        let scope = ScopeRef::tenant(tenant_id(1));
        let key = AvailabilityKey::new(scope, observation_target());
        // Nanoseconds the column cannot keep, as `SystemTime::now()` supplies.
        let taken = UNIX_EPOCH + Duration::new(1_700_000_000, 123_456_789);
        let observation = look(
            scope,
            DiscoveryResult::Present,
            DiscoveryCompleteness::Complete,
            taken,
        )
        .expiring_at(taken + Duration::from_secs(600));
        store
            .save(&EvidenceWrite::of_rows(vec![StoredObservation {
                key,
                slot: ObservationSlot::Current,
                observation: observation.clone(),
                definitive_at: Some(observation.observed_at),
            }]))
            .await
            .expect("evidence is written");

        let read = store.load(None).await.expect("evidence is read back");
        assert_eq!(read.len(), 1);
        assert!(
            read[0].observation.is_same_look(&observation),
            "a stored look is the look that was stored"
        );
        assert_eq!(read[0].definitive_at, Some(observation.observed_at));
    }

    /// A record is replaced whole, so a retained look a later conclusion
    /// discredited does not sit in storage waiting for the next restart.
    #[tokio::test]
    async fn saving_a_record_replaces_the_evidence_it_held() {
        let Some((store, _dsn, _schema)) = journal().await else {
            return;
        };
        store
            .publish_revision(candidate(ExpectedRevision::Empty, "state", state()))
            .await
            .expect("a tenant exists to own evidence");

        let scope = ScopeRef::tenant(tenant_id(1));
        let key = AvailabilityKey::new(scope, observation_target());
        let positive = look(
            scope,
            DiscoveryResult::Present,
            DiscoveryCompleteness::Complete,
            instant(100),
        );
        store
            .save(&EvidenceWrite::of_rows(vec![
                StoredObservation {
                    key: key.clone(),
                    slot: ObservationSlot::Current,
                    observation: positive.clone(),
                    definitive_at: Some(instant(100)),
                },
                StoredObservation {
                    key: key.clone(),
                    slot: ObservationSlot::LastKnownGood,
                    observation: positive,
                    definitive_at: Some(instant(100)),
                },
            ]))
            .await
            .expect("write");

        let dropped = look(
            scope,
            DiscoveryResult::Absent,
            DiscoveryCompleteness::Complete,
            instant(300),
        );
        store
            .save(&EvidenceWrite::of_rows(vec![StoredObservation {
                key: key.clone(),
                slot: ObservationSlot::Current,
                observation: dropped.clone(),
                definitive_at: Some(instant(300)),
            }]))
            .await
            .expect("write the conclusion");

        let read = store.load(Some(scope)).await.expect("read");
        assert_eq!(read.len(), 1, "the discredited fallback is gone");
        assert_eq!(read[0].slot, ObservationSlot::Current);
        assert!(read[0].observation.is_same_look(&dropped));
        assert_eq!(read[0].definitive_at, Some(instant(300)));
    }

    /// A key that stopped holding evidence carries no row to replace, so it is
    /// named as cleared: otherwise the one case where stored evidence must
    /// disappear is the one case a write cannot express, and every restart would
    /// restore the look a listing removed.
    #[tokio::test]
    async fn a_key_that_holds_no_evidence_holds_none_in_storage_either() {
        let Some((store, _dsn, _schema)) = journal().await else {
            return;
        };
        store
            .publish_revision(candidate(ExpectedRevision::Empty, "state", state()))
            .await
            .expect("a tenant exists to own evidence");

        let scope = ScopeRef::tenant(tenant_id(1));
        let key = AvailabilityKey::new(scope, observation_target());
        store
            .save(&EvidenceWrite::of_rows(vec![StoredObservation {
                key: key.clone(),
                slot: ObservationSlot::Current,
                observation: look(
                    scope,
                    DiscoveryResult::Present,
                    DiscoveryCompleteness::Complete,
                    instant(100),
                ),
                definitive_at: Some(instant(100)),
            }]))
            .await
            .expect("evidence is written");

        store
            .save(
                &EvidenceWrite::default().clearing([EvidenceClear::new(key.clone(), instant(100))]),
            )
            .await
            .expect("the key is cleared");

        assert!(
            store.load(None).await.expect("read").is_empty(),
            "a cleared key does not survive the process that cleared it"
        );

        // A definitive watermark is explicit ownership of a key whose looks were
        // discredited, so an index asks for the same cleanup without a separate
        // caller-side ledger.
        let write = EvidenceWrite::of_index(
            &AvailabilityIndex::builder()
                .record(
                    key.clone(),
                    AvailabilityRecord {
                        definitive_at: Some(instant(100)),
                        ..AvailabilityRecord::enabled()
                    },
                )
                .build(),
        );
        assert!(write.rows().is_empty());
        assert_eq!(write.cleared(), &[EvidenceClear::new(key, instant(100))]);
    }

    /// A replica that has only projected an empty record does not own that
    /// evidence key. Its durable write must not clear a look another replica
    /// recorded for the same scope and target.
    #[tokio::test]
    async fn an_empty_replica_write_does_not_clear_another_replicas_evidence() {
        let Some((store, dsn, schema)) = journal().await else {
            return;
        };
        store
            .publish_revision(candidate(ExpectedRevision::Empty, "state", state()))
            .await
            .expect("a tenant exists to own evidence");

        let scope = ScopeRef::tenant(tenant_id(1));
        let key = AvailabilityKey::new(scope, observation_target());
        store
            .save(&EvidenceWrite::of_rows(vec![StoredObservation {
                key: key.clone(),
                slot: ObservationSlot::Current,
                observation: look(
                    scope,
                    DiscoveryResult::Present,
                    DiscoveryCompleteness::Complete,
                    instant(100),
                ),
                definitive_at: Some(instant(100)),
            }]))
            .await
            .expect("replica A writes its look");

        let replica_b = second_store(&dsn, &schema).await;
        let empty_projection = AvailabilityIndex::builder()
            .record(key.clone(), AvailabilityRecord::enabled())
            .build();
        replica_b
            .save(&EvidenceWrite::of_index(&empty_projection))
            .await
            .expect("replica B has no owned evidence to replace");

        let rows = store.load(Some(scope)).await.expect("read evidence");
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].key, key);
        assert_eq!(rows[0].observation.result, DiscoveryResult::Present);
    }

    /// A replica can also be behind while still holding a look. Its whole-key
    /// replacement must not displace a newer look that another replica has
    /// already durably written, and its stale clear must not remove it either.
    #[tokio::test]
    async fn a_stale_replica_write_does_not_replace_newer_evidence() {
        let Some((store, dsn, schema)) = journal().await else {
            return;
        };
        store
            .publish_revision(candidate(ExpectedRevision::Empty, "state", state()))
            .await
            .expect("a tenant exists to own evidence");

        let scope = ScopeRef::tenant(tenant_id(1));
        let key = AvailabilityKey::new(scope, observation_target());
        let replica_b = second_store(&dsn, &schema).await;
        let newer = look(
            scope,
            DiscoveryResult::Present,
            DiscoveryCompleteness::Complete,
            instant(200),
        );
        store
            .save(&EvidenceWrite::of_rows(vec![StoredObservation {
                key: key.clone(),
                slot: ObservationSlot::Current,
                observation: newer.clone(),
                definitive_at: Some(instant(200)),
            }]))
            .await
            .expect("replica A writes the newer look");

        let stale = look(
            scope,
            DiscoveryResult::Absent,
            DiscoveryCompleteness::Complete,
            instant(100),
        );
        replica_b
            .save(&EvidenceWrite::of_rows(vec![StoredObservation {
                key: key.clone(),
                slot: ObservationSlot::Current,
                observation: stale,
                definitive_at: Some(instant(100)),
            }]))
            .await
            .expect("replica B writes its stale view");
        replica_b
            .save(
                &EvidenceWrite::default().clearing([EvidenceClear::new(key.clone(), instant(100))]),
            )
            .await
            .expect("replica B clears only evidence it knows about");

        let rows = store.load(Some(scope)).await.expect("read evidence");
        assert_eq!(rows.len(), 1);
        assert!(rows[0].observation.is_same_look(&newer));
    }

    /// A behind replica's current look can be newer than another replica's stored
    /// fallback and older than its stored current one. Replacing the key at that
    /// watermark would delete the fallback on behalf of a row the newer current
    /// look then keeps out, so the key is left alone entirely instead.
    #[tokio::test]
    async fn a_behind_replicas_write_does_not_erase_a_newer_records_fallback() {
        let Some((store, dsn, schema)) = journal().await else {
            return;
        };
        store
            .publish_revision(candidate(ExpectedRevision::Empty, "state", state()))
            .await
            .expect("a tenant exists to own evidence");

        let scope = ScopeRef::tenant(tenant_id(1));
        let key = AvailabilityKey::new(scope, observation_target());
        let current = look(
            scope,
            DiscoveryResult::Absent,
            DiscoveryCompleteness::Complete,
            instant(300),
        );
        let fallback = look(
            scope,
            DiscoveryResult::Present,
            DiscoveryCompleteness::Complete,
            instant(200),
        );
        store
            .save(&EvidenceWrite::of_rows(vec![
                StoredObservation {
                    key: key.clone(),
                    slot: ObservationSlot::Current,
                    observation: current.clone(),
                    definitive_at: Some(instant(300)),
                },
                StoredObservation {
                    key: key.clone(),
                    slot: ObservationSlot::LastKnownGood,
                    observation: fallback.clone(),
                    definitive_at: Some(instant(200)),
                },
            ]))
            .await
            .expect("replica A writes both slots");

        let replica_b = second_store(&dsn, &schema).await;
        replica_b
            .save(&EvidenceWrite::of_rows(vec![StoredObservation {
                key: key.clone(),
                slot: ObservationSlot::Current,
                observation: look(
                    scope,
                    DiscoveryResult::Absent,
                    DiscoveryCompleteness::Complete,
                    instant(250),
                ),
                definitive_at: Some(instant(250)),
            }]))
            .await
            .expect("replica B writes the only look it holds");

        let rows = store.load(Some(scope)).await.expect("read evidence");
        assert_eq!(
            rows.len(),
            2,
            "a behind replica replaces neither slot of a record it knows less about"
        );
        let held: Vec<(ObservationSlot, SystemTime)> = rows
            .iter()
            .map(|row| (row.slot, row.observation.observed_at))
            .collect();
        assert_eq!(
            held,
            vec![
                (ObservationSlot::Current, instant(300)),
                (ObservationSlot::LastKnownGood, instant(200)),
            ]
        );
    }

    /// A stale discovery writer naming a tenant the journal no longer owns is a
    /// permanent FK refusal, not a retryable control-plane outage.
    #[tokio::test]
    async fn an_observation_foreign_key_failure_is_permanent_not_unavailable() {
        let Some((store, _dsn, _schema)) = journal().await else {
            return;
        };
        let scope = ScopeRef::tenant(tenant_id(999));
        let error = store
            .save(&EvidenceWrite::of_rows(vec![StoredObservation {
                key: AvailabilityKey::new(scope, observation_target()),
                slot: ObservationSlot::Current,
                observation: look(
                    scope,
                    DiscoveryResult::Present,
                    DiscoveryCompleteness::Complete,
                    instant(100),
                ),
                definitive_at: Some(instant(100)),
            }]))
            .await
            .expect_err("the observation FK refuses a missing tenant");

        assert!(matches!(error, ControlPlaneError::Denied { .. }));
        assert!(
            !error.retryable(),
            "an FK violation is not cleared by retrying"
        );
    }

    /// Availability is entitlement, and entitlement is a tenant's: which models
    /// another tenant's credentials can reach is exactly the enumeration the
    /// row-level policies exist to prevent.
    #[tokio::test]
    async fn one_tenants_discovery_evidence_is_not_another_tenants() {
        let Some((store, dsn, schema)) = journal().await else {
            return;
        };
        store
            .publish_revision(candidate(
                ExpectedRevision::Empty,
                "two tenants",
                two_tenant_directory_state(),
            ))
            .await
            .expect("two tenants publish");

        for tenant in [tenant_id(1), tenant_id(11)] {
            let scope = ScopeRef::tenant(tenant);
            store
                .save(&EvidenceWrite::of_rows(vec![StoredObservation {
                    key: AvailabilityKey::new(scope, observation_target()),
                    slot: ObservationSlot::Current,
                    observation: look(
                        scope,
                        DiscoveryResult::Present,
                        DiscoveryCompleteness::Complete,
                        instant(100),
                    ),
                    definitive_at: Some(instant(100)),
                }]))
                .await
                .expect("write");
        }

        let mine = store
            .load(Some(ScopeRef::tenant(tenant_id(1))))
            .await
            .expect("read");
        assert_eq!(mine.len(), 1);
        assert_eq!(mine[0].key.scope.tenant, tenant_id(1));
        assert_eq!(store.load(None).await.expect("read").len(), 2);

        // And behind the service layer, the same wall the rest of the tenant-owned
        // tables are behind: a session pinned to one tenant, asking for everything.
        let role = format!("{schema}_availability_reader");
        store
            .attempt(&format!(
                "CREATE ROLE {role} LOGIN PASSWORD 'reader'; \
                 GRANT USAGE ON SCHEMA {schema} TO {role}; \
                 GRANT SELECT ON ALL TABLES IN SCHEMA {schema} TO {role}"
            ))
            .await
            .expect("create the reading role");
        let mut config: Config = dsn.parse().expect("test dsn");
        config.user(&role).password("reader");
        config.connect_timeout(Duration::from_secs(5));
        let (client, connection) = config
            .connect(crate::usage::tls_connector())
            .await
            .expect("connect as the reading role");
        tokio::spawn(async move {
            let _ = connection.await;
        });
        client
            .batch_execute(&format!(
                "SET search_path TO {schema}; SET axond.tenant_id = '{}'",
                tenant_id(11)
            ))
            .await
            .expect("pin the session to one tenant");
        let rows: Vec<String> = client
            .query(
                "SELECT tenant_id FROM axond_cp_availability_observation",
                &[],
            )
            .await
            .expect("read as the pinned session")
            .iter()
            .map(|row| row.get(0))
            .collect();
        assert_eq!(
            rows,
            vec![tenant_id(11).to_string()],
            "the observation table leaked another tenant's evidence"
        );
    }

    #[test]
    fn a_journal_timestamp_survives_a_round_trip_through_the_column_it_is_stored_in() {
        let now = journal_now();
        let since = now.duration_since(UNIX_EPOCH).expect("after the epoch");
        assert_eq!(
            since.subsec_nanos() % 1_000,
            0,
            "a timestamp with sub-microsecond precision cannot be read back as itself"
        );
    }
}