saddle-boundary 0.3.23

//! Required request-path diagnostics. Source ownership survives error mapping
//! and cancellation; Service owns the actual task, write loop and deadline.
use crate::source_diagnostics::{SourceCapture, SourceReceipt, code};
use crate::{
    BoundaryError, ExecutionCertainty, InvokeRequest, InvokeResponse,
    ProfuseContractAuthorityTemplate, TechnicalCode, TonicBoundary,
};
use saddle_core::{
    BoundedDiagnosticCause, DiagnosticCategory, DiagnosticOutcomeAxes, DiagnosticStage,
    InlineDiagnosticText, OperationOutcome, ResponseDelivery,
};
use saddle_observability::{
    DiagnosticSubmission, EmergencyDiagnosticHandle, FrameworkRequestFailure,
    RequestDiagnosticScope, RequestSourceReceipt,
};
use std::sync::Mutex;

/// Original technical classification, never a remote message or response body.
#[derive(Debug, PartialEq)]
pub struct BoundaryFailure {
    pub code: TechnicalCode,
    pub certainty: ExecutionCertainty,
}
/// ```compile_fail
/// use saddle_boundary::request_diagnostics::{RequiredBoundaryFailure, BoundaryFailure};
/// fn naked(e: BoundaryFailure) -> Result<(), RequiredBoundaryFailure> { Err(e) }
/// ```
/// ```compile_fail
/// use saddle_boundary::request_diagnostics::RequiredBoundaryFailure;
/// fn duplicate(e: RequiredBoundaryFailure) { let _ = e.clone(); }
/// ```
pub type RequiredBoundaryFailure = FrameworkRequestFailure<BoundaryFailure>;

fn classification(error: &BoundaryError) -> BoundaryFailure {
    BoundaryFailure {
        code: error.code,
        certainty: error.certainty,
    }
}

struct AttemptState<'a> {
    scope: RequestDiagnosticScope<'a>,
    receipt: Option<RequestSourceReceipt>,
    started: bool,
    completed: bool,
}

/// Caller-owned, outside the outbound future. No Clone/Default or naked-error
/// conversion. `finish` consumes cancellation evidence after dropping the future.
pub struct RequestAttempt {
    state: Mutex<AttemptState<'static>>,
    output: Option<EmergencyDiagnosticHandle>,
}

#[must_use]
pub enum AttemptCompletion {
    /// The returned Result already owns any technical failure credential.
    Returned,
    Interrupted(RequiredBoundaryFailure),
}

impl RequestAttempt {
    /// Keep the actual child projection for facade decoding after transport.
    pub fn diagnostic_scope(&self) -> RequestDiagnosticScope<'_> {
        self.state
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .scope
            .reborrow()
            .with_output(self.output.as_ref())
    }

    /// Capture a legacy/fake or pre-invoke failure without abandoning the
    /// attempt owner. The returned carrier is the sole failure credential.
    pub fn capture_boundary_failure(&mut self, error: BoundaryError) -> RequiredBoundaryFailure {
        let mut state = self.state.lock().unwrap_or_else(|p| p.into_inner());
        state.completed = true;
        state
            .scope
            .reborrow()
            .with_output(self.output.as_ref())
            .fail(
                classification(&error),
                DiagnosticCategory::UnexpectedError,
                cause("transport.adapter_failure"),
            )
    }

    pub fn complete_adapter(&mut self) {
        self.state
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .completed = true;
    }
    /// Supply the same provider's scope/output. None is a real degraded outlet,
    /// never permission to omit capture. This retains the scope's facts.
    pub fn new(
        scope: RequestDiagnosticScope<'_>,
        output: Option<&EmergencyDiagnosticHandle>,
    ) -> Self {
        Self {
            state: Mutex::new(AttemptState {
                scope: scope.with_output(None),
                receipt: None,
                started: false,
                completed: false,
            }),
            output: output.cloned(),
        }
    }

    pub(crate) fn bind_child(
        &self,
        call: &saddle_core::CallContext,
        event: &saddle_observability::EventContext,
    ) -> Result<(), BoundaryError> {
        let mut state = self.state.lock().unwrap_or_else(|p| p.into_inner());
        match state.scope.derive_outbound_child(call, event) {
            Ok(child) => {
                state.scope = child;
                Ok(())
            }
            Err(_) => {
                // Invalid supplied correlation is rejected before connecting;
                // the original parent scope is retained, never overwritten.
                state.receipt = Some(
                    state
                        .scope
                        .reborrow()
                        .with_output(self.output.as_ref())
                        .fail(
                            (),
                            DiagnosticCategory::UnexpectedError,
                            cause("transport.child_context_conflict"),
                        ),
                );
                Err(BoundaryError::invalid_request(
                    "outbound diagnostic context conflict",
                ))
            }
        }
    }

    /// For construction/fake errors in S before the real transport call. Moves
    /// only safe classification; never clones or formats the foreign carrier.
    #[track_caller]
    pub fn construction_failure(self, error: BoundaryError) -> RequiredBoundaryFailure {
        let state = self.state.into_inner().unwrap_or_else(|p| p.into_inner());
        state.scope.with_output(self.output.as_ref()).fail(
            classification(&error),
            DiagnosticCategory::ExpectedRejection,
            cause("transport.request_invalid"),
        )
    }

    /// Call after dropping an interrupted invoke future (including before its
    /// first poll). No remote certainty is upgraded and no retry is performed.
    pub fn finish(self) -> AttemptCompletion {
        let state = self.state.into_inner().unwrap_or_else(|p| p.into_inner());
        if state.completed {
            return AttemptCompletion::Returned;
        }
        let receipt = state.receipt.unwrap_or_else(|| {
            state
                .scope
                .reborrow()
                .with_output(self.output.as_ref())
                .fail(
                    (),
                    DiagnosticCategory::ExpectedRejection,
                    cause("transport.cancelled"),
                )
        });
        AttemptCompletion::Interrupted(receipt.map_error(|()| BoundaryFailure {
            code: TechnicalCode::TransportFailure,
            certainty: if state.started {
                ExecutionCertainty::MayHaveExecuted
            } else {
                ExecutionCertainty::NotExecuted
            },
        }))
    }
}

fn cause(reason: &'static str) -> BoundedDiagnosticCause {
    BoundedDiagnosticCause::new(DiagnosticStage::RequestOutbound, code(reason))
        .with_object(InlineDiagnosticText::metadata("profusecontract"))
}

impl SourceCapture for RequestAttempt {
    fn required(&self) -> bool {
        true
    }
    #[track_caller]
    fn submit_cause(
        &self,
        category: DiagnosticCategory,
        cause: BoundedDiagnosticCause,
    ) -> SourceReceipt {
        let mut state = self.state.lock().unwrap_or_else(|p| p.into_inner());
        if state.receipt.is_none() {
            state.receipt = Some(
                state
                    .scope
                    .reborrow()
                    .with_output(self.output.as_ref())
                    .fail((), category, cause),
            );
        }
        let receipt = state
            .receipt
            .as_ref()
            .expect("source captured before conversion");
        SourceReceipt {
            diagnostic_id: receipt.source_diagnostic().id(),
            submission: receipt.submission(),
            occurrence: receipt.source_diagnostic().occurrence(),
        }
    }
    fn boundary(
        &self,
        _: Option<SourceReceipt>,
        axes: &DiagnosticOutcomeAxes,
    ) -> DiagnosticSubmission {
        let state = self.state.lock().unwrap_or_else(|p| p.into_inner());
        state
            .receipt
            .as_ref()
            .map_or(DiagnosticSubmission::OutputUnavailable, |r| {
                r.record_boundary_optional(self.output.as_ref(), axes)
            })
    }
}

impl TonicBoundary {
    /// Already-connected facade path: same unary classifier, original supplied
    /// scope, no invented child call when this legacy transport owns none.
    pub async fn invoke_required(
        &self,
        request: InvokeRequest,
        attempt: &mut RequestAttempt,
    ) -> Result<InvokeResponse, RequiredBoundaryFailure> {
        attempt
            .state
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .started = true;
        let result = crate::source_diagnostics::observe_future(
            Some(attempt),
            self.invoke_with_source(request, Some(attempt)),
        )
        .await;
        let mut state = attempt.state.lock().unwrap_or_else(|p| p.into_inner());
        state.completed = true;
        result.map_err(|error| {
            state
                .receipt
                .take()
                .expect("unary classification captures before returning")
                .map_error(|()| classification(&error))
        })
    }

    /// Required production seam for S. All original validation/connect/unary/
    /// response classifiers and network operations are shared with the old path.
    pub async fn invoke_routed_required(
        template: &ProfuseContractAuthorityTemplate,
        request: InvokeRequest,
        attempt: &mut RequestAttempt,
    ) -> Result<InvokeResponse, RequiredBoundaryFailure> {
        {
            let mut state = attempt.state.lock().unwrap_or_else(|p| p.into_inner());
            if state.started {
                return Err(state
                    .scope
                    .reborrow()
                    .with_output(attempt.output.as_ref())
                    .fail(
                        BoundaryFailure {
                            code: TechnicalCode::FunctionRequestInvalid,
                            certainty: ExecutionCertainty::NotExecuted,
                        },
                        DiagnosticCategory::UnexpectedError,
                        cause("transport.attempt_reused"),
                    ));
            }
            state.started = true;
        }
        let result =
            Self::invoke_routed_inner(template, request, attempt.output.as_ref(), Some(attempt))
                .await;
        let mut state = attempt.state.lock().unwrap_or_else(|p| p.into_inner());
        state.completed = true;
        match result {
            Ok(response) => Ok(response),
            Err(error) => {
                let receipt = state
                    .receipt
                    .take()
                    .expect("every internal error captures at its source");
                Err(receipt.map_error(|()| classification(&error)))
            }
        }
    }
}

/// Mandatory response result. LocalWriteComplete is not peer acknowledgement.
pub type DeliveryFailure = FrameworkRequestFailure<DeliveryReason>;
#[derive(Clone, Copy, Debug, PartialEq)]
pub enum DeliveryReason {
    Io(std::io::ErrorKind),
    Timeout,
    Cancelled,
}
#[must_use]
/// ```compile_fail
/// use saddle_boundary::request_diagnostics::DeliveryOutcome;
/// fn discard_delivery(result: DeliveryOutcome) -> bool { result }
/// ```
pub enum DeliveryOutcome {
    Complete(DiagnosticOutcomeAxes),
    Failed {
        axes: DiagnosticOutcomeAxes,
        failure: DeliveryFailure,
    },
}
#[must_use]
pub enum WriteStep {
    Progress,
    Stopped,
}

/// Keep outside any select!/timeout future. No timer, task, queue or network
/// ownership. A failed/partial write never licenses a second response.
pub struct RequestDelivery<'a> {
    scope: RequestDiagnosticScope<'a>,
    written: u64,
    complete: bool,
    failure: Option<DeliveryFailure>,
}

#[cfg(test)]
mod tests {
    use super::*;
    use saddle_observability::{
        DiagnosticTaskId, DiagnosticZone, EarlyRequestContext, EventContext, RequestIdentity,
        RouteIdentity,
    };
    use std::time::{SystemTime, UNIX_EPOCH};

    fn early() -> RequestDiagnosticScope<'static> {
        RequestDiagnosticScope::early(None, EarlyRequestContext::socket_accepted("fixture"))
    }
    #[test]
    fn required_delivery_retains_partial_and_failure() {
        let mut progress = RequestDelivery::new(early());
        use std::io::Write;
        let (mut socket, peer) = std::os::unix::net::UnixStream::pair().unwrap();
        assert!(matches!(
            progress.record_write(socket.write(b"part")),
            WriteStep::Progress
        ));
        drop(peer);
        assert!(matches!(
            progress.record_write(socket.write(b"remaining")),
            WriteStep::Stopped
        ));
        progress.local_write_complete(); // cannot overwrite a failed terminal
        progress.timed_out();
        let DeliveryOutcome::Failed { axes, failure } = progress.finish() else {
            panic!("lost failure")
        };
        assert_eq!(axes.bytes_written, Some(4));
        assert!(matches!(axes.delivery, ResponseDelivery::Failed));
        let id = failure.source_diagnostic().id();
        let (retained, delivery) = failure
            .map_error(|e| e)
            .finish_boundary_retained(None, &axes);
        assert_eq!(retained.source_diagnostic().id(), id);
        assert_eq!(
            delivery.source_submission(),
            DiagnosticSubmission::OutputUnavailable
        );
        assert_eq!(
            *retained.error(),
            DeliveryReason::Io(std::io::ErrorKind::BrokenPipe)
        );

        let mut complete = RequestDelivery::new(early());
        assert!(matches!(complete.record_write(Ok(8)), WriteStep::Progress));
        complete.local_write_complete();
        complete.io_failure(&std::io::Error::from(std::io::ErrorKind::BrokenPipe));
        let DeliveryOutcome::Failed { axes, .. } = complete.finish() else {
            panic!("shutdown lost")
        };
        assert!(matches!(
            axes.delivery,
            ResponseDelivery::LocalWriteComplete
        ));
        assert!(matches!(axes.operation, OperationOutcome::Failed));

        let mut cancelled = RequestDelivery::new(early());
        assert!(matches!(cancelled.record_write(Ok(2)), WriteStep::Progress));
        let DeliveryOutcome::Failed { axes, .. } = cancelled.finish() else {
            panic!("incomplete succeeded")
        };
        assert_eq!(axes.bytes_written, Some(2));
        assert!(matches!(axes.operation, OperationOutcome::Cancelled));
    }

    #[tokio::test]
    async fn required_cancel_and_early_failure() {
        let mut attempt = RequestAttempt::new(early(), None);
        let template = ProfuseContractAuthorityTemplate::new("http://localhost:1").unwrap();
        let result = TonicBoundary::invoke_routed_required(
            &template,
            InvokeRequest::default(),
            &mut attempt,
        )
        .await;
        let failure = match result {
            Err(e) => e,
            Ok(_) => panic!("invalid accepted"),
        };
        assert_eq!(failure.error().code, TechnicalCode::FunctionRequestInvalid);
        assert_eq!(
            failure.submission(),
            DiagnosticSubmission::OutputUnavailable
        );
        assert!(matches!(attempt.finish(), AttemptCompletion::Returned));
        let attempt = RequestAttempt::new(early(), None);
        {
            let future = crate::source_diagnostics::observe_future(
                Some(&attempt),
                std::future::pending::<()>(),
            );
            tokio::pin!(future);
            tokio::select! { biased; _ = &mut future => unreachable!(), _ = tokio::task::yield_now() => () }
        }
        let AttemptCompletion::Interrupted(failure) = attempt.finish() else {
            panic!("cancel lost")
        };
        assert_eq!(
            failure.submission(),
            DiagnosticSubmission::OutputUnavailable
        );
    }

    fn request(id: &str) -> InvokeRequest {
        InvokeRequest::unary(
            id,
            "call-1",
            crate::InvocationTarget::new("fixture", "查询").unwrap(),
            SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .unwrap()
                .as_millis() as i64
                + 3000,
            crate::CallerContext {
                trace_info: Some(crate::proto::TraceInfo {
                    trace_id: "trace-1".into(),
                    rpc_id: "0.1".into(),
                }),
                ldc_info: Some(crate::proto::LdcInfo {
                    zone: "LoCaLhOsT".into(),
                    idc: "i1".into(),
                    env: "test".into(),
                }),
            },
            vec![8, 7],
        )
        .unwrap()
    }
    #[tokio::test]
    #[ignore = "audit-owned tonic server only"]
    async fn required_real_tonic() {
        use saddle_observability::{EmergencyDiagnostics, FileLoggingConfig, Rotation};
        saddle_observability::init(Default::default()).unwrap();
        let root = std::path::PathBuf::from(std::env::var("X_REQUIRED_LOG_ROOT").unwrap());
        let mut writer =
            EmergencyDiagnostics::start(&FileLoggingConfig::new(&root, Rotation::Daily)).unwrap();
        let handle = writer.handle();
        let port = std::env::var("SADDLE_DNS_TEST_PORT").unwrap();
        let template =
            ProfuseContractAuthorityTemplate::new(format!("http://{{zone}}:{port}")).unwrap();
        let mut failures = Vec::new();
        for id in [
            "absent",
            "diagnostic-error",
            "diagnostic-malformed",
            "diagnostic-technical",
            "expired",
            "refused",
        ] {
            let observer = saddle_observability::global().unwrap();
            let (parent, _) = observer
                .start_external_call_with_rpc(
                    "fixture",
                    "ingress",
                    "handler",
                    "query",
                    Some("trace-1"),
                    saddle_core::RpcCorrelationId::new("0").unwrap(),
                )
                .unwrap();
            let event = EventContext::new(
                RequestIdentity::new(id).unwrap(),
                RouteIdentity::new("incoming").unwrap(),
                1,
            )
            .unwrap();
            let scope = RequestDiagnosticScope::established(&handle, parent.context(), &event)
                .with_task(DiagnosticTaskId::from_runtime_id("42").unwrap())
                .unwrap_or_else(|_| panic!("task"))
                .with_zone(DiagnosticZone::from_validated_ingress("localhost").unwrap());
            let mut attempt = RequestAttempt::new(scope, Some(&handle));
            let mut req = request(id);
            if id == "expired" {
                req.deadline_unix_ms = 1;
            }
            let refused = ProfuseContractAuthorityTemplate::new("http://127.0.0.1:1").unwrap();
            let result = TonicBoundary::invoke_routed_required(
                if id == "refused" { &refused } else { &template },
                req,
                &mut attempt,
            )
            .await;
            if id == "absent" {
                assert!(result.is_ok());
            } else {
                let failure = match result {
                    Err(e) => e,
                    Ok(_) => panic!("missing failure"),
                };
                let source_id = failure.source_diagnostic().id();
                if id == "diagnostic-technical" {
                    assert_eq!(failure.error().code, TechnicalCode::TransportFailure);
                    assert_eq!(
                        failure.error().certainty,
                        ExecutionCertainty::MayHaveExecuted
                    );
                }
                let (failure, delivery) = failure
                    .map_error(|e| e)
                    .finish_boundary_retained(Some(&handle), &DiagnosticOutcomeAxes::default());
                assert_eq!(failure.source_diagnostic().id(), source_id);
                assert_eq!(delivery.source_submission(), DiagnosticSubmission::Enqueued);
                failures.push(source_id);
            }
            assert!(matches!(attempt.finish(), AttemptCompletion::Returned));
            parent.succeed();
        }
        for _ in 0..200 {
            if !matches!(
                writer.shutdown(),
                saddle_observability::DiagnosticShutdown::Pending
            ) {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
        }
        let raw = std::fs::read_to_string(writer.target()).unwrap();
        for secret in ["DO_NOT_LOG", "Bearer secret", "secret.invalid"] {
            assert!(!raw.contains(secret));
        }
        for expected in [
            "transport.rpc.internal",
            "transport.response_invalid",
            "transport.deadline_elapsed",
            "transport.connect_failed",
            "trace-1",
            "0.1",
            "42",
            "localhost",
        ] {
            assert!(raw.contains(expected), "missing {expected}");
        }
        assert_eq!(failures.len(), 5);
        let rows: Vec<serde_json::Value> = raw
            .lines()
            .map(|s| serde_json::from_str(s).unwrap())
            .collect();
        let sources: Vec<_> = rows
            .iter()
            .filter(|r| r["event"] == "framework.diagnostic")
            .collect();
        assert_eq!(sources.len(), 5);
        for source in sources {
            let boundaries: Vec<_> = rows
                .iter()
                .filter(|r| {
                    r["diagnostic"].is_null()
                        && r["diagnostic_reference"] == source["diagnostic_reference"]
                        && r.get("source_submission").is_some()
                })
                .collect();
            assert_eq!(boundaries.len(), 1);
            assert_eq!(boundaries[0]["context"], source["context"]);
            assert_eq!(source["context"]["rpc_id"]["value"], "0.1");
            assert_eq!(source["context"]["task"]["value"]["value"], "42");
            assert_eq!(source["context"]["zone"]["value"], "localhost");
        }
        println!(
            "X_REQUIRED_REAL_GO source-before-map retained child/task/zone IO/RPC/protocol/deadline normal"
        );
    }
}
impl<'a> RequestDelivery<'a> {
    pub fn new(scope: RequestDiagnosticScope<'a>) -> Self {
        Self {
            scope,
            written: 0,
            complete: false,
            failure: None,
        }
    }
    #[track_caller]
    pub fn record_write(&mut self, result: std::io::Result<usize>) -> WriteStep {
        if self.failure.is_some() || self.complete {
            return WriteStep::Stopped;
        }
        match result {
            Ok(n) if n > 0 => {
                self.written = self.written.saturating_add(n as u64);
                WriteStep::Progress
            }
            Ok(_) => {
                self.io_failure(&std::io::Error::from(std::io::ErrorKind::WriteZero));
                WriteStep::Stopped
            }
            Err(error) => {
                self.io_failure(&error);
                WriteStep::Stopped
            }
        }
    }
    /// Called only after all actual positive writes of this response completed.
    pub fn local_write_complete(&mut self) {
        if self.failure.is_none() {
            self.complete = true;
        }
    }
    /// Also handles shutdown error after local completion without erasing bytes
    /// or falsely claiming that previously written bytes were not delivered.
    #[track_caller]
    pub fn io_failure(&mut self, error: &std::io::Error) {
        if self.failure.is_some() {
            return;
        }
        self.failure = Some(
            self.scope.fail(
                DeliveryReason::Io(error.kind()),
                if matches!(
                    error.kind(),
                    std::io::ErrorKind::BrokenPipe
                        | std::io::ErrorKind::ConnectionReset
                        | std::io::ErrorKind::ConnectionAborted
                        | std::io::ErrorKind::TimedOut
                        | std::io::ErrorKind::UnexpectedEof
                        | std::io::ErrorKind::Interrupted
                        | std::io::ErrorKind::WouldBlock
                        | std::io::ErrorKind::InvalidData
                        | std::io::ErrorKind::InvalidInput
                ) {
                    DiagnosticCategory::ExpectedRejection
                } else {
                    DiagnosticCategory::UnexpectedError
                },
                BoundedDiagnosticCause::new(
                    DiagnosticStage::RequestResponse,
                    code("transport.write_io"),
                )
                .with_object(InlineDiagnosticText::metadata("profusegw"))
                .with_system(
                    crate::source_diagnostics::io_kind(error),
                    error.raw_os_error(),
                ),
            ),
        );
    }
    #[track_caller]
    pub fn timed_out(&mut self) {
        self.stop(DeliveryReason::Timeout);
    }
    #[track_caller]
    pub fn cancelled(&mut self) {
        self.stop(DeliveryReason::Cancelled);
    }
    #[track_caller]
    fn stop(&mut self, reason: DeliveryReason) {
        if self.failure.is_some() {
            return;
        }
        self.failure = Some(
            self.scope.fail(
                reason,
                DiagnosticCategory::ExpectedRejection,
                BoundedDiagnosticCause::new(
                    DiagnosticStage::RequestResponse,
                    code(if reason == DeliveryReason::Timeout {
                        "transport.delivery_timeout"
                    } else {
                        "transport.delivery_cancelled"
                    }),
                )
                .with_object(InlineDiagnosticText::metadata("profusegw")),
            ),
        );
    }
    pub fn bytes_written(&self) -> u64 {
        self.written
    }
    pub fn finish(mut self) -> DeliveryOutcome {
        if !self.complete && self.failure.is_none() {
            self.cancelled();
        }
        let mut axes = DiagnosticOutcomeAxes {
            bytes_written: Some(self.written),
            ..Default::default()
        };
        match self.failure {
            None => {
                axes.operation = OperationOutcome::Succeeded;
                axes.delivery = ResponseDelivery::LocalWriteComplete;
                DeliveryOutcome::Complete(axes)
            }
            Some(failure) => {
                axes.operation = match failure.error() {
                    DeliveryReason::Timeout => OperationOutcome::TimedOut,
                    DeliveryReason::Cancelled => OperationOutcome::Cancelled,
                    DeliveryReason::Io(_) => OperationOutcome::Failed,
                };
                axes.delivery = if self.complete {
                    ResponseDelivery::LocalWriteComplete
                } else {
                    match failure.error() {
                        DeliveryReason::Timeout => ResponseDelivery::TimedOut,
                        DeliveryReason::Cancelled => ResponseDelivery::Cancelled,
                        DeliveryReason::Io(_) => ResponseDelivery::Failed,
                    }
                };
                DeliveryOutcome::Failed { axes, failure }
            }
        }
    }
}