diff --git a/.env.example b/.env.example index 1405b45193..5dc0a6f0f7 100644 --- a/.env.example +++ b/.env.example @@ -240,8 +240,7 @@ RUST_LOG=buzz_relay=debug,buzz_datastore=info,buzz_db=debug,buzz_auth=debug,buzz # `req`, `count`, or `close`. `http` is one line per HTTP request, written # by one layer when the response is ready (not proof the body was delivered): # `route` is the matched route template (never the raw path), with the response -# `status` and `time_to_response_ms` on every line; the method is the -# enclosing `http.request` span's `http.request.method`. It carries the login +# `status` and `time_to_response_ms` on every line. It carries the login # (`method`, `pubkey`, `pubkey_verified`) and `outcome` (`success`, `denied`, # `failed`, `challenged`, or `undecided` when the route recorded no decision; a # refusal is final). `/query` and `/count` add the filter summary and diff --git a/crates/buzz-relay/src/access.rs b/crates/buzz-relay/src/access.rs index c83f5b554c..1412785e9c 100644 --- a/crates/buzz-relay/src/access.rs +++ b/crates/buzz-relay/src/access.rs @@ -15,8 +15,7 @@ //! matched template, never the raw path. //! A successful bridge read that returned nothing (`result_count=0`) logs at //! DEBUG: polling clients make most of them. WebSocket `req` lines stay at -//! INFO, because an empty subscription can still deliver live events. The -//! request method is the enclosing span's `http.request.method`. +//! INFO, because an empty subscription can still deliver live events. //! //! With `BUZZ_CLIENT_IP_HEADER` set, `http`, `auth` and `close` lines also //! carry `client_ip`: the rightmost entry of that request header, only when it @@ -1087,6 +1086,7 @@ impl AccessSlot { drop(HttpLine { slot: self.clone(), route: Some(route.to_owned()), + http_method: axum::http::Method::POST, peer_addr: None, client_ip: None, started: std::time::Instant::now(), @@ -1101,6 +1101,7 @@ impl AccessSlot { drop(HttpLine { slot: self.clone(), route: Some(route.to_owned()), + http_method: axum::http::Method::GET, peer_addr: None, client_ip: None, started: std::time::Instant::now(), @@ -1140,6 +1141,7 @@ pub(crate) async fn http_access_log( let mut line = HttpLine { slot, route, + http_method: req.method().clone(), peer_addr: req .extensions() .get::>() @@ -1165,6 +1167,7 @@ struct HttpLine { slot: AccessSlot, /// The matched template; `None` for a request that matched no route. route: Option, + http_method: axum::http::Method, peer_addr: Option, client_ip: Option, started: std::time::Instant, @@ -1211,6 +1214,7 @@ impl Drop for HttpLine { level, event = HTTP_EVENT, route = self.route.as_deref().unwrap_or(UNMATCHED_ROUTE), + http_method = %self.http_method, peer_addr = self.peer_addr.map(tracing::field::display), client_ip = self.client_ip.map(tracing::field::display), method = note.method.map(AuthMethod::as_str), @@ -1329,6 +1333,17 @@ pub(crate) mod tests { (tracing::subscriber::set_default(json_subscriber(&out)), out) } + /// Like [`capture_scoped`], with only the access target on, at `level`: + /// the HTTP span (target `buzz_relay`) is filtered out. + pub(crate) fn capture_access_only( + level: Level, + ) -> (tracing::subscriber::DefaultGuard, Captured) { + let out = Captured::default(); + let subscriber = json_subscriber(&out) + .with(tracing_subscriber::filter::Targets::new().with_target(TARGET, level)); + (tracing::subscriber::set_default(subscriber), out) + } + /// A replay guard whose check never finishes, so a request parks in its /// replay wait; `arrived` fires once a request reaches it. #[derive(Default)] diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 669b319037..6701864cb1 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -6145,6 +6145,111 @@ pub(crate) mod postgres_tests { assert_eq!(line["authors"], "*", "an open dimension is `*`: {line}"); } + /// Every HTTP line names its verb itself, since the `http.request` span + /// is absent under an access-only filter and when a request is dropped + /// (Tower enters the span only while polling). + #[tokio::test(flavor = "current_thread")] + #[ignore = "requires Postgres"] + async fn http_lines_carry_the_verb_without_the_span() { + use crate::access::tests::{assert_one_aborted_line, capture_access_only}; + use tracing::Level; + let state = nip_fi_off_test_state().await.expect("local Postgres"); + let host = with_community(&state).await; + let community = crate::tenant::bind_community(&state.db, &host) + .await + .expect("bind community") + .community(); + let keys = Keys::generate(); + + // A served, non-empty read under an access-only INFO filter. + let event = EventBuilder::new(Kind::TextNote, "") + .sign_with_keys(&keys) + .expect("sign event"); + state + .db + .insert_event(community, &event, None) + .await + .expect("insert event"); + let body = format!( + r#"[{{"kinds":[1],"authors":["{}"]}}]"#, + keys.public_key().to_hex() + ); + let headers = make_nip98_headers( + &keys, + &format!("https://{host}/query"), + "POST", + body.as_bytes(), + ); + let (guard, out) = capture_access_only(Level::INFO); + let status = oneshot_request( + Arc::clone(&state), + "POST", + "/query", + &host, + headers, + body.as_bytes(), + ) + .await; + drop(guard); + assert_eq!(status, axum::http::StatusCode::OK); + let lines = out.lines(); + assert_eq!(lines.len(), 1, "exactly one access line: {lines:?}"); + let line = &lines[0]; + assert_eq!( + (&line["level"], &line["result_count"]), + (&"INFO".into(), &1.into()), + "{line}" + ); + assert_eq!(line["http_method"], "POST", "{line}"); + assert_eq!(line["pubkey"], keys.public_key().to_hex(), "{line}"); + assert!(!out.raw().contains("http.request"), "{}", out.raw()); + + // A refusal under an access-only WARN filter. + let mut members_only = (*state).clone(); + Arc::make_mut(&mut members_only.config).require_relay_membership = true; + let outsider = Keys::generate(); + let headers = make_nip98_headers( + &outsider, + &format!("https://{host}/query"), + "POST", + QUERY_BODY, + ); + let (guard, out) = capture_access_only(Level::WARN); + let status = oneshot_request( + Arc::new(members_only), + "POST", + "/query", + &host, + headers, + QUERY_BODY, + ) + .await; + drop(guard); + assert_eq!(status, axum::http::StatusCode::FORBIDDEN); + let line = assert_one_auth_line(&out.lines(), "denied", Some("not_relay_member")); + assert_eq!(line["level"], "WARN", "{line}"); + assert_eq!(line["http_method"], "POST", "{line}"); + assert_eq!(line["pubkey"], outsider.public_key().to_hex(), "{line}"); + + // A request dropped mid-replay, with every target on. + let mut stalled = (*state).clone(); + let replay = Arc::new(crate::access::tests::StallingReplayGuard::default()); + stalled.nip98_replay = replay.clone(); + let headers = + make_nip98_headers(&keys, &format!("https://{host}/query"), "POST", QUERY_BODY); + let (guard, out) = crate::access::tests::capture_scoped(); + tokio::select! { + _ = oneshot_request(Arc::new(stalled), "POST", "/query", &host, headers, QUERY_BODY) => { + panic!("the replay check must stall"); + } + () = replay.arrived.notified() => {} + } + drop(guard); + let line = assert_one_aborted_line(&out, &keys.public_key()); + assert_eq!(line["http_method"], "POST", "{line}"); + assert_eq!(line["route"], "/query", "{line}"); + } + /// POST `body` to `route` signed by `keys` for `signed_url`, and return /// the status and every access line the request wrote. async fn post_access_lines( diff --git a/crates/buzz-relay/src/connection.rs b/crates/buzz-relay/src/connection.rs index 7ede782b1f..793d10ee74 100644 --- a/crates/buzz-relay/src/connection.rs +++ b/crates/buzz-relay/src/connection.rs @@ -2400,6 +2400,37 @@ pub(crate) mod tests { assert_eq!(frame[2], "rate-limited: too many concurrent requests"); } + /// Each parsed REQ and COUNT frame advances its read counter by exactly + /// one, even when refused; other frames count nothing. + #[tokio::test(flavor = "current_thread")] + async fn dispatcher_counts_each_req_and_count_frame_once() { + let (recorder, handle) = crate::metrics::readiness_test_recorder(); + let _recorder_guard = metrics::set_default_local_recorder(&recorder); + let state = crate::state::tests::test_state().await; + let (conn, _rx) = test_conn_with_auth(AuthState::Failed); + let permits = state.handler_semaphore.available_permits(); + let _held = Arc::clone(&state.handler_semaphore) + .acquire_many_owned(permits as u32) + .await + .expect("hold every handler permit"); + + for raw in [ + serde_json::json!(["REQ", "r", {"kinds": [1]}]), + serde_json::json!(["COUNT", "c", {"kinds": [1]}]), + serde_json::json!(["CLOSE", "r"]), + ] { + handle_text_message(raw.to_string(), Arc::clone(&conn), Arc::clone(&state)).await; + } + + let scrape = handle.render(); + for kind in ["req", "count"] { + assert!( + scrape.contains(&format!("buzz_ws_reads_total{{type=\"{kind}\"}} 1\n")), + "{scrape}" + ); + } + } + #[derive(Debug, Default)] struct MockSinkState { messages: Vec, diff --git a/crates/buzz-relay/src/router.rs b/crates/buzz-relay/src/router.rs index e2bfd8f9ef..89181b4ca8 100644 --- a/crates/buzz-relay/src/router.rs +++ b/crates/buzz-relay/src/router.rs @@ -545,10 +545,8 @@ fn http_trace_layer() -> TraceLayer) -> tr TraceLayer::new_for_http().make_span_with(make_http_span as fn(&Request) -> tracing::Span) } -/// ERROR level, so the span is on whenever an access line is: those lines -/// take the request method from its `http.request.method`. fn make_http_span(request: &Request) -> tracing::Span { - tracing::error_span!( + tracing::info_span!( target: "buzz_relay", "http.request", otel.kind = "server",