road-runner-common 0.31.0

//! Typed failure of a read against cex-policy's internal API.
//!
//! A governance read has three outcomes, and a consumer must never confuse the
//! last two:
//!
//! - `Ok(Some(_))` — cex-policy answered with the record.
//! - `Ok(None)` — cex-policy answered `NOT_FOUND`: the record does not exist.
//!   Definitive; a redelivery would read the same answer.
//! - `Err(PolicyFetchError)` — cex-policy did **not** answer the question.
//!   Nothing is known about the record, so the event that triggered the read
//!   must not be acknowledged (hold the Kafka offset, return `Err` from the
//!   [`crate::governance::GovernanceHandler`]).
//!
//! [`PolicyFetchError::is_retryable`] only says whether a retry is *expected* to
//! help on its own (an outage, a timeout) or whether someone has to fix
//! configuration first (a missing URL, a refused mesh identity). Both hold the
//! offset: an approved governance decision is never dropped because cex-policy
//! could not be asked.

/// Why a read against cex-policy produced no answer. See the module docs.
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum PolicyFetchError {
    /// The service was started without a cex-policy gRPC target, or with one
    /// that is not a valid URI. Not retryable until the deployment is fixed.
    #[error("cex-policy gRPC target not configured: {0}")]
    NotConfigured(String),

    /// cex-policy was unreachable, timed out, or failed while answering
    /// (`UNAVAILABLE`, `DEADLINE_EXCEEDED`, `INTERNAL`, `UNKNOWN`,
    /// `RESOURCE_EXHAUSTED`, `ABORTED`, `CANCELLED`, or a transport error).
    /// Expected to clear on its own; retry.
    #[error("cex-policy {rpc} for {subject} failed transiently ({code:?}): {message}")]
    Transient {
        rpc: &'static str,
        subject: String,
        code: tonic::Code,
        message: String,
    },

    /// cex-policy answered with a status that a retry will not change on its
    /// own (`PERMISSION_DENIED`, `UNAUTHENTICATED`, `INVALID_ARGUMENT`,
    /// `UNIMPLEMENTED`, `FAILED_PRECONDITION`, ...). Usually a mesh policy or a
    /// version skew between this service and cex-policy. Still not an answer
    /// about the record: hold the offset and page someone.
    #[error("cex-policy {rpc} for {subject} was refused ({code:?}): {message}")]
    Refused {
        rpc: &'static str,
        subject: String,
        code: tonic::Code,
        message: String,
    },
}

impl PolicyFetchError {
    /// Classify a non-`NOT_FOUND` gRPC status. `NOT_FOUND` is an answer, not a
    /// failure, and is handled by the caller before this is reached; if it ever
    /// does arrive here it is treated as refused rather than as "absent".
    pub(crate) fn from_status(rpc: &'static str, subject: &str, status: &tonic::Status) -> Self {
        let code = status.code();
        let subject = subject.to_string();
        let message = status.message().to_string();
        if is_transient_code(code) {
            Self::Transient {
                rpc,
                subject,
                code,
                message,
            }
        } else {
            Self::Refused {
                rpc,
                subject,
                code,
                message,
            }
        }
    }

    /// `true` when waiting and retrying is expected to succeed without anyone
    /// changing configuration. Every variant still means "no answer": callers
    /// must hold the triggering event either way.
    pub fn is_retryable(&self) -> bool {
        matches!(self, Self::Transient { .. })
    }

    /// The gRPC status code cex-policy (or the transport) produced, if any.
    pub fn code(&self) -> Option<tonic::Code> {
        match self {
            Self::NotConfigured(_) => None,
            Self::Transient { code, .. } | Self::Refused { code, .. } => Some(*code),
        }
    }
}

/// Codes that describe the call failing rather than cex-policy deciding.
/// Transport failures (connection refused, reset, DNS) surface from tonic as
/// `UNAVAILABLE` or `UNKNOWN`, so they land here too.
fn is_transient_code(code: tonic::Code) -> bool {
    matches!(
        code,
        tonic::Code::Unavailable
            | tonic::Code::DeadlineExceeded
            | tonic::Code::Internal
            | tonic::Code::Unknown
            | tonic::Code::ResourceExhausted
            | tonic::Code::Aborted
            | tonic::Code::Cancelled
    )
}