polyc-state-connect 2026.9.0

//! 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},
    journal::JournalDirectorySnapshotId,
    revision::{CommitRoot, JournalPosition, JournalSource, 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::SourceIncarnationChanged { .. }
        | 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 { .. }
        | StateError::SnapshotUnavailable { .. } => ErrorCode::OutOfRange,
        StateError::Malformed { .. } => ErrorCode::InvalidArgument,
        // The authority answered, and what it holds is damaged. Not
        // InvalidArgument, which blames the caller for storage it never
        // touched, and not Unavailable, which promises that waiting helps.
        StateError::JournalDamaged => ErrorCode::DataLoss,
    }
}

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::SnapshotUnavailable { snapshot } => {
                Outcome::from(pb::SnapshotUnavailableDetail {
                    snapshot: snapshot.as_str().to_owned(),
                    __buffa_unknown_fields: unknown(),
                })
            }
            StateError::SourceIncarnationChanged { expected } => {
                Outcome::from(pb::SourceIncarnationChangedDetail {
                    expected: buffa::MessageField::some(pb::JournalSource::from(Kernel(expected))),
                    __buffa_unknown_fields: unknown(),
                })
            }
            // Nothing to name. The detail is empty by contract, so no
            // server-held identifier can travel with it.
            StateError::JournalDamaged => Outcome::from(pb::JournalDamagedDetail {
                __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::DirectoryPageEntries => Self::BOUND_KIND_DIRECTORY_PAGE_ENTRIES,
            BoundKind::ChunkRecords => Self::BOUND_KIND_CHUNK_RECORDS,
            BoundKind::DirectorySnapshotEntries => Self::BOUND_KIND_DIRECTORY_SNAPSHOT_ENTRIES,
            BoundKind::DirectorySnapshotNameBytes => Self::BOUND_KIND_DIRECTORY_SNAPSHOT_NAME_BYTES,
            BoundKind::ActiveDirectorySnapshots => Self::BOUND_KIND_ACTIVE_DIRECTORY_SNAPSHOTS,
            BoundKind::ActiveDirectorySnapshotEntries => {
                Self::BOUND_KIND_ACTIVE_DIRECTORY_SNAPSHOT_ENTRIES
            }
            BoundKind::ActiveDirectorySnapshotNameBytes => {
                Self::BOUND_KIND_ACTIVE_DIRECTORY_SNAPSHOT_NAME_BYTES
            }
        }
    }
}

impl From<Kernel<OutageReach>> for pb::OutageReach {
    fn from(value: Kernel<OutageReach>) -> Self {
        match value.0 {
            OutageReach::NeverDispatched => Self::OUTAGE_REACH_NEVER_DISPATCHED,
            OutageReach::NoDurableEffect => Self::OUTAGE_REACH_NO_DURABLE_EFFECT,
            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_DIRECTORY_PAGE_ENTRIES) => BoundKind::DirectoryPageEntries,
        Some(pb::BoundKind::BOUND_KIND_CHUNK_RECORDS) => BoundKind::ChunkRecords,
        Some(pb::BoundKind::BOUND_KIND_DIRECTORY_SNAPSHOT_ENTRIES) => {
            BoundKind::DirectorySnapshotEntries
        }
        Some(pb::BoundKind::BOUND_KIND_DIRECTORY_SNAPSHOT_NAME_BYTES) => {
            BoundKind::DirectorySnapshotNameBytes
        }
        Some(pb::BoundKind::BOUND_KIND_ACTIVE_DIRECTORY_SNAPSHOTS) => {
            BoundKind::ActiveDirectorySnapshots
        }
        Some(pb::BoundKind::BOUND_KIND_ACTIVE_DIRECTORY_SNAPSHOT_ENTRIES) => {
            BoundKind::ActiveDirectorySnapshotEntries
        }
        Some(pb::BoundKind::BOUND_KIND_ACTIVE_DIRECTORY_SNAPSHOT_NAME_BYTES) => {
            BoundKind::ActiveDirectorySnapshotNameBytes
        }
        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_NO_DURABLE_EFFECT) => OutageReach::NoDurableEffect,
        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.
    #[allow(
        clippy::too_many_lines,
        reason = "one exhaustive match keeps every wire outcome and detail field compile checked"
    )]
    fn try_from(value: pb::StateErrorDetail) -> Result<Self, Self::Error> {
        use pb::__buffa::oneof::state_error_detail::Outcome;
        let pb::StateErrorDetail {
            outcome,
            __buffa_unknown_fields,
        } = value;
        let Some(outcome) = outcome else {
            return Err(pb::StateErrorDetail {
                outcome: None,
                __buffa_unknown_fields,
            });
        };
        let original = pb::StateErrorDetail {
            outcome: Some(outcome.clone()),
            __buffa_unknown_fields,
        };
        let error = match outcome {
            Outcome::RevisionConflict(detail) => {
                let pb::RevisionConflictDetail {
                    command_id,
                    expected,
                    observed,
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::RevisionConflict {
                    command_id: CommandId::new(command_id),
                    expected: expected
                        .into_option()
                        .and_then(|expected| Kernel::<Precondition>::try_from(expected).ok())
                        .map_or(Precondition::Unconditional, Kernel::into_inner),
                    observed: observed
                        .into_option()
                        .and_then(|observed| Kernel::<ObservedState>::try_from(observed).ok())
                        .map_or(ObservedState::Absent, Kernel::into_inner),
                }
            }
            Outcome::IncarnationConflict(detail) => {
                let pb::IncarnationConflictDetail {
                    command_id,
                    partition,
                    expected_root,
                    observed_root,
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::IncarnationConflict {
                    command_id: CommandId::new(command_id),
                    partition: PartitionId::new(partition),
                    expected: CommitRoot::from_bytes(
                        <[u8; CommitRoot::LEN]>::try_from(expected_root.as_slice())
                            .unwrap_or([0; CommitRoot::LEN]),
                    ),
                    observed: observed_root.map(|bytes| {
                        CommitRoot::from_bytes(
                            <[u8; CommitRoot::LEN]>::try_from(bytes.as_slice())
                                .unwrap_or([0; CommitRoot::LEN]),
                        )
                    }),
                }
            }
            Outcome::StaleFence(detail) => {
                let pb::StaleFenceDetail {
                    command_id,
                    presented,
                    current,
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::StaleFence {
                    command_id: CommandId::new(command_id),
                    presented: FencingToken::new(presented),
                    current: FencingToken::new(current),
                }
            }
            Outcome::DuplicateCommand(detail) => {
                let pb::DuplicateCommandDetail {
                    command_id,
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::DuplicateCommand {
                    command_id: CommandId::new(command_id),
                }
            }
            Outcome::PartitionHeld(detail) => {
                let pb::PartitionHeldDetail {
                    partition,
                    command_id,
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::PartitionHeld {
                    partition: PartitionId::new(partition),
                    command_id: CommandId::new(command_id),
                }
            }
            Outcome::DigestConflict(detail) => {
                let pb::DigestConflictDetail {
                    command_id,
                    recorded,
                    presented,
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::DigestConflict {
                    command_id: CommandId::new(command_id),
                    recorded: digest_or_zero(&recorded),
                    presented: digest_or_zero(&presented),
                }
            }
            Outcome::DeadlineExpired(detail) => {
                let pb::DeadlineExpiredDetail {
                    family,
                    overrun_nanos,
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::DeadlineExpired {
                    family: OperationFamily::new(family),
                    overrun: duration_from_nanos(overrun_nanos),
                }
            }
            Outcome::BoundsExceeded(detail) => {
                let pb::BoundsExceededDetail {
                    bound,
                    limit,
                    requested,
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::BoundsExceeded {
                    bound: bound_kind(bound),
                    limit,
                    requested,
                }
            }
            Outcome::Unavailable(detail) => {
                let pb::UnavailableDetail {
                    family,
                    reach,
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::Unavailable {
                    family: OperationFamily::new(family),
                    reach: outage_reach(reach),
                }
            }
            Outcome::Cancelled(detail) => {
                let pb::CancelledDetail {
                    family,
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::Cancelled {
                    family: OperationFamily::new(family),
                }
            }
            Outcome::AmbiguousOutcome(detail) => {
                let pb::AmbiguousOutcomeDetail {
                    command_id,
                    reason,
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::AmbiguousOutcome {
                    command_id: CommandId::new(command_id),
                    reason: ambiguity_reason(reason),
                }
            }
            Outcome::Denied(detail) => {
                let pb::DeniedDetail {
                    family,
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::Denied {
                    family: OperationFamily::new(family),
                }
            }
            Outcome::CompactedRange(detail) => {
                let pb::CompactedRangeDetail {
                    partition,
                    requested,
                    earliest,
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::CompactedRange {
                    partition: PartitionId::new(partition),
                    requested: JournalPosition::new(requested),
                    earliest: JournalPosition::new(earliest),
                }
            }
            Outcome::RetiredRevision(detail) => {
                let pb::RetiredRevisionDetail {
                    requested,
                    earliest,
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::RetiredRevision {
                    requested: Revision::new(requested),
                    earliest: Revision::new(earliest),
                }
            }
            Outcome::SnapshotUnavailable(detail) => {
                let pb::SnapshotUnavailableDetail {
                    snapshot,
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::SnapshotUnavailable {
                    snapshot: JournalDirectorySnapshotId::new(snapshot),
                }
            }
            Outcome::JournalDamaged(detail) => {
                let pb::JournalDamagedDetail {
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::JournalDamaged
            }
            Outcome::SourceIncarnationChanged(detail) => {
                let pb::SourceIncarnationChangedDetail {
                    expected,
                    __buffa_unknown_fields: _,
                } = *detail;
                let expected = expected
                    .into_option()
                    .and_then(|source| Kernel::<JournalSource>::try_from(source).ok())
                    .map(Kernel::into_inner)
                    .ok_or_else(|| original.clone())?;
                StateError::SourceIncarnationChanged { expected }
            }
            Outcome::Malformed(detail) => {
                let pb::MalformedDetail {
                    field,
                    reason,
                    __buffa_unknown_fields: _,
                } = *detail;
                StateError::Malformed { field, reason }
            }
        };
        Ok(Self(error))
    }
}

/// 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(),
            },
            // `DataLoss` is deliberately ABSENT here, so it reaches the
            // conservative fallback below.
            //
            // Only a valid typed `JournalDamagedDetail` may decode to
            // `StateError::JournalDamaged`, because only State attaches that
            // detail and only State can know its own storage is damaged. A bare
            // code answers a producer that is not State: `connectrpc` raises
            // `DataLoss` when a response body cannot be decoded, and for a
            // mutating call the command may have committed before the reply was
            // corrupted. That is an ambiguous outcome, and a caller may retry it
            // under the same command identity.
            //
            // Mapping the bare code to the terminal outcome would also let a
            // broken detail path launder itself: decoding fails, the fallback
            // answers, and the caller cannot tell that from a detail that
            // traveled. Terminal damage must be something State said, never
            // something this side inferred from a status code.

            // 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, PartitionIncarnation, 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::IncarnationConflict {
                command_id: CommandId::new("cmd-incarnation"),
                partition: PartitionId::new("conv-incarnation"),
                expected: CommitRoot::from_bytes([1; CommitRoot::LEN]),
                observed: Some(CommitRoot::from_bytes([2; CommitRoot::LEN])),
            },
            StateError::SourceIncarnationChanged {
                expected: JournalSource::new(
                    PartitionId::new("conv-source-present"),
                    PartitionIncarnation::from_bytes([3; PartitionIncarnation::LEN]),
                ),
            },
            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::PartitionHeld {
                partition: PartitionId::new("conv-held"),
                command_id: CommandId::new("cmd-held"),
            },
            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),
            },
            StateError::SnapshotUnavailable {
                snapshot: JournalDirectorySnapshotId::new("directory-1"),
            },
            // Carries nothing, which is exactly why it belongs here. Nothing
            // in the reply distinguishes it but the typed detail itself.
            StateError::JournalDamaged,
        ]
    }

    /// 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.
    /// A lineage refusal carries no lineage the caller did not send.
    ///
    /// It once carried the observed source beside the expected one, read out of
    /// storage. That let a caller name a partition it does not own and learn
    /// from the refusal whether that partition has a live lineage and which one
    /// — before any ownership check ran, because the source read is what the
    /// projection RPCs do first.
    ///
    /// This drives the real encoder and reads the real wire bytes, because the
    /// question is what LEAVES the process, not what the kernel holds.
    #[test]
    fn a_lineage_refusal_discloses_no_source_the_caller_did_not_name() {
        let expected = JournalSource::new(
            PartitionId::new("conv-owned"),
            PartitionIncarnation::from_bytes([3; PartitionIncarnation::LEN]),
        );
        // The lineage a server would hold for a partition the caller does not
        // own. It is never constructed into the refusal; it exists here only so
        // the assertions below can look for it.
        let secret = PartitionIncarnation::from_bytes([9; PartitionIncarnation::LEN]);

        let wire = to_connect_error(&StateError::SourceIncarnationChanged {
            expected: expected.clone(),
        });
        let encoded = format!("{wire:?}");

        assert!(
            encoded.contains("conv-owned"),
            "the caller's own source still travels, or it cannot act: {encoded}"
        );
        assert!(
            !encoded.contains(&format!("{secret}")),
            "no foreign incarnation may travel: {encoded}"
        );
        assert!(
            !encoded.to_lowercase().contains("observed"),
            "the observed lineage has no field to travel in: {encoded}"
        );

        // And the round trip preserves exactly the one source it carries.
        let StateError::SourceIncarnationChanged { expected: back } =
            from_connect_error(&wire, &fallback())
        else {
            panic!("the refusal decodes as itself: {wire:?}");
        };
        assert_eq!(back, expected);
    }

    #[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::NoDurableEffect,
            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());
    }

    /// The damaged-storage outcome travels as its own typed detail, and the
    /// caller reads it from that detail rather than from the bare code.
    ///
    /// The typed detail is the only thing that produces this outcome. A bare
    /// `DataLoss` is ambiguous instead, which
    /// `a_bare_data_loss_is_ambiguous_not_terminal` pins.
    ///
    /// This decodes the detail off the wire and names the oneof, and it pins
    /// the redaction where it is observable: the encoded wire bytes, not just
    /// the rendered message.
    #[test]
    fn damaged_storage_travels_as_a_typed_detail_and_names_nothing() {
        use pb::__buffa::oneof::state_error_detail::Outcome;

        let wire = to_connect_error(&StateError::JournalDamaged);
        assert_eq!(wire.code, ErrorCode::DataLoss);

        let detail = wire
            .details
            .iter()
            .find(|detail| detail.type_url == STATE_ERROR_DETAIL_TYPE)
            .expect("the refusal carries this protocol's detail");
        let bytes = decode_detail(detail.value.as_deref().expect("the detail has a value"))
            .expect("the detail decodes");
        let decoded = <pb::StateErrorDetail as buffa::Message>::decode_from_slice(&bytes)
            .expect("the detail parses");
        assert!(
            matches!(decoded.outcome, Some(Outcome::JournalDamaged(_))),
            "the typed outcome must be on the wire, or the fallback is what answered: {:?}",
            decoded.outcome
        );

        // Nothing server-held travels, in the rendered message or the bytes.
        let rendered = format!(
            "{}|{:?}",
            StateError::JournalDamaged,
            StateError::JournalDamaged
        );
        for leaked in [
            "conv-",
            "_offsets",
            "_data",
            "/Users",
            "partition",
            "remnant",
        ] {
            assert!(
                !rendered.contains(leaked),
                "the refusal renders {leaked:?}: {rendered}"
            );
            assert!(
                !String::from_utf8_lossy(&bytes).contains(leaked),
                "the encoded detail carries {leaked:?}"
            );
        }
        assert!(
            bytes.len() <= 4,
            "an empty detail encodes to almost nothing; {} bytes means a field traveled",
            bytes.len()
        );
    }

    /// A bare `DataLoss` is ambiguous, never terminal damage.
    ///
    /// Only State can know its own storage is damaged, and it says so with a
    /// typed detail. A bare code answers a producer that is not State:
    /// `connectrpc` raises `DataLoss` when it cannot decode a response body,
    /// and for a mutating call the command may have committed before the reply
    /// was corrupted.
    ///
    /// Reading that as terminal would abandon a mutation whose outcome is
    /// unknown, and would point an operator at a storage runbook for a
    /// transport fault.
    #[test]
    fn a_bare_data_loss_is_ambiguous_not_terminal() {
        let bare = from_connect_error(
            &ConnectError::new(ErrorCode::DataLoss, "no detail"),
            &fallback(),
        );

        assert!(
            matches!(
                bare,
                StateError::AmbiguousOutcome {
                    reason: AmbiguityReason::CommitUnknown,
                    ..
                }
            ),
            "a bare data-loss code is an unknown commit, not damaged storage: {bare:?}"
        );
        assert_eq!(bare.retry_class(), RetryClass::Ambiguous);
        assert!(
            bare.is_retry_safe(),
            "the caller may replay the same command identity, which is what makes an ambiguous \
             outcome recoverable"
        );
        assert_ne!(
            bare,
            StateError::JournalDamaged,
            "terminal damage must be something State said, never something inferred from a code"
        );
    }

    /// A detail that cannot be decoded cannot become terminal damage.
    ///
    /// This is the laundering path the bare-code mapping would have opened: if
    /// decoding fails and the code alone answered `JournalDamaged`, a caller
    /// could not tell a detail that traveled from one that was unreadable.
    /// The conservative outcome must win instead.
    #[test]
    fn an_undecodable_detail_under_data_loss_stays_ambiguous() {
        let corrupt = ConnectError::new(ErrorCode::DataLoss, "damaged").with_detail(ErrorDetail {
            type_url: STATE_ERROR_DETAIL_TYPE.to_owned(),
            value: Some("!!!not base64!!!".to_owned()),
            debug: None,
        });
        let refused = from_connect_error(&corrupt, &fallback());
        assert!(
            matches!(refused, StateError::AmbiguousOutcome { .. }),
            "an unreadable detail falls back to the conservative outcome: {refused:?}"
        );
        assert!(refused.is_retry_safe());

        // The same code carrying a detail with no outcome set: still not
        // enough to claim damage.
        let empty = ConnectError::new(ErrorCode::DataLoss, "damaged").with_detail(
            ErrorDetail::from_message(STATE_ERROR_DETAIL_TYPE, &pb::StateErrorDetail::default()),
        );
        let refused = from_connect_error(&empty, &fallback());
        assert!(
            matches!(refused, StateError::AmbiguousOutcome { .. }),
            "an empty detail claims nothing: {refused:?}"
        );
    }

    /// 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
        );
    }
}