perf(aglake): id lookups fan out, because every search term costs the same

.github/workflows/ci.yml

@@ -139,6 +139,12 @@ jobs:
         # twelve hours or a daemon restart. Getting the retry-after-401 wrong
         # is a backend that reads fine all day and then stops.
         #
+        # The concurrency rules are about load Heron puts on the daemon: the
+        # id-chunked reads fan out and rely on a semaphore to bound it. A
+        # semaphore that admitted everyone would pass every other test and show
+        # up only as an unattributable load spike in production, so the tests
+        # assert the in-flight high-water mark against a mock that measures it.
+        #
         # The live half of the suite self-skips without AGLAKE_TEST_URL and is
         # covered by the staging job instead.
         working-directory: server
@@ -147,6 +153,7 @@ jobs:
         run: |
           cargo test -p h-storage-aglake retry_tests --quiet
           cargo test -p h-storage-aglake auth_tests --quiet
+          cargo test -p h-storage-aglake concurrency_tests --quiet
 
       - name: cargo test (schema-migration golden DBs)
         # Locks in the auto-migration code in h-storage-duckdb/src/schema.rs

CHANGELOG.md

@@ -6,6 +6,54 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
 
+### Performance
+
+- **Opening a large agent turn, and the services topology graph, were spending
+  almost all of their time in one search.** aglake charges a fixed cost per
+  search-position term, linear in the number of terms and independent of how
+  many events match, so the no-JOIN read pattern — fetch the parent, then
+  `id IN (?, ?, …)` for its children — bought nothing from batching the ids into
+  one query and paid for every one of them serially. Measured on a production
+  instance: a 376-call turn took 15.7 s to open, of which 14.1 s was the single
+  bodies lookup; `/api/services/topology` took 6.0–6.5 s on **every** call, all
+  of it one `id IN (…)` carrying one term per turn in the window. The cost is
+  not the payload — the same terms with `| stats count` and no rows returned
+  still took 12.8 s, while a term-free scan of the same window took 302 ms.
+
+  Those lookups now split into smaller chunks (`ID_CHUNK` 512 → 64) and run the
+  chunks concurrently instead of in a `for` loop, bounded by
+  `storage.aglake.max_concurrent_searches`. On the same production data the
+  bodies lookup goes 14.1 s → 4.1 s at the default limit of 8 and the topology
+  lookup 7.1 s → 1.2 s; end to end, a 418-call turn opens in 2.6 s instead of
+  14.6 s and the topology graph answers in 1.7 s instead of 7.5 s.
+
+  64 is a measured optimum in both directions: fewer, bigger searches leave the
+  permits idle, and more, smaller ones pay aglaked's per-search overhead more
+  times than the terms save — chunking below ~16 is slower than not chunking at
+  all. It also pairs with the default concurrency, since 8 × 64 covers any turn
+  observed so far in one wave.
+
+  Upstream has since fixed the per-term cost on unsealed buckets, where it was
+  quadratic per event (`sglog-ystd`, prompted by this investigation). That is
+  worth ~4× on its own and does **not** make the fan-out redundant: measured
+  against a daemon carrying the fix, the concurrent chunks are still 3.5× faster
+  than one search, and 64 is still the optimum.
+
+  The `heron_traces` fan-out in the session list keeps the large chunk: it
+  matches many rows per id rather than one, and its `| head max_sessions_scan`
+  budget is written per search, so splitting the id set would have quietly
+  multiplied that budget by the number of chunks.
+
+### Fixed
+
+- **`storage.aglake.max_concurrent_searches` was accepted and ignored.** It is
+  now the real limit on searches Heron has in flight, enforced on the one method
+  every read path goes through — so it bounds the total rather than being a
+  per-query number that k concurrent console requests multiply by k. `0` is
+  clamped to 1 (a zero-permit budget would block the first read forever) and
+  reported by `heron config validate`.
+
+
 ## [0.8.1] — 2026-09-12
 
 `v0.8.0` was tagged from this same content but never released: the

server/Cargo.toml

@@ -28,6 +28,9 @@ opt-level = 1
 # Async runtime
 tokio = { version = "1", features = ["rt-multi-thread", "macros", "signal", "sync", "time"] }
 tokio-util = { version = "0.7", features = ["rt"] }
+# `try_join_all` over a chunked id lookup, and the pcap-extract stream combinators.
+# tokio has no join-all for a runtime-sized set of futures that borrow `&self`.
+futures = "0.3"
 
 # Serialization
 serde = { version = "1", features = ["derive"] }

server/config/default.toml

@@ -185,7 +185,16 @@ path = "data/heron.duckdb"
 # search_timeout_secs    = 60
 # max_page_offset        = 100000    # deep paging errors out rather than running for minutes
 # max_sessions_scan      = 200000
-# max_concurrent_searches = 8   # NOT YET IMPLEMENTED — accepted and ignored
+# max_concurrent_searches = 8   # searches in flight at once, across ALL read paths
+#   An id lookup (a turn's calls, its bodies, the topology graph) is split into
+#   64-id searches that run concurrently, because an N-id lookup costs N term
+#   probes however it is spelled: one search carrying 400 ids took 6.4 s where
+#   seven carrying 64 each took 0.9 s. This bounds that fan-out.
+#   8 x 64 = up to 512 ids in a single wave, which covers any turn seen so far.
+#   The default also matches aglaked's own default scan pool (`--search-threads 0`
+#   = half the cores, capped at 8). Raising it measured faster still, with no
+#   change in aglaked's RSS, but it oversubscribes that pool; raise
+#   `--search-threads` alongside it on a search-heavy box.
 # trace_time_skew_hours  = 24        # widens end-time windows; a trace's _time is its start
 # metrics_dedup          = false     # only needed if duplicate metric rows are observed;
 #                                    #   costs a sort and the columnar fast path

server/h-common/src/config.rs

@@ -879,8 +879,18 @@ pub struct AglakeConfig {
     /// session in the window before it can page.
     #[serde(default = "default_aglake_max_sessions_scan")]
     pub max_sessions_scan: u64,
-    /// Concurrency limit for the multi-request read paths (id-chunked point
-    /// lookups, the three-step session list).
+    /// Ceiling on searches Heron has in flight against aglaked at once.
+    ///
+    /// A global cap, not a per-query one: the id-chunked point lookups hand
+    /// every chunk to the search client at once and this decides how many run,
+    /// so k concurrent console requests share the budget instead of
+    /// multiplying it. `0` is treated as `1` — see
+    /// [`ConfigIssue::AglakeZeroConcurrentSearches`].
+    ///
+    /// The default matches aglaked's own default scan pool (`--search-threads
+    /// 0` = half the logical cores, capped at 8). Raising it past that pool
+    /// still measured faster — the per-search work outside the scan path
+    /// overlaps — but it is oversubscription, so it is opt-in.
     #[serde(default = "default_aglake_max_concurrent_searches")]
     pub max_concurrent_searches: usize,
     /// A trace's `_time` is its start; queries that filter on end time widen
@@ -1321,6 +1331,12 @@ pub enum ConfigIssue {
     /// nothing to log in as and the password is dead config. Only emitted when
     /// `storage.backend == "aglake"`.
     AglakePasswordWithoutUsername,
+    /// `storage.aglake.max_concurrent_searches = 0`. Read as "no limit" it
+    /// would be a deadlock on the first query, so the backend clamps it to 1 —
+    /// which is a real setting, just an unusually slow one, and worth saying
+    /// out loud rather than leaving the operator to wonder why reads serialized.
+    /// Only emitted when `storage.backend == "aglake"`.
+    AglakeZeroConcurrentSearches,
 }
 
 impl ConfigIssue {
@@ -1343,6 +1359,8 @@ impl ConfigIssue {
             // that has users — and Heron cannot tell which from here, so it
             // warns rather than blocking `config validate`.
             | Self::AglakePasswordWithoutUsername
+            // Clamped to 1, so reads still work — slowly.
+            | Self::AglakeZeroConcurrentSearches
             | Self::AglakeBodyRetentionExceedsParent { .. } => IssueSeverity::Warn,
             Self::DuplicatePipelineName(_)
             | Self::DuplicateSourceId { .. }
@@ -1484,6 +1502,15 @@ impl std::fmt::Display for ConfigIssue {
                  but only because the credential is ignored. Set username, or \
                  remove password."
             ),
+            Self::AglakeZeroConcurrentSearches => write!(
+                f,
+                "storage.aglake.max_concurrent_searches is 0, which is not \
+                 'unlimited' — a zero-permit budget would block the first read \
+                 forever, so it is clamped to 1. Every id-chunked lookup then \
+                 runs its chunks one at a time; opening a large agent turn is \
+                 several times slower. Remove the key to get the default, or \
+                 set the concurrency you want."
+            ),
             Self::LegacyBackendName { found, use_instead } => write!(
                 f,
                 "'{found}' is what the storage backend was called before the \
@@ -1715,6 +1742,9 @@ impl AppConfig {
             if !sg.password.is_empty() && sg.username.is_empty() {
                 issues.push(ConfigIssue::AglakePasswordWithoutUsername);
             }
+            if sg.max_concurrent_searches == 0 {
+                issues.push(ConfigIssue::AglakeZeroConcurrentSearches);
+            }
             // Deliberately not `|| !password.is_empty()`: a password with no
             // username establishes no session, so the loopback risk is
             // unchanged and both findings are reported together on purpose.
@@ -2950,6 +2980,50 @@ mod phase2_tests {
         }
     }
 
+    /// `max_concurrent_searches = 0` is the setting that reads as "no limit"
+    /// and is not one. The backend clamps it to 1, which keeps reads working,
+    /// so the only way an operator learns their reads are serialized is this
+    /// finding.
+    #[test]
+    fn a_zero_search_concurrency_is_reported() {
+        let cfg = |extra: &str| {
+            AppConfig::from_toml(&format!(
+                r#"
+                [[pipeline]]
+                name = "p"
+                [[pipeline.sources]]
+                type = "pcap"
+                interface = "eth0"
+
+                [storage]
+                backend = "aglake"
+
+                [storage.aglake]
+                {extra}
+                "#
+            ))
+        };
+
+        let issues = cfg("max_concurrent_searches = 0").validate();
+        let found: Vec<_> = issues
+            .iter()
+            .filter(|i| matches!(i, ConfigIssue::AglakeZeroConcurrentSearches))
+            .collect();
+        assert_eq!(found.len(), 1);
+        assert_eq!(found[0].severity(), IssueSeverity::Warn);
+
+        // A real limit, and the default, are both silent.
+        for extra in ["max_concurrent_searches = 1", ""] {
+            assert!(
+                !cfg(extra)
+                    .validate()
+                    .iter()
+                    .any(|i| matches!(i, ConfigIssue::AglakeZeroConcurrentSearches)),
+                "unexpected finding for {extra:?}"
+            );
+        }
+    }
+
     /// Only when aglake is the active backend — an unused `[storage.aglake]`
     /// block is not a finding.
     #[test]

server/h-pcap-extract/Cargo.toml

@@ -11,7 +11,7 @@ snap = "1"
 bytes = { workspace = true }
 tokio = { workspace = true }
 tokio-util = { workspace = true }
-futures = "0.3"
+futures.workspace = true
 tracing = { workspace = true }
 thiserror = { workspace = true }
 

server/h-storage-aglake/Cargo.toml

@@ -12,6 +12,8 @@ h-protocol.workspace = true
 h-turn.workspace = true
 h-storage.workspace = true
 reqwest.workspace = true
+# `try_join_all`, so the id-chunked point lookups overlap instead of queueing.
+futures.workspace = true
 # Request-body gzip. reqwest's `gzip` feature only covers *response*
 # decompression, so the HEC request bodies are compressed here.
 flate2.workspace = true