diff --git a/src/api/mod.rs b/src/api/mod.rs index 79cab6d..976447b 100644 --- a/src/api/mod.rs +++ b/src/api/mod.rs @@ -1583,6 +1583,10 @@ mod tests { .unwrap(); } engine.barrier().await; + // `barrier()` only confirms the primary's queue drained -- ticket 108's deterministic + // wait for the profile actor (which computes `requests`' buckets asynchronously) to + // have caught up, rather than relying on scheduler luck. + engine.wait_for_profile_sequence(2).await; let (status, body) = get_json(&app, "/api/query?promql=requests%7Bpath%3D%22%2Fa%22%7D").await; diff --git a/src/engine/handle.rs b/src/engine/handle.rs index fad8d1d..e421b82 100644 --- a/src/engine/handle.rs +++ b/src/engine/handle.rs @@ -96,6 +96,9 @@ pub struct EngineHandle { /// queries read through this instead of `reader`. Not part of the write path; the profile /// actor's own `mpsc` channel (proxied through `EngineMessage`'s metric variants) is that. pub(crate) profile_reader: IndexReader, + /// The profile actor's own processed-sequence watermark (`ProfileHandle::processed_sequence`) + /// — see `wait_for_profile_sequence` below. + pub(crate) profile_processed_sequence: watch::Receiver>, } impl Clone for EngineHandle { @@ -104,6 +107,7 @@ impl Clone for EngineHandle { tx: self.tx.clone(), reader: self.reader.clone(), profile_reader: self.profile_reader.clone(), + profile_processed_sequence: self.profile_processed_sequence.clone(), } } } @@ -120,6 +124,20 @@ impl EngineHandle { let _ = response.await; } } + + /// Waits until the profile actor has finished processing primary sequence `sequence` (metrics + /// computed, alerts evaluated) — ticket 108's deterministic alternative to a sleep for tests + /// that assert on profile-actor-owned data after ingesting; `barrier()` alone only confirms + /// the *primary*'s queue drained, not that the profile actor has caught up to what it published. + #[cfg(test)] + pub(crate) async fn wait_for_profile_sequence(&self, sequence: u64) { + let mut rx = self.profile_processed_sequence.clone(); + while *rx.borrow() < Some(sequence) { + if rx.changed().await.is_err() { + return; // profile actor gone + } + } + } } /// Builds the primary actor's sender/receiver/reader/writer/version-receiver, shared by `spawn` @@ -161,6 +179,7 @@ pub fn spawn(settings: SettingsHandle, channel_capacity: usize) -> EngineHandle tx, reader, profile_reader: profile.reader.clone(), + profile_processed_sequence: profile.processed_sequence.clone(), }; tokio::spawn(run(rx, write, profile)); handle @@ -171,35 +190,18 @@ pub(crate) fn test_handle( channel_capacity: usize, ) -> (EngineHandle, mpsc::Receiver) { let (tx, rx, reader, _write, _version_rx) = new_channel_and_handle(channel_capacity); + // No actor loop is spawned here at all, so nothing ever publishes -- an inert reader over + // an always-empty snapshot is all a query through this handle could see anyway. + let (_processed_tx, processed_rx) = watch::channel(None); let handle = EngineHandle { tx, reader, - // No actor loop is spawned here at all, so nothing ever publishes -- an inert reader - // over an always-empty snapshot is all a query through this handle could see anyway. profile_reader: crate::engine::profile::inert_reader(), + profile_processed_sequence: processed_rx, }; (handle, rx) } -/// Like `spawn`, but wires the profile actor's test-only observer instead of the normal -/// fire-and-forget one, so a test can assert the primary->profile wake/read loop actually ran. -#[cfg(test)] -pub(crate) fn spawn_with_profile_observer( - settings: SettingsHandle, - channel_capacity: usize, -) -> (EngineHandle, tokio::sync::mpsc::UnboundedReceiver) { - let (tx, rx, reader, write, version_rx) = new_channel_and_handle(channel_capacity); - let (profile, observed) = - crate::engine::profile::spawn_observed(reader.clone(), version_rx, settings); - let handle = EngineHandle { - tx, - reader, - profile_reader: profile.reader.clone(), - }; - tokio::spawn(run(rx, write, profile)); - (handle, observed) -} - async fn run( mut rx: mpsc::Receiver, mut write: IndexWriter, @@ -427,7 +429,7 @@ mod tests { #[tokio::test] async fn profile_actor_observes_newly_ingested_lines_via_the_primary_watch_signal() { let settings: SettingsHandle = Arc::new(Mutex::new(Settings::default())); - let (engine, mut observed) = spawn_with_profile_observer(settings, 8); + let engine = spawn(settings, 8); engine .tx @@ -439,10 +441,8 @@ mod tests { .await .unwrap(); - let observed_sequence = tokio::time::timeout(Duration::from_secs(2), observed.recv()) + tokio::time::timeout(Duration::from_secs(2), engine.wait_for_profile_sequence(0)) .await - .expect("timed out waiting for the profile actor to observe the new line") - .expect("profile actor's observer channel closed"); - assert_eq!(observed_sequence, 0); + .expect("timed out waiting for the profile actor to observe the new line"); } } diff --git a/src/engine/profile.rs b/src/engine/profile.rs index d9a4d88..53027fc 100644 --- a/src/engine/profile.rs +++ b/src/engine/profile.rs @@ -62,6 +62,11 @@ pub(crate) struct ProfileHandle { /// `MetricDeclarations`/`BucketValues`/`Promql` queries read through this (see /// `EngineHandle::profile_reader` in `handle.rs`) instead of the primary's reader. pub(crate) reader: IndexReader, + /// The highest primary log sequence this actor has finished processing (metrics computed, + /// alerts evaluated), or `None` before it's processed anything -- ticket 108's deterministic + /// alternative to a sleep for tests that need to wait until the profile actor has caught up + /// to a specific point, mirroring what `EngineHandle::barrier` gives the primary. + pub(crate) processed_sequence: watch::Receiver>, } /// A read-only view over a fresh, never-written `EngineSnapshot` -- used by `handle::test_handle`, @@ -91,41 +96,20 @@ pub(crate) fn spawn( let reader = IndexReader { factory: read.factory(), }; + let (processed_tx, processed_rx) = watch::channel(None); tokio::spawn(run( primary_reader, primary_version, rx, write, settings, - None, + processed_tx, )); - ProfileHandle { tx, reader } -} - -/// Test-only variant of `spawn` that reports every ingest-driven watermark advance on an -/// unbounded channel, so a test can assert the wake/read loop actually ran without reaching -/// into actor-private state. -#[cfg(test)] -pub(crate) fn spawn_observed( - primary_reader: IndexReader, - primary_version: watch::Receiver, - settings: SettingsHandle, -) -> (ProfileHandle, mpsc::UnboundedReceiver) { - let (tx, rx) = mpsc::channel(crate::consts::ENGINE_CHANNEL_CAPACITY); - let (write, read) = left_right::new::(); - let reader = IndexReader { - factory: read.factory(), - }; - let (advance_tx, advance_rx) = mpsc::unbounded_channel(); - tokio::spawn(run( - primary_reader, - primary_version, - rx, - write, - settings, - Some(advance_tx), - )); - (ProfileHandle { tx, reader }, advance_rx) + ProfileHandle { + tx, + reader, + processed_sequence: processed_rx, + } } async fn run( @@ -134,7 +118,7 @@ async fn run( mut rx: mpsc::Receiver, mut write: left_right::WriteHandle, settings: SettingsHandle, - on_advance: Option>, + processed_tx: watch::Sender>, ) { let mut last_seen_sequence: Option = None; let mut alert_states: HashMap = HashMap::new(); @@ -197,8 +181,8 @@ async fn run( )); } } - if let (Some(sequence), Some(tx)) = (last_seen_sequence, on_advance.as_ref()) { - let _ = tx.send(sequence); + if last_seen_sequence.is_some() { + processed_tx.send_replace(last_seen_sequence); } } message = rx.recv() => { diff --git a/tickets/108-rework-engine-test-harness-for-actor-split.md b/tickets/108-rework-engine-test-harness-for-actor-split.md deleted file mode 100644 index eb2c1b9..0000000 --- a/tickets/108-rework-engine-test-harness-for-actor-split.md +++ /dev/null @@ -1,57 +0,0 @@ -# 108 — Rework the engine test harness for the primary/profile actor split - -## Summary - -`src/engine/tests.rs` (2300+ lines) and the `test_handle` pattern are -built entirely around one actor with one channel and one snapshot. -After tickets 103-107 land, tests need to exercise three shapes that -don't exist today: the primary actor alone, the profile actor alone, -and primary+profile interaction (including lag/backfill scenarios). -This is explicitly called out as a gap in [[0001-actor-split]]'s "Not -yet considered" section — real rework, not just extension. - -## Where - -- `src/engine/tests.rs` — the whole file; most existing tests - construct an `EngineSnapshot` directly or drive it through a single - `test_handle`-obtained channel and assert against one snapshot's - query results. Every test that declares a metric, ingests a - matching line, and asserts on bucket/query output now spans two - actors. -- `src/engine/handle.rs` `test_handle` (see ticket 099's history for - the existing single-actor test-visibility pattern) — needs an - equivalent for the profile actor, and likely a combined test harness - helper that spins up both and wires them together the way `spawn` - does in production. -- `src/api/mod.rs`'s test module (`ingest_lines`/similar helpers, e.g. - around `:1817` per prior research) and `src/config/profile.rs`'s - test module — both currently assume one engine handle; they need - updating wherever they exercise metric/alert behavior that's now - profile-actor-owned. - -## Scope - -- Add primary-only test helpers (construct a primary actor, ingest - lines, assert on raw storage/param extraction only — no metrics). -- Add profile-only test helpers (construct a profile actor against a - fake/minimal primary read-side, declare metrics, assert on bucket - computation and alert evaluation in isolation). -- Add combined-harness helpers for tests that need real end-to-end - behavior (ingest via primary, wait for the profile actor to catch up - via the watch signal — needs a deterministic "wait until profile - actor has processed up to sequence N" helper rather than a sleep, to - avoid flaky tests) — a `barrier`-equivalent for the profile actor, - mirroring `EngineHandle::barrier` (`src/engine/handle.rs:89-101` per - prior research). -- Add explicit tests for the accepted lag/eviction tradeoff from - [[0001-actor-split]]'s resolved decisions: primary evicts on - schedule regardless of profile-actor progress, and a lagging profile - actor silently misses data for the evicted range — this behavior - should be asserted, not just assumed. -- Migrate existing tests in `src/engine/tests.rs` to the new harness - incrementally; tests that only exercise primary-side behavior (param - declaration, log storage, eviction) need minimal changes, while - tests exercising metrics/alerts need the combined harness. -- Out of scope: adding new test *coverage* beyond what's needed to - preserve today's behavior guarantees across the split — this is a - harness migration, not a coverage expansion project.