saddle-boundary 0.3.37

//! Original-request-owned standard unary attempt and supervised tonic driver.
//! The attempt remains outside cancellation; native messages retain their bills.
use crate::{
    ExecutionCertainty, ProfuseContractAuthorityTemplate,
    standard_pool::{StandardConnectionPool, TransportFailure},
    standard_wire::MethodSpec,
};
use saddle_admission::{ReadOnlyInput, RequestMemory, RpcStagePermit};
use saddle_core::{
    BoundedDiagnostic, BoundedDiagnosticCause, CaptureSite, DiagnosticCategory, DiagnosticStage,
    OperationOutcome,
};
use saddle_observability::{
    EmergencyDiagnosticHandle,
    root_diagnostic::{GrpcStatusFacts, GrpcTargetContext, RootOutcomeFacts, RootRequestEvent},
};
use saddle_runtime::request_task::reserved::{ReservedRequestFailure, ReservedRequestView};
use std::time::Instant;

#[derive(Clone, Copy, serde::Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum MetadataSource {
    TraceId,
    RpcId,
}
#[derive(serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MetadataConfig {
    pub header: String,
    pub source: MetadataSource,
}
enum MetadataHeader {
    Ascii(tonic::metadata::MetadataKey<tonic::metadata::Ascii>),
    Binary(tonic::metadata::MetadataKey<tonic::metadata::Binary>),
}
pub struct FrozenMetadata {
    header: MetadataHeader,
    source: MetadataSource,
}
impl FrozenMetadata {
    pub fn freeze(config: MetadataConfig) -> Result<Self, &'static str> {
        let key = config.header.as_str();
        if key != key.to_ascii_lowercase()
            || key.starts_with("grpc-")
            || ["authorization", "content-type", "te", "host", "user-agent", "trace-info-bin", "ldc-info-bin", "request-info-bin", "profuse-context-bin"].contains(&key)
        {
            return Err("reserved or noncanonical trace metadata header");
        }
        let header = if key.ends_with("-bin") {
            MetadataHeader::Binary(
                tonic::metadata::MetadataKey::from_bytes(key.as_bytes())
                    .map_err(|_| "invalid binary trace metadata header")?,
            )
        } else {
            MetadataHeader::Ascii(
                tonic::metadata::MetadataKey::from_bytes(key.as_bytes())
                    .map_err(|_| "invalid ASCII trace metadata header")?,
            )
        };
        Ok(Self {
            header,
            source: config.source,
        })
    }
}
#[derive(Clone, Copy)]
pub struct TraceContext<'a> {
    pub trace_id: &'a str,
    pub rpc_id: &'a str,
    pub request_id: &'a str,
    pub call_id: &'a str,
    pub zone: &'a str,
    pub idc: &'a str,
    pub env: &'a str,
    pub deadline_unix_ms: i64,
    pub number: u64,
    pub json: Option<&'a crate::ingress::IngressContextJson>,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum FailureKind {
    Configuration,
    Connection,
    Remote,
    Protocol,
    Resource,
    Deadline,
    Cancelled,
    Supervision,
}
#[must_use]
pub struct Failure {
    pub kind: FailureKind,
    pub certainty: ExecutionCertainty,
    source: ReservedRequestFailure,
}
impl Failure {
    pub fn into_parts(self) -> (FailureKind, ExecutionCertainty, ReservedRequestFailure) {
        (self.kind, self.certainty, self.source)
    }
}
#[derive(Debug)]
struct Local(&'static str);
impl std::fmt::Display for Local {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(self.0)
    }
}
impl std::error::Error for Local {}
/// Inline source owner with no unbilled dynamic strings or response copies.
pub struct Attempt<'a> {
    view: ReservedRequestView,
    output: Option<&'a EmergencyDiagnosticHandle>,
    spec: &'static MethodSpec,
    template: &'a str,
    business_unit: &'a str,
    authority: Option<ReadOnlyInput<String>>,
    number: u64,
    deadline: Instant,
    started: bool,
    sent: bool,
    completed: bool,
    phase: &'static str,
    response_bytes: Option<usize>,
}
impl<'a> Attempt<'a> {
    pub fn new(
        view: ReservedRequestView,
        output: Option<&'a EmergencyDiagnosticHandle>,
        spec: &'static MethodSpec,
        template: &'a str,
        business_unit: &'a str,
        number: u64,
        deadline: Instant,
    ) -> Self {
        Self {
            view,
            output,
            spec,
            template,
            business_unit,
            authority: None,
            number,
            deadline,
            started: false,
            sent: false,
            completed: false,
            phase: "preparation",
            response_bytes: None,
        }
    }
    pub fn certainty(&self) -> ExecutionCertainty {
        if self.sent {
            ExecutionCertainty::MayHaveExecuted
        } else {
            ExecutionCertainty::NotExecuted
        }
    }
    pub fn capture<E: std::error::Error>(
        &mut self,
        error: &E,
        kind: FailureKind,
        status: Option<&tonic::Status>,
    ) -> Failure {
        let reason = match kind {
            FailureKind::Configuration => "grpc.configuration",
            FailureKind::Connection => "grpc.connection",
            FailureKind::Remote => "grpc.remote_status",
            FailureKind::Protocol => "grpc.response_protocol",
            FailureKind::Resource => "grpc.resource",
            FailureKind::Deadline => "grpc.deadline",
            FailureKind::Cancelled => "grpc.cancelled",
            FailureKind::Supervision => "grpc.supervision",
        };
        let code = crate::source_diagnostics::code(reason);
        let operation = match kind {
            FailureKind::Deadline => OperationOutcome::TimedOut,
            FailureKind::Cancelled => OperationOutcome::Cancelled,
            FailureKind::Configuration | FailureKind::Resource => OperationOutcome::Rejected,
            _ => OperationOutcome::Failed,
        };
        let target = GrpcTargetContext {
            business_unit: self.business_unit,
            dependency_alias: self.spec.dependency,
            contract_sha256: &self.spec.contract,
            service: self.spec.service,
            method: self.spec.method,
            authority_template: self.template,
            authority: self.authority.as_ref().map(|v| v.get().as_str()),
            attempt: self.number,
            deadline_remaining_ns: self
                .deadline
                .saturating_duration_since(Instant::now())
                .as_nanos(),
            phase: self.phase,
            send_progress: if self.sent {
                "initiated_not_observed"
            } else {
                "not_started"
            },
            response_wire_bytes: self.response_bytes,
            status: status.map(|status| GrpcStatusFacts {
                code: status.code() as i32,
                message: status.message(),
                details: status.details(),
            }),
        };
        let source = match self.view.source_grpc_error_written(
            error,
            BoundedDiagnostic::capture(
                DiagnosticCategory::UnexpectedError,
                CaptureSite::FirstObserved,
                BoundedDiagnosticCause::new(DiagnosticStage::RequestOutbound, code),
            ),
            code,
            self.output,
            RootRequestEvent::Outbound,
            RootOutcomeFacts {
                axes: saddle_core::DiagnosticOutcomeAxes {
                    operation,
                    ..Default::default()
                },
                ..Default::default()
            },
            target,
        ) {
            Ok(written) => written.into_source(),
            Err(unconfirmed) => unconfirmed,
        };
        self.completed = true;
        Failure {
            kind,
            certainty: self.certainty(),
            source,
        }
    }
    pub fn finish(mut self) -> Option<Failure> {
        if self.completed {
            None
        } else {
            let kind = if Instant::now() >= self.deadline {
                FailureKind::Deadline
            } else {
                FailureKind::Cancelled
            };
            Some(self.capture(
                &Local("standard RPC interrupted before completion"),
                kind,
                None,
            ))
        }
    }
    fn remaining(&mut self) -> Result<std::time::Duration, Failure> {
        self.deadline
            .checked_duration_since(Instant::now())
            .filter(|v| !v.is_zero())
            .ok_or_else(|| {
                self.capture(
                    &Local("original request deadline elapsed"),
                    FailureKind::Deadline,
                    None,
                )
            })
    }
}

/// Only framework-owned raw wire crosses the dependency-audited executor.
/// All application registration/schema checks and generated callbacks occur
/// before this function. Physical drivers are destroyed before stage refund.
pub async fn unary(
    pool: Option<StandardConnectionPool>,
    template: &ProfuseContractAuthorityTemplate,
    zone: &str,
    spec: &'static MethodSpec,
    request: ReadOnlyInput<Vec<u8>>,
    memory: RequestMemory,
    attempt: &mut Attempt<'_>,
    stage: RpcStagePermit,
    mappings: &[FrozenMetadata],
    trace: TraceContext<'_>,
) -> Result<ReadOnlyInput<Vec<u8>>, Failure> {
    // The validated absolute fact is inherited unchanged. The original
    // monotonic owner independently bounds every local wait below; rebuilding
    // UTC from its sampled remaining duration loses sub-millisecond precision.
    attempt.remaining()?;
    let effective = trace.deadline_unix_ms;
    let ids = crate::grouped_context::ChildIds::new(trace.rpc_id, trace.call_id, trace.number)
        .map_err(|error| attempt.capture(&error, FailureKind::Configuration, None))?;
    let encoded = crate::grouped_context::encode(trace.json, trace.trace_id, ids.rpc(), trace.zone, trace.idc, trace.env, trace.request_id, ids.call(), effective)
        .map_err(|error| attempt.capture(&error, FailureKind::Resource, None))?;
    let grouped_metadata = encoded.metadata(&memory)
        .map_err(|error| attempt.capture(&error, FailureKind::Resource, None))?;
    let pool = pool.ok_or_else(|| attempt.capture(&Local("standard pool not started"), FailureKind::Resource, None))?;
    let endpoint = resolve_transport_endpoint(template, zone, &stage)
        .map_err(|error| attempt.capture(&error, FailureKind::Configuration, None))?;
    let target = pool.target_index(spec.dependency)
        .ok_or_else(|| attempt.capture(&Local("unknown frozen dependency"), FailureKind::Configuration, None))?;
    let authority = endpoint.0.strip_prefix("http://").expect("validated HTTP authority");
    attempt.authority = Some(memory.framework_output(|b| b.copy_text(&[authority]))
        .map_err(|error| attempt.capture(&error, FailureKind::Resource, None))?);
    let mut lease = pool.try_acquire_authority(target, &endpoint.0)
        .map_err(|error| attempt.capture(&error, FailureKind::Resource, None))?;
    let audit = stage.dependency_audit();
    lease.attach_stage(stage).map_err(|error| attempt.capture(&error, FailureKind::Resource, None))?;
    let authorization = endpoint.1;
    attempt.started = true;
    attempt.phase = "connect";
    lease.connect(attempt.deadline).await.map_err(|error| transport_failure(attempt, error))?;
    let request = prepare_transport_request(request, spec, authorization, grouped_metadata.get(),
        mappings, trace.trace_id, ids.rpc(), &audit)
        .map_err(|error| attempt.capture(&error, FailureKind::Configuration, None))?;
    attempt.phase = "unary";
    let response = lease.unary_transport_recorded(request, memory, spec.path, attempt.deadline, &mut attempt.sent).await
        .map_err(|error| transport_failure(attempt, error))?;
    attempt.phase = "decode";
    attempt.response_bytes = Some(response.get().len());
    spec.response.validate(response.get())
        .map_err(|error| attempt.capture(&error, FailureKind::Protocol, None))?;
    attempt.phase = "driver_cleanup";
    lease.drain_response(attempt.deadline).await.map_err(|error| transport_failure(attempt, error))?;
    lease.finish(true);
    attempt.completed = true;
    Ok(response)
}

// These stack-pinned, synchronous dependency futures perform only fixed tonic
// URI/header bookkeeping. No user callback or schema conversion enters them.
fn resolve_transport_endpoint(template: &ProfuseContractAuthorityTemplate, zone: &str,
    stage: &RpcStagePermit) -> Result<crate::ProfuseContractEndpoint, crate::BoundaryError> {
    let mut result = None;
    {
        let future = std::pin::pin!(async { result = Some(template.resolve(zone)); });
        let audit = stage.dependency_audit();
        assert!(audit.poll_driver(future, &mut std::task::Context::from_waker(std::task::Waker::noop())).is_ready());
    }
    result.unwrap()
}
fn prepare_transport_request(wire: ReadOnlyInput<Vec<u8>>, spec: &'static MethodSpec,
    authorization: Option<crate::BearerMetadata>,
    grouped: &[tonic::metadata::MetadataValue<tonic::metadata::Binary>], mappings: &[FrozenMetadata],
    trace_id: &str, rpc_id: &str, audit: &saddle_admission::RpcDependencyAudit,
) -> Result<tonic::Request<ReadOnlyInput<Vec<u8>>>, tonic::metadata::errors::InvalidMetadataValue> {
    let mut result = None;
    {
        let future = std::pin::pin!(async {
            let mut request = tonic::Request::new(wire);
            if let Some(authorization) = authorization { request.metadata_mut().insert("authorization", authorization.0); }
            for (key, value) in crate::grouped_context::KEYS.iter().zip(grouped) {
                request.metadata_mut().insert_bin(*key, value.clone());
            }
            for mapping in mappings {
                let value = match mapping.source { MetadataSource::TraceId => trace_id, MetadataSource::RpcId => rpc_id };
                match &mapping.header {
                    MetadataHeader::Ascii(key) => match tonic::metadata::MetadataValue::try_from(value) {
                        Ok(value) => { request.metadata_mut().insert(key.clone(), value); }
                        Err(error) => { result = Some(Err(error)); return; }
                    },
                    MetadataHeader::Binary(key) => { request.metadata_mut().insert_bin(key.clone(), tonic::metadata::MetadataValue::from_bytes(value.as_bytes())); }
                }
            }
            request.extensions_mut().insert(tonic::GrpcMethod::new(spec.service, spec.method));
            result = Some(Ok(request));
        });
        assert!(audit.poll_driver(future, &mut std::task::Context::from_waker(std::task::Waker::noop())).is_ready());
    }
    result.unwrap()
}

fn transport_failure(attempt: &mut Attempt<'_>, error: TransportFailure) -> Failure {
    match error {
        TransportFailure::Resource(error) => attempt.capture(&error, FailureKind::Resource, None),
        TransportFailure::Connection(error) | TransportFailure::Ready(error) => attempt.capture(&error, FailureKind::Connection, None),
        TransportFailure::Status(status) => attempt.capture(&status, FailureKind::Remote, Some(&status)),
        TransportFailure::Deadline => attempt.capture(&Local("original RPC deadline expired"), FailureKind::Deadline, None),
    }
}

#[cfg(test)]
mod grouped_metadata_config_tests {
    use super::*;
    #[test]
    fn mappings_cannot_replace_grouped_context_or_authorization() {
        for header in crate::grouped_context::KEYS.into_iter().chain(["authorization", "grpc-timeout"]) {
            for source in [MetadataSource::TraceId, MetadataSource::RpcId] {
                assert!(FrozenMetadata::freeze(MetadataConfig { header: header.into(), source }).is_err());
            }
        }
        assert!(FrozenMetadata::freeze(MetadataConfig {
            header: "x-user-trace-bin".into(), source: MetadataSource::TraceId,
        }).is_ok());
    }
}