From a932c49ca8a37f218093f89728e8920460099235 Mon Sep 17 00:00:00 2001
From: tipogi <tipogi@protonmail.com>
Date: Thu, 20 Aug 2026 07:34:12 +0200
Subject: [PATCH 1/5] feat: add HTTP RED metrics and bound label cardinality in
 webapi

---
 .../src/routes/middlewares/tracing.rs         | 202 ++++++++++++++++--
 nexus-webapi/src/routes/mod.rs                |   2 +
 2 files changed, 188 insertions(+), 16 deletions(-)

diff --git a/nexus-webapi/src/routes/middlewares/tracing.rs b/nexus-webapi/src/routes/middlewares/tracing.rs
index 918181751..7e00feed4 100644
--- a/nexus-webapi/src/routes/middlewares/tracing.rs
+++ b/nexus-webapi/src/routes/middlewares/tracing.rs
@@ -1,41 +1,211 @@
+//! HTTP tracing and metrics middleware.
+//!
+//! Incoming W3C `traceparent` is intentionally ignored. This is a public
+//! unauthenticated API, so clients must not inject trace context; each request
+//! starts a new root span via the tracing subscriber.
+
+use std::{sync::LazyLock, time::Instant};
+
 use axum::{
     extract::{MatchedPath, Request},
+    http::Method,
     middleware::Next,
     response::Response,
 };
-
+use opentelemetry::metrics::{Counter, Histogram};
+use opentelemetry::{global, KeyValue};
 use tracing::Instrument;
 
-// middleware for tracing
+const METER_NAME: &str = "nexus";
+
+/// HTTP request instruments. No-ops until an `SdkMeterProvider` is installed.
+///
+/// `requests` duplicates the histogram's count (same attributes, same call
+/// site) but is kept for cheaper PromQL. `errors` is the availability SLI
+/// (see [`is_failed_request`]). `duration` runs until the handler returns a
+/// `Response` — headers ready, not end-of-body.
+struct HttpMetrics {
+    requests: Counter<u64>,
+    errors: Counter<u64>,
+    duration: Histogram<f64>,
+}
+
+impl HttpMetrics {
+    fn new() -> Self {
+        let meter = global::meter(METER_NAME);
+        Self {
+            requests: meter
+                .u64_counter("http.server.requests")
+                .with_description(
+                    "Total HTTP requests handled by Nexus, by method/route/status",
+                )
+                .build(),
+            errors: meter
+                .u64_counter("http.server.errors")
+                .with_description(
+                    "HTTP requests that failed from the server's perspective (5xx and 408 timeouts), by method/route/status",
+                )
+                .build(),
+            duration: meter
+                .f64_histogram("http.server.request.duration")
+                .with_description("Duration of HTTP server requests")
+                .with_unit("s")
+                // Explicit buckets from the HTTP semantic conventions (seconds).
+                .with_boundaries(vec![
+                    0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1.0, 2.5, 5.0, 7.5,
+                    10.0,
+                ])
+                .build(),
+        }
+    }
+}
+
+/// Lazy on purpose: instruments bind to whatever meter provider is global at
+/// creation time, and stay no-ops forever if that is still the default one.
+/// First use is the first HTTP request, safely after `setup_metrics`.
+static METRICS: LazyLock<HttpMetrics> = LazyLock::new(HttpMetrics::new);
+
+/// Availability failure: 5xx, plus 408 because it times out *our* work.
+///
+/// Other 4xx (bad input, missing entities, blacklist 403, oversized 413,
+/// rate-limit 429) are ordinary on a public unauthenticated API. They stay on
+/// `http.server.requests` via `http.response.status_code` without polluting
+/// the error rate — same policy as `Error` logging in `error.rs`.
+fn is_failed_request(status: u16) -> bool {
+    matches!(status, 408 | 500..=599)
+}
+
+/// Low-cardinality route template, or `unmatched`. Never the raw URI:
+/// `/v0/user/<pubkey>` would explode series cardinality on scanner 404s.
+///
+/// `MatchedPath` is only set for middleware on the router that declared the
+/// routes (see `build_app`). A nested copy of this layer would see `unmatched`.
+fn http_route(request: &Request) -> String {
+    request
+        .extensions()
+        .get::<MatchedPath>()
+        .map(|pattern| pattern.as_str().to_string())
+        .unwrap_or_else(|| "unmatched".to_string())
+}
+
+/// Known methods per OTel semconv, `_OTHER` for the rest. Clients can send
+/// arbitrary extension methods, which would otherwise mint unbounded metric
+/// series. The raw value goes on the span as `http.request.method_original`.
+fn http_method(method: &Method) -> &'static str {
+    match *method {
+        Method::GET => "GET",
+        Method::HEAD => "HEAD",
+        Method::POST => "POST",
+        Method::PUT => "PUT",
+        Method::DELETE => "DELETE",
+        Method::CONNECT => "CONNECT",
+        Method::OPTIONS => "OPTIONS",
+        Method::TRACE => "TRACE",
+        Method::PATCH => "PATCH",
+        _ => "_OTHER",
+    }
+}
+
+fn record_http_request(method: &str, route: &str, status: u16, elapsed_secs: f64) {
+    let attrs = [
+        KeyValue::new("http.request.method", method.to_string()),
+        KeyValue::new("http.route", route.to_string()),
+        KeyValue::new("http.response.status_code", i64::from(status)),
+    ];
+    METRICS.requests.add(1, &attrs);
+    METRICS.duration.record(elapsed_secs, &attrs);
+    if is_failed_request(status) {
+        METRICS.errors.add(1, &attrs);
+    }
+}
+
+/// Root span plus request/error/duration metrics for every request.
 pub async fn tracing_middleware(request: Request, next: Next) -> Response {
-    let route = request.uri().path().to_string();
-    let route_pattern = request.extensions().get::<MatchedPath>();
-    let span_name = match route_pattern {
-        Some(pattern) => pattern.as_str().to_string(),
-        _ => route.clone(),
-    };
-    let query = request.uri().query().unwrap_or("").to_string();
-    let method = request.method().to_string();
+    let route = http_route(&request);
+    let method = http_method(request.method());
 
+    // Query strings are deliberately not recorded: they carry search text,
+    // pubkeys, and filter values, with no cardinality bound in Tempo.
     let span = tracing::info_span!(
         "http.request",
-        otel.name = %span_name,
+        // Semconv wants a bare `{method}` when no route matched; we keep
+        // `unmatched` so scanner 404s stay searchable in Tempo.
+        otel.name = %format!("{method} {route}"),
         http.request.method = %method,
+        http.request.method_original = tracing::field::Empty,
         http.route = %route,
-        http.query = %query,
         http.response.status_code = tracing::field::Empty,
         otel.status_code = tracing::field::Empty,
         otel.status_message = tracing::field::Empty,
     );
+    if method == "_OTHER" {
+        span.record(
+            "http.request.method_original",
+            request.method().to_string(),
+        );
+    }
 
+    let started = Instant::now();
     let response = next.run(request).instrument(span.clone()).await;
 
-    let status = response.status().as_u16();
-    span.record("http.response.status_code", status);
-    if (500..=599).contains(&status) {
+    let status = response.status();
+    span.record("http.response.status_code", status.as_u16());
+    if is_failed_request(status.as_u16()) {
         span.record("otel.status_code", "ERROR");
-        span.record("otel.status_message", "Internal Server Error");
+        span.record(
+            "otel.status_message",
+            status.canonical_reason().unwrap_or("error"),
+        );
     }
 
+    record_http_request(
+        method,
+        &route,
+        status.as_u16(),
+        started.elapsed().as_secs_f64(),
+    );
+
     response
 }
+
+#[cfg(test)]
+mod tests {
+    use super::{http_method, http_route, is_failed_request, Method, Request};
+    use axum::body::Body;
+
+    // The matched case can't be unit-tested: `MatchedPath` has no public
+    // constructor and only exists inside a real router.
+    #[test]
+    fn route_without_matched_path_is_unmatched() {
+        let request = Request::builder()
+            .uri("/wp-admin")
+            .body(Body::empty())
+            .unwrap();
+        assert_eq!(http_route(&request), "unmatched");
+    }
+
+    #[test]
+    fn extension_methods_collapse_to_other() {
+        assert_eq!(http_method(&Method::GET), "GET");
+        assert_eq!(http_method(&Method::PATCH), "PATCH");
+
+        let custom = Method::from_bytes(b"PROPFIND").unwrap();
+        assert_eq!(http_method(&custom), "_OTHER");
+    }
+
+    #[test]
+    fn failed_request_is_5xx_and_timeout_only() {
+        assert!(is_failed_request(500));
+        assert!(is_failed_request(502));
+        assert!(is_failed_request(503));
+        assert!(is_failed_request(408));
+
+        assert!(!is_failed_request(200));
+        assert!(!is_failed_request(204));
+        assert!(!is_failed_request(400));
+        assert!(!is_failed_request(403));
+        assert!(!is_failed_request(404));
+        assert!(!is_failed_request(413));
+        assert!(!is_failed_request(429));
+    }
+}
diff --git a/nexus-webapi/src/routes/mod.rs b/nexus-webapi/src/routes/mod.rs
index ba2524c0f..476d3b540 100644
--- a/nexus-webapi/src/routes/mod.rs
+++ b/nexus-webapi/src/routes/mod.rs
@@ -168,6 +168,8 @@ pub fn build_app(
             Duration::from_secs(request_timeout_secs.max(1)),
         ))
         .layer(cors)
+        // Must stay on this outer router: `MatchedPath` is only populated for
+        // middleware layered here, not if a nested sub-router copies this layer.
         .layer(axum::middleware::from_fn(
             middlewares::tracing::tracing_middleware,
         ))

From a36df273d4fb5c008d5980411f179a41bfcc4271 Mon Sep 17 00:00:00 2001
From: tipogi <tipogi@protonmail.com>
Date: Thu, 20 Aug 2026 08:22:23 +0200
Subject: [PATCH 2/5] feat: add neo4j.query.requests counter for a first-class
 query total

---
 nexus-common/src/db/graph/instrumented.rs      |  9 +++++++++
 nexus-webapi/src/routes/middlewares/tracing.rs | 13 +++----------
 2 files changed, 12 insertions(+), 10 deletions(-)

diff --git a/nexus-common/src/db/graph/instrumented.rs b/nexus-common/src/db/graph/instrumented.rs
index c3e409df1..ba17c44db 100644
--- a/nexus-common/src/db/graph/instrumented.rs
+++ b/nexus-common/src/db/graph/instrumented.rs
@@ -33,6 +33,8 @@ struct GraphMetrics {
     execute_duration: Histogram<f64>,
     /// Number of rows returned by a query.
     rows: Histogram<u64>,
+    /// Query completions (stream drop, execute failure, or run); error-rate denominator.
+    requests: Counter<u64>,
     /// Incremented on every failed `execute()` or `run()` call.
     errors: Counter<u64>,
     /// Incremented when a query's total duration exceeds the slow-query threshold.
@@ -73,6 +75,10 @@ impl GraphMetrics {
                 .with_description("Number of rows returned per Neo4j query")
                 .with_unit("{row}")
                 .build(),
+            requests: meter
+                .u64_counter("neo4j.query.requests")
+                .with_description("Total Neo4j query executions, by query label")
+                .build(),
             errors: meter
                 .u64_counter("neo4j.query.errors")
                 .with_description("Total number of failed Neo4j query executions")
@@ -155,6 +161,7 @@ impl Drop for InstrumentedStream {
         let attrs: &[KeyValue] = &query_attrs(self.label);
 
         // Always record metrics (no-op when OTLP is not configured).
+        self.metrics.requests.add(1, attrs);
         self.metrics.duration.record(ms(total), attrs);
         self.metrics
             .execute_duration
@@ -302,6 +309,7 @@ impl<G: GraphOps> GraphOps for InstrumentedGraph<G> {
             }
             Err(e) => {
                 let attrs: &[KeyValue] = &query_attrs(label);
+                self.metrics.requests.add(1, attrs);
                 self.metrics.duration.record(ms(execute_duration), attrs);
                 self.metrics
                     .execute_duration
@@ -339,6 +347,7 @@ impl<G: GraphOps> GraphOps for InstrumentedGraph<G> {
         let elapsed = start.elapsed();
 
         let attrs: &[KeyValue] = &query_attrs(label);
+        self.metrics.requests.add(1, attrs);
         self.metrics.duration.record(ms(elapsed), attrs);
         self.metrics.execute_duration.record(ms(elapsed), attrs);
 
diff --git a/nexus-webapi/src/routes/middlewares/tracing.rs b/nexus-webapi/src/routes/middlewares/tracing.rs
index 7e00feed4..0da8a9359 100644
--- a/nexus-webapi/src/routes/middlewares/tracing.rs
+++ b/nexus-webapi/src/routes/middlewares/tracing.rs
@@ -18,12 +18,8 @@ use tracing::Instrument;
 
 const METER_NAME: &str = "nexus";
 
-/// HTTP request instruments. No-ops until an `SdkMeterProvider` is installed.
-///
-/// `requests` duplicates the histogram's count (same attributes, same call
-/// site) but is kept for cheaper PromQL. `errors` is the availability SLI
-/// (see [`is_failed_request`]). `duration` runs until the handler returns a
-/// `Response` — headers ready, not end-of-body.
+/// Request total, availability failures, and time until response headers are ready.
+/// Instruments are no-ops until an `SdkMeterProvider` is installed.
 struct HttpMetrics {
     requests: Counter<u64>,
     errors: Counter<u64>,
@@ -139,10 +135,7 @@ pub async fn tracing_middleware(request: Request, next: Next) -> Response {
         otel.status_message = tracing::field::Empty,
     );
     if method == "_OTHER" {
-        span.record(
-            "http.request.method_original",
-            request.method().to_string(),
-        );
+        span.record("http.request.method_original", request.method().to_string());
     }
 
     let started = Instant::now();

From 4c48f15550202ec67f94667a21e3d91346cf5764 Mon Sep 17 00:00:00 2001
From: tipogi <tipogi@protonmail.com>
Date: Thu, 20 Aug 2026 08:59:42 +0200
Subject: [PATCH 3/5] feat: split reach-scoped Neo4j query labels by reach and
 WoT depth

---
 nexus-common/src/db/graph/instrumented.rs     |  2 +-
 nexus-common/src/db/graph/queries/get.rs      | 78 ++++++++++++++++---
 .../src/routes/middlewares/tracing.rs         | 44 ++++-------
 nexus-webapi/src/routes/mod.rs                |  3 +-
 4 files changed, 81 insertions(+), 46 deletions(-)

diff --git a/nexus-common/src/db/graph/instrumented.rs b/nexus-common/src/db/graph/instrumented.rs
index ba17c44db..bfca91c1f 100644
--- a/nexus-common/src/db/graph/instrumented.rs
+++ b/nexus-common/src/db/graph/instrumented.rs
@@ -33,7 +33,7 @@ struct GraphMetrics {
     execute_duration: Histogram<f64>,
     /// Number of rows returned by a query.
     rows: Histogram<u64>,
-    /// Query completions (stream drop, execute failure, or run); error-rate denominator.
+    /// Total queries (error-rate denominator). Same sites as `duration`.
     requests: Counter<u64>,
     /// Incremented on every failed `execute()` or `run()` call.
     errors: Counter<u64>,
diff --git a/nexus-common/src/db/graph/queries/get.rs b/nexus-common/src/db/graph/queries/get.rs
index 8c3a03624..a9f665412 100644
--- a/nexus-common/src/db/graph/queries/get.rs
+++ b/nexus-common/src/db/graph/queries/get.rs
@@ -758,6 +758,25 @@ pub fn get_tags() -> Query {
     )
 }
 
+/// Metric label `{base}_{reach}` / `{base}_wot_{depth}`. Depth lives here
+/// (unlike `post_stream_wot`, these queries have no companion meter). A new
+/// WoT depth must get its own arm — never reuse an existing label.
+macro_rules! reach_query_label {
+    ($base:literal, $reach:expr) => {
+        match $reach {
+            StreamReach::Followers => concat!($base, "_followers"),
+            StreamReach::Following => concat!($base, "_following"),
+            StreamReach::Friends => concat!($base, "_friends"),
+            StreamReach::Wot(depth) => match depth.get() {
+                1 => concat!($base, "_wot_1"),
+                2 => concat!($base, "_wot_2"),
+                3 => concat!($base, "_wot_3"),
+                other => unreachable!("WotDepth out of range: {other}"),
+            },
+        }
+    };
+}
+
 pub fn get_tag_taggers_by_reach(
     label: &str,
     user_id: &str,
@@ -785,11 +804,14 @@ pub fn get_tag_taggers_by_reach(
             ",
         stream_reach_to_graph_subquery(&reach)
     );
-    Query::new("get_tag_taggers_by_reach", &cypher)
-        .param("label", label)
-        .param("user_id", user_id)
-        .param("skip", skip as i64)
-        .param("limit", limit as i64)
+    Query::new(
+        reach_query_label!("get_tag_taggers_by_reach", reach),
+        &cypher,
+    )
+    .param("label", label)
+    .param("user_id", user_id)
+    .param("skip", skip as i64)
+    .param("limit", limit as i64)
 }
 
 pub fn get_hot_tags_by_reach(
@@ -827,7 +849,7 @@ pub fn get_hot_tags_by_reach(
         input_tagged_type,
         tags_query.taggers_limit
     );
-    Query::new("get_hot_tags_by_reach", &cypher)
+    Query::new(reach_query_label!("get_hot_tags_by_reach", reach), &cypher)
         .param("user_id", user_id)
         .param("skip", tags_query.skip as i64)
         .param("limit", tags_query.limit as i64)
@@ -911,12 +933,15 @@ pub fn get_influencers_by_reach(
     ",
         stream_reach_to_graph_subquery(&reach),
     );
-    Query::new("get_influencers_by_reach", &cypher)
-        .param("user_id", user_id)
-        .param("skip", skip as i64)
-        .param("limit", limit as i64)
-        .param("from", from)
-        .param("to", to)
+    Query::new(
+        reach_query_label!("get_influencers_by_reach", reach),
+        &cypher,
+    )
+    .param("user_id", user_id)
+    .param("skip", skip as i64)
+    .param("limit", limit as i64)
+    .param("from", from)
+    .param("to", to)
 }
 
 pub fn get_global_influencers(skip: usize, limit: usize, timeframe: &Timeframe) -> Query {
@@ -1480,6 +1505,35 @@ mod tests {
         .to_cypher_populated()
     }
 
+    #[test]
+    fn reach_queries_split_label_by_reach_and_depth() {
+        let cases = [
+            (StreamReach::Followers, "get_influencers_by_reach_followers"),
+            (StreamReach::Following, "get_influencers_by_reach_following"),
+            (StreamReach::Friends, "get_influencers_by_reach_friends"),
+            (
+                StreamReach::Wot(WotDepth::new(1).unwrap()),
+                "get_influencers_by_reach_wot_1",
+            ),
+            (
+                StreamReach::Wot(WotDepth::new(3).unwrap()),
+                "get_influencers_by_reach_wot_3",
+            ),
+        ];
+        for (reach, expected) in cases {
+            let query = get_influencers_by_reach("user", reach, 0, 10, &Timeframe::AllTime);
+            assert_eq!(query.label(), Some(expected));
+        }
+
+        let taggers =
+            get_tag_taggers_by_reach("tag", "user", StreamReach::Wot(WotDepth::default()), 0, 10);
+        assert_eq!(taggers.label(), Some("get_tag_taggers_by_reach_wot_2"));
+
+        let hot_tags_input = HotTagsInputDTO::new(Timeframe::AllTime, 10, 0, 5, None);
+        let hot_tags = get_hot_tags_by_reach("user", StreamReach::Friends, &hot_tags_input);
+        assert_eq!(hot_tags.label(), Some("get_hot_tags_by_reach_friends"));
+    }
+
     /// The trust traversal must bind and dedupe authors before the posts MATCH.
     /// A CALL subquery runs once per incoming row (the planner cannot hoist it)
     /// and a variable-length traversal yields one row per path, so either one
diff --git a/nexus-webapi/src/routes/middlewares/tracing.rs b/nexus-webapi/src/routes/middlewares/tracing.rs
index 0da8a9359..e81a13443 100644
--- a/nexus-webapi/src/routes/middlewares/tracing.rs
+++ b/nexus-webapi/src/routes/middlewares/tracing.rs
@@ -56,26 +56,16 @@ impl HttpMetrics {
     }
 }
 
-/// Lazy on purpose: instruments bind to whatever meter provider is global at
-/// creation time, and stay no-ops forever if that is still the default one.
-/// First use is the first HTTP request, safely after `setup_metrics`.
+/// Bound after `setup_metrics`; no-ops if OTLP was never configured.
 static METRICS: LazyLock<HttpMetrics> = LazyLock::new(HttpMetrics::new);
 
-/// Availability failure: 5xx, plus 408 because it times out *our* work.
-///
-/// Other 4xx (bad input, missing entities, blacklist 403, oversized 413,
-/// rate-limit 429) are ordinary on a public unauthenticated API. They stay on
-/// `http.server.requests` via `http.response.status_code` without polluting
-/// the error rate — same policy as `Error` logging in `error.rs`.
+/// 5xx, plus 408 (our timeout). Other 4xx stay on `requests` by status code.
 fn is_failed_request(status: u16) -> bool {
     matches!(status, 408 | 500..=599)
 }
 
-/// Low-cardinality route template, or `unmatched`. Never the raw URI:
-/// `/v0/user/<pubkey>` would explode series cardinality on scanner 404s.
-///
-/// `MatchedPath` is only set for middleware on the router that declared the
-/// routes (see `build_app`). A nested copy of this layer would see `unmatched`.
+/// Matched route template, or `unmatched`. Never the raw URI (cardinality).
+/// Only set when this middleware sits on the router that declared the routes.
 fn http_route(request: &Request) -> String {
     request
         .extensions()
@@ -84,9 +74,7 @@ fn http_route(request: &Request) -> String {
         .unwrap_or_else(|| "unmatched".to_string())
 }
 
-/// Known methods per OTel semconv, `_OTHER` for the rest. Clients can send
-/// arbitrary extension methods, which would otherwise mint unbounded metric
-/// series. The raw value goes on the span as `http.request.method_original`.
+/// Semconv well-known methods; anything else is `_OTHER` (raw value on the span).
 fn http_method(method: &Method) -> &'static str {
     match *method {
         Method::GET => "GET",
@@ -115,17 +103,15 @@ fn record_http_request(method: &str, route: &str, status: u16, elapsed_secs: f64
     }
 }
 
-/// Root span plus request/error/duration metrics for every request.
 pub async fn tracing_middleware(request: Request, next: Next) -> Response {
     let route = http_route(&request);
     let method = http_method(request.method());
 
-    // Query strings are deliberately not recorded: they carry search text,
-    // pubkeys, and filter values, with no cardinality bound in Tempo.
+    // No query string: search text / pubkeys / filters, unbounded in Tempo.
     let span = tracing::info_span!(
         "http.request",
-        // Semconv wants a bare `{method}` when no route matched; we keep
-        // `unmatched` so scanner 404s stay searchable in Tempo.
+        // Semconv would use a bare `{method}` when unmatched; keep the token
+        // so scanner 404s stay searchable.
         otel.name = %format!("{method} {route}"),
         http.request.method = %method,
         http.request.method_original = tracing::field::Empty,
@@ -135,15 +121,16 @@ pub async fn tracing_middleware(request: Request, next: Next) -> Response {
         otel.status_message = tracing::field::Empty,
     );
     if method == "_OTHER" {
-        span.record("http.request.method_original", request.method().to_string());
+        span.record("http.request.method_original", request.method().as_str());
     }
 
     let started = Instant::now();
     let response = next.run(request).instrument(span.clone()).await;
 
     let status = response.status();
-    span.record("http.response.status_code", status.as_u16());
-    if is_failed_request(status.as_u16()) {
+    let status_code = status.as_u16();
+    span.record("http.response.status_code", status_code);
+    if is_failed_request(status_code) {
         span.record("otel.status_code", "ERROR");
         span.record(
             "otel.status_message",
@@ -151,12 +138,7 @@ pub async fn tracing_middleware(request: Request, next: Next) -> Response {
         );
     }
 
-    record_http_request(
-        method,
-        &route,
-        status.as_u16(),
-        started.elapsed().as_secs_f64(),
-    );
+    record_http_request(method, &route, status_code, started.elapsed().as_secs_f64());
 
     response
 }
diff --git a/nexus-webapi/src/routes/mod.rs b/nexus-webapi/src/routes/mod.rs
index 476d3b540..cdf723ef6 100644
--- a/nexus-webapi/src/routes/mod.rs
+++ b/nexus-webapi/src/routes/mod.rs
@@ -168,8 +168,7 @@ pub fn build_app(
             Duration::from_secs(request_timeout_secs.max(1)),
         ))
         .layer(cors)
-        // Must stay on this outer router: `MatchedPath` is only populated for
-        // middleware layered here, not if a nested sub-router copies this layer.
+        // Outer router only: a nested copy of this layer would see `unmatched`.
         .layer(axum::middleware::from_fn(
             middlewares::tracing::tracing_middleware,
         ))

From 74717389705ab964ae89487efaf437a3e9e1404e Mon Sep 17 00:00:00 2001
From: tipogi <tipogi@protonmail.com>
Date: Fri, 21 Aug 2026 09:37:04 +0200
Subject: [PATCH 4/5] chore: drop http.server.errors; status on requests is
 enough

---
 .../src/routes/middlewares/tracing.rs         | 21 ++++---------------
 1 file changed, 4 insertions(+), 17 deletions(-)

diff --git a/nexus-webapi/src/routes/middlewares/tracing.rs b/nexus-webapi/src/routes/middlewares/tracing.rs
index e81a13443..6842ef062 100644
--- a/nexus-webapi/src/routes/middlewares/tracing.rs
+++ b/nexus-webapi/src/routes/middlewares/tracing.rs
@@ -18,11 +18,10 @@ use tracing::Instrument;
 
 const METER_NAME: &str = "nexus";
 
-/// Request total, availability failures, and time until response headers are ready.
+/// Request count and time until response headers are ready.
 /// Instruments are no-ops until an `SdkMeterProvider` is installed.
 struct HttpMetrics {
     requests: Counter<u64>,
-    errors: Counter<u64>,
     duration: Histogram<f64>,
 }
 
@@ -32,15 +31,7 @@ impl HttpMetrics {
         Self {
             requests: meter
                 .u64_counter("http.server.requests")
-                .with_description(
-                    "Total HTTP requests handled by Nexus, by method/route/status",
-                )
-                .build(),
-            errors: meter
-                .u64_counter("http.server.errors")
-                .with_description(
-                    "HTTP requests that failed from the server's perspective (5xx and 408 timeouts), by method/route/status",
-                )
+                .with_description("Total HTTP requests handled by Nexus, by method/route/status")
                 .build(),
             duration: meter
                 .f64_histogram("http.server.request.duration")
@@ -48,8 +39,7 @@ impl HttpMetrics {
                 .with_unit("s")
                 // Explicit buckets from the HTTP semantic conventions (seconds).
                 .with_boundaries(vec![
-                    0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1.0, 2.5, 5.0, 7.5,
-                    10.0,
+                    0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1.0, 2.5, 5.0, 7.5, 10.0,
                 ])
                 .build(),
         }
@@ -59,7 +49,7 @@ impl HttpMetrics {
 /// Bound after `setup_metrics`; no-ops if OTLP was never configured.
 static METRICS: LazyLock<HttpMetrics> = LazyLock::new(HttpMetrics::new);
 
-/// 5xx, plus 408 (our timeout). Other 4xx stay on `requests` by status code.
+/// 5xx, plus 408 (our timeout). Used to mark the span ERROR; metrics use status labels.
 fn is_failed_request(status: u16) -> bool {
     matches!(status, 408 | 500..=599)
 }
@@ -98,9 +88,6 @@ fn record_http_request(method: &str, route: &str, status: u16, elapsed_secs: f64
     ];
     METRICS.requests.add(1, &attrs);
     METRICS.duration.record(elapsed_secs, &attrs);
-    if is_failed_request(status) {
-        METRICS.errors.add(1, &attrs);
-    }
 }
 
 pub async fn tracing_middleware(request: Request, next: Next) -> Response {

From e435f2e7f904cf45fca840f2bd5af26b51d63fa9 Mon Sep 17 00:00:00 2001
From: tipogi <tipogi@protonmail.com>
Date: Tue, 25 Aug 2026 13:01:36 +0200
Subject: [PATCH 5/5] refactor: replace encoded Neo4j query labels with
 structured telemetry attributes

---
 Cargo.lock                                |   2 +-
 nexus-common/src/db/graph/instrumented.rs | 190 +++++++++++++------
 nexus-common/src/db/graph/queries/get.rs  | 214 +++++++++++++++-------
 nexus-common/src/db/graph/query.rs        | 142 +++++++++++---
 nexus-common/src/models/post/stream.rs    |  23 +++
 nexus-common/src/types/mod.rs             |  12 ++
 6 files changed, 432 insertions(+), 151 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index 26becff36..f59c1a034 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3064,7 +3064,7 @@ dependencies = [
  "redis",
  "serde",
  "tempfile",
- "thiserror 2.0.18",
+ "thiserror 2.0.20",
  "tokio",
  "tokio-shared-rt",
  "tracing",
diff --git a/nexus-common/src/db/graph/instrumented.rs b/nexus-common/src/db/graph/instrumented.rs
index bfca91c1f..8e2125cf9 100644
--- a/nexus-common/src/db/graph/instrumented.rs
+++ b/nexus-common/src/db/graph/instrumented.rs
@@ -11,7 +11,7 @@ use std::time::{Duration, Instant};
 use tracing::warn;
 
 use super::ops::{Graph, GraphOps};
-use super::query::Query;
+use super::query::{Query, TelemetryValue};
 use crate::utils::ms;
 
 /// The OpenTelemetry meter name used by all Neo4j graph metrics.
@@ -41,9 +41,30 @@ struct GraphMetrics {
     slow: Counter<u64>,
 }
 
-/// Returns the single-element attribute slice used for all Neo4j metric recordings.
-fn query_attrs(label: Option<&'static str>) -> [KeyValue; 1] {
-    [KeyValue::new("query", label.unwrap_or("unknown"))]
+/// Builds the shared metric, span, and log attributes: `query=<label>` first,
+/// then the query's telemetry attributes.
+fn build_attrs(query: &Query) -> Vec<KeyValue> {
+    std::iter::once(KeyValue::new("query", query.label()))
+        .chain(
+            query
+                .telemetry_attrs()
+                .iter()
+                .map(|(key, value)| match value {
+                    TelemetryValue::Str(s) => KeyValue::new(*key, *s),
+                    TelemetryValue::Int(i) => KeyValue::new(*key, i64::from(*i)),
+                }),
+        )
+        .collect()
+}
+
+/// `key=value` pairs after the leading `query` attribute, for slow-query logs.
+fn format_attrs(attrs: &[KeyValue]) -> String {
+    attrs
+        .iter()
+        .skip(1)
+        .map(|kv| format!("{}={}", kv.key, kv.value))
+        .collect::<Vec<_>>()
+        .join(" ")
 }
 
 impl GraphMetrics {
@@ -77,7 +98,7 @@ impl GraphMetrics {
                 .build(),
             requests: meter
                 .u64_counter("neo4j.query.requests")
-                .with_description("Total Neo4j query executions, by query label")
+                .with_description("Total Neo4j query executions")
                 .build(),
             errors: meter
                 .u64_counter("neo4j.query.errors")
@@ -97,7 +118,9 @@ impl GraphMetrics {
 /// records OpenTelemetry metrics when dropped.
 struct InstrumentedStream {
     inner: BoxStream<'static, Result<Row, neo4rs::Error>>,
-    label: Option<&'static str>,
+    label: &'static str,
+    /// Prebuilt metric/span attributes: `query=<label>` + telemetry attributes.
+    attrs: Vec<KeyValue>,
     /// Populated cypher text for debug logging (only set when `slow_query_logging_include_cypher` is enabled).
     cypher: Option<String>,
     /// Pool-acquire + Bolt RUN round-trip (query planning & start of execution).
@@ -114,9 +137,11 @@ struct InstrumentedStream {
 }
 
 impl InstrumentedStream {
+    #[allow(clippy::too_many_arguments)]
     fn new(
         inner: BoxStream<'static, Result<Row, neo4rs::Error>>,
-        label: Option<&'static str>,
+        label: &'static str,
+        attrs: Vec<KeyValue>,
         cypher: Option<String>,
         execute_duration: Duration,
         threshold: Option<Duration>,
@@ -126,6 +151,7 @@ impl InstrumentedStream {
         Self {
             inner,
             label,
+            attrs,
             cypher,
             execute_duration,
             stream_start: Instant::now(),
@@ -158,7 +184,7 @@ impl Drop for InstrumentedStream {
         let fetch_duration = self.stream_start.elapsed();
         let total = self.execute_duration + fetch_duration;
 
-        let attrs: &[KeyValue] = &query_attrs(self.label);
+        let attrs: &[KeyValue] = &self.attrs;
 
         // Always record metrics (no-op when OTLP is not configured).
         self.metrics.requests.add(1, attrs);
@@ -176,18 +202,16 @@ impl Drop for InstrumentedStream {
 
         if is_slow {
             self.metrics.slow.add(1, attrs);
-
-            if let Some(label) = &self.label {
-                warn!(
-                    total_ms = total.as_millis(),
-                    execute_ms = self.execute_duration.as_millis(),
-                    fetch_ms = fetch_duration.as_millis(),
-                    rows = self.row_count,
-                    query = %label,
-                    cypher = self.cypher.as_deref().unwrap_or(""),
-                    "Slow Neo4j query"
-                );
-            }
+            warn!(
+                total_ms = total.as_millis(),
+                execute_ms = self.execute_duration.as_millis(),
+                fetch_ms = fetch_duration.as_millis(),
+                rows = self.row_count,
+                query = %self.label,
+                attrs = %format_attrs(attrs),
+                cypher = self.cypher.as_deref().unwrap_or(""),
+                "Slow Neo4j query"
+            );
         }
 
         // End the OTel span with query timing attributes
@@ -248,22 +272,21 @@ impl<G: GraphOps> InstrumentedGraph<G> {
         &self,
         elapsed: Duration,
         attrs: &[KeyValue],
-        label: Option<&'static str>,
+        label: &'static str,
         cypher: Option<&str>,
         suffix: &'static str,
     ) {
         if let Some(threshold) = self.slow_query_threshold {
             if elapsed > threshold {
                 self.metrics.slow.add(1, attrs);
-                if let Some(lbl) = label {
-                    warn!(
-                        elapsed_ms = elapsed.as_millis(),
-                        query = %lbl,
-                        cypher = cypher.unwrap_or(""),
-                        "Slow Neo4j query{}",
-                        suffix
-                    );
-                }
+                warn!(
+                    elapsed_ms = elapsed.as_millis(),
+                    query = %label,
+                    attrs = %format_attrs(attrs),
+                    cypher = cypher.unwrap_or(""),
+                    "Slow Neo4j query{}",
+                    suffix
+                );
             }
         }
     }
@@ -276,16 +299,19 @@ impl<G: GraphOps> GraphOps for InstrumentedGraph<G> {
         query: Query,
     ) -> neo4rs::Result<BoxStream<'static, Result<Row, neo4rs::Error>>> {
         let label = query.label();
-        let span_name = label.unwrap_or("neo4j.query");
+        let attrs = build_attrs(&query);
         let cypher = self.log_cypher.then(|| query.to_cypher_populated());
 
         // Create an OTel span for this query (child of the current context).
         let tracer = global::tracer(TRACER_NAME);
         let mut span = tracer
-            .span_builder(span_name)
+            .span_builder(label)
             .with_kind(SpanKind::Client)
             .start(&tracer);
         span.set_attribute(KeyValue::new("db.system", "neo4j"));
+        for attr in &attrs {
+            span.set_attribute(attr.clone());
+        }
 
         let start = Instant::now();
         let result = self.inner.execute(query).await;
@@ -299,6 +325,7 @@ impl<G: GraphOps> GraphOps for InstrumentedGraph<G> {
                 let instrumented = InstrumentedStream::new(
                     stream,
                     label,
+                    attrs,
                     cypher,
                     execute_duration,
                     self.slow_query_threshold,
@@ -308,12 +335,13 @@ impl<G: GraphOps> GraphOps for InstrumentedGraph<G> {
                 Ok(instrumented.boxed())
             }
             Err(e) => {
-                let attrs: &[KeyValue] = &query_attrs(label);
+                let attrs: &[KeyValue] = &attrs;
                 self.metrics.requests.add(1, attrs);
                 self.metrics.duration.record(ms(execute_duration), attrs);
                 self.metrics
                     .execute_duration
                     .record(ms(execute_duration), attrs);
+                self.metrics.rows.record(0, attrs);
                 self.metrics.errors.add(1, attrs);
                 span.set_status(Status::error(e.to_string()));
                 span.end();
@@ -332,21 +360,24 @@ impl<G: GraphOps> GraphOps for InstrumentedGraph<G> {
 
     async fn run(&self, query: Query) -> neo4rs::Result<()> {
         let label = query.label();
-        let span_name = label.unwrap_or("neo4j.query");
+        let attrs = build_attrs(&query);
         let cypher = self.log_cypher.then(|| query.to_cypher_populated());
 
         let tracer = global::tracer(TRACER_NAME);
         let mut span = tracer
-            .span_builder(span_name)
+            .span_builder(label)
             .with_kind(SpanKind::Client)
             .start(&tracer);
         span.set_attribute(KeyValue::new("db.system", "neo4j"));
+        for attr in &attrs {
+            span.set_attribute(attr.clone());
+        }
 
         let start = Instant::now();
         let result = self.inner.run(query).await;
         let elapsed = start.elapsed();
 
-        let attrs: &[KeyValue] = &query_attrs(label);
+        let attrs: &[KeyValue] = &attrs;
         self.metrics.requests.add(1, attrs);
         self.metrics.duration.record(ms(elapsed), attrs);
         self.metrics.execute_duration.record(ms(elapsed), attrs);
@@ -382,14 +413,75 @@ mod tests {
         Row::new(fields, data)
     }
 
+    fn test_attrs(label: &'static str) -> Vec<KeyValue> {
+        vec![KeyValue::new("query", label)]
+    }
+
+    // ── build_attrs / format_attrs ──────────────────────────────────
+
+    #[test]
+    fn build_attrs_prepends_query_label_and_converts_values() {
+        let q = Query::new("post_stream", "RETURN 1")
+            .telemetry_attr("source", "wot")
+            .telemetry_attr("depth", 2_u8);
+        let attrs = build_attrs(&q);
+
+        let as_strings: Vec<(String, String)> = attrs
+            .iter()
+            .map(|kv| (kv.key.to_string(), kv.value.to_string()))
+            .collect();
+        assert_eq!(
+            as_strings,
+            vec![
+                ("query".into(), "post_stream".into()),
+                ("source".into(), "wot".into()),
+                ("depth".into(), "2".into()),
+            ]
+        );
+    }
+
+    #[test]
+    fn format_attrs_skips_leading_query() {
+        let attrs = vec![
+            KeyValue::new("query", "post_stream"),
+            KeyValue::new("source", "wot"),
+            KeyValue::new("depth", 2_i64),
+        ];
+        assert_eq!(format_attrs(&attrs), "source=wot depth=2");
+        assert_eq!(format_attrs(&[KeyValue::new("query", "post_stream")]), "");
+    }
+
+    #[tokio::test]
+    #[traced_test]
+    async fn slow_warning_includes_telemetry_attrs() {
+        let q = Query::new("post_stream", "RETURN 1")
+            .telemetry_attr("source", "wot")
+            .telemetry_attr("depth", 2_u8);
+        let ts = InstrumentedStream::new(
+            stream::empty().boxed(),
+            q.label(),
+            build_attrs(&q),
+            None,
+            Duration::ZERO,
+            Some(Duration::ZERO),
+            GraphMetrics::new(),
+            None,
+        );
+        drop(ts);
+
+        assert!(logs_contain("Slow Neo4j query"));
+        assert!(logs_contain("source=wot depth=2"));
+    }
+
     fn make_instrumented_stream(
         inner: BoxStream<'static, Result<Row, neo4rs::Error>>,
-        label: Option<&'static str>,
+        label: &'static str,
         threshold: Option<Duration>,
     ) -> InstrumentedStream {
         InstrumentedStream::new(
             inner,
             label,
+            test_attrs(label),
             None,
             Duration::ZERO,
             threshold,
@@ -404,7 +496,7 @@ mod tests {
     async fn counts_rows_from_inner_stream() {
         let rows = vec![Ok(dummy_row()), Ok(dummy_row()), Ok(dummy_row())];
         let inner = stream::iter(rows).boxed();
-        let mut ts = make_instrumented_stream(inner, Some("test"), Some(Duration::from_secs(100)));
+        let mut ts = make_instrumented_stream(inner, "test", Some(Duration::from_secs(100)));
 
         while ts.next().await.is_some() {}
         assert_eq!(ts.row_count, 3);
@@ -413,7 +505,7 @@ mod tests {
     #[tokio::test]
     async fn empty_stream_yields_none_and_zero_rows() {
         let inner = stream::empty().boxed();
-        let mut ts = make_instrumented_stream(inner, Some("test"), Some(Duration::from_secs(100)));
+        let mut ts = make_instrumented_stream(inner, "test", Some(Duration::from_secs(100)));
         assert!(ts.next().await.is_none());
         assert_eq!(ts.row_count, 0);
     }
@@ -425,7 +517,7 @@ mod tests {
         // Threshold is 0ms — every query is "slow", but all rows must still be returned.
         let rows = vec![Ok(dummy_row()), Ok(dummy_row())];
         let inner = stream::iter(rows).boxed();
-        let mut ts = make_instrumented_stream(inner, Some("slow_q"), Some(Duration::ZERO));
+        let mut ts = make_instrumented_stream(inner, "slow_q", Some(Duration::ZERO));
 
         let mut collected = Vec::new();
         while let Some(item) = ts.next().await {
@@ -442,7 +534,7 @@ mod tests {
     async fn emits_warning_when_threshold_exceeded() {
         let inner = stream::iter(vec![Ok(dummy_row())]).boxed();
         // threshold = 0 → always slow
-        let mut ts = make_instrumented_stream(inner, Some("slow_label"), Some(Duration::ZERO));
+        let mut ts = make_instrumented_stream(inner, "slow_label", Some(Duration::ZERO));
         while ts.next().await.is_some() {}
         drop(ts);
 
@@ -454,18 +546,7 @@ mod tests {
     #[traced_test]
     async fn no_warning_when_under_threshold() {
         let inner = stream::empty().boxed();
-        let ts = make_instrumented_stream(inner, Some("fast_q"), Some(Duration::from_secs(600)));
-        drop(ts);
-
-        assert!(!logs_contain("Slow Neo4j query"));
-    }
-
-    #[tokio::test]
-    #[traced_test]
-    async fn no_warning_when_label_is_none() {
-        let inner = stream::empty().boxed();
-        // threshold = 0 but no label → drop skips logging entirely
-        let ts = make_instrumented_stream(inner, None, Some(Duration::ZERO));
+        let ts = make_instrumented_stream(inner, "fast_q", Some(Duration::from_secs(600)));
         drop(ts);
 
         assert!(!logs_contain("Slow Neo4j query"));
@@ -476,7 +557,8 @@ mod tests {
     async fn warning_includes_cypher_when_set() {
         let ts = InstrumentedStream::new(
             stream::empty().boxed(),
-            Some("cypher_q"),
+            "cypher_q",
+            test_attrs("cypher_q"),
             Some("MATCH (n) RETURN n".into()),
             Duration::ZERO,
             Some(Duration::ZERO),
diff --git a/nexus-common/src/db/graph/queries/get.rs b/nexus-common/src/db/graph/queries/get.rs
index a9f665412..5404e8d2a 100644
--- a/nexus-common/src/db/graph/queries/get.rs
+++ b/nexus-common/src/db/graph/queries/get.rs
@@ -758,23 +758,11 @@ pub fn get_tags() -> Query {
     )
 }
 
-/// Metric label `{base}_{reach}` / `{base}_wot_{depth}`. Depth lives here
-/// (unlike `post_stream_wot`, these queries have no companion meter). A new
-/// WoT depth must get its own arm — never reuse an existing label.
-macro_rules! reach_query_label {
-    ($base:literal, $reach:expr) => {
-        match $reach {
-            StreamReach::Followers => concat!($base, "_followers"),
-            StreamReach::Following => concat!($base, "_following"),
-            StreamReach::Friends => concat!($base, "_friends"),
-            StreamReach::Wot(depth) => match depth.get() {
-                1 => concat!($base, "_wot_1"),
-                2 => concat!($base, "_wot_2"),
-                3 => concat!($base, "_wot_3"),
-                other => unreachable!("WotDepth out of range: {other}"),
-            },
-        }
-    };
+fn reach_attrs(query: Query, reach: &StreamReach) -> Query {
+    let (name, depth) = reach.telemetry_dimensions();
+    query
+        .telemetry_attr("reach", name)
+        .telemetry_attr_opt("depth", depth)
 }
 
 pub fn get_tag_taggers_by_reach(
@@ -804,14 +792,11 @@ pub fn get_tag_taggers_by_reach(
             ",
         stream_reach_to_graph_subquery(&reach)
     );
-    Query::new(
-        reach_query_label!("get_tag_taggers_by_reach", reach),
-        &cypher,
-    )
-    .param("label", label)
-    .param("user_id", user_id)
-    .param("skip", skip as i64)
-    .param("limit", limit as i64)
+    reach_attrs(Query::new("get_tag_taggers_by_reach", &cypher), &reach)
+        .param("label", label)
+        .param("user_id", user_id)
+        .param("skip", skip as i64)
+        .param("limit", limit as i64)
 }
 
 pub fn get_hot_tags_by_reach(
@@ -849,7 +834,7 @@ pub fn get_hot_tags_by_reach(
         input_tagged_type,
         tags_query.taggers_limit
     );
-    Query::new(reach_query_label!("get_hot_tags_by_reach", reach), &cypher)
+    reach_attrs(Query::new("get_hot_tags_by_reach", &cypher), &reach)
         .param("user_id", user_id)
         .param("skip", tags_query.skip as i64)
         .param("limit", tags_query.limit as i64)
@@ -933,15 +918,12 @@ pub fn get_influencers_by_reach(
     ",
         stream_reach_to_graph_subquery(&reach),
     );
-    Query::new(
-        reach_query_label!("get_influencers_by_reach", reach),
-        &cypher,
-    )
-    .param("user_id", user_id)
-    .param("skip", skip as i64)
-    .param("limit", limit as i64)
-    .param("from", from)
-    .param("to", to)
+    reach_attrs(Query::new("get_influencers_by_reach", &cypher), &reach)
+        .param("user_id", user_id)
+        .param("skip", skip as i64)
+        .param("limit", limit as i64)
+        .param("from", from)
+        .param("to", to)
 }
 
 pub fn get_global_influencers(skip: usize, limit: usize, timeframe: &Timeframe) -> Query {
@@ -1237,25 +1219,17 @@ pub fn post_stream(
         cypher.push_str(&format!("LIMIT {}\n", limit.min(MAX_QUERY_LIMIT)));
     }
 
-    let query_name = match &source {
-        StreamSource::Following { .. } => "post_stream_following",
-        StreamSource::Followers { .. } => "post_stream_followers",
-        StreamSource::Friends { .. } => "post_stream_friends",
-        StreamSource::Bookmarks { .. } => "post_stream_bookmarks",
-        StreamSource::Author { .. } => "post_stream_author",
-        StreamSource::AuthorReplies { .. } => "post_stream_author_replies",
-        StreamSource::PostReplies { .. } => "post_stream_post_replies",
-        // Short-circuited upstream in collect_post_keys.
-        StreamSource::Collection { .. } => {
-            return Err(GraphError::QueryBuildError(
-                "StreamSource::Collection must be served by collect_post_keys".to_string(),
-            ));
-        }
-        StreamSource::Wot { .. } => "post_stream_wot",
-        StreamSource::WotDomain { .. } => "post_stream_wot_domain",
-        StreamSource::All => "post_stream_all",
-    };
-    let query = Query::new(query_name, &cypher);
+    // Short-circuited upstream in collect_post_keys.
+    if matches!(&source, StreamSource::Collection { .. }) {
+        return Err(GraphError::QueryBuildError(
+            "StreamSource::Collection must be served by collect_post_keys".to_string(),
+        ));
+    }
+
+    let (source_attr, depth) = source.telemetry_dimensions();
+    let query = Query::new("post_stream", &cypher)
+        .telemetry_attr("source", source_attr)
+        .telemetry_attr_opt("depth", depth);
     Ok(build_query_with_params(
         query,
         &source,
@@ -1487,9 +1461,11 @@ pub fn get_tag_by_tagger_and_id(tagger_id: &str, tag_id: &str) -> Query {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::db::graph::query::TelemetryValue;
     use crate::types::WotDepth;
+    use TelemetryValue::{Int, Str};
 
-    fn build(source: StreamSource) -> String {
+    fn build_query(source: StreamSource) -> Query {
         post_stream(
             source,
             StreamSorting::Timeline,
@@ -1502,36 +1478,132 @@ mod tests {
             None,
         )
         .unwrap()
-        .to_cypher_populated()
+    }
+
+    fn build(source: StreamSource) -> String {
+        build_query(source).to_cypher_populated()
     }
 
     #[test]
-    fn reach_queries_split_label_by_reach_and_depth() {
+    fn reach_queries_use_base_label_with_reach_and_depth_attrs() {
         let cases = [
-            (StreamReach::Followers, "get_influencers_by_reach_followers"),
-            (StreamReach::Following, "get_influencers_by_reach_following"),
-            (StreamReach::Friends, "get_influencers_by_reach_friends"),
+            (StreamReach::Followers, vec![("reach", Str("followers"))]),
+            (StreamReach::Following, vec![("reach", Str("following"))]),
+            (StreamReach::Friends, vec![("reach", Str("friends"))]),
             (
                 StreamReach::Wot(WotDepth::new(1).unwrap()),
-                "get_influencers_by_reach_wot_1",
+                vec![("reach", Str("wot")), ("depth", Int(1))],
+            ),
+            (
+                StreamReach::Wot(WotDepth::new(2).unwrap()),
+                vec![("reach", Str("wot")), ("depth", Int(2))],
             ),
             (
                 StreamReach::Wot(WotDepth::new(3).unwrap()),
-                "get_influencers_by_reach_wot_3",
+                vec![("reach", Str("wot")), ("depth", Int(3))],
             ),
         ];
         for (reach, expected) in cases {
-            let query = get_influencers_by_reach("user", reach, 0, 10, &Timeframe::AllTime);
-            assert_eq!(query.label(), Some(expected));
+            let influencers =
+                get_influencers_by_reach("user", reach.clone(), 0, 10, &Timeframe::AllTime);
+            assert_eq!(influencers.label(), "get_influencers_by_reach");
+            assert_eq!(influencers.telemetry_attrs(), expected.as_slice());
+
+            let taggers = get_tag_taggers_by_reach("tag", "user", reach.clone(), 0, 10);
+            assert_eq!(taggers.label(), "get_tag_taggers_by_reach");
+            assert_eq!(taggers.telemetry_attrs(), expected.as_slice());
+
+            let hot_tags_input = HotTagsInputDTO::new(Timeframe::AllTime, 10, 0, 5, None);
+            let hot_tags = get_hot_tags_by_reach("user", reach, &hot_tags_input);
+            assert_eq!(hot_tags.label(), "get_hot_tags_by_reach");
+            assert_eq!(hot_tags.telemetry_attrs(), expected.as_slice());
         }
+    }
 
-        let taggers =
-            get_tag_taggers_by_reach("tag", "user", StreamReach::Wot(WotDepth::default()), 0, 10);
-        assert_eq!(taggers.label(), Some("get_tag_taggers_by_reach_wot_2"));
-
-        let hot_tags_input = HotTagsInputDTO::new(Timeframe::AllTime, 10, 0, 5, None);
-        let hot_tags = get_hot_tags_by_reach("user", StreamReach::Friends, &hot_tags_input);
-        assert_eq!(hot_tags.label(), Some("get_hot_tags_by_reach_friends"));
+    #[test]
+    fn post_stream_uses_base_label_with_source_and_depth_attrs() {
+        let observer = || "obs".to_string();
+        let cases: Vec<(StreamSource, Vec<(&'static str, TelemetryValue)>)> = vec![
+            (StreamSource::All, vec![("source", Str("all"))]),
+            (
+                StreamSource::Following {
+                    observer_id: observer(),
+                },
+                vec![("source", Str("following"))],
+            ),
+            (
+                StreamSource::Followers {
+                    observer_id: observer(),
+                },
+                vec![("source", Str("followers"))],
+            ),
+            (
+                StreamSource::Friends {
+                    observer_id: observer(),
+                },
+                vec![("source", Str("friends"))],
+            ),
+            (
+                StreamSource::Bookmarks {
+                    observer_id: observer(),
+                },
+                vec![("source", Str("bookmarks"))],
+            ),
+            (
+                StreamSource::Author {
+                    author_id: observer(),
+                },
+                vec![("source", Str("author"))],
+            ),
+            (
+                StreamSource::AuthorReplies {
+                    author_id: observer(),
+                },
+                vec![("source", Str("author_replies"))],
+            ),
+            (
+                StreamSource::PostReplies {
+                    post_id: "post".to_string(),
+                    author_id: observer(),
+                },
+                vec![("source", Str("post_replies"))],
+            ),
+            (
+                StreamSource::Wot {
+                    observer_id: observer(),
+                    depth: WotDepth::new(1).unwrap(),
+                },
+                vec![("source", Str("wot")), ("depth", Int(1))],
+            ),
+            (
+                StreamSource::Wot {
+                    observer_id: observer(),
+                    depth: WotDepth::new(3).unwrap(),
+                },
+                vec![("source", Str("wot")), ("depth", Int(3))],
+            ),
+            (
+                StreamSource::WotDomain {
+                    observer_id: observer(),
+                    trust: DomainTrust::Me,
+                    domain_tags: vec!["bitcoin".to_string()],
+                },
+                vec![("source", Str("wot_domain")), ("depth", Int(0))],
+            ),
+            (
+                StreamSource::WotDomain {
+                    observer_id: observer(),
+                    trust: DomainTrust::Network(WotDepth::new(2).unwrap()),
+                    domain_tags: vec!["bitcoin".to_string()],
+                },
+                vec![("source", Str("wot_domain")), ("depth", Int(2))],
+            ),
+        ];
+        for (source, expected) in cases {
+            let query = build_query(source);
+            assert_eq!(query.label(), "post_stream");
+            assert_eq!(query.telemetry_attrs(), expected.as_slice());
+        }
     }
 
     /// The trust traversal must bind and dedupe authors before the posts MATCH.
diff --git a/nexus-common/src/db/graph/query.rs b/nexus-common/src/db/graph/query.rs
index 75d055d05..1f94e9213 100644
--- a/nexus-common/src/db/graph/query.rs
+++ b/nexus-common/src/db/graph/query.rs
@@ -2,28 +2,75 @@ use std::fmt::Write;
 
 use neo4rs::{BoltList, BoltMap, BoltString, BoltType};
 
-/// Our own `Query` type that mirrors `neo4rs::Query` but exposes
-/// `cypher()` and `params_map()` for logging and tracing.
+/// Bounded telemetry value: `'static` strings and `u8` (WoT depth). Wider
+/// types would let request data become metric dimensions.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum TelemetryValue {
+    Str(&'static str),
+    Int(u8),
+}
+
+impl From<&'static str> for TelemetryValue {
+    fn from(value: &'static str) -> Self {
+        Self::Str(value)
+    }
+}
+
+impl From<u8> for TelemetryValue {
+    fn from(value: u8) -> Self {
+        Self::Int(value)
+    }
+}
+
+/// Query builder with a stable telemetry label and bounded attributes.
 #[derive(Clone)]
 pub struct Query {
-    label: Option<&'static str>,
+    label: &'static str,
     cypher: String,
     params: BoltMap,
+    /// Telemetry-only attributes; never sent to Neo4j.
+    telemetry_attrs: Vec<(&'static str, TelemetryValue)>,
 }
 
 impl Query {
     pub fn new(label: &'static str, cypher: impl Into<String>) -> Self {
         Self {
-            label: Some(label),
+            label,
             cypher: cypher.into(),
             params: BoltMap::default(),
+            telemetry_attrs: Vec::new(),
         }
     }
 
-    pub fn label(&self) -> Option<&'static str> {
+    pub fn label(&self) -> &'static str {
         self.label
     }
 
+    /// Attach a telemetry dimension. Never sent to Neo4j.
+    pub(crate) fn telemetry_attr(
+        mut self,
+        key: &'static str,
+        value: impl Into<TelemetryValue>,
+    ) -> Self {
+        self.telemetry_attrs.push((key, value.into()));
+        self
+    }
+
+    pub(crate) fn telemetry_attr_opt(
+        self,
+        key: &'static str,
+        value: Option<impl Into<TelemetryValue>>,
+    ) -> Self {
+        match value {
+            Some(value) => self.telemetry_attr(key, value),
+            None => self,
+        }
+    }
+
+    pub(crate) fn telemetry_attrs(&self) -> &[(&'static str, TelemetryValue)] {
+        &self.telemetry_attrs
+    }
+
     pub fn param<T: Into<BoltType>>(mut self, key: &str, value: T) -> Self {
         self.params.put(key.into(), value.into());
         self
@@ -49,22 +96,27 @@ impl Query {
 
 /// Replaces `$param` placeholders in `cypher` with literal values from `params`.
 pub fn populate_cypher(cypher: &str, params: &BoltMap) -> String {
-    let mut out = cypher.to_owned();
-    // Sort keys by length descending so `$skip` is replaced before a
-    // hypothetical `$s`, avoiding partial substitutions.
-    //
-    // NOTE: There is still a potential issue with this approach: if a
-    // parameter *value* happens to contain text matching another parameter
-    // name (e.g. param "a" has value "$b"), a later replacement pass will
-    // substitute inside the already-replaced value. A proper fix would
-    // require single-pass replacement or placeholder-based substitution.
-    let mut entries: Vec<_> = params.value.iter().collect();
-    entries.sort_by_key(|b| std::cmp::Reverse(b.0.value.len()));
-    for (k, v) in entries {
-        let placeholder = format!("${}", k.value);
-        let literal = bolt_to_cypher_literal(v);
-        out = out.replace(&placeholder, &literal);
+    let mut out = String::with_capacity(cypher.len());
+    let mut rest = cypher;
+    while let Some(i) = rest.find('$') {
+        out.push_str(&rest[..i]);
+        rest = &rest[i + 1..];
+        let len = rest
+            .find(|c: char| c != '_' && !c.is_alphanumeric())
+            .unwrap_or(rest.len());
+        let name = &rest[..len];
+        rest = &rest[len..];
+        match params.value.iter().find(|(key, _)| key.value == name) {
+            Some((_, value)) if !name.is_empty() => {
+                out.push_str(&bolt_to_cypher_literal(value));
+            }
+            _ => {
+                out.push('$');
+                out.push_str(name);
+            }
+        }
     }
+    out.push_str(rest);
     out
 }
 
@@ -126,11 +178,7 @@ impl From<Query> for neo4rs::Query {
 
 #[cfg(test)]
 fn query(cypher: impl Into<String>) -> Query {
-    Query {
-        label: None,
-        cypher: cypher.into(),
-        params: BoltMap::default(),
-    }
+    Query::new("test", cypher)
 }
 
 #[cfg(test)]
@@ -290,6 +338,20 @@ mod tests {
         assert_eq!(result, "SET u.bio = 'it\\'s a\\nnew \"line\"'");
     }
 
+    #[test]
+    fn populate_does_not_rescan_inserted_literals() {
+        let q = query("RETURN $a AS a, $b AS b")
+            .param("a", "$b")
+            .param("b", "secret");
+        assert_eq!(q.to_cypher_populated(), "RETURN '$b' AS a, 'secret' AS b");
+    }
+
+    #[test]
+    fn populate_preserves_unknown_parameters() {
+        let q = query("RETURN $known, $unknown").param("known", 1_i64);
+        assert_eq!(q.to_cypher_populated(), "RETURN 1, $unknown");
+    }
+
     // ── From<Query> for neo4rs::Query ───────────────────────────────
 
     #[test]
@@ -301,6 +363,36 @@ mod tests {
         let _ = nq;
     }
 
+    // ── Telemetry attributes ────────────────────────────────────────
+
+    #[test]
+    fn telemetry_attrs_are_exposed_in_insertion_order() {
+        let q = Query::new("test", "RETURN 1")
+            .telemetry_attr("reach", "wot")
+            .telemetry_attr("depth", 2_u8);
+        assert_eq!(
+            q.telemetry_attrs(),
+            &[
+                ("reach", TelemetryValue::Str("wot")),
+                ("depth", TelemetryValue::Int(2)),
+            ]
+        );
+    }
+
+    #[test]
+    fn telemetry_attrs_are_not_query_params() {
+        let q = Query::new(
+            "test",
+            "MATCH (n) WHERE n.r = $reach AND n.d = $depth RETURN n",
+        )
+        .telemetry_attr("reach", "wot")
+        .telemetry_attr("depth", 2_u8);
+
+        let populated = q.to_cypher_populated();
+        assert!(populated.contains("$reach"));
+        assert!(populated.contains("$depth"));
+    }
+
     // ── Query builder ───────────────────────────────────────────────
 
     #[test]
diff --git a/nexus-common/src/models/post/stream.rs b/nexus-common/src/models/post/stream.rs
index 436c868b2..6aecc7913 100644
--- a/nexus-common/src/models/post/stream.rs
+++ b/nexus-common/src/models/post/stream.rs
@@ -73,6 +73,29 @@ pub enum StreamSource {
 }
 
 impl StreamSource {
+    /// Low-cardinality source value and optional WoT depth for telemetry.
+    pub(crate) fn telemetry_dimensions(&self) -> (&'static str, Option<u8>) {
+        match self {
+            StreamSource::PostReplies { .. } => ("post_replies", None),
+            StreamSource::Following { .. } => ("following", None),
+            StreamSource::Followers { .. } => ("followers", None),
+            StreamSource::Friends { .. } => ("friends", None),
+            StreamSource::Bookmarks { .. } => ("bookmarks", None),
+            StreamSource::Author { .. } => ("author", None),
+            StreamSource::AuthorReplies { .. } => ("author_replies", None),
+            StreamSource::Collection { .. } => ("collection", None),
+            StreamSource::Wot { depth, .. } => ("wot", Some(depth.get())),
+            StreamSource::WotDomain { trust, .. } => (
+                "wot_domain",
+                Some(match trust {
+                    DomainTrust::Me => 0,
+                    DomainTrust::Network(depth) => depth.get(),
+                }),
+            ),
+            StreamSource::All => ("all", None),
+        }
+    }
+
     pub fn get_observer(&self) -> Option<&str> {
         match self {
             StreamSource::Followers { observer_id }
diff --git a/nexus-common/src/types/mod.rs b/nexus-common/src/types/mod.rs
index c392b5793..2c4412f40 100644
--- a/nexus-common/src/types/mod.rs
+++ b/nexus-common/src/types/mod.rs
@@ -93,6 +93,18 @@ pub enum StreamReach {
     Wot(WotDepth),
 }
 
+impl StreamReach {
+    /// Low-cardinality reach value and optional WoT depth for telemetry.
+    pub(crate) fn telemetry_dimensions(&self) -> (&'static str, Option<u8>) {
+        match self {
+            StreamReach::Followers => ("followers", None),
+            StreamReach::Following => ("following", None),
+            StreamReach::Friends => ("friends", None),
+            StreamReach::Wot(depth) => ("wot", Some(depth.get())),
+        }
+    }
+}
+
 impl<'de> Deserialize<'de> for StreamReach {
     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
     where