lodestone-proto 0.0.16

//! Composition helpers for the authenticated P6 transport runtime.
//!
//! P6 deliberately has one wire body codec. The fixed [`P6RpcPrelude`] is
//! read and authorized first; only then may the runtime allocate/read the
//! declared bytes and pass them to [`InternalRpcFrame::decode`]. This module
//! does not allocate from a peer-declared length and does not introduce a
//! second frame decoder.

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

use lodestone_types::NodeRole;

use crate::{
    INTERNAL_RPC_FORMAT_VERSION, InternalRpcError, InternalRpcFrame, InternalRpcMethod, InternalRpcStatus,
    MAX_INTERNAL_RPC_FRAME_BYTES, MetastoreCommandFamily, RPC_MAX_CONTROL_PAYLOAD_BYTES, RPC_MAX_PAYLOAD_BYTES,
    RPC_METASTORE_SNAPSHOT_REQUEST_FRAME_BYTES, RPC_METASTORE_SNAPSHOT_RESPONSE_FRAME_BYTES, RPC_PRELUDE_BYTES,
    RPC_SEARCH_REQUEST_FRAME_BYTES, RequestId, RpcAuthorizationError, RpcMethod, RpcNegotiatedProfile, RpcPrelude,
    WorkloadIdentity, authorize_rpc,
};

/// Correlation used when the transport must refuse a request before reading
/// the inner frame and therefore cannot know the caller's request ID.
///
/// The value is fixed and non-secret. It is never an idempotency key and must
/// never be submitted to an application handler.
pub const TRANSPORT_DENIAL_REQUEST_ID: &str = "p6-preallocation-denial-v1";
pub const RESPONSE_LENGTH_BYTES: usize = 4;

/// Decodes the response length prefix without allocating from the declared
/// value. The caller may allocate only after this cap passes.
pub fn decode_response_length(bytes: &[u8]) -> Result<usize, TransportError> {
    let array: [u8; RESPONSE_LENGTH_BYTES] = bytes
        .try_into()
        .map_err(|_| TransportError::ResponseLengthPrefix { actual: bytes.len() })?;
    let length =
        usize::try_from(u32::from_be_bytes(array)).map_err(|_| TransportError::ResponseLengthTooLarge(usize::MAX))?;
    if length > MAX_INTERNAL_RPC_FRAME_BYTES {
        return Err(TransportError::ResponseLengthTooLarge(length));
    }
    Ok(length)
}

/// Canonical coarse mapping from the P6 authorization method to the existing
/// internal frame method. The prelude remains authoritative for ACL decisions;
/// this mapping prevents a permitted prelude from smuggling a differently
/// classified inner frame.
#[must_use]
pub const fn internal_rpc_method_for(method: RpcMethod) -> InternalRpcMethod {
    match method {
        RpcMethod::RaftVote => InternalRpcMethod::RaftVote,
        RpcMethod::RaftAppend => InternalRpcMethod::RaftAppend,
        RpcMethod::RaftInstallSnapshot => InternalRpcMethod::RaftInstallSnapshot,
        RpcMethod::MetastoreCommand
        | RpcMethod::OrchestratorPlacementIntent
        | RpcMethod::OrchestratorCatalogIntent
        | RpcMethod::ProposePlacementRecovery
        | RpcMethod::CoreApiAuthority => InternalRpcMethod::MetastoreCommand,
        RpcMethod::MetastoreSnapshot | RpcMethod::MetastorePointRead | RpcMethod::MetastoreWatch => {
            InternalRpcMethod::MetastoreSnapshot
        }
        RpcMethod::NodeSelfHealth | RpcMethod::InternalHealth => InternalRpcMethod::Health,
        RpcMethod::WriterIngest | RpcMethod::TaskDestinationMutation | RpcMethod::CompactorExecuteTask => {
            InternalRpcMethod::WriterIngest
        }
        RpcMethod::WriterRealtimeRead | RpcMethod::SearchCoordinate | RpcMethod::SearchShardExecute => {
            InternalRpcMethod::Search
        }
        RpcMethod::WriterDrainAdvisory => InternalRpcMethod::WriterFence,
        RpcMethod::SearchMount => InternalRpcMethod::MountManifest,
    }
}

/// Channel-derived identity failures. Payload fields never construct one of
/// these values; the mTLS identity verifier supplies them to the runtime.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum ChannelIdentityError {
    Unauthenticated,
    VerifierUnavailable,
    WrongCluster,
    WrongRole,
    StaleNodeIncarnation,
    CredentialMismatch,
    UnsupportedIdentityVersion,
}

/// Stable, detail-free transport status for a channel identity refusal.
#[must_use]
pub const fn channel_identity_error_status(error: ChannelIdentityError) -> InternalRpcStatus {
    match error {
        ChannelIdentityError::Unauthenticated => InternalRpcStatus::Unauthenticated,
        ChannelIdentityError::VerifierUnavailable => InternalRpcStatus::Unavailable,
        ChannelIdentityError::WrongCluster => InternalRpcStatus::WrongCluster,
        ChannelIdentityError::WrongRole => InternalRpcStatus::WrongRole,
        ChannelIdentityError::StaleNodeIncarnation => InternalRpcStatus::StaleNodeIncarnation,
        ChannelIdentityError::CredentialMismatch => InternalRpcStatus::CredentialMismatch,
        ChannelIdentityError::UnsupportedIdentityVersion => InternalRpcStatus::UnknownVersion,
    }
}

/// Stable, detail-free transport status for a fixed-prelude refusal.
#[must_use]
pub const fn authorization_error_status(error: &RpcAuthorizationError) -> InternalRpcStatus {
    match error {
        RpcAuthorizationError::UnsupportedIdentityVersion(_)
        | RpcAuthorizationError::UnsupportedPreludeVersion(_)
        | RpcAuthorizationError::IncompatibleWireVersion => InternalRpcStatus::UnknownVersion,
        RpcAuthorizationError::CredentialMismatch => InternalRpcStatus::CredentialMismatch,
        RpcAuthorizationError::IncompatibleAuthProfile { .. } => InternalRpcStatus::IncompatibleAuthProfile,
        RpcAuthorizationError::PermissionDenied => InternalRpcStatus::PermissionDenied,
        RpcAuthorizationError::WrongRole => InternalRpcStatus::WrongRole,
        RpcAuthorizationError::BadMagic
        | RpcAuthorizationError::MalformedCommandFamily
        | RpcAuthorizationError::PayloadTooLarge { .. }
        | RpcAuthorizationError::PreludeLength { .. }
        | RpcAuthorizationError::ReservedBits
        | RpcAuthorizationError::UnknownCommandFamily(_)
        | RpcAuthorizationError::UnknownMethod(_)
        | RpcAuthorizationError::UnknownRole(_) => InternalRpcStatus::Malformed,
    }
}

/// Statuses which may be emitted without dispatching an application handler.
/// Handler outcomes such as `not_leader` and `stale_epoch` use ordinary
/// response framing instead.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum TransportDenialStatus {
    Overloaded,
    Unavailable,
    Malformed,
    UnknownVersion,
    Unauthenticated,
    PermissionDenied,
    WrongCluster,
    WrongRole,
    StaleNodeIncarnation,
    CredentialMismatch,
    IncompatibleAuthProfile,
}

impl TransportDenialStatus {
    #[must_use]
    pub const fn as_internal(self) -> InternalRpcStatus {
        match self {
            Self::Overloaded => InternalRpcStatus::Overloaded,
            Self::Unavailable => InternalRpcStatus::Unavailable,
            Self::Malformed => InternalRpcStatus::Malformed,
            Self::UnknownVersion => InternalRpcStatus::UnknownVersion,
            Self::Unauthenticated => InternalRpcStatus::Unauthenticated,
            Self::PermissionDenied => InternalRpcStatus::PermissionDenied,
            Self::WrongCluster => InternalRpcStatus::WrongCluster,
            Self::WrongRole => InternalRpcStatus::WrongRole,
            Self::StaleNodeIncarnation => InternalRpcStatus::StaleNodeIncarnation,
            Self::CredentialMismatch => InternalRpcStatus::CredentialMismatch,
            Self::IncompatibleAuthProfile => InternalRpcStatus::IncompatibleAuthProfile,
        }
    }

    pub fn from_internal(status: InternalRpcStatus) -> Result<Self, TransportError> {
        match status {
            InternalRpcStatus::Overloaded => Ok(Self::Overloaded),
            InternalRpcStatus::Unavailable => Ok(Self::Unavailable),
            InternalRpcStatus::Malformed => Ok(Self::Malformed),
            InternalRpcStatus::UnknownVersion => Ok(Self::UnknownVersion),
            InternalRpcStatus::Unauthenticated => Ok(Self::Unauthenticated),
            InternalRpcStatus::PermissionDenied => Ok(Self::PermissionDenied),
            InternalRpcStatus::WrongCluster => Ok(Self::WrongCluster),
            InternalRpcStatus::WrongRole => Ok(Self::WrongRole),
            InternalRpcStatus::StaleNodeIncarnation => Ok(Self::StaleNodeIncarnation),
            InternalRpcStatus::CredentialMismatch => Ok(Self::CredentialMismatch),
            InternalRpcStatus::IncompatibleAuthProfile => Ok(Self::IncompatibleAuthProfile),
            InternalRpcStatus::Ok | InternalRpcStatus::NotLeader | InternalRpcStatus::StaleEpoch => {
                Err(TransportError::InvalidDenialStatus(status))
            }
        }
    }

    #[must_use]
    pub const fn from_channel_identity(error: ChannelIdentityError) -> Self {
        match error {
            ChannelIdentityError::Unauthenticated => Self::Unauthenticated,
            ChannelIdentityError::VerifierUnavailable => Self::Unavailable,
            ChannelIdentityError::WrongCluster => Self::WrongCluster,
            ChannelIdentityError::WrongRole => Self::WrongRole,
            ChannelIdentityError::StaleNodeIncarnation => Self::StaleNodeIncarnation,
            ChannelIdentityError::CredentialMismatch => Self::CredentialMismatch,
            ChannelIdentityError::UnsupportedIdentityVersion => Self::UnknownVersion,
        }
    }

    #[must_use]
    pub const fn from_authorization(error: &RpcAuthorizationError) -> Self {
        match authorization_error_status(error) {
            InternalRpcStatus::Malformed => Self::Malformed,
            InternalRpcStatus::UnknownVersion => Self::UnknownVersion,
            InternalRpcStatus::PermissionDenied => Self::PermissionDenied,
            InternalRpcStatus::WrongRole => Self::WrongRole,
            InternalRpcStatus::CredentialMismatch => Self::CredentialMismatch,
            InternalRpcStatus::IncompatibleAuthProfile => Self::IncompatibleAuthProfile,
            // The authorization mapper above cannot yield any other status.
            _ => Self::Malformed,
        }
    }
}

/// Two wire buffers written in order by the client runtime.
///
/// Cloning or retaining this value for a transport retry preserves the exact
/// caller-supplied [`RequestId`]. That token is correlation only; application
/// idempotency remains owned by the request payload's durable contract.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct EncodedRequest {
    prelude: [u8; RPC_PRELUDE_BYTES],
    frame: Vec<u8>,
    response: ResponseExpectation,
    method: RpcMethod,
    command_family: Option<MetastoreCommandFamily>,
}

impl EncodedRequest {
    #[must_use]
    pub const fn prelude(&self) -> &[u8; RPC_PRELUDE_BYTES] {
        &self.prelude
    }

    #[must_use]
    pub fn frame(&self) -> &[u8] {
        &self.frame
    }

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

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

    #[must_use]
    pub const fn response_expectation(&self) -> &ResponseExpectation {
        &self.response
    }

    #[must_use]
    pub fn into_parts(self) -> ([u8; RPC_PRELUDE_BYTES], Vec<u8>, ResponseExpectation) {
        (self.prelude, self.frame, self.response)
    }
}

/// Constructs the fixed prelude and existing checksummed inner frame for one
/// outbound request. No caller-provided payload actor participates in ACL.
pub fn encode_request(
    target_role: NodeRole,
    method: RpcMethod,
    command_family: Option<MetastoreCommandFamily>,
    request_id: RequestId,
    payload: Vec<u8>,
) -> Result<EncodedRequest, TransportError> {
    if request_id.as_str() == TRANSPORT_DENIAL_REQUEST_ID {
        return Err(TransportError::ReservedRequestId);
    }
    let inner_method = internal_rpc_method_for(method);
    let response = ResponseExpectation {
        request_id: request_id.clone(),
        method: inner_method,
    };
    let frame = InternalRpcFrame::request(request_id, inner_method, payload)
        .encode()
        .map_err(TransportError::Frame)?;
    let maximum = request_frame_limit(method);
    if frame.len() > maximum {
        return Err(TransportError::FrameForMethodTooLarge {
            actual: frame.len(),
            maximum,
        });
    }
    let declared_payload_bytes = u32::try_from(frame.len()).map_err(|_| TransportError::FrameForMethodTooLarge {
        actual: frame.len(),
        maximum,
    })?;
    let prelude = RpcPrelude {
        format_version: crate::RPC_PRELUDE_FORMAT_VERSION,
        authorization_profile: crate::RPC_AUTH_PROFILE_V1,
        minimum_wire_version: crate::RPC_WIRE_VERSION_MIN,
        maximum_wire_version: crate::RPC_WIRE_VERSION_MAX,
        target_role,
        method,
        command_family,
        declared_payload_bytes,
    }
    .encode()
    .map_err(TransportError::Authorization)?;
    Ok(EncodedRequest {
        prelude,
        frame,
        response,
        method,
        command_family,
    })
}

/// Proof that the fixed prelude has passed version negotiation and ACL.
/// Runtime code must obtain this value before allocating the declared frame.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct AuthorizedPrelude {
    prelude: RpcPrelude,
    negotiated_profile: RpcNegotiatedProfile,
    expected_inner_method: InternalRpcMethod,
    declared_frame_bytes: usize,
}

impl AuthorizedPrelude {
    #[must_use]
    pub const fn prelude(self) -> RpcPrelude {
        self.prelude
    }

    #[must_use]
    pub const fn negotiated_profile(self) -> RpcNegotiatedProfile {
        self.negotiated_profile
    }

    /// Exact number of already-bounded bytes the runtime may now allocate and
    /// read. This value is unavailable until authorization succeeds.
    #[must_use]
    pub const fn declared_frame_bytes(self) -> usize {
        self.declared_frame_bytes
    }

    /// Decodes the existing length/CRC frame only after the ACL type gate.
    pub fn decode_request(self, frame_bytes: &[u8]) -> Result<AuthorizedRequest, TransportError> {
        if frame_bytes.len() != self.declared_frame_bytes {
            return Err(TransportError::DeclaredFrameLengthMismatch {
                declared: self.declared_frame_bytes,
                actual: frame_bytes.len(),
            });
        }
        let frame = InternalRpcFrame::decode(frame_bytes).map_err(TransportError::Frame)?;
        if frame.status != InternalRpcStatus::Ok {
            return Err(TransportError::RequestStatus(frame.status));
        }
        if frame.method != self.expected_inner_method {
            return Err(TransportError::InnerMethodMismatch {
                prelude_method: self.prelude.method,
                expected: self.expected_inner_method,
                actual: frame.method,
            });
        }
        Ok(AuthorizedRequest {
            prelude: self.prelude,
            negotiated_profile: self.negotiated_profile,
            request_id: frame.request_id,
            inner_method: frame.method,
            payload: frame.payload,
        })
    }
}

/// Decodes and authorizes only the fixed prelude. The runtime is responsible
/// for reading exactly 24 bytes before this call and for allocating the inner
/// frame only after this call succeeds.
pub fn authorize_request_prelude(
    identity: &WorkloadIdentity,
    listener_role: NodeRole,
    prelude_bytes: &[u8],
) -> Result<AuthorizedPrelude, TransportError> {
    let prelude = RpcPrelude::decode(prelude_bytes).map_err(TransportError::Authorization)?;
    let negotiated_profile = authorize_rpc(identity, listener_role, prelude).map_err(TransportError::Authorization)?;
    let declared_frame_bytes =
        usize::try_from(prelude.declared_payload_bytes).map_err(|_| TransportError::FrameForMethodTooLarge {
            actual: usize::MAX,
            maximum: request_frame_limit(prelude.method),
        })?;
    Ok(AuthorizedPrelude {
        prelude,
        negotiated_profile,
        expected_inner_method: internal_rpc_method_for(prelude.method),
        declared_frame_bytes,
    })
}

/// Application-dispatch request produced only after channel authentication,
/// prelude authorization, and inner-frame integrity/method checks.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct AuthorizedRequest {
    pub prelude: RpcPrelude,
    pub negotiated_profile: RpcNegotiatedProfile,
    pub request_id: RequestId,
    pub inner_method: InternalRpcMethod,
    pub payload: Vec<u8>,
}

impl AuthorizedRequest {
    /// Frames one handler response with the request's exact correlation and
    /// coarse method identity.
    pub fn encode_response(&self, status: InternalRpcStatus, payload: Vec<u8>) -> Result<Vec<u8>, TransportError> {
        encode_response_frame(
            self.request_id.clone(),
            self.inner_method,
            self.prelude.method,
            status,
            payload,
        )
    }
}

/// Exact client-side correlation and method expectation for one response.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ResponseExpectation {
    request_id: RequestId,
    method: InternalRpcMethod,
}

impl ResponseExpectation {
    #[must_use]
    pub fn request_id(&self) -> &RequestId {
        &self.request_id
    }

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

    pub fn decode(&self, bytes: &[u8]) -> Result<DecodedResponse, TransportError> {
        let frame = InternalRpcFrame::decode(bytes).map_err(TransportError::Frame)?;
        if frame.request_id != self.request_id {
            return Err(TransportError::ResponseRequestIdMismatch {
                expected: self.request_id.clone(),
                actual: frame.request_id,
            });
        }
        if frame.method != self.method {
            return Err(TransportError::ResponseMethodMismatch {
                expected: self.method,
                actual: frame.method,
            });
        }
        Ok(DecodedResponse {
            status: frame.status,
            payload: frame.payload,
        })
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct DecodedResponse {
    pub status: InternalRpcStatus,
    pub payload: Vec<u8>,
}

/// Frames a denial before the caller's inner request is read. Denials carry
/// no payload details and use one fixed, non-secret correlation token.
pub fn encode_transport_denial(method: RpcMethod, status: TransportDenialStatus) -> Result<Vec<u8>, TransportError> {
    let request_id = RequestId::new(TRANSPORT_DENIAL_REQUEST_ID)
        .map_err(|error| TransportError::FixedDenialRequestId(error.to_string()))?;
    encode_response_frame(
        request_id,
        internal_rpc_method_for(method),
        method,
        status.as_internal(),
        Vec::new(),
    )
}

/// Decodes a pre-dispatch denial and refuses ordinary handler statuses or a
/// substituted correlation/method.
pub fn decode_transport_denial(method: RpcMethod, bytes: &[u8]) -> Result<TransportDenialStatus, TransportError> {
    let frame = InternalRpcFrame::decode(bytes).map_err(TransportError::Frame)?;
    if frame.request_id.as_str() != TRANSPORT_DENIAL_REQUEST_ID {
        return Err(TransportError::DenialRequestIdMismatch(frame.request_id));
    }
    let expected_method = internal_rpc_method_for(method);
    if frame.method != expected_method {
        return Err(TransportError::ResponseMethodMismatch {
            expected: expected_method,
            actual: frame.method,
        });
    }
    if !frame.payload.is_empty() {
        return Err(TransportError::DenialPayload(frame.payload.len()));
    }
    TransportDenialStatus::from_internal(frame.status)
}

fn encode_response_frame(
    request_id: RequestId,
    inner_method: InternalRpcMethod,
    rpc_method: RpcMethod,
    status: InternalRpcStatus,
    payload: Vec<u8>,
) -> Result<Vec<u8>, TransportError> {
    let frame = InternalRpcFrame {
        format_version: INTERNAL_RPC_FORMAT_VERSION,
        status,
        request_id,
        method: inner_method,
        payload,
    }
    .encode()
    .map_err(TransportError::Frame)?;
    let maximum = response_frame_limit(rpc_method);
    if frame.len() > maximum {
        return Err(TransportError::FrameForMethodTooLarge {
            actual: frame.len(),
            maximum,
        });
    }
    Ok(frame)
}

#[must_use]
pub const fn request_frame_limit(method: RpcMethod) -> usize {
    match method {
        RpcMethod::RaftInstallSnapshot => RPC_MAX_PAYLOAD_BYTES,
        RpcMethod::MetastoreSnapshot => RPC_METASTORE_SNAPSHOT_REQUEST_FRAME_BYTES,
        RpcMethod::SearchShardExecute => RPC_SEARCH_REQUEST_FRAME_BYTES,
        _ => RPC_MAX_CONTROL_PAYLOAD_BYTES,
    }
}

#[must_use]
pub const fn response_frame_limit(method: RpcMethod) -> usize {
    match method {
        RpcMethod::RaftInstallSnapshot => RPC_MAX_PAYLOAD_BYTES,
        RpcMethod::MetastoreSnapshot => RPC_METASTORE_SNAPSHOT_RESPONSE_FRAME_BYTES,
        RpcMethod::WriterRealtimeRead | RpcMethod::SearchCoordinate | RpcMethod::SearchShardExecute => {
            crate::RPC_SEARCH_RESPONSE_FRAME_BYTES
        }
        _ => RPC_MAX_CONTROL_PAYLOAD_BYTES,
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum TransportError {
    Authorization(RpcAuthorizationError),
    DeclaredFrameLengthMismatch {
        declared: usize,
        actual: usize,
    },
    DenialPayload(usize),
    DenialRequestIdMismatch(RequestId),
    FixedDenialRequestId(String),
    Frame(InternalRpcError),
    FrameForMethodTooLarge {
        actual: usize,
        maximum: usize,
    },
    InnerMethodMismatch {
        prelude_method: RpcMethod,
        expected: InternalRpcMethod,
        actual: InternalRpcMethod,
    },
    InvalidDenialStatus(InternalRpcStatus),
    RequestStatus(InternalRpcStatus),
    ReservedRequestId,
    ResponseMethodMismatch {
        expected: InternalRpcMethod,
        actual: InternalRpcMethod,
    },
    ResponseRequestIdMismatch {
        expected: RequestId,
        actual: RequestId,
    },
    ResponseLengthPrefix {
        actual: usize,
    },
    ResponseLengthTooLarge(usize),
}

impl TransportError {
    #[must_use]
    pub const fn status(&self) -> InternalRpcStatus {
        match self {
            Self::Authorization(error) => authorization_error_status(error),
            Self::Frame(InternalRpcError::UnsupportedVersion(_)) => InternalRpcStatus::UnknownVersion,
            Self::Frame(InternalRpcError::FrameTooLarge(_) | InternalRpcError::PayloadTooLarge { .. })
            | Self::FrameForMethodTooLarge { .. } => InternalRpcStatus::Malformed,
            Self::DeclaredFrameLengthMismatch { .. }
            | Self::DenialPayload(_)
            | Self::DenialRequestIdMismatch(_)
            | Self::FixedDenialRequestId(_)
            | Self::Frame(_)
            | Self::InnerMethodMismatch { .. }
            | Self::InvalidDenialStatus(_)
            | Self::RequestStatus(_)
            | Self::ReservedRequestId
            | Self::ResponseLengthPrefix { .. }
            | Self::ResponseLengthTooLarge(_)
            | Self::ResponseMethodMismatch { .. }
            | Self::ResponseRequestIdMismatch { .. } => InternalRpcStatus::Malformed,
        }
    }
}

impl Display for TransportError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::Authorization(error) => write!(formatter, "RPC authorization failed: {error}"),
            Self::DeclaredFrameLengthMismatch { declared, actual } => write!(
                formatter,
                "RPC prelude declares {declared} frame bytes but received {actual}"
            ),
            Self::DenialPayload(actual) => write!(formatter, "transport denial has {actual} payload bytes"),
            Self::DenialRequestIdMismatch(_) => formatter.write_str("transport denial used an unexpected request ID"),
            Self::FixedDenialRequestId(reason) => {
                write!(formatter, "fixed transport denial request ID is invalid: {reason}")
            }
            Self::Frame(error) => write!(formatter, "inner RPC frame failed: {error}"),
            Self::FrameForMethodTooLarge { actual, maximum } => {
                write!(formatter, "method frame is {actual} bytes; maximum is {maximum}")
            }
            Self::InnerMethodMismatch {
                prelude_method,
                expected,
                actual,
            } => write!(
                formatter,
                "prelude method {prelude_method:?} requires inner method {expected:?}, got {actual:?}"
            ),
            Self::InvalidDenialStatus(status) => {
                write!(formatter, "RPC status {status:?} is not a transport denial")
            }
            Self::RequestStatus(status) => write!(formatter, "request frame has response status {status:?}"),
            Self::ReservedRequestId => formatter.write_str("request used the reserved transport denial request ID"),
            Self::ResponseMethodMismatch { expected, actual } => write!(
                formatter,
                "response method mismatch: expected {expected:?}, got {actual:?}"
            ),
            Self::ResponseLengthPrefix { actual } => {
                write!(formatter, "response length prefix has {actual} bytes")
            }
            Self::ResponseLengthTooLarge(actual) => {
                write!(formatter, "response declares {actual} bytes")
            }
            Self::ResponseRequestIdMismatch { .. } => formatter.write_str("response request ID mismatch"),
        }
    }
}

impl Error for TransportError {}

#[cfg(test)]
mod tests {
    use lodestone_types::{
        ClusterIdentity, CredentialGeneration, NodeId, NodeIncarnation, NodeRole, NodeToken, RecoveryEra,
        RecoveryGeneration,
    };

    use super::{
        ChannelIdentityError, TRANSPORT_DENIAL_REQUEST_ID, TransportDenialStatus, TransportError,
        authorization_error_status, authorize_request_prelude, channel_identity_error_status, decode_response_length,
        decode_transport_denial, encode_request, encode_transport_denial, internal_rpc_method_for, request_frame_limit,
        response_frame_limit,
    };
    use crate::{
        INTERNAL_RPC_FORMAT_VERSION, InternalRpcError, InternalRpcFrame, InternalRpcMethod, InternalRpcStatus,
        MetastoreCommandFamily, RequestId, RpcAuthorizationError, RpcMethod, RpcPrelude, WorkloadIdentity,
    };

    fn identity(role: NodeRole) -> Result<WorkloadIdentity, Box<dyn std::error::Error>> {
        let cluster = ClusterIdentity::new(RecoveryEra::new(1)?, RecoveryGeneration::genesis());
        Ok(WorkloadIdentity {
            format_version: 1,
            cluster_id: lodestone_types::ClusterStorageId::derive_from_genesis(cluster)?,
            node_token: NodeToken::new(RecoveryEra::new(1)?, NodeId::new(7), NodeIncarnation::new(1)?)?,
            node_role: role,
            credential_generation: CredentialGeneration::new(1),
            credential_binding_digest: [9; 32],
        })
    }

    #[test]
    fn canonical_method_mapping_covers_every_method() {
        let cases = [
            (RpcMethod::RaftVote, InternalRpcMethod::RaftVote),
            (RpcMethod::RaftAppend, InternalRpcMethod::RaftAppend),
            (RpcMethod::RaftInstallSnapshot, InternalRpcMethod::RaftInstallSnapshot),
            (RpcMethod::MetastoreCommand, InternalRpcMethod::MetastoreCommand),
            (RpcMethod::MetastoreSnapshot, InternalRpcMethod::MetastoreSnapshot),
            (RpcMethod::MetastorePointRead, InternalRpcMethod::MetastoreSnapshot),
            (RpcMethod::MetastoreWatch, InternalRpcMethod::MetastoreSnapshot),
            (RpcMethod::NodeSelfHealth, InternalRpcMethod::Health),
            (
                RpcMethod::OrchestratorPlacementIntent,
                InternalRpcMethod::MetastoreCommand,
            ),
            (
                RpcMethod::OrchestratorCatalogIntent,
                InternalRpcMethod::MetastoreCommand,
            ),
            (RpcMethod::ProposePlacementRecovery, InternalRpcMethod::MetastoreCommand),
            (RpcMethod::WriterIngest, InternalRpcMethod::WriterIngest),
            (RpcMethod::WriterRealtimeRead, InternalRpcMethod::Search),
            (RpcMethod::WriterDrainAdvisory, InternalRpcMethod::WriterFence),
            (RpcMethod::TaskDestinationMutation, InternalRpcMethod::WriterIngest),
            (RpcMethod::SearchCoordinate, InternalRpcMethod::Search),
            (RpcMethod::SearchShardExecute, InternalRpcMethod::Search),
            (RpcMethod::SearchMount, InternalRpcMethod::MountManifest),
            (RpcMethod::InternalHealth, InternalRpcMethod::Health),
            (RpcMethod::CompactorExecuteTask, InternalRpcMethod::WriterIngest),
            (RpcMethod::CoreApiAuthority, InternalRpcMethod::MetastoreCommand),
        ];
        for (method, expected) in cases {
            assert_eq!(internal_rpc_method_for(method), expected);
        }
    }

    #[test]
    fn every_noncanonical_exact_to_coarse_pair_is_rejected() -> Result<(), Box<dyn std::error::Error>> {
        use NodeRole::{Compactor, Gateway, Indexer, MetastoreVoter, Orchestrator, Placement, Searcher};

        let exact_cases = [
            (MetastoreVoter, MetastoreVoter, RpcMethod::RaftVote, None),
            (MetastoreVoter, MetastoreVoter, RpcMethod::RaftAppend, None),
            (MetastoreVoter, MetastoreVoter, RpcMethod::RaftInstallSnapshot, None),
            (
                Gateway,
                MetastoreVoter,
                RpcMethod::MetastoreCommand,
                Some(MetastoreCommandFamily::Catalog),
            ),
            (MetastoreVoter, MetastoreVoter, RpcMethod::MetastoreSnapshot, None),
            (Gateway, MetastoreVoter, RpcMethod::MetastorePointRead, None),
            (Gateway, MetastoreVoter, RpcMethod::MetastoreWatch, None),
            (Gateway, MetastoreVoter, RpcMethod::NodeSelfHealth, None),
            (Orchestrator, Placement, RpcMethod::OrchestratorPlacementIntent, None),
            (Orchestrator, Gateway, RpcMethod::OrchestratorCatalogIntent, None),
            (Orchestrator, MetastoreVoter, RpcMethod::ProposePlacementRecovery, None),
            (Gateway, Indexer, RpcMethod::WriterIngest, None),
            (Gateway, Indexer, RpcMethod::WriterRealtimeRead, None),
            (Placement, Indexer, RpcMethod::WriterDrainAdvisory, None),
            (Compactor, Indexer, RpcMethod::TaskDestinationMutation, None),
            (Placement, Compactor, RpcMethod::CompactorExecuteTask, None),
            (Gateway, MetastoreVoter, RpcMethod::CoreApiAuthority, None),
            (Gateway, Searcher, RpcMethod::SearchCoordinate, None),
            (Searcher, Searcher, RpcMethod::SearchShardExecute, None),
            (Placement, Searcher, RpcMethod::SearchMount, None),
            (Gateway, Indexer, RpcMethod::InternalHealth, None),
        ];
        let coarse_methods = [
            InternalRpcMethod::RaftAppend,
            InternalRpcMethod::RaftVote,
            InternalRpcMethod::RaftInstallSnapshot,
            InternalRpcMethod::MetastoreCommand,
            InternalRpcMethod::MetastoreSnapshot,
            InternalRpcMethod::WriterIngest,
            InternalRpcMethod::WriterFence,
            InternalRpcMethod::Search,
            InternalRpcMethod::MountManifest,
            InternalRpcMethod::Health,
        ];
        let expected_rejected = exact_cases.len() * coarse_methods.len().saturating_sub(1);
        let mut rejected = 0_usize;
        for (source, target, method, family) in exact_cases {
            let expected = internal_rpc_method_for(method);
            for coarse in coarse_methods {
                if coarse == expected {
                    continue;
                }
                let frame = InternalRpcFrame::request(RequestId::new("noncanonical-exhaustive")?, coarse, Vec::new())
                    .encode()?;
                let prelude = RpcPrelude {
                    format_version: 1,
                    authorization_profile: 1,
                    minimum_wire_version: 1,
                    maximum_wire_version: 1,
                    target_role: target,
                    method,
                    command_family: family,
                    declared_payload_bytes: u32::try_from(frame.len())?,
                }
                .encode()?;
                let authorized = authorize_request_prelude(&identity(source)?, target, &prelude)?;
                assert!(matches!(
                    authorized.decode_request(&frame),
                    Err(TransportError::InnerMethodMismatch { .. })
                ));
                rejected += 1;
            }
        }
        assert_eq!(rejected, expected_rejected);
        Ok(())
    }

    #[test]
    fn request_round_trip_authorizes_before_frame_decode_and_preserves_retry_identity()
    -> Result<(), Box<dyn std::error::Error>> {
        let request_id = RequestId::new("correlation-only-7")?;
        let encoded = encode_request(
            NodeRole::Indexer,
            RpcMethod::WriterIngest,
            None,
            request_id.clone(),
            b"body".to_vec(),
        )?;
        let retry = encoded.clone();
        assert_eq!(retry.response_expectation().request_id(), &request_id);
        assert_eq!(retry.prelude(), encoded.prelude());
        assert_eq!(retry.frame(), encoded.frame());

        let authorized =
            authorize_request_prelude(&identity(NodeRole::Gateway)?, NodeRole::Indexer, encoded.prelude())?;
        assert_eq!(authorized.declared_frame_bytes(), encoded.frame().len());
        let decoded = authorized.decode_request(encoded.frame())?;
        assert_eq!(decoded.request_id, request_id);
        assert_eq!(decoded.payload, b"body");
        Ok(())
    }

    #[test]
    fn acl_denial_precedes_any_inner_frame_decode_or_allocation() -> Result<(), Box<dyn std::error::Error>> {
        let prelude = RpcPrelude {
            format_version: 1,
            authorization_profile: 1,
            minimum_wire_version: 1,
            maximum_wire_version: 1,
            target_role: NodeRole::MetastoreVoter,
            method: RpcMethod::MetastoreCommand,
            command_family: Some(MetastoreCommandFamily::WriterLease),
            declared_payload_bytes: u32::MAX,
        }
        .encode()?;
        let error = authorize_request_prelude(&identity(NodeRole::Gateway)?, NodeRole::MetastoreVoter, &prelude);
        assert!(matches!(
            error,
            Err(TransportError::Authorization(RpcAuthorizationError::PermissionDenied))
        ));
        Ok(())
    }

    #[test]
    fn authorized_decode_refuses_inner_method_length_crc_and_request_status_mismatches()
    -> Result<(), Box<dyn std::error::Error>> {
        let frame = InternalRpcFrame::request(RequestId::new("mismatch")?, InternalRpcMethod::Search, b"body".to_vec())
            .encode()?;
        let prelude = RpcPrelude {
            format_version: 1,
            authorization_profile: 1,
            minimum_wire_version: 1,
            maximum_wire_version: 1,
            target_role: NodeRole::Indexer,
            method: RpcMethod::WriterIngest,
            command_family: None,
            declared_payload_bytes: u32::try_from(frame.len())?,
        }
        .encode()?;
        let authorized = authorize_request_prelude(&identity(NodeRole::Gateway)?, NodeRole::Indexer, &prelude)?;
        assert!(matches!(
            authorized.decode_request(&frame),
            Err(TransportError::InnerMethodMismatch { .. })
        ));

        let encoded = encode_request(
            NodeRole::Indexer,
            RpcMethod::WriterIngest,
            None,
            RequestId::new("corrupt")?,
            b"body".to_vec(),
        )?;
        let authorized =
            authorize_request_prelude(&identity(NodeRole::Gateway)?, NodeRole::Indexer, encoded.prelude())?;
        assert!(matches!(
            authorized.decode_request(&encoded.frame()[..encoded.frame().len() - 1]),
            Err(TransportError::DeclaredFrameLengthMismatch { .. })
        ));
        let mut corrupt = encoded.frame().to_vec();
        let Some(last) = corrupt.last_mut() else {
            return Err("encoded frame was empty".into());
        };
        *last ^= 0xff;
        let authorized =
            authorize_request_prelude(&identity(NodeRole::Gateway)?, NodeRole::Indexer, encoded.prelude())?;
        assert!(matches!(
            authorized.decode_request(&corrupt),
            Err(TransportError::Frame(InternalRpcError::ChecksumMismatch { .. }))
        ));

        let response_shaped_request = InternalRpcFrame {
            format_version: INTERNAL_RPC_FORMAT_VERSION,
            status: InternalRpcStatus::Unavailable,
            request_id: RequestId::new("status")?,
            method: InternalRpcMethod::WriterIngest,
            payload: Vec::new(),
        }
        .encode()?;
        let status_prelude = RpcPrelude {
            format_version: 1,
            authorization_profile: 1,
            minimum_wire_version: 1,
            maximum_wire_version: 1,
            target_role: NodeRole::Indexer,
            method: RpcMethod::WriterIngest,
            command_family: None,
            declared_payload_bytes: u32::try_from(response_shaped_request.len())?,
        }
        .encode()?;
        let authorized = authorize_request_prelude(&identity(NodeRole::Gateway)?, NodeRole::Indexer, &status_prelude)?;
        assert_eq!(
            authorized.decode_request(&response_shaped_request),
            Err(TransportError::RequestStatus(InternalRpcStatus::Unavailable))
        );
        Ok(())
    }

    #[test]
    fn response_checks_exact_request_id_and_method() -> Result<(), Box<dyn std::error::Error>> {
        let encoded = encode_request(
            NodeRole::Indexer,
            RpcMethod::WriterIngest,
            None,
            RequestId::new("response-7")?,
            Vec::new(),
        )?;
        let authorized =
            authorize_request_prelude(&identity(NodeRole::Gateway)?, NodeRole::Indexer, encoded.prelude())?
                .decode_request(encoded.frame())?;
        let response = authorized.encode_response(InternalRpcStatus::Ok, b"done".to_vec())?;
        let decoded = encoded.response_expectation().decode(&response)?;
        assert_eq!(decoded.status, InternalRpcStatus::Ok);
        assert_eq!(decoded.payload, b"done");

        let wrong_id = InternalRpcFrame {
            format_version: INTERNAL_RPC_FORMAT_VERSION,
            status: InternalRpcStatus::Ok,
            request_id: RequestId::new("substituted")?,
            method: InternalRpcMethod::WriterIngest,
            payload: Vec::new(),
        }
        .encode()?;
        assert!(matches!(
            encoded.response_expectation().decode(&wrong_id),
            Err(TransportError::ResponseRequestIdMismatch { .. })
        ));

        let wrong_method = InternalRpcFrame {
            format_version: INTERNAL_RPC_FORMAT_VERSION,
            status: InternalRpcStatus::Ok,
            request_id: RequestId::new("response-7")?,
            method: InternalRpcMethod::Search,
            payload: Vec::new(),
        }
        .encode()?;
        assert!(matches!(
            encoded.response_expectation().decode(&wrong_method),
            Err(TransportError::ResponseMethodMismatch { .. })
        ));
        Ok(())
    }

    #[test]
    fn typed_denials_use_fixed_non_secret_identity_and_refuse_handler_statuses()
    -> Result<(), Box<dyn std::error::Error>> {
        let denial = encode_transport_denial(RpcMethod::WriterIngest, TransportDenialStatus::PermissionDenied)?;
        let frame = InternalRpcFrame::decode(&denial)?;
        assert_eq!(frame.request_id.as_str(), TRANSPORT_DENIAL_REQUEST_ID);
        assert!(frame.payload.is_empty());
        assert_eq!(
            decode_transport_denial(RpcMethod::WriterIngest, &denial)?,
            TransportDenialStatus::PermissionDenied
        );
        assert_eq!(
            TransportDenialStatus::from_internal(InternalRpcStatus::NotLeader),
            Err(TransportError::InvalidDenialStatus(InternalRpcStatus::NotLeader))
        );
        Ok(())
    }

    #[test]
    fn reserved_transport_denial_request_id_is_never_admitted() -> Result<(), Box<dyn std::error::Error>> {
        let result = encode_request(
            NodeRole::Indexer,
            RpcMethod::WriterIngest,
            None,
            RequestId::new(TRANSPORT_DENIAL_REQUEST_ID)?,
            Vec::new(),
        );
        let error = match result {
            Ok(_) => return Err("reserved denial correlation was caller-selectable".into()),
            Err(error) => error,
        };
        assert_eq!(error, TransportError::ReservedRequestId);
        Ok(())
    }

    #[test]
    fn status_mappings_are_fixed_cardinality_and_detail_free() {
        let channel_cases = [
            (
                ChannelIdentityError::Unauthenticated,
                InternalRpcStatus::Unauthenticated,
            ),
            (
                ChannelIdentityError::VerifierUnavailable,
                InternalRpcStatus::Unavailable,
            ),
            (ChannelIdentityError::WrongCluster, InternalRpcStatus::WrongCluster),
            (ChannelIdentityError::WrongRole, InternalRpcStatus::WrongRole),
            (
                ChannelIdentityError::StaleNodeIncarnation,
                InternalRpcStatus::StaleNodeIncarnation,
            ),
            (
                ChannelIdentityError::CredentialMismatch,
                InternalRpcStatus::CredentialMismatch,
            ),
            (
                ChannelIdentityError::UnsupportedIdentityVersion,
                InternalRpcStatus::UnknownVersion,
            ),
        ];
        for (error, expected) in channel_cases {
            assert_eq!(channel_identity_error_status(error), expected);
        }
        assert_eq!(
            authorization_error_status(&RpcAuthorizationError::IncompatibleAuthProfile { offered: 7 }),
            InternalRpcStatus::IncompatibleAuthProfile
        );
        assert_eq!(
            authorization_error_status(&RpcAuthorizationError::PayloadTooLarge {
                actual: usize::MAX,
                maximum: 1,
            }),
            InternalRpcStatus::Malformed
        );
    }

    #[test]
    fn response_length_decoder_refuses_truncated_and_oversized_prefixes() {
        assert_eq!(decode_response_length(&[0, 0, 0, 9]), Ok(9));
        assert_eq!(
            decode_response_length(&[0, 0, 0]),
            Err(TransportError::ResponseLengthPrefix { actual: 3 })
        );
        assert!(matches!(
            decode_response_length(&u32::MAX.to_be_bytes()),
            Err(TransportError::ResponseLengthTooLarge(_))
        ));
    }

    #[test]
    fn oversized_control_frame_is_refused_at_construction() -> Result<(), Box<dyn std::error::Error>> {
        let payload = vec![0_u8; crate::RPC_MAX_CONTROL_PAYLOAD_BYTES];
        let result = encode_request(
            NodeRole::Indexer,
            RpcMethod::WriterIngest,
            None,
            RequestId::new("oversized")?,
            payload,
        );
        assert!(matches!(result, Err(TransportError::FrameForMethodTooLarge { .. })));
        Ok(())
    }

    #[test]
    fn metastore_snapshot_uses_asymmetric_frame_caps() -> Result<(), Box<dyn std::error::Error>> {
        assert_eq!(
            request_frame_limit(RpcMethod::MetastoreSnapshot),
            crate::RPC_METASTORE_SNAPSHOT_REQUEST_FRAME_BYTES
        );
        assert_eq!(
            response_frame_limit(RpcMethod::MetastoreSnapshot),
            crate::RPC_METASTORE_SNAPSHOT_RESPONSE_FRAME_BYTES
        );
        let valid = encode_request(
            NodeRole::MetastoreVoter,
            RpcMethod::MetastoreSnapshot,
            None,
            RequestId::new(format!("p6-recovery-snapshot-{}", u64::MAX))?,
            vec![0; 42],
        )?;
        assert!(valid.frame().len() <= crate::RPC_METASTORE_SNAPSHOT_REQUEST_FRAME_BYTES);
        let result = encode_request(
            NodeRole::MetastoreVoter,
            RpcMethod::MetastoreSnapshot,
            None,
            RequestId::new("oversized-snapshot-request")?,
            vec![0; crate::RPC_METASTORE_SNAPSHOT_REQUEST_FRAME_BYTES],
        );
        assert!(matches!(result, Err(TransportError::FrameForMethodTooLarge { .. })));
        Ok(())
    }

    #[test]
    fn search_data_methods_keep_bounded_directional_frames() {
        for method in [RpcMethod::WriterRealtimeRead, RpcMethod::SearchCoordinate] {
            assert_eq!(request_frame_limit(method), crate::RPC_MAX_CONTROL_PAYLOAD_BYTES);
            assert_eq!(response_frame_limit(method), crate::RPC_SEARCH_RESPONSE_FRAME_BYTES);
        }
        assert_eq!(
            request_frame_limit(RpcMethod::SearchShardExecute),
            crate::RPC_SEARCH_REQUEST_FRAME_BYTES
        );
        assert_eq!(
            response_frame_limit(RpcMethod::SearchShardExecute),
            crate::RPC_SEARCH_RESPONSE_FRAME_BYTES
        );
    }
}