mirror of
https://github.com/block/buzz.git
synced 2026-11-04 00:12:02 +00:00
fix(relay): restore http_method on HTTP access lines
The http.request span is absent under an access-only filter and when a pending request is dropped, so lines relying on it lost the verb. Keep the span at INFO and the verb on the line; also test the WS read counter through the dispatcher. Co-authored-by: Will Pfleger <pfleger.will@gmail.com> Signed-off-by: Will Pfleger <pfleger.will@gmail.com>
This commit is contained in:
Duncan
@@ -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
|
||||
|
||||
2
Justfile
2
Justfile
@@ -554,7 +554,7 @@ test-unit:
|
||||
# subscription::tests covers the in-memory registry and its fan-out
|
||||
# indexes; it is infra-free.
|
||||
cargo nextest run -p buzz-relay --lib --bin buzz-relay \
|
||||
-E 'test(/^subscription::tests::/) + test(/^api::admin::/) + test(/^access::tests::/) + test(/^api::git::transport::access_log_tests::/) + test(=api::bridge::postgres_tests::nip98_replay_check_logs_only_its_refusal) + test(=api::bridge::postgres_tests::warn_only_artifact_read_lines_keep_the_filter_summary) + test(=connection::tests::close_totals_count_frames_the_writer_flushed) + test(=connection::tests::startup_cancellation_writes_one_close_line) + test(=connection::tests::connection_limit_refusal_writes_one_close_line) + test(=connection::tests::dropped_authenticated_handler_writes_one_close_line) + test(=connection::tests::close_totals_count_a_batch_flushed_by_the_drain_after_cancellation) + test(=connection::tests::close_totals_skip_a_frame_the_socket_rejects) + test(=connection::tests::close_totals_skip_frames_stranded_by_a_failed_flush) + test(=connection::tests::read_before_login_logs_one_denial_from_the_handler) + test(=connection::tests::read_refused_by_a_full_handler_semaphore_logs_one_denial) + test(=connection::tests::read_refused_by_unavailable_admission_logs_one_denial) + test(=handlers::auth::tests::repeat_auth_after_a_terminal_logs_one_denial) + test(=nip_fi_upgrade::tests::deny_protected_logs_one_denied_auth_line) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::coalesced_refresh::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^nip11::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::profiles::/) + test(/^api::bridge::search_slots::tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^mesh_boot::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::accessible_channel_invalidation_is_scoped_to_community) + test(=state::tests::accessible_channels_join_member_and_open_halves) + test(=state::tests::channel_deleted_invalidation_is_scoped_to_community) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::failed_search_closes_without_eose_or_trying_later_filters) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + test(=api::invites::postgres_tests::restriction_tests::postgres_tests::destructive_fixture_rejects_shared_database_urls) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))'
|
||||
-E 'test(/^subscription::tests::/) + test(/^api::admin::/) + test(/^access::tests::/) + test(/^api::git::transport::access_log_tests::/) + test(=api::bridge::postgres_tests::nip98_replay_check_logs_only_its_refusal) + test(=api::bridge::postgres_tests::warn_only_artifact_read_lines_keep_the_filter_summary) + test(=connection::tests::close_totals_count_frames_the_writer_flushed) + test(=connection::tests::dispatcher_counts_each_req_and_count_frame_once) + test(=connection::tests::startup_cancellation_writes_one_close_line) + test(=connection::tests::connection_limit_refusal_writes_one_close_line) + test(=connection::tests::dropped_authenticated_handler_writes_one_close_line) + test(=connection::tests::close_totals_count_a_batch_flushed_by_the_drain_after_cancellation) + test(=connection::tests::close_totals_skip_a_frame_the_socket_rejects) + test(=connection::tests::close_totals_skip_frames_stranded_by_a_failed_flush) + test(=connection::tests::read_before_login_logs_one_denial_from_the_handler) + test(=connection::tests::read_refused_by_a_full_handler_semaphore_logs_one_denial) + test(=connection::tests::read_refused_by_unavailable_admission_logs_one_denial) + test(=handlers::auth::tests::repeat_auth_after_a_terminal_logs_one_denial) + test(=nip_fi_upgrade::tests::deny_protected_logs_one_denied_auth_line) + test(/^push_runtime::/) + test(/^operator_listener::tests::/) + test(/^handlers::channel_authz::/) + test(/^handlers::moderation_authz::/) + test(/^handlers::side_effects::tests::/) + test(/^handlers::coalesced_refresh::tests::/) + test(/^handlers::system_channels::tests::/) + test(/^nip11::tests::/) + test(/^storage_sweep::tests::/) + test(/^nip_fi_core::tests::/) + test(/^nip_fi_http::tests::/) + test(/^nip_fi_config::tests::/) + test(/^nip_fi_resolver::tests::/) + test(/^readiness::tests::/) + test(/^router::tests::/) + test(/^api::parse_query_tests::/) + test(/^api::profiles::/) + test(/^api::bridge::search_slots::tests::/) + test(/^api::git::transport::off_mode_precedence_tests::/) + test(/^audio::join::tests::/) + test(/^audio::handler::tests::/) + test(/^nip_fi_gate::tests::/) + test(/^nip_fi_session::tests::/) + test(/^nip_fi_shadow::tests::/) + test(/^nip_fi_shadow_session::tests::/) + test(/^startup_steps::tests::/) + test(/^mesh_boot::tests::/) + test(/^telemetry::tests::/) + test(/^config::tests::/) + test(=state::tests::accessible_channel_invalidation_is_scoped_to_community) + test(=state::tests::accessible_channels_join_member_and_open_halves) + test(=state::tests::channel_deleted_invalidation_is_scoped_to_community) + test(=state::tests::neither_a_confirmed_inactive_community_nor_a_failed_lookup_admits_the_socket) + test(=state::tests::periodic_revalidation_disconnects_inside_the_fenced_callback) + test(=state::tests::bare_community_disconnect_fails_closed_when_the_fence_is_unavailable) + test(=connection::tests::send_loop_sends_policy_close_when_community_is_archived) + test(=state::tests::community_disconnect_then_nip_fi_keeps_community_deleted_reason) + test(=state::tests::disconnect_community_wins_reason_losing_nip_fi_does_not_enqueue_frame) + test(=handlers::req::tests::timed_out_historical_read_deregisters_before_closed) + test(=handlers::req::tests::superseded_timeout_leaves_replacement_intact) + test(=handlers::req::tests::search_claim_retires_live_and_yields_to_replacement) + test(=handlers::req::tests::failed_search_closes_without_eose_or_trying_later_filters) + test(=handlers::req::tests::concurrent_claims_and_stale_teardowns_keep_the_last_owner) + test(=handlers::req::tests::timeout_closed_is_emitted_before_a_replacement_can_claim) + test(=handlers::req::tests::revoke_then_replacement_keeps_replacement_whole) + test(=handlers::req::tests::claims_after_connection_cleanup_are_refused) + test(=handlers::req::tests::dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::revoke_dropped_terminal_frame_cancels_connection) + test(=handlers::req::tests::count_fallback_covers_filters_that_can_match_private_deletions) + test(=handlers::req::tests::group_state_t_and_creator_tags_reach_sql_only_for_group_state_kinds) + test(=api::bridge::postgres_tests::bridge_search_mode_extension_accepts_prefix_snake_or_camel_case) + test(=api::bridge::postgres_tests::bridge_search_sort_extension_selects_newest_only_when_named) + test(=api::invites::postgres_tests::restriction_tests::postgres_tests::destructive_fixture_rejects_shared_database_urls) + (kind(bin) & (test(/^tests::/) + test(/^composition_tests::/) + test(/^env_filter_tests::/)) - test(/^tests::postgres_tests::/))'
|
||||
# Note on audio::join::tests scope: the full suite is infra-free (no DB,
|
||||
# no Redis). The infra-free audio/FI regression witnesses — bootstrap
|
||||
# ordering barrier, CommitConfirmed arm, pending-close invisibility,
|
||||
|
||||
@@ -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
|
||||
@@ -1116,6 +1115,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(),
|
||||
@@ -1130,6 +1130,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(),
|
||||
@@ -1169,6 +1170,7 @@ pub(crate) async fn http_access_log(
|
||||
let mut line = HttpLine {
|
||||
slot,
|
||||
route,
|
||||
http_method: req.method().clone(),
|
||||
peer_addr: req
|
||||
.extensions()
|
||||
.get::<axum::extract::ConnectInfo<SocketAddr>>()
|
||||
@@ -1194,6 +1196,7 @@ struct HttpLine {
|
||||
slot: AccessSlot,
|
||||
/// The matched template; `None` for a request that matched no route.
|
||||
route: Option<String>,
|
||||
http_method: axum::http::Method,
|
||||
peer_addr: Option<SocketAddr>,
|
||||
client_ip: Option<IpAddr>,
|
||||
started: std::time::Instant,
|
||||
@@ -1240,6 +1243,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),
|
||||
@@ -1358,6 +1362,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)]
|
||||
|
||||
@@ -6164,6 +6164,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(
|
||||
|
||||
@@ -2697,6 +2697,37 @@ pub(crate) mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
/// 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<WsMessage>,
|
||||
|
||||
@@ -546,10 +546,8 @@ fn http_trace_layer() -> TraceLayer<HttpMakeClassifier, fn(&Request<Body>) -> tr
|
||||
TraceLayer::new_for_http().make_span_with(make_http_span as fn(&Request<Body>) -> 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<Body>) -> tracing::Span {
|
||||
tracing::error_span!(
|
||||
tracing::info_span!(
|
||||
target: "buzz_relay",
|
||||
"http.request",
|
||||
otel.kind = "server",
|
||||
|
||||
Reference in New Issue
Block a user