Seek log/metric-bucket queries by key range instead of scanning

Log reads (logs_since, the unsorted Logs query path) now seek the
IngestedLog range directly from since_sequence/before_sequence instead
of scanning from IndexKey::Start and filtering every prior entry out
in a closure. Metric-comparison filters and bucket_values now seek to
a single metric's Bucket key range instead of scanning every declared
metric's buckets. evaluate_promql sits on top of bucket_values, so its
per-poll dashboard evaluation is narrowed by the same fix without
further changes; caching prior evaluations across polls is left for a
follow-up if profiling ever shows it's needed (per ticket 081's own
scope, not built speculatively).

Resolves tickets 077, 078, 081.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TeyhRfMYjMditek6geDRoQ

src/engine/snapshot/metrics.rs

@@ -298,8 +298,18 @@ impl EngineSnapshot {
         labels_filter: Option<&'a BTreeMap<String, MetricPrimitive>>,
     ) -> impl Iterator<Item = (&'a BTreeMap<String, MetricPrimitive>, &'a (u64, Timestamp, MetricPrimitive))> + 'a
     {
+        // IndexKey::Bucket sorts by (metric_name, labels), so bound the range to just this
+        // metric's buckets instead of ranging MetricBucketSplit..End over every declared
+        // metric. '\u{10FFFF}' is the maximum Unicode scalar value, so `metric_name` followed
+        // by it sorts after every real key for this metric (any labels) and before any key
+        // for a lexicographically greater metric name.
+        let lower = IndexKey::Bucket { metric_name: metric_name.to_string(), labels: BTreeMap::new() };
+        let upper = IndexKey::Bucket {
+            metric_name: format!("{metric_name}\u{10FFFF}"),
+            labels: BTreeMap::new(),
+        };
         self.index
-            .range(IndexKey::MetricBucketSplit..IndexKey::End)
+            .range(lower..upper)
             .filter_map(move |(key, value)| match (key, value) {
                 (IndexKey::Bucket { metric_name: name, labels }, DataValue::BucketValue { values })
                     if name == metric_name && labels_filter.is_none_or(|filter| filter == labels) =>

src/engine/snapshot/mod.rs

@@ -157,13 +157,41 @@ impl EngineSnapshot {
     }
 
     pub(crate) fn logs_range(&self) -> impl Iterator<Item = (u64, Timestamp, &str)> {
-        self.index.range(IndexKey::Start..IndexKey::LogParamSplit).filter_map(|(key, value)| {
-            match (key, value) {
-                (IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text }) => {
-                    Some((*sequence, *timestamp, text.as_str()))
-                }
-                _ => None,
+        self.logs_in_sequence_range(..)
+    }
+
+    /// Seeks directly to `sequence_range` within the `IngestedLog` key range instead of
+    /// scanning from `IndexKey::Start` and filtering after the fact. `IngestedLog` sorts by
+    /// `(sequence, timestamp)`, and `sequence` is assigned monotonically at ingest
+    /// (`handle.rs`), so a bound on `sequence` alone is an exact seek — `Timestamp::MIN` is
+    /// used as the synthetic sub-key since it sorts below any real timestamp for that sequence.
+    pub(crate) fn logs_in_sequence_range(
+        &self,
+        sequence_range: impl std::ops::RangeBounds<u64>,
+    ) -> impl Iterator<Item = (u64, Timestamp, &str)> {
+        let lower = match sequence_range.start_bound() {
+            std::ops::Bound::Included(&sequence) => {
+                IndexKey::IngestedLog { sequence, timestamp: Timestamp::MIN }
             }
+            std::ops::Bound::Excluded(&sequence) => {
+                IndexKey::IngestedLog { sequence: sequence.saturating_add(1), timestamp: Timestamp::MIN }
+            }
+            std::ops::Bound::Unbounded => IndexKey::Start,
+        };
+        let upper = match sequence_range.end_bound() {
+            std::ops::Bound::Included(&sequence) => {
+                IndexKey::IngestedLog { sequence: sequence.saturating_add(1), timestamp: Timestamp::MIN }
+            }
+            std::ops::Bound::Excluded(&sequence) => {
+                IndexKey::IngestedLog { sequence, timestamp: Timestamp::MIN }
+            }
+            std::ops::Bound::Unbounded => IndexKey::LogParamSplit,
+        };
+        self.index.range(lower..upper).filter_map(|(key, value)| match (key, value) {
+            (IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text }) => {
+                Some((*sequence, *timestamp, text.as_str()))
+            }
+            _ => None,
         })
     }
 
@@ -173,10 +201,8 @@ impl EngineSnapshot {
         since_ms: i64,
         now_ms: i64,
     ) -> impl Iterator<Item = (u64, Timestamp, &str)> {
-        self.logs_range().filter(move |(sequence, timestamp, _)| {
-            *sequence < before_sequence
-                && timestamp.as_millis() > since_ms
-                && timestamp.as_millis() <= now_ms
+        self.logs_in_sequence_range(..before_sequence).filter(move |(_, timestamp, _)| {
+            timestamp.as_millis() > since_ms && timestamp.as_millis() <= now_ms
         })
     }
 }

src/engine/snapshot/query.rs

@@ -56,11 +56,14 @@ pub(super) fn handle_query_inner(
             }
             let limit = params.limit.unwrap_or(500).min(10_000);
             let compiled_params = snapshot.compiled_param_declarations();
+            let since_bound = match params.since_sequence {
+                Some(since) => std::ops::Bound::Excluded(since),
+                None => std::ops::Bound::Unbounded,
+            };
             let filtered = snapshot
-                .logs_range()
-                .filter(|(sequence, timestamp, _)| {
-                    params.since_sequence.is_none_or(|since| *sequence > since)
-                        && params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
+                .logs_in_sequence_range((since_bound, std::ops::Bound::Unbounded))
+                .filter(|(_, timestamp, _)| {
+                    params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
                         && params.end_ms.is_none_or(|end| timestamp.as_millis() <= end)
                 })
                 .filter(|(sequence, timestamp, text)| {
@@ -80,6 +83,10 @@ pub(super) fn handle_query_inner(
                     })
                     .collect(),
                 Some(sort_by) => {
+                    // Not range-bounded by since_sequence/start_ms: SortHelperKey::WithParam
+                    // sorts by the param's value first (that's the point — it's what gives
+                    // sort-by-param order), so sequence/time bounds can't narrow a BTreeMap
+                    // range seek here the way they do for the unsorted path above.
                     static EMPTY_HELPER: std::sync::LazyLock<BTreeMap<SortHelperKey, Timestamp>> =
                         std::sync::LazyLock::new(BTreeMap::new);
                     let helper_index =