saddle-boundary 0.3.23

//! Borrowed unified outlet: source facts are submitted before conversion.
//! No new logger, task, queue, timer or response ownership.
use saddle_core::{
    BoundedDiagnostic, BoundedDiagnosticCause, CaptureSite, DiagnosticCategory, DiagnosticCode,
    DiagnosticOccurrence, DiagnosticOutcomeAxes, DiagnosticStage, InlineDiagnosticText,
    OperationOutcome, ResponseDelivery,
};
use saddle_observability::{DiagnosticSubmission, EmergencyDiagnosticHandle, EventContext};
use std::io;

/// No task or timer: observes cancellation of the existing caller-owned future.
/// Drop cannot prove whether the caller selected timeout, disconnect, or cancellation.
pub(crate) async fn observe_future<T>(
    source: Option<&dyn SourceCapture>,
    future: impl std::future::Future<Output = T>,
) -> T {
    struct Guard<'a> {
        source: Option<&'a dyn SourceCapture>,
        completed: bool,
    }
    impl Drop for Guard<'_> {
        fn drop(&mut self) {
            if !self.completed {
                if let Some(source) = self.source {
                    let panicked = std::thread::panicking();
                    let receipt = if panicked {
                        source.submit_cause(
                            DiagnosticCategory::UnexpectedError,
                            BoundedDiagnosticCause::new(
                                DiagnosticStage::RequestOutbound,
                                code("transport.future_panicked"),
                            )
                            .with_object(InlineDiagnosticText::metadata("profusecontract")),
                        )
                    } else {
                        source.local_failure(LocalFailure::Cancelled)
                    };
                    let axes = DiagnosticOutcomeAxes {
                        operation: if panicked {
                            OperationOutcome::Panicked
                        } else {
                            OperationOutcome::Cancelled
                        },
                        ..Default::default()
                    };
                    // Degradation is exposed by the shared handle snapshot; no business retry.
                    // Required callers retain the receipt outside this future
                    // and own its single boundary after the borrow is dropped.
                    if !source.required() {
                        let _ = source.boundary(Some(receipt), &axes);
                    }
                }
            }
        }
    }
    let mut guard = Guard {
        source,
        completed: false,
    };
    let result = future.await;
    guard.completed = true;
    result
}

pub struct SourceContext<'a> {
    pub output: &'a EmergencyDiagnosticHandle,
    pub association: Option<(&'a saddle_core::CallContext, &'a EventContext)>,
}

/// Enqueued is not written; retain this status independently of business Result.
#[derive(Clone, Copy)]
pub struct SourceReceipt {
    pub(crate) diagnostic_id: u64,
    pub(crate) submission: DiagnosticSubmission,
    pub(crate) occurrence: DiagnosticOccurrence,
}

impl SourceReceipt {
    pub fn diagnostic_id(&self) -> u64 {
        self.diagnostic_id
    }
    pub fn submission(&self) -> DiagnosticSubmission {
        self.submission
    }
    pub fn occurrence(&self) -> DiagnosticOccurrence {
        self.occurrence
    }
}

impl std::fmt::Debug for SourceReceipt {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SourceReceipt")
            .field("diagnostic_id", &self.diagnostic_id)
            .field("submission", &self.submission)
            .finish_non_exhaustive()
    }
}

#[derive(Clone, Copy)]
pub enum IoOperation {
    Accept,
    ReadHead,
    ReadBody,
    WriteResponse,
    ShutdownResponse,
    DeadlineWrite,
}

#[derive(Clone, Copy)]
pub enum LocalFailure {
    InvalidRequest,
    InvalidAuthority,
    Deadline,
    Cancelled,
}

pub(crate) fn code(value: &'static str) -> DiagnosticCode {
    DiagnosticCode::new(value).expect("closed transport code")
}

pub(crate) fn io_kind(error: &io::Error) -> DiagnosticCode {
    code(match error.kind() {
        io::ErrorKind::ConnectionRefused => "connection_refused",
        io::ErrorKind::ConnectionReset => "connection_reset",
        io::ErrorKind::ConnectionAborted => "connection_aborted",
        io::ErrorKind::BrokenPipe => "broken_pipe",
        io::ErrorKind::UnexpectedEof => "unexpected_eof",
        io::ErrorKind::WriteZero => "write_zero",
        io::ErrorKind::TimedOut => "timed_out",
        io::ErrorKind::Interrupted => "interrupted",
        io::ErrorKind::WouldBlock => "would_block",
        io::ErrorKind::InvalidData => "invalid_data",
        io::ErrorKind::InvalidInput => "invalid_input",
        io::ErrorKind::PermissionDenied => "permission_denied",
        io::ErrorKind::NotConnected => "not_connected",
        io::ErrorKind::AddrInUse => "addr_in_use",
        io::ErrorKind::AddrNotAvailable => "addr_not_available",
        _ => "unknown",
    })
}

impl SourceContext<'_> {
    #[cfg(test)]
    #[track_caller]
    pub(crate) fn rpc_failure(&self, status: &tonic::Status) -> SourceReceipt {
        (self as &dyn SourceCapture).rpc_failure(status)
    }
    #[track_caller]
    pub fn local_failure(&self, failure: LocalFailure) -> SourceReceipt {
        SourceCapture::local_failure(self, failure)
    }
    #[track_caller]
    pub fn io_failure(&self, operation: IoOperation, error: &io::Error) -> SourceReceipt {
        (self as &dyn SourceCapture).io_failure(operation, error)
    }
    pub fn boundary(
        &self,
        source: Option<SourceReceipt>,
        axes: &DiagnosticOutcomeAxes,
    ) -> DiagnosticSubmission {
        SourceCapture::boundary(self, source, axes)
    }
}

/// Internal common source path: legacy callers and required callers use the
/// same classifiers, before conversion. No network implementation is copied.
pub(crate) trait SourceCapture: Sync {
    fn required(&self) -> bool {
        false
    }
    #[track_caller]
    fn submit_cause(
        &self,
        category: DiagnosticCategory,
        cause: BoundedDiagnosticCause,
    ) -> SourceReceipt;
    fn boundary(
        &self,
        source: Option<SourceReceipt>,
        axes: &DiagnosticOutcomeAxes,
    ) -> DiagnosticSubmission;
    #[track_caller]
    fn local_failure(&self, failure: LocalFailure) -> SourceReceipt {
        let reason = match failure {
            LocalFailure::InvalidRequest => "transport.request_invalid",
            LocalFailure::InvalidAuthority => "transport.authority_invalid",
            LocalFailure::Deadline => "transport.deadline_elapsed",
            LocalFailure::Cancelled => "transport.cancelled",
        };
        self.submit_cause(
            DiagnosticCategory::ExpectedRejection,
            BoundedDiagnosticCause::new(DiagnosticStage::RequestOutbound, code(reason))
                .with_object(InlineDiagnosticText::metadata("profusecontract")),
        )
    }
    #[track_caller]
    fn protocol_failure(&self) -> SourceReceipt {
        self.submit_cause(
            DiagnosticCategory::UnexpectedError,
            BoundedDiagnosticCause::new(
                DiagnosticStage::RequestOutbound,
                code("transport.response_invalid"),
            )
            .with_object(InlineDiagnosticText::metadata("profusecontract")),
        )
    }
}

impl SourceCapture for SourceContext<'_> {
    fn boundary(
        &self,
        source: Option<SourceReceipt>,
        axes: &DiagnosticOutcomeAxes,
    ) -> DiagnosticSubmission {
        self.output
            .submit_boundary(source.map(|value| value.occurrence), axes, self.association)
    }

    fn submit_cause(
        &self,
        category: DiagnosticCategory,
        cause: BoundedDiagnosticCause,
    ) -> SourceReceipt {
        let diagnostic = BoundedDiagnostic::capture(category, CaptureSite::FirstObserved, cause);
        let submission = self.output.submit_bounded(
            Some(&diagnostic),
            &DiagnosticOutcomeAxes::default(),
            self.association,
        );
        SourceReceipt {
            diagnostic_id: diagnostic.id(),
            occurrence: diagnostic.occurrence(),
            submission,
        }
    }
}

impl dyn SourceCapture + '_ {
    /// Inspect at most sixteen source links, never Display/Debug a foreign error.
    #[track_caller]
    pub(crate) fn connect_failure(
        &self,
        error: &(dyn std::error::Error + 'static),
    ) -> SourceReceipt {
        let cause = BoundedDiagnosticCause::new(
            DiagnosticStage::RequestOutbound,
            code("transport.connect_failed"),
        )
        .with_object(InlineDiagnosticText::metadata("profusecontract"));
        let (cause, expected) = error_cause(cause, error);
        self.submit_cause(
            if expected {
                DiagnosticCategory::ExpectedRejection
            } else {
                DiagnosticCategory::UnexpectedError
            },
            cause,
        )
    }

    #[track_caller]
    pub(crate) fn rpc_failure(&self, status: &tonic::Status) -> SourceReceipt {
        use tonic::Code;
        let reason = 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 cause = BoundedDiagnosticCause::new(DiagnosticStage::RequestOutbound, code(reason))
            .with_object(InlineDiagnosticText::metadata("profusecontract"));
        let (cause, _) = error_cause(cause, status);
        let expected = matches!(
            status.code(),
            Code::Cancelled
                | Code::DeadlineExceeded
                | Code::ResourceExhausted
                | Code::InvalidArgument
                | Code::Unauthenticated
                | Code::PermissionDenied
        );
        self.submit_cause(
            if expected {
                DiagnosticCategory::ExpectedRejection
            } else {
                DiagnosticCategory::UnexpectedError
            },
            cause,
        )
    }
    #[track_caller]
    pub fn io_failure(&self, operation: IoOperation, error: &io::Error) -> SourceReceipt {
        let (stage, reason) = match operation {
            IoOperation::Accept => (DiagnosticStage::StartupListener, "transport.accept_io"),
            IoOperation::ReadHead => (DiagnosticStage::RequestDecode, "transport.read_head_io"),
            IoOperation::ReadBody => (DiagnosticStage::RequestDecode, "transport.read_body_io"),
            IoOperation::WriteResponse => (DiagnosticStage::RequestResponse, "transport.write_io"),
            IoOperation::ShutdownResponse => {
                (DiagnosticStage::RequestResponse, "transport.shutdown_io")
            }
            IoOperation::DeadlineWrite => (
                DiagnosticStage::RequestResponse,
                "transport.deadline_write_io",
            ),
        };
        let expected = !matches!(operation, IoOperation::Accept)
            && 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
                    | io::ErrorKind::InvalidInput
            );
        self.submit_cause(
            if expected {
                DiagnosticCategory::ExpectedRejection
            } else {
                DiagnosticCategory::UnexpectedError
            },
            BoundedDiagnosticCause::new(stage, code(reason))
                .with_system(io_kind(error), error.raw_os_error())
                .with_object(InlineDiagnosticText::metadata("profusegw")),
        )
    }
}

fn error_cause(
    mut cause: BoundedDiagnosticCause,
    error: &(dyn std::error::Error + 'static),
) -> (BoundedDiagnosticCause, bool) {
    let mut current = Some(error);
    for _ in 0..16 {
        let Some(error) = current else { break };
        if let Some(io) = error.downcast_ref::<io::Error>() {
            cause = cause.with_system(io_kind(io), io.raw_os_error());
            return (
                cause,
                matches!(
                    io.kind(),
                    io::ErrorKind::TimedOut | io::ErrorKind::Interrupted
                ),
            );
        }
        current = error.source();
    }
    (cause.with_system(code("unknown"), None), false)
}

/// Kept outside the write future, so cancelling that future cannot erase progress.
/// Observer only: caller retains the existing deadline and response ownership.
pub struct DeliveryProgress {
    written: u64,
    state: ResponseDelivery,
    operation: OperationOutcome,
    first_failure: Option<SourceReceipt>,
    terminal: bool,
}
impl Default for DeliveryProgress {
    fn default() -> Self {
        Self {
            written: 0,
            state: ResponseDelivery::NotStarted,
            operation: OperationOutcome::Unknown,
            first_failure: None,
            terminal: false,
        }
    }
}
impl DeliveryProgress {
    /// Call only for actual positive write results, never planned body length.
    pub fn wrote(&mut self, count: usize) {
        if self.terminal {
            return;
        }
        self.written = self.written.saturating_add(count as u64);
        if count > 0 {
            self.state = ResponseDelivery::Partial;
        }
    }
    pub fn local_write_complete(&mut self) {
        if self.terminal {
            return;
        }
        self.terminal = true;
        self.state = ResponseDelivery::LocalWriteComplete;
        self.operation = OperationOutcome::Succeeded;
    }
    pub fn failed(&mut self, receipt: SourceReceipt) {
        if self.terminal {
            return;
        }
        self.terminal = true;
        if self.first_failure.is_none() {
            self.first_failure = Some(receipt);
        }
        self.state = ResponseDelivery::Failed;
        self.operation = OperationOutcome::Failed;
    }
    pub fn timed_out(&mut self) {
        if self.terminal {
            return;
        }
        self.terminal = true;
        self.state = ResponseDelivery::TimedOut;
        self.operation = OperationOutcome::TimedOut;
    }
    pub fn cancelled(&mut self) {
        if self.terminal {
            return;
        }
        self.terminal = true;
        self.state = ResponseDelivery::Cancelled;
        self.operation = OperationOutcome::Cancelled;
    }
    pub fn first_failure(&self) -> Option<SourceReceipt> {
        self.first_failure
    }
    pub fn axes(&self) -> DiagnosticOutcomeAxes {
        DiagnosticOutcomeAxes {
            operation: self.operation,
            delivery: self.state,
            bytes_written: Some(self.written),
            ..DiagnosticOutcomeAxes::default()
        }
    }
}

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

    #[tokio::test]
    async fn source_immediate_submission_and_reference_are_secret_free() {
        use saddle_observability::{EmergencyDiagnostics, FileLoggingConfig, Rotation};
        let directory = std::env::temp_dir().join(format!(
            "transport-source-{}-{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        std::fs::create_dir(&directory).unwrap();
        let mut writer =
            EmergencyDiagnostics::start(&FileLoggingConfig::new(&directory, Rotation::Daily))
                .unwrap();
        let handle = writer.handle();
        let source = SourceContext {
            output: &handle,
            association: None,
        };
        let receipt = source.rpc_failure(&tonic::Status::internal(
            "Bearer SECRET_TOKEN https://secret.invalid/payload",
        ));
        assert_eq!(receipt.submission, DiagnosticSubmission::Enqueued);
        assert_eq!(
            source.boundary(Some(receipt), &DiagnosticOutcomeAxes::default()),
            DiagnosticSubmission::Enqueued
        );
        // Actual connection error is captured before conversion, without formatting URL.
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let address = listener.local_addr().unwrap();
        drop(listener);
        let endpoint = crate::ProfuseContractEndpoint::new(format!("http://{address}")).unwrap();
        let deadline = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis() as i64
            + 1000;
        let error =
            match crate::TonicBoundary::connect_until(endpoint, deadline, Some(&source)).await {
                Ok(_) => panic!("unexpected connect"),
                Err(error) => error,
            };
        assert!(error.diagnostic().is_none());
        assert_eq!(
            error.source_receipt().unwrap().submission,
            DiagnosticSubmission::Enqueued
        );
        use std::io::Write;
        let (mut socket, peer) = std::os::unix::net::UnixStream::pair().unwrap();
        let mut delivery = DeliveryProgress::default();
        delivery.wrote(socket.write(b"part").unwrap());
        drop(peer);
        let io = socket.write(b"remaining").unwrap_err();
        delivery.failed(source.io_failure(IoOperation::WriteResponse, &io));
        let axes = serde_json::to_value(delivery.axes()).unwrap();
        assert_eq!(axes["bytes_written"], 4);
        assert_eq!(axes["delivery"], "failed");
        assert_eq!(axes["business"], "unknown");
        // Actual cancellation of a polled future emits source then ID-only result.
        let mut future = Box::pin(observe_future(Some(&source), std::future::pending::<()>()));
        use std::future::Future;
        assert!(
            future
                .as_mut()
                .poll(&mut std::task::Context::from_waker(std::task::Waker::noop()))
                .is_pending()
        );
        drop(future);
        let stop = std::time::Instant::now() + std::time::Duration::from_secs(2);
        while matches!(
            writer.shutdown(),
            saddle_observability::DiagnosticShutdown::Pending
        ) {
            assert!(
                std::time::Instant::now() < stop,
                "test writer did not drain"
            );
            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
        }
        let raw = std::fs::read_to_string(writer.target()).unwrap();
        for secret in [
            "SECRET_TOKEN",
            "secret.invalid",
            "payload",
            &address.to_string(),
        ] {
            assert!(!raw.contains(secret));
        }
        let rows: Vec<serde_json::Value> = raw
            .lines()
            .map(|line| serde_json::from_str(line).unwrap())
            .collect();
        assert_eq!(rows.len(), 6);
        assert!(rows[0]["diagnostic"].is_object());
        assert!(rows[1]["diagnostic"].is_null());
        assert_eq!(
            rows[0]["diagnostic_reference"],
            rows[1]["diagnostic_reference"]
        );
        assert!(raw.contains("connection_refused"));
        assert!(raw.contains("broken_pipe"));
        assert!(raw.contains("transport.cancelled"));
        assert_eq!(rows[5]["axes"]["operation"], "cancelled");
        assert!(rows[5]["diagnostic"].is_null());
        assert_eq!(writer.snapshot().written, 6);
        assert_eq!(
            source
                .rpc_failure(&tonic::Status::cancelled("SECRET_TOKEN"))
                .submission,
            DiagnosticSubmission::Closed
        );
        std::fs::remove_dir_all(&directory).unwrap();
    }

    #[test]
    fn delivery_progress_survives_cancel_without_business_claim() {
        let mut progress = DeliveryProgress::default();
        progress.wrote(7);
        progress.timed_out();
        progress.wrote(99);
        progress.local_write_complete();
        let axes = serde_json::to_value(progress.axes()).unwrap();
        assert_eq!(axes["bytes_written"], 7);
        assert_eq!(axes["delivery"], "timed_out");
        assert_eq!(axes["business"], "unknown");
        assert_eq!(axes["physical"], "unknown");
        assert_eq!(axes["cleanup"], "unknown");
        let mut complete = DeliveryProgress::default();
        complete.wrote(9);
        complete.local_write_complete();
        // A later shutdown must not erase successful local write facts.
        complete.cancelled();
        let axes = serde_json::to_value(complete.axes()).unwrap();
        assert_eq!(axes["delivery"], "local_write_complete");
        assert_eq!(axes["business"], "unknown");
    }
}