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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TeyhRfMYjMditek6geDRoQ

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<dyn FnMut(&mut EngineSnapshot) + Send + 'static>);
 
@@ -32,12 +32,19 @@ impl Absorb<EngineOp> for EngineSnapshot {
 
 struct IndexWriter {
     handle: left_right::WriteHandle<EngineSnapshot, EngineOp>,
+    /// 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<u64>,
 }
 
 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<left_right::ReadGuard<'_, EngineSnapshot>> {
@@ -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<EngineMessage>, IndexWriter) {
+) -> (
+    EngineHandle,
+    mpsc::Receiver<EngineMessage>,
+    IndexWriter,
+    watch::Receiver<u64>,
+) {
     let (tx, rx) = mpsc::channel(channel_capacity);
     let (write, read) = left_right::new::<EngineSnapshot, EngineOp>();
     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<EngineMessage>) {
-    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<u64>) {
+    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<EngineMessage>,
     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);
+    }
+}

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;

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<dyn FnMut(&mut ProfileSnapshot) + Send + 'static>);
+
+impl Absorb<ProfileOp> for ProfileSnapshot {
+    fn absorb_first(&mut self, operation: &mut ProfileOp, _other: &Self) {
+        (operation.0)(self);
+    }
+
+    fn drop_first(self: Box<Self>) {}
+
+    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<u64>,
+}
+
+/// 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<u64>) {
+    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<u64>,
+) -> tokio::sync::mpsc::UnboundedReceiver<u64> {
+    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<u64>,
+    #[cfg_attr(not(test), allow(unused_variables))] on_advance: Option<
+        tokio::sync::mpsc::UnboundedSender<u64>,
+    >,
+) {
+    tokio::spawn(async move {
+        let (mut write, _read) = left_right::new::<ProfileSnapshot, ProfileOp>();
+
+        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);
+                }
+            }
+        }
+    });
+}