saddle-boundary 0.3.18

//! Source facts only: no logger, response, retry, or lifecycle authority.
use crate::{BoundaryError, ExecutionCertainty};
use saddle_core::{
    CaptureSite, Diagnostic, DiagnosticCategory, DiagnosticCause, DiagnosticCode, DiagnosticObject,
    DiagnosticObjectKind, DiagnosticStage,
};
use std::{error::Error, io};

fn cause(stage: DiagnosticStage, code: &'static str, alias: &'static str) -> DiagnosticCause {
    DiagnosticCause::new(
        stage,
        DiagnosticCode::new(code).expect("fixed diagnostic code"),
    )
    .with_object(
        DiagnosticObject::new(DiagnosticObjectKind::TargetAlias, alias).expect("fixed alias"),
    )
}

pub(crate) fn invalid_response() -> Diagnostic {
    Diagnostic::capture(
        DiagnosticCategory::UnexpectedError,
        CaptureSite::FirstObserved,
        cause(
            DiagnosticStage::RequestOutbound,
            "transport.response_invalid",
            "profusecontract",
        ),
    )
}

// Never format a foreign error. Bounded traversal also tolerates cyclic source chains.
fn io_source<'a>(error: &'a (dyn Error + 'static)) -> Option<&'a io::Error> {
    let mut current = Some(error);
    for _ in 0..16 {
        let error = current?;
        if let Some(io) = error.downcast_ref::<io::Error>() {
            return Some(io);
        }
        current = error.source();
    }
    None
}

pub(crate) fn connect_error(error: &(dyn Error + 'static)) -> BoundaryError {
    let mut cause = cause(
        DiagnosticStage::RequestOutbound,
        "transport.connect_failed",
        "profusecontract",
    );
    let mut category = DiagnosticCategory::UnexpectedError;
    if let Some(io) = io_source(error) {
        cause = cause.with_io(io);
        if matches!(
            io.kind(),
            io::ErrorKind::TimedOut | io::ErrorKind::Interrupted
        ) {
            category = DiagnosticCategory::ExpectedRejection;
        }
    }
    BoundaryError::transport(
        "profusecontract connection failed",
        ExecutionCertainty::NotExecuted,
    )
    .with_diagnostic(Diagnostic::capture(
        category,
        CaptureSite::FirstObserved,
        cause,
    ))
}

pub(crate) fn rpc_error(status: &tonic::Status) -> BoundaryError {
    use tonic::Code;
    let code = match status.code() {
        Code::Ok => "transport.rpc.ok",
        Code::Cancelled => "transport.rpc.cancelled",
        Code::Unknown => "transport.rpc.unknown",
        Code::InvalidArgument => "transport.rpc.invalid_argument",
        Code::DeadlineExceeded => "transport.rpc.deadline_exceeded",
        Code::NotFound => "transport.rpc.not_found",
        Code::AlreadyExists => "transport.rpc.already_exists",
        Code::PermissionDenied => "transport.rpc.permission_denied",
        Code::ResourceExhausted => "transport.rpc.resource_exhausted",
        Code::FailedPrecondition => "transport.rpc.failed_precondition",
        Code::Aborted => "transport.rpc.aborted",
        Code::OutOfRange => "transport.rpc.out_of_range",
        Code::Unimplemented => "transport.rpc.unimplemented",
        Code::Internal => "transport.rpc.internal",
        Code::Unavailable => "transport.rpc.unavailable",
        Code::DataLoss => "transport.rpc.data_loss",
        Code::Unauthenticated => "transport.rpc.unauthenticated",
    };
    let category = if matches!(
        status.code(),
        Code::Cancelled
            | Code::DeadlineExceeded
            | Code::ResourceExhausted
            | Code::InvalidArgument
            | Code::Unauthenticated
            | Code::PermissionDenied
    ) {
        DiagnosticCategory::ExpectedRejection
    } else {
        DiagnosticCategory::UnexpectedError
    };
    let mut cause = cause(DiagnosticStage::RequestOutbound, code, "profusecontract");
    if let Some(io) = io_source(status) {
        cause = cause.with_io(io);
    }
    BoundaryError::transport(
        "profusecontract RPC failed",
        ExecutionCertainty::MayHaveExecuted,
    )
    .with_diagnostic(Diagnostic::capture(
        category,
        CaptureSite::FirstObserved,
        cause,
    ))
}

/// Closed physical IO source stage. It cannot send or retry a response.
#[derive(Clone, Copy)]
pub enum HttpIoStage {
    Listener,
    IngressRead,
    ResponseWrite,
}

/// Call while the original IO error still exists. Facade supplies its existing
/// request context to the unified writer separately; never fabricate correlation.
/// Peer disconnect/cancellation/deadline are expected, not internal exceptions.
pub fn http_io(stage: HttpIoStage, error: &io::Error) -> Diagnostic {
    let (stage, code) = match stage {
        HttpIoStage::Listener => (DiagnosticStage::StartupListener, "transport.listener_io"),
        HttpIoStage::IngressRead => (DiagnosticStage::RequestDecode, "transport.ingress_io"),
        HttpIoStage::ResponseWrite => (DiagnosticStage::RequestResponse, "transport.response_io"),
    };
    let category = if matches!(
        error.kind(),
        io::ErrorKind::ConnectionReset
            | io::ErrorKind::ConnectionAborted
            | io::ErrorKind::BrokenPipe
            | io::ErrorKind::UnexpectedEof
            | io::ErrorKind::TimedOut
            | io::ErrorKind::Interrupted
            | io::ErrorKind::WouldBlock
            | io::ErrorKind::InvalidData
    ) {
        DiagnosticCategory::ExpectedRejection
    } else {
        DiagnosticCategory::UnexpectedError
    };
    Diagnostic::capture(
        category,
        CaptureSite::FirstObserved,
        cause(stage, code, "profusegw").with_io(error),
    )
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn source_diagnostics_real_refused_connect() {
        tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap()
            .block_on(async {
                let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
                let addr = listener.local_addr().unwrap();
                drop(listener);
                let result = crate::TonicBoundary::connect(
                    crate::ProfuseContractEndpoint::new(format!("http://{addr}")).unwrap(),
                )
                .await;
                let error = match result {
                    Err(error) => error,
                    Ok(_) => panic!("connection unexpectedly succeeded"),
                };
                let json = serde_json::to_string(error.diagnostic().unwrap()).unwrap();
                assert!(json.contains("connection_refused"), "{json}");
                assert!(json.contains("os_code"));
                assert!(json.contains("profusecontract"));
                assert!(!json.contains(&addr.to_string()));
            });
    }

    #[test]
    fn source_diagnostics_status_and_peer_are_safe() {
        let status = tonic::Status::internal("Bearer DO_NOT_LOG https://secret.invalid user-body");
        let error = rpc_error(&status);
        let json = serde_json::to_string(error.diagnostic().unwrap()).unwrap();
        assert!(json.contains("transport.rpc.internal"));
        for secret in ["DO_NOT_LOG", "secret.invalid", "user-body"] {
            assert!(!format!("{error:?}{error}{json}").contains(secret));
        }
        let cancelled = rpc_error(&tonic::Status::cancelled("DO_NOT_LOG"));
        assert!(
            serde_json::to_string(cancelled.diagnostic().unwrap())
                .unwrap()
                .contains("expected_rejection")
        );
        let (reader, writer) = std::os::unix::net::UnixStream::pair().unwrap();
        drop(reader);
        use std::io::Write;
        let io = (&writer).write_all(b"response").unwrap_err();
        let diagnostic = http_io(HttpIoStage::ResponseWrite, &io);
        let json = serde_json::to_string(&diagnostic).unwrap();
        assert!(json.contains("broken_pipe"));
        assert!(json.contains("expected_rejection"));
        // Observation does not own a socket and cannot write a second response.
    }
}