diff --git a/Cargo.toml b/Cargo.toml index 954185a..15403b9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,6 +15,11 @@ smarm-trace = [] # (D6). The `ActorInfo.budget_cycles` field exists regardless; it just stays 0 # unless this is enabled. budget-accounting = [] +# RFC 016 Chunk 4: the live observer gen_server (src/observer.rs). Off by +# default (DECISION D10) — the read primitive (Chunks 1–3) is always present +# and unflagged; only the optional gen_server transport sits behind this, so a +# release build pays nothing for an observer it never starts. +observer = [] # Run-queue selection: exactly one, compile-time (see src/run_queue.rs). # Non-default variants need --no-default-features (features are additive). rq-mutex = [] diff --git a/src/lib.rs b/src/lib.rs index f60056b..0552e09 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -28,6 +28,8 @@ pub mod pg; pub mod link; pub mod gen_server; pub mod introspect; +#[cfg(feature = "observer")] +pub mod observer; pub mod runtime; pub(crate) mod raw_mutex; pub(crate) mod slot_state; @@ -59,6 +61,8 @@ pub use introspect::{ actor_info, snapshot, tree, tree_from, ActorInfo, ActorState, RuntimeSnapshot, RuntimeTree, TreeNode, SNAPSHOT_FORMAT_VERSION, }; +#[cfg(feature = "observer")] +pub use observer::{ObserverReply, ObserverRequest}; pub use link::{link, trap_exit, unlink, ExitSignal}; pub use monitor::{demonitor, monitor, Down, DownReason, Monitor, MonitorId}; pub use mutex::{LockTimeout, Mutex, MutexGuard}; diff --git a/src/observer.rs b/src/observer.rs new file mode 100644 index 0000000..0f77120 --- /dev/null +++ b/src/observer.rs @@ -0,0 +1,114 @@ +//! RFC 016 — runtime observability (Chunk 4: the observer gen_server). +//! +//! A thin [`GenServer`] that consumes the Chunk-1 read primitive +//! ([`snapshot`](crate::snapshot) / [`tree`](crate::tree) / +//! [`actor_info`](crate::actor_info)) over a message interface — the live +//! `observer` process, in the OTP sense. It is a *transport*, not the +//! mechanism: the synchronous internal read stays the primitive, and the +//! observer is just one more consumer of it alongside the test suite. This is +//! also the read half of the future RFC 003 control plane — the same actor +//! gains write verbs there rather than a second consumer being spun up +//! (DECISION D12). +//! +//! ## Why it is feature-gated (DECISION D10) +//! +//! The read primitive (Chunks 1–3) is always present and unflagged: it is pure +//! reads and the test suite leans on it. The *gen_server* sits behind the +//! `observer` Cargo feature, off by default, matching RFC 003's dev-only +//! feature-flag stance — a release build pays nothing for a live observer it +//! never starts. +//! +//! ## The protocol is the contract (DECISION D11) +//! +//! [`ObserverRequest`] / [`ObserverReply`] *are* the wire contract. They carry +//! no version field of their own because the payloads already do: +//! [`RuntimeSnapshot`](crate::RuntimeSnapshot) and +//! [`RuntimeTree`](crate::RuntimeTree) each carry +//! [`SNAPSHOT_FORMAT_VERSION`](crate::SNAPSHOT_FORMAT_VERSION) (D1). The owned +//! snapshot — a potentially large `Vec` — travels over the call +//! channel by value; that is intended, it is exactly what a remote observer +//! (RFC 011) will serialize across a node boundary. + +use crate::gen_server::{GenServer, ServerBuilder, ServerRef}; +use crate::introspect::{actor_info, snapshot, tree}; +use crate::introspect::{ActorInfo, RuntimeSnapshot, RuntimeTree}; +use crate::pid::Pid; + +/// A read-only request to the observer. Each verb maps one-to-one onto a +/// Chunk-1 read; there are deliberately no mutating verbs here (those are RFC +/// 003, D12). +#[derive(Debug, Clone)] +pub enum ObserverRequest { + /// Whole-runtime [`snapshot`]. + Snapshot, + /// Parentage forest, folded from a snapshot ([`tree`]). + Tree, + /// Coherent view of one actor ([`actor_info`]); `None` reply if the pid is + /// stale, forged, or names a vacant slot. + ActorInfo(Pid), +} + +/// The observer's reply, tagged to match the [`ObserverRequest`] verb. Each +/// variant wraps the owned Chunk-1 read result unchanged — the observer adds no +/// interpretation, it is pure transport. +#[derive(Debug, Clone)] +pub enum ObserverReply { + Snapshot(RuntimeSnapshot), + Tree(RuntimeTree), + ActorInfo(Option), +} + +/// The observer server. Stateless by construction (a ZST): every reply is +/// derived freshly from the live runtime on each call, so there is nothing to +/// keep between requests. +pub struct Observer; + +impl GenServer for Observer { + type Call = ObserverRequest; + type Reply = ObserverReply; + /// No async verbs: the observer is request/reply only. `Infallible` is + /// uninhabited, so a `cast` can never be constructed and + /// [`handle_cast`](GenServer::handle_cast) is statically unreachable. + type Cast = core::convert::Infallible; + type Info = (); + type Timer = (); + + fn handle_call(&mut self, request: ObserverRequest) -> ObserverReply { + match request { + ObserverRequest::Snapshot => ObserverReply::Snapshot(snapshot()), + ObserverRequest::Tree => ObserverReply::Tree(tree()), + ObserverRequest::ActorInfo(pid) => ObserverReply::ActorInfo(actor_info(pid)), + } + } + + fn handle_cast(&mut self, request: core::convert::Infallible) { + // Uninhabited: this match has no arms because `Cast` cannot be + // constructed. It documents at the type level that the observer takes + // no fire-and-forget traffic. + match request {} + } +} + +/// Spawn the observer under the current actor and hand back its [`ServerRef`]. +/// Shorthand for `ServerBuilder::new(Observer).start()`; use the builder +/// directly (e.g. `.under(sup)`) to slot it into a supervision tree. +/// +/// ``` +/// use smarm::run; +/// use smarm::observer::{self, ObserverRequest, ObserverReply}; +/// +/// run(|| { +/// let obs = observer::start(); +/// +/// // Ask for a whole-runtime snapshot over the call channel. +/// let ObserverReply::Snapshot(snap) = obs.call(ObserverRequest::Snapshot).unwrap() +/// else { panic!("snapshot verb must reply with a snapshot") }; +/// +/// // The observer is itself a scheduled actor, so it appears in the very +/// // snapshot it produced — transport over the same read every consumer sees. +/// assert!(snap.actors.iter().any(|a| a.pid == obs.pid())); +/// }); +/// ``` +pub fn start() -> ServerRef { + ServerBuilder::new(Observer).start() +} diff --git a/tests/observer.rs b/tests/observer.rs new file mode 100644 index 0000000..909cf07 --- /dev/null +++ b/tests/observer.rs @@ -0,0 +1,121 @@ +//! RFC 016 Chunk 4 — the observer gen_server. The whole file is gated on the +//! `observer` feature (run with `cargo test --features observer`); without it +//! the module does not exist and there is nothing to compile. +//! +//! The point these prove: the observer is *transport over the same reads*. Each +//! verb returns exactly what the corresponding Chunk-1 primitive would, just +//! marshalled over the gen_server call channel — so a known spawned actor that +//! `snapshot()` / `actor_info()` would see is equally visible through the +//! observer. +#![cfg(feature = "observer")] + +use smarm::observer::{self, ObserverReply, ObserverRequest}; +use smarm::{channel, run, ActorState, SNAPSHOT_FORMAT_VERSION}; + +#[test] +fn observer_relays_snapshot_tree_and_actor_info() { + run(|| { + // A worker parked on an empty gate: a known, stable actor for the + // observer to find across all three verbs. + let (ready_tx, ready_rx) = channel::<()>(); + let (gate_tx, gate_rx) = channel::<()>(); + let worker = smarm::spawn(move || { + ready_tx.send(()).unwrap(); + gate_rx.recv().unwrap(); + }); + ready_rx.recv().unwrap(); + + let obs = observer::start(); + + // Snapshot: the worker is present, and so is the observer itself — it + // is a scheduled actor like any other. + let ObserverReply::Snapshot(snap) = obs.call(ObserverRequest::Snapshot).unwrap() else { + panic!("Snapshot verb must reply Snapshot"); + }; + assert_eq!(snap.format_version, SNAPSHOT_FORMAT_VERSION); + assert!( + snap.actors.iter().any(|a| a.pid == worker.pid()), + "observer's snapshot should contain the spawned worker" + ); + assert!( + snap.actors.iter().any(|a| a.pid == obs.pid()), + "observer should appear in the snapshot it produced" + ); + + // Tree: same data folded into the parentage forest, same version. + let ObserverReply::Tree(t) = obs.call(ObserverRequest::Tree).unwrap() else { + panic!("Tree verb must reply Tree"); + }; + assert_eq!(t.format_version, SNAPSHOT_FORMAT_VERSION); + fn contains(nodes: &[smarm::TreeNode], pid: smarm::Pid) -> bool { + nodes + .iter() + .any(|n| n.info.pid == pid || contains(&n.children, pid)) + } + assert!( + contains(&t.roots, worker.pid()), + "worker should appear somewhere in the observer's tree" + ); + + // ActorInfo: coherent single-actor view, matching a direct read. + let ObserverReply::ActorInfo(Some(info)) = + obs.call(ObserverRequest::ActorInfo(worker.pid())).unwrap() + else { + panic!("ActorInfo verb must reply ActorInfo(Some) for a live worker"); + }; + assert_eq!(info.pid, worker.pid()); + + gate_tx.send(()).unwrap(); + worker.join().unwrap(); + }); +} + +#[test] +fn observer_reports_none_for_a_forged_pid() { + run(|| { + let obs = observer::start(); + // An index that is not in the slab at all — the verb relays the + // primitive's `None` faithfully. + let forged = smarm::Pid::new(u32::MAX - 1, 0); + let ObserverReply::ActorInfo(none) = + obs.call(ObserverRequest::ActorInfo(forged)).unwrap() + else { + panic!("ActorInfo verb must reply ActorInfo"); + }; + assert!(none.is_none(), "a forged pid should relay as None"); + }); +} + +#[test] +fn observer_sees_a_parked_actor_as_parked() { + run(|| { + let (ready_tx, ready_rx) = channel::<()>(); + let (gate_tx, gate_rx) = channel::<()>(); + let worker = smarm::spawn(move || { + ready_tx.send(()).unwrap(); + gate_rx.recv().unwrap(); // nothing to do but park on the gate + }); + ready_rx.recv().unwrap(); + + let obs = observer::start(); + // Bounded poll through the observer until the worker reaches Parked — + // proving the live state classification rides the call channel intact. + let mut parked = false; + for _ in 0..100_000 { + let ObserverReply::ActorInfo(info) = + obs.call(ObserverRequest::ActorInfo(worker.pid())).unwrap() + else { + panic!("ActorInfo verb must reply ActorInfo"); + }; + if matches!(info, Some(i) if i.state == ActorState::Parked) { + parked = true; + break; + } + smarm::yield_now(); + } + assert!(parked, "observer should eventually report the worker as Parked"); + + gate_tx.send(()).unwrap(); + worker.join().unwrap(); + }); +}