gunnar-grpc 2.0.0

//! **A request id and a span around every gRPC call** — the tower layer, not an
//! interceptor.
//!
//! # What an operator could not do before this
//!
//! SSH has carried `conn=<id>` on a span since the connection span landed, and
//! smart HTTP has carried `req=<id>` — plus `x-gunnar-request-id` back on the wire — since
//! 2026-08-05. The gRPC control plane had **neither**. Its one observable was
//! [`crate::AUDIT_TARGET`]'s record per authorised call, and that record could
//! not be tied to a duration, to an HTTP status, to the refusal a caller was
//! looking at, or to a second line from the same call. An operator holding a
//! failed `gunnar control` invocation had no value to grep for.
//!
//! # Why a layer and not an interceptor
//!
//! [`crate::AuthGate`] is a `tonic` interceptor, which is the obvious place to
//! put this and the wrong one. An interceptor is
//! `FnMut(Request<()>) -> Result<Request<()>, Status>`: it is handed the request
//! **before** the handler and never sees the response. It therefore cannot time
//! a call, cannot read the status the call ended with, and cannot keep a span
//! open across the handler — a span it entered would close when the interceptor
//! returned, so every line the handler emitted would fall outside it. That is
//! the same defect `.instrument(span)` fixes on the HTTP side, where a
//! `tokio::spawn`ed transfer starts with no span at all.
//!
//! A `tower_layer::Layer` wraps the whole `Routes` service, so one
//! [`RequestSpan::call`] covers request, handler and response. It is applied in
//! [`crate::router`], once, to every service including the ungated
//! `Bootstrap` — the same by-construction placement as the gate itself, so
//! there is no way to build a gunnar gRPC router without it.
//!
//! # Where the id comes from, and why it is passed in
//!
//! The id **must come from the same counter SSH and HTTP mint from**, or two
//! lines in one log could carry the same `req=` for a git transfer and a
//! control call, which is worse than no id. That counter is
//! `gunnar_core::observe::RequestId`, and this crate deliberately has no
//! `gunnar-core` edge: `gunnar-core` pulls `gunnar-store`, `aws-lc-rs`, `sha2`
//! and `base64ct`, none of which the `--features client,control` graph carries
//! today, and `aws-lc-rs` is C with a build script.
//!
//! So the source is a [`RequestIdSource`] the caller names, and it is a
//! **required argument of [`crate::router`]** rather than a defaulted field.
//! There is no fallback counter in this crate, because a fallback is exactly how
//! two sequences come to exist without anyone deciding that they should.
//! `gunnar-cli` links both crates and
//! `the_grpc_request_id_renders_exactly_as_gunnar_core_renders_one` asserts the
//! two spellings agree, the same way
//! `the_audit_target_is_the_same_string_in_both_crates` pins the one other
//! constant this crate has to state twice.
//!
//! # What is on the wire
//!
//! Every response carries [`REQUEST_ID_METADATA_KEY`] as response metadata, on
//! **every outcome including every refusal**, stamped on the way out in one
//! place so no arm can forget it. `gunnar control` prints it on a failure, which
//! is what makes the id reach the person reading the error rather than only the
//! log they may not have.
//!
//! # What this layer does NOT see
//!
//! * **The principal.** The gate names it, inside the span, on
//!   [`crate::AUDIT_TARGET`] — so the audit record inherits `req` and the join
//!   is made there. Recording it here would need a second identity decision.
//! * **`grpc-status` on a success.** tonic sends `grpc-status: 0` in the
//!   **trailers** of a successful unary call, and a trailer is not readable
//!   without wrapping the response body. A *refused* call is a trailers-only
//!   response whose `grpc-status` is in the headers, and that is the case an
//!   operator is reading at 03:00, so it is the case this reports. A success
//!   logs `http_status=200` and no `grpc_status`, and that absence is rendered
//!   as an absent field rather than as `0`.
//! * **The body of a server stream.** `Events.Watch` returns its response
//!   immediately and then streams; the polling happens after
//!   [`RequestSpan::call`]'s future resolves, so `elapsed_ms` on a stream is the
//!   time to the first response and not the life of the subscription.

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

use tonic::body::Body;
use tonic::codegen::http;
use tonic::codegen::Service;
use tracing::Instrument as _;

use crate::auth::AUDIT_TARGET;

/// The response metadata key the request id goes back to the caller in.
///
/// Lower-case ASCII with no `-bin` suffix, so it is a legal gRPC text metadata
/// key and `tonic`'s `MetadataMap` will hand it to a client as a `&str`. Not
/// `x-`-prefixed, unlike the HTTP header: gRPC metadata is already a private
/// namespace per service, and `gunnar-` says whose it is.
pub const REQUEST_ID_METADATA_KEY: &str = "gunnar-request-id";

/// [`REQUEST_ID_METADATA_KEY`] as an `http::HeaderName`, built once at compile
/// time so the hot path does no parsing and cannot fail.
const REQUEST_ID_HEADER: http::HeaderName = http::HeaderName::from_static(REQUEST_ID_METADATA_KEY);

/// Where the ids come from. See the module docs for why this is injected and
/// not defaulted.
///
/// A plain `fn` pointer rather than a boxed closure: it is `Copy`, so the layer
/// is `Copy`, which is what `Router::serve` wants, and it costs no allocation
/// per call.
pub type RequestIdSource = fn() -> u64;

/// Render an id the way `gunnar_core::observe::RequestId` renders one: sixteen
/// lower-case hex digits, fixed width, greppable.
///
/// Stated here because this crate cannot link the type that owns the rendering,
/// and asserted equal to it by a test in `gunnar-cli`, which links both.
pub fn render_request_id(id: u64) -> String {
    format!("{id:016x}")
}

/// The layer. Wraps `Routes`, so it is entered once per **request**.
#[derive(Clone, Copy, Debug)]
pub struct RequestSpanLayer {
    ids: RequestIdSource,
}

impl RequestSpanLayer {
    /// Mint ids from `ids`.
    pub fn new(ids: RequestIdSource) -> RequestSpanLayer {
        RequestSpanLayer { ids }
    }
}

impl<S> tower_layer::Layer<S> for RequestSpanLayer {
    type Service = RequestSpan<S>;

    fn layer(&self, inner: S) -> RequestSpan<S> {
        RequestSpan {
            inner,
            ids: self.ids,
        }
    }
}

/// One gRPC call, under a span that names it.
#[derive(Clone, Copy, Debug)]
pub struct RequestSpan<S> {
    inner: S,
    ids: RequestIdSource,
}

impl<S, ResBody> Service<http::Request<Body>> for RequestSpan<S>
where
    S: Service<http::Request<Body>, Response = http::Response<ResBody>> + Clone + Send + 'static,
    S::Future: Send + 'static,
{
    type Response = http::Response<ResBody>;
    type Error = S::Error;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, S::Error>> + Send>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, mut request: http::Request<Body>) -> Self::Future {
        let id = (self.ids)();
        let rendered = render_request_id(id);
        // The gRPC path IS the method: `/gunnar.v1.Repos/Create`. Owned, because
        // the request is moved into the future.
        let method = request.uri().path().to_owned();
        // ── and the method the AUTH GATE checks the credential against ─────
        //
        // This layer is the only thing in the control plane that sees the URI
        // AND runs outside the interceptor: `InterceptedService` hands the
        // interceptor a `Request<()>` built by `Request::from_http`, which
        // drops the URI and keeps the extensions. So the one fact GUN-001 said
        // could not reach the gate travels the one way it can. Inserted for
        // every route including `Bootstrap` and including a path that matches
        // nothing, so a credential can never be checked against an absence.
        if let Some(called) = crate::auth::method_name_from_path(&method) {
            request
                .extensions_mut()
                .insert(crate::auth::CalledMethod(called));
        }
        let span = tracing::info_span!(
            target: AUDIT_TARGET,
            "grpc.request",
            req = %rendered,
            method = %method,
        );

        // The canonical tower clone dance: `self.inner` is the copy `poll_ready`
        // was called on, so it is the one that must serve this request. A clone
        // is left behind for the next one.
        let clone = self.inner.clone();
        let mut inner = std::mem::replace(&mut self.inner, clone);

        Box::pin(
            async move {
                // `debug`, not `info`. The audit record the gate writes on
                // `gunnar::audit` already marks arrival at `info` and names the
                // principal, so an entry line here would double every
                // control-plane log for nothing. The receive-pack phase log is
                // at `info` on entry because those phases run for minutes with
                // nothing else to show; a unary RPC does not.
                tracing::debug!(
                    target: AUDIT_TARGET,
                    event = "grpc.request.begin",
                    "a control-plane call arrived"
                );
                let started = std::time::Instant::now();
                let mut response = inner.call(request).await;
                let elapsed_ms = started.elapsed().as_millis() as u64;

                match &mut response {
                    Ok(response) => {
                        let http_status = response.status().as_u16();
                        // Trailers-only refusals put `grpc-status` here. A
                        // success puts it in the trailers, which this cannot
                        // read; the field is then ABSENT rather than rendered
                        // as `0`, because an absence rendered as a value is how
                        // `pack_bytes` came to log a failed `stat` as zero.
                        let grpc_status = response
                            .headers()
                            .get("grpc-status")
                            .and_then(|v| v.to_str().ok())
                            .and_then(|v| v.parse::<i32>().ok());
                        let grpc_message = response
                            .headers()
                            .get("grpc-message")
                            .and_then(|v| v.to_str().ok())
                            .map(str::to_owned);
                        // Stamped here, once, on the one return every arm goes
                        // through, so a refusal carries the id as surely as a
                        // success does.
                        response.headers_mut().insert(
                            REQUEST_ID_HEADER,
                            http::HeaderValue::from_str(&rendered)
                                .expect("a request id is ascii hex"),
                        );
                        // A refused call is `warn`: it is the line an operator
                        // is looking for, and `RUST_LOG=warn` must not hide it.
                        // M1: a box growing its data set answers `Unavailable
                        // [primary-growing]` by design; the caller waits. That is
                        // the box doing what it was told, said at INFO.
                        // The control store's own words for the same growth (`Error::OffStripe`,
                        // "off the stripe while the data set grows") are the same deferral: a
                        // browser retrying its login through a growth must not trip WARN spam.
                        let growing = crate::quiet::growth_hold(grpc_status, grpc_message.as_deref());
                        // A key the installer fed and the pairing loop has not enrolled
                        // yet (`auth::KEY_ENROLLING`): the caller asks again, by design.
                        let enrolling = grpc_status == Some(14) && grpc_message.as_deref().is_some_and(|m| m.contains("key-enrolling"));
                        if enrolling {
                            tracing::info!(
                                target: AUDIT_TARGET,
                                event = "grpc.request.done",
                                http_status,
                                grpc_status,
                                elapsed_ms,
                                "a control-plane call was deferred: the caller's key was fed to this box and is being enrolled"
                            );
                        } else if growing {
                            tracing::info!(
                                target: AUDIT_TARGET,
                                event = "grpc.request.done",
                                http_status,
                                grpc_status,
                                elapsed_ms,
                                "a control-plane call was deferred: this box is growing its data set"
                            );
                        } else if crate::quiet::unverified_answer(grpc_status, grpc_message.as_deref()) && crate::quiet::loud("grpc.request.done:not-verified").is_none() {
                            // A key not installed YET, retried (crate::quiet): said at
                            // WARN once a minute; this one at INFO.
                            tracing::info!(
                                target: AUDIT_TARGET,
                                event = "grpc.request.done",
                                http_status,
                                grpc_status,
                                grpc_message,
                                elapsed_ms,
                                "a control-plane call was refused (said at WARN this minute already)"
                            );
                        } else if grpc_status.is_some_and(|status| status != 0) {
                            tracing::warn!(
                                target: AUDIT_TARGET,
                                event = "grpc.request.done",
                                http_status,
                                grpc_status,
                                grpc_message,
                                elapsed_ms,
                                "a control-plane call was refused"
                            );
                        } else {
                            tracing::info!(
                                target: AUDIT_TARGET,
                                event = "grpc.request.done",
                                http_status,
                                grpc_status,
                                elapsed_ms,
                                "a control-plane call was served"
                            );
                        }
                    }
                    Err(_) => {
                        // `Routes`'s error is `Infallible`, so this arm is
                        // unreachable through `router`. It is written anyway
                        // because the layer is generic over the inner service
                        // and a silent hole here would be invisible.
                        tracing::warn!(
                            target: AUDIT_TARGET,
                            event = "grpc.request.failed",
                            elapsed_ms,
                            "the gRPC stack returned an error instead of a response"
                        );
                    }
                }
                response
            }
            .instrument(span),
        )
    }
}

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

    #[test]
    fn an_id_renders_fixed_width_lower_case_hex() {
        assert_eq!(render_request_id(0), "0000000000000000");
        assert_eq!(render_request_id(1), "0000000000000001");
        assert_eq!(render_request_id(u64::MAX), "ffffffffffffffff");
        // The width is the property a human reads a column of these by, and
        // `{:x}` without the `016` would pass every other assertion here.
        assert_eq!(render_request_id(0xdead_beef).len(), 16);
        assert!(render_request_id(0xABCD_EF01_2345_6789)
            .chars()
            .all(|c| c.is_ascii_hexdigit() && !c.is_ascii_uppercase()));
    }

    #[test]
    fn the_metadata_key_is_a_legal_grpc_text_key() {
        // gRPC metadata keys are lower-case ASCII; a `-bin` suffix would make
        // this a BINARY key, which a client reads as bytes and cannot print.
        assert_eq!(
            REQUEST_ID_METADATA_KEY,
            REQUEST_ID_METADATA_KEY.to_ascii_lowercase()
        );
        assert!(!REQUEST_ID_METADATA_KEY.ends_with("-bin"));
        // And the compile-time header name is the same string, not a second
        // spelling of it.
        assert_eq!(REQUEST_ID_HEADER.as_str(), REQUEST_ID_METADATA_KEY);
    }

    /// **The one mechanism GUN-001 said did not exist.**
    ///
    /// The old note in `auth` argued that a server interceptor is handed a
    /// `Request<()>` and cannot see the `:path`, so the credential could not
    /// bind the method. True of the interceptor; false of this layer, which is
    /// outside it and does see the URI. This drives the real `RequestSpan`
    /// over a real `http::Request` and asserts the inner service — which is
    /// where `InterceptedService`, and therefore `AuthGate`, sits — receives
    /// the method in the extensions.
    ///
    /// Mutation-proven 2026-09-11: with the `extensions_mut().insert(...)` in
    /// `call` removed, the first assertion reads `None` and the test is red.
    #[tokio::test]
    async fn the_span_layer_hands_the_called_method_down_to_the_interceptor() {
        use std::sync::{Arc, Mutex};

        #[derive(Clone)]
        struct Spy(Arc<Mutex<Option<Option<crate::auth::CalledMethod>>>>);

        impl Service<http::Request<Body>> for Spy {
            type Response = http::Response<Body>;
            type Error = std::convert::Infallible;
            type Future = std::future::Ready<Result<Self::Response, Self::Error>>;

            fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
                Poll::Ready(Ok(()))
            }

            fn call(&mut self, request: http::Request<Body>) -> Self::Future {
                *self.0.lock().expect("the spy") = Some(
                    request
                        .extensions()
                        .get::<crate::auth::CalledMethod>()
                        .cloned(),
                );
                std::future::ready(Ok(http::Response::new(Body::empty())))
            }
        }

        let seen = Arc::new(Mutex::new(None));
        let mut service =
            tower_layer::Layer::layer(&RequestSpanLayer::new(|| 1), Spy(Arc::clone(&seen)));
        let request = http::Request::builder()
            .uri("http://gunnar.invalid/gunnar.v1.Access/AddKey")
            .body(Body::empty())
            .expect("a request");
        service.call(request).await.expect("infallible");
        assert_eq!(
            seen.lock().expect("the spy").clone(),
            Some(Some(crate::auth::CalledMethod("Access.AddKey".to_owned()))),
            "the method the router dispatches on did not reach the gate"
        );

        // The other direction: a path that is not a gRPC method carries no
        // `CalledMethod` at all, rather than a made-up one a credential could
        // be matched against.
        let seen = Arc::new(Mutex::new(None));
        let mut service =
            tower_layer::Layer::layer(&RequestSpanLayer::new(|| 2), Spy(Arc::clone(&seen)));
        let request = http::Request::builder()
            .uri("http://gunnar.invalid/healthz")
            .body(Body::empty())
            .expect("a request");
        service.call(request).await.expect("infallible");
        assert_eq!(
            seen.lock().expect("the spy").clone(),
            Some(None),
            "a non-method path was given a method name"
        );
    }

    /// **M1 (release-3 test2 mock 2026-09-29): a box growing its data set is no WARN.** Main
    /// answered the twin's `Mirror.LogFetch` during its growth, and every answer was `WARN a
    /// control-plane call was refused` (7/min, the log gate's spam). A growing box answers
    /// `Unavailable` carrying [`crate::PRIMARY_GROWING`] by design and the caller waits: that
    /// line is INFO. Any other refusal stays WARN.
    #[tokio::test]
    async fn a_growing_boxs_deferral_is_info_and_any_other_refusal_stays_warn() {
        use std::sync::{Arc, Mutex};

        #[derive(Clone)]
        struct Answers(&'static str, &'static str);

        impl Service<http::Request<Body>> for Answers {
            type Response = http::Response<Body>;
            type Error = std::convert::Infallible;
            type Future = std::future::Ready<Result<Self::Response, Self::Error>>;

            fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
                Poll::Ready(Ok(()))
            }

            fn call(&mut self, _: http::Request<Body>) -> Self::Future {
                let mut r = http::Response::new(Body::empty());
                r.headers_mut().insert("grpc-status", http::HeaderValue::from_static(self.0));
                r.headers_mut().insert("grpc-message", http::HeaderValue::from_static(self.1));
                std::future::ready(Ok(r))
            }
        }

        #[derive(Clone, Default)]
        struct Heard(Arc<Mutex<Vec<u8>>>);
        impl std::io::Write for Heard {
            fn write(&mut self, b: &[u8]) -> std::io::Result<usize> {
                self.0.lock().unwrap().extend_from_slice(b);
                Ok(b.len())
            }
            fn flush(&mut self) -> std::io::Result<()> {
                Ok(())
            }
        }
        let heard = Heard::default();
        let h = heard.clone();
        let sub = tracing_subscriber::fmt().with_writer(move || h.clone()).with_ansi(false).finish();
        let _g = tracing::subscriber::set_default(sub);
        // Round 4's fresh UpCloud mockup: a login while the set grows is answered from the control
        // store's OffStripe (gunnar-core error.rs) — the same growth, said in the store's words.
        let off_stripe = "unavailable:%20the%20control%20store%20is%20off%20the%20stripe%20while%20the%20data%20set%20grows%20(the%20transaction%20log%20is%20released%20for%20the%20remount);%20writes%20are%20refused%20until%20/data%20is%20mounted%20again%20%E2%80%94%20retry%20in%20a%20few%20minutes";
        for (status, message) in [("14", "%5Bprimary-growing%5D%20this%20box's%20data%20set%20is%20being%20grown"), ("7", "principal%20x%20has%20no%20grant"), ("14", off_stripe)] {
            let mut service = tower_layer::Layer::layer(&RequestSpanLayer::new(|| 3), Answers(status, message));
            let request = http::Request::builder().uri("http://gunnar.invalid/gunnar.v1.Mirror/LogFetch").body(Body::empty()).expect("a request");
            service.call(request).await.expect("infallible");
        }
        let said = String::from_utf8(heard.0.lock().unwrap().clone()).unwrap();
        let growing: Vec<&str> = said.lines().filter(|l| l.contains("grpc_status=14")).collect();
        assert!(growing.len() == 2 && growing.iter().all(|l| l.contains(" INFO ")), "the growing box's deferral is INFO, in either words (a WARN per retried login trips the log gate's spam rule):\n{said}");
        let refused: Vec<&str> = said.lines().filter(|l| l.contains("grpc_status=7")).collect();
        assert!(refused.len() == 1 && refused[0].contains(" WARN "), "any other refusal stays WARN:\n{said}");
    }
}