Left-right single-writer, multi-reader index

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<EngineResponse, StatusCode> {
-    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<ApiState>) -> 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]

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<usize>),
-    Query {
-        query: EngineQuery,
-        respond_to: oneshot::Sender<EngineResponse>,
+    /// 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<EngineMessage>;
+/// 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<IndexKey, DataValue>,
+    sort_helper_indexes: HashMap<String, SortHelperIndexEntry>,
+    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<EngineMessage>, settings: SettingsHandle) {
-    let mut index: BTreeMap<IndexKey, DataValue> = 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<dyn FnMut(&mut EngineSnapshot) + Send>` 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<dyn FnMut(&mut EngineSnapshot) + Send + 'static>);
+
+impl Absorb<EngineOp> for EngineSnapshot {
+    fn absorb_first(&mut self, operation: &mut EngineOp, _other: &Self) {
+        (operation.0)(self);
+    }
+
+    fn drop_first(self: Box<Self>) {}
+
+    /// 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<EngineMessage>,
+    pub reader_factory: left_right::ReadHandleFactory<EngineSnapshot>,
+}
+
+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::<EngineSnapshot, EngineOp>();
+    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<EngineMessage>) {
+    let (tx, rx) = mpsc::channel(channel_capacity);
+    let (_write, read) = left_right::new::<EngineSnapshot, EngineOp>();
+    (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<EngineMessage>,
+    settings: SettingsHandle,
+    mut write: left_right::WriteHandle<EngineSnapshot, EngineOp>,
+) {
     let mut max_logs: Option<usize> = 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<String, AlertRuntimeState> = 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<String, SortHelperIndexEntry> = 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<crate::logfmt::LogStreamHeader> = 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<EngineMessage>, 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<String, MetricPrimitive> = 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, &param_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<IndexKey, DataValue>,
-    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<IndexKey, DataValue>,
+    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<IndexKey, DataValue>, 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<IndexKey, DataValue>,
+    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<IndexKey, DataValue>, 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<crate::logfmt::LogStreamHeader>,
+    text: &str,
+) -> Option<crate::logfmt::LogStreamHeader> {
+    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<IndexKey, DataValue>,
-    header: &mut Option<crate::logfmt::LogStreamHeader>,
+    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<IndexKey, DataValue>,
+    header: &mut Option<crate::logfmt::LogStreamHeader>,
+    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<IndexKey, DataValue>,
+/// 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<IndexKey, DataValue>,
     existing_name: &str,
-    request: DeclareMetricRequest,
-) -> Result<(), String> {
+    request: &DeclareMetricRequest,
+) -> Result<Regex, String> {
     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<IndexKey, DataValue>,
+    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<SortHelperKey, Timestamp>,
     last_accessed: std::time::Instant,
@@ -1469,6 +1741,24 @@ fn purge_stale_sort_helper_indexes(sort_helper_indexes: &mut HashMap<String, Sor
     sort_helper_indexes.retain(|_, entry| entry.last_accessed.elapsed() < SORT_HELPER_INDEX_PURGE_AFTER);
 }
 
+/// Builds (if missing) or touches (if present) the helper index for `sort_by`, and bumps its
+/// `last_accessed` either way. Ticket 075: this is now the *only* place a sort helper index gets
+/// built or access-time-refreshed — a plain read query can no longer do so itself, since reads go
+/// straight through a `left_right::ReadHandle` and can't mutate `EngineSnapshot`. A sorted `Logs`
+/// query whose helper index is missing has the API layer send `EngineMessage::EnsureSortHelperIndex`
+/// (handled by calling this) before it re-enters its `ReadHandle` and proceeds.
+fn ensure_sort_helper_index_built(
+    sort_helper_indexes: &mut HashMap<String, SortHelperIndexEntry>,
+    index: &BTreeMap<IndexKey, DataValue>,
+    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<IndexKey, DataValue>) -> Vec<CompiledParamDeclaration> {
     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<String, SortHelperIndexEntry>),
+    Immutable(&'a HashMap<String, SortHelperIndexEntry>),
+}
+
+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<IndexKey, DataValue>,
+        sort_by: &str,
+    ) -> Option<&'a BTreeMap<SortHelperKey, Timestamp>> {
+        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<IndexKey, DataValue>,
     query: EngineQuery,
     approx_bytes_total: u64,
     ingest_lag_high_watermark_ms: i64,
     sort_helper_indexes: &mut HashMap<String, SortHelperIndexEntry>,
+) -> 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<IndexKey, DataValue>,
+    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<BTreeMap<SortHelperKey, Timestamp>> =
+                        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) => {

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<EngineMessage>) {
+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<EngineMessage>) {
         };
         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<EngineMessage>, 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<EngineMessage>, 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<EngineMessage>, 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<EngineMessage>, 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<EngineMessage>, 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<EngineMessage>) -> 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())

src/logfmt.rs

@@ -181,6 +181,10 @@ static COLUMN_NAME: LazyLock<Regex> =
 /// 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<LogStreamColumn>,
     tab_delimited: bool,

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

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<dx>\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) =