polyc-state-connect 2026.8.3

//! Carrying a typed outcome across the wire, and reading it back.
//!
//! A transport status code is not a State outcome. `permission_denied` says
//! nothing about whether a retry is safe, and `deadline_exceeded` says nothing
//! about whether the operation committed — the kernel's own variant does. So a
//! refusal travels as a Connect error detail holding the full
//! [`StateError`], and the caller reconstructs the identical variant on the
//! far side rather than inferring one from the code.
//!
//! The code still matters: it is what a peer that does not speak this protocol
//! sees, and it is what the transport itself produces when it refuses a
//! message before any handler runs. [`from_connect_error`] therefore reads the
//! detail when one is present and falls back to the code when none is, so an
//! oversized message the transport refused and a bound the module refused
//! arrive as the same typed outcome.

use base64::Engine as _;
use connectrpc::{ConnectError, ErrorCode, ErrorDetail};
use polyc_proto::proto::polychrome::state::v1 as pb;
use polyc_state::{
    command::{FencingToken, ObservedState, Precondition},
    digest::ContentDigest,
    error::{AmbiguityReason, BoundKind, OutageReach, StateError},
    id::{CommandId, OperationFamily, PartitionId},
    revision::{CommitRoot, JournalPosition, Revision},
};

use crate::wire::{Kernel, duration_from_nanos, nanos_from_duration};

/// The type name the typed outcome travels under, as a Connect error detail.
pub const STATE_ERROR_DETAIL_TYPE: &str = "polychrome.state.v1.StateErrorDetail";

/// Returns the transport status code that best describes `error`.
///
/// The code is a courtesy for peers that read only codes; the detail is the
/// contract. Both are set on every refusal this crate produces.
#[must_use]
pub const fn code_for(error: &StateError) -> ErrorCode {
    match error {
        StateError::RevisionConflict { .. }
        | StateError::IncarnationConflict { .. }
        | StateError::StaleFence { .. } => ErrorCode::FailedPrecondition,
        StateError::DuplicateCommand { .. } | StateError::PartitionHeld { .. } => {
            ErrorCode::Aborted
        }
        StateError::DigestConflict { .. } => ErrorCode::AlreadyExists,
        StateError::DeadlineExpired { .. } => ErrorCode::DeadlineExceeded,
        StateError::BoundsExceeded { .. } => ErrorCode::ResourceExhausted,
        StateError::Cancelled { .. } => ErrorCode::Canceled,
        // An outage and an unknown commit share one code because the code has
        // no room for the distinction — `unavailable` is what a peer reading
        // codes alone can act on for both. The detail is what keeps them apart,
        // which is precisely why the detail is the contract.
        StateError::Unavailable { .. } | StateError::AmbiguousOutcome { .. } => {
            ErrorCode::Unavailable
        }
        StateError::Denied { .. } => ErrorCode::PermissionDenied,
        // The one code that says "the thing you asked for is outside what still
        // exists" rather than "you asked wrongly": both a compacted cursor and
        // a retired revision were valid when issued, and the data under them is
        // what went away. OutOfRange also preserves the terminal retry class for
        // an older peer that cannot decode the typed detail.
        StateError::CompactedRange { .. } | StateError::RetiredRevision { .. } => {
            ErrorCode::OutOfRange
        }
        StateError::Malformed { .. } => ErrorCode::InvalidArgument,
    }
}

impl From<Kernel<&StateError>> for pb::StateErrorDetail {
    // A flat, exhaustive match: one arm per outcome, every field named. Split
    // across helpers it would stop failing to compile when a variant is added.
    #[allow(clippy::too_many_lines)]
    fn from(value: Kernel<&StateError>) -> Self {
        use pb::__buffa::oneof::state_error_detail::Outcome;
        let unknown = buffa::UnknownFields::default;
        let outcome = match value.0 {
            StateError::RevisionConflict {
                command_id,
                expected,
                observed,
            } => Outcome::from(pb::RevisionConflictDetail {
                command_id: command_id.as_str().to_owned(),
                expected: buffa::MessageField::some(pb::Precondition::from(Kernel(*expected))),
                observed: buffa::MessageField::some(pb::ObservedState::from(Kernel(*observed))),
                __buffa_unknown_fields: unknown(),
            }),
            StateError::IncarnationConflict {
                command_id,
                partition,
                expected,
                observed,
            } => Outcome::from(pb::IncarnationConflictDetail {
                command_id: command_id.as_str().to_owned(),
                partition: partition.as_str().to_owned(),
                expected_root: expected.as_bytes().to_vec(),
                observed_root: observed.map(|root| root.as_bytes().to_vec()),
                __buffa_unknown_fields: unknown(),
            }),
            StateError::StaleFence {
                command_id,
                presented,
                current,
            } => Outcome::from(pb::StaleFenceDetail {
                command_id: command_id.as_str().to_owned(),
                presented: presented.get(),
                current: current.get(),
                __buffa_unknown_fields: unknown(),
            }),
            StateError::DuplicateCommand { command_id } => {
                Outcome::from(pb::DuplicateCommandDetail {
                    command_id: command_id.as_str().to_owned(),
                    __buffa_unknown_fields: unknown(),
                })
            }
            StateError::PartitionHeld {
                partition,
                command_id,
            } => Outcome::from(pb::PartitionHeldDetail {
                partition: partition.as_str().to_owned(),
                command_id: command_id.as_str().to_owned(),
                __buffa_unknown_fields: unknown(),
            }),
            StateError::DigestConflict {
                command_id,
                recorded,
                presented,
            } => Outcome::from(pb::DigestConflictDetail {
                command_id: command_id.as_str().to_owned(),
                recorded: recorded.as_bytes().to_vec(),
                presented: presented.as_bytes().to_vec(),
                __buffa_unknown_fields: unknown(),
            }),
            StateError::DeadlineExpired { family, overrun } => {
                Outcome::from(pb::DeadlineExpiredDetail {
                    family: family.as_str().to_owned(),
                    overrun_nanos: nanos_from_duration(*overrun),
                    __buffa_unknown_fields: unknown(),
                })
            }
            StateError::BoundsExceeded {
                bound,
                limit,
                requested,
            } => Outcome::from(pb::BoundsExceededDetail {
                bound: pb::BoundKind::from(Kernel(*bound)).into(),
                limit: *limit,
                requested: *requested,
                __buffa_unknown_fields: unknown(),
            }),
            StateError::Unavailable { family, reach } => Outcome::from(pb::UnavailableDetail {
                family: family.as_str().to_owned(),
                reach: pb::OutageReach::from(Kernel(*reach)).into(),
                __buffa_unknown_fields: unknown(),
            }),
            StateError::Cancelled { family } => Outcome::from(pb::CancelledDetail {
                family: family.as_str().to_owned(),
                __buffa_unknown_fields: unknown(),
            }),
            StateError::AmbiguousOutcome { command_id, reason } => {
                Outcome::from(pb::AmbiguousOutcomeDetail {
                    command_id: command_id.as_str().to_owned(),
                    reason: pb::AmbiguityReason::from(Kernel(*reason)).into(),
                    __buffa_unknown_fields: unknown(),
                })
            }
            StateError::Denied { family } => Outcome::from(pb::DeniedDetail {
                family: family.as_str().to_owned(),
                __buffa_unknown_fields: unknown(),
            }),
            StateError::CompactedRange {
                partition,
                requested,
                earliest,
            } => Outcome::from(pb::CompactedRangeDetail {
                partition: partition.as_str().to_owned(),
                requested: requested.get(),
                earliest: earliest.get(),
                __buffa_unknown_fields: unknown(),
            }),
            StateError::RetiredRevision {
                requested,
                earliest,
            } => Outcome::from(pb::RetiredRevisionDetail {
                requested: requested.get(),
                earliest: earliest.get(),
                __buffa_unknown_fields: unknown(),
            }),
            StateError::Malformed { field, reason } => Outcome::from(pb::MalformedDetail {
                field: field.clone(),
                reason: reason.clone(),
                __buffa_unknown_fields: unknown(),
            }),
        };
        Self {
            outcome: Some(outcome),
            __buffa_unknown_fields: unknown(),
        }
    }
}

impl From<Kernel<BoundKind>> for pb::BoundKind {
    fn from(value: Kernel<BoundKind>) -> Self {
        match value.0 {
            BoundKind::PayloadBytes => Self::BOUND_KIND_PAYLOAD_BYTES,
            BoundKind::CommandRecords => Self::BOUND_KIND_COMMAND_RECORDS,
            BoundKind::PageRecords => Self::BOUND_KIND_PAGE_RECORDS,
            BoundKind::ChunkRecords => Self::BOUND_KIND_CHUNK_RECORDS,
        }
    }
}

impl From<Kernel<OutageReach>> for pb::OutageReach {
    fn from(value: Kernel<OutageReach>) -> Self {
        match value.0 {
            OutageReach::NeverDispatched => Self::OUTAGE_REACH_NEVER_DISPATCHED,
            OutageReach::PossiblyApplied => Self::OUTAGE_REACH_POSSIBLY_APPLIED,
        }
    }
}

impl From<Kernel<AmbiguityReason>> for pb::AmbiguityReason {
    fn from(value: Kernel<AmbiguityReason>) -> Self {
        match value.0 {
            AmbiguityReason::ResponseLost => Self::AMBIGUITY_REASON_RESPONSE_LOST,
            AmbiguityReason::CommitUnknown => Self::AMBIGUITY_REASON_COMMIT_UNKNOWN,
            AmbiguityReason::EffectDeliveryUnknown => {
                Self::AMBIGUITY_REASON_EFFECT_DELIVERY_UNKNOWN
            }
        }
    }
}

/// Reads a bound kind off the wire, defaulting to the payload bound when the
/// sender named none.
fn bound_kind(value: buffa::EnumValue<pb::BoundKind>) -> BoundKind {
    match value.as_known() {
        Some(pb::BoundKind::BOUND_KIND_COMMAND_RECORDS) => BoundKind::CommandRecords,
        Some(pb::BoundKind::BOUND_KIND_PAGE_RECORDS) => BoundKind::PageRecords,
        Some(pb::BoundKind::BOUND_KIND_CHUNK_RECORDS) => BoundKind::ChunkRecords,
        Some(pb::BoundKind::BOUND_KIND_PAYLOAD_BYTES | pb::BoundKind::BOUND_KIND_UNSPECIFIED)
        | None => BoundKind::PayloadBytes,
    }
}

/// Reads an outage's reach off the wire, defaulting to a request that may
/// already have applied.
///
/// A sender this build does not understand — an older peer, or a newer one
/// naming a reach this build has no name for — must never be read as proof
/// that nothing landed. Resolving the unknown to the weaker claim keeps a
/// version skew from turning an ambiguous outcome into a retryable one.
fn outage_reach(value: buffa::EnumValue<pb::OutageReach>) -> OutageReach {
    match value.as_known() {
        Some(pb::OutageReach::OUTAGE_REACH_NEVER_DISPATCHED) => OutageReach::NeverDispatched,
        Some(
            pb::OutageReach::OUTAGE_REACH_POSSIBLY_APPLIED
            | pb::OutageReach::OUTAGE_REACH_UNSPECIFIED,
        )
        | None => OutageReach::PossiblyApplied,
    }
}

/// Reads an ambiguity reason off the wire, defaulting to an unknown commit
/// when the sender named none — the weakest claim, and the only safe one.
fn ambiguity_reason(value: buffa::EnumValue<pb::AmbiguityReason>) -> AmbiguityReason {
    match value.as_known() {
        Some(pb::AmbiguityReason::AMBIGUITY_REASON_RESPONSE_LOST) => AmbiguityReason::ResponseLost,
        Some(pb::AmbiguityReason::AMBIGUITY_REASON_EFFECT_DELIVERY_UNKNOWN) => {
            AmbiguityReason::EffectDeliveryUnknown
        }
        Some(
            pb::AmbiguityReason::AMBIGUITY_REASON_COMMIT_UNKNOWN
            | pb::AmbiguityReason::AMBIGUITY_REASON_UNSPECIFIED,
        )
        | None => AmbiguityReason::CommitUnknown,
    }
}

/// Reads a digest out of a detail, falling back to the zero digest when the
/// sender's bytes were the wrong width.
///
/// A conflict detail whose digests could not be parsed is still a conflict:
/// downgrading it to "malformed" would hand the caller a terminal outcome with
/// the wrong meaning.
fn digest_or_zero(bytes: &[u8]) -> ContentDigest {
    <[u8; ContentDigest::LEN]>::try_from(bytes).map_or_else(
        |_| ContentDigest::from_bytes([0; ContentDigest::LEN]),
        ContentDigest::from_bytes,
    )
}

impl TryFrom<pb::StateErrorDetail> for Kernel<StateError> {
    type Error = pb::StateErrorDetail;

    /// Rebuilds the typed outcome the far side named.
    ///
    /// # Errors
    ///
    /// Returns the detail unchanged when its `outcome` names no variant — a
    /// peer speaking a protocol this build does not know, which the caller
    /// resolves by falling back to the transport code.
    fn try_from(value: pb::StateErrorDetail) -> Result<Self, Self::Error> {
        use pb::__buffa::oneof::state_error_detail::Outcome;
        let Some(outcome) = value.outcome.clone() else {
            return Err(value);
        };
        Ok(Self(match outcome {
            Outcome::RevisionConflict(detail) => StateError::RevisionConflict {
                command_id: CommandId::new(detail.command_id),
                expected: detail
                    .expected
                    .into_option()
                    .and_then(|expected| Kernel::<Precondition>::try_from(expected).ok())
                    .map_or(Precondition::Unconditional, Kernel::into_inner),
                observed: detail
                    .observed
                    .into_option()
                    .and_then(|observed| Kernel::<ObservedState>::try_from(observed).ok())
                    .map_or(ObservedState::Absent, Kernel::into_inner),
            },
            Outcome::IncarnationConflict(detail) => StateError::IncarnationConflict {
                command_id: CommandId::new(detail.command_id),
                partition: PartitionId::new(detail.partition),
                expected: CommitRoot::from_bytes(
                    <[u8; CommitRoot::LEN]>::try_from(detail.expected_root.as_slice())
                        .unwrap_or([0; CommitRoot::LEN]),
                ),
                observed: detail.observed_root.map(|bytes| {
                    CommitRoot::from_bytes(
                        <[u8; CommitRoot::LEN]>::try_from(bytes.as_slice())
                            .unwrap_or([0; CommitRoot::LEN]),
                    )
                }),
            },
            Outcome::StaleFence(detail) => StateError::StaleFence {
                command_id: CommandId::new(detail.command_id),
                presented: FencingToken::new(detail.presented),
                current: FencingToken::new(detail.current),
            },
            Outcome::DuplicateCommand(detail) => StateError::DuplicateCommand {
                command_id: CommandId::new(detail.command_id),
            },
            Outcome::PartitionHeld(detail) => StateError::PartitionHeld {
                partition: PartitionId::new(detail.partition),
                command_id: CommandId::new(detail.command_id),
            },
            Outcome::DigestConflict(detail) => StateError::DigestConflict {
                command_id: CommandId::new(detail.command_id),
                recorded: digest_or_zero(&detail.recorded),
                presented: digest_or_zero(&detail.presented),
            },
            Outcome::DeadlineExpired(detail) => StateError::DeadlineExpired {
                family: OperationFamily::new(detail.family),
                overrun: duration_from_nanos(detail.overrun_nanos),
            },
            Outcome::BoundsExceeded(detail) => StateError::BoundsExceeded {
                bound: bound_kind(detail.bound),
                limit: detail.limit,
                requested: detail.requested,
            },
            Outcome::Unavailable(detail) => StateError::Unavailable {
                family: OperationFamily::new(detail.family),
                reach: outage_reach(detail.reach),
            },
            Outcome::Cancelled(detail) => StateError::Cancelled {
                family: OperationFamily::new(detail.family),
            },
            Outcome::AmbiguousOutcome(detail) => StateError::AmbiguousOutcome {
                command_id: CommandId::new(detail.command_id),
                reason: ambiguity_reason(detail.reason),
            },
            Outcome::Denied(detail) => StateError::Denied {
                family: OperationFamily::new(detail.family),
            },
            Outcome::CompactedRange(detail) => StateError::CompactedRange {
                partition: PartitionId::new(detail.partition),
                requested: JournalPosition::new(detail.requested),
                earliest: JournalPosition::new(detail.earliest),
            },
            Outcome::RetiredRevision(detail) => StateError::RetiredRevision {
                requested: Revision::new(detail.requested),
                earliest: Revision::new(detail.earliest),
            },
            Outcome::Malformed(detail) => StateError::Malformed {
                field: detail.field,
                reason: detail.reason,
            },
        }))
    }
}

/// Returns the Connect error that carries `error` across the wire.
///
/// The message is the kernel's own `Display` text and the detail is the whole
/// variant, so the far side reconstructs exactly what this side refused with.
#[must_use]
pub fn to_connect_error(error: &StateError) -> ConnectError {
    ConnectError::new(code_for(error), error.to_string()).with_detail(ErrorDetail::from_message(
        STATE_ERROR_DETAIL_TYPE,
        &pb::StateErrorDetail::from(Kernel(error)),
    ))
}

/// Returns the typed outcome the transport carried, given the family the call
/// belonged to and the wire bound the caller was working under.
///
/// A detail wins when one is present. Without one the transport itself
/// produced the failure — it refused an oversized message, a spent deadline,
/// or a listener that is draining — and `fallback` decides what that means in
/// the kernel's vocabulary. Every fallback keeps the retry class the design
/// assigns: a refused message is terminal, a spent budget or a withdrawn call
/// is ambiguous, an unreachable listener is an outage whose request may
/// already have applied, and anything else is an unknown commit, which is the
/// weakest and therefore the only safe claim.
#[must_use]
pub fn from_connect_error(error: &ConnectError, fallback: &TransportFallback) -> StateError {
    for detail in &error.details {
        if detail.type_url != STATE_ERROR_DETAIL_TYPE {
            continue;
        }
        let Some(encoded) = detail.value.as_deref() else {
            continue;
        };
        let Ok(bytes) = decode_detail(encoded) else {
            continue;
        };
        let Ok(message) = <pb::StateErrorDetail as buffa::Message>::decode_from_slice(&bytes)
        else {
            continue;
        };
        if let Ok(typed) = Kernel::<StateError>::try_from(message) {
            return typed.into_inner();
        }
    }
    fallback.for_code(error.code)
}

/// Decodes an error detail's value, accepting the padded form as well.
pub(crate) fn decode_detail(encoded: &str) -> Result<Vec<u8>, base64::DecodeError> {
    base64::engine::general_purpose::STANDARD_NO_PAD
        .decode(encoded)
        .or_else(|_| base64::engine::general_purpose::STANDARD.decode(encoded))
}

/// What a transport-generated failure means for one call.
///
/// The transport refuses some requests before any handler runs, so those
/// failures carry no detail. This names the call they belonged to, which is
/// everything the kernel's vocabulary needs to describe them.
#[derive(Debug, Clone)]
pub struct TransportFallback {
    family: OperationFamily,
    wire_bound_bytes: u64,
    attempted_bytes: u64,
}

impl TransportFallback {
    /// Describes a call of `family` that put `attempted_bytes` on the wire
    /// under a `wire_bound_bytes` message bound.
    ///
    /// The bound applies to the reply as well, whose size this side never
    /// observes when the transport refuses it; see
    /// [`TransportFallback::for_code`] for what gets reported then.
    #[must_use]
    pub const fn new(family: OperationFamily, wire_bound_bytes: u64, attempted_bytes: u64) -> Self {
        Self {
            family,
            wire_bound_bytes,
            attempted_bytes,
        }
    }

    /// Returns the typed outcome a bare transport code means for this call.
    #[must_use]
    pub fn for_code(&self, code: ErrorCode) -> StateError {
        match code {
            // A message was refused against the declared wire bound before
            // any handler ran. Same bound the module enforces, reported the
            // same way.
            //
            // The bound runs in both directions, and only one of them is
            // measurable from here. When the request itself broke it, its own
            // size is what to report. When the reply did, its size never
            // reached this side at all, so the smallest value consistent with
            // the refusal stands in: one byte past the limit. Under-reporting
            // is the safe direction — a caller that trims to fit is simply
            // refused again rather than told a bound it never broke.
            ErrorCode::ResourceExhausted => StateError::BoundsExceeded {
                bound: BoundKind::PayloadBytes,
                limit: self.wire_bound_bytes,
                requested: self
                    .attempted_bytes
                    .max(self.wire_bound_bytes.saturating_add(1)),
            },
            ErrorCode::DeadlineExceeded => StateError::DeadlineExpired {
                family: self.family.clone(),
                overrun: std::time::Duration::ZERO,
            },
            ErrorCode::Canceled => StateError::Cancelled {
                family: self.family.clone(),
            },
            ErrorCode::PermissionDenied | ErrorCode::Unauthenticated => StateError::Denied {
                family: self.family.clone(),
            },
            ErrorCode::InvalidArgument => StateError::Malformed {
                field: "request".to_owned(),
                reason: "the listener could not interpret the request".to_owned(),
            },
            // A bare out-of-range code with no detail: the listener refused a
            // resume point, and this side does not know which range it named.
            // Reported with an unnamed partition and coincident positions,
            // which keeps the terminal retry class — the part that decides what
            // the caller does next — without inventing a floor it never saw.
            ErrorCode::OutOfRange => StateError::CompactedRange {
                partition: PartitionId::new(""),
                requested: JournalPosition::ORIGIN,
                earliest: JournalPosition::ORIGIN,
            },
            // An unreachable, draining, or failed listener. This transport
            // reports every network failure under one code — a connection the
            // peer refused, a stream that died mid-request, and a reply that
            // never came back are indistinguishable here — so the reach is the
            // weaker of the two. A caller that observed the refusal itself, and
            // can prove the request never went out, says so through
            // [`TransportFallback::never_dispatched`] instead of inferring it
            // from a code that cannot carry the claim.
            ErrorCode::Unavailable => StateError::Unavailable {
                family: self.family.clone(),
                reach: OutageReach::PossiblyApplied,
            },
            // Everything else leaves the outcome genuinely unknown, and unknown
            // is not the same claim as unreachable: a bare code this side did
            // not expect is no evidence the authority is down, so it must not
            // be counted as an outage.
            _ => StateError::AmbiguousOutcome {
                command_id: CommandId::new(""),
                reason: AmbiguityReason::CommitUnknown,
            },
        }
    }

    /// Returns the typed outcome for a call the transport never sent.
    ///
    /// The one condition that earns [`OutageReach::NeverDispatched`], and it is
    /// a claim about what this side observed rather than about a status code: a
    /// dial the peer refused, or a connection this side gave up on, strictly
    /// before any byte of the request went out. Nothing reached the authority,
    /// so the same command identity is sent again as a first attempt.
    ///
    /// A caller that cannot prove the request stayed home uses
    /// [`TransportFallback::for_code`], which reports the weaker reach.
    #[must_use]
    pub fn never_dispatched(&self) -> StateError {
        StateError::Unavailable {
            family: self.family.clone(),
            reach: OutageReach::NeverDispatched,
        }
    }
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs, clippy::unwrap_used)]

    use super::*;
    use polyc_state::{
        conformance::family,
        error::RetryClass,
        revision::{JournalPosition, Revision},
    };
    use std::time::Duration;

    fn fallback() -> TransportFallback {
        TransportFallback::new(OperationFamily::new(family::FAMILY), 64, 128)
    }

    fn every_variant() -> Vec<StateError> {
        vec![
            StateError::RevisionConflict {
                command_id: CommandId::new("cmd-1"),
                expected: Precondition::Revision(Revision::new(2)),
                observed: ObservedState::JournalHead(JournalPosition::new(5)),
            },
            StateError::StaleFence {
                command_id: CommandId::new("cmd-2"),
                presented: FencingToken::new(1),
                current: FencingToken::new(2),
            },
            StateError::DuplicateCommand {
                command_id: CommandId::new("cmd-3"),
            },
            StateError::DigestConflict {
                command_id: CommandId::new("cmd-4"),
                recorded: ContentDigest::from_bytes([1; ContentDigest::LEN]),
                presented: ContentDigest::from_bytes([2; ContentDigest::LEN]),
            },
            StateError::DeadlineExpired {
                family: OperationFamily::new(family::FAMILY),
                overrun: Duration::from_millis(7),
            },
            StateError::BoundsExceeded {
                bound: BoundKind::ChunkRecords,
                limit: 4,
                requested: 9,
            },
            StateError::Cancelled {
                family: OperationFamily::new(family::FAMILY),
            },
            StateError::Unavailable {
                family: OperationFamily::new(family::FAMILY),
                reach: OutageReach::NeverDispatched,
            },
            StateError::Unavailable {
                family: OperationFamily::new(family::FAMILY),
                reach: OutageReach::PossiblyApplied,
            },
            StateError::AmbiguousOutcome {
                command_id: CommandId::new("cmd-5"),
                reason: AmbiguityReason::ResponseLost,
            },
            StateError::Denied {
                family: OperationFamily::new(family::FAMILY),
            },
            StateError::Malformed {
                field: "protocol_version".to_owned(),
                reason: "this module speaks v1".to_owned(),
            },
            StateError::CompactedRange {
                partition: PartitionId::new("conv-1"),
                requested: JournalPosition::new(3),
                earliest: JournalPosition::new(9),
            },
            StateError::RetiredRevision {
                requested: Revision::new(3),
                earliest: Revision::new(9),
            },
        ]
    }

    /// The charter requirement: the variant a module refused with is the
    /// variant its caller reads, field for field, across a real encode and
    /// decode of the detail.
    #[test]
    fn every_variant_round_trips_through_a_connect_error() {
        for error in every_variant() {
            let wire = to_connect_error(&error);
            let back = from_connect_error(&wire, &fallback());
            assert_eq!(back, error, "variant did not survive the wire");
            assert_eq!(back.retry_class(), error.retry_class());
        }
    }

    /// Detail-free codes are the transport's own refusals, and each still
    /// lands on a typed outcome with the right retry class.
    #[test]
    fn a_bare_transport_code_falls_back_to_a_typed_outcome() {
        let refused = from_connect_error(
            &ConnectError::resource_exhausted("message size 128 exceeds limit 64"),
            &fallback(),
        );
        assert_eq!(
            refused,
            StateError::BoundsExceeded {
                bound: BoundKind::PayloadBytes,
                limit: 64,
                requested: 128,
            }
        );
        assert_eq!(refused.retry_class(), RetryClass::Terminal);

        let expired = from_connect_error(&ConnectError::deadline_exceeded("gone"), &fallback());
        assert!(matches!(expired, StateError::DeadlineExpired { .. }));
        assert!(expired.is_ambiguous());

        let withdrawn = from_connect_error(&ConnectError::canceled("gone"), &fallback());
        assert!(matches!(withdrawn, StateError::Cancelled { .. }));
        assert!(withdrawn.is_ambiguous());

        let draining = from_connect_error(&ConnectError::unavailable("draining"), &fallback());
        assert_eq!(
            draining,
            StateError::Unavailable {
                family: OperationFamily::new(family::FAMILY),
                reach: OutageReach::PossiblyApplied,
            }
        );
        assert!(draining.is_outage(), "a draining listener is an outage");
        assert!(
            draining.is_ambiguous(),
            "an unreachable listener settles nothing"
        );

        // A code this side did not expect is not evidence the listener is
        // down, so it must not be counted as an outage.
        let unexpected = from_connect_error(
            &ConnectError::new(ErrorCode::FailedPrecondition, "no detail"),
            &fallback(),
        );
        assert!(!unexpected.is_outage());
        assert!(unexpected.is_ambiguous());
    }

    /// The distinction the kernel gained, carried end to end: an outage and a
    /// spent bound share no class, and a reach survives a real encode and
    /// decode rather than being re-derived from a status code that cannot hold
    /// it.
    #[test]
    fn an_outage_crosses_the_wire_as_an_outage_and_keeps_its_reach() {
        for reach in [OutageReach::NeverDispatched, OutageReach::PossiblyApplied] {
            let refused = StateError::Unavailable {
                family: OperationFamily::new(family::FAMILY),
                reach,
            };
            let wire = to_connect_error(&refused);
            assert_eq!(wire.code, ErrorCode::Unavailable);

            let back = from_connect_error(&wire, &fallback());
            assert_eq!(back, refused, "the reach did not survive the wire");
            assert!(back.is_outage());
            assert_eq!(back.retry_class(), refused.retry_class());
        }

        // The two conditions C0 could not tell apart, now told apart on both
        // sides of the wire.
        let outage = from_connect_error(
            &to_connect_error(&StateError::Unavailable {
                family: OperationFamily::new(family::FAMILY),
                reach: OutageReach::PossiblyApplied,
            }),
            &fallback(),
        );
        let exhausted = from_connect_error(
            &to_connect_error(&StateError::BoundsExceeded {
                bound: BoundKind::PayloadBytes,
                limit: 4,
                requested: 9,
            }),
            &fallback(),
        );
        assert!(outage.is_outage() && !exhausted.is_outage());
        assert_eq!(outage.retry_class(), RetryClass::Ambiguous);
        assert_eq!(exhausted.retry_class(), RetryClass::Terminal);
    }

    /// The retry rule that must never invert across the wire: only a reach the
    /// sender actually proved arrives as the transient one, and a reach this
    /// build cannot read falls back to the weaker claim rather than the
    /// stronger.
    #[test]
    fn an_unreadable_reach_never_arrives_as_a_request_that_stayed_home() {
        let never = from_connect_error(
            &to_connect_error(&StateError::Unavailable {
                family: OperationFamily::new(family::FAMILY),
                reach: OutageReach::NeverDispatched,
            }),
            &fallback(),
        );
        assert_eq!(never.retry_class(), RetryClass::Transient);
        assert!(!never.is_ambiguous());

        // A sender that named a reach this build has no name for: a newer peer
        // across a version skew, which must not be read as the stronger claim.
        for unreadable in [
            buffa::EnumValue::from(pb::OutageReach::OUTAGE_REACH_UNSPECIFIED),
            buffa::EnumValue::from(97),
        ] {
            let skewed = pb::StateErrorDetail {
                outcome: Some(pb::__buffa::oneof::state_error_detail::Outcome::from(
                    pb::UnavailableDetail {
                        family: family::FAMILY.to_owned(),
                        reach: unreadable,
                        __buffa_unknown_fields: buffa::UnknownFields::default(),
                    },
                )),
                __buffa_unknown_fields: buffa::UnknownFields::default(),
            };
            let wire = ConnectError::unavailable("gone")
                .with_detail(ErrorDetail::from_message(STATE_ERROR_DETAIL_TYPE, &skewed));
            let read_back = from_connect_error(&wire, &fallback());
            assert_eq!(
                read_back,
                StateError::Unavailable {
                    family: OperationFamily::new(family::FAMILY),
                    reach: OutageReach::PossiblyApplied,
                },
                "an unrecognized reach must resolve to the weaker claim"
            );
            assert!(read_back.is_outage() && read_back.is_ambiguous());
        }
    }

    /// A transport that watched its own dial get refused is the one caller that
    /// may say the request never left.
    #[test]
    fn a_refused_dial_is_the_one_outage_that_proves_nothing_landed() {
        let never = fallback().never_dispatched();
        assert_eq!(
            never,
            StateError::Unavailable {
                family: OperationFamily::new(family::FAMILY),
                reach: OutageReach::NeverDispatched,
            }
        );
        assert_eq!(never.retry_class(), RetryClass::Transient);
        assert!(never.is_outage() && never.is_retry_safe() && !never.is_ambiguous());
    }

    /// The bound runs in both directions. A reply refused against it never
    /// reveals its size to this side, so the reported request is a floor —
    /// past the limit, which is the only thing the refusal actually proves.
    #[test]
    fn a_reply_refused_against_the_bound_reports_a_floor_not_the_request() {
        // A tiny request under a 64-byte bound: the request cannot be what
        // broke it, so the reply was.
        let tiny_request = TransportFallback::new(OperationFamily::new(family::FAMILY), 64, 12);
        let refused = from_connect_error(
            &ConnectError::resource_exhausted("message size 900 exceeds limit 64"),
            &tiny_request,
        );
        match refused {
            StateError::BoundsExceeded {
                bound,
                limit,
                requested,
            } => {
                assert_eq!(bound, BoundKind::PayloadBytes);
                assert_eq!(limit, 64);
                assert!(
                    requested > limit,
                    "a bound reported as unbroken would be nonsense: {requested} vs {limit}"
                );
            }
            other => panic!("expected the wire bound to be exceeded, got {other}"),
        }
        assert_eq!(refused.retry_class(), RetryClass::Terminal);
    }

    /// A compacted range crosses the wire whole. A consumer that read only the
    /// status code would know it was refused but not from where, and "from
    /// where" is the entire difference between rebootstrapping and retrying.
    #[test]
    fn a_compacted_range_survives_the_wire_with_the_floor_it_named() {
        let refused = StateError::CompactedRange {
            partition: PartitionId::new("conv-1"),
            requested: JournalPosition::new(3),
            earliest: JournalPosition::new(9),
        };
        let wire = to_connect_error(&refused);
        assert_eq!(wire.code, ErrorCode::OutOfRange);

        let back = from_connect_error(&wire, &fallback());
        assert_eq!(back, refused);
        assert_eq!(back.retry_class(), RetryClass::Terminal);
        assert!(!back.is_retry_safe(), "retrying the same cursor is futile");

        // Without a detail the code alone still lands on the same variant and
        // the same terminal class, which is what stops a caller retrying.
        let bare = from_connect_error(
            &ConnectError::new(ErrorCode::OutOfRange, "gone"),
            &fallback(),
        );
        assert!(matches!(bare, StateError::CompactedRange { .. }));
        assert_eq!(bare.retry_class(), RetryClass::Terminal);
    }

    #[test]
    fn a_retired_revision_keeps_a_terminal_code_without_its_detail() {
        let retired = StateError::RetiredRevision {
            requested: Revision::new(3),
            earliest: Revision::new(9),
        };
        let wire = to_connect_error(&retired);
        assert_eq!(wire.code, ErrorCode::OutOfRange);

        let bare = from_connect_error(
            &ConnectError::new(wire.code, "detail removed by an older peer"),
            &fallback(),
        );
        assert_eq!(bare.retry_class(), RetryClass::Terminal);
        assert!(!bare.is_retry_safe());
    }

    /// A detail this build cannot read must not be mistaken for success or
    /// for a different variant — the transport code decides instead.
    #[test]
    fn an_unreadable_detail_falls_back_to_the_code() {
        let error = ConnectError::deadline_exceeded("gone").with_detail(ErrorDetail {
            type_url: STATE_ERROR_DETAIL_TYPE.to_owned(),
            value: Some("!!!not base64!!!".to_owned()),
            debug: None,
        });
        assert!(matches!(
            from_connect_error(&error, &fallback()),
            StateError::DeadlineExpired { .. }
        ));

        let empty = ConnectError::permission_denied("no").with_detail(ErrorDetail::from_message(
            STATE_ERROR_DETAIL_TYPE,
            &pb::StateErrorDetail::default(),
        ));
        assert!(matches!(
            from_connect_error(&empty, &fallback()),
            StateError::Denied { .. }
        ));
    }

    /// A detail carried under someone else's type name is not this protocol's
    /// and is ignored.
    #[test]
    fn a_foreign_detail_is_ignored() {
        let error = ConnectError::canceled("gone").with_detail(ErrorDetail {
            type_url: "google.rpc.RetryInfo".to_owned(),
            value: Some(String::new()),
            debug: None,
        });
        assert!(matches!(
            from_connect_error(&error, &fallback()),
            StateError::Cancelled { .. }
        ));
    }

    #[test]
    fn each_variant_carries_a_distinct_transport_code() {
        assert_eq!(
            code_for(&StateError::Denied {
                family: OperationFamily::new(family::FAMILY)
            }),
            ErrorCode::PermissionDenied
        );
        assert_eq!(
            code_for(&StateError::Malformed {
                field: "f".to_owned(),
                reason: "r".to_owned()
            }),
            ErrorCode::InvalidArgument
        );
        assert_eq!(
            code_for(&StateError::BoundsExceeded {
                bound: BoundKind::PayloadBytes,
                limit: 1,
                requested: 2
            }),
            ErrorCode::ResourceExhausted
        );
        // The one code two variants share, and the reason the detail rather
        // than the code is the contract.
        assert_eq!(
            code_for(&StateError::Unavailable {
                family: OperationFamily::new(family::FAMILY),
                reach: OutageReach::NeverDispatched,
            }),
            ErrorCode::Unavailable
        );
        assert_eq!(
            code_for(&StateError::RetiredRevision {
                requested: Revision::new(3),
                earliest: Revision::new(9),
            }),
            ErrorCode::OutOfRange
        );
    }
}