From 6d52b3c62343ad306182416f88ee54aa0e8c7595 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Sun, 16 Aug 2026 20:30:52 +0400 Subject: [PATCH] provisioning-reconciler-demo: wire-announce readiness + --docker node kind MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the per-attempt key-file side channel (DEMO_NODE_KEY_FILE, read_key_report, JoinCheck) with a control-plane announce: node roles send a tagged gossip frame {attempt, logical_node, key_hex, endpoint_addr_json} after joining and every heartbeat thereafter. - iroh-driver: IrohDriver::send_tagged_gossip — fire-and-forget tag-routed gossip egress for bridge-less clients (reuses cached/join connections, dials with backoff). - provisioning: BootstrapMsg::Announce — first delivery while bootstrapping completes the attempt (collector + exactly-once Bootstrapped report); duplicates, misrouted attempts, and terminal-phase announces drop. Unit-tested. - xtask demo: AnnounceActor decodes the tag-routed frame and forwards by attempt to the owning bootstrap actor; last_announce_ms is the wire heartbeat. LocalProcessLogic keeps only process lifecycle. - --docker: DockerProcessLogic (kind "docker") — attached "docker run --rm" child on a per-run labeled bridge network (foreign-node masking: per-container IPs, gateway-dialed supervisor). Standalone scratch image from the static-musl xtask binary (37MB), staged one-file build context. The container is force-removed on every terminal path so a SIGKILLed docker CLI cannot orphan a running container. - Cleanup: no volumes/mounts; label-filtered exit sweep (SIGINT and SIGTERM both drain first) + startup sweep of stale demo resources; images persist per run token. Verified live: process kind (kill -> replacement in 3.4s, provision/ remove/kill waves, zero orphans) and docker kind (8-node abuse waves across docker kill, mid-provision control kills, CLI SIGKILL orphans force-removed, SIGKILL-crash leftovers swept on restart, clean exits leave zero containers/networks/CLIs). provisioning 22 + iroh-driver 13 tests pass. --- crates/iroh-driver/src/iroh_driver.rs | 174 ++++- crates/iroh-driver/src/lib.rs | 11 +- crates/iroh-driver/src/telemetry_transport.rs | 150 ++++- crates/provisioning/Cargo.toml | 2 + crates/provisioning/src/bootstrap.rs | 536 ++++++++++++++++ crates/provisioning/src/lib.rs | 2 + xtask/src/main.rs | 89 ++- xtask/src/provisioning_demo/bootstrap.rs | 112 ++++ xtask/src/provisioning_demo/docker.rs | 428 +++++++++++++ xtask/src/provisioning_demo/docker/Dockerfile | 9 + xtask/src/provisioning_demo/feed.rs | 598 ++++++++++++------ xtask/src/provisioning_demo/mod.rs | 264 ++++++-- xtask/src/provisioning_demo/node.rs | 346 ++++++++-- xtask/src/provisioning_demo/provider.rs | 108 +++- xtask/src/provisioning_demo/view.rs | 3 +- 15 files changed, 2428 insertions(+), 404 deletions(-) create mode 100644 crates/provisioning/src/bootstrap.rs create mode 100644 xtask/src/provisioning_demo/bootstrap.rs create mode 100644 xtask/src/provisioning_demo/docker.rs create mode 100644 xtask/src/provisioning_demo/docker/Dockerfile diff --git a/crates/iroh-driver/src/iroh_driver.rs b/crates/iroh-driver/src/iroh_driver.rs index 68a5371..2d2c045 100644 --- a/crates/iroh-driver/src/iroh_driver.rs +++ b/crates/iroh-driver/src/iroh_driver.rs @@ -31,14 +31,14 @@ use distribution::swim::actor::SwimIn; use distribution::transport_bridge::{OutFrame, Outbox, RelayMirror, RouteView, peer_addr}; use distribution::types::NodeId; -use crate::telemetry_transport::{ - TELEMETRY_ALPN, TelemetryQuicHeader, TelemetryQuicRead, read_events_from_stream, - spawn_subscription_writer, -}; use crate::edge_transport::{ EDGE_ALPN, EdgeSendHandle, EdgeTransportEvent, spawn_edge_recv_pump, spawn_edge_send_pump as spawn_edge_sender_task, }; +use crate::telemetry_transport::{ + TELEMETRY_ALPN, TelemetryQuicHeader, TelemetryQuicRead, read_events_from_stream, + spawn_subscription_writer, +}; use swactor::actor::ActorAddress; use swactor::runtime::Runtime; use swactor_transport::CodecRegistry; @@ -402,15 +402,15 @@ impl IrohDriver { // Only relax relay-cert verification for a custom relay; Default / // Staging relays keep full WebPKI verification. - if custom_relay { - builder = builder.ca_roots_config(iroh::tls::CaRootsConfig::insecure_skip_verify()); - // Relay-only: drop direct IP transports so the endpoint neither - // advertises nor chases direct addresses. Without this, iroh learns - // a peer's NAT-obscured/container-local direct addr via discovery - // and prefers it over the relay, black-holing all data. Applied to - // every endpoint so neither side has a direct addr to publish. - builder = builder.clear_ip_transports(); - } + if custom_relay { + builder = builder.ca_roots_config(iroh::tls::CaRootsConfig::insecure_skip_verify()); + // Relay-only: drop direct IP transports so the endpoint neither + // advertises nor chases direct addresses. Without this, iroh learns + // a peer's NAT-obscured/container-local direct addr via discovery + // and prefers it over the relay, black-holing all data. Applied to + // every endpoint so neither side has a direct addr to publish. + builder = builder.clear_ip_transports(); + } if let Some(key) = secret_key { builder = builder.secret_key(key); @@ -444,8 +444,7 @@ impl IrohDriver { Arc::new(Mutex::new(Vec::new())); let other_accepted_conns: Arc, Connection)>>> = Arc::new(Mutex::new(Vec::new())); - let telemetry_reads: Arc>> = - Arc::new(Mutex::new(Vec::new())); + let telemetry_reads: Arc>> = Arc::new(Mutex::new(Vec::new())); let edge_events: Arc>> = Arc::new(Mutex::new(Vec::new())); { let ep = endpoint.clone(); @@ -509,7 +508,6 @@ impl IrohDriver { }) } - /// Clone the iroh endpoint for creating outbound connections. pub fn endpoint(&self) -> Endpoint { self.endpoint.clone() @@ -851,11 +849,12 @@ impl IrohDriver { ); } - let connect_result = engine.timeout( - per_attempt_timeout, - endpoint.connect(seed_addr.clone(), ALPN), - ) - .await; + let connect_result = engine + .timeout( + per_attempt_timeout, + endpoint.connect(seed_addr.clone(), ALPN), + ) + .await; match connect_result { Ok(Ok(conn)) => { @@ -949,7 +948,117 @@ impl IrohDriver { } }); } - + /// Fire-and-forget a single tagged gossip frame to the peer behind + /// `addr`. + /// + /// The frame format matches the actor-bridge ingress (`write_message`) + /// with the destination set to the peer's peer-mailbox, so a receiving + /// bridge routes it by `type_tag` through its `routes` table. This is + /// lightweight egress for clients that never install the actor bridge — + /// e.g. a bootstrap node announcing its identity to a supervisor. The + /// send reuses an open cached connection (folding completed + /// [`Self::join`] dials first), connects fresh otherwise, and retries + /// with backoff; callers that need delivery guarantees re-send at their + /// own cadence. + pub fn send_tagged_gossip(&self, addr: EndpointAddr, tag: &[u8], payload: Vec) { + let tag = tag.to_vec(); + let peer = NodeId(*addr.id.as_bytes()); + let endpoint = self.endpoint.clone(); + let engine = self.engine.clone(); + let conns = Arc::clone(&self.conns); + let pending_joins = Arc::clone(&self.pending_joins); + self.engine.spawn(async move { + let dest = peer_addr(peer); + let tag_len = (tag.len() as u32).to_be_bytes(); + let mut delay = Duration::from_millis(250); + let max_delay = Duration::from_secs(2); + for _ in 0..4 { + // Reuse an open cached connection, or fold a completed (but + // uncached) join dial into the cache and take it. Sync-only, + // so no guard crosses an await. + let reused = { + let mut cache = conns.lock(); + if let Some(cached) = cache.connections.get(&peer) { + if cached.conn.close_reason().is_none() { + Some(cached.conn.clone()) + } else { + cache.connections.remove(&peer); + None + } + } else { + None + } + } + .or_else(|| { + let mut pending = pending_joins.lock(); + let index = pending.iter().position(|result| result.node_id == peer)?; + let result = pending.remove(index); + let mut cache = conns.lock(); + let generation = cache.next_generation; + cache.next_generation = cache.next_generation.wrapping_add(1).max(1); + cache.connections.insert( + peer, + CachedConnection { + generation, + conn: result.conn.clone(), + }, + ); + Some(result.conn) + }); + let conn = match reused { + Some(conn) => conn, + None => { + // No connection yet: dial one (no locks held). The + // fresh connection is cached so later sends reuse it. + let connect = engine + .timeout( + Duration::from_secs(10), + endpoint.connect(addr.clone(), ALPN), + ) + .await; + match connect { + Ok(Ok(conn)) => { + let mut cache = conns.lock(); + let generation = cache.next_generation; + cache.next_generation = + cache.next_generation.wrapping_add(1).max(1); + cache.connections.insert( + peer, + CachedConnection { + generation, + conn: conn.clone(), + }, + ); + conn + } + _ => { + engine.timer(delay).await; + delay = (delay * 2).min(max_delay); + continue; + } + } + } + }; + let result: Result<(), Box> = async { + let mut send = conn.open_uni().await?; + send.write_all(&dest.0).await?; + send.write_all(&tag_len).await?; + send.write_all(&tag).await?; + send.write_all(&payload).await?; + send.finish()?; + Ok(()) + } + .await; + if result.is_ok() { + return; + } + // Write failed: evict so the next attempt re-dials. + conns.lock().connections.remove(&peer); + engine.timer(delay).await; + delay = (delay * 2).min(max_delay); + } + }); + } // ─── Actor bridge: iroh ⇄ swactor runtime ───────────────────────── /// Install the actor-bridge wiring so the driver shuttles frames between iroh @@ -1129,9 +1238,13 @@ impl AdapterPump { let mut cache = self.conns.lock(); let generation = cache.next_generation; cache.next_generation = cache.next_generation.wrapping_add(1).max(1); - cache - .connections - .insert(node_id, CachedConnection { generation, conn: conn.clone() }); + cache.connections.insert( + node_id, + CachedConnection { + generation, + conn: conn.clone(), + }, + ); generation }; self.spawn_reader(node_id, generation, conn); @@ -1310,11 +1423,12 @@ impl AdapterPump { const ATTEMPTS: u32 = 3; let per_attempt_timeout = Duration::from_secs(10); for attempt in 1..=ATTEMPTS { - let result = engine.timeout( - per_attempt_timeout, - endpoint.connect(dial_addr.clone(), ALPN), - ) - .await; + let result = engine + .timeout( + per_attempt_timeout, + endpoint.connect(dial_addr.clone(), ALPN), + ) + .await; if let Ok(Ok(conn)) = result { pending.lock().push(JoinResult { node_id, conn }); break; diff --git a/crates/iroh-driver/src/lib.rs b/crates/iroh-driver/src/lib.rs index 38b5906..24e66f3 100644 --- a/crates/iroh-driver/src/lib.rs +++ b/crates/iroh-driver/src/lib.rs @@ -9,17 +9,17 @@ // work goes through `EngineHandle`. #![deny(clippy::disallowed_methods)] -pub mod telemetry_transport; pub mod driver_pumps; pub mod edge_transport; pub mod endpoint_advertisement; pub mod iroh_driver; +pub mod telemetry_transport; pub use endpoint_advertisement::{ EndpointAddrMask, MVP_IROH_ENDPOINT_ADDR_MASK_ENV, advertised_endpoint, }; pub use iroh_driver::{ - ConnType, TelemetryPublishHandle, IrohDriver, IrohDriverConfig, JoinPhase, JoinStatus, + ConnType, IrohDriver, IrohDriverConfig, JoinPhase, JoinStatus, TelemetryPublishHandle, conn_type_of, discover_lan_ips, }; @@ -27,7 +27,8 @@ pub use edge_transport::{EDGE_ALPN, EdgeSendHandle, EdgeTransportEvent, EdgeTran pub use telemetry_transport::{ TELEMETRY_ALPN, TelemetryQuicHeader, TelemetryQuicRead, TelemetryQuicWriteStats, - read_events_from_stream, read_next_event, read_next_uni_from_connection, read_stream_header, - read_stream_into_fanout, spawn_connection_reader, spawn_subscription_writer, - write_available_subscription, write_event, write_subscription_until_closed, + read_events_from_stream, read_next_event, read_next_uni_from_connection, read_pull_request, + read_stream_header, read_stream_into_fanout, spawn_connection_reader, spawn_pull_collector, + spawn_pull_server, spawn_subscription_writer, write_available_subscription, write_event, + write_pull_request, write_subscription_until_closed, }; diff --git a/crates/iroh-driver/src/telemetry_transport.rs b/crates/iroh-driver/src/telemetry_transport.rs index de4bd0a..603b87d 100644 --- a/crates/iroh-driver/src/telemetry_transport.rs +++ b/crates/iroh-driver/src/telemetry_transport.rs @@ -5,12 +5,14 @@ use std::sync::Arc; use std::time::Duration; use crossbeam_channel::TryRecvError; -use telemetry::{TelemetrySnapshot, TelemetrySubscription}; -use telemetry::frame::{ChannelDescriptor, ChannelId, ChannelRef, TelemetryEvent, FrameDelivery, Position, -StreamDescriptor,}; use iroh::endpoint::{Connection, RecvStream, SendStream}; use iroh::{Endpoint, EndpointAddr}; use swactor_engine::EngineHandle; +use telemetry::frame::{ + ChannelDescriptor, ChannelId, ChannelRef, FrameDelivery, Position, StreamDescriptor, + TelemetryEvent, +}; +use telemetry::{TelemetrySnapshot, TelemetrySubscription}; pub const TELEMETRY_ALPN: &[u8] = b"swactor/telemetry/0"; @@ -18,6 +20,134 @@ const MAGIC: &[u8; 4] = b"DSQ1"; const TAG_CHANNEL_DECLARED: u8 = 0x01; const TAG_FRAME: u8 = 0x02; const TAG_STREAM_ENDED: u8 = 0x03; + +// ─── Pull model: collector-initiated subscriptions ─────────────────────────── +// +// A supervising node dials a freshly-bootstrapped node on `TELEMETRY_ALPN`, +// sends one subscription request on the first uni stream, and the node answers +// by writing the existing header+events stream shape on a uni stream of the +// same connection. Subscription lifetime = connection lifetime. + +/// Write a pull request (magic + flow id + token + `SubscriptionRequest`) and +/// finish the stream so the serving side's read completes. +const REQUEST_MAGIC: &[u8; 4] = b"DSQR"; +pub async fn write_pull_request( + send: &mut SendStream, + flow_id: [u8; 16], + token: &[u8], + request: &telemetry::SubscriptionRequest, +) -> Result<(), BoxError> { + if token.len() > u16::MAX as usize { + return Err("telemetry pull token exceeds u16 length prefix".into()); + } + send.write_all(REQUEST_MAGIC).await?; + send.write_all(&flow_id).await?; + send.write_all(&(token.len() as u16).to_le_bytes()).await?; + send.write_all(token).await?; + write_json(send, request).await?; + send.finish()?; + Ok(()) +} + +/// Read a pull request written by [`write_pull_request`]. +pub async fn read_pull_request( + recv: &mut RecvStream, +) -> Result<([u8; 16], Vec, telemetry::SubscriptionRequest), BoxError> { + let mut magic = [0u8; 4]; + recv.read_exact(&mut magic).await?; + if &magic != REQUEST_MAGIC { + return Err("invalid telemetry pull request magic".into()); + } + let mut flow_id = [0u8; 16]; + recv.read_exact(&mut flow_id).await?; + let mut token_len = [0u8; 2]; + recv.read_exact(&mut token_len).await?; + let token_len = u16::from_le_bytes(token_len) as usize; + let mut token = vec![0u8; token_len]; + recv.read_exact(&mut token).await?; + let request = read_json(recv).await?; + Ok((flow_id, token, request)) +} + +/// Node side: serve one accepted `TELEMETRY_ALPN` connection. Reads the pull +/// request from the first uni stream, subscribes the local endpoint, and +/// writes the answering subscription stream (header + events, until the +/// subscription ends or the connection drops) on a uni stream of the same +/// connection. One request per connection; the task exits when the writer +/// ends or the connection fails. +pub fn spawn_pull_server( + engine: &EngineHandle, + conn: Connection, + endpoint: std::sync::Arc, + idle_sleep: Duration, +) { + let engine_handle = engine.clone(); + engine.spawn(async move { + let Ok(mut recv) = conn.accept_uni().await else { + return; + }; + let Ok((_flow_id, token, request)) = read_pull_request(&mut recv).await else { + return; + }; + let subscription = endpoint.subscribe("supervisor-pull", request); + let header = + match TelemetryQuicHeader::from_snapshot(_flow_id, token, subscription.snapshot()) { + Ok(header) => header, + Err(_) => return, + }; + let Ok(send) = conn.open_uni().await else { + return; + }; + let _ = + write_subscription_until_closed(&engine_handle, send, header, subscription, idle_sleep) + .await; + }); +} + +/// Supervisor side: dial a node on `TELEMETRY_ALPN`, send the pull request, +/// and stream answering events into `fanout` as they arrive (incrementally, +/// not buffered until stream end). The header is reported through +/// `on_header` first so the caller can register stream/channel metadata +/// before any frame lands. +pub fn spawn_pull_collector( + engine: &EngineHandle, + endpoint: Endpoint, + peer: EndpointAddr, + flow_id: [u8; 16], + token: Vec, + request: telemetry::SubscriptionRequest, + fanout: std::sync::Arc, + on_header: std::sync::mpsc::Sender, +) { + engine.spawn(async move { + let peer_id = peer.id.to_string(); + let Ok(conn) = endpoint.connect(peer, TELEMETRY_ALPN).await else { + eprintln!("telemetry-pull: connect to {peer_id} failed"); + return; + }; + let Ok(mut req) = conn.open_uni().await else { + eprintln!("telemetry-pull: open request stream to {peer_id} failed"); + return; + }; + if let Err(error) = write_pull_request(&mut req, flow_id, &token, &request).await { + eprintln!("telemetry-pull: write request to {peer_id} failed: {error}"); + return; + } + let Ok(mut recv) = conn.accept_uni().await else { + eprintln!("telemetry-pull: no answer stream from {peer_id}"); + return; + }; + let Ok(header) = read_header(&mut recv).await else { + eprintln!("telemetry-pull: answer header from {peer_id} unreadable"); + return; + }; + let _ = on_header.send(header.clone()); + let stream = header.stream.clone(); + while let Ok(Some(event)) = read_next_event(&mut recv, &stream).await { + fanout.publish(event); + } + }); +} const MAX_RECORD_BYTES: usize = 16 * 1024 * 1024; type BoxError = Box; @@ -87,7 +217,9 @@ pub fn spawn_subscription_writer( let Ok(send) = conn.open_uni().await else { return; }; - let _ = write_subscription_until_closed(&engine_handle, send, header, subscription, idle_sleep).await; + let _ = + write_subscription_until_closed(&engine_handle, send, header, subscription, idle_sleep) + .await; }); } @@ -157,10 +289,7 @@ async fn write_subscription_inner( Ok(stats) } -pub async fn write_event( - send: &mut SendStream, - event: &TelemetryEvent, -) -> Result { +pub async fn write_event(send: &mut SendStream, event: &TelemetryEvent) -> Result { let mut bytes = Vec::new(); match event { TelemetryEvent::StreamDeclared(_) => return Ok(0), @@ -232,10 +361,7 @@ pub fn spawn_connection_reader( }); } -async fn write_header( - send: &mut SendStream, - header: &TelemetryQuicHeader, -) -> Result<(), BoxError> { +async fn write_header(send: &mut SendStream, header: &TelemetryQuicHeader) -> Result<(), BoxError> { if header.token.len() > u16::MAX as usize { return Err("telemetry token exceeds u16 length prefix".into()); } diff --git a/crates/provisioning/Cargo.toml b/crates/provisioning/Cargo.toml index 70a8976..4a894d4 100644 --- a/crates/provisioning/Cargo.toml +++ b/crates/provisioning/Cargo.toml @@ -7,6 +7,8 @@ publish = false [dependencies] serde = { version = "1", features = ["derive"] } +swactor = { path = "../.." } +swactor-engine = { path = "../engine" } [dev-dependencies] parking_lot = "0.12" diff --git a/crates/provisioning/src/bootstrap.rs b/crates/provisioning/src/bootstrap.rs new file mode 100644 index 0000000..3023d3a --- /dev/null +++ b/crates/provisioning/src/bootstrap.rs @@ -0,0 +1,536 @@ +//! Actor-controlled node bootstrap. +//! +//! One [`BootstrapActor`] owns one node provision attempt end-to-end. The +//! supervision (reconciler loop) talks to every bootstrap actor through the +//! same standard interface — [`BootstrapMsg`] in, [`BootstrapEvent`] out — and +//! never learns *how* a node is launched. Target-specific behavior (local OS +//! process, docker container over ssh, leased remote host over ssh, …) lives +//! behind the private [`BootstrapLogic`] extension point, resolved from a +//! [`BootstrapRegistry`] by the spec's `kind`. +//! +//! The generic actor owns the universal lifecycle state machine +//! (`Bootstrapping → Ready → Stopping → Dead`), self-drives its probes on the +//! engine, starts telemetry collection once a node reports bootstrapped, and +//! guarantees idempotent termination. Logic implementations report raw +//! observations; the actor alone decides what the supervision sees. +//! +//! Telemetry collection is injected as a [`NodeTelemetryCollector`] hook so +//! this crate stays independent of any transport; the demo's collector dials +//! the node on `TELEMETRY_ALPN` and pulls its subscription +//! (`iroh_driver::spawn_pull_collector`). + +use std::collections::BTreeMap; +use std::sync::Arc; +use std::time::{Duration, SystemTime}; + +use swactor::actor::{ActorAddress, ActorInterface, Ctx}; +use swactor::runtime::ExternalSender; +use swactor_engine::EngineHandle; + +/// Default probe period for the self-driven state machine. +pub const DEFAULT_PROBE_PERIOD: Duration = Duration::from_millis(250); + +/// What the bootstrap actor learned about a node attempt. +#[derive(Clone, Debug)] +pub struct NodeIdentity { + pub attempt: u64, + pub logical_node: String, + /// Transport-facing node key (e.g. iroh node id, hex); opaque here. + pub key_hex: String, + /// Advertised transport address of the node (format owned by the + /// collector implementation); opaque here. + pub transport_addr: String, +} + +/// Standard events every bootstrap actor reports upward. These are the +/// supervision-facing interface: identical for every bootstrap kind. +#[derive(Clone, Debug)] +pub enum BootstrapEvent { + /// The node process came up and joined: identity known, telemetry + /// collectible. Reported at most once per attempt. + Bootstrapped(NodeIdentity), + /// The attempt failed before becoming ready (spawn failure, early exit, + /// terminal transport error). The reconciler retries with a fresh lease. + Failed { attempt: u64, reason: String }, + /// A previously-bootstrapped node process exited (death, not bootstrap + /// failure). The reconciler replaces it. + Exited { attempt: u64, reason: String }, +} + +/// How a report leaves the actor: an app-supplied closure that routes the +/// event into the supervision's message enum. +pub type BootstrapReporter = Arc; + +/// Standard commands the supervision sends to any bootstrap actor. +#[derive(Clone, Debug)] +pub enum BootstrapMsg { + /// Begin the attempt: launch the node process. + Start, + /// Terminate the node process. Idempotent; `kill_after` escalates. + Stop { kill_after: Option }, + /// Internal: self-driven state probe. Sending it externally is harmless. + Probe, + /// The node announced itself over the control plane: its transport + /// identity facts arrived from the wire (key + advertised address), + /// routed here by the app's announce plumbing. First delivery while + /// bootstrapping completes the attempt; later deliveries are heartbeat + /// duplicates and are ignored. + Announce(NodeIdentity), +} + +#[derive(Clone, Debug)] +/// Neutral description of one node launch, independent of bootstrap kind. +pub struct NodeLaunchSpec { + /// Bootstrap kind; selects the logic from the registry. + pub kind: String, + pub attempt: u64, + pub logical_node: String, + /// Semantic argv the *node* runs (before any transport encoding). + pub argv: Vec, + pub env: Vec<(String, String)>, + pub workdir: Option, + pub label: Option, +} + +/// One raw probe observation from a logic implementation. The actor applies +/// state-machine rules (once-only reporting, phase filtering); logics stay +/// dumb readers of their foreign process. +pub enum LogicProbe { + /// Still coming up; nothing to report. + Pending, + /// Node process is up and joined. + Bootstrapped(NodeIdentity), + /// Attempt failed while bootstrapping. + Failed(String), + /// Node process exited after having reported bootstrapped. + Exited(String), +} + +/// Target-specific bootstrap behavior. Implementations live in application +/// crates and may capture app state (registries, join checks, key files). +pub trait BootstrapLogic: Send { + /// Launch the node process. Called once, in actor context, with the + /// owning actor's address (`owner`) so the logic can register itself in + /// app-side registries. An error is a terminal attempt failure. + fn start( + &mut self, + ctx: &Ctx, + owner: ActorAddress, + sender: &ExternalSender, + ) -> Result<(), String>; + /// Non-blocking state probe (key-file reads, registry lookups, exit + /// polls). Never blocks; called every probe period. + fn probe(&mut self, now: SystemTime) -> LogicProbe; + /// Terminate the node process (best effort, idempotent). + fn terminate(&mut self, sender: &ExternalSender, kill_after: Option); +} + +/// Telemetry collection hook the actor fires once per attempt, right after +/// `Bootstrapped` is reported. Implementations dial the node and pull. +pub trait NodeTelemetryCollector: Send + Sync { + fn collect(&self, identity: &NodeIdentity); +} + +/// Creates a [`BootstrapLogic`] for one launch spec. +pub type BootstrapFactory = + Arc Result, String> + Send + Sync>; + +/// Kind-keyed logic registry: adding bootstrap type #N is a new factory +/// registration; the supervision never changes. +#[derive(Default)] +pub struct BootstrapRegistry { + factories: BTreeMap, +} + +impl BootstrapRegistry { + pub fn new() -> Self { + Self::default() + } + + pub fn register(&mut self, kind: &str, factory: BootstrapFactory) { + self.factories.insert(kind.to_owned(), factory); + } + + /// Resolve the logic named by `spec.kind`. + pub fn create(&self, spec: &NodeLaunchSpec) -> Result, String> { + self.factories + .get(&spec.kind) + .ok_or_else(|| format!("no bootstrap logic registered for kind '{}'", spec.kind))? + .clone()(spec) + } +} + +/// Universal per-attempt lifecycle state. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Phase { + /// Launched, no readiness signal yet. + Bootstrapping, + /// Node joined and reported bootstrapped. + Ready, + /// Termination requested, waiting for the exit observation. + Stopping, + /// Terminal: exit observed (or start failed). No further reports. + Dead, +} + +/// Configuration for one spawned bootstrap actor. +pub struct BootstrapConfig { + pub spec: NodeLaunchSpec, + /// Routes events into the supervision's message enum. + pub reporter: BootstrapReporter, + /// Sender the logic uses for process commands. + pub sender: ExternalSender, + /// Optional telemetry collection hook, fired once after `Bootstrapped`. + pub collector: Option>, + pub probe_period: Duration, +} + +/// The generic, transport-agnostic bootstrap actor. See the module docs. +pub struct BootstrapActor { + attempt: u64, + logic: Box, + reporter: BootstrapReporter, + sender: ExternalSender, + collector: Option>, + phase: Phase, + /// `Bootstrapped` reported (and collector fired) exactly once. + announced: bool, + /// Terminal report (Failed/Exited) emitted exactly once. + closed: bool, +} + +impl BootstrapActor { + pub fn new(logic: Box, config: BootstrapConfig) -> Self { + Self { + attempt: config.spec.attempt, + logic, + reporter: config.reporter, + sender: config.sender, + collector: config.collector, + phase: Phase::Bootstrapping, + announced: false, + closed: false, + } + } + + fn report(&self, event: BootstrapEvent) { + (self.reporter)(event); + } + + fn fail(&mut self, reason: String) { + if self.closed { + return; + } + self.closed = true; + self.phase = Phase::Dead; + self.report(BootstrapEvent::Failed { + attempt: self.attempt, + reason, + }); + } + + fn exit(&mut self, reason: String) { + if self.closed { + return; + } + self.closed = true; + self.phase = Phase::Dead; + self.report(BootstrapEvent::Exited { + attempt: self.attempt, + reason, + }); + } + + fn probe(&mut self, now: SystemTime) { + match self.phase { + Phase::Dead => {} + Phase::Bootstrapping => match self.logic.probe(now) { + LogicProbe::Pending => {} + LogicProbe::Bootstrapped(identity) => self.become_ready(identity), + LogicProbe::Failed(reason) => self.fail(reason), + // An exit before the join signal is a failed attempt. + LogicProbe::Exited(reason) => self.fail(format!("node process exited: {reason}")), + }, + Phase::Ready | Phase::Stopping => match self.logic.probe(now) { + LogicProbe::Exited(reason) => { + if self.phase == Phase::Stopping && !self.announced { + // Stopped before it ever joined: failed attempt. + self.fail(format!("stopped before join: {reason}")); + } else { + self.exit(reason); + } + } + _ => {} + }, + } + } + + /// First readiness signal for the attempt (probe or wire announce): + /// fire the collector and report `Bootstrapped` exactly once. + fn become_ready(&mut self, identity: NodeIdentity) { + if self.announced { + return; + } + self.announced = true; + self.phase = Phase::Ready; + if let Some(collector) = &self.collector { + collector.collect(&identity); + } + self.report(BootstrapEvent::Bootstrapped(identity)); + } + + /// Apply a wire announce: first delivery for this attempt completes the + /// bootstrap; anything else (mismatched attempt, heartbeat duplicates, + /// terminal phases) is a drop. + fn apply_announce(&mut self, identity: NodeIdentity) { + if identity.attempt != self.attempt { + return; // misrouted: drop rather than misreport + } + if self.phase == Phase::Bootstrapping { + self.become_ready(identity); + } + } +} + +impl ActorInterface for BootstrapActor { + type Incoming = BootstrapMsg; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, msg: BootstrapMsg) { + match msg { + BootstrapMsg::Start => { + if self.phase != Phase::Bootstrapping { + return; // Restart of a started/stopped attempt: ignore. + } + if let Err(reason) = self.logic.start(ctx, ctx.self_addr(), &self.sender) { + self.fail(format!("launch failed: {reason}")); + } + } + BootstrapMsg::Stop { kill_after } => { + if self.phase == Phase::Dead { + return; + } + self.phase = Phase::Stopping; + self.logic.terminate(&self.sender, kill_after); + } + BootstrapMsg::Probe => { + self.probe(SystemTime::now()); + if self.phase == Phase::Dead { + ctx.stop_self(); + } + } + BootstrapMsg::Announce(identity) => self.apply_announce(identity), + } + } +} + +/// Spawn one bootstrap actor with a self-driven probe interval: the actor, +/// not the supervision tick, owns its progression. `Start` is sent after the +/// interval is installed so the logic never races its own probes. +pub fn spawn_bootstrap_actor( + ctx: &Ctx, + engine: &EngineHandle, + logic: Box, + config: BootstrapConfig, +) -> Result { + let sender = config.sender.clone(); + let start_sender = sender.clone(); + let period = config.probe_period; + let actor = ctx + .spawn(BootstrapActor::new(logic, config)) + .map_err(|error| format!("spawn bootstrap actor: {error}"))?; + let probe_engine = engine.clone(); + engine.spawn(async move { + let mut interval = probe_engine.interval(period); + loop { + (&mut interval).await; + if sender.send_to(actor, BootstrapMsg::Probe).is_err() { + return; + } + } + }); + let _ = start_sender.send_to(actor, BootstrapMsg::Start); + Ok(actor) +} + +#[cfg(test)] +mod tests { + use super::*; + use parking_lot::Mutex; + use std::sync::atomic::{AtomicUsize, Ordering}; + + struct ScriptedLogic { + probes: Mutex>, + started: AtomicUsize, + terminated: AtomicUsize, + } + + impl ScriptedLogic { + fn new(probes: Vec) -> Self { + Self { + probes: Mutex::new(probes), + started: AtomicUsize::new(0), + terminated: AtomicUsize::new(0), + } + } + } + + impl BootstrapLogic for ScriptedLogic { + fn start( + &mut self, + _ctx: &Ctx, + _owner: ActorAddress, + _sender: &ExternalSender, + ) -> Result<(), String> { + self.started.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + fn probe(&mut self, _now: SystemTime) -> LogicProbe { + self.probes.lock().remove(0) + } + fn terminate(&mut self, _sender: &ExternalSender, _kill_after: Option) { + self.terminated.fetch_add(1, Ordering::SeqCst); + } + } + + fn identity(attempt: u64) -> NodeIdentity { + NodeIdentity { + attempt, + logical_node: format!("node-{attempt}"), + key_hex: format!("key{attempt}"), + transport_addr: format!("addr{attempt}"), + } + } + + #[test] + fn registry_resolves_by_kind() { + let mut registry = BootstrapRegistry::new(); + registry.register( + "process", + Arc::new(|_spec| Ok(Box::new(ScriptedLogic::new(vec![])) as Box)), + ); + let spec = NodeLaunchSpec { + kind: "process".to_owned(), + attempt: 1, + logical_node: "node-1".to_owned(), + argv: vec![], + env: vec![], + workdir: None, + label: None, + }; + assert!(registry.create(&spec).is_ok()); + let unknown = NodeLaunchSpec { + kind: "nope".to_owned(), + ..spec + }; + assert!(registry.create(&unknown).is_err()); + } + #[test] + fn reporter_contract_carries_identity() { + let events: Arc>> = Arc::new(Mutex::new(Vec::new())); + let reporter: BootstrapReporter = { + let events = events.clone(); + Arc::new(move |event| { + let name = match event { + BootstrapEvent::Bootstrapped(_) => "bootstrapped", + BootstrapEvent::Failed { .. } => "failed", + BootstrapEvent::Exited { .. } => "exited", + }; + events.lock().push(name.to_owned()); + }) + }; + reporter(BootstrapEvent::Exited { + attempt: 7, + reason: "test".to_owned(), + }); + assert_eq!(events.lock().as_slice(), ["exited"]); + let id = identity(7); + assert_eq!(id.logical_node, "node-7"); + } + + #[test] + fn announce_completes_attempt_exactly_once() { + let events: Arc>> = Arc::new(Mutex::new(Vec::new())); + let mut actor = announce_actor(7, events.clone(), vec![]); + + actor.apply_announce(identity(7)); + assert_eq!(actor.phase, Phase::Ready); + // Heartbeat duplicates: ignored. + actor.apply_announce(identity(7)); + actor.apply_announce(identity(7)); + let logged = events.lock(); + assert_eq!(logged.len(), 1); + assert!(matches!(&logged[0], BootstrapEvent::Bootstrapped(id) if id.key_hex == "key7")); + } + + #[test] + fn announce_with_mismatched_attempt_is_dropped() { + let events: Arc>> = Arc::new(Mutex::new(Vec::new())); + let mut actor = announce_actor(7, events.clone(), vec![]); + + actor.apply_announce(identity(8)); + assert_eq!(actor.phase, Phase::Bootstrapping); + assert!(events.lock().is_empty()); + } + + #[test] + fn announce_after_terminal_phase_is_dropped() { + let events: Arc>> = Arc::new(Mutex::new(Vec::new())); + let mut actor = announce_actor(7, events.clone(), vec![]); + + actor.apply_announce(identity(7)); + actor.exit("killed".to_owned()); + // A stale container from the dead attempt still announcing. + actor.apply_announce(identity(7)); + let logged = events.lock(); + assert_eq!(logged.len(), 2); + assert!(matches!(&logged[1], BootstrapEvent::Exited { attempt, .. } if *attempt == 7)); + } + + #[test] + fn announce_after_probe_bootstrapped_does_not_double_report() { + let events: Arc>> = Arc::new(Mutex::new(Vec::new())); + // Probe path reports Bootstrapped first… + let mut actor = announce_actor( + 9, + events.clone(), + vec![LogicProbe::Bootstrapped(identity(9))], + ); + actor.probe(SystemTime::now()); + // …then the announce for the same attempt arrives late. + actor.apply_announce(identity(9)); + let logged = events.lock(); + assert_eq!(logged.len(), 1); + } + + /// An actor over a scripted logic whose events land in `events`. + fn announce_actor( + attempt: u64, + events: Arc>>, + probes: Vec, + ) -> BootstrapActor { + let reporter: BootstrapReporter = { + let events = events.clone(); + Arc::new(move |event| events.lock().push(event)) + }; + let sender = swactor::runtime::RuntimeParts::new(swactor::config::RuntimeConfig::default()) + .runtime() + .create_sender(); + let config = BootstrapConfig { + spec: NodeLaunchSpec { + kind: "process".to_owned(), + attempt, + logical_node: format!("node-{attempt}"), + argv: vec![], + env: vec![], + workdir: None, + label: None, + }, + reporter, + sender, + collector: None, + probe_period: DEFAULT_PROBE_PERIOD, + }; + BootstrapActor::new(Box::new(ScriptedLogic::new(probes)), config) + } +} diff --git a/crates/provisioning/src/lib.rs b/crates/provisioning/src/lib.rs index 6b065a7..ee60967 100644 --- a/crates/provisioning/src/lib.rs +++ b/crates/provisioning/src/lib.rs @@ -5,10 +5,12 @@ //! provider adapters live in application crates and implement the executor //! backend contract. +pub mod bootstrap; pub mod executor; pub mod node; pub mod plugin; pub mod reconciler; +pub use bootstrap::*; pub use executor::*; pub use node::*; diff --git a/xtask/src/main.rs b/xtask/src/main.rs index 72aa5a0..030c094 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -266,10 +266,12 @@ COMMANDS: Run real cargo myelin-chat acceptance check and write benchmark artifacts. myelin-chat-compare Compare two benchmark summaries and report comparable deltas. - provisioning-reconciler-demo [--port n] [--nodes n] + provisioning-reconciler-demo [--port n] [--nodes n] [--docker] Run the visual E2E provisioning reconciler sanity demo (supervisor + dashboard on localhost, node children - join over iroh). Ctrl-C tears down. + join over iroh; --docker launches nodes as scratch + containers on a per-run bridge network). Ctrl-C tears + down and sweeps. check-telemetry-isolation Verify no frame types appear in control-plane modules. test Run the basic non-binding test barrier: root crate plus each non-binding repository package with `cargo test -p`." @@ -364,9 +366,12 @@ fn run_myelin_chat(args: Vec) -> ExitCode { "args": MYELIN_CHAT_CARGO_RUN_ARGS, }), ); - if let Err(error) = - append_synthetic_benchmark_frame(path, "xtask-myelin-chat", "myelin.xtask.benchmark", event) - { + if let Err(error) = append_synthetic_benchmark_frame( + path, + "xtask-myelin-chat", + "myelin.xtask.benchmark", + event, + ) { eprintln!("Failed to write myelin-chat benchmark frame: {error}"); return ExitCode::from(1); } @@ -381,9 +386,12 @@ fn run_myelin_chat(args: Vec) -> ExitCode { Err(error) => ("failed", json!({"error": error.to_string()})), }; let event = xtask_myelin_chat_benchmark_event(run_id, status, detail); - if let Err(error) = - append_synthetic_benchmark_frame(path, "xtask-myelin-chat", "myelin.xtask.benchmark", event) - { + if let Err(error) = append_synthetic_benchmark_frame( + path, + "xtask-myelin-chat", + "myelin.xtask.benchmark", + event, + ) { eprintln!("Failed to write myelin-chat benchmark frame: {error}"); return ExitCode::from(1); } @@ -1098,7 +1106,9 @@ fn terminate_myelin_chat_child(child: &mut Child) -> Result Ok(Some(status)) => return Ok(status), Ok(None) => thread::sleep(Duration::from_millis(MYELIN_CHAT_CHECK_POLL_MS)), Err(error) => { - return Err(format!("myelin-chat-check: poll child after SIGTERM: {error}")); + return Err(format!( + "myelin-chat-check: poll child after SIGTERM: {error}" + )); } } } @@ -1347,7 +1357,9 @@ fn find_stdout_marker( stdout[start..] .find(marker) .map(|offset| start + offset) - .ok_or_else(|| format!("myelin-chat-check: missing {label} marker for prompt cycle {cycle}")) + .ok_or_else(|| { + format!("myelin-chat-check: missing {label} marker for prompt cycle {cycle}") + }) } #[derive(Clone)] @@ -2196,12 +2208,22 @@ fn validate_benchmark_observability( validation.edges_with_consumer.insert(edge_id); } } - ("myelin.chat.prompt", Some("ChatProgress"), Some("prompt_submitted"), Some("ready")) => { + ( + "myelin.chat.prompt", + Some("ChatProgress"), + Some("prompt_submitted"), + Some("ready"), + ) => { if let Some(request_id) = benchmark_request_id(&record.event) { validation.requests_started.insert(request_id); } } - ("myelin.chat.prompt", Some("ChatProgress"), Some("request_completed"), Some("ready")) => { + ( + "myelin.chat.prompt", + Some("ChatProgress"), + Some("request_completed"), + Some("ready"), + ) => { if let Some(request_id) = benchmark_request_id(&record.event) { validation.requests_completed.insert(request_id); } @@ -3945,7 +3967,10 @@ fn pipeline_stage_index(event: &Value) -> Option { .or_else(|| event_u64(event, "role_id").and_then(|role| role.checked_sub(1))) } -fn benchmark_invariants_json(facts: &DumpLogFacts, scenario: MyelinChatCheckScenario) -> Vec { +fn benchmark_invariants_json( + facts: &DumpLogFacts, + scenario: MyelinChatCheckScenario, +) -> Vec { let mut invariants = vec![ invariant_json("chat_config_ready", facts.chat_config_ready), invariant_json("prepare_runtime_ready", facts.prepare_runtime_ready), @@ -4494,7 +4519,9 @@ fn record_gpu_dump_log_event(channel: &str, event: &Value, facts: &mut DumpLogFa { facts.gpu_probe_ready = true; } - ("myelin.worker.initialize", Some("WorkerReady")) if worker_ready_backend_is_cuda(event) => { + ("myelin.worker.initialize", Some("WorkerReady")) + if worker_ready_backend_is_cuda(event) => + { facts.gpu_worker_ready = true; } ("myelin.worker.prompt", Some("DecodeStarted")) @@ -4862,7 +4889,10 @@ mod tests { let baseline = MyelinChatCheckInvocation::parse_args(Vec::new()).expect("default scenario parses"); - assert_eq!(baseline.scenario(), MyelinChatCheckScenario::ProcessBaseline); + assert_eq!( + baseline.scenario(), + MyelinChatCheckScenario::ProcessBaseline + ); assert_eq!( baseline.myelin_chat_args(42, dump_log), strings(&[ @@ -4904,8 +4934,9 @@ mod tests { ]) ); - let multinode_docker = MyelinChatCheckInvocation::parse_args(strings(&["--multinode-docker"])) - .expect("multinode docker parses"); + let multinode_docker = + MyelinChatCheckInvocation::parse_args(strings(&["--multinode-docker"])) + .expect("multinode docker parses"); assert_eq!( multinode_docker.scenario(), MyelinChatCheckScenario::MultinodeDocker @@ -4959,9 +4990,12 @@ mod tests { "--dump-logs=/tmp/myelin-chat-check.ndjson", ]) ); - let vastai_parallel = - MyelinChatCheckInvocation::parse_args(strings(&["--vastai", "--pipeline-parallel", "4"])) - .expect("vastai pipeline-parallel parses"); + let vastai_parallel = MyelinChatCheckInvocation::parse_args(strings(&[ + "--vastai", + "--pipeline-parallel", + "4", + ])) + .expect("vastai pipeline-parallel parses"); assert_eq!( vastai_parallel.myelin_chat_args(42, dump_log), strings(&[ @@ -5597,7 +5631,10 @@ mod tests { include_cpu_fallback: bool, ) -> Vec<(&'static str, Value)> { let mut events = vec![ - ("myelin.chat.lifecycle", chat_span("config", "ready", 1_000, 0)), + ( + "myelin.chat.lifecycle", + chat_span("config", "ready", 1_000, 0), + ), ( "myelin.chat.benchmark", stamped( @@ -6421,8 +6458,9 @@ mod tests { fn benchmark_observability_report_requires_granular_decode_events() { let events = parse_synthetic_events("missing-first-token", benchmark_report_events(false)); - let error = build_benchmark_report(&events, 80, 9, MyelinChatCheckScenario::ProcessBaseline) - .expect_err("missing first token should fail"); + let error = + build_benchmark_report(&events, 80, 9, MyelinChatCheckScenario::ProcessBaseline) + .expect_err("missing first token should fail"); assert!( error.starts_with("myelin-chat-check: missing benchmark event "), @@ -6439,8 +6477,9 @@ mod tests { let events = parse_synthetic_events("pipeline-report", pipeline_benchmark_report_events(true)); - let report = build_benchmark_report(&events, 80, 9, MyelinChatCheckScenario::ProcessBaseline) - .expect("pipeline report builds"); + let report = + build_benchmark_report(&events, 80, 9, MyelinChatCheckScenario::ProcessBaseline) + .expect("pipeline report builds"); assert!( report diff --git a/xtask/src/provisioning_demo/bootstrap.rs b/xtask/src/provisioning_demo/bootstrap.rs new file mode 100644 index 0000000..d88dd84 --- /dev/null +++ b/xtask/src/provisioning_demo/bootstrap.rs @@ -0,0 +1,112 @@ +//! Demo bootstrap logic #1: local OS foreign process. +//! +//! Implements [`BootstrapLogic`] for `"process"`-kind specs: the node is a +//! re-exec'd child of this binary supervised by a `swactor-process` actor +//! (stdio null). Readiness arrives over the control plane: the child's node +//! role joins the supervisor's iroh endpoint and announces itself, and the +//! supervisor's announce relay delivers [`BootstrapMsg::Announce`] to the +//! owning bootstrap actor. This logic only owns process lifecycle (spawn, +//! exit observation, termination). The docker variant sits beside it in the +//! registry — same interface, different foreign process. + +use std::time::SystemTime; + +use swactor::actor::{ActorAddress, Ctx}; +use swactor::runtime::ExternalSender; +use swactor_process::{ProcessOutputConfig, ProcessSpec, spawn_local_process}; + +use provisioning::bootstrap::{BootstrapLogic, LogicProbe, NodeLaunchSpec}; + +use crate::provisioning_demo::provider::{NodeManager, NodeRelayActor, NodeRuntime}; + +/// Local-process bootstrap logic. The spawned process actor's lifecycle +/// reports fold into the shared [`NodeManager`] registry via +/// [`NodeRelayActor`]; probes read that registry only — readiness is the +/// wire announce, not a probe observation. +pub struct LocalProcessLogic { + spec: NodeLaunchSpec, + manager: NodeManager, + process_actor: Option, +} + +impl LocalProcessLogic { + pub fn new(spec: NodeLaunchSpec, manager: NodeManager) -> Self { + Self { + spec, + manager, + process_actor: None, + } + } +} + +impl BootstrapLogic for LocalProcessLogic { + fn start( + &mut self, + ctx: &Ctx, + owner: ActorAddress, + sender: &ExternalSender, + ) -> Result<(), String> { + let attempt = self.spec.attempt; + let relay = ctx + .spawn(NodeRelayActor::new(self.manager.clone(), attempt)) + .map_err(|error| format!("spawn relay actor: {error}"))?; + + let (command, args) = self + .spec + .argv + .split_first() + .map(|(head, tail)| (head.clone(), tail.to_vec())) + .ok_or_else(|| "process spec missing argv".to_owned())?; + let spec = ProcessSpec { + command, + args, + env: self.spec.env.clone().into_iter().collect(), + working_dir: self.spec.workdir.clone(), + label: self.spec.label.clone(), + }; + let output = ProcessOutputConfig::disabled(relay); + let process_actor = spawn_local_process(ctx, sender, spec, output) + .map_err(|error| format!("spawn process actor: {error}"))?; + self.process_actor = Some(process_actor); + + self.manager.register(NodeRuntime { + attempt, + logical_node: self.spec.logical_node.clone(), + bootstrap: owner, + pid: None, + exited: None, + spawn_failed: None, + last_announce_ms: None, + }); + Ok(()) + } + + fn probe(&mut self, _now: SystemTime) -> LogicProbe { + let Some(runtime) = self.manager.get(self.spec.attempt) else { + // Deregistered (lease destroyed / supervisor teardown): the + // node is gone by definition. The relay's exit observation can + // lose the race against deregistration, so this is the + // reliable terminal report. The actor's phase machine turns it + // into Failed (before join) or Exited (after). + return LogicProbe::Exited("deregistered (lease destroyed)".to_owned()); + }; + if let Some(error) = &runtime.spawn_failed { + return LogicProbe::Failed(format!("node process failed to spawn: {error}")); + } + if let Some(status) = &runtime.exited { + return LogicProbe::Exited(format!("{status:?}")); + } + // Still coming up; readiness is the wire announce. + LogicProbe::Pending + } + + fn terminate(&mut self, sender: &ExternalSender, kill_after: Option) { + if let Some(process_actor) = self.process_actor { + let _ = swactor_process::send_process_command( + sender, + process_actor, + swactor_process::ProcessCommand::Stop { kill_after }, + ); + } + } +} diff --git a/xtask/src/provisioning_demo/docker.rs b/xtask/src/provisioning_demo/docker.rs new file mode 100644 index 0000000..540b830 --- /dev/null +++ b/xtask/src/provisioning_demo/docker.rs @@ -0,0 +1,428 @@ +//! Demo bootstrap logic #2: docker container as foreign node. +//! +//! Implements [`BootstrapLogic`] for `"docker"`-kind specs: the node is a +//! `scratch` container running the statically-linked xtask binary in node +//! role, launched attached (`docker run --rm`) so the supervised `docker` +//! CLI child's lifetime tracks the container's — its exit *is* the node +//! exit. Readiness is the same control-plane announce as the process kind. +//! +//! Foreign-node masking: every container runs on a per-run user-defined +//! bridge network, so each node gets its own bridge IP and the supervisor is +//! reached through the bridge gateway — no localhost shortcuts, UDP +//! hole-punching over a real (if virtual) network. +//! +//! Termination always goes through `docker rm -f` (force-remove): signals to +//! the attached CLI are unreliable proxies, and a force-remove both kills +//! the container and satisfies `--rm` cleanup. Zero wastage by construction: +//! no volumes, no mounts, `--rm` containers, and label-filtered startup + +//! exit sweeps that remove anything a SIGKILLed supervisor left behind. +//! Images persist across runs (rebuilds are content-addressed by run token). + +use std::path::Path; +use std::time::SystemTime; + +use swactor::actor::{ActorAddress, Ctx}; +use swactor::runtime::ExternalSender; +use swactor_process::{ProcessOutputConfig, ProcessSpec, spawn_local_process}; + +use provisioning::bootstrap::{BootstrapLogic, LogicProbe, NodeLaunchSpec}; + +use crate::provisioning_demo::provider::{NodeManager, NodeRelayActor, NodeRuntime}; + +/// Generic label present on every demo container/network (sweep key). +pub const SWEEP_LABEL: &str = "swactor-demo"; +/// Per-run label value: only this run's resources. +pub const RUN_LABEL: &str = "swactor-demo-run"; +/// Image repository (tagged per run token). +pub const IMAGE_REPO: &str = "swactor-demo-node"; +/// Static-musl target the node image is built from. +pub const IMAGE_TARGET: &str = "x86_64-unknown-linux-musl"; + +const DOCKERFILE: &str = include_str!("docker/Dockerfile"); + +/// Everything the docker launch style needs, produced by [`preflight`]. +#[derive(Clone)] +pub struct DockerLaunch { + /// Fully-qualified image ref (`swactor-demo-node:`). + pub image: String, + /// Per-run user-defined bridge network name. + pub network: String, + /// Per-run label value used by the exit sweep. + pub run_token: String, + /// Serde `iroh::EndpointAddr` of the supervisor, rebuilt to advertise + /// the bridge gateway address (containers cannot use the host's + /// localhost or LAN addrs meaningfully). + pub supervisor_addr_json: String, +} + +/// Container name for one provision attempt (unique per attempt). +pub fn container_name(attempt: u64) -> String { + format!("{SWEEP_LABEL}-node-{attempt}") +} + +/// Docker bootstrap logic. Identical lifecycle shape to the local process +/// kind — the supervised child is the attached `docker run` CLI; the +/// container is force-removed on every terminal path (terminate, exit, +/// spawn failure, deregistration) so it can never outlive the attempt. +pub struct DockerProcessLogic { + spec: NodeLaunchSpec, + manager: NodeManager, + process_actor: Option, + /// The attempt's container has been force-removed (or removal was + /// spawned); guards idempotency across probe/terminate paths. + removed: bool, +} + +impl DockerProcessLogic { + pub fn new(spec: NodeLaunchSpec, manager: NodeManager) -> Self { + Self { + spec, + manager, + process_actor: None, + removed: false, + } + } + + /// Idempotent force-remove of this attempt's container (kills it if + /// running, satisfies --rm, releases the name; no-op if already gone). + fn force_remove(&mut self) { + if self.removed { + return; + } + self.removed = true; + let name = container_name(self.spec.attempt); + std::thread::spawn(move || { + let status = std::process::Command::new("docker") + .args(["rm", "-f", "--", &name]) + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .status(); + if let Err(error) = status { + eprintln!("demo: docker rm -f {name} failed: {error}"); + } + }); + } +} + +impl BootstrapLogic for DockerProcessLogic { + fn start( + &mut self, + ctx: &Ctx, + owner: ActorAddress, + sender: &ExternalSender, + ) -> Result<(), String> { + let attempt = self.spec.attempt; + let relay = ctx + .spawn(NodeRelayActor::new(self.manager.clone(), attempt)) + .map_err(|error| format!("spawn relay actor: {error}"))?; + + let (command, args) = self + .spec + .argv + .split_first() + .map(|(head, tail)| (head.clone(), tail.to_vec())) + .ok_or_else(|| "docker spec missing argv".to_owned())?; + let spec = ProcessSpec { + command, + args, + env: self.spec.env.clone().into_iter().collect(), + working_dir: self.spec.workdir.clone(), + label: self.spec.label.clone(), + }; + let output = ProcessOutputConfig::disabled(relay); + let process_actor = spawn_local_process(ctx, sender, spec, output) + .map_err(|error| format!("spawn docker process actor: {error}"))?; + self.process_actor = Some(process_actor); + + self.manager.register(NodeRuntime { + attempt, + logical_node: self.spec.logical_node.clone(), + bootstrap: owner, + pid: None, + exited: None, + spawn_failed: None, + last_announce_ms: None, + }); + Ok(()) + } + + fn probe(&mut self, _now: SystemTime) -> LogicProbe { + let Some(runtime) = self.manager.get(self.spec.attempt) else { + self.force_remove(); + return LogicProbe::Exited("deregistered (lease destroyed)".to_owned()); + }; + if let Some(error) = &runtime.spawn_failed { + self.force_remove(); + return LogicProbe::Failed(format!("docker run failed to spawn: {error}")); + } + if let Some(status) = &runtime.exited { + // The attached CLI's exit mirrors the container's (or is the + // launch failure itself); the actor's phase machine classifies + // Failed-vs-Exited against the announce. Either way the + // container must not outlive the attempt: a SIGKILLed CLI + // orphans a *running* container (signal-proxy never fired), so + // the exit path force-removes too — not just terminate(). + self.force_remove(); + return LogicProbe::Exited(format!("docker run {status:?}")); + } + LogicProbe::Pending + } + + fn terminate(&mut self, sender: &ExternalSender, kill_after: Option) { + // Force-remove the container: kills it regardless of signal-proxy + // semantics, satisfies --rm, and releases the name. The attached + // CLI child then exits on its own; the explicit Stop below is + // actor-tree hygiene (the kill_after escalation applies to the + // CLI, not the container). + self.force_remove(); + if let Some(process_actor) = self.process_actor { + let _ = swactor_process::send_process_command( + sender, + process_actor, + swactor_process::ProcessCommand::Stop { kill_after }, + ); + } + } +} + +// ─── Preflight: image + network ───────────────────────────────────────────── + +/// Build the node image and the per-run network: sweep stale demo resources, +/// compile the static-musl xtask binary, `docker build` it from a staging +/// dir, create the labeled bridge network, and resolve the gateway address +/// the containers will dial the supervisor on. +pub fn preflight( + root: &Path, + supervisor_pubkey: iroh::PublicKey, + port: u16, +) -> Result { + docker_version()?; + let run_token = format!("{}-{}", std::process::id(), unix_ms()); + sweep_stale(); + let image = format!("{IMAGE_REPO}:{run_token}"); + build_image(root, &image)?; + let network = format!("{SWEEP_LABEL}-net-{run_token}"); + docker_ok( + &[ + "network", + "create", + "--label", + &format!("{SWEEP_LABEL}=1"), + "--label", + &format!("{RUN_LABEL}={run_token}"), + "--", + &network, + ], + "create demo network", + )?; + let gateway = docker_output( + &[ + "network", + "inspect", + "--format", + "{{(index .IPAM.Config 0).Gateway}}", + "--", + &network, + ], + "read demo network gateway", + )?; + let gateway_ip: std::net::IpAddr = gateway + .trim() + .parse() + .map_err(|error| format!("network gateway {gateway:?}: {error}"))?; + let supervisor_addr = iroh::EndpointAddr::new(supervisor_pubkey) + .with_ip_addr(std::net::SocketAddr::new(gateway_ip, port)); + let supervisor_addr_json = serde_json::to_string(&supervisor_addr) + .map_err(|error| format!("serialize supervisor addr: {error}"))?; + Ok(DockerLaunch { + image, + network, + run_token, + supervisor_addr_json, + }) +} + +/// Remove every container and network carrying the generic demo label. All +/// such resources found here are stale: this runs before the current run +/// creates anything, so anything matched is a leftover (e.g. of a SIGKILLed +/// supervisor). One demo instance at a time. +fn sweep_stale() { + let containers = docker_output( + &["ps", "-aq", "--filter", &format!("label={SWEEP_LABEL}=1")], + "list stale demo containers", + ) + .unwrap_or_default(); + let ids: Vec = containers + .lines() + .map(str::trim) + .filter(|line| !line.is_empty()) + .map(str::to_owned) + .collect(); + if !ids.is_empty() { + let mut args = vec!["rm".to_owned(), "-f".to_owned()]; + args.extend(ids.iter().map(|id| id.to_string())); + let refs: Vec<&str> = args.iter().map(String::as_str).collect(); + let _ = docker_ok(&refs, "remove stale demo containers"); + } + let networks = docker_output( + &[ + "network", + "ls", + "-q", + "--filter", + &format!("label={SWEEP_LABEL}=1"), + ], + "list stale demo networks", + ) + .unwrap_or_default(); + let nets: Vec = networks + .lines() + .map(str::trim) + .filter(|line| !line.is_empty()) + .map(str::to_owned) + .collect(); + if !nets.is_empty() { + let mut args = vec!["network".to_owned(), "rm".to_owned()]; + args.extend(nets.iter().map(|id| id.to_string())); + let refs: Vec<&str> = args.iter().map(String::as_str).collect(); + let _ = docker_ok(&refs, "remove stale demo networks"); + } +} + +/// Exit-path sweep: remove this run's containers (by run token) and its +/// network. The image persists (per-run tokens make old images trivially +/// identifiable; they are tiny scratch images). +pub fn sweep_run(launch: &DockerLaunch) { + let containers = docker_output( + &[ + "ps", + "-aq", + "--filter", + &format!("label={RUN_LABEL}={}", launch.run_token), + ], + "list run containers", + ) + .unwrap_or_default(); + let ids: Vec<&str> = containers + .lines() + .map(str::trim) + .filter(|l| !l.is_empty()) + .collect(); + if !ids.is_empty() { + let mut args = vec!["rm".to_owned(), "-f".to_owned()]; + args.extend(ids.iter().map(|id| id.to_string())); + let refs: Vec<&str> = args.iter().map(String::as_str).collect(); + let _ = docker_ok(&refs, "remove run containers"); + } + let _ = docker_ok( + &["network", "rm", "--", &launch.network], + "remove run network", + ); +} + +/// Build the scratch image: stage the static binary + Dockerfile in a temp +/// dir (keeps the build context to one file), then `docker build`. +fn build_image(root: &Path, image: &str) -> Result<(), String> { + println!("provisioning-reconciler-demo --docker: building static node binary…"); + let bin = root + .join("target") + .join(IMAGE_TARGET) + .join("release") + .join("xtask"); + // Build when the static binary is missing; DEMO_DOCKER_REBUILD=1 forces + // a rebuild (e.g. after source changes). + let force_rebuild = std::env::var("DEMO_DOCKER_REBUILD").as_deref() == Ok("1"); + if force_rebuild || !bin.exists() { + run_cargo_build(root)?; + } + if !bin.exists() { + return Err(format!( + "node binary missing after build: {}", + bin.display() + )); + } + let staging = std::env::temp_dir().join(format!("{SWEEP_LABEL}-image-{}", unix_ms())); + std::fs::create_dir_all(&staging).map_err(|e| format!("staging dir: {e}"))?; + let result = (|| { + std::fs::copy(&bin, staging.join("xtask")).map_err(|e| format!("stage binary: {e}"))?; + std::fs::write(staging.join("Dockerfile"), DOCKERFILE) + .map_err(|e| format!("stage Dockerfile: {e}"))?; + docker_ok( + &["build", "-q", "-t", image, "--", &staging.to_string_lossy()], + "build demo node image", + ) + })(); + let _ = std::fs::remove_dir_all(&staging); + result.map(|_| { + println!("provisioning-reconciler-demo --docker: image {image} ready"); + }) +} + +fn run_cargo_build(root: &Path) -> Result<(), String> { + let output = std::process::Command::new("cargo") + .args([ + "build", + "--release", + "--target", + IMAGE_TARGET, + "--package", + "xtask", + ]) + .current_dir(root) + .stdin(std::process::Stdio::null()) + .output() + .map_err(|e| format!("run cargo: {e}"))?; + if !output.status.success() { + let mut stderr = String::from_utf8_lossy(&output.stderr).into_owned(); + if stderr.len() > 4000 { + stderr.truncate(4000); + } + return Err(format!("cargo build for {IMAGE_TARGET} failed:\n{stderr}")); + } + Ok(()) +} + +/// Docker CLI + daemon reachable? +fn docker_version() -> Result<(), String> { + docker_output( + &["version", "--format", "{{.Server.Version}}"], + "docker daemon", + ) + .map(|_| ()) +} + +fn docker_ok(args: &[&str], label: &str) -> Result<(), String> { + let output = docker_raw(args)?; + if output.status.success() { + Ok(()) + } else { + let stderr = String::from_utf8_lossy(&output.stderr); + Err(format!("{label} failed: {stderr}")) + } +} + +fn docker_output(args: &[&str], label: &str) -> Result { + let output = docker_raw(args)?; + if output.status.success() { + Ok(String::from_utf8_lossy(&output.stdout).into_owned()) + } else { + let stderr = String::from_utf8_lossy(&output.stderr); + Err(format!("{label} failed: {stderr}")) + } +} + +fn docker_raw(args: &[&str]) -> Result { + std::process::Command::new("docker") + .args(args) + .stdin(std::process::Stdio::null()) + .output() + .map_err(|e| format!("run docker {args:?}: {e}")) +} + +fn unix_ms() -> u64 { + SystemTime::now() + .duration_since(SystemTime::UNIX_EPOCH) + .map(|duration| duration.as_millis() as u64) + .unwrap_or(0) +} diff --git a/xtask/src/provisioning_demo/docker/Dockerfile b/xtask/src/provisioning_demo/docker/Dockerfile new file mode 100644 index 0000000..6004db7 --- /dev/null +++ b/xtask/src/provisioning_demo/docker/Dockerfile @@ -0,0 +1,9 @@ +# Demo node image: the static-musl xtask binary and nothing else. +# Staged into a one-file build context by +# xtask/src/provisioning_demo/docker.rs (`build_image`); never built from +# the repo root. The binary is fully static (musl + ring), so `scratch` +# needs no libc, no CA bundle (relay disabled; direct iroh only), and no +# shell — kill signals reach PID 1 = the node role directly. +FROM scratch +COPY xtask /xtask +ENTRYPOINT ["/xtask"] diff --git a/xtask/src/provisioning_demo/feed.rs b/xtask/src/provisioning_demo/feed.rs index 6d0f655..b8dd6e1 100644 --- a/xtask/src/provisioning_demo/feed.rs +++ b/xtask/src/provisioning_demo/feed.rs @@ -4,36 +4,30 @@ //! Each tick mirrors the production `ClusterReconciler` poll semantics: //! drain executor results (closing bootstrap sessions after convergence), //! classify due operations, requeue, drive until blocked — then feeds the -//! real world back in (key-file observations, iroh join checks, child exits), -//! emits `prov.reconciler.*` telemetry, and drains every telemetry endpoint -//! into the dashboard. +//! real world back in (wire announces, child exits), emits +//! `prov.reconciler.*` telemetry, and drains every telemetry endpoint into +//! the dashboard. use std::collections::BTreeMap; use std::time::{Duration, SystemTime}; +use provisioning::executor::{ + BlockingEffectSpawner, BlockingEffectWork, ExecutorOperationStatus, IdempotentEffectExecutor, +}; +use provisioning::node::{ + BootstrapObservation, BootstrapStage, NodeGroupId, NodeStage, RoleId, RunId, SwactorId, +}; +use provisioning::reconciler::{ClusterDriver, EffectExecutor}; +use provisioning::reconciler::{ + ClusterShape, NodeObservation, OperationOutcome, PlannedEffect, RetryPolicy, +}; use serde_json::json; use swactor::actor::{ActorInterface, Ctx}; use swactor_engine::EngineHandle; -use swactor_process::{spawn_local_process, ProcessOutputConfig, ProcessSpec}; - -use provisioning::executor::{ - BlockingEffectSpawner, BlockingEffectWork, ExecutorOperationStatus, - IdempotentEffectExecutor, -}; -use provisioning::node::{ - BootstrapObservation, BootstrapStage, NodeGroupId, - NodeStage, RoleId, RunId, SwactorId, -}; -use provisioning::reconciler::{ - ClusterShape, NodeObservation, RetryPolicy, PlannedEffect, OperationOutcome, -}; -use provisioning::reconciler::{ClusterDriver, EffectExecutor}; use telemetry::{ChannelContent, StreamDescriptor, TelemetryEndpoint, TelemetryProducer}; -use crate::provisioning_demo::node::read_key_report; use crate::provisioning_demo::provider::{ - register_node_channels, unix_ms, DemoBackend, NodeManager, NodeRelayActor, - NodeTelemetry, + DemoBackend, NodeManager, NodeTelemetry, register_node_channels, unix_ms, }; /// Supervisor telemetry: channels + name resolution for the dashboard path. @@ -136,12 +130,32 @@ pub enum SupervisorMsg { Tick, Control(dashboard::control::ControlCommand), Spawn(crate::provisioning_demo::provider::SpawnNodeRequest), + /// Event from a per-node bootstrap actor. + Bootstrap(provisioning::BootstrapEvent), + /// A remote node's telemetry pull stream registered its header (stream + /// descriptor + channel catalog), associated with its logical node and + /// provision attempt. Frames are fused onto the logical node's stream. + NodeStream { + header: iroh_driver::TelemetryQuicHeader, + logical_node: String, + attempt: u64, + }, /// Drain the cluster: desired → empty, stop every child, flag when done. Shutdown { drained: std::sync::Arc, }, } +/// Metadata for one remote node telemetry pull, keyed by the node's hex +/// transport key: channel catalog + which logical node stream its frames +/// are fused onto. +#[derive(Clone, Default)] +struct RemoteStreamMeta { + logical_node: String, + attempt: u64, + channels: BTreeMap, +} + /// Per-node telemetry handle kept while the attempt is live. struct NodeStreams { telemetry: NodeTelemetry, @@ -158,6 +172,18 @@ pub struct SupervisorActor { pub events_channel: telemetry::ChannelId, pub snapshot_channel: telemetry::ChannelId, pub sender: swactor::runtime::ExternalSender, + /// Bootstrap kind registry (spec.kind → logic). + registry: provisioning::BootstrapRegistry, + /// Pull collector fired by bootstrap actors on Bootstrapped. + collector: std::sync::Arc, + /// Engine handle for bootstrap actor probe intervals. + engine: EngineHandle, + /// Subscription draining the remote-stream fanout into the dashboard. + remote_sub: telemetry::TelemetrySubscription, + remote_streams: BTreeMap, + /// Bootstrap reports parked until the reconciler opens the attempt's + /// bootstrap session (early joins / failures). + pending_reports: BTreeMap, /// Desired shape slots: singleton groups, one per logical node. slots: Vec, slot_seq: u64, @@ -167,7 +193,7 @@ pub struct SupervisorActor { last_stages: BTreeMap)>, pub dashboard: dashboard::DashboardHandle, status_tick: u64, - exe: std::path::PathBuf, + launch: crate::provisioning_demo::LaunchStyle, } impl SupervisorActor { @@ -180,9 +206,13 @@ impl SupervisorActor { mut telemetry: SupervisorTelemetry, dashboard: dashboard::DashboardHandle, sender: swactor::runtime::ExternalSender, + registry: provisioning::BootstrapRegistry, + collector: std::sync::Arc, + engine: EngineHandle, + remote_sub: telemetry::TelemetrySubscription, initial_slots: Vec, run_id: RunId, - exe: std::path::PathBuf, + launch: crate::provisioning_demo::LaunchStyle, ) -> Self { let events_channel = telemetry.register("prov.reconciler.events"); let snapshot_channel = telemetry.register("prov.reconciler.snapshot"); @@ -195,6 +225,12 @@ impl SupervisorActor { events_channel, snapshot_channel, sender, + registry, + collector, + engine, + remote_sub, + remote_streams: BTreeMap::new(), + pending_reports: BTreeMap::new(), nodes: BTreeMap::new(), run_id, slot_seq: initial_slots.len() as u64, @@ -203,7 +239,7 @@ impl SupervisorActor { last_stages: BTreeMap::new(), dashboard, status_tick: 0, - exe, + launch, } } @@ -223,49 +259,103 @@ impl SupervisorActor { "detail": detail, }); let bytes = serde_json::to_vec(&payload).expect("event serializes"); - self.telemetry.producer.submit_bytes(self.events_channel, bytes); + self.telemetry + .producer + .submit_bytes(self.events_channel, bytes); } - /// Handle a spawn request from the provider (runs in actor context). - fn spawn_node(&mut self, ctx: &Ctx, request: crate::provisioning_demo::provider::SpawnNodeRequest) { + /// Handle a spawn request from the provider (runs in actor context): + /// spawn the per-attempt bootstrap actor through the registry — the + /// supervision never launches nodes directly. + fn spawn_node( + &mut self, + ctx: &Ctx, + request: crate::provisioning_demo::provider::SpawnNodeRequest, + ) { let attempt = request.attempt; - let relay = match ctx.spawn(NodeRelayActor::new(self.manager.clone(), attempt)) { - Ok(addr) => addr, - Err(error) => { - let _ = request.reply.send(Err(format!("spawn relay actor: {error}"))); - return; - } - }; - - let spec = ProcessSpec { - command: self.exe.to_string_lossy().to_string(), - args: vec![ - "provisioning-reconciler-demo".to_owned(), - "--demo-node".to_owned(), - self.driver_handle.supervisor_addr_json.clone(), - ], - env: [ - ( - "DEMO_NODE_KEY_FILE".to_owned(), - request.key_file.to_string_lossy().to_string(), - ), - ("DEMO_NODE_ID".to_owned(), request.logical_node.clone()), - ] - .into_iter() - .collect(), - working_dir: None, - label: Some(request.logical_node.clone()), - }; + // Supervisor-authored lifecycle stream (kept alive independent of + // the node: it reports death even when the node cannot). self.node_life += 1; let telemetry = NodeTelemetry::new(&request.logical_node, self.node_life); let status_channel = register_node_channels(&telemetry.producer); - // The lifecycle channel is registered by the process crate with the - // sanitized label; mirror it for name resolution during drain. - let output = - ProcessOutputConfig::telemetry_mirror(relay, telemetry.producer.clone()); - match spawn_local_process(ctx, &self.sender, spec, output) { - Ok(process_actor) => { + + let (kind, argv, mut env) = match &self.launch { + crate::provisioning_demo::LaunchStyle::Process { exe } => ( + "process", + vec![ + exe.to_string_lossy().to_string(), + "provisioning-reconciler-demo".to_owned(), + "--demo-node".to_owned(), + self.driver_handle.supervisor_addr_json.clone(), + "--demo-attempt".to_owned(), + attempt.to_string(), + ], + Vec::new(), + ), + crate::provisioning_demo::LaunchStyle::Docker(docker) => ( + "docker", + vec![ + "docker".to_owned(), + "run".to_owned(), + "--rm".to_owned(), + "--name".to_owned(), + crate::provisioning_demo::docker::container_name(attempt), + "--label".to_owned(), + format!("{}=1", crate::provisioning_demo::docker::SWEEP_LABEL), + "--label".to_owned(), + format!( + "{}={}", + crate::provisioning_demo::docker::RUN_LABEL, + docker.run_token + ), + "--network".to_owned(), + docker.network.clone(), + docker.image.clone(), + "provisioning-reconciler-demo".to_owned(), + "--demo-node".to_owned(), + docker.supervisor_addr_json.clone(), + "--demo-attempt".to_owned(), + attempt.to_string(), + ], + Vec::new(), + ), + }; + env.push(("DEMO_NODE_ID".to_owned(), request.logical_node.clone())); + + let spec = provisioning::NodeLaunchSpec { + kind: kind.to_owned(), + attempt, + logical_node: request.logical_node.clone(), + argv, + env, + workdir: None, + label: Some(request.logical_node.clone()), + }; + + let reporter: provisioning::BootstrapReporter = { + let sender = self.sender.clone(); + let supervisor = ctx.self_addr(); + std::sync::Arc::new(move |event| { + let _ = sender.send_to(supervisor, SupervisorMsg::Bootstrap(event)); + }) + }; + let config = provisioning::BootstrapConfig { + reporter, + sender: self.sender.clone(), + collector: Some(std::sync::Arc::clone(&self.collector)), + probe_period: provisioning::DEFAULT_PROBE_PERIOD, + spec: spec.clone(), + }; + let logic = match self.registry.create(&spec) { + Ok(logic) => logic, + Err(error) => { + let _ = request.reply.send(Err(format!("bootstrap logic: {error}"))); + return; + } + }; + match provisioning::spawn_bootstrap_actor(ctx, &self.engine, logic, config) { + Ok(bootstrap) => { self.nodes.insert( attempt, NodeStreams { @@ -273,36 +363,27 @@ impl SupervisorActor { status_channel, }, ); - self.manager.register( - crate::provisioning_demo::provider::NodeRuntime { - attempt, - logical_node: request.logical_node.clone(), - process_actor, - key_file: request.key_file.clone(), - pid: None, - exited: None, - spawn_failed: None, - }, - ); - let _ = request.reply.send(Ok( - crate::provisioning_demo::provider::NodeRuntime { - attempt, - logical_node: request.logical_node, - process_actor, - key_file: request.key_file, - pid: None, - exited: None, - spawn_failed: None, - }, - )); + let runtime = crate::provisioning_demo::provider::NodeRuntime { + attempt, + logical_node: request.logical_node.clone(), + bootstrap, + pid: None, + exited: None, + spawn_failed: None, + last_announce_ms: None, + }; + let _ = request.reply.send(Ok(runtime)); } Err(error) => { - let _ = request.reply.send(Err(format!("spawn process actor: {error}"))); + let _ = request + .reply + .send(Err(format!("spawn bootstrap actor: {error}"))); } } } - /// Feed real-world observations into the driver. + /// Feed bootstrap-progress observations into the driver. Join detection + /// and exit classification moved into the per-node bootstrap actors fn observe_world(&mut self, now: SystemTime) { let node_ids: Vec = self .driver @@ -312,87 +393,246 @@ impl SupervisorActor { .map(|id| id.0.clone()) .collect(); for node_id in node_ids { - let Some(managed) = self.driver.state().nodes.get(&provisioning::node::LogicalNodeId(node_id.clone())) else { + let Some(managed) = self + .driver + .state() + .nodes + .get(&provisioning::node::LogicalNodeId(node_id.clone())) + else { continue; }; let attempt = managed.attempt; - let stage = managed.record.stage; - let active_bootstrap = managed.active_bootstrap; - let runtime = match self.manager.get(attempt.0) { - Some(runtime) => runtime, - None => continue, + let Some(session_id) = managed.active_bootstrap else { + continue; }; + let Some(runtime) = self.manager.get(attempt.0) else { + continue; + }; + // Stage evidence: the supervised child started (process pid / + // docker CLI pid observed) means the node runtime is coming up + // and we are waiting for its control-plane announce; before + // that the lease exists but the foreign process is not up yet. + let stage_seen = if runtime.pid.is_some() { + BootstrapStage::WaitingForSwactorJoin + } else { + BootstrapStage::SshReady + }; + self.driver.apply_observation( + &provisioning::node::LogicalNodeId(node_id.clone()), + attempt, + NodeObservation::BootstrapObserved { + session_id, + observation: BootstrapObservation::stage(stage_seen), + }, + now, + ); + } + } - // Child exit or spawn failure: fail the attempt while it is - // still bootstrapping so the reconciler retries with a fresh - // lease instead of wedging at SshReady forever. - if runtime.exited.is_some() || runtime.spawn_failed.is_some() { - if stage != NodeStage::Failed && active_bootstrap.is_some() { - let reason = if let Some(status) = &runtime.exited { - format!("node process exited: {status:?}") - } else { - format!( - "node process failed to spawn: {}", - runtime.spawn_failed.as_deref().unwrap_or("unknown") - ) - }; - self.emit_event("observation", &node_id, reason.clone()); + /// Fold bootstrap-actor events into the reconciler. A report that + /// arrives before the reconciler has an active bootstrap session for + /// the attempt (the actor starts at spawn, `StartBootstrap` comes + /// later) is parked and re-delivered on the next tick — the join signal + /// is exactly-once, so dropping an early one wedges the session. + fn handle_bootstrap_event(&mut self, now: SystemTime, event: provisioning::BootstrapEvent) { + let attempt = match &event { + provisioning::BootstrapEvent::Bootstrapped(identity) => identity.attempt, + provisioning::BootstrapEvent::Failed { attempt, .. } + | provisioning::BootstrapEvent::Exited { attempt, .. } => *attempt, + }; + if !self.deliver_bootstrap_report(now, event.clone()) { + self.pending_reports.insert(attempt, event); + } + } + + /// Deliver one bootstrap report to the reconciler driver. Returns false + /// when the attempt has no active bootstrap session yet. + fn deliver_bootstrap_report( + &mut self, + now: SystemTime, + event: provisioning::BootstrapEvent, + ) -> bool { + match event { + provisioning::BootstrapEvent::Bootstrapped(identity) => { + let Some((node_id, session_id)) = self.node_for_attempt(identity.attempt) else { + return false; + }; + let key_prefix = &identity.key_hex[..8.min(identity.key_hex.len())]; + self.emit_event( + "observation", + &node_id, + format!("swactor join confirmed (key {key_prefix}…)"), + ); + self.driver.apply_observation( + &provisioning::node::LogicalNodeId(node_id), + provisioning::NodeAttemptId(identity.attempt), + NodeObservation::SwactorJoined { + session_id, + swactor_id: SwactorId(identity.key_hex), + }, + now, + ); + } + provisioning::BootstrapEvent::Failed { attempt, reason } => { + let Some((node_id, session_id)) = self.node_for_attempt(attempt) else { + return false; + }; + self.emit_event("observation", &node_id, reason.clone()); + self.driver.apply_observation( + &provisioning::node::LogicalNodeId(node_id), + provisioning::NodeAttemptId(attempt), + NodeObservation::BootstrapFailed { session_id, reason }, + now, + ); + } + provisioning::BootstrapEvent::Exited { attempt, reason } => { + // An exit while a session is still open is a bootstrap + // failure (death before join); otherwise it is a plain + // death observation. + if let Some((node_id, session_id)) = self.node_for_attempt(attempt) { + self.emit_event("observation", &node_id, format!("node runtime: {reason}")); self.driver.apply_observation( - &provisioning::node::LogicalNodeId(node_id.clone()), - attempt, + &provisioning::node::LogicalNodeId(node_id), + provisioning::NodeAttemptId(attempt), NodeObservation::BootstrapFailed { - session_id: active_bootstrap.expect("checked above"), - reason, + session_id, + reason: format!("node process exited: {reason}"), }, now, ); } - continue; - } - - // Bootstrap progression from the key file and iroh join state. - if let Some(session_id) = active_bootstrap { - let report = read_key_report(&runtime.key_file); - let mut stage_seen = BootstrapStage::SshReady; - if let Some(report) = &report { - let connected = report - .node_hex - .parse_key() - .is_some_and(|key| self.driver_handle.has_active_connection(key)); - if connected { - let heartbeat_age = unix_ms(now).saturating_sub(report.last_seen_ms); - self.emit_event( - "observation", - &node_id, - format!( - "swactor join confirmed (key {}…, heartbeat {}ms old)", - &report.node_hex[..8.min(report.node_hex.len())], - heartbeat_age - ), - ); - let swactor_id = SwactorId(report.node_hex.clone()); - self.driver.apply_observation( - &provisioning::node::LogicalNodeId(node_id.clone()), - attempt, - NodeObservation::SwactorJoined { - session_id, - swactor_id, - }, - now, - ); - continue; + // Publish the terminal status now: once the reconciler + // destroys the lease the runtime is deregistered, and the + // Fleet Control table would otherwise keep its last + // "running" state forever. Derive the logical name from the + // lifecycle stream itself — the registry entry can already + // be gone (lease destroy races the bootstrap probe). + if let Some(streams) = self.nodes.get(&attempt) { + let logical = streams + .telemetry + .endpoint + .stream_id() + .node + .as_str() + .to_owned(); + let pid = self.manager.get(attempt).and_then(|r| r.pid); + let payload = json!({ + "at_ms": unix_ms(now), + "node": logical, + "alive": false, + "pid": pid, + "event": "exited", + }); + if let Ok(bytes) = serde_json::to_vec(&payload) { + streams + .telemetry + .producer + .submit_bytes(streams.status_channel, bytes); } - stage_seen = BootstrapStage::WaitingForSwactorJoin; } - self.driver.apply_observation( - &provisioning::node::LogicalNodeId(node_id.clone()), - attempt, - NodeObservation::BootstrapObserved { - session_id, - observation: BootstrapObservation::stage(stage_seen), - }, - now, - ); + // Death-replacement is shape logic: `replace_dead_ready_nodes` + // reads the exit from the shared registry. + } + } + true + } + + /// Re-deliver parked bootstrap reports whose session has appeared. + fn deliver_pending_reports(&mut self, now: SystemTime) { + let attempts: Vec = self.pending_reports.keys().copied().collect(); + for attempt in attempts { + let Some(event) = self.pending_reports.get(&attempt).cloned() else { + continue; + }; + if self.deliver_bootstrap_report(now, event) { + self.pending_reports.remove(&attempt); + } + } + } + + /// Resolve a driver node (id + active bootstrap session) by attempt. + fn node_for_attempt( + &self, + attempt: u64, + ) -> Option<(String, provisioning::node::BootstrapSessionId)> { + self.driver + .state() + .nodes + .iter() + .find(|(_, managed)| managed.attempt.0 == attempt) + .and_then(|(id, managed)| { + managed + .active_bootstrap + .map(|session| (id.0.clone(), session)) + }) + } + + /// Register a remote node telemetry stream header (pull side). + fn register_remote_stream( + &mut self, + header: iroh_driver::TelemetryQuicHeader, + logical_node: String, + attempt: u64, + ) { + let key = header.stream.stream.node.as_str().to_string(); + let meta = self.remote_streams.entry(key.clone()).or_default(); + meta.logical_node = logical_node.clone(); + meta.attempt = attempt; + for channel in &header.channels { + meta.channels.insert(channel.id, channel.name.clone()); + } + println!( + "demo: fusing telemetry of node {logical_node} (stream {key}, {} channels)", + meta.channels.len() + ); + } + /// Drain remote-node telemetry frames into the dashboard, fusing them + /// onto the logical node's stream: one card per node, carrying both the + /// supervisor-authored lifecycle channels and the node's real runtime + /// channels (runtime.actors, node.beat, node.status). + fn flush_remote_streams(&mut self) { + for event in self.remote_sub.drain_available() { + match event { + telemetry::frame::TelemetryEvent::Frame(delivery) => { + let key = delivery.channel.stream.node.as_str().to_string(); + let Some(meta) = self.remote_streams.get(&key).cloned() else { + continue; + }; + let Some(node_stream) = self.nodes.get(&meta.attempt) else { + continue; + }; + // Fuse: republish on the logical node's stream. + let target_stream = node_stream.telemetry.endpoint.stream_id().clone(); + let origin = node_stream.telemetry.origin; + let label = node_stream.telemetry.label.clone(); + let channel_name = meta + .channels + .get(&delivery.channel.channel) + .cloned() + .unwrap_or_else(|| format!("channel#{}", delivery.channel.channel.0)); + let frame = telemetry::frame::Frame { + channel: delivery.channel.channel, + position: delivery.position, + payload: delivery.payload, + }; + publish_frame( + &self.dashboard, + &target_stream, + &channel_name, + &frame, + origin, + &label, + ); + } + telemetry::frame::TelemetryEvent::ChannelDeclared(descriptor) => { + let key = descriptor.stream.node.as_str().to_string(); + self.remote_streams + .entry(key) + .or_default() + .channels + .insert(descriptor.id, descriptor.name); + } + _ => {} } } } @@ -403,7 +643,8 @@ impl SupervisorActor { let state = self.driver.state().clone(); let mut replacements: Vec<(String, String)> = Vec::new(); for (id, managed) in &state.nodes { - if managed.record.ready && managed.intent == provisioning::reconciler::NodeIntent::Active + if managed.record.ready + && managed.intent == provisioning::reconciler::NodeIntent::Active { let Some(runtime) = self.manager.get(managed.attempt.0) else { continue; @@ -419,7 +660,11 @@ impl SupervisorActor { return; } for (dead, fresh) in &replacements { - self.emit_event("control", dead, format!("runtime death; replacing as {fresh}")); + self.emit_event( + "control", + dead, + format!("runtime death; replacing as {fresh}"), + ); self.slots.retain(|slot| slot_group_id(slot) != *dead); self.slots.push(fresh.clone()); } @@ -514,16 +759,20 @@ impl SupervisorActor { let Some(runtime) = self.manager.get(attempt) else { continue; }; - let report = read_key_report(&runtime.key_file); - let heartbeat_ms_ago = report - .as_ref() - .map(|r| unix_ms(now).saturating_sub(r.last_seen_ms)) - .unwrap_or(u64::MAX); + // Wire liveness: the node re-announces every heartbeat period, + // so the age of the last announce is the control-plane + // heartbeat. `null` until the first announce. + let heartbeat_ms_ago = runtime + .last_announce_ms + .map(|ms| unix_ms(now).saturating_sub(ms)); let payload = json!({ "at_ms": unix_ms(now), "node": runtime.logical_node, "alive": runtime.exited.is_none(), "pid": runtime.pid, + // Process state for the Fleet Control table (same shape the + // process lifecycle mirror used to publish). + "event": if runtime.exited.is_some() { "exited" } else { "started" }, "heartbeat_ms_ago": heartbeat_ms_ago, }); let bytes = serde_json::to_vec(&payload).expect("status serializes"); @@ -681,6 +930,7 @@ impl ActorInterface for SupervisorActor { match msg { SupervisorMsg::Tick => { let now = SystemTime::now(); + self.deliver_pending_reports(now); self.poll(now); self.observe_world(now); self.replace_dead_ready_nodes(now); @@ -688,9 +938,20 @@ impl ActorInterface for SupervisorActor { self.emit_node_status(now); self.emit_feed(now); self.flush_telemetry(); + self.flush_remote_streams(); } SupervisorMsg::Control(command) => self.handle_control(command), SupervisorMsg::Spawn(request) => self.spawn_node(ctx, request), + SupervisorMsg::Bootstrap(event) => { + let now = SystemTime::now(); + self.handle_bootstrap_event(now, event); + self.poll(now); + } + SupervisorMsg::NodeStream { + header, + logical_node, + attempt, + } => self.register_remote_stream(header, logical_node, attempt), SupervisorMsg::Shutdown { drained } => { self.slots.clear(); let generation = self.driver.desired().generation.saturating_add(1); @@ -715,10 +976,9 @@ impl SupervisorActor { &node, format!("kill requested (pid {:?})", runtime.pid), ); - let _ = swactor_process::send_process_command( - &self.sender, - runtime.process_actor, - swactor_process::ProcessCommand::Stop { + let _ = self.sender.send_to( + runtime.bootstrap, + provisioning::BootstrapMsg::Stop { kill_after: Some(Duration::ZERO), }, ); @@ -760,19 +1020,6 @@ impl SupervisorActor { } } -/// Parse a hex node key into a transport NodeId. -trait ParseKey { - fn parse_key(&self) -> Option; -} - -impl ParseKey for String { - fn parse_key(&self) -> Option { - let bytes = swactor_transport::hex_decode(self)?; - let array: [u8; 32] = bytes.try_into().ok()?; - Some(swactor_transport::NodeId(array)) - } -} - fn slot_group(slot: &str) -> provisioning::node::RunNodeGroupSpec { let mut group = demo_group(slot, 1); group.group_id = NodeGroupId(slot.to_owned()); @@ -832,4 +1079,3 @@ pub fn demo_retry_policy() -> RetryPolicy { endpoint_probe_interval: Duration::from_secs(1), } } - diff --git a/xtask/src/provisioning_demo/mod.rs b/xtask/src/provisioning_demo/mod.rs index ff2b281..d53b8c6 100644 --- a/xtask/src/provisioning_demo/mod.rs +++ b/xtask/src/provisioning_demo/mod.rs @@ -12,12 +12,26 @@ //! (per-node PID/state), kill nodes from Fleet Control or a shell, and watch //! the reconciler replace them for real. +pub mod bootstrap; pub mod control; +pub mod docker; pub mod feed; pub mod node; pub mod provider; pub mod view; +/// How the supervisor launches node-role children (bootstrap kind + launch +/// facts). Selected by `--docker`; default is local processes. +#[derive(Clone)] +pub enum LaunchStyle { + /// Re-exec this binary as a local child process (`kind: "process"`). + Process { exe: std::path::PathBuf }, + /// Run the demo docker image per node on the per-run bridge network + /// (`kind: "docker"`) — nodes are foreign: own IPs, gateway-dialed + /// supervisor, no shared filesystem. + Docker(docker::DockerLaunch), +} + use std::process::ExitCode; use std::sync::Arc; use std::time::Duration; @@ -30,12 +44,12 @@ use swactor_engine::{Engine, TokioBackend, TokioConfig}; use distribution::node::DistributedNodeConfig; use iroh_driver::{IrohDriver, IrohDriverConfig}; use provisioning::executor::IdempotentEffectExecutor; -use provisioning::{ClusterShape, RunId}; use provisioning::reconciler::ClusterDriver; +use provisioning::{ClusterShape, RunId}; use feed::{ - demo_retry_policy, initial_slots, EngineSpawner, SupervisorActor, SupervisorMsg, - SupervisorTelemetry, + EngineSpawner, SupervisorActor, SupervisorMsg, SupervisorTelemetry, demo_retry_policy, + initial_slots, }; use provider::{DemoBackend, DemoProvider, NodeManager}; @@ -81,27 +95,92 @@ fn resolve_exe() -> std::path::PathBuf { std::path::PathBuf::from("xtask") } -/// Shared handle the supervisor actor uses to check iroh connections. +/// Shared iroh driver handle: the supervisor's endpoint address (handed to +/// node roles) and the endpoint for outbound telemetry pulls. pub struct DemoDriverHandle { pub supervisor_addr_json: String, - driver: IrohDriver, + pub(crate) driver: IrohDriver, } impl DemoDriverHandle { - pub fn has_active_connection(&self, node: swactor_transport::NodeId) -> bool { - self.driver.has_active_connection(&node) + /// Clone the iroh endpoint for outbound telemetry pulls. + pub fn endpoint(&self) -> iroh::Endpoint { + self.driver.endpoint() } } -/// Entry point: `provisioning-reconciler-demo [--port N] [--nodes N]` for the -/// supervisor, or `--demo-node ` for node children. +/// Pull collector fired by bootstrap actors: dials the freshly-bootstrapped +/// node on `TELEMETRY_ALPN`, sends the subscription request, and streams its +/// telemetry into the supervisor's remote-stream fanout. The header lands in +/// the supervisor actor so frames can be classified before publishing. +struct DemoTelemetryCollector { + engine: swactor_engine::EngineHandle, + endpoint: iroh::Endpoint, + fanout: Arc, + sender: swactor::runtime::ExternalSender, + supervisor_slot: Arc>, +} + +impl provisioning::NodeTelemetryCollector for DemoTelemetryCollector { + fn collect(&self, identity: &provisioning::NodeIdentity) { + let Ok(addr) = serde_json::from_str::(&identity.transport_addr) else { + eprintln!( + "demo: node {} advertised unparsable endpoint address", + identity.logical_node + ); + return; + }; + println!( + "demo: pulling telemetry from node {} (key {}…)", + identity.logical_node, + &identity.key_hex[..8.min(identity.key_hex.len())] + ); + let mut flow_id = [0u8; 16]; + flow_id[..8].copy_from_slice(&identity.attempt.to_le_bytes()); + let (header_tx, header_rx) = std::sync::mpsc::channel(); + iroh_driver::spawn_pull_collector( + &self.engine, + self.endpoint.clone(), + addr, + flow_id, + Vec::new(), + telemetry::SubscriptionRequest::all(), + Arc::clone(&self.fanout), + header_tx, + ); + let sender = self.sender.clone(); + if let Some(supervisor) = self.supervisor_slot.get() { + let supervisor = supervisor.clone(); + let logical_node = identity.logical_node.clone(); + let attempt = identity.attempt; + std::thread::spawn(move || { + if let Ok(header) = header_rx.recv() { + let _ = sender.send_to( + supervisor, + feed::SupervisorMsg::NodeStream { + header, + logical_node, + attempt, + }, + ); + } + }); + } + } +} + +/// Entry point: `provisioning-reconciler-demo [--port N] [--nodes N] +/// [--docker]` for the supervisor, or `--demo-node +/// --demo-attempt ` for node children. pub fn run(args: &[String]) -> ExitCode { if let Some(index) = args.iter().position(|arg| arg == "--demo-node") { - let addr = args - .get(index + 1) - .map(String::as_str) - .unwrap_or_default(); - return match node::run_node_role(addr) { + let addr = args.get(index + 1).map(String::as_str).unwrap_or_default(); + let attempt = arg_value(args, "--demo-attempt").and_then(|value| value.parse::().ok()); + let Some(attempt) = attempt else { + eprintln!("demo node: --demo-attempt is required"); + return ExitCode::FAILURE; + }; + return match node::run_node_role(addr, attempt) { Ok(()) => ExitCode::SUCCESS, Err(error) => { eprintln!("demo node: {error}"); @@ -130,6 +209,13 @@ fn run_supervisor(args: &[String]) -> Result<(), String> { let nodes: u64 = arg_value(args, "--nodes") .and_then(|value| value.parse().ok()) .unwrap_or(DEFAULT_NODES); + let docker_mode = args.iter().any(|arg| arg == "--docker"); + + // Node registry + spawn channel: needed before the actor bridge so the + // announce relay can route wire announces into bootstrap actors. + let manager = NodeManager::new(); + let (spawn_tx, spawn_rx) = std::sync::mpsc::channel::(); + manager.set_spawn_channel(spawn_tx); // Telemetry endpoint with the runtime stats hook attached before the // engine takes the parts. @@ -168,29 +254,42 @@ fn run_supervisor(args: &[String]) -> Result<(), String> { use distribution::transport_bridge::{Outbox, RelayMirror, RouteView}; use swactor_transport::CodecRegistry; - struct NoopActor; - impl swactor::actor::ActorInterface for NoopActor { - type Incoming = (); - type Response = (); - fn handle(&mut self, _ctx: &swactor::actor::Ctx, _msg: ()) {} - } - let noop = runtime - .spawn(NoopActor) - .map_err(|e| format!("spawn noop actor: {e}"))?; - let relay_mirror: RelayMirror = Arc::new(std::sync::RwLock::new(std::collections::HashMap::new())); - let route_view: RouteView = Arc::new(std::sync::RwLock::new(std::collections::HashMap::new())); + // The announce relay is the bridge's only decoded ingress: node + // roles announce over the control plane (tag-routed gossip), and + // the relay correlates by attempt → bootstrap actor. The + // `swim_addr` argument only names a SendFailed target, which never + // fires for inbound frames; the relay doubles for it. Undecodable + // frames (e.g. the driver's own JoinRequest) still drop harmlessly. + let announce = runtime + .spawn(provider::AnnounceActor::new( + manager.clone(), + sender.clone(), + )) + .map_err(|e| format!("spawn announce actor: {e}"))?; + let mut codec = CodecRegistry::new(); + codec.register_decoder::(node::ANNOUNCE_TAG, |bytes| { + serde_json::from_slice(bytes) + .map_err(|e| swactor::Error::from(format!("announce decode: {e}"))) + }); + let mut routes = std::collections::HashMap::new(); + routes.insert(node::ANNOUNCE_TAG.to_owned(), announce); + let relay_mirror: RelayMirror = + Arc::new(std::sync::RwLock::new(std::collections::HashMap::new())); + let route_view: RouteView = + Arc::new(std::sync::RwLock::new(std::collections::HashMap::new())); let outbox: Outbox = Arc::new(std::sync::Mutex::new(Vec::new())); driver.enable_actor_bridge( runtime.clone(), - Arc::new(CodecRegistry::new()), - std::collections::HashMap::new(), - noop, + Arc::new(codec), + routes, + announce, relay_mirror, route_view, outbox, ); driver.install_actor_bridge_pump(Duration::from_millis(250)); } + let driver_handle = Arc::new(DemoDriverHandle { supervisor_addr_json: supervisor_addr_json.clone(), driver, @@ -205,15 +304,23 @@ fn run_supervisor(args: &[String]) -> Result<(), String> { engine.handle().spawn(dashboard.http_server()); // Provisioning: driver + plugin + executor. - let keys_dir = std::env::temp_dir().join(format!( - "provisioning-demo-{}", - std::process::id() - )); - std::fs::create_dir_all(&keys_dir).map_err(|e| format!("keys dir: {e}"))?; - - let manager = NodeManager::new(); - let (spawn_tx, spawn_rx) = std::sync::mpsc::channel::(); - manager.set_spawn_channel(spawn_tx); + let launch = if docker_mode { + let addrs = driver_handle.driver.direct_addresses(); + let port = addrs + .iter() + .find(|addr| addr.is_ipv4()) + .or_else(|| addrs.first()) + .map(|addr| addr.port()) + .ok_or_else(|| "supervisor has no bound direct address".to_owned())?; + let root = std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .parent() + .expect("xtask manifest dir has a parent") + .to_path_buf(); + let launch = docker::preflight(&root, driver_handle.driver.endpoint().id(), port)?; + LaunchStyle::Docker(launch) + } else { + LaunchStyle::Process { exe: resolve_exe() } + }; let slots = initial_slots(nodes); let shape = ClusterShape { @@ -224,7 +331,7 @@ fn run_supervisor(args: &[String]) -> Result<(), String> { let cluster_driver = ClusterDriver::new(shape, demo_retry_policy()).map_err(|e| format!("driver: {e}"))?; - let plugin = DemoProvider::new(manager.clone(), keys_dir.clone()); + let plugin = DemoProvider::new(manager.clone()); let spawner = EngineSpawner::new(engine.handle()); let executor = IdempotentEffectExecutor::new( DemoBackend { @@ -235,6 +342,50 @@ fn run_supervisor(args: &[String]) -> Result<(), String> { spawner, ); + // The supervisor's address travels through a shared slot; long-lived + // engine tasks installed below wait for it lazily. (Spawning engine + // tasks after the actor spawn proved flaky at startup.) + let supervisor_slot: Arc> = + Arc::new(std::sync::OnceLock::new()); + + // Bootstrap machinery: kind registry with the process logic, the + // remote-stream fanout, and the pull collector. + let fanout = Arc::new(telemetry::DeliveryFanout::new(1024)); + let remote_sub = fanout.subscribe_all( + "dashboard", + telemetry::TelemetrySnapshot { + streams: Vec::new(), + channels: Vec::new(), + }, + ); + let mut bootstrap_registry = provisioning::BootstrapRegistry::new(); + bootstrap_registry.register("process", { + let manager = manager.clone(); + Arc::new(move |spec| { + Ok(Box::new(bootstrap::LocalProcessLogic::new( + spec.clone(), + manager.clone(), + )) as Box) + }) + }); + bootstrap_registry.register("docker", { + let manager = manager.clone(); + Arc::new(move |spec| { + Ok(Box::new(docker::DockerProcessLogic::new( + spec.clone(), + manager.clone(), + )) as Box) + }) + }); + let collector: Arc = + Arc::new(DemoTelemetryCollector { + engine: engine.handle(), + endpoint: driver_handle.endpoint(), + fanout: Arc::clone(&fanout), + sender: sender.clone(), + supervisor_slot: supervisor_slot.clone(), + }); + let supervisor = SupervisorActor::new( cluster_driver, executor, @@ -243,16 +394,14 @@ fn run_supervisor(args: &[String]) -> Result<(), String> { telemetry, dashboard.clone(), sender.clone(), + bootstrap_registry, + collector, + engine.handle(), + remote_sub, slots, RunId(1), - resolve_exe(), + launch.clone(), ); - // Long-lived engine tasks are installed BEFORE the actor spawn: the - // supervisor's address travels through a shared slot, and the tasks wait - // for it lazily. (Spawning engine tasks after the actor spawn proved - // flaky at startup.) - let supervisor_slot: Arc> = - Arc::new(std::sync::OnceLock::new()); // Control plane: dashboard → supervisor. control::install(&engine.handle(), sender.clone(), supervisor_slot.clone()); @@ -260,14 +409,16 @@ fn run_supervisor(args: &[String]) -> Result<(), String> { // Spawn-request pump: provider blocking threads → supervisor actor. let pump_sender = sender.clone(); let pump_slot = supervisor_slot.clone(); - engine.handle().spawn_blocking(move || loop { - match spawn_rx.recv() { - Ok(request) => { - if let Some(addr) = pump_slot.get() { - let _ = pump_sender.send_to(addr.clone(), SupervisorMsg::Spawn(request)); + engine.handle().spawn_blocking(move || { + loop { + match spawn_rx.recv() { + Ok(request) => { + if let Some(addr) = pump_slot.get() { + let _ = pump_sender.send_to(addr.clone(), SupervisorMsg::Spawn(request)); + } } + Err(_) => return, } - Err(_) => return, } }); @@ -328,8 +479,11 @@ fn run_supervisor(args: &[String]) -> Result<(), String> { std::thread::sleep(TICK); } // Best-effort drain window closed: children still alive (if any) are - // killed by the kernel parent-death signal armed in the node role. - let _ = std::fs::remove_dir_all(&keys_dir); + // killed by the kernel parent-death signal armed in the node role + // (process kind) or swept by label below (docker kind). + if let LaunchStyle::Docker(docker) = &launch { + docker::sweep_run(docker); + } std::process::exit(0); } @@ -344,6 +498,10 @@ fn install_sigint_flag() { unsafe { let handler: extern "C" fn(libc::c_int) = sigint_handler; libc::signal(libc::SIGINT, handler as usize); + // Supervisors under process managers (systemd, container runtimes, + // harnesses) stop children with SIGTERM; treat it exactly like + // Ctrl-C so the drain + sweep path runs instead of a default kill. + libc::signal(libc::SIGTERM, handler as usize); } } diff --git a/xtask/src/provisioning_demo/node.rs b/xtask/src/provisioning_demo/node.rs index 105e367..fd6111e 100644 --- a/xtask/src/provisioning_demo/node.rs +++ b/xtask/src/provisioning_demo/node.rs @@ -1,38 +1,85 @@ //! Node role: a real swactor runtime that joins the supervisor over iroh. //! -//! Re-exec'd from the xtask binary by the demo provider. Builds an engine + -//! `IrohDriver` (relay disabled), joins the supervisor's endpoint, and -//! reports its swactor node key + heartbeats to the key file given in -//! `DEMO_NODE_KEY_FILE` (the process actor supervises children with stdio -//! null, so stdout is not observable). +//! Re-exec'd from the xtask binary by the demo provider (local process kind) +//! or launched as the entrypoint of the demo docker image. Builds an engine + +//! `IrohDriver` (relay disabled), joins the supervisor's endpoint, and then +//! **announces itself over the control plane**: a tagged gossip frame +//! ([`ANNOUNCE_TAG`]) carrying the provision attempt token, its iroh node +//! key, and its advertised endpoint address. The supervisor's announce relay +//! routes that frame by attempt to the owning bootstrap actor — the wire +//! announce *is* the readiness signal (no shared filesystem, no stdout +//! capture), and it repeats every [`HEARTBEAT_PERIOD`] as a liveness +//! heartbeat. +//! +//! Telemetry: the node owns a real [`TelemetryEndpoint`] whose stream id is +//! its transport identity (iroh node key). The runtime stats hook feeds +//! `runtime.actors` rosters, a `node.status` channel carries liveness +//! records, and beat actors keep the roster visibly alive. The supervisor +//! pulls this endpoint over `TELEMETRY_ALPN` +//! ([`iroh_driver::spawn_pull_server`]) — the node never dials out for +//! telemetry. -use std::io::Write; +use std::sync::Arc; +use std::time::Duration; use iroh::RelayMode; +use serde_json::json; +use swactor::actor::{ActorInterface, Ctx}; use swactor::config::RuntimeConfig; use swactor::runtime::RuntimeParts; use swactor_engine::{Engine, TokioBackend, TokioConfig}; use distribution::node::DistributedNodeConfig; -use iroh_driver::{IrohDriver, IrohDriverConfig}; +use iroh_driver::{IrohDriver, IrohDriverConfig, TELEMETRY_ALPN, spawn_pull_server}; +use telemetry::{ChannelContent, TelemetryEndpoint, TelemetryProducer}; use crate::provisioning_demo::HEARTBEAT_PERIOD; -/// Key-file line marking the node key (first line): ` KEY `. -pub const KEY_LINE_KIND: &str = "KEY"; +/// Wire tag of the announce gossip frame (`CodecRegistry` decode key on the +/// supervisor side; raw tag bytes on the node side). +pub const ANNOUNCE_TAG: &str = "xtask_demo/NodeAnnounce/1"; -/// Key-file heartbeat line: ` ALIVE`. -pub const ALIVE_LINE_KIND: &str = "ALIVE"; +/// A node's self-introduction on the supervisor's control plane. Sent right +/// after joining and then every heartbeat period; the supervisor routes it +/// by `attempt` to the owning bootstrap actor (first delivery = readiness, +/// later deliveries = liveness). +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] +pub struct NodeAnnounce { + /// Provision attempt token (`--demo-attempt`); correlates the announce + /// with the supervisor-side bootstrap actor that launched this node. + pub attempt: u64, + /// Logical node name (slot id) for dashboard labeling. + pub logical_node: String, + /// The node's transport key (iroh node id, hex). + pub key_hex: String, + /// Serde-serialized `iroh::EndpointAddr` the supervisor dials for + /// telemetry pulls. + pub endpoint_addr_json: String, + pub at_ms: u64, +} + +/// How often the node polls its driver for accepted telemetry pull +/// connections. +const PULL_POLL_PERIOD: Duration = Duration::from_millis(200); + +/// Idle sleep for the pull answer writer between subscription batches. +const WRITER_IDLE: Duration = Duration::from_millis(100); + +/// Number of beat actors on the node — real actors whose message flow keeps +/// the `runtime.actors` roster visibly alive. +const BEAT_ACTORS: u64 = 2; /// Run the node role. `supervisor_addr_json` is a serde-serialized -/// `iroh::EndpointAddr` of the supervisor's iroh endpoint. -pub fn run_node_role(supervisor_addr_json: &str) -> Result<(), String> { +/// `iroh::EndpointAddr` of the supervisor's iroh endpoint; `attempt` is the +/// provision attempt token echoed back in every announce. +pub fn run_node_role(supervisor_addr_json: &str, attempt: u64) -> Result<(), String> { + install_parent_death_signal()?; let supervisor_addr: iroh::EndpointAddr = serde_json::from_str(supervisor_addr_json) .map_err(|error| format!("invalid supervisor endpoint address: {error}"))?; - let key_file = std::env::var("DEMO_NODE_KEY_FILE") - .map_err(|_| "DEMO_NODE_KEY_FILE not set".to_owned())?; + let logical_node = std::env::var("DEMO_NODE_ID").unwrap_or_else(|_| "node".to_owned()); let parts = RuntimeParts::new(RuntimeConfig::default()); + let runtime = parts.runtime().clone(); let engine = Engine::new( parts, TokioBackend::new(TokioConfig::default()).map_err(|error| format!("backend: {error}"))?, @@ -40,30 +87,170 @@ pub fn run_node_role(supervisor_addr_json: &str) -> Result<(), String> { .map_err(|error| format!("engine: {error}"))?; // Bind the driver, then keep it alive for the process lifetime: dropping - // it closes the endpoint. - let driver = IrohDriver::with_engine( - engine.handle(), - IrohDriverConfig { - secret_key: None, - relay_mode: RelayMode::Disabled, - node: DistributedNodeConfig::default(), - peer_auth: None, - additional_alpns: vec![], - }, - ) - .map_err(|error| format!("iroh driver: {error}"))?; - + // it closes the endpoint. The endpoint must advertise TELEMETRY_ALPN so + // the supervisor's pull connection can negotiate it. + let driver = Arc::new( + IrohDriver::with_engine( + engine.handle(), + IrohDriverConfig { + secret_key: None, + relay_mode: RelayMode::Disabled, + node: DistributedNodeConfig::default(), + peer_auth: None, + additional_alpns: vec![TELEMETRY_ALPN.to_vec()], + }, + ) + .map_err(|error| format!("iroh driver: {error}"))?, + ); let node_hex = swactor_transport::hex_encode(&driver.node_id().0); - driver.join(&[supervisor_addr]); - append_key_line(&key_file, KEY_LINE_KIND, &node_hex); - let heartbeat_file = key_file.clone(); - let interval_handle = engine.handle(); - interval_handle.clone().spawn(async move { - let mut interval = interval_handle.interval(HEARTBEAT_PERIOD); + // Telemetry endpoint: stream identity is the transport node key, so + // every provisioned process lands on its own dashboard stream. + let stream = telemetry::frame::StreamId::new( + telemetry::frame::NodeId::new(&node_hex), + telemetry::frame::Lifetime(1), + ); + let endpoint = TelemetryEndpoint::with_descriptor( + telemetry::frame::StreamDescriptor { + stream, + label: Some(format!("demo node {logical_node} runtime")), + origin: telemetry::frame::StreamOrigin::RemoteNode, + }, + 256, + 16, + ); + let producer = endpoint.producer(); + let actors_channel = endpoint.register_channel( + "runtime.actors", + ChannelContent::JsonRecord { + schema: Some("runtime.actors".to_owned()), + }, + ); + let status_channel = endpoint.register_channel( + "node.status", + ChannelContent::JsonRecord { + schema: Some("demo.node.status.v1".to_owned()), + }, + ); + let beats_channel = endpoint.register_channel( + "node.beat", + ChannelContent::JsonRecord { + schema: Some("demo.node.beat.v1".to_owned()), + }, + ); + runtime.set_stats_hook(producer.stats_hook_on(actors_channel)); + + // Join, then announce identity + advertised address to the supervisor's + // bootstrap actor over the control plane (readiness + telemetry dial). + driver.join(&[supervisor_addr.clone()]); + let addr_json = + serde_json::to_string(&driver.endpoint_addr()).map_err(|e| format!("addr: {e}"))?; + let announce_driver = Arc::clone(&driver); + let announce_logical = logical_node.clone(); + let announce_key = node_hex.clone(); + let announce_engine = engine.handle(); + announce_engine.clone().spawn(async move { + let mut interval = announce_engine.interval(HEARTBEAT_PERIOD); loop { (&mut interval).await; - append_key_line(&heartbeat_file, ALIVE_LINE_KIND, ""); + let announce = NodeAnnounce { + attempt, + logical_node: announce_logical.clone(), + key_hex: announce_key.clone(), + endpoint_addr_json: addr_json.clone(), + at_ms: unix_ms(), + }; + let Ok(bytes) = serde_json::to_vec(&announce) else { + continue; + }; + announce_driver.send_tagged_gossip( + supervisor_addr.clone(), + ANNOUNCE_TAG.as_bytes(), + bytes, + ); + } + }); + + // Beat actors: real actors with real message flow, so the node's + // runtime.actors roster (pulled by the supervisor) is visibly alive. + let sender = runtime.create_sender(); + let mut beat_addrs = Vec::new(); + for index in 0..BEAT_ACTORS { + let name = format!("beat-{index}"); + let producer = producer.clone(); + let addr = runtime + .spawn(BeatActor { + name: name.clone(), + producer, + channel: beats_channel, + }) + .map_err(|error| format!("spawn beat actor: {error}"))?; + beat_addrs.push((addr, name)); + } + + // Heartbeats: a node.status telemetry record (the wire announce above is + // the supervisor-facing liveness channel). + let heartbeat_producer = producer.clone(); + let heartbeat_logical = logical_node.clone(); + let heartbeat_key = node_hex.clone(); + let interval_engine = engine.handle(); + interval_engine.clone().spawn(async move { + let mut interval = interval_engine.interval(HEARTBEAT_PERIOD); + let mut seq: u64 = 0; + loop { + (&mut interval).await; + seq += 1; + let payload = json!({ + "at_ms": unix_ms(), + "node": heartbeat_logical, + "key": heartbeat_key, + "seq": seq, + "alive": true, + "pid": std::process::id(), + "beats": BEAT_ACTORS, + }); + let bytes = serde_json::to_vec(&payload).expect("status serializes"); + heartbeat_producer.submit_bytes(status_channel, bytes); + } + }); + + // Beat driver: engine intervals poking the beat actors. + for (index, (addr, _name)) in beat_addrs.iter().enumerate() { + let addr = *addr; + let beat_sender = sender.clone(); + let beat_engine = engine.handle(); + // Stagger the beat periods so roster entries tick at different rates. + let period = Duration::from_millis(1000 + 500 * index as u64); + beat_engine.clone().spawn(async move { + let mut interval = beat_engine.interval(period); + loop { + (&mut interval).await; + let _ = beat_sender.send_to(addr, BeatMsg::Wake); + } + }); + } + + // Serve telemetry pulls: accepted TELEMETRY_ALPN connections answer with + // this endpoint's subscription stream. + let telemetry_endpoint = Arc::new(endpoint); + let serve_engine = engine.handle(); + let serve_driver = Arc::clone(&driver); + serve_engine.clone().spawn(async move { + let mut interval = serve_engine.interval(PULL_POLL_PERIOD); + loop { + (&mut interval).await; + // Drain the mux into the endpoint fanout so producer frames + // reach the pull subscription; without this tick the mux fills + // and nothing is ever streamed. + let _ = telemetry_endpoint.tick(); + for (_node, conn) in serve_driver.drain_accepted_for_alpn(TELEMETRY_ALPN) { + spawn_pull_server( + &serve_engine, + conn, + Arc::clone(&telemetry_endpoint), + WRITER_IDLE, + ); + } } }); @@ -75,45 +262,68 @@ pub fn run_node_role(supervisor_addr_json: &str) -> Result<(), String> { } } -fn append_key_line(path: &str, kind: &str, value: &str) { - let Ok(mut file) = std::fs::OpenOptions::new().create(true).append(true).open(path) else { - return; - }; - let now = std::time::SystemTime::now() +/// Message that keeps a beat actor's mailbox flowing (and therefore its +/// `runtime.actors` stats current). +#[derive(Clone, Debug)] +pub enum BeatMsg { + Wake, +} + +/// A deliberately trivial actor whose only job is existing: every Wake it +/// submits one `node.beat` record, and the runtime stats hook records its +/// productive tick on `runtime.actors`. +pub struct BeatActor { + name: String, + producer: TelemetryProducer, + channel: telemetry::ChannelId, +} + +fn unix_ms() -> u64 { + std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .map(|duration| duration.as_millis() as u64) - .unwrap_or(0); - let _ = writeln!(file, "{now} {kind} {value}"); + .unwrap_or(0) } -/// Parsed key-file contents: the child's node key and its last heartbeat. -pub struct NodeKeyReport { - pub node_hex: String, - pub last_seen_ms: u64, +impl ActorInterface for BeatActor { + type Incoming = BeatMsg; + type Response = (); + + fn handle(&mut self, _ctx: &Ctx, _msg: BeatMsg) { + let payload = json!({ + "at_ms": unix_ms(), + "actor": self.name, + }); + let bytes = serde_json::to_vec(&payload).expect("beat serializes"); + self.producer.submit_bytes(self.channel, bytes); + } } -/// Read a node's key file: its node key and last heartbeat. -pub fn read_key_report(path: &std::path::Path) -> Option { - let contents = std::fs::read_to_string(path).ok()?; - let mut node_hex: Option = None; - let mut last_seen_ms = 0_u64; - for line in contents.lines() { - let mut parts = line.split_whitespace(); - let Some(stamp) = parts.next().and_then(|value| value.parse::().ok()) else { - continue; - }; - let Some(kind) = parts.next() else { - continue; - }; - if kind == KEY_LINE_KIND { - node_hex = parts.next().map(str::to_owned); - last_seen_ms = last_seen_ms.max(stamp); - } else if kind == ALIVE_LINE_KIND { - last_seen_ms = last_seen_ms.max(stamp); +/// Arm the kernel's parent-death signal so a hard-killed supervisor +/// (SIGKILL, crash, wedged teardown) cannot leave this node orphaned and +/// parked forever. The `parent_id` check closes the fork/prctl race: if the +/// supervisor died before the prctl landed, the signal would never fire. +/// Inside a container the node is PID 1 (parent 0), so the check passes and +/// the prctl is a harmless no-op — the docker bootstrap's `docker rm -f` +/// owns termination there. +#[cfg(target_os = "linux")] +fn install_parent_death_signal() -> Result<(), String> { + unsafe { + if libc::prctl(libc::PR_SET_PDEATHSIG, libc::SIGTERM) != 0 { + return Err(format!( + "prctl(PR_SET_PDEATHSIG): {}", + std::io::Error::last_os_error() + )); } } - Some(NodeKeyReport { - node_hex: node_hex?, - last_seen_ms, - }) + if std::os::unix::process::parent_id() == 1 { + return Err("supervisor exited before node startup".to_owned()); + } + Ok(()) +} + +#[cfg(not(target_os = "linux"))] +fn install_parent_death_signal() -> Result<(), String> { + // Non-Linux builds rely on the supervisor's graceful teardown path. + Ok(()) } diff --git a/xtask/src/provisioning_demo/provider.rs b/xtask/src/provisioning_demo/provider.rs index c29eb7f..a4580b8 100644 --- a/xtask/src/provisioning_demo/provider.rs +++ b/xtask/src/provisioning_demo/provider.rs @@ -12,13 +12,12 @@ //! channel to the supervisor actor; the plugin call blocks for the reply on //! an executor blocking thread. //! -//! Because the process actor supervises children with stdio null, each child -//! publishes its swactor node key and heartbeats to a per-attempt key file -//! (`DEMO_NODE_KEY_FILE`); the supervisor reads it to learn the child's iroh -//! identity and liveness. +//! Children run with stdio null, so the supervisor learns a node's iroh +//! identity and liveness from the wire announce (see `node.rs`): the +//! [`AnnounceActor`] routes it to the owning bootstrap actor and records +//! `last_announce_ms` in the [`NodeManager`] registry. use std::collections::BTreeMap; -use std::path::PathBuf; use std::sync::{Arc, Mutex}; use std::time::{Duration, SystemTime}; @@ -31,8 +30,8 @@ use provisioning::node::{ BootstrapSessionId, CreateLeaseResult, DestroyHandle, LeaseFacts, NodeManagerCommand, ProviderKind, ProviderLeaseId, SshEndpoint, }; -use provisioning::reconciler::{OperationId, OperationOutcome, PlannedEffect}; use provisioning::plugin::{NodeProvisionSpec, PluginNodeHandle, PluginSink, ProvisionPlugin}; +use provisioning::reconciler::{OperationId, OperationOutcome, PlannedEffect}; use telemetry::{ChannelContent, StreamDescriptor, TelemetryEndpoint, TelemetryProducer}; /// How long a blocking plugin call waits for the supervisor actor. @@ -43,11 +42,15 @@ pub const BACKEND_WAIT: Duration = Duration::from_secs(20); pub struct NodeRuntime { pub attempt: u64, pub logical_node: String, - pub process_actor: ActorAddress, - pub key_file: PathBuf, + /// The bootstrap actor that owns this attempt's lifecycle. + pub bootstrap: ActorAddress, pub pid: Option, pub exited: Option, pub spawn_failed: Option, + /// Wall-clock ms of the last wire announce from the node (None until + /// the first announce). The node re-announces every heartbeat period, + /// so staleness here means the control-plane path is dead. + pub last_announce_ms: Option, } /// Spawn request from the plugin (blocking thread) to the supervisor actor. @@ -55,7 +58,6 @@ pub struct NodeRuntime { pub struct SpawnNodeRequest { pub attempt: u64, pub logical_node: String, - pub key_file: PathBuf, pub reply: std::sync::mpsc::Sender>, } @@ -81,17 +83,11 @@ impl NodeManager { self.inner.lock().expect("node manager").spawn_tx = Some(sender); } - pub fn request_spawn( - &self, - attempt: u64, - logical_node: String, - key_file: PathBuf, - ) -> Result { + pub fn request_spawn(&self, attempt: u64, logical_node: String) -> Result { let (reply_tx, reply_rx) = std::sync::mpsc::channel(); let request = SpawnNodeRequest { attempt, logical_node, - key_file, reply: reply_tx, }; { @@ -160,23 +156,27 @@ impl NodeManager { self.update(attempt, |runtime| runtime.spawn_failed = Some(reason)); } - pub fn remove(&self, attempt: u64) { - self.inner.lock().expect("node manager").nodes.remove(&attempt); + pub fn set_announce(&self, attempt: u64, at_ms: u64) { + self.update(attempt, |runtime| runtime.last_announce_ms = Some(at_ms)); + } + pub fn remove(&self, attempt: u64) { + self.inner + .lock() + .expect("node manager") + .nodes + .remove(&attempt); } - } /// The demo `ProvisionPlugin`: resources are node-role child processes. pub struct DemoProvider { manager: NodeManager, - keys_dir: PathBuf, } impl DemoProvider { - pub fn new(manager: NodeManager, keys_dir: PathBuf) -> Self { - Self { manager, keys_dir } + pub fn new(manager: NodeManager) -> Self { + Self { manager } } - } impl ProvisionPlugin for DemoProvider { @@ -199,10 +199,7 @@ impl ProvisionPlugin for DemoProvider { .find(|(key, _)| key == "DEMO_LOGICAL_NODE") .map(|(_, value)| value.clone()) .ok_or_else(|| "spec missing DEMO_LOGICAL_NODE".to_owned())?; - let key_file = self.keys_dir.join(format!("node-{attempt}.key")); - let runtime = self - .manager - .request_spawn(attempt, logical_node, key_file)?; + let runtime = self.manager.request_spawn(attempt, logical_node)?; Ok(PluginNodeHandle { id: attempt, provider_process_id: runtime.pid, @@ -297,6 +294,51 @@ impl ActorInterface for NodeRelayActor { } } +/// Supervisor-side announce relay. The iroh actor bridge decodes inbound +/// [`NodeAnnounce`] gossip frames and routes them here by wire tag; this +/// actor correlates by attempt token and forwards to the owning bootstrap +/// actor ([`provisioning::BootstrapMsg::Announce`]) — the handoff from +/// bootstrap to control plane. Announces for unknown attempts (deregistered +/// lease, stale container from a dead attempt) are dropped with a log line. +pub struct AnnounceActor { + manager: NodeManager, + sender: ExternalSender, +} + +impl AnnounceActor { + pub fn new(manager: NodeManager, sender: ExternalSender) -> Self { + Self { manager, sender } + } +} + +impl ActorInterface for AnnounceActor { + type Incoming = crate::provisioning_demo::node::NodeAnnounce; + type Response = (); + + fn handle(&mut self, _ctx: &Ctx, announce: crate::provisioning_demo::node::NodeAnnounce) { + self.manager.set_announce(announce.attempt, announce.at_ms); + let Some(runtime) = self.manager.get(announce.attempt) else { + let key = &announce.key_hex; + eprintln!( + "demo: announce for unknown attempt {} (key {}…) dropped", + announce.attempt, + &key[..8.min(key.len())] + ); + return; + }; + let identity = provisioning::NodeIdentity { + attempt: announce.attempt, + logical_node: announce.logical_node, + key_hex: announce.key_hex, + transport_addr: announce.endpoint_addr_json, + }; + let _ = self.sender.send_to( + runtime.bootstrap, + provisioning::BootstrapMsg::Announce(identity), + ); + } +} + /// Demo lease/session identity (attempt-encoded, kit conventions). pub fn demo_lease(attempt: u64) -> LeaseFacts { let provider = ProviderKind::new("demo"); @@ -331,7 +373,9 @@ pub fn session_id_for(operation: OperationId) -> BootstrapSessionId { } fn runtime_exited(manager: &NodeManager, attempt: u64) -> bool { - manager.get(attempt).is_some_and(|runtime| runtime.exited.is_some()) + manager + .get(attempt) + .is_some_and(|runtime| runtime.exited.is_some()) } /// The demo `EffectBackend`: routes effects through the plugin. @@ -353,7 +397,6 @@ impl DemoBackend { } } - impl EffectBackend for DemoBackend { fn execute(&self, effect: &PlannedEffect) -> Result { let attempt = effect.operation.attempt.0; @@ -433,10 +476,9 @@ fn stop_node_with_sender( ) -> Result<(), String> { if let Some(runtime) = manager.get(handle.id) { if runtime.pid.is_some() && runtime.exited.is_none() { - let _ = swactor_process::send_process_command( - sender, - runtime.process_actor, - swactor_process::ProcessCommand::Stop { + let _ = sender.send_to( + runtime.bootstrap, + provisioning::BootstrapMsg::Stop { kill_after: Some(Duration::from_secs(1)), }, ); diff --git a/xtask/src/provisioning_demo/view.rs b/xtask/src/provisioning_demo/view.rs index 3c437a0..c8dafda 100644 --- a/xtask/src/provisioning_demo/view.rs +++ b/xtask/src/provisioning_demo/view.rs @@ -13,8 +13,8 @@ use serde::Serialize; use serde_json::Value; use telemetry::frame::{Frame, StreamId}; -use dashboard::view::DashboardView; use dashboard::FrameEvent; +use dashboard::view::DashboardView; const EVENTS_CHANNEL: &str = "prov.reconciler.events"; const SNAPSHOT_CHANNEL: &str = "prov.reconciler.snapshot"; @@ -171,4 +171,3 @@ impl DashboardView for ReconcilerDashboardView { None } } -