feat: identify slow reads in bounded recovery signals

crates/lance-context-metrics/src/lib.rs

@@ -344,10 +344,11 @@ pub async fn http_metrics_layer(req: Request<axum::body::Body>, next: Next) -> R
 
     let recovery_target = recovery_signals::target(req.uri().path(), &path, &method);
     let response = next.run(req).await;
-    recovery_signals::record(recovery_target, response.status());
+    let elapsed = start.elapsed();
+    recovery_signals::record(recovery_target, response.status(), elapsed);
 
     let status = response.status().as_u16().to_string();
-    let elapsed = start.elapsed().as_secs_f64();
+    let elapsed = elapsed.as_secs_f64();
     let labels = [
         ("method", method.to_string()),
         ("path", path),

crates/lance-context-metrics/src/recovery_signals.rs

@@ -21,12 +21,15 @@ const PER_SHARD: usize = 128;
 const WINDOW: Duration = Duration::from_secs(300);
 const MAX_REPLY: usize = 1024 * 1024;
 const MAX_TARGET: usize = 1032;
+// Observation threshold, not a deadline: never cancel or retry a request here.
+const SLOW_READ: Duration = Duration::from_secs(5);
 
 #[derive(Clone)]
 struct Entry {
     touched: Instant,
     active_at: u64,
     read_error: Option<(Instant, u64, u16)>,
+    read_slow: Option<(Instant, u64, u16, u64)>,
 }
 
 struct Signals {
@@ -81,8 +84,24 @@ impl Signals {
     }
 
     fn record(&self, target: String, read: bool, status: StatusCode, now: Instant, wall: u64) {
+        self.record_with_latency(target, read, status, now, wall, None);
+    }
+
+    fn record_with_latency(
+        &self,
+        target: String,
+        read: bool,
+        status: StatusCode,
+        now: Instant,
+        wall: u64,
+        elapsed: Option<Duration>,
+    ) {
+        let slow = elapsed.filter(|elapsed| read && *elapsed >= SLOW_READ);
         if !self.enabled
-            || !(status.is_success() || status.is_redirection() || status.is_server_error())
+            || !(status.is_success()
+                || status.is_redirection()
+                || status.is_server_error()
+                || slow.is_some())
         {
             return;
         }
@@ -105,12 +124,21 @@ impl Signals {
             touched: now,
             active_at: wall,
             read_error: None,
+            read_slow: None,
         });
         entry.touched = now;
         entry.active_at = wall;
         if read && status.is_server_error() {
             entry.read_error = Some((now, wall, status.as_u16()));
         }
+        if let Some(elapsed) = slow {
+            entry.read_slow = Some((
+                now,
+                wall,
+                status.as_u16(),
+                elapsed.as_millis().min(u64::MAX as u128) as u64,
+            ));
+        }
     }
 
     fn snapshot(&self, nonce: &str, now: Instant) -> Response {
@@ -134,13 +162,17 @@ impl Signals {
                 }
             }
         }
-        // Preserve recent read failures first if long names fill the byte bound.
+        // Preserve recent failures, then slow reads, before ordinary activity.
         entries.sort_by_key(|(_, entry)| {
             let error = entry
                 .read_error
                 .filter(|(at, _, _)| now.saturating_duration_since(*at) < WINDOW);
+            let slow = entry
+                .read_slow
+                .filter(|(at, _, _, _)| now.saturating_duration_since(*at) < WINDOW);
             (
                 std::cmp::Reverse(error.map(|(_, at, _)| at)),
+                std::cmp::Reverse(slow.map(|(_, at, _, _)| at)),
                 std::cmp::Reverse(entry.active_at),
             )
         });
@@ -150,9 +182,15 @@ impl Signals {
             let error = entry
                 .read_error
                 .filter(|(at, _, _)| now.saturating_duration_since(*at) < WINDOW);
+            let slow = entry
+                .read_slow
+                .filter(|(at, _, _, _)| now.saturating_duration_since(*at) < WINDOW);
             let row = json!({"name": name, "active_at_unix_millis": entry.active_at,
                 "read_error_at_unix_millis": error.map(|(_, at, _)| at),
-                "read_error_status": error.map(|(_, _, status)| status)});
+                "read_error_status": error.map(|(_, _, status)| status),
+                "read_slow_at_unix_millis": slow.map(|(_, at, _, _)| at),
+                "read_slow_status": slow.map(|(_, _, status, _)| status),
+                "read_slow_header_millis": slow.map(|(_, _, _, millis)| millis)});
             let size = serde_json::to_vec(&row).expect("bounded scalar row").len() + 1;
             if bytes + size > MAX_REPLY {
                 complete = false;
@@ -170,6 +208,7 @@ impl Signals {
         complete &= dropped == dropped_before && !recent_loss;
         let value = json!({"schema":1, "nonce":nonce,"sampled_at_unix_millis":wall_millis(),
             "coverage_started_at_unix_millis":self.started_at,"window_seconds":WINDOW.as_secs(),
+            "slow_read_header_threshold_millis": SLOW_READ.as_millis() as u64,
             "complete":complete,"dropped_total":dropped,"tables":tables});
         let raw = serde_json::to_vec(&value).expect("bounded scalar envelope");
         if raw.len() > MAX_REPLY {
@@ -298,10 +337,17 @@ fn parse_target(path: &str, matched: &str, method: &Method) -> Option<(String, b
     Some((format!("{kind}{name}"), read))
 }
 
-pub(super) fn record(target: Option<(String, bool)>, status: StatusCode) {
+pub(super) fn record(target: Option<(String, bool)>, status: StatusCode, elapsed: Duration) {
     if let Some((name, true)) = target {
-        if status.is_server_error() {
-            global().record(name, true, status, Instant::now(), wall_millis());
+        if status.is_server_error() || elapsed >= SLOW_READ {
+            global().record_with_latency(
+                name,
+                true,
+                status,
+                Instant::now(),
+                wall_millis(),
+                Some(elapsed),
+            );
         }
     }
 }
@@ -521,6 +567,95 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn slow_read_hints_include_404_preserve_status_and_expire_independently() {
+        let signals = Signals::new(true);
+        let now = Instant::now();
+        for (name, read, status, elapsed) in [
+            (
+                "missing",
+                true,
+                StatusCode::NOT_FOUND,
+                Duration::from_secs(241),
+            ),
+            ("successful", true, StatusCode::OK, SLOW_READ),
+            ("failed", true, StatusCode::SERVICE_UNAVAILABLE, SLOW_READ),
+            (
+                "fast-miss",
+                true,
+                StatusCode::NOT_FOUND,
+                SLOW_READ - Duration::from_millis(1),
+            ),
+            ("write", false, StatusCode::OK, Duration::from_secs(241)),
+        ] {
+            signals.record_with_latency(name.into(), read, status, now, 100, Some(elapsed));
+        }
+        // Continued activity and a successful read must not erase either hint.
+        signals.record(
+            "missing".into(),
+            false,
+            StatusCode::OK,
+            now + Duration::from_secs(200),
+            200,
+        );
+        signals.record_with_latency(
+            "failed".into(),
+            true,
+            StatusCode::OK,
+            now + Duration::from_secs(200),
+            200,
+            Some(Duration::from_millis(1)),
+        );
+        let v = value(&signals, now + Duration::from_secs(201)).await;
+        let rows = v["tables"].as_array().unwrap();
+        let missing = rows.iter().find(|r| r["name"] == "missing").unwrap();
+        assert_eq!(missing["read_slow_header_millis"], 241_000);
+        assert_eq!(missing["read_slow_status"], 404);
+        assert_eq!(missing["read_error_status"], Value::Null);
+        assert_eq!(missing["active_at_unix_millis"], 200);
+        assert_eq!(missing["read_slow_at_unix_millis"], 100);
+        assert!(!rows.iter().any(|r| r["name"] == "fast-miss"));
+        assert_eq!(
+            rows.iter().find(|r| r["name"] == "write").unwrap()["read_slow_status"],
+            Value::Null
+        );
+        assert_eq!(
+            rows.iter().find(|r| r["name"] == "successful").unwrap()["read_slow_header_millis"],
+            5000
+        );
+        let failed = rows.iter().find(|r| r["name"] == "failed").unwrap();
+        assert_eq!(failed["read_error_status"], 503);
+        assert_eq!(failed["read_slow_status"], 503);
+        let v = value(&signals, now + WINDOW).await;
+        assert_eq!(v["tables"].as_array().unwrap().len(), 2);
+        for row in v["tables"].as_array().unwrap() {
+            assert_eq!(row["read_slow_at_unix_millis"], Value::Null);
+            assert_eq!(row["read_slow_header_millis"], Value::Null);
+            assert_eq!(row["read_error_status"], Value::Null);
+        }
+    }
+
+    #[tokio::test]
+    async fn slow_read_recording_keeps_the_nonblocking_loss_contract() {
+        let signals = Signals::new(true);
+        let now = Instant::now();
+        {
+            let _held = signals.shards[Signals::shard("slow")].lock().unwrap();
+            signals.record_with_latency(
+                "slow".into(),
+                true,
+                StatusCode::NOT_FOUND,
+                now,
+                1,
+                Some(SLOW_READ),
+            );
+        }
+        let v = value(&signals, now).await;
+        assert_eq!(v["complete"], false);
+        assert_eq!(v["dropped_total"], 1);
+        assert!(v["tables"].as_array().unwrap().is_empty());
+    }
+
     #[tokio::test]
     async fn cardinality_bytes_and_ttl_are_bounded_and_errors_win_truncation() {
         let signals = Signals::new(true);
@@ -533,6 +668,14 @@ mod tests {
             now,
             1,
         );
+        signals.record_with_latency(
+            "slow".into(),
+            true,
+            StatusCode::NOT_FOUND,
+            now,
+            1,
+            Some(SLOW_READ),
+        );
         for n in 0..20_000 {
             signals.record(
                 format!("{n}_{}", "x".repeat(1000)),
@@ -550,6 +693,7 @@ mod tests {
             .all(|s| s.lock().unwrap().len() <= PER_SHARD));
         let v = value(&signals, now).await;
         assert_eq!(v["tables"][0]["name"], "urgent");
+        assert_eq!(v["tables"][1]["name"], "slow");
         assert_eq!(v["complete"], false);
         assert!(v["dropped_total"].as_u64().unwrap() > 0);
         assert!(v["tables"].as_array().unwrap().len() < count);

crates/lance-context-metrics/tests/recovery_signals.rs

@@ -32,6 +32,13 @@ async fn middleware_records_inflight_demand_and_read_failures_without_payloads()
     let entered = Arc::new(Notify::new());
     let release = Arc::new(Notify::new());
     let app = Router::new()
+        .route(
+            "/api/v1/rollouts/{name}/records",
+            get(|| async {
+                tokio::time::sleep(std::time::Duration::from_millis(5010)).await;
+                StatusCode::NOT_FOUND
+            }),
+        )
         .route(
             "/api/v1/rollouts/{name}/records/{id}",
             get({
@@ -104,6 +111,41 @@ async fn middleware_records_inflight_demand_and_read_failures_without_payloads()
             assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
         }
     }
+    if std::env::var("RECOVERY_SIGNALS_ENABLED").unwrap() == "true" {
+        let response = app
+            .clone()
+            .oneshot(
+                Request::builder()
+                    .uri("/api/v1/rollouts/slow_missing/records?secret=query")
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_eq!(response.status(), StatusCode::NOT_FOUND);
+        let response = app
+            .clone()
+            .oneshot(
+                Request::builder()
+                    .uri("/api/v1/internal/recovery-signals?nonce=slow")
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        let bytes = to_bytes(response.into_body(), 1024 * 1024).await.unwrap();
+        assert!(!String::from_utf8_lossy(&bytes).contains("secret"));
+        let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
+        let slow = v["tables"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .find(|r| r["name"] == "slow_missing")
+            .unwrap();
+        assert_eq!(slow["read_error_status"], serde_json::Value::Null);
+        assert_eq!(slow["read_slow_status"], 404);
+        assert!(slow["read_slow_header_millis"].as_u64().unwrap() >= 5000);
+    }
     let invalid = app
         .oneshot(
             Request::builder()

docs/recovery-request-signals.md

@@ -0,0 +1,37 @@
+# Request hints for recovery
+
+With `RECOVERY_SIGNALS_ENABLED=1`, a worker exposes recent per-table request
+hints at `GET /api/v1/internal/recovery-signals?nonce=<caller-nonce>`.
+The existing schema 1 response now also includes slow-read observations:
+
+| Field | Meaning |
+| --- | --- |
+| `slow_read_header_threshold_millis` | Envelope threshold: 5,000 ms. |
+| `read_slow_at_unix_millis` | Completion time of the latest qualifying read. |
+| `read_slow_status` | That read's HTTP status, including 2xx or 4xx. |
+| `read_slow_header_millis` | Time from middleware entry until response headers. |
+
+These fields are additive. Consumers of older workers must treat absent fields
+as unavailable observations, not proof of fast reads. A null per-table field
+means there is no retained slow-read observation in the five-minute window.
+Successful activity does not clear unexpired errors or slow-read observations.
+
+This catches cases such as a `/records` request that spends four minutes scanning
+before returning 404: it was absent from the 5xx hints and could not be assigned
+to a table from route-level latency histograms. Neither request parameters nor
+record IDs or payloads enter this cache. Maintenance routes remain excluded.
+
+Request recording uses a nonblocking lock and at most 4,096 entries. Snapshots
+remain capped at 1 MiB; recent 5xx hints are preferred, then slow reads, then
+ordinary activity. Check `complete`, `dropped_total`, coverage start, and nonce
+before interpreting an empty snapshot. Loss is explicitly reported.
+
+The threshold only selects observations. It does not cancel requests, retry
+writes, declare a merge stalled, or initiate a merge. Consumers should correlate
+the table with fresh WAL metadata and actual execution progress before choosing
+a recovery action. A slow read can have causes other than WAL backlog.
+
+These observations exclude response-body streaming and requests that have not
+returned headers (including disconnected requests). Use client or ingress
+measurements for end-to-end latency. An empty window with no reads is not proof
+that the read path is healthy.