saddle-boundary 0.3.34

//! Original-request-owned standard unary attempt and supervised tonic driver.
//! The attempt remains outside cancellation; native messages retain their bills.
use crate::{
    ExecutionCertainty, ProfuseContractAuthorityTemplate, rpc_driver::RpcCall, standard_codec,
    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(
    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> {
    let remaining = attempt.remaining()?;
    let wall = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)
        .map_err(|error| attempt.capture(&error, FailureKind::Configuration, None))?;
    let effective = trace.deadline_unix_ms.min(i64::try_from(wall.as_millis().saturating_add(remaining.as_millis())).unwrap_or(i64::MAX));
    let until = std::time::Duration::from_millis(effective.max(0) as u64).saturating_sub(wall);
    attempt.deadline = attempt.deadline.min(Instant::now() + until);
    attempt.remaining()?;
    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 result = {
        let call = RpcCall::new(stage, |executor| async {
            attempt.started = true;
            attempt.remaining()?;
            let endpoint = template
                .resolve(zone)
                .map_err(|error| attempt.capture(&error, FailureKind::Configuration, None))?;
            let authorization = endpoint.1;
            // URI is a trusted driver object; the separately retained diagnostic
            // authority is copied under an independent original-account bill.
            let endpoint = tonic::transport::Endpoint::from_shared(endpoint.0)
                .map_err(|error| attempt.capture(&error, FailureKind::Configuration, None))?;
            let authority = endpoint
                .uri()
                .authority()
                .expect("validated endpoint")
                .as_str();
            attempt.authority = Some(
                memory
                    .framework_output(|b| b.copy_text(&[authority]))
                    .map_err(|error| attempt.capture(&error, FailureKind::Resource, None))?,
            );
            attempt.phase = "connect";
            let remaining = attempt.remaining()?;
            let channel = endpoint
                .connect_timeout(remaining)
                .timeout(remaining)
                .concurrency_limit(1)
                .buffer_size(1)
                .executor(executor)
                .connect()
                .await
                .map_err(|error| attempt.capture(&error, FailureKind::Connection, None))?;
            attempt.phase = "ready";
            let mut client = tonic::client::Grpc::new(channel)
                .max_encoding_message_size(crate::MAX_UNARY_PAYLOAD_BYTES)
                .max_decoding_message_size(crate::MAX_UNARY_PAYLOAD_BYTES);
            client
                .ready()
                .await
                .map_err(|error| attempt.capture(&error, FailureKind::Connection, None))?;
            let mut request = tonic::Request::new(request);
            request.set_timeout(attempt.remaining()?);
            if let Some(authorization) = authorization {
                request
                    .metadata_mut()
                    .insert("authorization", authorization.0);
            }
            for (key, value) in crate::grouped_context::KEYS.iter().zip(grouped_metadata.get()) {
                request.metadata_mut().insert_bin(*key, value.clone());
            }
            for mapping in mappings {
                let value = match mapping.source {
                    MetadataSource::TraceId => trace.trace_id,
                    MetadataSource::RpcId => ids.rpc(),
                };
                match &mapping.header {
                    MetadataHeader::Ascii(key) => {
                        let value =
                            tonic::metadata::MetadataValue::try_from(value).map_err(|error| {
                                attempt.capture(&error, FailureKind::Configuration, None)
                            })?;
                        request.metadata_mut().insert(key.clone(), value);
                    }
                    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));
            attempt.phase = "unary";
            attempt.sent = true;
            let response = client
                .unary(
                    request,
                    tonic::codegen::http::uri::PathAndQuery::from_static(spec.path),
                    standard_codec::WireCodec { memory },
                )
                .await
                .map_err(|status| attempt.capture(&status, FailureKind::Remote, Some(&status)))?
                .into_inner();
            let response =
                response.map_err(|error| attempt.capture(&error, FailureKind::Resource, None))?;
            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";
            Ok(response)
        });
        match call {
            Err(error) => Err(error),
            Ok(call) => call.await,
        }
    };
    match result {
        Ok(result) => {
            if result.is_ok() {
                attempt.completed = true;
            }
            result
        }
        Err(error) => Err(attempt.capture(&error, FailureKind::Resource, 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());
    }
}