diff --git a/crates/lance-context-metrics/src/lib.rs b/crates/lance-context-metrics/src/lib.rs index 597e5281..a30b282a 100644 --- a/crates/lance-context-metrics/src/lib.rs +++ b/crates/lance-context-metrics/src/lib.rs @@ -344,10 +344,11 @@ pub async fn http_metrics_layer(req: Request, 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), diff --git a/crates/lance-context-metrics/src/recovery_signals.rs b/crates/lance-context-metrics/src/recovery_signals.rs index f9052b0b..eb3f07b0 100644 --- a/crates/lance-context-metrics/src/recovery_signals.rs +++ b/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, + ) { + 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); diff --git a/crates/lance-context-metrics/tests/recovery_signals.rs b/crates/lance-context-metrics/tests/recovery_signals.rs index 28f47018..d144fc4c 100644 --- a/crates/lance-context-metrics/tests/recovery_signals.rs +++ b/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() diff --git a/docs/recovery-request-signals.md b/docs/recovery-request-signals.md new file mode 100644 index 00000000..096ffeb2 --- /dev/null +++ b/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=`. +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.