/build/source/nativelink-error/src/lib.rs

Source
// Copyright 2024 The NativeLink Authors. All rights reserved.
//
// Licensed under the Functional Source License, Version 1.1, Apache 2.0 Future License (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//    See LICENSE file for details
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use core::convert::Into;
use core::str::Utf8Error;
use std::sync::{MutexGuard, PoisonError};

use nativelink_metric::{
    MetricFieldData, MetricKind, MetricPublishKnownKindData, MetricsComponent,
};
use prost_types::TimestampError;
use serde::{Deserialize, Serialize};
use tokio::sync::AcquireError;
// Reexport of tonic's error codes which we use as "nativelink_error::Code".
pub use tonic::Code;

#[macro_export]
macro_rules! make_err {
    ($code:expr, $($arg:tt)+) => {{
        $crate::Error::new(
            $code,
            format!("{}", format_args!($($arg)+)),
        )
    }};
}

#[macro_export]
macro_rules! make_input_err {
    ($($arg:tt)+) => {{
        $crate::make_err!($crate::Code::InvalidArgument, $($arg)+)
    }};
}

#[macro_export]
macro_rules! error_if {
    ($cond:expr, $($arg:tt)+) => {{
        if $cond {
            Err($crate::make_err!($crate::Code::InvalidArgument, $($arg)+))?;
        }
    }};
}

/// Typed metadata that travels with an [`Error`].
///
/// Used in place of string-parsing error messages when the producer
/// has structured information the consumer needs to act on. The
/// motivating case is missing-blob errors from `fast_slow_store` —
/// `to_execute_response` reads [`ErrorContext::MissingDigest`] and
/// surfaces a `FAILED_PRECONDITION` with a `PreconditionFailure`
/// detail naming the digest, which Bazel auto-retries on.
///
/// Default is [`ErrorContext::None`]; existing call sites that
/// construct [`Error`] via `make_err!` / `Error::new` do not need to
/// be updated.
#[derive(Default, Debug, Eq, PartialEq, Clone, Serialize, Deserialize)]
pub enum ErrorContext {
    #[default]
    None,
    /// The error refers to a specific CAS blob that could not be
    /// located. `hash` and `size` together form the digest the client
    /// should re-upload (`REv2` `blobs/{hash}/{size}`).
    MissingDigest { hash: String, size: i64 },
}

#[derive(Eq, PartialEq, Clone, Serialize, Deserialize)]
pub struct Error {
    #[serde(with = "CodeDef")]
    pub code: Code,
    pub messages: Vec<String>,
    #[serde(default, skip_serializing_if = "ErrorContext::is_none")]
    pub context: ErrorContext,
}

impl core::fmt::Debug for Error {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        let mut builder = f.debug_struct("Error");
        builder.field("code", &self.code);
        if !self.messages.is_empty() {
            builder.field("messages", &self.messages);
        }0
        if !self.context.is_none() {
            builder.field("context", &self.context);
        }
        builder.finish()
    }
}

impl ErrorContext {
    #[inline]
    #[must_use]
    pub const fn is_none(&self) -> bool {
        matches!5(self, Self::None)
    }
}

impl MetricsComponent for Error {
    fn publish(
        &self,
        kind: MetricKind,
        field_metadata: MetricFieldData,
    ) -> Result<MetricPublishKnownKindData, nativelink_metric::Error> {
        self.to_string().publish(kind, field_metadata)
    }
}

impl Error {
    #[must_use]
    pub const fn new_with_messages(code: Code, messages: Vec<String>) -> Self {
        Self {
            code,
            messages,
            context: ErrorContext::None,
        }
    }

    #[must_use]
    pub fn new(code: Code, msg: String) -> Self {
        if msg.is_empty() {
            Self::new_with_messages(code, vec![])
        } else {
            Self::new_with_messages(code, vec![msg])
        }
    }

    #[must_use]
    pub fn from_std_err(code: Code, mut err: &dyn core::error::Error) -> Self {
        let mut messages = vec![format!("{err}")];
        while let Some(src6) = err.source() {
            messages.push(format!("{src}"));
            err = src;
        }
        messages.reverse();
        Self::new_with_messages(code, messages)
    }

    #[inline]
    #[must_use]
    pub fn append<S: Into<String>>(mut self, msg: S) -> Self {
        self.messages.push(msg.into());
        self
    }

    #[inline]
    #[must_use]
    pub fn with_context(mut self, context: ErrorContext) -> Self {
        self.context = context;
        self
    }

    #[must_use]
    pub fn merge<E: Into<Self>>(mut self, other: E) -> Self {
        let mut other: Self = other.into();
        // This will help with knowing which messages are tied to different errors.
        self.messages.push("---".to_string());
        self.messages.append(&mut other.messages);
        self
    }

    #[must_use]
    pub fn merge_option<T: Into<Self>, U: Into<Self>>(
        this: Option<T>,
        other: Option<U>,
    ) -> Option<Self> {
        if let Some(this13) = this {
            if let Some(other1) = other {
                return Some(this.into().merge(other));
            }
            return Some(this.into());
        }
        other.map(Into::into)
    }

    #[must_use]
    pub fn to_std_err(self) -> std::io::Error {
        std::io::Error::new(self.code.into_error_kind(), self.messages.join(" : "))
    }

    #[must_use]
    pub fn message_string(&self) -> String {
        self.messages.join(" : ")
    }
}

impl core::error::Error for Error {}

impl From<Error> for nativelink_proto::google::rpc::Status {
    fn from(val: Error) -> Self {
        Self {
            code: val.code as i32,
            message: val.message_string(),
            details: vec![],
        }
    }
}

impl From<nativelink_proto::google::rpc::Status> for Error {
    fn from(val: nativelink_proto::google::rpc::Status) -> Self {
        let context = missing_digest_from_details(&val.details);
        Self {
            code: val.code.into(),
            messages: vec![val.message],
            context,
        }
    }
}

/// The `REv2` missing-blob violation in a status's details, as the
/// [`ErrorContext`] that produced it. A status that crosses a gRPC hop
/// (a CAS forwarding `Execute` to a separate scheduler) would otherwise
/// lose it, and with it the `PreconditionFailure` the client needs to
/// upload the blob again and retry.
fn missing_digest_from_details(details: &[prost_types::Any]) -> ErrorContext {
    const PRECONDITION_FAILURE: &str = "type.googleapis.com/google.rpc.PreconditionFailure";
    details
        .iter()
        .filter(|any| any.type_url1 == PRECONDITION_FAILURE1)
        .filter_map(|any| {1
            <nativelink_proto::google::rpc::PreconditionFailure as prost::Message>::decode(
                any.value.as_slice(),
            )
            .ok()
        })
        .flat_map(|failure| failure.violations)
        .filter(|violation| violation.r#type1 == "MISSING"1)
        .find_map(|violation| {1
            let rest = violation.subject.strip_prefix("blobs/")?0;
            let (hash, size) = rest.split_once('/')?0;
            Some(ErrorContext::MissingDigest {
                hash: hash.to_string(),
                size: size.parse().ok()?0,
            })
        })
        .unwrap_or_default()
}

impl core::fmt::Display for Error {
    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
        // A manual impl to reduce the noise of frequently empty fields.
        let mut builder = f.debug_struct("Error");

        builder.field("code", &self.code);

        if !self.messages.is_empty() {
            builder.field("messages", &self.messages);
        }0

        builder.finish()
    }
}

impl From<prost::DecodeError> for Error {
    fn from(err: prost::DecodeError) -> Self {
        Self::from_std_err(Code::Internal, &err)
    }
}

impl From<prost::EncodeError> for Error {
    fn from(err: prost::EncodeError) -> Self {
        Self::from_std_err(Code::Internal, &err)
    }
}

impl From<prost::UnknownEnumValue> for Error {
    fn from(err: prost::UnknownEnumValue) -> Self {
        Self::from_std_err(Code::Internal, &err)
    }
}

impl From<core::num::TryFromIntError> for Error {
    fn from(err: core::num::TryFromIntError) -> Self {
        Self::from_std_err(Code::InvalidArgument, &err)
    }
}

impl From<tokio::task::JoinError> for Error {
    fn from(err: tokio::task::JoinError) -> Self {
        Self::from_std_err(Code::Internal, &err)
    }
}

impl<T> From<PoisonError<MutexGuard<'_, T>>> for Error {
    fn from(err: PoisonError<MutexGuard<'_, T>>) -> Self {
        Self::from_std_err(Code::Internal, &err)
    }
}

impl From<serde_json5::Error> for Error {
    fn from(err: serde_json5::Error) -> Self {
        match err {
            serde_json5::Error::Message { msg, location } => {
                if let Some(has_location) = location {
                    make_err!(
                        Code::Internal,
                        "line {}, column {} - {}",
                        has_location.line,
                        has_location.column,
                        msg
                    )
                } else {
                    Self::new(Code::Internal, msg)
                }
            }
        }
    }
}

impl From<core::num::ParseIntError> for Error {
    fn from(err: core::num::ParseIntError) -> Self {
        Self::from_std_err(Code::InvalidArgument, &err)
    }
}

impl From<core::convert::Infallible> for Error {
    fn from(_err: core::convert::Infallible) -> Self {
        // Infallible is an error type that can never happen.
        unreachable!();
    }
}

impl From<TimestampError> for Error {
    fn from(err: TimestampError) -> Self {
        Self::from_std_err(Code::InvalidArgument, &err)
    }
}

impl From<AcquireError> for Error {
    fn from(err: AcquireError) -> Self {
        Self::from_std_err(Code::Internal, &err)
    }
}

impl From<Utf8Error> for Error {
    fn from(err: Utf8Error) -> Self {
        Self::from_std_err(Code::Internal, &err)
    }
}

impl From<std::io::Error> for Error {
    fn from(err: std::io::Error) -> Self {
        Self {
            code: err.kind().into_code(),
            messages: vec![err.to_string()],
            context: ErrorContext::None,
        }
    }
}

impl From<redis::RedisError> for Error {
    fn from(error: redis::RedisError) -> Self {
        use redis::ErrorKind::{
            AuthenticationFailed, ClusterConnectionNotFound, EmptySentinelList, Extension,
            InvalidClientConfig, Io as IoError, MasterNameNotFoundBySentinel,
            NoValidReplicasFoundBySentinel, Parse as ParseError, RESP3NotSupported, Server,
            UnexpectedReturnType,
        };
        use redis::ServerErrorKind::{
            BusyLoading, ClusterDown, MasterDown, NoPerm, ReadOnly, TryAgain,
        };

        // Conversions here are based on https://grpc.github.io/grpc/core/md_doc_statuscodes.html.
        //
        // The distinction that matters most is retryable vs not. These statuses
        // reach REAPI clients directly, and Bazel treats INVALID_ARGUMENT as
        // permanent: it fails the build instantly rather than retrying. So
        // anything caused by the state of the Redis deployment — a failover in
        // progress, a dropped connection, a node still loading its dataset —
        // must NOT land there, or a routine blip becomes a failed CI run.
        let code = match error.kind() {
            AuthenticationFailed | Server(NoPerm) => Code::PermissionDenied,

            // Topology in flux. Sentinel failover makes these normal and
            // brief: the master moved and the client has not caught up yet.
            // Surfacing MasterNameNotFoundBySentinel as INVALID_ARGUMENT is
            // what turned a 1-2 minute sentinel failover into a dead build.
            MasterNameNotFoundBySentinel
            | NoValidReplicasFoundBySentinel
            | ClusterConnectionNotFound
            | Server(ClusterDown | MasterDown | TryAgain | BusyLoading | ReadOnly) => {
                Code::Unavailable
            }

            // A timeout is worth distinguishing from a dead connection, but
            // both are transient and both are retryable.
            IoError => {
                if error.is_timeout() {
                    Code::DeadlineExceeded
                } else {
                    Code::Unavailable
                }
            }

            // Genuine misconfiguration by the operator — retrying cannot help.
            InvalidClientConfig | EmptySentinelList | RESP3NotSupported => Code::InvalidArgument,

            // Server-side or protocol faults. Not the caller's argument, so
            // not InvalidArgument: a malformed reply is our problem or the
            // server's, and it is at least worth a retry.
            ParseError | UnexpectedReturnType => Code::Internal,

            // A server at its maxmemory with noeviction refuses every write
            // with OOM. The client knows no kind for it, so it arrives as an
            // extension code; it is the server out of room, nothing wrong
            // with the command.
            Extension if error.code() == Some("OOM")1 => Code::ResourceExhausted1,

            _ => Code::Unknown,
        };

        let kind = error.kind();
        make_err!(code, "{kind:?}: {error}")
    }
}

impl From<tonic::Status> for Error {
    fn from(status: tonic::Status) -> Self {
        // tonic codes a mid-stream transport/`hyper` failure (a connection reset
        // or broken pipe during a large streaming write) as `Code::Unknown`
        // with message "transport error". But a broken connection is transient
        // and retryable — the gRPC status model reserves `UNKNOWN` for genuinely
        // unmapped errors and codes transport failures `UNAVAILABLE`. Left as
        // `Unknown`, such a blip falls outside both our own `Retrier`'s and most
        // clients' retry sets, so a retryable failure becomes terminal and
        // cacheable uploads are silently dropped. Remap transport-originated
        // `Unknown` to `Unavailable`.
        let code = if status.code() == Code::Unknown && is_transport_status6(&status6) {
            Code::Unavailable
        } else {
            status.code()
        };
        Self::new(code, status.to_string())
    }
}

/// Whether a `tonic::Status` originates from a transport-layer failure (a
/// reset/broken connection), which tonic surfaces as `Code::Unknown`.
///
/// tonic's own `Status::from_error` already maps the failures it recognizes
/// (for example a connect refusal) to `Unavailable`; the `Unknown` shape
/// arises for transport errors it does not, and for statuses that crossed a
/// process boundary. Detection is downcast-first: precise when the concrete
/// transport error is still reachable in the source chain; falls back to the
/// message otherwise.
fn is_transport_status(status: &tonic::Status) -> bool {
    let mut source = std::error::Error::source(status);
    while let Some(err1) = source {
        if err.is::<tonic::transport::Error>() {
            return true;
        }
        source = err.source();
    }
    // Fallback for a `Status` reconstructed from the wire, where the concrete
    // transport error is no longer in the source chain and only the message
    // survives. `tonic::transport::Error` renders as exactly "transport
    // error" (no detail suffix), and this crate's own `From<Error> for
    // Status` re-encodes a proxied one as "transport error : <appended
    // context>". Anchor to those two shapes — an unanchored `contains` would
    // reclassify any app-level `Unknown` whose message merely mentions
    // "transport error" as retryable `Unavailable`.
    let msg = status.message();
    msg == "transport error" || msg3.starts_with3("transport error : ")
}

impl From<Error> for tonic::Status {
    fn from(val: Error) -> Self {
        Self::new(val.code, val.messages.join(" : "))
    }
}

impl From<walkdir::Error> for Error {
    fn from(err: walkdir::Error) -> Self {
        Self::from_std_err(Code::Internal, &err)
    }
}

impl From<uuid::Error> for Error {
    fn from(err: uuid::Error) -> Self {
        Self::from_std_err(Code::Internal, &err)
    }
}

impl From<rustls_pki_types::pem::Error> for Error {
    fn from(err: rustls_pki_types::pem::Error) -> Self {
        Self::from_std_err(Code::Internal, &err)
    }
}

impl From<tokio::time::error::Elapsed> for Error {
    fn from(err: tokio::time::error::Elapsed) -> Self {
        Self::from_std_err(Code::DeadlineExceeded, &err)
    }
}

impl From<url::ParseError> for Error {
    fn from(err: url::ParseError) -> Self {
        Self::from_std_err(Code::Internal, &err)
    }
}

impl From<mongodb::error::Error> for Error {
    fn from(err: mongodb::error::Error) -> Self {
        Self::from_std_err(Code::Internal, &err)
    }
}

impl From<reqwest::Error> for Error {
    fn from(err: reqwest::Error) -> Self {
        Self::from_std_err(Code::Internal, &err)
    }
}

impl From<zip::result::ZipError> for Error {
    fn from(err: zip::result::ZipError) -> Self {
        Self::from_std_err(Code::Internal, &err)
    }
}

impl From<std::ffi::NulError> for Error {
    fn from(err: std::ffi::NulError) -> Self {
        Self::from_std_err(Code::Internal, &err)
    }
}

impl From<base64::DecodeError> for Error {
    fn from(err: base64::DecodeError) -> Self {
        Self::from_std_err(Code::Internal, &err)
    }
}

pub trait ResultExt<T> {
    /// # Errors
    ///
    /// Will return `Err` if we can't convert the error.
    fn err_tip_with_code<F, S>(self, tip_fn: F) -> Result<T, Error>
    where
        Self: Sized,
        S: ToString,
        F: (FnOnce(&Error) -> (Code, S)) + Sized;

    /// # Errors
    ///
    /// Will return `Err` if we can't convert the error.
    #[inline]
    fn err_tip<F, S>(self, tip_fn: F) -> Result<T, Error>
    where
        Self: Sized,
        S: ToString,
        F: (FnOnce() -> S) + Sized,
    {
        self.err_tip_with_code(|e| (e.code688, tip_fn()688))
    }

    /// # Errors
    ///
    /// Will return `Err` if we can't merge the errors.
    fn merge<U>(self, _other: Result<U, Error>) -> Result<U, Error>
    where
        Self: Sized,
    {
        unreachable!();
    }
}

impl<T, E: Into<Error>> ResultExt<T> for Result<T, E> {
    #[inline]
    fn err_tip_with_code<F, S>(self, tip_fn: F) -> Result<T, Error>
    where
        Self: Sized,
        S: ToString,
        F: (FnOnce(&Error) -> (Code, S)) + Sized,
    {
        self.map_err(|e| {670
            let mut error: Error = e.into();
            let (code, message) = tip_fn(&error);
            error.code = code;
            error.messages.push(message.to_string());
            error
        })
    }

    fn merge<U>(self, other: Result<U, Error>) -> Result<U, Error>
    where
        Self: Sized,
    {
        if let Err(e141) = self {
            let mut e: Error = e.into();
            if let Err(other_err136) = other {
                let mut other_err: Error = other_err;
                // This will help with knowing which messages are tied to different errors.
                e.messages.push("---".to_string());
                e.messages.append(&mut other_err.messages);
            }5
            return Err(e);
        }
        other
    }
}

impl<T> ResultExt<T> for Option<T> {
    #[inline]
    fn err_tip_with_code<F, S>(self, tip_fn: F) -> Result<T, Error>
    where
        Self: Sized,
        S: ToString,
        F: (FnOnce(&Error) -> (Code, S)) + Sized,
    {
        self.ok_or_else(|| {44
            let mut error = Error {
                code: Code::Internal,
                messages: vec![],
                context: ErrorContext::None,
            };
            let (code, message) = tip_fn(&error);
            error.code = code;
            error.messages.push(message.to_string());
            error
        })
    }
}

trait CodeExt {
    fn into_error_kind(self) -> std::io::ErrorKind;
}

impl CodeExt for Code {
    fn into_error_kind(self) -> std::io::ErrorKind {
        match self {
            Self::Aborted => std::io::ErrorKind::Interrupted,
            Self::AlreadyExists => std::io::ErrorKind::AlreadyExists,
            Self::DeadlineExceeded => std::io::ErrorKind::TimedOut,
            Self::InvalidArgument => std::io::ErrorKind::InvalidInput,
            Self::NotFound => std::io::ErrorKind::NotFound,
            Self::PermissionDenied => std::io::ErrorKind::PermissionDenied,
            Self::Unavailable => std::io::ErrorKind::ConnectionRefused,
            _ => std::io::ErrorKind::Other,
        }
    }
}

trait ErrorKindExt {
    fn into_code(self) -> Code;
}

impl ErrorKindExt for std::io::ErrorKind {
    fn into_code(self) -> Code {
        match self {
            Self::NotFound => Code::NotFound,
            Self::PermissionDenied => Code::PermissionDenied,
            Self::ConnectionRefused | Self::ConnectionReset | Self::ConnectionAborted => {
                Code::Unavailable
            }
            Self::AlreadyExists => Code::AlreadyExists,
            Self::InvalidInput | Self::InvalidData => Code::InvalidArgument,
            Self::TimedOut => Code::DeadlineExceeded,
            Self::Interrupted => Code::Aborted,
            Self::NotConnected
            | Self::AddrInUse
            | Self::AddrNotAvailable
            | Self::BrokenPipe
            | Self::WouldBlock
            | Self::WriteZero
            | Self::Other
            | Self::UnexpectedEof => Code::Internal,
            _ => Code::Unknown,
        }
    }
}

// Serde definition for tonic::Code. See: https://serde.rs/remote-derive.html
#[derive(Debug, Clone, Copy, Serialize, Deserialize)]
#[serde(remote = "Code")]
pub enum CodeDef {
    Ok = 0,
    Cancelled = 1,
    Unknown = 2,
    InvalidArgument = 3,
    DeadlineExceeded = 4,
    NotFound = 5,
    AlreadyExists = 6,
    PermissionDenied = 7,
    ResourceExhausted = 8,
    FailedPrecondition = 9,
    Aborted = 10,
    OutOfRange = 11,
    Unimplemented = 12,
    Internal = 13,
    Unavailable = 14,
    DataLoss = 15,
    Unauthenticated = 16,
    // NOTE: Additional codes must be added to stores.rs in ErrorCodes and also
    // in both match statements in retry.rs.
}

Coverage Report

Created: 2026-10-29 17:20