lodestone_proto/

cluster_control_protocol.rs

//! Strict cluster-control session wire values shared by servers and external clients.
//!
//! Cluster control is deliberately disjoint from [`crate::PublicSessionV1`].
//! A cluster-control credential has no Collection grant, and the action enum
//! makes each credential namespace's closed capability set explicit before a
//! listener or durable credential authority is allowed to accept it.

use std::error::Error;
use std::fmt::{self, Display, Formatter};

use serde::{Deserialize, Serialize};

pub use lodestone_types::{CollectionId, CollectionSecurityGeneration, LogicalMillis};

pub const CLUSTER_CONTROL_SESSION_FORMAT_VERSION: u16 = 2;
pub const MAX_CLUSTER_CONTROL_SESSION_BYTES: usize = 1_024;
pub const MAX_CLUSTER_CONTROL_CREDENTIAL_BYTES: usize = 512;
pub const CLUSTER_CONTROL_CREDENTIAL_PREFIX: &[u8] = b"lcc1_";
pub const CLUSTER_CONTROL_CREDENTIAL_ENCODED_BYTES: usize = 48;
pub const CLUSTER_CONTROL_REQUEST_FORMAT_VERSION: u16 = 2;
pub const CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION: u16 = 2;
pub const MAX_CLUSTER_CONTROL_FRAME_BYTES: usize = 2 * 1024 * 1024;
pub const MAX_CLUSTER_CONTROL_REQUEST_ID_BYTES: usize = 128;
pub const MAX_CLUSTER_CONTROL_ERROR_CODE_BYTES: usize = 96;
pub const COLLECTION_RETIREMENT_REQUEST_FORMAT_VERSION: u16 = 1;
pub const COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION: u16 = 1;
pub const COLLECTION_RETIREMENT_RECEIPT_FORMAT_VERSION: u16 = 1;
pub const MAX_COLLECTION_RETIREMENT_CURSOR_BYTES: usize = 8 * 1024;
pub const MAX_COLLECTION_RETIREMENT_BLOCKERS: usize = 32;

const COLLECTION_RETIREMENT_REQUEST_DIGEST_DOMAIN: &[u8] =
    b"lodestone/cluster-control/collection-retirement-request/v1\0";
const COLLECTION_RETIREMENT_CLOSURE_DIGEST_DOMAIN: &[u8] =
    b"lodestone/cluster-control/collection-retirement-closure/v1\0";

/// Policy admitted by the Collection provisioner. Protection and retention
/// authorities remain blockers; this policy never grants permission to
/// override them.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CollectionRetirementClosurePolicyV1 {
    RetireOwnedChildren,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CollectionRetirementRequestV1 {
    pub format_version: u16,
    pub operation_id: [u8; 32],
    pub collection_id: CollectionId,
    pub expected_security_generation: CollectionSecurityGeneration,
    pub closure_policy: CollectionRetirementClosurePolicyV1,
}

impl CollectionRetirementRequestV1 {
    pub fn validate(&self) -> Result<(), CollectionRetirementContractError> {
        if self.format_version != COLLECTION_RETIREMENT_REQUEST_FORMAT_VERSION
            || self.operation_id == [0; 32]
            || self.expected_security_generation.get() == 0
        {
            return Err(CollectionRetirementContractError::InvalidRequest);
        }
        Ok(())
    }

    pub fn canonical_request_digest(&self) -> Result<[u8; 32], CollectionRetirementContractError> {
        self.validate()?;
        canonical_collection_retirement_digest(COLLECTION_RETIREMENT_REQUEST_DIGEST_DOMAIN, self)
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CollectionRetirementOwnerV1 {
    Admission,
    Administration,
    Index,
    Security,
    Operation,
    Portability,
    Retention,
    Protection,
    Usage,
    Task,
    ResidentRoute,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum CollectionRetirementBlockerKindV1 {
    OwnerBusy,
    ActiveAuthority,
    PolicyRefusal,
    RetainedEvidence,
    PendingOperation,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CollectionRetirementBlockerV1 {
    pub owner: CollectionRetirementOwnerV1,
    pub kind: CollectionRetirementBlockerKindV1,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CollectionRetirementCursorV1 {
    pub owner: CollectionRetirementOwnerV1,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub after: Option<Vec<u8>>,
}

impl CollectionRetirementCursorV1 {
    fn validate(&self) -> Result<(), CollectionRetirementContractError> {
        if self
            .after
            .as_ref()
            .is_some_and(|after| after.is_empty() || after.len() > MAX_COLLECTION_RETIREMENT_CURSOR_BYTES)
        {
            return Err(CollectionRetirementContractError::InvalidStatus);
        }
        Ok(())
    }
}

#[derive(Clone, Copy, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CollectionRetirementChildSummaryV1 {
    pub administration_resources: u64,
    pub indexes: u64,
    pub security_resources: u64,
    pub operations: u64,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct CollectionRetirementReceiptV1 {
    pub format_version: u16,
    pub operation_id: [u8; 32],
    pub request_digest: [u8; 32],
    pub collection_id: CollectionId,
    pub admitted_security_generation: CollectionSecurityGeneration,
    pub retired_security_generation: CollectionSecurityGeneration,
    pub admitted_at: LogicalMillis,
    pub retired_at: LogicalMillis,
    pub final_root_digest: [u8; 32],
    pub closure_digest: [u8; 32],
    pub child_summary: CollectionRetirementChildSummaryV1,
}

impl CollectionRetirementReceiptV1 {
    pub fn validate(&self) -> Result<(), CollectionRetirementContractError> {
        if self.format_version != COLLECTION_RETIREMENT_RECEIPT_FORMAT_VERSION
            || self.operation_id == [0; 32]
            || self.request_digest == [0; 32]
            || self.admitted_security_generation.get() == 0
            || self.retired_security_generation.get() <= self.admitted_security_generation.get()
            || self.admitted_at.get() == 0
            || self.retired_at < self.admitted_at
            || self.final_root_digest == [0; 32]
            || self.closure_digest == [0; 32]
        {
            return Err(CollectionRetirementContractError::InvalidReceipt);
        }
        if self.canonical_closure_digest()? != self.closure_digest {
            return Err(CollectionRetirementContractError::InvalidReceipt);
        }
        Ok(())
    }

    pub fn canonical_closure_digest(&self) -> Result<[u8; 32], CollectionRetirementContractError> {
        canonical_collection_retirement_digest(
            COLLECTION_RETIREMENT_CLOSURE_DIGEST_DOMAIN,
            &(
                self.operation_id,
                self.request_digest,
                self.collection_id,
                self.admitted_security_generation,
                self.retired_security_generation,
                self.admitted_at,
                self.retired_at,
                self.final_root_digest,
                self.child_summary,
            ),
        )
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
pub enum CollectionRetirementOutcomeV1 {
    NotSubmitted {
        format_version: u16,
        request_digest: [u8; 32],
    },
    Blocked {
        format_version: u16,
        request_digest: [u8; 32],
        admitted_security_generation: CollectionSecurityGeneration,
        cursor: CollectionRetirementCursorV1,
        child_summary: CollectionRetirementChildSummaryV1,
        blockers: Vec<CollectionRetirementBlockerV1>,
    },
    InProgress {
        format_version: u16,
        request_digest: [u8; 32],
        admitted_security_generation: CollectionSecurityGeneration,
        cursor: CollectionRetirementCursorV1,
        child_summary: CollectionRetirementChildSummaryV1,
    },
    /// Client-only projection used when a call may have submitted but no
    /// authoritative response was received. The exact request is retained for
    /// resolver replay and is never safe to replace with a fresh identity.
    Indeterminate {
        format_version: u16,
        request: Box<CollectionRetirementRequestV1>,
    },
    Resolved {
        format_version: u16,
        receipt: Box<CollectionRetirementReceiptV1>,
    },
    KnownRetired {
        format_version: u16,
        request_digest: [u8; 32],
        collection_id: CollectionId,
        retired_security_generation: CollectionSecurityGeneration,
        retired_at: LogicalMillis,
    },
    ExpiredUnknown {
        format_version: u16,
        request_digest: [u8; 32],
        collection_id: CollectionId,
        operation_id: [u8; 32],
    },
}

impl CollectionRetirementOutcomeV1 {
    pub fn validate(&self) -> Result<(), CollectionRetirementContractError> {
        match self {
            Self::NotSubmitted {
                format_version,
                request_digest,
            } if *format_version == COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION && *request_digest != [0; 32] => Ok(()),
            Self::Blocked {
                format_version,
                request_digest,
                admitted_security_generation,
                cursor,
                blockers,
                ..
            } if *format_version == COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION
                && *request_digest != [0; 32]
                && admitted_security_generation.get() > 0
                && !blockers.is_empty()
                && blockers.len() <= MAX_COLLECTION_RETIREMENT_BLOCKERS
                && blockers.windows(2).all(|pair| pair[0] < pair[1]) =>
            {
                cursor.validate()
            }
            Self::InProgress {
                format_version,
                request_digest,
                admitted_security_generation,
                cursor,
                ..
            } if *format_version == COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION
                && *request_digest != [0; 32]
                && admitted_security_generation.get() > 0 =>
            {
                cursor.validate()
            }
            Self::Indeterminate {
                format_version,
                request,
            } if *format_version == COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION => request.validate(),
            Self::Resolved {
                format_version,
                receipt,
            } if *format_version == COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION => receipt.validate(),
            Self::KnownRetired {
                format_version,
                request_digest,
                retired_security_generation,
                retired_at,
                ..
            } if *format_version == COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION
                && *request_digest != [0; 32]
                && retired_security_generation.get() > 0
                && retired_at.get() > 0 =>
            {
                Ok(())
            }
            Self::ExpiredUnknown {
                format_version,
                request_digest,
                operation_id,
                ..
            } if *format_version == COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION
                && *request_digest != [0; 32]
                && *operation_id != [0; 32] =>
            {
                Ok(())
            }
            Self::NotSubmitted { .. }
            | Self::Blocked { .. }
            | Self::InProgress { .. }
            | Self::Indeterminate { .. }
            | Self::Resolved { .. }
            | Self::KnownRetired { .. }
            | Self::ExpiredUnknown { .. } => Err(CollectionRetirementContractError::InvalidStatus),
        }
    }
}

fn canonical_collection_retirement_digest(
    domain: &[u8],
    value: &impl Serialize,
) -> Result<[u8; 32], CollectionRetirementContractError> {
    let encoded = serde_json::to_vec(value).map_err(|_| CollectionRetirementContractError::Codec)?;
    let mut hasher = blake3::Hasher::new();
    hasher.update(domain);
    hasher.update(&encoded);
    Ok(*hasher.finalize().as_bytes())
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CollectionRetirementContractError {
    InvalidRequest,
    InvalidStatus,
    InvalidReceipt,
    Codec,
}

impl Display for CollectionRetirementContractError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter.write_str(match self {
            Self::InvalidRequest => "Collection-retirement request is invalid",
            Self::InvalidStatus => "Collection-retirement status is invalid",
            Self::InvalidReceipt => "Collection-retirement receipt is invalid",
            Self::Codec => "Collection-retirement canonical encoding failed",
        })
    }
}

impl Error for CollectionRetirementContractError {}

#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ClusterControlCredentialKindV1 {
    CollectionProvisioner,
    CapacityObserver,
    RecoveryOperator,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ClusterControlActionV2 {
    InitializeCollectionAuthority,
    ActivateCollectionAuthority,
    ResolveCollectionAuthority,
    InspectCollectionRetirement,
    RetireCollectionAuthority,
    ObserveClusterCapacity,
    ObserveResourceDiagnostics,
    ReserveClusterControlCredentialRotation,
    ActivateClusterControlCredentialRotation,
    RevokeClusterControlCredential,
    ResolveClusterControlCredential,
    ListClusterBackups,
    GetClusterBackup,
    CreateClusterBackup,
    ResolveClusterBackup,
    DeleteClusterBackup,
    ResolveClusterBackupDeletion,
    GetClusterRestoreReadiness,
    PrepareOfflineRecovery,
}

impl ClusterControlCredentialKindV1 {
    #[must_use]
    pub const fn admits(self, action: ClusterControlActionV2) -> bool {
        match self {
            Self::CollectionProvisioner => matches!(
                action,
                ClusterControlActionV2::InitializeCollectionAuthority
                    | ClusterControlActionV2::ActivateCollectionAuthority
                    | ClusterControlActionV2::ResolveCollectionAuthority
                    | ClusterControlActionV2::InspectCollectionRetirement
                    | ClusterControlActionV2::RetireCollectionAuthority
                    | ClusterControlActionV2::ReserveClusterControlCredentialRotation
                    | ClusterControlActionV2::ActivateClusterControlCredentialRotation
                    | ClusterControlActionV2::RevokeClusterControlCredential
                    | ClusterControlActionV2::ResolveClusterControlCredential
            ),
            Self::CapacityObserver => matches!(
                action,
                ClusterControlActionV2::ObserveClusterCapacity | ClusterControlActionV2::ObserveResourceDiagnostics
            ),
            Self::RecoveryOperator => matches!(
                action,
                ClusterControlActionV2::ReserveClusterControlCredentialRotation
                    | ClusterControlActionV2::ActivateClusterControlCredentialRotation
                    | ClusterControlActionV2::RevokeClusterControlCredential
                    | ClusterControlActionV2::ResolveClusterControlCredential
                    | ClusterControlActionV2::ListClusterBackups
                    | ClusterControlActionV2::GetClusterBackup
                    | ClusterControlActionV2::CreateClusterBackup
                    | ClusterControlActionV2::ResolveClusterBackup
                    | ClusterControlActionV2::DeleteClusterBackup
                    | ClusterControlActionV2::ResolveClusterBackupDeletion
                    | ClusterControlActionV2::GetClusterRestoreReadiness
                    | ClusterControlActionV2::PrepareOfflineRecovery
            ),
        }
    }
}

#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ClusterControlSessionV2 {
    pub format_version: u16,
    pub credential_kind: ClusterControlCredentialKindV1,
    pub credential: Vec<u8>,
    pub request_id: String,
    pub action: ClusterControlActionV2,
    pub request_correlation_id: [u8; 16],
}

impl fmt::Debug for ClusterControlSessionV2 {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("ClusterControlSessionV2")
            .field("format_version", &self.format_version)
            .field("credential_kind", &self.credential_kind)
            .field("credential", &"[REDACTED]")
            .field("request_id", &self.request_id)
            .field("action", &self.action)
            .field("request_correlation_id", &self.request_correlation_id)
            .finish()
    }
}

impl ClusterControlSessionV2 {
    pub fn validate(&self) -> Result<(), ClusterControlWireError> {
        if self.format_version != CLUSTER_CONTROL_SESSION_FORMAT_VERSION {
            return Err(ClusterControlWireError::UnsupportedVersion(self.format_version));
        }
        validate_cluster_control_credential_v1(&self.credential)?;
        if self.request_id.is_empty()
            || self.request_id.len() > MAX_CLUSTER_CONTROL_REQUEST_ID_BYTES
            || self.request_id.bytes().any(|byte| byte.is_ascii_control())
        {
            return Err(ClusterControlWireError::InvalidRequestIdentity);
        }
        if !self.credential_kind.admits(self.action) {
            return Err(ClusterControlWireError::ActionDenied);
        }
        Ok(())
    }
}

pub fn validate_cluster_control_credential_v1(credential: &[u8]) -> Result<(), ClusterControlWireError> {
    if credential.len() != CLUSTER_CONTROL_CREDENTIAL_ENCODED_BYTES
        || !credential.starts_with(CLUSTER_CONTROL_CREDENTIAL_PREFIX)
        || !credential[CLUSTER_CONTROL_CREDENTIAL_PREFIX.len()..]
            .iter()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
    {
        return Err(ClusterControlWireError::InvalidCredentialShape);
    }
    Ok(())
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ClusterControlRequestV2 {
    pub format_version: u16,
    pub session: ClusterControlSessionV2,
    pub body: serde_json::Value,
}

impl ClusterControlRequestV2 {
    pub fn validate(&self) -> Result<(), ClusterControlWireError> {
        if self.format_version != CLUSTER_CONTROL_REQUEST_FORMAT_VERSION {
            return Err(ClusterControlWireError::UnsupportedVersion(self.format_version));
        }
        self.session.validate()?;
        if !self.body.is_object() {
            return Err(ClusterControlWireError::InvalidBody);
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ClusterControlResponseDispositionV2 {
    RetrySameRequest,
    ResolveOperation,
    CorrectRequest,
    Reauthenticate,
    Terminal,
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(tag = "outcome", rename_all = "snake_case", deny_unknown_fields)]
pub enum ClusterControlResponseOutcomeV2 {
    Succeeded {
        body: serde_json::Value,
    },
    Refused {
        code: String,
        disposition: ClusterControlResponseDispositionV2,
    },
}

#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct ClusterControlResponseV2 {
    pub format_version: u16,
    pub request_id: String,
    pub request_correlation_id: [u8; 16],
    pub outcome: ClusterControlResponseOutcomeV2,
}

impl ClusterControlResponseV2 {
    pub fn validate(&self) -> Result<(), ClusterControlWireError> {
        if self.format_version != CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION {
            return Err(ClusterControlWireError::UnsupportedVersion(self.format_version));
        }
        validate_request_id(&self.request_id)?;
        match &self.outcome {
            ClusterControlResponseOutcomeV2::Succeeded { body } if body.is_object() => Ok(()),
            ClusterControlResponseOutcomeV2::Succeeded { .. } => Err(ClusterControlWireError::InvalidBody),
            ClusterControlResponseOutcomeV2::Refused { code, .. }
                if !code.is_empty()
                    && code.len() <= MAX_CLUSTER_CONTROL_ERROR_CODE_BYTES
                    && code
                        .bytes()
                        .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_') =>
            {
                Ok(())
            }
            ClusterControlResponseOutcomeV2::Refused { .. } => Err(ClusterControlWireError::InvalidErrorCode),
        }
    }
}

pub fn encode_cluster_control_request_v2(
    request: &ClusterControlRequestV2,
) -> Result<Vec<u8>, ClusterControlWireError> {
    request.validate()?;
    encode_frame(request)
}

pub fn decode_cluster_control_request_v2(bytes: &[u8]) -> Result<ClusterControlRequestV2, ClusterControlWireError> {
    let request: ClusterControlRequestV2 = decode_frame(bytes)?;
    request.validate()?;
    Ok(request)
}

pub fn encode_cluster_control_response_v2(
    response: &ClusterControlResponseV2,
) -> Result<Vec<u8>, ClusterControlWireError> {
    response.validate()?;
    encode_frame(response)
}

pub fn decode_cluster_control_response_v2(bytes: &[u8]) -> Result<ClusterControlResponseV2, ClusterControlWireError> {
    let response: ClusterControlResponseV2 = decode_frame(bytes)?;
    response.validate()?;
    Ok(response)
}

fn encode_frame(value: &impl Serialize) -> Result<Vec<u8>, ClusterControlWireError> {
    let bytes = serde_json::to_vec(value).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
    if bytes.is_empty() || bytes.len() > MAX_CLUSTER_CONTROL_FRAME_BYTES {
        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
    }
    Ok(bytes)
}

fn decode_frame<T: for<'de> Deserialize<'de> + Serialize>(bytes: &[u8]) -> Result<T, ClusterControlWireError> {
    if bytes.is_empty() || bytes.len() > MAX_CLUSTER_CONTROL_FRAME_BYTES {
        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
    }
    let value = serde_json::from_slice(bytes).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
    if serde_json::to_vec(&value).map_err(|error| ClusterControlWireError::Codec(error.to_string()))? != bytes {
        return Err(ClusterControlWireError::NonCanonical);
    }
    Ok(value)
}

fn validate_request_id(request_id: &str) -> Result<(), ClusterControlWireError> {
    if request_id.is_empty()
        || request_id.len() > MAX_CLUSTER_CONTROL_REQUEST_ID_BYTES
        || request_id.bytes().any(|byte| byte.is_ascii_control())
    {
        return Err(ClusterControlWireError::InvalidRequestIdentity);
    }
    Ok(())
}

pub fn encode_cluster_control_session_v2(
    session: &ClusterControlSessionV2,
) -> Result<Vec<u8>, ClusterControlWireError> {
    session.validate()?;
    let bytes = serde_json::to_vec(session).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
    if bytes.len() > MAX_CLUSTER_CONTROL_SESSION_BYTES {
        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
    }
    Ok(bytes)
}

pub fn decode_cluster_control_session_v2(bytes: &[u8]) -> Result<ClusterControlSessionV2, ClusterControlWireError> {
    if bytes.is_empty() || bytes.len() > MAX_CLUSTER_CONTROL_SESSION_BYTES {
        return Err(ClusterControlWireError::FrameTooLarge(bytes.len()));
    }
    let session: ClusterControlSessionV2 =
        serde_json::from_slice(bytes).map_err(|error| ClusterControlWireError::Codec(error.to_string()))?;
    session.validate()?;
    if serde_json::to_vec(&session).map_err(|error| ClusterControlWireError::Codec(error.to_string()))? != bytes {
        return Err(ClusterControlWireError::NonCanonical);
    }
    Ok(session)
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum ClusterControlWireError {
    UnsupportedVersion(u16),
    InvalidCredentialShape,
    InvalidRequestIdentity,
    InvalidBody,
    InvalidErrorCode,
    ActionDenied,
    FrameTooLarge(usize),
    NonCanonical,
    Codec(String),
}

impl Display for ClusterControlWireError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::UnsupportedVersion(version) => {
                write!(formatter, "unsupported cluster-control wire version {version}")
            }
            Self::InvalidCredentialShape => formatter.write_str("invalid cluster-control credential shape"),
            Self::InvalidRequestIdentity => formatter.write_str("invalid cluster-control request identity"),
            Self::InvalidBody => formatter.write_str("invalid cluster-control request or response body"),
            Self::InvalidErrorCode => formatter.write_str("invalid cluster-control error code"),
            Self::ActionDenied => formatter.write_str("cluster-control credential kind does not admit the action"),
            Self::FrameTooLarge(size) => write!(formatter, "cluster-control frame size {size} exceeds the bound"),
            Self::NonCanonical => formatter.write_str("cluster-control frame is not canonical"),
            Self::Codec(error) => write!(formatter, "cluster-control frame codec failed: {error}"),
        }
    }
}

impl Error for ClusterControlWireError {}

#[cfg(test)]
mod tests {
    use super::*;

    fn retirement_request() -> Result<CollectionRetirementRequestV1, Box<dyn Error>> {
        Ok(CollectionRetirementRequestV1 {
            format_version: COLLECTION_RETIREMENT_REQUEST_FORMAT_VERSION,
            operation_id: [11; 32],
            collection_id: CollectionId::new(7)?,
            expected_security_generation: CollectionSecurityGeneration::new(4),
            closure_policy: CollectionRetirementClosurePolicyV1::RetireOwnedChildren,
        })
    }

    fn session(
        credential_kind: ClusterControlCredentialKindV1,
        action: ClusterControlActionV2,
    ) -> ClusterControlSessionV2 {
        ClusterControlSessionV2 {
            format_version: CLUSTER_CONTROL_SESSION_FORMAT_VERSION,
            credential_kind,
            credential: b"lcc1_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".to_vec(),
            request_id: "capacity-sample-1".to_owned(),
            action,
            request_correlation_id: [7; 16],
        }
    }

    #[test]
    fn credential_namespaces_are_disjoint_and_canonical() -> Result<(), Box<dyn Error>> {
        let observer = session(
            ClusterControlCredentialKindV1::CapacityObserver,
            ClusterControlActionV2::ObserveClusterCapacity,
        );
        let bytes = encode_cluster_control_session_v2(&observer)?;
        assert_eq!(decode_cluster_control_session_v2(&bytes)?, observer);

        let provisioner_backup = session(
            ClusterControlCredentialKindV1::CollectionProvisioner,
            ClusterControlActionV2::GetClusterBackup,
        );
        assert_eq!(
            provisioner_backup.validate(),
            Err(ClusterControlWireError::ActionDenied)
        );

        let observer_mutation = session(
            ClusterControlCredentialKindV1::CapacityObserver,
            ClusterControlActionV2::InitializeCollectionAuthority,
        );
        assert_eq!(observer_mutation.validate(), Err(ClusterControlWireError::ActionDenied));

        let recovery_capacity = session(
            ClusterControlCredentialKindV1::RecoveryOperator,
            ClusterControlActionV2::ObserveClusterCapacity,
        );
        assert_eq!(recovery_capacity.validate(), Err(ClusterControlWireError::ActionDenied));

        let recovery_delete = session(
            ClusterControlCredentialKindV1::RecoveryOperator,
            ClusterControlActionV2::DeleteClusterBackup,
        );
        assert!(recovery_delete.validate().is_ok());
        let provisioner_delete = session(
            ClusterControlCredentialKindV1::CollectionProvisioner,
            ClusterControlActionV2::DeleteClusterBackup,
        );
        assert_eq!(
            provisioner_delete.validate(),
            Err(ClusterControlWireError::ActionDenied)
        );

        let mut public_shape = observer;
        public_shape.credential = b"1:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".to_vec();
        assert_eq!(
            public_shape.validate(),
            Err(ClusterControlWireError::InvalidCredentialShape)
        );
        Ok(())
    }

    #[test]
    fn decoder_refuses_predecessor_newer_unknown_and_malformed_frames() -> Result<(), Box<dyn Error>> {
        let observer = session(
            ClusterControlCredentialKindV1::CapacityObserver,
            ClusterControlActionV2::ObserveClusterCapacity,
        );
        let predecessor = String::from_utf8(encode_cluster_control_session_v2(&observer)?)?.replacen(
            &format!("\"format_version\":{CLUSTER_CONTROL_SESSION_FORMAT_VERSION}"),
            &format!("\"format_version\":{}", CLUSTER_CONTROL_SESSION_FORMAT_VERSION - 1),
            1,
        );
        assert_eq!(
            decode_cluster_control_session_v2(predecessor.as_bytes()),
            Err(ClusterControlWireError::UnsupportedVersion(1))
        );
        let mut newer = serde_json::to_value(&observer)?;
        newer["format_version"] = serde_json::json!(CLUSTER_CONTROL_SESSION_FORMAT_VERSION + 1);
        assert!(matches!(
            decode_cluster_control_session_v2(&serde_json::to_vec(&newer)?),
            Err(ClusterControlWireError::UnsupportedVersion(_))
        ));

        let mut unknown = serde_json::to_value(observer)?;
        unknown["collection_id"] = serde_json::json!(1);
        assert!(decode_cluster_control_session_v2(&serde_json::to_vec(&unknown)?).is_err());

        for malformed in [
            Vec::new(),
            vec![0xff],
            vec![b'x'; MAX_CLUSTER_CONTROL_SESSION_BYTES + 1],
        ] {
            assert!(decode_cluster_control_session_v2(&malformed).is_err());
        }
        Ok(())
    }

    #[test]
    fn request_and_response_envelopes_are_strict_bounded_and_correlated() -> Result<(), Box<dyn Error>> {
        let request = ClusterControlRequestV2 {
            format_version: CLUSTER_CONTROL_REQUEST_FORMAT_VERSION,
            session: session(
                ClusterControlCredentialKindV1::CapacityObserver,
                ClusterControlActionV2::ObserveClusterCapacity,
            ),
            body: serde_json::json!({}),
        };
        let encoded = encode_cluster_control_request_v2(&request)?;
        assert_eq!(decode_cluster_control_request_v2(&encoded)?, request);

        let predecessor_request = String::from_utf8(encoded.clone())?.replacen(
            &format!("\"format_version\":{CLUSTER_CONTROL_REQUEST_FORMAT_VERSION}"),
            &format!("\"format_version\":{}", CLUSTER_CONTROL_REQUEST_FORMAT_VERSION - 1),
            1,
        );
        assert_eq!(
            decode_cluster_control_request_v2(predecessor_request.as_bytes()),
            Err(ClusterControlWireError::UnsupportedVersion(1))
        );

        let response = ClusterControlResponseV2 {
            format_version: CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION,
            request_id: request.session.request_id.clone(),
            request_correlation_id: request.session.request_correlation_id,
            outcome: ClusterControlResponseOutcomeV2::Succeeded {
                body: serde_json::json!({"observation_id": "sample-1"}),
            },
        };
        let encoded = encode_cluster_control_response_v2(&response)?;
        assert_eq!(decode_cluster_control_response_v2(&encoded)?, response);

        let predecessor_response = String::from_utf8(encoded)?.replacen(
            &format!("\"format_version\":{CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION}"),
            &format!("\"format_version\":{}", CLUSTER_CONTROL_RESPONSE_FORMAT_VERSION - 1),
            1,
        );
        assert_eq!(
            decode_cluster_control_response_v2(predecessor_response.as_bytes()),
            Err(ClusterControlWireError::UnsupportedVersion(1))
        );

        let mut unknown = serde_json::to_value(&request)?;
        unknown["unexpected"] = serde_json::json!(true);
        assert!(decode_cluster_control_request_v2(&serde_json::to_vec(&unknown)?).is_err());
        assert!(matches!(
            decode_cluster_control_request_v2(&vec![b'x'; MAX_CLUSTER_CONTROL_FRAME_BYTES + 1]),
            Err(ClusterControlWireError::FrameTooLarge(_))
        ));
        Ok(())
    }

    #[test]
    fn collection_retirement_identity_binds_every_admitted_semantic() -> Result<(), Box<dyn Error>> {
        let request = retirement_request()?;
        let digest = request.canonical_request_digest()?;
        let mut changed = request.clone();
        changed.expected_security_generation = CollectionSecurityGeneration::new(5);
        assert_ne!(changed.canonical_request_digest()?, digest);

        let mut invalid = request;
        invalid.operation_id = [0; 32];
        assert_eq!(
            invalid.validate(),
            Err(CollectionRetirementContractError::InvalidRequest)
        );
        Ok(())
    }

    #[test]
    fn collection_retirement_receipt_and_recovery_statuses_fail_closed() -> Result<(), Box<dyn Error>> {
        let request = retirement_request()?;
        let mut receipt = CollectionRetirementReceiptV1 {
            format_version: COLLECTION_RETIREMENT_RECEIPT_FORMAT_VERSION,
            operation_id: request.operation_id,
            request_digest: request.canonical_request_digest()?,
            collection_id: request.collection_id,
            admitted_security_generation: CollectionSecurityGeneration::new(5),
            retired_security_generation: CollectionSecurityGeneration::new(6),
            admitted_at: LogicalMillis::new(12),
            retired_at: LogicalMillis::new(18),
            final_root_digest: [21; 32],
            closure_digest: [0; 32],
            child_summary: CollectionRetirementChildSummaryV1 {
                administration_resources: 2,
                indexes: 3,
                security_resources: 4,
                operations: 0,
            },
        };
        receipt.closure_digest = receipt.canonical_closure_digest()?;
        receipt.validate()?;
        let closure_digest = receipt.closure_digest;
        receipt.child_summary.indexes += 1;
        assert_eq!(
            receipt.validate(),
            Err(CollectionRetirementContractError::InvalidReceipt)
        );
        assert_ne!(receipt.canonical_closure_digest()?, closure_digest);

        let blocked = CollectionRetirementOutcomeV1::Blocked {
            format_version: COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION,
            request_digest: request.canonical_request_digest()?,
            admitted_security_generation: CollectionSecurityGeneration::new(5),
            cursor: CollectionRetirementCursorV1 {
                owner: CollectionRetirementOwnerV1::Protection,
                after: None,
            },
            child_summary: CollectionRetirementChildSummaryV1::default(),
            blockers: vec![CollectionRetirementBlockerV1 {
                owner: CollectionRetirementOwnerV1::Protection,
                kind: CollectionRetirementBlockerKindV1::PolicyRefusal,
            }],
        };
        blocked.validate()?;
        let indeterminate = CollectionRetirementOutcomeV1::Indeterminate {
            format_version: COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION,
            request: Box::new(request),
        };
        indeterminate.validate()?;
        Ok(())
    }
}