fix(server): extend query-window clamp to /api/logs; cap WATCHER_MAX_QUERY_HOURS (JEF-546)

docs/architecture.md

@@ -82,7 +82,7 @@ last rollup bucket, so it stays continuous after pruning.
 | `WATCHER_RETENTION_METRICS_DAYS` | — | per-table override of `WATCHER_RETENTION_DAYS` for `metric_series_rollups`; unset falls back to the default |
 | `WATCHER_METRICS_RAW_DAYS` | `2` | prune age for raw metric points (rollups keep history); `0` = same as retention |
 | `WATCHER_ROLLUP_BUCKET_SECS` | `300` | downsample bucket width; `0` disables rollups |
-| `WATCHER_MAX_QUERY_HOURS` | `168` (7d) | max look-back for `/api/traces` and `/api/services`; an explicit `from` is clamped to this ceiling, not honored verbatim (JEF-532) |
+| `WATCHER_MAX_QUERY_HOURS` | `168` (7d), clamped to `[1, 8760]` | max look-back for `/api/traces`, `/api/services`, and `/api/logs`; an explicit `from` (or its absence) is clamped to this ceiling, not honored verbatim (JEF-532, JEF-546) |
 | `WATCHER_ALERT_INTERVAL_SECS` | `30` | how often alert rules are evaluated (min 5) |
 | `WATCHER_ALERT_WEBHOOK` | — | optional URL to POST on alert fire/resolve |
 | `WATCHER_ALERT_SMTP_HOST` | — | SMTP relay host; setting it enables emailing alert fire/resolve (STARTTLS) |

server/src/api.rs

@@ -86,19 +86,27 @@ pub async fn list_traces(
     Ok(Json(query_traces(&pool, q).await.map_err(internal)?))
 }
 
-/// Hard ceiling (hours) on how far back a `spans`-table query may reach, even
-/// when the caller passes an explicit `from` (JEF-532). Without this, the
-/// `COALESCE($n, now() - interval '24 hours')` default-window floor below only
-/// applies when `from` is unset — an arbitrarily-far-past explicit `from`
-/// would defeat it and full-scan the retention-deep `spans` table. Defaults to
+/// Hard ceiling (hours) on how far back a `spans`/`logs`-table query may
+/// reach, even when the caller passes an explicit `from` (JEF-532). Without
+/// this, the `COALESCE($n, now() - interval '24 hours')` default-window floor
+/// below only applies when `from` is unset — an arbitrarily-far-past explicit
+/// `from` would defeat it and full-scan the retention-deep table. Defaults to
 /// the same 7-day ceiling most metric endpoints use (`resolve_window`'s
 /// `max_hours` for facet/histogram); override with `WATCHER_MAX_QUERY_HOURS`.
+/// Clamped to `[1, 8760]` (1 hour .. 1 year, JEF-546) so a misconfigured env
+/// value can't disable the ceiling (0/negative) or resolve it to the distant
+/// past and re-open the full-scan it exists to prevent (an absurdly large
+/// value).
 fn max_lookback_hours() -> i32 {
-    std::env::var("WATCHER_MAX_QUERY_HOURS")
-        .ok()
-        .and_then(|s| s.parse::<i32>().ok())
-        .filter(|h| *h > 0)
+    parse_max_lookback_hours(std::env::var("WATCHER_MAX_QUERY_HOURS").ok().as_deref())
+}
+
+/// Pure helper behind `max_lookback_hours()`, extracted so the clamp can be
+/// unit-tested without mutating process-global env state.
+fn parse_max_lookback_hours(raw: Option<&str>) -> i32 {
+    raw.and_then(|s| s.parse::<i32>().ok())
         .unwrap_or(24 * 7)
+        .clamp(1, 8760)
 }
 
 /// Recent-traces query shared by the HTTP handler and the MCP `search_traces`
@@ -262,7 +270,15 @@ pub async fn query_logs(pool: &PgPool, q: LogQuery) -> Result<Vec<LogRow>, sqlx:
            AND ($2::text IS NULL OR trace_id = $2)
            AND ($3::text IS NULL OR span_id = $3)
            AND ($4::text IS NULL OR body ILIKE '%' || $4 || '%')
-           AND ($5::timestamptz IS NULL OR time >= $5)
+           -- default to a recent window when unbounded, and clamp an explicit
+           -- `from` to the max-lookback ceiling too, so a body ILIKE search for
+           -- a rare/absent term can't full-scan the retention-deep `logs` table
+           -- to satisfy ORDER BY time DESC LIMIT (JEF-546, mirrors JEF-532's
+           -- query_traces clamp).
+           AND time >= GREATEST(
+                 COALESCE($5::timestamptz, now() - interval '24 hours'),
+                 now() - make_interval(hours => $9::int)
+               )
            AND ($6::timestamptz IS NULL OR time <= $6)
            AND ($7::jsonb IS NULL OR attributes @> $7)
          ORDER BY time DESC
@@ -276,6 +292,7 @@ pub async fn query_logs(pool: &PgPool, q: LogQuery) -> Result<Vec<LogRow>, sqlx:
     .bind(q.to)
     .bind(attr_json)
     .bind(limit)
+    .bind(max_lookback_hours())
     .fetch_all(pool)
     .instrument(tracing::info_span!("db.query"))
     .await
@@ -1363,4 +1380,29 @@ mod tests {
         assert_eq!(resolved_hi, hi);
         assert_eq!(lo, hi - Duration::hours(48));
     }
+
+    // JEF-546: a pathologically large WATCHER_MAX_QUERY_HOURS must not resolve
+    // the ceiling to the distant past and re-open the full-scan JEF-532 closed.
+    #[test]
+    fn parse_max_lookback_hours_clamps_huge_value_to_one_year() {
+        assert_eq!(parse_max_lookback_hours(Some("999999999")), 8760);
+    }
+
+    // Zero/negative must not disable the ceiling entirely.
+    #[test]
+    fn parse_max_lookback_hours_clamps_zero_and_negative_to_one_hour() {
+        assert_eq!(parse_max_lookback_hours(Some("0")), 1);
+        assert_eq!(parse_max_lookback_hours(Some("-5")), 1);
+    }
+
+    #[test]
+    fn parse_max_lookback_hours_defaults_to_one_week_when_unset_or_invalid() {
+        assert_eq!(parse_max_lookback_hours(None), 24 * 7);
+        assert_eq!(parse_max_lookback_hours(Some("not a number")), 24 * 7);
+    }
+
+    #[test]
+    fn parse_max_lookback_hours_leaves_in_range_value_unchanged() {
+        assert_eq!(parse_max_lookback_hours(Some("48")), 48);
+    }
 }

server/tests/smoke.rs

@@ -364,7 +364,10 @@ async fn ingest_and_query_a_log() {
             }),
             scope_logs: vec![ScopeLogs {
                 log_records: vec![LogRecord {
-                    time_unix_nano: 1_000_000_000,
+                    // Real "now" rather than a fixed epoch value: /api/logs floors
+                    // to a recent default window (JEF-546), so a fake 1970
+                    // timestamp would fall outside it and never come back.
+                    time_unix_nano: now_nanos(),
                     severity_number: 9, // INFO
                     severity_text: "INFO".to_string(),
                     body: Some(AnyValue {
@@ -425,9 +428,13 @@ async fn ingest_a_log_batch_persists_every_row_in_one_call() {
 
     const N: usize = 250;
     let drops_before = selfmon::DROP_INSERT.load(std::sync::atomic::Ordering::Relaxed);
+    // Real "now" rather than a fixed epoch value: /api/logs floors to a recent
+    // default window (JEF-546), so fake 1970 timestamps would fall outside it
+    // and never come back via the `/api/logs?service=batcher` check below.
+    let base_nanos = now_nanos();
     let records: Vec<LogRecord> = (0..N)
         .map(|i| LogRecord {
-            time_unix_nano: 1_000_000_000 + i as u64,
+            time_unix_nano: base_nanos + i as u64,
             severity_number: 9,
             severity_text: "INFO".to_string(),
             trace_id: vec![(i % 251) as u8 + 1; 16],
@@ -2876,6 +2883,46 @@ async fn services_from_beyond_max_lookback_is_clamped() {
     );
 }
 
+// JEF-546: same clamp as JEF-532's query_traces, extended to /api/logs — an
+// explicit `from` far beyond the max-lookback ceiling must not defeat it, and
+// with no `from` at all a full scan (e.g. an ILIKE search for a rare/absent
+// term) must not walk the whole retention window either.
+#[tokio::test]
+#[serial]
+async fn logs_from_beyond_max_lookback_is_clamped() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let day = 86_400.0;
+    // Lower the ceiling to 2 days so the test doesn't need to insert 7 days of
+    // data; #[serial] keeps this process-global env change from racing other
+    // tests.
+    std::env::set_var("WATCHER_MAX_QUERY_HOURS", "48");
+    insert_log_at(&pool, "recent", 3600.0).await; // 1h ago
+    insert_log_at(&pool, "old", 10.0 * day).await; // 10 days ago
+    let router = app(pool);
+
+    // Ask for a window starting 30 days ago — far past the 48h ceiling.
+    let from = (chrono::Utc::now() - chrono::Duration::days(30))
+        .to_rfc3339_opts(chrono::SecondsFormat::Secs, true);
+    let (status, logs) = get_json(&router, &format!("/api/logs?from={from}")).await;
+    std::env::remove_var("WATCHER_MAX_QUERY_HOURS");
+
+    assert_eq!(status, StatusCode::OK);
+    let services: Vec<_> = logs
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(|l| l["service"].as_str().unwrap().to_string())
+        .collect();
+    assert_eq!(
+        services,
+        vec!["recent"],
+        "the 10-day-old log is outside the max-lookback ceiling and must stay \
+         excluded even though `from` asked for 30 days back"
+    );
+}
+
 #[tokio::test]
 #[serial]
 async fn logs_attribute_filter() {