diff --git a/Cargo.lock b/Cargo.lock index 2469cb5..6faf83d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -272,6 +272,12 @@ dependencies = [ "libc", ] +[[package]] +name = "crossbeam-utils" +version = "0.8.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a31eee39dddec8330830986fcd7625edb5a24ec90ea038215273bbc3adb08ac6" + [[package]] name = "darling" version = "0.24.1" @@ -496,6 +502,21 @@ dependencies = [ "slab", ] +[[package]] +name = "generator" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3b854b0e584ead1a33f18b2fcad7cf7be18b3875c78816b753639aa501513ae" +dependencies = [ + "cc", + "cfg-if", + "libc", + "log", + "rustversion", + "windows-link", + "windows-result", +] + [[package]] name = "getrandom" version = "0.2.17" @@ -840,6 +861,23 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "lazy_static" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" + +[[package]] +name = "left-right" +version = "0.11.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8bc015ded5d9b3054dbbdb63332cdd6ee42352ccef19e911e25117490e2f48ee" +dependencies = [ + "crossbeam-utils", + "loom", + "slab", +] + [[package]] name = "libc" version = "0.2.189" @@ -882,12 +920,34 @@ version = "0.4.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" +[[package]] +name = "loom" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "419e0dc8046cb947daa77eb95ae174acfbddb7673b4151f56d1eed8e93fbfaca" +dependencies = [ + "cfg-if", + "generator", + "scoped-tls", + "tracing", + "tracing-subscriber", +] + [[package]] name = "lru-slab" version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + [[package]] name = "matchit" version = "0.8.4" @@ -933,6 +993,15 @@ dependencies = [ "minimal-lexical", ] +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "num-traits" version = "0.2.19" @@ -973,6 +1042,7 @@ dependencies = [ "http-body-util", "humantime", "include_dir", + "left-right", "nom", "pest", "pest_derive", @@ -1481,6 +1551,12 @@ dependencies = [ "syn 3.0.5", ] +[[package]] +name = "scoped-tls" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1cf6437eb19a8f4a6cc0f7dca544973b0b78843adbfeb3683d1a94a0024a294" + [[package]] name = "scopeguard" version = "1.2.0" @@ -1564,6 +1640,15 @@ dependencies = [ "serde", ] +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shlex" version = "2.0.1" @@ -1701,6 +1786,15 @@ dependencies = [ "syn 3.0.5", ] +[[package]] +name = "thread_local" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad99c4c6d32803332c548b1af0540b357b3f5fc0be8f6c6bfe8b2e6ae784070" +dependencies = [ + "cfg-if", +] + [[package]] name = "tinystr" version = "0.8.4" @@ -1854,6 +1948,36 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" dependencies = [ "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7f578e5945fb242538965c2d0b04418d38ec25c79d160cd279bf0731c8d319" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", ] [[package]] @@ -1915,6 +2039,12 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "walkdir" version = "2.5.0" diff --git a/Cargo.toml b/Cargo.toml index 8e8a5d2..cd49a40 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -22,6 +22,7 @@ reqwest = { version = "0.12", default-features = false, features = ["rustls-tls" petname = { version = "3.2.0", default-features = false, features = ["default-rng", "default-words"] } rmcp = { version = "3.2.0", default-features = false, features = ["server", "macros", "transport-io"] } anyhow = "1" +left-right = "0.11.8" [dev-dependencies] tempfile = "3.27.0" diff --git a/src/api.rs b/src/api.rs index 0f9250d..c3e83c5 100644 --- a/src/api.rs +++ b/src/api.rs @@ -234,13 +234,32 @@ fn mime_for_extension(extension: &str) -> &'static str { } } +/// Ticket 075: every read query is served straight off a `left_right::ReadHandle` obtained from +/// `engine.reader_factory` — wait-free, and never funneled through the actor's mpsc queue, so a slow +/// query here can no longer block an unrelated cheap one. The one exception is a sorted `Logs` query +/// the first time its `sort_by` is used: no helper index exists yet for it, and only the actor (the +/// sole owner of the write side) can build one, so this sends `EnsureSortHelperIndex`, awaits the +/// ack, then re-enters a fresh `ReadHandle` (now caught up) before proceeding — every other read, +/// and every sorted query once the helper already exists, never reaches the actor at all. async fn ask(engine: &EngineHandle, query: EngineQuery) -> Result { - let (respond_to, response) = oneshot::channel(); - engine - .send(EngineMessage::Query { query, respond_to }) - .await - .map_err(|_| StatusCode::SERVICE_UNAVAILABLE)?; - response.await.map_err(|_| StatusCode::SERVICE_UNAVAILABLE) + let read_handle = engine.reader_factory.handle(); + if let Some(sort_by) = crate::engine::query_sort_by(&query) { + let needs_build = match read_handle.enter() { + Some(guard) => !crate::engine::has_sort_helper_index(&guard, sort_by), + None => return Err(StatusCode::SERVICE_UNAVAILABLE), + }; + if needs_build { + let (respond_to, response) = oneshot::channel(); + engine + .tx + .send(EngineMessage::EnsureSortHelperIndex { param_name: sort_by.to_string(), respond_to }) + .await + .map_err(|_| StatusCode::SERVICE_UNAVAILABLE)?; + response.await.map_err(|_| StatusCode::SERVICE_UNAVAILABLE)?; + } + } + let guard = read_handle.enter().ok_or(StatusCode::SERVICE_UNAVAILABLE)?; + Ok(crate::engine::handle_query_ro(&guard, query)) } /// Any of `request`'s referenced params (`label_params`/`value_param`) not already present in @@ -385,6 +404,7 @@ async fn declare_param( let request = body.clone(); if state .engine + .tx .send(EngineMessage::DeclareParam { request: body, respond_to }) .await .is_err() @@ -420,7 +440,7 @@ async fn declare_param( /// can't regain their params — this is expected behavior, not a partial failure. async fn clear_params(State(state): State) -> impl IntoResponse { let (respond_to, response) = oneshot::channel(); - if state.engine.send(EngineMessage::ClearParams { respond_to }).await.is_err() { + if state.engine.tx.send(EngineMessage::ClearParams { respond_to }).await.is_err() { return StatusCode::SERVICE_UNAVAILABLE.into_response(); } match response.await { @@ -448,6 +468,7 @@ async fn declare_metric( let request = body.clone(); if state .engine + .tx .send(EngineMessage::DeclareMetric { request: body, respond_to }) .await .is_err() @@ -482,6 +503,7 @@ async fn update_metric( let request = body.clone(); if state .engine + .tx .send(EngineMessage::UpdateMetric { name: name.clone(), request: body, respond_to }) .await .is_err() @@ -519,6 +541,7 @@ async fn delete_metric( let (respond_to, response) = oneshot::channel(); if state .engine + .tx .send(EngineMessage::RemoveMetricDeclaration { name: name.clone(), respond_to }) .await .is_err() @@ -754,8 +777,13 @@ async fn update_settings( if max_logs_changed { let _ = state .engine + .tx .send(EngineMessage::SetMaxLogs(new_max_logs.1)) .await; + // `SetMaxLogs` has no response to await on its own (ticket 075: it's fire-and-forget, same as + // `Ingest`), but this endpoint's caller reasonably expects a subsequent read to already + // reflect the new cap's eviction — barrier so it does. + state.engine.barrier().await; } Json(new_max_logs.0).into_response() } @@ -812,6 +840,7 @@ async fn reload_engine_declarations(engine: &EngineHandle, profile: &Profile) { for param in params.into_iter().filter(|param| !param.auto_derived) { let (respond_to, response) = oneshot::channel(); if engine + .tx .send(EngineMessage::RemoveParamDeclaration { name: param.name, respond_to }) .await .is_ok() @@ -826,6 +855,7 @@ async fn reload_engine_declarations(engine: &EngineHandle, profile: &Profile) { for metric in metrics { let (respond_to, response) = oneshot::channel(); if engine + .tx .send(EngineMessage::RemoveMetricDeclaration { name: metric.name, respond_to }) .await .is_ok() @@ -940,18 +970,15 @@ mod tests { use http_body_util::BodyExt; use serde_json::{json, Value}; use std::sync::Mutex; - use tokio::sync::mpsc; use tower::ServiceExt; /// Spins up a fresh engine + router pair backed by a temp settings file. fn test_app() -> (Router, EngineHandle, SettingsHandle, tempfile::TempDir) { - let (engine_tx, engine_rx) = mpsc::channel(128); - let dir = tempfile::tempdir().unwrap(); let settings_path = dir.path().join("settings.json"); let settings = Arc::new(Mutex::new(Settings::load(&settings_path))); - tokio::spawn(engine::run(engine_rx, settings.clone())); + let engine_tx = engine::spawn(settings.clone(), 128); let app = router(engine_tx.clone(), settings.clone(), settings_path); (app, engine_tx, settings, dir) @@ -1383,6 +1410,7 @@ mod tests { ) .await; engine + .tx .send(EngineMessage::Ingest(IngestedLine { sequence: 0, timestamp: Timestamp::from_millis(0), @@ -1390,6 +1418,7 @@ mod tests { })) .await .unwrap(); + engine.barrier().await; // Sanity check: the dashboard-panel-style query works before deletion. let (status, _body) = get_json(&app, "/api/query?promql=requests").await; @@ -1429,6 +1458,7 @@ mod tests { (2u64, "request path=/a"), ] { engine + .tx .send(EngineMessage::Ingest(IngestedLine { sequence, timestamp: Timestamp::from_millis(sequence as i64 * 1000), @@ -1437,6 +1467,7 @@ mod tests { .await .unwrap(); } + engine.barrier().await; let (status, body) = get_json(&app, "/api/metrics/requests").await; assert_eq!(status, StatusCode::OK); @@ -1482,6 +1513,7 @@ mod tests { (2u64, "request path=/a"), ] { engine + .tx .send(EngineMessage::Ingest(IngestedLine { sequence, timestamp: Timestamp::from_millis(sequence as i64 * 1000), @@ -1490,6 +1522,7 @@ mod tests { .await .unwrap(); } + engine.barrier().await; let (status, body) = get_json(&app, "/api/query?promql=requests%7Bpath%3D%22%2Fa%22%7D").await; @@ -1585,17 +1618,12 @@ mod tests { assert_eq!(profile.params[0].name, "dx"); // Simulate a restart: replay the persisted profile into a fresh engine with no prior ingestion. - let (fresh_engine_tx, fresh_engine_rx) = mpsc::channel(16); let fresh_settings_handle = std::sync::Arc::new(std::sync::Mutex::new(Settings::default())); - tokio::spawn(crate::engine::run(fresh_engine_rx, fresh_settings_handle)); + let fresh_engine_tx = crate::engine::spawn(fresh_settings_handle, 16); profile::load_into_engine(&fresh_engine_tx, profile).await; - let (respond_to, response) = oneshot::channel(); - fresh_engine_tx - .send(EngineMessage::Query { query: EngineQuery::MetricDeclarations, respond_to }) - .await - .unwrap(); - let EngineResponse::MetricDeclarations(metrics) = response.await.unwrap() else { + let response = ask(&fresh_engine_tx, EngineQuery::MetricDeclarations).await.unwrap(); + let EngineResponse::MetricDeclarations(metrics) = response else { panic!("unexpected response"); }; assert_eq!(metrics.len(), 1); @@ -1716,10 +1744,14 @@ mod tests { assert_eq!(dashboards[0].panels[0].promql, "requests"); } + /// Ticket 075: `Ingest` has no response to await on its own, and reads no longer share a queue + /// with it — every caller here immediately reads back what it just ingested, so this barriers + /// once at the end rather than making every call site remember to. async fn ingest_lines(engine: &EngineHandle, start_sequence: u64, texts: &[&str]) { for (offset, text) in texts.iter().enumerate() { let sequence = start_sequence + offset as u64; engine + .tx .send(EngineMessage::Ingest(IngestedLine { sequence, timestamp: Timestamp::from_millis(sequence as i64 * 1000), @@ -1728,6 +1760,7 @@ mod tests { .await .unwrap(); } + engine.barrier().await; } #[tokio::test] diff --git a/src/engine.rs b/src/engine.rs index df2dc2d..4fa7174 100644 --- a/src/engine.rs +++ b/src/engine.rs @@ -12,6 +12,7 @@ use crate::promql::{self, AggOp, LabelMatcher, PromqlQuery}; use crate::rate; use crate::settings::SettingsHandle; use crate::timestamp::Timestamp; +use left_right::Absorb; use regex::Regex; use serde::Deserialize; use std::cmp::Ordering; @@ -129,10 +130,23 @@ pub enum EngineMessage { }, /// `None` means unlimited; sent on startup and whenever the setting changes so the cap applies without a restart. SetMaxLogs(Option), - Query { - query: EngineQuery, - respond_to: oneshot::Sender, + /// Ticket 075: reads no longer funnel through this actor loop (see `EngineHandle::reader_factory`) + /// — a sorted `Logs` query is the one read that still needs the actor, and only when no helper + /// index exists yet for `param_name`: the API layer sends this first, awaits the ack, then re-enters + /// its `ReadHandle` and proceeds with the query against the now-built (and now-cached) index. Every + /// other read, and every sorted query once the helper already exists, never reaches this loop at all. + EnsureSortHelperIndex { + param_name: String, + respond_to: oneshot::Sender<()>, }, + /// Ticket 075: acks immediately, with no mutation — a read via `EngineHandle::reader_factory` no + /// longer shares one FIFO queue with `EngineMessage::Ingest`/declaration sends, so nothing + /// guarantees a read right after a `tx.send(...).await` observes that send's effects. Sending + /// this and awaiting its ack is a barrier: since the actor drains messages strictly in order and + /// fully finishes one (including its `publish()`) before starting the next, every message sent + /// before this one is guaranteed visible to readers by the time this ack arrives. See + /// `EngineHandle::barrier`. + Barrier(oneshot::Sender<()>), } #[allow(dead_code)] @@ -149,27 +163,125 @@ pub enum EngineResponse { Error(String), } -pub type EngineHandle = mpsc::Sender; +/// Ticket 075: state a read query (`handle_query`) touches — split out of `run`'s own locals so it +/// can live behind a `left_right` pair instead of the actor's exclusive ownership. Everything else +/// `run` owns (`alert_states`, `ingest_lag_samples`, `metric_sample_sequence`, `log_stream_header`, +/// `max_logs`) is write-only bookkeeping no query ever reads, so it stays a plain local instead. +#[derive(Debug, Clone, Default)] +pub(crate) struct EngineSnapshot { + index: BTreeMap, + sort_helper_indexes: HashMap, + log_count: usize, + approx_bytes_total: u64, + /// Recomputed on every ingest from `ingest_lag_samples` (a run-loop-local, not part of this + /// snapshot) and stashed here as a plain value — purely informational, so per-request freshness + /// down to the millisecond doesn't matter, and stashing it avoids threading `ingest_lag_samples` + /// itself (and `Timestamp::now()`) into a snapshot that must stay deterministic under replay. + ingest_lag_high_watermark_ms: i64, +} -/// Sole owner of the primary index; runs single-threaded so the map never needs a lock, and builds DTOs itself so callers never clone the index. -pub async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle) { - let mut index: BTreeMap = BTreeMap::new(); +/// A single deferred mutation to `EngineSnapshot`, applied once per copy (see `Absorb` below). Modeled +/// as a boxed closure rather than an enum mirroring every `EngineMessage` mutation shape — the +/// pragmatic choice given how many distinct mutations `run` already performs. Each closure must be +/// self-contained and deterministic: every value it needs (including anything that would otherwise +/// require `Timestamp::now()` or a mutable run-loop-local) is resolved to a plain value *before* the +/// closure is constructed, since left-right may invoke it once now and once later (when catching the +/// stale copy up), and it must produce the same effect both times. +/// Wrapped in a named struct (rather than a bare `Box` type +/// alias) to sidestep a known rustc limitation ("implementation of `Absorb` is not general enough") +/// where a trait impl'd directly over a boxed `dyn FnMut(&mut T)` trait object confuses the borrow +/// checker's higher-ranked-lifetime check when that type also needs to prove `Send` through +/// `tokio::spawn` — hiding the trait object behind a concrete field sidesteps it. +struct EngineOp(Box); + +impl Absorb for EngineSnapshot { + fn absorb_first(&mut self, operation: &mut EngineOp, _other: &Self) { + (operation.0)(self); + } + + fn drop_first(self: Box) {} + + /// Only runs once, when a fresh `ReadHandle` is created from the factory — cloning the whole + /// snapshot here (rather than replaying the oplog) is the simplest correct way to bring a new + /// reader copy up to date with the write side, and the cost is paid once per new handle, not per + /// read. + fn sync_with(&mut self, first: &Self) { + *self = first.clone(); + } +} + +/// Sole owner of `EngineMessage::Ingest`/mutation handling; a `left_right::WriteHandle` (see +/// `EngineSnapshot`) still gives every read query wait-free, lock-free access to a consistent +/// snapshot without going through this channel at all. +pub struct EngineHandle { + pub tx: mpsc::Sender, + pub reader_factory: left_right::ReadHandleFactory, +} + +impl Clone for EngineHandle { + fn clone(&self) -> Self { + EngineHandle { tx: self.tx.clone(), reader_factory: self.reader_factory.clone() } + } +} + +impl EngineHandle { + /// Waits until every `EngineMessage` sent before this call has been fully applied and published + /// — see `EngineMessage::Barrier`'s doc comment for why this is needed at all now that reads + /// bypass the actor's queue. Mainly for tests that ingest/declare then immediately read and + /// expect to see it; production reads generally accept the small, bounded staleness instead. + pub async fn barrier(&self) { + let (respond_to, response) = oneshot::channel(); + if self.tx.send(EngineMessage::Barrier(respond_to)).await.is_ok() { + let _ = response.await; + } + } +} + +/// Spawns the actor and returns the `EngineHandle` callers use to send mutations and (via +/// `reader_factory`) read from — the left-right pair has to be constructed here, before the task +/// starts, since `run` itself has no way to hand its internals back out once spawned. +/// `channel_capacity` is exposed mainly so tests can keep using the small capacities they always +/// have; production always passes the same value main.rs used pre-ticket-075. +pub fn spawn(settings: SettingsHandle, channel_capacity: usize) -> EngineHandle { + let (tx, rx) = mpsc::channel(channel_capacity); + let (write, read) = left_right::new::(); + let reader_factory = read.factory(); + tokio::spawn(run(rx, settings, write)); + EngineHandle { tx, reader_factory } +} + +/// Test-only constructor for callers (`ingest.rs`'s own tests) that want to intercept +/// `EngineMessage`s on a plain `mpsc::Receiver` themselves rather than driving a real `run` loop — +/// they never read through `reader_factory`, so the paired `WriteHandle` is simply dropped. +#[cfg(test)] +pub(crate) fn test_handle(channel_capacity: usize) -> (EngineHandle, mpsc::Receiver) { + let (tx, rx) = mpsc::channel(channel_capacity); + let (_write, read) = left_right::new::(); + (EngineHandle { tx, reader_factory: read.factory() }, rx) +} + +/// Sole owner of `EngineSnapshot`'s `left_right::WriteHandle`; runs single-threaded so appending an +/// op and publishing it never needs a lock. Every read query instead goes straight to a +/// `left_right::ReadHandle` obtained from `EngineHandle::reader_factory`, wait-free and off this +/// channel entirely (ticket 075) — the one exception is a sorted `Logs` query the first time a given +/// `sort_by` is used, which still needs this loop to build (and cache) its helper index; see +/// `EngineMessage::EnsureSortHelperIndex`. +async fn run( + mut rx: mpsc::Receiver, + settings: SettingsHandle, + mut write: left_right::WriteHandle, +) { let mut max_logs: Option = None; - let mut log_count: usize = 0; // Its own counter, independent of log `sequence` — scraped samples aren't tied to any ingested line. let mut metric_sample_sequence: u64 = 0; // Per-rule rolling windows/seen-value sets, keyed by rule id; lives only as long as this loop (no restart persistence). let mut alert_states: HashMap = HashMap::new(); - // Ticket 011: running total of `approx_log_bytes` (see that function) across - // every currently-ingested log, kept in lockstep with `log_count` — incremented - // here on ingest, decremented in `evict_excess_logs` when a log is dropped. - let mut approx_bytes_total: u64 = 0; - // Ticket 060/061: one entry per param name that's ever been used as a `sort_by`, populated - // lazily on first sorted query and kept incrementally up to date on every ingest thereafter; - // swept for staleness on ingest too (see `purge_stale_sort_helper_indexes`). - let mut sort_helper_indexes: HashMap = HashMap::new(); // Ticket 059: the most recently seen `log stream` header line's column names, cached across // ingested lines since a `log stream` pipe only prints its header once, before any data line. + // Kept as a plain run-loop local (not part of `EngineSnapshot`) since it's derived purely from + // `text` (see `log_stream_header_transition`) and never read by a query — moving it into the + // snapshot's left-right op closures would mean re-deriving it non-deterministically across two + // separately-timed closure invocations instead of resolving it once, synchronously, right here. let mut log_stream_header: Option = None; // Ticket 073: rolling high-watermark of ingestion-to-engine-processing lag samples — see // `record_ingest_lag`/`ingest_lag_high_watermark_ms`. @@ -185,104 +297,206 @@ pub async fn run(mut rx: mpsc::Receiver, settings: SettingsHandle processed_at, (processed_at.as_millis() - timestamp.as_millis()).max(0), ); - approx_bytes_total += approx_log_bytes(&text); - index.insert( - IndexKey::IngestedLog { sequence, timestamp }, - DataValue::LogLiteral { text: text.clone() }, - ); - log_count += 1; - auto_declare_log_stream_params(&mut index, &mut log_stream_header, &text); - auto_declare_params(&mut index, &text); - apply_all_metrics(&mut index, sequence, timestamp, &text); - update_sort_helper_indexes_on_ingest(&mut sort_helper_indexes, &index, sequence, timestamp, &text); - purge_stale_sort_helper_indexes(&mut sort_helper_indexes); - evict_excess_logs(&mut index, &mut log_count, max_logs, &mut approx_bytes_total); + let ingest_lag_high_watermark_ms = + ingest_lag_high_watermark_ms(&ingest_lag_samples, processed_at); + + // The header transition is a pure function of `text` alone, so it's resolved here, + // synchronously, against this loop's own `log_stream_header` — not inside the op + // closure below, which left-right may invoke a second time (to catch the stale + // snapshot copy up) and so must never reach back into run-loop-local state. + let is_new_header = crate::logfmt::detect_log_stream_header(&text).is_some(); + log_stream_header = log_stream_header_transition(&log_stream_header, &text); + let row_header = if is_new_header { None } else { log_stream_header.clone() }; - // Evaluate alert rules against just this line, then hand any fired rule off to delivery via tokio::spawn so it stays off the hot path. let (rules, ntfy_topic, ntfy_server) = { let settings = settings.lock().unwrap(); (settings.alert_rules.clone(), settings.ntfy_topic.clone(), settings.ntfy_server.clone()) }; - let fired = - alerts::evaluate_alerts(&rules, &mut alert_states, &index, sequence, timestamp, &text); + + let text_for_op = text.clone(); + write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + snap.approx_bytes_total += approx_log_bytes(&text_for_op); + snap.index.insert( + IndexKey::IngestedLog { sequence, timestamp }, + DataValue::LogLiteral { text: text_for_op.clone() }, + ); + snap.log_count += 1; + if let Some(header) = row_header.as_ref() { + derive_log_stream_params_from_row(&mut snap.index, header, &text_for_op); + } + auto_declare_params(&mut snap.index, &text_for_op); + apply_all_metrics(&mut snap.index, sequence, timestamp, &text_for_op); + update_sort_helper_indexes_on_ingest( + &mut snap.sort_helper_indexes, + &snap.index, + sequence, + timestamp, + &text_for_op, + ); + purge_stale_sort_helper_indexes(&mut snap.sort_helper_indexes); + evict_excess_logs(&mut snap.index, &mut snap.log_count, max_logs, &mut snap.approx_bytes_total); + snap.ingest_lag_high_watermark_ms = ingest_lag_high_watermark_ms; + }))); + write.publish(); + + // Alert evaluation reads the index for context (e.g. windowed rate lookups) — peek + // the snapshot just published rather than threading `&index` through the closure + // above, since firing an alert spawns delivery off the hot path regardless. + let fired = write + .enter() + .map(|guard| { + alerts::evaluate_alerts(&rules, &mut alert_states, &guard.index, sequence, timestamp, &text) + }) + .unwrap_or_default(); for (rule, message) in fired { tokio::spawn(alerts::deliver_for_rule(rule.name, message, ntfy_topic.clone(), ntfy_server.clone())); } } EngineMessage::IngestMetricSample(sample) => { - let labels = sample + let labels: BTreeMap = sample .labels .into_iter() .map(|(name, value)| (name, MetricPrimitive::String(value))) .collect(); - push_bucket_value( - &mut index, - &sample.metric_name, - labels, - metric_sample_sequence, - sample.timestamp, - MetricPrimitive::Number(Number::Float(sample.value)), - ); + let metric_name = sample.metric_name; + let sequence = metric_sample_sequence; + let timestamp = sample.timestamp; + let value = MetricPrimitive::Number(Number::Float(sample.value)); + write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + push_bucket_value(&mut snap.index, &metric_name, labels.clone(), sequence, timestamp, value.clone()); + }))); + write.publish(); metric_sample_sequence += 1; } EngineMessage::SetMaxLogs(new_max) => { max_logs = new_max; - evict_excess_logs(&mut index, &mut log_count, max_logs, &mut approx_bytes_total); + write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + evict_excess_logs(&mut snap.index, &mut snap.log_count, new_max, &mut snap.approx_bytes_total); + }))); + write.publish(); } EngineMessage::DeclareParam { request, respond_to } => { - // A manual re-declaration can change the regex/kind under the same name — drop any - // helper index built under the old declaration so a sorted query rebuilds it fresh - // rather than serving values derived from the now-stale regex. - sort_helper_indexes.remove(&request.name); - let result = declare_param(&mut index, request); + // Peeked against the write side's currently-published state (always fully caught up, + // since every message here appends then immediately publishes) rather than computed + // inside the op closure below — a closure must not itself decide whether to mutate + // from a `Result` it computes, since a failed validation must append no op at all. + let result = write + .enter() + .map(|guard| validate_param_declaration(&guard.index, &request)) + .unwrap_or_else(|| Err("engine is shutting down".to_string())); + let succeeded = result.is_ok(); + let name = request.name.clone(); + let request_for_op = request.clone(); + write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + // A manual re-declaration can change the regex/kind under the same name — drop + // any helper index built under the old declaration so a sorted query rebuilds it + // fresh rather than serving values derived from the now-stale regex. + snap.sort_helper_indexes.remove(&name); + if succeeded { + insert_param_declaration(&mut snap.index, request_for_op.clone()); + } + }))); + write.publish(); let _ = respond_to.send(result); } EngineMessage::RemoveParamDeclaration { name, respond_to } => { - let existed = index.remove(&IndexKey::ParamDeclaration { name: name.clone() }).is_some(); - // A helper index built under the old declaration (or a same-named one re-declared - // with a different regex/kind later) would otherwise keep serving stale derived - // values until ticket 061's sweep happens to purge it — dropping it here instead - // means the next sorted query for this param rebuilds it fresh. - sort_helper_indexes.remove(&name); + let existed = write + .enter() + .is_some_and(|guard| guard.index.contains_key(&IndexKey::ParamDeclaration { name: name.clone() })); + let name_for_op = name.clone(); + write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + snap.index.remove(&IndexKey::ParamDeclaration { name: name_for_op.clone() }); + // A helper index built under the old declaration (or a same-named one re-declared + // with a different regex/kind later) would otherwise keep serving stale derived + // values until ticket 061's sweep happens to purge it — dropping it here instead + // means the next sorted query for this param rebuilds it fresh. + snap.sort_helper_indexes.remove(&name_for_op); + }))); + write.publish(); let _ = respond_to.send(existed); } EngineMessage::ClearParams { respond_to } => { - let removed = clear_all_param_declarations(&mut index); - let (reextracted, logs_considered) = reextract_params_from_retained_logs(&mut index); - sort_helper_indexes.clear(); + // `clear_all_param_declarations`/`reextract_params_from_retained_logs`'s returned + // counts depend on the exact current param/log state, so they're computed once here + // against a scratch clone of the write side's current index (cheap relative to how + // rarely a user triggers a full param clear) and reused for the response; the closure + // below re-runs the same two calls "for real" against each `EngineSnapshot` copy. + let mut scratch = write.enter().map(|guard| guard.index.clone()).unwrap_or_default(); + let removed = clear_all_param_declarations(&mut scratch); + let (reextracted, logs_considered) = reextract_params_from_retained_logs(&mut scratch); + write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + clear_all_param_declarations(&mut snap.index); + reextract_params_from_retained_logs(&mut snap.index); + snap.sort_helper_indexes.clear(); + }))); + write.publish(); let _ = respond_to.send(ClearParamsResult { removed, reextracted, logs_considered }); } EngineMessage::DeclareMetric { request, respond_to } => { - let result = declare_metric(&mut index, request); + let result = write + .enter() + .map(|guard| validate_metric_request(&guard.index, &request).map(|_| ())) + .unwrap_or_else(|| Err("engine is shutting down".to_string())); + let succeeded = result.is_ok(); + let request_for_op = request.clone(); + write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + if succeeded { + let _ = declare_metric(&mut snap.index, request_for_op.clone()); + } + }))); + write.publish(); let _ = respond_to.send(result); } EngineMessage::UpdateMetric { name, request, respond_to } => { - let result = update_metric(&mut index, &name, request); + let result = write + .enter() + .map(|guard| validate_update_metric(&guard.index, &name, &request).map(|_| ())) + .unwrap_or_else(|| Err("engine is shutting down".to_string())); + let succeeded = result.is_ok(); + let existing_name = name.clone(); + let request_for_op = request.clone(); + write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + if succeeded { + let _ = update_metric(&mut snap.index, &existing_name, request_for_op.clone()); + } + }))); + write.publish(); let _ = respond_to.send(result); } EngineMessage::RemoveMetricDeclaration { name, respond_to } => { - let existed = remove_metric_declaration(&mut index, &name); + let existed = write + .enter() + .is_some_and(|guard| guard.index.contains_key(&IndexKey::MetricDeclaration { name: name.clone() })); + let name_for_op = name.clone(); + write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + remove_metric_declaration(&mut snap.index, &name_for_op); + }))); + write.publish(); let _ = respond_to.send(existed); } - EngineMessage::Query { query, respond_to } => { - let ingest_lag_high_watermark_ms = ingest_lag_high_watermark_ms(&ingest_lag_samples, Timestamp::now()); - let response = handle_query( - &index, - query, - approx_bytes_total, - ingest_lag_high_watermark_ms, - &mut sort_helper_indexes, - ); - let _ = respond_to.send(response); + EngineMessage::EnsureSortHelperIndex { param_name, respond_to } => { + write.append(EngineOp(Box::new(move |snap: &mut EngineSnapshot| { + ensure_sort_helper_index_built(&mut snap.sort_helper_indexes, &snap.index, ¶m_name); + }))); + write.publish(); + let _ = respond_to.send(()); + } + EngineMessage::Barrier(respond_to) => { + let _ = respond_to.send(()); } } } } /// Validates the regex and capture group before storing, so a bad declaration is rejected up front rather than discovered later. -fn declare_param( - index: &mut BTreeMap, - request: DeclareParamRequest, +/// Read-only half of `declare_param` (ticket 075's left-right conversion): validates without +/// mutating so the actor loop can compute `EngineMessage::DeclareParam`'s response by peeking the +/// write-side snapshot (`WriteHandle::enter`) *before* appending the actual insert as a left-right +/// op — an op closure must not itself decide whether to mutate based on a `Result` it computes +/// internally, since a failed validation must produce no oplog entry at all. +fn validate_param_declaration( + index: &BTreeMap, + request: &DeclareParamRequest, ) -> Result<(), String> { if OTTYL_RESERVED_PARAM_NAMES.contains(&request.name.as_str()) { return Err(format!( @@ -322,12 +536,26 @@ fn declare_param( request.capture_group )); } + Ok(()) +} +/// Write-only half of `declare_param` — assumes `validate_param_declaration` already passed. +fn insert_param_declaration(index: &mut BTreeMap, request: DeclareParamRequest) { index.insert( IndexKey::ParamDeclaration { name: request.name.clone() }, // `auto_derived` is never trusted from the request — only `auto_declare_params` may set it. DataValue::ParamDeclaration(ParamDeclaration { auto_derived: false, ..request }), ); +} + +/// Validates the regex and capture group before storing, so a bad declaration is rejected up front rather than discovered later. Thin wrapper over `validate_param_declaration`/`insert_param_declaration` kept for callers (tests) that don't need the two halves split across a left-right op boundary — see `validate_param_declaration`'s doc comment. +#[cfg(test)] +fn declare_param( + index: &mut BTreeMap, + request: DeclareParamRequest, +) -> Result<(), String> { + validate_param_declaration(index, &request)?; + insert_param_declaration(index, request); Ok(()) } @@ -486,18 +714,29 @@ fn auto_declare_params(index: &mut BTreeMap, text: &str) { /// count against that shared ceiling, tightening (slightly) the budget left for generic `key=value` /// pairs. Like `auto_declare_params`, never overwrites an existing declaration (manual or auto) for /// any of these names. -fn auto_declare_log_stream_params( +/// Pure header-value transition for one ingested line (ticket 075's left-right conversion): a +/// header-detecting line replaces the current header outright; anything else leaves it unchanged. +/// Split out from the old combined `auto_declare_log_stream_params` so `EngineMessage::Ingest` can +/// resolve this synchronously against the run loop's own `log_stream_header` var (kept as a plain +/// local, not part of `EngineSnapshot`), then hand the *resulting* header value to the left-right op +/// closure by value — the closure itself must never reach back into run-loop-local state, since it +/// may be replayed a second time (once per `EngineSnapshot` copy). +fn log_stream_header_transition( + current: &Option, + text: &str, +) -> Option { + crate::logfmt::detect_log_stream_header(text).or_else(|| current.clone()) +} + +/// Derives and inserts any not-yet-declared `log stream` column params for one row, given the +/// header already in effect for it (see `log_stream_header_transition`) — the index-mutating half of +/// the old combined `auto_declare_log_stream_params`, kept separate so it can run inside a left-right +/// op closure that only ever touches `EngineSnapshot::index`. +fn derive_log_stream_params_from_row( index: &mut BTreeMap, - header: &mut Option, + header: &crate::logfmt::LogStreamHeader, text: &str, ) { - if let Some(detected) = crate::logfmt::detect_log_stream_header(text) { - *header = Some(detected); - return; - } - let Some(header) = header.as_ref() else { - return; - }; if !crate::logfmt::is_log_stream_line(text) { return; } @@ -540,6 +779,26 @@ fn auto_declare_log_stream_params( } } +/// Convenience wrapper combining `log_stream_header_transition` and `derive_log_stream_params_from_row` +/// for callers (tests) that don't need to split them across a left-right op boundary. +/// `EngineMessage::Ingest`'s handler calls the two halves directly instead — see +/// `log_stream_header_transition`'s doc comment for why. +#[cfg(test)] +fn auto_declare_log_stream_params( + index: &mut BTreeMap, + header: &mut Option, + text: &str, +) { + let is_new_header = crate::logfmt::detect_log_stream_header(text).is_some(); + *header = log_stream_header_transition(header, text); + if is_new_header { + return; + } + if let Some(header) = header.as_ref() { + derive_log_stream_params_from_row(index, header, text); + } +} + /// Ticket 052: removes every `IndexKey::ParamDeclaration` entry — manual and auto-derived alike. /// Returns how many were removed. Unlike `RemoveParamDeclaration`'s single-name removal (used by /// `reload_engine_declarations` during a profile switch, which deliberately keeps auto-derived @@ -638,11 +897,14 @@ fn declare_metric( } /// Updates the metric declaration named `existing_name` in place, reusing `declare_metric`'s validation so an edit is rejected the same way a bad new declaration would be. Fails (rather than creating a fresh declaration) if no metric named `existing_name` exists yet. Existing bucket/last-event data for this metric is cleared first and rebuilt from the full log history under the new regex/params, so edited matchers don't leave stale buckets from the previous declaration mixed in with fresh ones. Supports renaming (when `request.name` differs from `existing_name`) as long as the new name isn't already taken by another declaration. -fn update_metric( - index: &mut BTreeMap, +/// Read-only checks `update_metric` performs before mutating anything — split out (ticket 075) so +/// the actor loop can peek the write-side snapshot for `EngineMessage::UpdateMetric`'s `Result` +/// before deciding whether to append a mutating op at all. +fn validate_update_metric( + index: &BTreeMap, existing_name: &str, - request: DeclareMetricRequest, -) -> Result<(), String> { + request: &DeclareMetricRequest, +) -> Result { if !index.contains_key(&IndexKey::MetricDeclaration { name: existing_name.to_string() }) { return Err(format!("no metric declaration named '{existing_name}'")); } @@ -651,8 +913,15 @@ fn update_metric( { return Err(format!("a metric named '{}' already exists", request.name)); } + validate_metric_request(index, request) +} - let re = validate_metric_request(index, &request)?; +fn update_metric( + index: &mut BTreeMap, + existing_name: &str, + request: DeclareMetricRequest, +) -> Result<(), String> { + let re = validate_update_metric(index, existing_name, &request)?; let resolved_params = resolved_param_declarations(index); clear_metric_data(index, existing_name); @@ -1381,6 +1650,9 @@ enum SortHelperKey { /// whose `last_accessed` is stale. The map's value is that line's `Timestamp`, not `()`, purely so a /// query can reconstruct the exact `IndexKey::IngestedLog { sequence, timestamp }` primary-index key /// and look its text up in O(log n) instead of re-scanning `logs_range` to find it by sequence alone. +/// `Clone`/`Debug` so `EngineSnapshot` (which holds these, ticket 075) can derive both — `sync_with` +/// clones the whole snapshot once per new `left_right::ReadHandle`, not per read. +#[derive(Debug, Clone)] struct SortHelperIndexEntry { index: BTreeMap, last_accessed: std::time::Instant, @@ -1469,6 +1741,24 @@ fn purge_stale_sort_helper_indexes(sort_helper_indexes: &mut HashMap, + index: &BTreeMap, + sort_by: &str, +) { + let entry = sort_helper_indexes.entry(sort_by.to_string()).or_insert_with(|| SortHelperIndexEntry { + index: build_sort_helper_index(index, sort_by), + last_accessed: std::time::Instant::now(), + }); + entry.last_accessed = std::time::Instant::now(); +} + fn compiled_param_declarations(index: &BTreeMap) -> Vec { param_declarations_range(index) .filter(|(_, _, _, _, _, elapsed_since_last, _, _)| !elapsed_since_last) @@ -1640,12 +1930,93 @@ pub(crate) fn logs_since( }) } +/// How `handle_query_inner`'s `Logs`/`sort_by` branch gets at a sort helper index — split into two +/// variants (ticket 075) because the two callers hold fundamentally different kinds of access to +/// `sort_helper_indexes`: a direct caller (tests, and formerly the actor) has `&mut` access and can +/// lazily build-or-touch a missing helper right there; the read-only API path only ever has a +/// `left_right::ReadGuard`'s `&EngineSnapshot`, which can't mutate anything — it relies on +/// `EngineMessage::EnsureSortHelperIndex` having built the helper through the actor beforehand, and +/// simply looks it up. `resolve` takes `self` by value (rather than `&mut self`) specifically so the +/// returned reference can carry the enum's own `'a` lifetime instead of a shorter reborrowed one. +enum SortHelperAccess<'a> { + #[cfg(test)] + Mutable(&'a mut HashMap), + Immutable(&'a HashMap), +} + +impl<'a> SortHelperAccess<'a> { + // `index` is only read by the `#[cfg(test)]` `Mutable` arm below — unused in a non-test build. + #[cfg_attr(not(test), allow(unused_variables))] + fn resolve( + self, + index: &BTreeMap, + sort_by: &str, + ) -> Option<&'a BTreeMap> { + match self { + #[cfg(test)] + SortHelperAccess::Mutable(map) => { + ensure_sort_helper_index_built(map, index, sort_by); + map.get(sort_by).map(|entry| &entry.index) + } + SortHelperAccess::Immutable(map) => map.get(sort_by).map(|entry| &entry.index), + } + } +} + +/// Test-facing entry point: lazily builds a missing sort helper index the same way ticket 060/061 +/// always has. `EngineMessage`'s handlers no longer call this (see `handle_query_ro`). +#[cfg(test)] fn handle_query( index: &BTreeMap, query: EngineQuery, approx_bytes_total: u64, ingest_lag_high_watermark_ms: i64, sort_helper_indexes: &mut HashMap, +) -> EngineResponse { + handle_query_inner( + index, + query, + approx_bytes_total, + ingest_lag_high_watermark_ms, + SortHelperAccess::Mutable(sort_helper_indexes), + ) +} + +/// Read-only entry point for the API layer's real query path (ticket 075): reads straight off a +/// `left_right::ReadGuard`'s `&EngineSnapshot`, no actor round-trip, no mutation. Callers must check +/// `has_sort_helper_index` and send `EngineMessage::EnsureSortHelperIndex` first when a sorted `Logs` +/// query's helper index doesn't exist yet — see `api.rs`'s `ask`. +pub(crate) fn handle_query_ro(snapshot: &EngineSnapshot, query: EngineQuery) -> EngineResponse { + handle_query_inner( + &snapshot.index, + query, + snapshot.approx_bytes_total, + snapshot.ingest_lag_high_watermark_ms, + SortHelperAccess::Immutable(&snapshot.sort_helper_indexes), + ) +} + +/// What `sort_by` a query needs a sort helper index for, if any — what the API layer checks +/// (alongside `has_sort_helper_index`) to decide whether `EngineMessage::EnsureSortHelperIndex` needs +/// sending before `handle_query_ro` can proceed. +pub(crate) fn query_sort_by(query: &EngineQuery) -> Option<&str> { + match query { + EngineQuery::Logs(params) => params.sort_by.as_deref(), + _ => None, + } +} + +/// Whether a sort helper index for `sort_by` already exists in `snapshot` — see `query_sort_by`. +pub(crate) fn has_sort_helper_index(snapshot: &EngineSnapshot, sort_by: &str) -> bool { + snapshot.sort_helper_indexes.contains_key(sort_by) +} + +fn handle_query_inner( + index: &BTreeMap, + query: EngineQuery, + approx_bytes_total: u64, + ingest_lag_high_watermark_ms: i64, + sort_helper_access: SortHelperAccess<'_>, ) -> EngineResponse { match query { EngineQuery::Logs(params) => { @@ -1696,17 +2067,14 @@ fn handle_query( // param ordered by ottyl_seq" (`sort_missing_fallback`'s semantics), so this just // filters/limits as it walks rather than partition-sort-concat afterward. Some(sort_by) => { - let helper = sort_helper_indexes.entry(sort_by.to_string()).or_insert_with(|| { - SortHelperIndexEntry { - index: build_sort_helper_index(index, sort_by), - last_accessed: std::time::Instant::now(), - } - }); - helper.last_accessed = std::time::Instant::now(); + static EMPTY_HELPER: std::sync::LazyLock> = + std::sync::LazyLock::new(BTreeMap::new); + let helper_index = + sort_helper_access.resolve(index, sort_by).unwrap_or(&EMPTY_HELPER); let sort_missing_fallback = params.sort_missing_fallback.unwrap_or(true); let mut entries = Vec::new(); - for (helper_key, timestamp) in helper.index.iter() { + for (helper_key, timestamp) in helper_index.iter() { let sequence = match helper_key { SortHelperKey::WithParam(_, sequence) => *sequence, SortHelperKey::WithoutParam(sequence) => { diff --git a/src/ingest.rs b/src/ingest.rs index 83b6acc..e9ab420 100644 --- a/src/ingest.rs +++ b/src/ingest.rs @@ -2,11 +2,10 @@ use std::collections::BTreeMap; use std::path::{Path, PathBuf}; use std::time::Duration; -use crate::engine::{EngineMessage, IngestedLine, MetricSample}; +use crate::engine::{EngineHandle, EngineMessage, IngestedLine, MetricSample}; use crate::timestamp::Timestamp; use tokio::fs::File; use tokio::io::{AsyncBufReadExt, BufReader}; -use tokio::sync::mpsc; const POLL_INTERVAL: Duration = Duration::from_millis(200); @@ -14,7 +13,7 @@ const METRICS_POLL_INTERVAL: Duration = Duration::from_secs(15); /// Reads stdin line by line, sequences and timestamps each line, and forwards it /// to the engine task. -pub async fn run(engine: mpsc::Sender) { +pub async fn run(engine: EngineHandle) { let stdin = tokio::io::stdin(); let mut lines = BufReader::new(stdin).lines(); let mut sequence: u64 = 0; @@ -27,7 +26,7 @@ pub async fn run(engine: mpsc::Sender) { }; sequence += 1; - if engine.send(EngineMessage::Ingest(line)).await.is_err() { + if engine.tx.send(EngineMessage::Ingest(line)).await.is_err() { break; } } @@ -59,7 +58,7 @@ async fn open(path: &Path) -> Option<(File, FileId)> { /// Tails `path` like `tail -f`, resuming from the start if the file is deleted/recreated or /// truncated in place rather than exiting or getting stuck; `sequence` keeps incrementing /// across a recreation since it's ottyl's own line counter, not tied to the source file. -pub async fn run_file(engine: mpsc::Sender, path: PathBuf) { +pub async fn run_file(engine: EngineHandle, path: PathBuf) { let mut sequence: u64 = 0; 'reopen: loop { @@ -109,7 +108,7 @@ pub async fn run_file(engine: mpsc::Sender, path: PathBuf) { }; sequence += 1; - if engine.send(EngineMessage::Ingest(line)).await.is_err() { + if engine.tx.send(EngineMessage::Ingest(line)).await.is_err() { return; } } @@ -124,7 +123,7 @@ pub async fn run_file(engine: mpsc::Sender, path: PathBuf) { /// Polls `url` on a fixed interval for a Prometheus/OpenMetrics-format response, pushing every parsed /// sample straight into the engine. Errors (unreachable endpoint, non-2xx response) are only logged — /// polling keeps retrying on the next tick rather than exiting. -pub async fn run_metrics_poll(engine: mpsc::Sender, url: String) { +pub async fn run_metrics_poll(engine: EngineHandle, url: String) { let client = reqwest::Client::new(); loop { match client.get(&url).send().await { @@ -133,7 +132,7 @@ pub async fn run_metrics_poll(engine: mpsc::Sender, url: String) Ok(body) => { let now = Timestamp::now(); for sample in parse_openmetrics(&body, now) { - if engine.send(EngineMessage::IngestMetricSample(sample)).await.is_err() { + if engine.tx.send(EngineMessage::IngestMetricSample(sample)).await.is_err() { return; } } @@ -239,6 +238,7 @@ mod openmetrics_tests { mod tests { use super::*; use std::io::Write; + use crate::engine::test_handle; use tokio::sync::mpsc; async fn recv_text(rx: &mut mpsc::Receiver) -> String { @@ -258,7 +258,7 @@ mod tests { let path = dir.path().join("app.log"); std::fs::write(&path, "first\n").unwrap(); - let (tx, mut rx) = mpsc::channel(16); + let (tx, mut rx) = test_handle(16); tokio::spawn(run_file(tx, path.clone())); assert_eq!(recv_text(&mut rx).await, "first"); @@ -274,7 +274,7 @@ mod tests { let path = dir.path().join("app.log"); std::fs::write(&path, "before-restart\n").unwrap(); - let (tx, mut rx) = mpsc::channel(16); + let (tx, mut rx) = test_handle(16); tokio::spawn(run_file(tx, path.clone())); assert_eq!(recv_text(&mut rx).await, "before-restart"); @@ -291,7 +291,7 @@ mod tests { let path = dir.path().join("app.log"); std::fs::write(&path, "long line before truncation\n").unwrap(); - let (tx, mut rx) = mpsc::channel(16); + let (tx, mut rx) = test_handle(16); tokio::spawn(run_file(tx, path.clone())); assert_eq!(recv_text(&mut rx).await, "long line before truncation"); @@ -312,7 +312,7 @@ mod tests { let path = dir.path().join("app.log"); std::fs::write(&path, "one\n").unwrap(); - let (tx, mut rx) = mpsc::channel(16); + let (tx, mut rx) = test_handle(16); tokio::spawn(run_file(tx, path.clone())); let first = match tokio::time::timeout(Duration::from_secs(2), rx.recv()) diff --git a/src/logfmt.rs b/src/logfmt.rs index c62f1e4..2fc3345 100644 --- a/src/logfmt.rs +++ b/src/logfmt.rs @@ -181,6 +181,10 @@ static COLUMN_NAME: LazyLock = /// One column named by a detected `log stream` header (ticket 059). `start` is that column's position /// in `char`s (not bytes -- see `char_slice`) within the header line -- meaningful only when /// `LogStreamHeader::tab_delimited` is `false` (see that field's doc comment). +/// `Clone` so the engine's ingest path (`crate::engine::run`) can snapshot the current header value +/// by-value before handing it to a left-right op closure — the closure must be replayable against +/// two separate `EngineSnapshot` copies without touching the run loop's own `log_stream_header` var. +#[derive(Clone)] pub struct LogStreamColumn { pub name: String, start: usize, @@ -195,6 +199,7 @@ pub struct LogStreamColumn { /// A tab-separated header carries no such alignment guarantee (values aren't padded to a shared /// width), but a lone tab is itself an unambiguous separator, so that case instead re-splits each data /// line directly. +#[derive(Clone)] pub struct LogStreamHeader { pub columns: Vec, tab_delimited: bool, diff --git a/src/main.rs b/src/main.rs index 10f575f..988f8f7 100644 --- a/src/main.rs +++ b/src/main.rs @@ -22,7 +22,6 @@ use std::sync::{Arc, LazyLock, Mutex}; use clap::{Parser, Subcommand}; use tokio::net::TcpListener; -use tokio::sync::mpsc; use settings::Settings; @@ -107,11 +106,10 @@ async fn main() { let settings = Arc::new(Mutex::new(Settings::load(&settings_path))); - let (engine_tx, engine_rx) = mpsc::channel(1024); - tokio::spawn(engine::run(engine_rx, settings.clone())); + let engine_tx = engine::spawn(settings.clone(), 1024); { let max_logs = settings.lock().unwrap().max_logs; - let _ = engine_tx.send(engine::EngineMessage::SetMaxLogs(max_logs)).await; + let _ = engine_tx.tx.send(engine::EngineMessage::SetMaxLogs(max_logs)).await; } // Reload the active profile's declarations before ingest starts: there's no way to diff --git a/src/profile.rs b/src/profile.rs index 93a95f6..5b8e1ef 100644 --- a/src/profile.rs +++ b/src/profile.rs @@ -400,6 +400,7 @@ pub async fn load_into_engine(engine: &EngineHandle, profile: &Profile) { let request: DeclareParamRequest = record.into(); let name = request.name.clone(); if engine + .tx .send(EngineMessage::DeclareParam { request, respond_to }) .await .is_err() @@ -425,6 +426,7 @@ pub async fn load_into_engine(engine: &EngineHandle, profile: &Profile) { let request: DeclareMetricRequest = record.into(); let name = request.name.clone(); if engine + .tx .send(EngineMessage::DeclareMetric { request, respond_to }) .await .is_err() @@ -818,7 +820,6 @@ mod tests { use crate::dto::LogsParams; use crate::engine::{EngineMessage, EngineQuery, EngineResponse, IngestedLine}; use crate::timestamp::Timestamp; - use tokio::sync::mpsc; fn param_request(name: &str, pattern: &str, capture_group: &str) -> DeclareParamRequest { DeclareParamRequest { @@ -844,10 +845,14 @@ mod tests { } } + /// Ticket 075: reads no longer go through `EngineHandle::tx` — this reads straight off + /// `EngineHandle::reader_factory`, same as the real API layer. Only used here for + /// `ParamDeclarations`/`MetricDeclarations` lookups, neither of which ever needs the + /// `EnsureSortHelperIndex` round-trip a sorted `Logs` query would (see `api.rs`'s own `ask`). async fn ask(engine: &EngineHandle, query: EngineQuery) -> EngineResponse { - let (respond_to, response) = oneshot::channel(); - engine.send(EngineMessage::Query { query, respond_to }).await.unwrap(); - response.await.unwrap() + let read_handle = engine.reader_factory.handle(); + let guard = read_handle.enter().expect("engine unavailable"); + crate::engine::handle_query_ro(&guard, query) } #[test] @@ -939,7 +944,6 @@ mod tests { async fn seeded_metric_renders_real_error_vs_other_data() { use crate::engine::{self, EngineMessage, EngineQuery, EngineResponse, IngestedLine}; use crate::timestamp::Timestamp; - use tokio::sync::mpsc; let dir = tempfile::tempdir().unwrap(); let settings_path = dir.path().join("settings.json"); @@ -947,13 +951,13 @@ mod tests { settings.ensure_default_profile(&settings_path).unwrap(); let default_profile = settings.profile(DEFAULT_PROFILE_NAME).unwrap().clone(); - let (engine_tx, engine_rx) = mpsc::channel(128); let settings_handle = std::sync::Arc::new(std::sync::Mutex::new(settings.clone())); - tokio::spawn(engine::run(engine_rx, settings_handle)); + let engine_tx = engine::spawn(settings_handle, 128); load_into_engine(&engine_tx, &default_profile).await; for (sequence, text) in [(0u64, "all good"), (1, "an ERROR happened"), (2, "fine again")] { engine_tx + .tx .send(EngineMessage::Ingest(IngestedLine { sequence, timestamp: Timestamp::from_millis(sequence as i64), @@ -962,6 +966,7 @@ mod tests { .await .unwrap(); } + engine_tx.barrier().await; let response = ask( &engine_tx, @@ -1143,13 +1148,13 @@ mod tests { let profile = settings.profile(&active_name).unwrap().clone(); // Spawn a fresh engine and replay the profile's declarations into it. - let (engine_tx, engine_rx) = mpsc::channel(16); let settings_handle = std::sync::Arc::new(std::sync::Mutex::new(settings.clone())); - tokio::spawn(crate::engine::run(engine_rx, settings_handle)); + let engine_tx = crate::engine::spawn(settings_handle, 16); load_into_engine(&engine_tx, &profile).await; // Ingest a line and confirm the engine works with the replayed declarations (no re-declaration needed). engine_tx + .tx .send(EngineMessage::Ingest(IngestedLine { sequence: 0, timestamp: Timestamp::from_millis(0), @@ -1157,6 +1162,7 @@ mod tests { })) .await .unwrap(); + engine_tx.barrier().await; let params = match ask(&engine_tx, EngineQuery::ParamDeclarations).await { EngineResponse::ParamDeclarations(params) => params, @@ -1308,12 +1314,12 @@ mod tests { // Replaying into a fresh engine behaves identically to before migration: the metric still ingests and buckets. let profile = profile.clone(); - let (engine_tx, engine_rx) = mpsc::channel(16); let settings_handle = std::sync::Arc::new(std::sync::Mutex::new(Settings::default())); tokio::runtime::Runtime::new().unwrap().block_on(async { - tokio::spawn(crate::engine::run(engine_rx, settings_handle)); + let engine_tx = crate::engine::spawn(settings_handle, 16); load_into_engine(&engine_tx, &profile).await; engine_tx + .tx .send(EngineMessage::Ingest(IngestedLine { sequence: 0, timestamp: Timestamp::from_millis(0), @@ -1489,10 +1495,10 @@ mod tests { // Simulate the auto-derive path: ingest a line carrying a logfmt key before any param is // manually declared, so `dx` only exists as engine-internal auto-derived state. - let (engine_tx, engine_rx) = mpsc::channel(16); let settings_handle = std::sync::Arc::new(std::sync::Mutex::new(settings.clone())); - tokio::spawn(crate::engine::run(engine_rx, settings_handle)); + let engine_tx = crate::engine::spawn(settings_handle, 16); engine_tx + .tx .send(EngineMessage::Ingest(IngestedLine { sequence: 0, timestamp: Timestamp::from_millis(0), @@ -1506,6 +1512,7 @@ mod tests { let metric_req = metric_request("direction_x", r"dx=(?P\S+) moved", "dx"); let (respond_to, response) = oneshot::channel(); engine_tx + .tx .send(EngineMessage::DeclareMetric { request: metric_req.clone(), respond_to }) .await .unwrap(); @@ -1536,9 +1543,8 @@ mod tests { assert_eq!(profile.params.len(), 1); assert_eq!(profile.params[0].name, "dx"); - let (fresh_engine_tx, fresh_engine_rx) = mpsc::channel(16); let fresh_settings_handle = std::sync::Arc::new(std::sync::Mutex::new(Settings::default())); - tokio::spawn(crate::engine::run(fresh_engine_rx, fresh_settings_handle)); + let fresh_engine_tx = crate::engine::spawn(fresh_settings_handle, 16); load_into_engine(&fresh_engine_tx, &profile).await; let EngineResponse::MetricDeclarations(metrics) = diff --git a/tickets/075-investigate-api-unresponsiveness.md b/tickets/075-investigate-api-unresponsiveness.md deleted file mode 100644 index c1594b0..0000000 --- a/tickets/075-investigate-api-unresponsiveness.md +++ /dev/null @@ -1,60 +0,0 @@ -# 075 — Investigate intermittent API unresponsiveness - -## Summary - -Wishlist feedback, verbatim: "The API is really unresponsive -sometimes. Create a ticket for me to investigate, not an agent. -Populate with some ideas." - -This ticket is explicitly for Hugh to investigate by hand, not for an -agent to pick up and fix — it's a starting list of hypotheses, not a -scoped implementation task. - -## Candidate causes to check - -- **Single actor task, single mpsc channel.** All engine state lives - behind one tokio task (`EngineHandle`/`EngineMessage`, - `engine.rs`), and every HTTP request in `api.rs` becomes a message - send + oneshot await. A single slow query (e.g. a large unsorted - timeline materialize, or the pre-060 sort path if ticket 074 turns - out to still be hitting a full-scan fallback) blocks the entire actor - loop — every other in-flight request queues behind it, including - cheap ones like stats or param-list lookups. Worth instrumenting: - time each `EngineMessage` variant's handling and log/expose the - slowest ones. -- **Sort helper index builds are synchronous full scans.** - `build_sort_helper_index` (`engine.rs:1383`) does a one-time full - scan of every currently-ingested log the first time a param is - sorted by, inline in the actor's message handling — with a large log - count this could itself be the stall, and would look exactly like - "API goes unresponsive briefly" right after picking a new sort - column. Related to ticket 074. -- **Prometheus/OpenMetrics polling (`ingest.rs::run_metrics_poll`)** - runs concurrently with log ingestion feeding the same channel — check - whether a slow or hanging `--metrics-url` endpoint can back up the - shared `EngineMessage` channel and starve log-ingestion/API-serving - throughput. -- **Unbounded per-request work scaling with log volume.** Any handler - that iterates the full retained log index (dashboards/PromQL - evaluation, `promql.rs`; alert rule evaluation, `alerts.rs`) rather - than an indexed range could scale badly as `max logs in memory` - grows, especially with multiple dashboard panels polling - concurrently. -- **Frontend polling cadence.** `webapp/src/lib/polling.ts` — check - whether multiple open tabs/panels each poll independently at a short - interval, potentially self-inflicting load rather than the backend - being intrinsically slow. -- Worth capturing wall-clock timing (both server-side, and via browser - devtools network timing) the next time it's reproduced, to know - whether the stall is queueing (actor busy), compute (one query is - slow), or genuinely I/O (waiting on `--metrics-url` or file tailing). - -## Scope - -Investigation only — no code changes implied by this ticket. Once a -concrete cause is found, split into a targeted fix ticket. - -## Depends on - -None, but related to ticket 074 (sort helper index scans) and touches -the same actor architecture ticket 060/061 modified.