From 033e8bbab5a06bb393068b4b92cda8301ff5edb4 Mon Sep 17 00:00:00 2001 From: Hugh Kaznowski Date: Mon, 7 Sep 2026 13:39:19 +0100 Subject: [PATCH] Stand up the profile actor skeleton and primary->profile watch signaling (ticket 103) Adds a second actor task with its own left_right-backed ProfileSnapshot and a tokio::sync::watch the primary's IndexWriter bumps after every publish, so the profile actor wakes and re-reads newly-visible primary lines through its own cloned IndexReader instead of polling. No metric computation, alert evaluation, or query routing moves yet -- this is pure plumbing, proven with a test that ingests a line on the primary and observes the profile actor's watermark advance in response. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_01TeyhRfMYjMditek6geDRoQ --- src/engine/handle.rs | 79 ++++++++++++++-- src/engine/mod.rs | 1 + src/engine/profile.rs | 93 +++++++++++++++++++ ...le-actor-skeleton-and-primary-signaling.md | 67 ------------- 4 files changed, 166 insertions(+), 74 deletions(-) create mode 100644 src/engine/profile.rs delete mode 100644 tickets/103-profile-actor-skeleton-and-primary-signaling.md diff --git a/src/engine/handle.rs b/src/engine/handle.rs index 167e4a4..1f75c8c 100644 --- a/src/engine/handle.rs +++ b/src/engine/handle.rs @@ -14,7 +14,7 @@ use crate::engine::{ClearParamsResult, IngestedLine}; use left_right::Absorb; use std::collections::{BTreeMap, HashMap}; use std::sync::Arc; -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::{mpsc, oneshot, watch}; struct EngineOp(Box); @@ -32,12 +32,19 @@ impl Absorb for EngineSnapshot { struct IndexWriter { handle: left_right::WriteHandle, + /// Bumped and re-sent on every publish so the profile actor (`super::profile`) can wake on + /// change rather than polling; the value itself carries no meaning beyond "something new is + /// visible; go re-read what you need through your own `IndexReader`." + version: u64, + version_tx: watch::Sender, } impl IndexWriter { fn mutate(&mut self, op: impl FnMut(&mut EngineSnapshot) + Send + 'static) { self.handle.append(EngineOp(Box::new(op))); self.handle.publish(); + self.version += 1; + let _ = self.version_tx.send(self.version); } fn peek(&mut self) -> Option> { @@ -99,25 +106,37 @@ impl EngineHandle { } } -/// Builds the handle/receiver/writer triple shared by `spawn` and, in tests, by callers that -/// want the raw receiver without an actor loop draining it. +/// Builds the handle/receiver/writer/version-receiver quadruple shared by `spawn` and, in +/// tests, by callers that want the raw receiver without an actor loop draining it. fn new_channel_and_handle( channel_capacity: usize, -) -> (EngineHandle, mpsc::Receiver, IndexWriter) { +) -> ( + EngineHandle, + mpsc::Receiver, + IndexWriter, + watch::Receiver, +) { let (tx, rx) = mpsc::channel(channel_capacity); let (write, read) = left_right::new::(); let reader = IndexReader { factory: read.factory(), }; + let (version_tx, version_rx) = watch::channel(0u64); ( EngineHandle { tx, reader }, rx, - IndexWriter { handle: write }, + IndexWriter { + handle: write, + version: 0, + version_tx, + }, + version_rx, ) } pub fn spawn(settings: SettingsHandle, channel_capacity: usize) -> EngineHandle { - let (handle, rx, write) = new_channel_and_handle(channel_capacity); + let (handle, rx, write, version_rx) = new_channel_and_handle(channel_capacity); + crate::engine::profile::spawn(handle.reader.clone(), version_rx); tokio::spawn(run(rx, settings, write)); handle } @@ -126,10 +145,23 @@ pub fn spawn(settings: SettingsHandle, channel_capacity: usize) -> EngineHandle pub(crate) fn test_handle( channel_capacity: usize, ) -> (EngineHandle, mpsc::Receiver) { - let (handle, rx, _write) = new_channel_and_handle(channel_capacity); + let (handle, rx, _write, _version_rx) = new_channel_and_handle(channel_capacity); (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 (handle, rx, write, version_rx) = new_channel_and_handle(channel_capacity); + let observed = crate::engine::profile::spawn_observed(handle.reader.clone(), version_rx); + tokio::spawn(run(rx, settings, write)); + (handle, observed) +} + async fn run( mut rx: mpsc::Receiver, settings: SettingsHandle, @@ -383,3 +415,36 @@ pub(crate) fn ingest_lag_high_watermark_ms(samples: &[(Timestamp, i64)], now: Ti .max() .unwrap_or(0) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::config::Settings; + use std::sync::Mutex; + use std::time::Duration; + + /// Proves ticket 103's wake/read loop end-to-end: the profile actor wakes on the primary's + /// publish signal and reads the newly-visible line through its own primary `IndexReader`, + /// without the primary ever being told the profile actor exists. + #[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); + + engine + .tx + .send(EngineMessage::Ingest(IngestedLine { + sequence: 0, + timestamp: Timestamp::from_millis(0), + text: "hello".into(), + })) + .await + .unwrap(); + + let observed_sequence = tokio::time::timeout(Duration::from_secs(2), observed.recv()) + .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); + } +} diff --git a/src/engine/mod.rs b/src/engine/mod.rs index 5aa305d..12ac9b6 100644 --- a/src/engine/mod.rs +++ b/src/engine/mod.rs @@ -8,6 +8,7 @@ pub mod ingest; pub mod logfmt; pub mod message; pub mod metric_primitive; +pub(crate) mod profile; pub mod promql; pub mod rate; pub mod snapshot; diff --git a/src/engine/profile.rs b/src/engine/profile.rs new file mode 100644 index 0000000..47bddd2 --- /dev/null +++ b/src/engine/profile.rs @@ -0,0 +1,93 @@ +//! The profile actor (ticket 103, first slice of the split described in +//! `ideas/0001-actor-split.md`). It wakes whenever the primary actor's `IndexWriter` +//! (`super::handle`) publishes a `left_right` change, reads whatever primary log lines have +//! become newly visible since it last looked, and — for now — does nothing with them beyond +//! tracking how far it's read. Metric computation (ticket 104) and alert evaluation +//! (ticket 105) land in this wake loop next; nothing outside `engine::handle::spawn` talks to +//! this actor yet, so it has no public handle of its own. + +use crate::engine::handle::IndexReader; +use left_right::Absorb; +use tokio::sync::watch; + +struct ProfileOp(Box); + +impl Absorb for ProfileSnapshot { + fn absorb_first(&mut self, operation: &mut ProfileOp, _other: &Self) { + (operation.0)(self); + } + + fn drop_first(self: Box) {} + + fn sync_with(&mut self, first: &Self) { + *self = first.clone(); + } +} + +/// Profile-actor-owned derived state. Empty besides a read watermark until ticket 104 moves +/// metric declarations/buckets here and ticket 105 moves alert runtime state here. +#[derive(Debug, Clone, Default)] +struct ProfileSnapshot { + /// Highest primary log sequence this actor has read so far, so the wake loop resumes from + /// there instead of rescanning the primary's whole retained range on every wake. + last_seen_sequence: Option, +} + +/// Spawns the profile actor. `primary_reader` is a clone of the primary's `IndexReader`, giving +/// this actor its own read-only view of the primary's index; `primary_version` is bumped by the +/// primary's `IndexWriter::mutate` after every `left_right` publish, so this actor blocks on +/// `changed()` between primary writes rather than polling. +pub(crate) fn spawn(primary_reader: IndexReader, primary_version: watch::Receiver) { + spawn_inner(primary_reader, primary_version, None); +} + +/// Test-only variant of `spawn` that reports every watermark advance on an unbounded channel, +/// so a test can assert the wake/read loop actually ran without reaching into actor-private +/// state (`ProfileSnapshot` has no public read API yet — this ticket is plumbing only). +#[cfg(test)] +pub(crate) fn spawn_observed( + primary_reader: IndexReader, + primary_version: watch::Receiver, +) -> tokio::sync::mpsc::UnboundedReceiver { + let (tx, rx) = tokio::sync::mpsc::unbounded_channel(); + spawn_inner(primary_reader, primary_version, Some(tx)); + rx +} + +fn spawn_inner( + primary_reader: IndexReader, + mut primary_version: watch::Receiver, + #[cfg_attr(not(test), allow(unused_variables))] on_advance: Option< + tokio::sync::mpsc::UnboundedSender, + >, +) { + tokio::spawn(async move { + let (mut write, _read) = left_right::new::(); + + while primary_version.changed().await.is_ok() { + let primary_handle = primary_reader.handle(); + let Some(primary) = primary_handle.enter() else { + continue; + }; + let last_seen = write.enter().and_then(|guard| guard.last_seen_sequence); + let range_start = last_seen.map_or(0, |sequence| sequence.saturating_add(1)); + + let mut new_last_seen = last_seen; + for (sequence, _timestamp, _text) in primary.logs_in_sequence_range(range_start..) { + new_last_seen = + Some(new_last_seen.map_or(sequence, |current| current.max(sequence))); + } + drop(primary); + + if new_last_seen != last_seen { + write.append(ProfileOp(Box::new(move |snap: &mut ProfileSnapshot| { + snap.last_seen_sequence = new_last_seen; + }))); + write.publish(); + if let (Some(sequence), Some(tx)) = (new_last_seen, on_advance.as_ref()) { + let _ = tx.send(sequence); + } + } + } + }); +} diff --git a/tickets/103-profile-actor-skeleton-and-primary-signaling.md b/tickets/103-profile-actor-skeleton-and-primary-signaling.md deleted file mode 100644 index 4e990bc..0000000 --- a/tickets/103-profile-actor-skeleton-and-primary-signaling.md +++ /dev/null @@ -1,67 +0,0 @@ -# 103 — Stand up the profile actor skeleton and primary→profile signaling, with no behavior moved yet - -## Summary - -First slice of [[0001-actor-split]]'s step 3 (splitting the single -engine actor into reader/primary/profile actors). This ticket is pure -plumbing: a second actor task with its own `left_right` write side for -profile-owned state, a way for it to read the primary's index -(`EngineHandle`'s `IndexReader` is already `Clone`, so no primary-side -change is needed to hand out a read-only view), and a -`tokio::sync::watch` the primary signals on after each `left_right` -sync so the profile actor wakes without busy-polling. No metric -computation, alert evaluation, or query routing moves in this ticket — -that's [[104-move-metric-computation-to-profile-actor]] and -[[105-move-alert-evaluation-to-profile-actor]]. Depends on -[[101-extract-metric-computation-into-separate-stage]] (done) for the -`ingest_params`/`apply_all_metrics` seam this split will eventually -route across two actors. - -## Where - -- `src/engine/handle.rs:33-47` (`IndexWriter`) — `mutate()` calls - `self.handle.append(...)` then `self.handle.publish()`; this is - where the new watch-sender notification hooks in, right after - `publish()`. -- `src/engine/handle.rs:48-73` (`IndexReadHandle`/`IndexReader`) — the - existing read-side pattern to mirror for the profile actor's own - derived state; `IndexReader` is already `Clone` - (`handle.rs:61-67` is a template, not something to reuse directly — - the profile actor needs its own `left_right` pair for its own - state, plus a cloned primary `IndexReader` for read access). -- `src/engine/handle.rs:74-78` (`EngineHandle`) and `:119-124` - (`spawn`) — the shape to parallel for a new profile-actor spawn - function. -- `src/engine/snapshot/mod.rs:208-255` (`logs_range`/ - `logs_in_sequence_range`) — the read-side primary API the profile - actor will use (via its cloned `IndexReader`) to pull newly-visible - lines after waking on the watch signal. - -## Scope - -- Add a `ProfileSnapshot` type (new module, e.g. - `src/engine/profile.rs`) with its own `left_right` write/read pair, - parallel to `EngineSnapshot`'s pattern in `handle.rs` — for now it - can be effectively empty (a placeholder struct), since no state - moves into it yet. -- Add a `tokio::sync::watch` (carrying, e.g., the primary's - latest processed sequence) that `IndexWriter::mutate` sends on after - `publish()` succeeds. -- Add a profile actor task: holds a cloned primary `IndexReader` plus - its own `ProfileSnapshot` write side, wakes on the watch channel, - and — for this ticket only — reads whatever new lines - `logs_in_sequence_range` exposes since its last-seen sequence and - does nothing with them (e.g. just advances its own watermark). The - goal is proving the wake/read loop works end-to-end, not doing real - work with the data yet. -- Wire `spawn` (`src/cli/mod.rs:95`) to also spawn the profile actor - alongside the primary, but do not yet change what `ApiState` holds - or what any API handler talks to — `EngineHandle` and its public API - stay exactly as they are today. That's - [[106-split-enginehandle-api-for-primary-and-profile]]'s concern. -- Out of scope: moving `apply_all_metrics`, metric declarations, or - alert evaluation; changing `EngineHandle`'s public surface; eviction - coordination beyond what already exists (primary still evicts on its - own schedule per [[0001-actor-split]]'s resolved decision — this - ticket doesn't need to handle the profile actor falling behind - eviction yet, since it isn't consuming anything meaningful).