polyc-state-connect 2026.9.0

//! Carrying one trace across the State hop.
//!
//! A call that crosses a process boundary starts a new trace tree unless
//! something carries the caller's context with it, and a plane whose every
//! durable write happens on the far side of that boundary is exactly where a
//! broken trace hurts most: the commit and the decision that caused it end up
//! in different traces.
//!
//! So both halves are wired here. The client injects the active span's W3C
//! `traceparent` into every request, and the listener extracts it and
//! re-parents the span its handler runs in. The shared adapters live in
//! `polyc_runtime::propagation`, which the control plane and the execution
//! plane already use, so all three hops stitch into one tree rather than three
//! conventions.
//!
//! Both directions are cheap when no propagator is installed: the global getter
//! is a no-op, no header is written, and an absent `traceparent` degrades to
//! "start a new trace tree" rather than failing the call.

use connectrpc::client::CallOptions;

use crate::wire::DeclaredCall;

/// Returns per-call options carrying the active span's `traceparent`.
///
/// A client calls this on every request. This helper carries tracing only; the
/// capability facade that owns the call must still enforce its own budget.
#[must_use]
fn traced_options() -> CallOptions {
    let mut headers = http::HeaderMap::new();
    polyc_runtime::propagation::inject_current_span_into(&mut headers);
    options_from_headers(headers)
}

/// Builds call options from an already-captured W3C carrier.
///
/// The isolated Control journal bridge captures this on the caller's runtime,
/// before a State-client future moves to the bridge runtime. Other clients use
/// [`bounded_traced_options`] and capture their active span as usual.
#[must_use]
pub fn bounded_traced_options_from_headers(
    declared: &DeclaredCall,
    headers: http::HeaderMap,
) -> CallOptions {
    options_from_headers(headers).with_timeout(declared.remaining_budget())
}

/// Converts a W3C carrier into Connect call options.
fn options_from_headers(headers: http::HeaderMap) -> CallOptions {
    CallOptions::default().with_headers(
        headers
            .into_iter()
            .filter_map(|(name, value)| name.map(|name| (name, value))),
    )
}

/// Returns traced call options with an end-to-end transport budget.
///
/// The Connect timeout becomes the request's `Connect-Timeout-Ms`, so the
/// listener can refuse an already-spent call as well as the client cancelling
/// its wait. Capability facades choose the budget; this module does not invent
/// one shared across unrelated operations.
#[must_use]
pub fn bounded_traced_options(declared: &DeclaredCall) -> CallOptions {
    traced_options().with_timeout(declared.remaining_budget())
}

/// Returns traced options for a deliberately long-lived server stream.
///
/// Unary State calls must use [`bounded_traced_options`]. This exception
/// exists only for the commit subscription: its lifetime is the projector's
/// lifetime, and each delivered chunk is governed by the stream contract.
#[must_use]
pub(crate) fn streaming_traced_options() -> CallOptions {
    traced_options()
}

/// Returns the span one served call runs in, re-parented on the caller's trace.
///
/// `method` names the RPC, and the returned span records the trace the call was
/// adopted into — which is what makes propagation observable rather than
/// merely configured. A handler enters the span for the duration of the call.
#[must_use]
pub fn adopt_caller_trace(headers: &http::HeaderMap, method: &'static str) -> tracing::Span {
    use opentelemetry::trace::TraceContextExt as _;

    let context = polyc_runtime::propagation::extract_context_from(headers);
    let remote = context.span().span_context().trace_id();
    let span = tracing::info_span!(
        "state.call",
        rpc = method,
        // Recorded, not inferred: a test reads this back to prove the listener
        // adopted the caller's trace rather than starting its own.
        caller_trace_id = %remote
    );
    span.in_scope(|| {
        polyc_runtime::propagation::extract_parent_into_current_span(headers);
    });
    span
}

/// Returns the trace identity `headers` carries, as the listener would read it.
///
/// The observation half of [`adopt_caller_trace`], separated so a test can
/// compare what a client injected against what a listener extracted without
/// standing up a tracing subscriber.
#[must_use]
pub fn carried_trace_id(headers: &http::HeaderMap) -> String {
    use opentelemetry::trace::TraceContextExt as _;

    polyc_runtime::propagation::extract_context_from(headers)
        .span()
        .span_context()
        .trace_id()
        .to_string()
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs, clippy::unwrap_used)]

    use super::*;
    use std::time::Duration;

    /// The end-to-end claim, at the seam: what the client injects is what the
    /// listener extracts, and it is the caller's trace rather than a fresh one.
    #[test]
    fn what_a_client_injects_is_what_a_listener_extracts() {
        opentelemetry::global::set_text_map_propagator(
            opentelemetry_sdk::propagation::TraceContextPropagator::new(),
        );

        let mut carried = http::HeaderMap::new();
        carried.insert(
            "traceparent",
            "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01"
                .parse()
                .unwrap(),
        );
        assert_eq!(
            carried_trace_id(&carried),
            "4bf92f3577b34da6a3ce929d0e0e4736",
            "the listener reads the trace the caller named"
        );

        // The listener's span records the trace it adopted, which is how a
        // served call's propagation is observable at all.
        let span = adopt_caller_trace(&carried, "CommitJournalBatch");
        span.in_scope(|| {
            assert_eq!(
                carried_trace_id(&carried),
                "4bf92f3577b34da6a3ce929d0e0e4736"
            );
        });
    }

    /// A call with no `traceparent` starts its own tree instead of failing.
    #[test]
    fn a_call_without_a_trace_context_still_runs() {
        let bare = http::HeaderMap::new();
        assert_eq!(
            carried_trace_id(&bare),
            "00000000000000000000000000000000",
            "no context is the invalid trace id, not an error"
        );
        let _span = adopt_caller_trace(&bare, "GetJournalHead");
    }

    /// The client half writes a header only when there is a context to write.
    #[test]
    fn the_client_half_produces_call_options() {
        let declared = DeclaredCall::bounded(crate::state_audience(), Duration::from_millis(37));
        let options = bounded_traced_options(&declared);
        assert_eq!(
            options.timeout(),
            Some(Duration::from_millis(37)),
            "the Connect timeout is the exact semantic budget"
        );
    }
}