From ab54986d318f2dde0476bf22d335c35ae6ad7fee Mon Sep 17 00:00:00 2001 From: Vader Yang Date: Mon, 14 Sep 2026 01:09:06 +0800 Subject: [PATCH 1/2] perf(aglake): id lookups fan out, because every search term costs the same MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit aglake charges a fixed cost per search-position term — linear in the term count, independent of how many events match — so the no-JOIN read pattern paid for every id serially. Batching the ids into one `id IN (…)` was the slowest available shape, not the cheapest. Measured on a production instance: opening a 376-call turn took 15.7 s, of which 14.1 s was the single bodies lookup, and the services topology graph took 6.0-6.5 s on every call. Not payload — the same terms with `| stats count` and no rows returned still took 12.8 s, where a term-free scan of the same window took 302 ms. So ID_CHUNK drops 512 -> 32 and the chunks run concurrently rather than in a `for` loop. 14.1 s -> 4.7 s for the bodies lookup at the default limit, 7.1 s -> 1.2 s for topology. Chunking further is slower: at 4 ids per search the per-search overhead dominates and the bodies lookup regresses to 11.2 s, so the constant is a measured optimum rather than a minimum. max_concurrent_searches stops being accepted-and-ignored and becomes that bound, enforced on the one method every read path goes through so it limits the total instead of being a per-query number that k concurrent console requests multiply by k. 0 clamps to 1 and is reported by `config validate`, because read as "unlimited" it would block the first read forever. The session-list fan-out keeps the large chunk on purpose: 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 multiplied that budget by the number of chunks. Tests: a mock that measures its own in-flight high-water mark (a no-op semaphore passes every other assertion and fails only that one), and a live turn of 2*ID_CHUNK+1 spans — every existing test used three, one chunk, so the split was never taken. Verified both fail without the change: dropping a chunk loses span-0032, the first id of the second chunk. Claude-Session: https://claude.ai/code/session_01GxGWpo3L4BRMkwtoa9X54W --- .github/workflows/ci.yml | 7 + CHANGELOG.md | 37 +++ server/Cargo.lock | 31 +-- server/Cargo.toml | 3 + server/config/default.toml | 10 +- server/h-common/src/config.rs | 78 ++++++- server/h-pcap-extract/Cargo.toml | 2 +- server/h-storage-aglake/Cargo.toml | 2 + server/h-storage-aglake/src/calls.rs | 47 ++-- server/h-storage-aglake/src/client.rs | 286 ++++++++++++++++++++++++ server/h-storage-aglake/src/it.rs | 85 +++++++ server/h-storage-aglake/src/services.rs | 17 +- server/h-storage-aglake/src/sessions.rs | 2 +- server/h-storage-aglake/src/spl.rs | 41 +++- 14 files changed, 604 insertions(+), 44 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7c916d31..597ccfba 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/CHANGELOG.md b/CHANGELOG.md index 095ebbd5..bcc59b62 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,43 @@ 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 much smaller chunks (`ID_CHUNK` 512 → 32) 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.7 s at the default limit of 8, and 3.0 s at 16; + the topology lookup 7.1 s → 1.2 s. Chunking further is *slower* — at 4 ids per + search the per-search overhead dominates and the bodies lookup regresses to + 11.2 s — so the constant is a measured optimum, not a minimum. + + 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 diff --git a/server/Cargo.lock b/server/Cargo.lock index 911c176a..54a5278a 100644 --- a/server/Cargo.lock +++ b/server/Cargo.lock @@ -1361,7 +1361,7 @@ dependencies = [ [[package]] name = "h-api" -version = "0.8.0" +version = "0.8.1" dependencies = [ "axum", "chrono", @@ -1385,7 +1385,7 @@ dependencies = [ [[package]] name = "h-capture" -version = "0.8.0" +version = "0.8.1" dependencies = [ "async-trait", "aya", @@ -1406,7 +1406,7 @@ dependencies = [ [[package]] name = "h-common" -version = "0.8.0" +version = "0.8.1" dependencies = [ "config", "serde", @@ -1420,11 +1420,11 @@ dependencies = [ [[package]] name = "h-ebpf-common" -version = "0.8.0" +version = "0.8.1" [[package]] name = "h-export" -version = "0.8.0" +version = "0.8.1" dependencies = [ "h-llm", "serde", @@ -1434,7 +1434,7 @@ dependencies = [ [[package]] name = "h-llm" -version = "0.8.0" +version = "0.8.1" dependencies = [ "bitflags 2.11.0", "bytes", @@ -1454,7 +1454,7 @@ dependencies = [ [[package]] name = "h-metrics" -version = "0.8.0" +version = "0.8.1" dependencies = [ "h-common", "h-llm", @@ -1465,7 +1465,7 @@ dependencies = [ [[package]] name = "h-pcap-extract" -version = "0.8.0" +version = "0.8.1" dependencies = [ "bytes", "futures", @@ -1482,7 +1482,7 @@ dependencies = [ [[package]] name = "h-protocol" -version = "0.8.0" +version = "0.8.1" dependencies = [ "bytemuck", "bytes", @@ -1500,7 +1500,7 @@ dependencies = [ [[package]] name = "h-storage" -version = "0.8.0" +version = "0.8.1" dependencies = [ "async-trait", "bytes", @@ -1519,11 +1519,12 @@ dependencies = [ [[package]] name = "h-storage-aglake" -version = "0.8.0" +version = "0.8.1" dependencies = [ "async-trait", "bytes", "flate2", + "futures", "h-common", "h-llm", "h-metrics", @@ -1541,7 +1542,7 @@ dependencies = [ [[package]] name = "h-storage-clickhouse" -version = "0.8.0" +version = "0.8.1" dependencies = [ "async-trait", "bytes", @@ -1561,7 +1562,7 @@ dependencies = [ [[package]] name = "h-storage-duckdb" -version = "0.8.0" +version = "0.8.1" dependencies = [ "async-trait", "bytes", @@ -1581,7 +1582,7 @@ dependencies = [ [[package]] name = "h-turn" -version = "0.8.0" +version = "0.8.1" dependencies = [ "h-capture", "h-common", @@ -1690,7 +1691,7 @@ checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" [[package]] name = "heron" -version = "0.8.0" +version = "0.8.1" dependencies = [ "axum", "clap", diff --git a/server/Cargo.toml b/server/Cargo.toml index 0e4ded6b..1a213d08 100644 --- a/server/Cargo.toml +++ b/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"] } diff --git a/server/config/default.toml b/server/config/default.toml index 2e77f74d..9392c146 100644 --- a/server/config/default.toml +++ b/server/config/default.toml @@ -185,7 +185,15 @@ 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 +# `ID_CHUNK`-sized searches that run concurrently, because aglaked charges a +# fixed cost per search term: one search carrying 376 ids took 14.6 s where +# twelve carrying 32 each took 4.7 s. This bounds that fan-out. +# The default matches aglaked's own default scan pool (`--search-threads 0` = +# half the cores, capped at 8). Raising it measured faster still (16 -> 3.0 s +# on the same lookup, 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 diff --git a/server/h-common/src/config.rs b/server/h-common/src/config.rs index c81dea88..b5414551 100644 --- a/server/h-common/src/config.rs +++ b/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] diff --git a/server/h-pcap-extract/Cargo.toml b/server/h-pcap-extract/Cargo.toml index 250160b5..2d5e9731 100644 --- a/server/h-pcap-extract/Cargo.toml +++ b/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 } diff --git a/server/h-storage-aglake/Cargo.toml b/server/h-storage-aglake/Cargo.toml index bed3f4cd..6f60f437 100644 --- a/server/h-storage-aglake/Cargo.toml +++ b/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 diff --git a/server/h-storage-aglake/src/calls.rs b/server/h-storage-aglake/src/calls.rs index 7b8b92bd..d4807bef 100644 --- a/server/h-storage-aglake/src/calls.rs +++ b/server/h-storage-aglake/src/calls.rs @@ -11,6 +11,7 @@ use std::collections::{HashMap, HashSet}; +use futures::future::try_join_all; use h_common::error::Result; use h_common::process::ProcessInfo; use h_llm::model::LlmCall; @@ -280,26 +281,32 @@ impl AglakeBackend { if span_ids.is_empty() { return Ok(Vec::new()); } - let mut events: Vec = Vec::with_capacity(span_ids.len()); - for chunk in span_ids.chunks(ID_CHUNK) { + // Chunks run concurrently. The whole point of a small `ID_CHUNK` is + // that N id probes overlap instead of queueing; `SearchClient`'s + // permits are what bound the fan-out. + let window = window.as_ref().map(|(e, l)| (e.as_str(), l.as_str())); + let per_chunk = span_ids.chunks(ID_CHUNK).map(|chunk| async move { let ix = &self.ix.spans; let Some(list) = in_list("id", chunk) else { - continue; + return Result::>::Ok(Vec::new()); }; let search = format!("search index={ix} sourcetype={ST_SPAN} {list}"); - let found: Vec = match &window { + match window { Some((earliest, latest)) => { self.fetch_raw("read_spans_by_ids", &search, chunk.len(), earliest, latest) - .await? + .await } // No caller window: bound by the ids themselves, unbounded on // a miss. `chunk[0]` is representative — a trace's calls are // minted within seconds of each other, well inside the skew. None => { self.fetch_raw_by_id("read_spans_by_ids", &search, chunk.len(), &chunk[0]) - .await? + .await } - }; + } + }); + let mut events: Vec = Vec::with_capacity(span_ids.len()); + for found in try_join_all(per_chunk).await? { events.extend(found); } @@ -425,22 +432,32 @@ impl AglakeBackend { span_ids: &[String], window: (String, String), ) -> Result> { - let mut out = HashMap::with_capacity(span_ids.len()); let (earliest, latest) = window; - for chunk in span_ids.chunks(ID_CHUNK) { + // The dominant cost of opening a large turn, and the reason the chunks + // are concurrent: bodies are matched by raw term, one term per span, + // over the largest index Heron writes. + let (earliest, latest) = (earliest.as_str(), latest.as_str()); + let per_chunk = span_ids.chunks(ID_CHUNK).map(|chunk| async move { let ix = &self.ix.bodies; let Some(terms) = spl::body_terms(chunk) else { - continue; + return Result::>::Ok(Vec::new()); }; let search = format!("search index={ix} sourcetype={ST_BODY} {terms}"); - let wanted: HashSet<&str> = chunk.iter().map(String::as_str).collect(); let found: Vec = self - .fetch_raw("fetch_bodies", &search, chunk.len(), &earliest, &latest) + .fetch_raw("fetch_bodies", &search, chunk.len(), earliest, latest) .await?; + // A raw-term match can land on an event that merely quotes the id, + // so the decoded `span_id` is checked against this chunk's set. + let wanted: HashSet<&str> = chunk.iter().map(String::as_str).collect(); + Ok(found + .into_iter() + .filter(|b| wanted.contains(b.span_id.as_str())) + .collect()) + }); + let mut out = HashMap::with_capacity(span_ids.len()); + for found in try_join_all(per_chunk).await? { for b in found { - if wanted.contains(b.span_id.as_str()) { - out.insert(b.span_id.clone(), b); - } + out.insert(b.span_id.clone(), b); } } Ok(out) diff --git a/server/h-storage-aglake/src/client.rs b/server/h-storage-aglake/src/client.rs index 9501ca40..20c4caa3 100644 --- a/server/h-storage-aglake/src/client.rs +++ b/server/h-storage-aglake/src/client.rs @@ -15,6 +15,7 @@ use std::time::Duration; use h_common::config::AglakeConfig; use h_common::error::{AppError, Result}; use serde::Deserialize; +use tokio::sync::Semaphore; fn err(ctx: &str, e: E) -> AppError { AppError::Storage(format!("aglake {ctx}: {e}")) @@ -762,6 +763,16 @@ pub(crate) struct SearchClient { auth: Arc, endpoint: String, ping_url: String, + /// Global cap on searches in flight, from + /// [`AglakeConfig::max_concurrent_searches`]. + /// + /// It lives here, on the one method every read path goes through, rather + /// than in each fan-out loop. The loops then need no limit of their own: + /// they hand every chunk to `search()` at once and the permits decide how + /// many run. That also makes the cap mean what its name says — a bound on + /// what Heron asks of the daemon, not a per-query bound that k concurrent + /// console requests multiply by k. + permits: Arc, } impl SearchClient { @@ -776,6 +787,10 @@ impl SearchClient { auth, endpoint: format!("{base}/api/v1/search"), ping_url: format!("{base}/api/v1/indexes"), + // `max(1)` because a zero-permit semaphore is not "no limit", it is + // a deadlock on the first read. Config validation warns about the + // 0 separately; this is the half that keeps the process useful. + permits: Arc::new(Semaphore::new(config.max_concurrent_searches.max(1))), }) } @@ -789,6 +804,14 @@ impl SearchClient { latest: &str, ) -> Result { let body = serde_json::json!({ "q": spl, "earliest": earliest, "latest": latest }); + // Held across the whole round-trip, response body included: the point + // is to bound concurrent work on the daemon, and a search that has + // returned headers is still streaming rows out of it. + let _permit = self + .permits + .acquire() + .await + .map_err(|e| err("search", format!("search permits closed: {e}")))?; let resp = send_authenticated(&self.auth, "search", |token| { with_token(self.http.post(&self.endpoint).json(&body), token) }) @@ -2131,3 +2154,266 @@ mod auth_tests { assert_eq!(parse("other=abc123; Path=/"), None); } } + +/// `max_concurrent_searches`, which is the whole reason the id lookups may fan +/// out freely. +/// +/// The chunked reads hand every chunk to [`SearchClient::search`] at once and +/// rely on this budget to decide how many actually run. So the limit is not a +/// tuning nicety — it is the only thing standing between a 10,000-turn topology +/// window and 313 simultaneous searches against the daemon. A semaphore that +/// silently admitted everyone would look identical in every functional test, +/// and show up in production as a load spike nobody could attribute. +#[cfg(test)] +mod concurrency_tests { + use super::*; + use std::io::{BufRead, BufReader, Read, Write}; + use std::net::{SocketAddr, TcpListener}; + use std::sync::atomic::{AtomicUsize, Ordering}; + + /// A mock that answers every search the same way, slowly, while recording + /// the high-water mark of requests it was handling at once. + /// + /// The delay is the measuring instrument: without it each request would be + /// served before the next arrived and the gauge would read 1 whatever the + /// limit was. + struct CountingAglake { + addr: SocketAddr, + peak: Arc, + total: Arc, + } + + impl CountingAglake { + fn start(hold: Duration) -> Self { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); + let addr = listener.local_addr().unwrap(); + let in_flight = Arc::new(AtomicUsize::new(0)); + let peak = Arc::new(AtomicUsize::new(0)); + let total = Arc::new(AtomicUsize::new(0)); + let (bg_in, bg_peak, bg_total) = ( + Arc::clone(&in_flight), + Arc::clone(&peak), + Arc::clone(&total), + ); + + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { break }; + let (bg_in, bg_peak, bg_total) = ( + Arc::clone(&bg_in), + Arc::clone(&bg_peak), + Arc::clone(&bg_total), + ); + std::thread::spawn(move || { + // Drain the request first. Counting at accept() time + // would count a connection the client has not yet + // spent a permit on. + let mut reader = BufReader::new(stream.try_clone().unwrap()); + let mut line = String::new(); + if reader.read_line(&mut line).is_err() { + return; + } + let mut len = 0usize; + loop { + let mut h = String::new(); + if reader.read_line(&mut h).is_err() || h.trim().is_empty() { + break; + } + if let Some(v) = h.to_ascii_lowercase().strip_prefix("content-length:") { + len = v.trim().parse().unwrap_or(0); + } + } + let mut body = vec![0u8; len]; + let _ = reader.read_exact(&mut body); + + let now = bg_in.fetch_add(1, Ordering::SeqCst) + 1; + bg_peak.fetch_max(now, Ordering::SeqCst); + bg_total.fetch_add(1, Ordering::SeqCst); + std::thread::sleep(hold); + bg_in.fetch_sub(1, Ordering::SeqCst); + + let payload = r#"{"mode":"results","rows":[]}"#; + let _ = stream.write_all( + format!( + "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n\ + Content-Length: {}\r\n\r\n{payload}", + payload.len() + ) + .as_bytes(), + ); + }); + } + }); + Self { addr, peak, total } + } + + fn config(&self, max_concurrent_searches: usize) -> AglakeConfig { + AglakeConfig { + url: format!("http://{}", self.addr), + request_timeout_secs: 10, + search_timeout_secs: 10, + max_concurrent_searches, + ..Default::default() + } + } + + fn peak(&self) -> usize { + self.peak.load(Ordering::SeqCst) + } + + fn total(&self) -> usize { + self.total.load(Ordering::SeqCst) + } + } + + fn searcher(config: &AglakeConfig) -> SearchClient { + let auth = Arc::new(AuthState::new(config).unwrap()); + SearchClient::new(config, auth).unwrap() + } + + async fn fire(client: &SearchClient, n: usize) { + let queries: Vec = (0..n).map(|i| format!("search index=x | head {i}")).collect(); + let all = queries.iter().map(|q| client.search(q, "0", "0")); + futures::future::try_join_all(all).await.unwrap(); + } + + /// The budget is honoured, and it is honoured at the configured number + /// rather than at some accidental smaller one. + #[tokio::test] + async fn concurrent_searches_are_capped_at_the_configured_limit() { + let mock = CountingAglake::start(Duration::from_millis(120)); + let config = mock.config(3); + let client = searcher(&config); + + fire(&client, 12).await; + + assert_eq!(mock.total(), 12, "every search must still be issued"); + assert_eq!( + mock.peak(), + 3, + "expected exactly the configured 3 in flight, saw {}", + mock.peak() + ); + } + + /// A limit of 1 serializes. Worth its own test: it is the shape that proves + /// the semaphore is load-bearing, since a no-op semaphore passes every + /// assertion about results and fails only this one. + #[tokio::test] + async fn a_limit_of_one_serializes_the_searches() { + let mock = CountingAglake::start(Duration::from_millis(40)); + let config = mock.config(1); + let client = searcher(&config); + + fire(&client, 6).await; + + assert_eq!(mock.total(), 6); + assert_eq!(mock.peak(), 1, "one permit means one search at a time"); + } + + /// `0` is the trap: read as "unlimited" it would be a zero-permit + /// semaphore, and the first read would wait forever. It must clamp to 1 and + /// come back — the process staying useful matters more than honouring a + /// setting that cannot mean what it looks like. `heron config validate` + /// reports it separately. + #[tokio::test] + async fn a_zero_limit_clamps_to_one_instead_of_deadlocking() { + let mock = CountingAglake::start(Duration::from_millis(20)); + let config = mock.config(0); + let client = searcher(&config); + + tokio::time::timeout(Duration::from_secs(10), fire(&client, 3)) + .await + .expect("a zero limit must not block forever"); + + assert_eq!(mock.total(), 3); + assert_eq!(mock.peak(), 1); + } + + /// Permits are released on the failure path too. A 500 that leaked its + /// permit would degrade the backend one search at a time until it stopped + /// reading altogether — a failure that only shows up after the errors have + /// stopped, which is the worst time to start diagnosing it. + #[tokio::test] + async fn a_failed_search_returns_its_permit() { + let mock = FailingAglake::start(2); + let config = AglakeConfig { + url: format!("http://{}", mock.addr), + request_timeout_secs: 10, + search_timeout_secs: 10, + max_concurrent_searches: 1, + ..Default::default() + }; + let client = searcher(&config); + + for _ in 0..2 { + assert!( + client.search("search index=x", "0", "0").await.is_err(), + "the mock is scripted to fail" + ); + } + // With a single permit, a leak on either failure makes this hang rather + // than fail — hence the timeout. + let ok = tokio::time::timeout( + Duration::from_secs(5), + client.search("search index=x", "0", "0"), + ) + .await + .expect("permit leaked: the search after two failures never got one"); + ok.expect("the third response is a success"); + } + + /// Answers `500` the first `fail` times, then `200`. + struct FailingAglake { + addr: SocketAddr, + } + + impl FailingAglake { + fn start(fail: usize) -> Self { + let listener = TcpListener::bind("127.0.0.1:0").expect("bind"); + let addr = listener.local_addr().unwrap(); + let seen = Arc::new(AtomicUsize::new(0)); + std::thread::spawn(move || { + for stream in listener.incoming() { + let Ok(mut stream) = stream else { break }; + let seen = Arc::clone(&seen); + std::thread::spawn(move || { + let mut reader = BufReader::new(stream.try_clone().unwrap()); + let mut line = String::new(); + if reader.read_line(&mut line).is_err() { + return; + } + let mut len = 0usize; + loop { + let mut h = String::new(); + if reader.read_line(&mut h).is_err() || h.trim().is_empty() { + break; + } + if let Some(v) = h.to_ascii_lowercase().strip_prefix("content-length:") { + len = v.trim().parse().unwrap_or(0); + } + } + let mut body = vec![0u8; len]; + let _ = reader.read_exact(&mut body); + + let n = seen.fetch_add(1, Ordering::SeqCst); + let (code, payload) = if n < fail { + (500, r#"{"error":"boom"}"#) + } else { + (200, r#"{"mode":"results","rows":[]}"#) + }; + let _ = stream.write_all( + format!( + "HTTP/1.1 {code} X\r\nContent-Type: application/json\r\n\ + Content-Length: {}\r\n\r\n{payload}", + payload.len() + ) + .as_bytes(), + ); + }); + } + }); + Self { addr } + } + } +} diff --git a/server/h-storage-aglake/src/it.rs b/server/h-storage-aglake/src/it.rs index 3b5911cc..d14f1808 100644 --- a/server/h-storage-aglake/src/it.rs +++ b/server/h-storage-aglake/src/it.rs @@ -268,6 +268,91 @@ async fn unknown_ids_return_none_not_an_error() { .is_empty()); } +/// A turn whose calls do not fit in one `ID_CHUNK`. +/// +/// `read_spans_by_ids` and `fetch_bodies` split an id set into chunks and run +/// them concurrently, and every existing test uses three spans — one chunk, so +/// the split is never taken. The failures this covers are the ones a single +/// chunk cannot express: a chunk whose rows are dropped on the floor, a body +/// map keyed from the wrong chunk's ids, and an ordering that only holds within +/// a chunk because the sort happens before the chunks are merged. +/// +/// The count is `2 * ID_CHUNK + 1` on purpose: two full chunks plus a remainder +/// of one, so an off-by-one in the chunking shows up as a missing span rather +/// than as a rebalanced set that still adds up. +#[tokio::test] +async fn a_turn_larger_than_one_id_chunk_comes_back_whole_and_in_order() { + let backend = require_backend!(); + let n = crate::spl::ID_CHUNK * 2 + 1; + + let mut trace = fixtures::full_trace(); + trace.turn_id = uuid::Uuid::now_v7().to_string(); + let base = trace.start_time_us; + + // Written in reverse so a correct result cannot be insertion order, and + // indexed into the id so the expected order is computable rather than + // observed. + let mut calls = Vec::new(); + for i in (0..n).rev() { + let mut c = fixtures::full_call(); + c.id = format!("{}-span-{i:04}", trace.turn_id); + c.request_time = base + (i as i64) * 1_000; + c.complete_time = Some(c.request_time + 500); + calls.push(c); + } + trace.call_count = n as u32; + trace.span_ids = { + let mut ids: Vec = calls.iter().map(|c| c.id.clone()).collect(); + ids.reverse(); + ids + }; + trace.end_time_us = base + (n as i64) * 1_000; + + backend.write_spans(calls).await.unwrap(); + backend.write_traces(vec![trace.clone()]).await.unwrap(); + + let spans = eventually("all spans of a multi-chunk turn", || async { + let s = backend + .query_trace_spans(&trace.turn_id, true) + .await + .unwrap(); + (s.len() == n).then_some(s) + }) + .await; + + let got: Vec = spans.iter().map(|s| s.id.clone()).collect(); + assert_eq!(got, trace.span_ids, "every chunk, merged in request order"); + assert_eq!( + got.iter().collect::>().len(), + n, + "a chunk fetched twice would duplicate rather than lose" + ); + assert_eq!( + spans.iter().map(|s| s.sequence).collect::>(), + (1..=n as u32).collect::>() + ); + // Bodies are the second, separately-chunked hop over a different index — + // the one that dominates a large turn. Each span must get its own. + for s in &spans { + assert!( + s.request_body.is_some(), + "span {} came back without its body", + s.id + ); + } + + // Same ids by the registry path, which chunks through `read_spans_by_ids` + // without a trace to bound the window. + let by_ids = backend + .query_spans_by_ids(&trace.span_ids, false) + .await + .unwrap(); + assert_eq!( + by_ids.iter().map(|s| s.id.clone()).collect::>(), + trace.span_ids + ); +} + #[tokio::test] async fn trace_round_trips_and_resolves_its_spans() { let backend = require_backend!(); diff --git a/server/h-storage-aglake/src/services.rs b/server/h-storage-aglake/src/services.rs index 42dada4c..faee5254 100644 --- a/server/h-storage-aglake/src/services.rs +++ b/server/h-storage-aglake/src/services.rs @@ -26,6 +26,7 @@ use std::collections::{HashMap, HashSet}; +use futures::future::try_join_all; use h_common::error::{AppError, Result}; use h_storage::query::*; @@ -455,8 +456,10 @@ impl AglakeBackend { ids: &[String], range: &TimeRange, ) -> Result> { - let mut out = HashMap::with_capacity(ids.len()); - for chunk in ids.chunks(spl::ID_CHUNK) { + // One `id` term per turn in the window, so on a busy day this is the + // widest id lookup Heron issues — it was the whole cost of + // `/api/services/topology` before the chunks ran concurrently. + let per_chunk = ids.chunks(spl::ID_CHUNK).map(|chunk| async move { let mut s = Search::new(&self.ix.spans, ST_SPAN); s.any_of("id", chunk); let spl_q = format!( @@ -464,15 +467,17 @@ impl AglakeBackend { s.build(), chunk.len() ); - let rows = self - .search + self.search .search( &spl_q, &spl::epoch_secs(range.start_us), &spl::epoch_secs(range.end_us), ) - .await? - .rows(); + .await + .map(|r| r.rows()) + }); + let mut out = HashMap::with_capacity(ids.len()); + for rows in try_join_all(per_chunk).await? { for r in rows { out.insert( string(&r, "id"), diff --git a/server/h-storage-aglake/src/sessions.rs b/server/h-storage-aglake/src/sessions.rs index 4ce743c9..6cb4cff8 100644 --- a/server/h-storage-aglake/src/sessions.rs +++ b/server/h-storage-aglake/src/sessions.rs @@ -262,7 +262,7 @@ impl AglakeBackend { if session_ids.is_empty() { return Ok(out); } - for chunk in session_ids.chunks(spl::ID_CHUNK) { + for chunk in session_ids.chunks(spl::FANOUT_ID_CHUNK) { let mut s = Search::new(&self.ix.traces, ST_TRACE); s.any_of("session_id", chunk); if let Some(sid) = source_id { diff --git a/server/h-storage-aglake/src/spl.rs b/server/h-storage-aglake/src/spl.rs index 0c9a1ac0..52b74163 100644 --- a/server/h-storage-aglake/src/spl.rs +++ b/server/h-storage-aglake/src/spl.rs @@ -257,9 +257,44 @@ pub(crate) fn epoch_secs(us: i64) -> String { ) } -/// Chunk size for `id IN (...)` point lookups. Keeps a single query string -/// bounded while staying far above the common trace size. -pub(crate) const ID_CHUNK: usize = 512; +/// Chunk size for `id IN (...)` point lookups. +/// +/// Not a query-string bound — a cost bound. aglaked charges a **fixed cost per +/// search-position term**, linear in term count and independent of how many +/// events match, so an N-id lookup is N probes however it is spelled. One +/// search carrying every id is therefore the slowest available shape, and +/// chunking exists to get the probes running concurrently rather than to keep +/// the request small. +/// +/// Measured on production (aglaked 1.5.0, 376 span ids, the bodies index, with +/// [`AglakeConfig::max_concurrent_searches`] = 8): +/// +/// | chunk | searches | wall | +/// |---|---|---| +/// | 512 (one search) | 1 | 14.6 s | +/// | 64 | 6 | 4.1 s | +/// | 32 | 12 | 4.7 s | +/// | 16 | 24 | 4.8 s | +/// | 8 | 47 | 6.9 s | +/// | 4 | 94 | 11.2 s | +/// +/// The curve has a floor, which is why this is 32 and not 1: below ~16 the +/// per-search overhead aglaked pays regardless of term count (opening the +/// window, pruning buckets) starts to dominate, and chunking too finely is +/// slower than not chunking enough. 32 is within noise of the optimum on the +/// spans and traces indexes too, where 16 measured marginally better. +pub(crate) const ID_CHUNK: usize = 32; + +/// Chunk size for id lookups that fan *out* — many rows per id, not one. +/// +/// `session_aggregates` is the only one: it matches turn rows by `session_id`, +/// where a session has tens to thousands of turns. Two reasons it keeps the +/// large chunk. The per-term cost is amortized against real row volume rather +/// than being the whole cost (one page of 100 sessions measures 293 ms end to +/// end), and its `| head max_sessions_scan` row budget is written per search — +/// so splitting the id set into k chunks would quietly multiply the budget by +/// k. +pub(crate) const FANOUT_ID_CHUNK: usize = 512; /// Build the offset-pagination pipeline. /// From bb14d543ca559554482ebc8be9409be2348cc3f2 Mon Sep 17 00:00:00 2001 From: Vader Yang Date: Mon, 14 Sep 2026 11:42:39 +0800 Subject: [PATCH 2/2] perf(aglake): ID_CHUNK 32 -> 64, the optimum on both daemon builds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Upstream fixed the cost this change was measured against. The per-term cost on dictionary-less (unsealed) buckets was quadratic per event — three linear scans in the residual lookup structures, no short-circuit on a miss — and `sglog-ystd` turns them into index lookups. That is worth ~4x on its own. So the constant had to be re-derived against a daemon carrying the fix rather than kept on numbers taken from one without it. 400 span ids against a hot bucket, 8 permits, five interleaved reps on one host: chunk searches 1.5.0.2674 sglog-ystd fixed 400 1 6423 ms 1596 ms 128 4 1670 ms 679 ms 96 5 1254 ms 555 ms 64 7 887 ms 455 ms 48 9 1047 ms 655 ms 32 13 1051 ms 658 ms 64 wins on both, and the optimum is sharp in both directions — fewer, bigger searches leave the permits idle; more, smaller ones pay aglaked's per-search overhead more times than the terms save. 32 was costing 1.19x the optimum on the unfixed daemon and 1.45x on the fixed one. I had this number in the first measurement and waved it off: 64 measured 4137 ms against 32's 4664 on the real bodies hop, and I called 13% "within noise across the three hops" and took the compromise. It was not noise. What the fix does not do is make the fan-out redundant, which was the live question for this branch: concurrency is still worth 3.5x after it, and the shape is unchanged. Per-term cost now falls with term count (4.02 ms/term at 400 against 10.1 at 32), so the marginal term is cheap — but the total still grows with N, and only the fan-out shortens the wall clock. 8 permits x 64 ids also happens to resolve up to 512 ids in one wave, covering every turn seen. Verified by building aglaked from source at the commit carrying the fix and A/B-ing it against the version in production on two copies of the same real index. Workspace 1236 passed; the live suite 141 passed against a current daemon build. Claude-Session: https://claude.ai/code/session_01GxGWpo3L4BRMkwtoa9X54W --- CHANGELOG.md | 23 +++++++++---- server/config/default.toml | 15 ++++---- server/h-storage-aglake/src/spl.rs | 55 ++++++++++++++++++------------ 3 files changed, 58 insertions(+), 35 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index bcc59b62..6269291f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,13 +20,24 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). 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 much smaller chunks (`ID_CHUNK` 512 → 32) and run - the chunks concurrently instead of in a `for` loop, bounded by + 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.7 s at the default limit of 8, and 3.0 s at 16; - the topology lookup 7.1 s → 1.2 s. Chunking further is *slower* — at 4 ids per - search the per-search overhead dominates and the bodies lookup regresses to - 11.2 s — so the constant is a measured optimum, not a minimum. + 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` diff --git a/server/config/default.toml b/server/config/default.toml index 9392c146..93fbbcac 100644 --- a/server/config/default.toml +++ b/server/config/default.toml @@ -187,13 +187,14 @@ path = "data/heron.duckdb" # max_sessions_scan = 200000 # 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 -# `ID_CHUNK`-sized searches that run concurrently, because aglaked charges a -# fixed cost per search term: one search carrying 376 ids took 14.6 s where -# twelve carrying 32 each took 4.7 s. This bounds that fan-out. -# The default matches aglaked's own default scan pool (`--search-threads 0` = -# half the cores, capped at 8). Raising it measured faster still (16 -> 3.0 s -# on the same lookup, with no change in aglaked's RSS), but it oversubscribes -# that pool; raise `--search-threads` alongside it on a search-heavy box. +# 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 diff --git a/server/h-storage-aglake/src/spl.rs b/server/h-storage-aglake/src/spl.rs index 52b74163..064a4014 100644 --- a/server/h-storage-aglake/src/spl.rs +++ b/server/h-storage-aglake/src/spl.rs @@ -259,31 +259,42 @@ pub(crate) fn epoch_secs(us: i64) -> String { /// Chunk size for `id IN (...)` point lookups. /// -/// Not a query-string bound — a cost bound. aglaked charges a **fixed cost per -/// search-position term**, linear in term count and independent of how many -/// events match, so an N-id lookup is N probes however it is spelled. One -/// search carrying every id is therefore the slowest available shape, and -/// chunking exists to get the probes running concurrently rather than to keep -/// the request small. +/// Not a query-string bound — a cost bound. An N-id lookup is N term probes +/// however it is spelled, and the wall clock is what a user waits on, so the +/// chunks exist to get those probes running concurrently. One search carrying +/// every id is the slowest available shape. /// -/// Measured on production (aglaked 1.5.0, 376 span ids, the bodies index, with -/// [`AglakeConfig::max_concurrent_searches`] = 8): +/// 64 is a **measured optimum, and a sharp one**. 400 span ids against a hot +/// bucket, with [`AglakeConfig::max_concurrent_searches`] = 8, five interleaved +/// reps on one host: /// -/// | chunk | searches | wall | -/// |---|---|---| -/// | 512 (one search) | 1 | 14.6 s | -/// | 64 | 6 | 4.1 s | -/// | 32 | 12 | 4.7 s | -/// | 16 | 24 | 4.8 s | -/// | 8 | 47 | 6.9 s | -/// | 4 | 94 | 11.2 s | +/// | chunk | searches | aglaked 1.5.0.2674 | with `sglog-ystd` fixed | +/// |---|---|---|---| +/// | 400 (one search) | 1 | 6423 ms | 1596 ms | +/// | 128 | 4 | 1670 ms | 679 ms | +/// | 96 | 5 | 1254 ms | 555 ms | +/// | **64** | **7** | **887 ms** | **455 ms** | +/// | 48 | 9 | 1047 ms | 655 ms | +/// | 32 | 13 | 1051 ms | 658 ms | /// -/// The curve has a floor, which is why this is 32 and not 1: below ~16 the -/// per-search overhead aglaked pays regardless of term count (opening the -/// window, pruning buckets) starts to dominate, and chunking too finely is -/// slower than not chunking enough. 32 is within noise of the optimum on the -/// spans and traces indexes too, where 16 measured marginally better. -pub(crate) const ID_CHUNK: usize = 32; +/// Both directions away from 64 are worse, for different reasons: fewer, bigger +/// searches leave the permits idle, and more, smaller ones pay aglaked's +/// per-search overhead (opening the window, pruning buckets) more times than the +/// terms save. Chunking below ~16 is slower than not chunking at all. +/// +/// Note what the two columns do *not* say. Upstream fixed the per-term cost on +/// dictionary-less (unsealed) buckets — it was quadratic per event, `sglog-ystd` +/// — which is the difference between the columns and is worth 4x on its own. It +/// does not make the chunking redundant: concurrency is still worth **3.5x** +/// after the fix, and 64 is still the optimum. Per-term cost falls with term +/// count on a fixed daemon (4.02 ms/term at 400 vs 10.1 at 32), so the marginal +/// term is cheap — but the total still grows with N, and only the fan-out +/// shortens the wall clock. +/// +/// 64 also pairs with the default concurrency: 8 permits x 64 ids means up to +/// 512 ids resolve in a single wave, which covers every agent turn observed so +/// far (the largest was 418 calls). +pub(crate) const ID_CHUNK: usize = 64; /// Chunk size for id lookups that fan *out* — many rows per id, not one. ///