diff --git a/README.md b/README.md index 9da222a..8946110 100644 --- a/README.md +++ b/README.md @@ -5,6 +5,30 @@ addresses, and a built-in cluster. Actors are ordinary structs; a message is any `Clone + Send + Sync` type; an address works the same whether the actor lives in this process, on a peer, or behind NAT on another node. +## See it run + +One command boots a supervisor — a real swactor engine with a real `iroh` +endpoint and the dashboard — plus a fleet of node processes that join it over +QUIC, then keeps that fleet alive: + +```sh +cargo xtask demo # --nodes 8 --port 9871 --docker also work +``` + +Open the dashboard it prints and watch the system act like one: + +- **Fleet Control** (`/view/demo-control`) — the reconciler's + current-vs-desired state as counters, each node's stage + (`New → LeaseRequested → … → SwactorJoined → HandedOff`), and the + command/result feeds. Press **kill** on a node: the reconciler notices the + death and provisions a replacement, for real. +- **Fleet** (`/view/fleet`) — per-node cards with PID and lifecycle; click + through to an actor roster and per-actor dossier. + +Nodes are ordinary child processes (or, with `--docker`, scratch containers +on their own bridge network), so `kill -9` from a shell triggers the same +recovery. Ctrl-C tears everything down. + ## What it is Every actor is one trait. There is no separate message trait to implement — diff --git a/crates/dashboard/src/control.rs b/crates/dashboard/src/control.rs index a36b59c..f84e7ff 100644 --- a/crates/dashboard/src/control.rs +++ b/crates/dashboard/src/control.rs @@ -25,6 +25,10 @@ pub enum ControlCommand { /// Lower the desired cluster size by `count` nodes (graceful scale /// down: teardown through the reconciler, not a kill). Remove { count: u32 }, + /// Establish (or replace) the data-plane edge toward one node. The + /// supervisor provisions the node's inbound edge over the control + /// plane and dials it over EDGE_ALPN. + EstablishEdge { node: String }, } static CONTROL_SENDER: OnceLock> = OnceLock::new(); diff --git a/crates/dashboard/src/demo_control_page.html b/crates/dashboard/src/demo_control_page.html index f163492..bfcf2bc 100644 --- a/crates/dashboard/src/demo_control_page.html +++ b/crates/dashboard/src/demo_control_page.html @@ -146,7 +146,7 @@ RISK: terminal cosplay if decoration creeps past data; held by the palette law.
- +
nodestagestatepidseen
nodestagestatepidseenedge
@@ -176,11 +176,20 @@ RISK: terminal cosplay if decoration creeps past data; held by the palette law. Failed: 'b-failed', Destroyed: 'b-destroyed' }; const INTENT_DELETING = new Set(['Deleting']); - - const rows = document.getElementById('rows'); + const EDGE_CLASS = { + provisioning: 'b-progress', + ready: 'b-ready', + faulted: 'b-failed' + }; const empty = document.getElementById('empty'); const status = document.getElementById('status'); const counter = document.getElementById('counter'); + // A1: last-rendered HTML per region — a poll that produces identical + // markup must not swap innerHTML (that destroyed hover/selection and any + // button press straddling the rebuild). + let lastRowsHtml = null; + let lastCommandsHtml = null; + let lastEventsHtml = null; function setStatus(text, isError) { status.textContent = text; @@ -230,27 +239,40 @@ RISK: terminal cosplay if decoration creeps past data; held by the palette law. ? `${escapeHtml(process.state)}` : '—'; const pid = process && process.pid != null ? process.pid : (reconciler && reconciler.pid != null ? reconciler.pid : null); - const seen = process ? fmtSeen(process.seen_ms_ago) : '—'; const attempt = reconciler ? `
attempt ${escapeHtml(reconciler.attempt)}${reconciler.ready ? ' · ready' : ''}
` : ''; const failure = reconciler && reconciler.failure ? `
${escapeHtml(reconciler.failure)}
` : ''; const killable = process && process.state === 'running'; - return ` + const edge = entry.edge + ? `e${escapeHtml(entry.edge.edge)} ${escapeHtml(entry.edge.state)}` + : '—'; + // "seen" is deliberately absent from this HTML: it is written per-poll + // as textContent by updateSeen() so a ticking relative age never + // invalidates the cached row markup. + return ` ${escapeHtml(node)}${attempt} ${stage} ${state} ${pid == null ? '—' : escapeHtml(pid)} - ${escapeHtml(seen)} - - ${failure ? `${failure}` : ''}`; + — + ${edge} + + + + + ${failure ? `${failure}` : ''}`; } - function fuse(processes, reconcilerNodes) { + function fuse(processes, reconcilerNodes, edges) { const byNode = new Map(); const order = []; + const edgeByNode = new Map(); + for (const edge of edges || []) { + if (edge && edge.node != null) edgeByNode.set(edge.node, edge); + } for (const node of reconcilerNodes || []) { - byNode.set(node.id, { reconciler: node, process: null }); + byNode.set(node.id, { reconciler: node, process: null, edge: edgeByNode.get(node.id) || null }); order.push(node.id); } for (const process of processes || []) { @@ -258,7 +280,7 @@ RISK: terminal cosplay if decoration creeps past data; held by the palette law. if (entry) { entry.process = process; } else { - byNode.set(process.node, { reconciler: null, process }); + byNode.set(process.node, { reconciler: null, process, edge: edgeByNode.get(process.node) || null }); order.push(process.node); } } @@ -268,21 +290,15 @@ RISK: terminal cosplay if decoration creeps past data; held by the palette law. function render(processSnapshot, reconcilerSnapshot) { const processes = (processSnapshot && processSnapshot.nodes) || []; const reconcilerNodes = (reconcilerSnapshot && reconcilerSnapshot.nodes) || []; - const fused = fuse(processes, reconcilerNodes); + const edges = (reconcilerSnapshot && reconcilerSnapshot.edges) || []; + const fused = fuse(processes, reconcilerNodes, edges); empty.hidden = fused.length > 0; - rows.innerHTML = fused.map(renderRow).join(''); - for (const button of rows.querySelectorAll('button[data-node]')) { - button.addEventListener('click', () => { - button.disabled = true; - fetch('/control/kill', { - method: 'POST', - headers: { 'content-type': 'application/json' }, - body: JSON.stringify({ Kill: { node: button.dataset.node } }) - }).then((response) => { - setStatus(response.ok ? 'kill issued' : 'kill failed: HTTP ' + response.status, !response.ok); - }).catch((error) => setStatus('kill failed: ' + error, true)); - }); + const rowsHtml = fused.map(renderRow).join(''); + if (rowsHtml !== lastRowsHtml) { + rows.innerHTML = rowsHtml; + lastRowsHtml = rowsHtml; } + updateSeen(fused); if (reconcilerSnapshot) { counter.hidden = false; setSeg('ready-seg', reconcilerSnapshot.ready); @@ -295,6 +311,17 @@ RISK: terminal cosplay if decoration creeps past data; held by the palette law. } } + // Relative "seen" ages tick every poll but must not churn row markup: + // write them as textContent against the row keyed by data-node. + function updateSeen(fused) { + const byNode = new Map(fused.map((entry) => [entry.reconciler ? entry.reconciler.id : entry.process.node, entry])); + for (const row of rows.querySelectorAll('tr[data-node]')) { + const entry = byNode.get(row.dataset.node); + const cell = row.querySelector('td[data-seen]'); + if (cell) cell.textContent = entry && entry.process ? fmtSeen(entry.process.seen_ms_ago) : '—'; + } + } + // Segment display: value over ghost eights; every unlit place stays drawn. function setSeg(id, value) { const text = String(value == null ? '—' : value); @@ -311,12 +338,18 @@ RISK: terminal cosplay if decoration creeps past data; held by the palette law. function renderFeeds(feed) { const recent = feed.slice().reverse(); - document.getElementById('commands').innerHTML = - recent.map((line) => feedLine(line, 'command')).join('') || '
  • none
    • '; + const commandsHtml = recent.map((line) => feedLine(line, 'command')).join('') || '
  • none
    • '; const eventKinds = new Set(['command']); - document.getElementById('events').innerHTML = - recent.filter((line) => !eventKinds.has(line.kind)).map((line) => feedLine(line)).join('') + const eventsHtml = recent.filter((line) => !eventKinds.has(line.kind)).map((line) => feedLine(line)).join('') || '
  • none
    • '; + if (commandsHtml !== lastCommandsHtml) { + document.getElementById('commands').innerHTML = commandsHtml; + lastCommandsHtml = commandsHtml; + } + if (eventsHtml !== lastEventsHtml) { + document.getElementById('events').innerHTML = eventsHtml; + lastEventsHtml = eventsHtml; + } } function nodeCount() { @@ -341,6 +374,32 @@ RISK: terminal cosplay if decoration creeps past data; held by the palette law. post('/control/remove', { Remove: { count: nodeCount() } }, 'remove'); }); + // A2: one delegated listener on the persistent tbody. Per-render binding + // dropped any press that straddled a row rebuild; delegation survives them. + rows.addEventListener('click', (event) => { + const button = event.target.closest('button[data-node]'); + if (!button || !rows.contains(button)) return; + const action = button.dataset.action || 'kill'; + const route = action === 'edge' ? '/control/edge' : '/control/kill'; + const body = action === 'edge' + ? { EstablishEdge: { node: button.dataset.node } } + : { Kill: { node: button.dataset.node } }; + const label = action === 'edge' ? 'edge' : 'kill'; + if (action === 'edge') button.disabled = false; // edges can be re-issued + else button.disabled = true; + fetch(route, { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(body) + }).then((response) => { + if (!response.ok && action !== 'edge') button.disabled = false; + setStatus(response.ok ? label + ' issued' : label + ' failed: HTTP ' + response.status, !response.ok); + }).catch((error) => { + if (action !== 'edge') button.disabled = false; + setStatus(label + ' failed: ' + error, true); + }); + }); + async function poll() { try { const [control, reconciler] = await Promise.allSettled([ diff --git a/crates/dashboard/src/server.rs b/crates/dashboard/src/server.rs index b93a7c1..794fd5b 100644 --- a/crates/dashboard/src/server.rs +++ b/crates/dashboard/src/server.rs @@ -46,7 +46,8 @@ fn router(state: AppState) -> Router { let router = router .route("/control/kill", axum::routing::post(control_kill)) .route("/control/provision", axum::routing::post(control_provision)) - .route("/control/remove", axum::routing::post(control_remove)); + .route("/control/remove", axum::routing::post(control_remove)) + .route("/control/edge", axum::routing::post(control_edge)); router.with_state(state) } @@ -98,6 +99,22 @@ async fn control_remove( } } +#[cfg(feature = "demo-control")] +async fn control_edge( + Json(command): Json, +) -> impl IntoResponse { + match command { + crate::control::ControlCommand::EstablishEdge { .. } => { + if crate::control::dispatch(command) { + StatusCode::ACCEPTED + } else { + StatusCode::SERVICE_UNAVAILABLE + } + } + _ => StatusCode::UNPROCESSABLE_ENTITY, + } +} + async fn root_page(State(state): State) -> Response { // Home is the fleet control-plane view. match state.views.html("fleet") { diff --git a/crates/iroh-driver/src/iroh_driver.rs b/crates/iroh-driver/src/iroh_driver.rs index 348d919..e234b3b 100644 --- a/crates/iroh-driver/src/iroh_driver.rs +++ b/crates/iroh-driver/src/iroh_driver.rs @@ -291,6 +291,10 @@ pub struct IrohDriver { telemetry_reads: Arc>>, /// Logical edge events emitted by driver-owned EDGE_ALPN byte pumps. edge_events: Arc>>, + /// When set, TELEMETRY_ALPN connections are NOT claimed by the + /// driver-owned telemetry ingress; the application drains them via + /// [`Self::drain_accepted_for_alpn`] (e.g. to serve pulls itself). + retain_telemetry_conns: Arc, next_edge_stream_group: Arc, /// Frames read by per-connection reader tasks, drained by the engine-hosted /// adapter pump ([`Self::install_actor_bridge_pump`]). This decouples network @@ -495,9 +499,10 @@ impl IrohDriver { pending_joins: Arc::new(Mutex::new(Vec::new())), dialing: Arc::new(Mutex::new(HashSet::new())), accepted_conns, + edge_events, + retain_telemetry_conns: Arc::new(std::sync::atomic::AtomicBool::new(false)), other_accepted_conns, telemetry_reads, - edge_events, next_edge_stream_group: Arc::new(AtomicU64::new(1)), incoming: Arc::new(Mutex::new(Vec::new())), evict: Arc::new(Mutex::new(Vec::new())), @@ -1124,6 +1129,15 @@ impl IrohDriver { let _ = rx.recv(); } + /// Keep TELEMETRY_ALPN connections unclaimed by the driver-owned + /// telemetry ingress, so the application can drain them with + /// [`Self::drain_accepted_for_alpn`] and serve pulls itself. Call + /// before [`Self::install_actor_bridge_pump`]. + pub fn retain_telemetry_connections(&self) { + self.retain_telemetry_conns + .store(true, std::sync::atomic::Ordering::Relaxed); + } + /// Install engine-hosted interval tasks that drive adapter progression /// (actor-bridge ingress/egress, telemetry ingress, edge ingress). After /// this call, the application must not manually pump these adapters @@ -1142,6 +1156,7 @@ impl IrohDriver { other_accepted_conns: Arc::clone(&self.other_accepted_conns), telemetry_reads: Arc::clone(&self.telemetry_reads), edge_events: Arc::clone(&self.edge_events), + retain_telemetry_conns: Arc::clone(&self.retain_telemetry_conns), next_edge_stream_group: Arc::clone(&self.next_edge_stream_group), dialing: Arc::clone(&self.dialing), peer_auth: self.peer_auth.clone(), @@ -1181,6 +1196,7 @@ struct AdapterPump { other_accepted_conns: Arc, Connection)>>>, telemetry_reads: Arc>>, edge_events: Arc>>, + retain_telemetry_conns: Arc, next_edge_stream_group: Arc, dialing: Arc>>, peer_auth: Option>>, @@ -1471,8 +1487,15 @@ impl AdapterPump { } /// Claim accepted telemetry connections and read them inside driver-owned - /// tasks. + /// tasks. Skipped when the application retained TELEMETRY_ALPN + /// connections to serve pulls itself. fn pump_telemetry_ingress(&self) { + if self + .retain_telemetry_conns + .load(std::sync::atomic::Ordering::Relaxed) + { + return; + } let drained = { let mut pending = self.other_accepted_conns.lock(); let mut keep = Vec::new(); diff --git a/xtask/Cargo.toml b/xtask/Cargo.toml index 1b1f6f8..b9a4cb1 100644 --- a/xtask/Cargo.toml +++ b/xtask/Cargo.toml @@ -14,6 +14,7 @@ swactor-process = { path = "../crates/process" } provisioning = { path = "../crates/provisioning" } telemetry = { path = "../crates/telemetry" } dashboard = { path = "../crates/dashboard", features = ["demo-control"] } +data-plane = { path = "../crates/data-plane" } iroh-driver = { path = "../crates/iroh-driver" } distribution = { path = "../crates/distribution" } iroh = "0.98" diff --git a/xtask/src/provisioning_demo/bootstrap.rs b/xtask/src/demo/bootstrap.rs similarity index 97% rename from xtask/src/provisioning_demo/bootstrap.rs rename to xtask/src/demo/bootstrap.rs index d88dd84..fe0a415 100644 --- a/xtask/src/provisioning_demo/bootstrap.rs +++ b/xtask/src/demo/bootstrap.rs @@ -17,7 +17,7 @@ use swactor_process::{ProcessOutputConfig, ProcessSpec, spawn_local_process}; use provisioning::bootstrap::{BootstrapLogic, LogicProbe, NodeLaunchSpec}; -use crate::provisioning_demo::provider::{NodeManager, NodeRelayActor, NodeRuntime}; +use crate::demo::provider::{NodeManager, NodeRelayActor, NodeRuntime}; /// Local-process bootstrap logic. The spawned process actor's lifecycle /// reports fold into the shared [`NodeManager`] registry via @@ -77,6 +77,7 @@ impl BootstrapLogic for LocalProcessLogic { exited: None, spawn_failed: None, last_announce_ms: None, + endpoint_addr: None, }); Ok(()) } diff --git a/xtask/src/provisioning_demo/control.rs b/xtask/src/demo/control.rs similarity index 71% rename from xtask/src/provisioning_demo/control.rs rename to xtask/src/demo/control.rs index b6a1ceb..c68a5a5 100644 --- a/xtask/src/provisioning_demo/control.rs +++ b/xtask/src/demo/control.rs @@ -1,15 +1,15 @@ //! Control plumbing: dashboard control commands → supervisor actor. //! //! The dashboard (under `demo-control`) dispatches into a std mpsc channel; -//! an engine task forwards each command as a `SupervisorMsg::Control` message -//! to the supervisor actor. +//! a blocking-pool task forwards each command as a `SupervisorMsg::Control` +//! message to the supervisor actor. use std::sync::mpsc as std_mpsc; use swactor::runtime::ExternalSender; use swactor_engine::EngineHandle; -use crate::provisioning_demo::feed::SupervisorMsg; +use crate::demo::feed::SupervisorMsg; /// Wire the dashboard control channel to the supervisor actor. pub fn install( @@ -21,7 +21,10 @@ pub fn install( dashboard::control::set_control_sender(control_tx); let engine = engine.clone(); - engine.spawn(async move { + // A std-mpsc recv blocks its thread, so this forwarder must live on the + // engine's blocking pool — as an async task it would park one of the + // (two) Tokio workers indefinitely and starve the reconciler ticks. + engine.spawn_blocking(move || { loop { match control_rx.recv() { Ok(command) => { diff --git a/xtask/src/provisioning_demo/docker.rs b/xtask/src/demo/docker.rs similarity index 98% rename from xtask/src/provisioning_demo/docker.rs rename to xtask/src/demo/docker.rs index 540b830..258692c 100644 --- a/xtask/src/provisioning_demo/docker.rs +++ b/xtask/src/demo/docker.rs @@ -27,7 +27,7 @@ use swactor_process::{ProcessOutputConfig, ProcessSpec, spawn_local_process}; use provisioning::bootstrap::{BootstrapLogic, LogicProbe, NodeLaunchSpec}; -use crate::provisioning_demo::provider::{NodeManager, NodeRelayActor, NodeRuntime}; +use crate::demo::provider::{NodeManager, NodeRelayActor, NodeRuntime}; /// Generic label present on every demo container/network (sweep key). pub const SWEEP_LABEL: &str = "swactor-demo"; @@ -143,6 +143,7 @@ impl BootstrapLogic for DockerProcessLogic { exited: None, spawn_failed: None, last_announce_ms: None, + endpoint_addr: None, }); Ok(()) } @@ -324,7 +325,7 @@ pub fn sweep_run(launch: &DockerLaunch) { /// 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…"); + println!("demo --docker: building static node binary…"); let bin = root .join("target") .join(IMAGE_TARGET) @@ -355,7 +356,7 @@ fn build_image(root: &Path, image: &str) -> Result<(), String> { })(); let _ = std::fs::remove_dir_all(&staging); result.map(|_| { - println!("provisioning-reconciler-demo --docker: image {image} ready"); + println!("demo --docker: image {image} ready"); }) } diff --git a/xtask/src/provisioning_demo/docker/Dockerfile b/xtask/src/demo/docker/Dockerfile similarity index 83% rename from xtask/src/provisioning_demo/docker/Dockerfile rename to xtask/src/demo/docker/Dockerfile index 6004db7..b67604c 100644 --- a/xtask/src/provisioning_demo/docker/Dockerfile +++ b/xtask/src/demo/docker/Dockerfile @@ -1,6 +1,6 @@ # 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 +# xtask/src/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. diff --git a/xtask/src/demo/edge.rs b/xtask/src/demo/edge.rs new file mode 100644 index 0000000..038df8b --- /dev/null +++ b/xtask/src/demo/edge.rs @@ -0,0 +1,749 @@ +//! Demo data-plane edges: real `EdgeRuntime` sessions over `EDGE_ALPN`. +//! +//! Supervisor side: one [`EdgeSession`] per established edge — an +//! `EdgeRuntime` (outbound) with its own arena and a recorder worker, +//! polled on the supervisor tick. Node side: [`NodeEdgeAgent`] — the actor +//! the node's actor bridge routes `EdgeProvision` gossip to; it provisions +//! the node's (single) inbound edge, polls it, mirrors observations onto +//! the `node.edge` telemetry channel, and answers control-plane +//! [`EdgeAck`] gossip which terminates the supervisor's provision +//! retries. No control flow reads telemetry: establishment decisions come +//! from the supervisor-local FSM plus the gossip ack path. `node.edge` +//! records are render-only dashboard material. +//! +//! Edge ids are supervisor-allocated and travel in the provision message, +//! so both sides agree on the wire preamble tag. Provision gossip is +//! retried by the supervisor until an ack lands (the node re-acks on +use std::sync::Arc; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; + +use data_plane::arena::{ArenaConfig, ArenaManager}; +use data_plane::edge_lifecycle::{ + DType, NodeId, ObjectKind, ObjectSpec as EdgeObjectSpec, ProvisionRx, ProvisionTx, RingSpec, +}; +use data_plane::edge_runtime::{EdgeRuntime, LoadedObject, WorkerPort}; +use data_plane::edge_wire::EdgeTransport; +use data_plane::ids::{EdgeId, RunId}; +pub use data_plane::edge_runtime::Observation; +use data_plane::object_record::{self, ObjectRecord}; +use iroh::EndpointAddr; +use telemetry::TelemetryProducer; + +/// Wire tag of the edge-provision gossip frame (supervisor → node). +pub const EDGE_PROVISION_TAG: &str = "xtask_demo/EdgeProvision/1"; +/// Wire tag of the edge-ack gossip frame (node → supervisor). +pub const EDGE_ACK_TAG: &str = "xtask_demo/EdgeAck/1"; + +/// Telemetry channel carrying node-side edge observations (render-only). +pub const NODE_EDGE_CHANNEL: &str = "node.edge"; +/// How long a supervisor session waits for an ack before faulting. +pub const PROVISION_ACK_TIMEOUT: Duration = Duration::from_secs(10); +/// Upper bound on one edge connect handshake before the session faults. +pub const EDGE_CONNECT_TIMEOUT: Duration = Duration::from_secs(10); + +/// Provision gossip retry period. +pub const PROVISION_RETRY_PERIOD: Duration = Duration::from_secs(1); + +/// Supervisor → node: arm your inbound edge `edge_id`. +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] +pub struct EdgeProvision { + pub attempt: u64, + pub edge_id: u64, + pub at_ms: u64, +} + +/// Node → supervisor: my inbound `edge_id` reached `outcome`. +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] +pub struct EdgeAck { + pub attempt: u64, + pub edge_id: u64, + /// `"ready"` or `"fault:"`. + pub outcome: String, + pub at_ms: u64, +} + +// Both binaries are the same crate, so the demo edge specs are shared +// constants — the provision message only carries identity, never specs. +pub fn demo_object_spec() -> EdgeObjectSpec { + EdgeObjectSpec { + kind: ObjectKind::Activation, + dtype: DType::F16, + max_extent_bytes: 4096, + } +} + +pub fn demo_ring_spec() -> RingSpec { + RingSpec { + header_bytes: 0, + data_bytes: 8192, + alignment: 64, + } +} + +pub fn demo_parse_spec() -> object_record::ObjectSpec { + object_record::ObjectSpec { + max_extent: 4096, + alignment: 16, + layout: object_record::ObjectLayout::Token, + } +} + +fn boot_arena(node_id: NodeId) -> ArenaManager { + ArenaManager::boot(ArenaConfig { + node_id, + reservation_ceiling: 1 << 20, + base_alignment: 64, + }) + .expect("demo arena boots") +} + +pub fn unix_ms(now: SystemTime) -> u64 { + now.duration_since(UNIX_EPOCH) + .map(|duration| duration.as_millis() as u64) + .unwrap_or(0) +} + +// ─── Shared transport ─────────────────────────────────────────────────────── + +/// `EdgeTransport` over a shared driver: outbound writers are driver-owned +/// send pumps (`&self` through the Arc); inbound events drain the shared +/// edge-event queue. Supervisor sessions (all outbound) and the node agent +/// (single inbound) both use this — outbound-only pollers never race on +/// `drain_events` because inbound events only exist where edges are +/// accepted. +pub struct DriverTransport(pub Arc); +impl EdgeTransport for DriverTransport { + type Writer = iroh_driver::EdgeSendHandle; + type PeerAddr = EndpointAddr; + + fn open_writer(&mut self, edge_id: EdgeId, peer: &EndpointAddr) -> Result { + // The driver's send pump blocks its calling thread until the + // connect handshake completes — indefinitely for a dead peer. + // Bound it: run the spawn on a helper thread and give up after + // EDGE_CONNECT_TIMEOUT. (This runs on the edge pump thread, but a + // EDGE_CONNECT_TIMEOUT. (This runs on the edge pump thread, but a + // dead node must not wedge edge polling for the whole cluster.) + let driver = Arc::clone(&self.0); + let peer = peer.clone(); + let (tx, rx) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + let _ = tx.send(driver.spawn_edge_send_pump(peer, edge_id.0)); + }); + rx.recv_timeout(EDGE_CONNECT_TIMEOUT) + .map_err(|error| format!("edge {} connect handshake: {error}", edge_id.0))? + } + + fn drain_events(&mut self) -> Vec { + self.0.drain_edge_events() + } +} + +// ─── Recorder worker ──────────────────────────────────────────────────────── + +/// A `WorkerPort` that records effects instead of touching a GPU: the demo +/// has no worker, but the real edge lifecycle leases and installs rings on +/// both sides, and those effects must succeed for the FSM to reach Ready. +#[derive(Default)] +pub struct DemoWorkerPort { + pub installed: Vec<(EdgeId, data_plane::ids::RingId)>, +} + +impl WorkerPort for DemoWorkerPort { + fn install_ring( + &mut self, + edge_id: EdgeId, + ring_id: data_plane::ids::RingId, + _direction: data_plane::edge_lifecycle::RingDirection, + _layout: &data_plane::arena::RingLayout, + _object_spec: &EdgeObjectSpec, + ) -> Result<(), String> { + self.installed.push((edge_id, ring_id)); + Ok(()) + } + + fn uninstall_ring(&mut self, _ring_id: data_plane::ids::RingId) -> Result<(), String> { + Ok(()) + } + + fn load_object( + &mut self, + _edge_id: EdgeId, + _ring_id: data_plane::ids::RingId, + record: &ObjectRecord, + _spec: &object_record::ObjectSpec, + ) -> Result { + Ok(LoadedObject { + object_id: record.object_id.0, + sequence: record.sequence, + handle_generation: 1, + handle_id: record.object_id.0, + }) + } +} + +/// Supervisor-side state for one outbound edge toward a node. +pub struct EdgeSession { + pub edge_id: EdgeId, + pub attempt: u64, + pub logical_node: String, + pub peer: EndpointAddr, + runtime: EdgeRuntime, + arena: ArenaManager, + worker: DemoWorkerPort, + /// Node ack outcome, once its gossip landed. + pub acked: Option, + /// Local FSM observed outbound readiness. + pub local_ready: bool, + pub faulted: bool, + /// First provision send; drives retry + timeout. + pub started: SystemTime, + pub last_send: SystemTime, +} + +impl EdgeSession { + pub fn new( + edge_id: EdgeId, + attempt: u64, + logical_node: String, + peer: EndpointAddr, + ) -> Self { + let mut runtime = EdgeRuntime::new(NodeId(0)); + runtime.establish_outbound( + ProvisionTx { + run_id: RunId(1), + edge_id, + local_node_id: NodeId(0), + consumer_node_id: NodeId(attempt), + object_spec: demo_object_spec(), + ring_spec: demo_ring_spec(), + }, + peer.clone(), + ); + Self { + edge_id, + attempt, + logical_node, + peer, + runtime, + arena: boot_arena(NodeId(0)), + worker: DemoWorkerPort::default(), + acked: None, + local_ready: false, + faulted: false, + started: SystemTime::now(), + last_send: SystemTime::now(), + } + } + + /// One poll: drive the lifecycle; return new observations. + pub fn poll(&mut self, driver: &Arc) -> (Vec, Option) { + let mut transport = DriverTransport(Arc::clone(driver)); + let result = self + .runtime + .poll(&mut transport, &mut self.arena, &mut self.worker); + let observations = self.runtime.take_observations(); + for observation in &observations { + if matches!(observation, Observation::EdgeReady { .. }) { + self.local_ready = true; + } + if matches!(observation, Observation::EdgeFaulted { .. }) { + self.faulted = true; + } + } + if result.is_err() { + self.faulted = true; + } + (observations, result.err()) + } + + /// Provision gossip payload for this session. + pub fn provision(&self) -> EdgeProvision { + EdgeProvision { + attempt: self.attempt, + edge_id: self.edge_id.0, + at_ms: unix_ms(SystemTime::now()), + } + } + + /// Display state for the dashboard snapshot. + pub fn state(&self) -> &'static str { + if self.faulted { + "faulted" + } else if self.acked.as_deref() == Some("ready") && self.local_ready { + "ready" + } else { + "provisioning" + } + } + + pub fn ring_id(&self) -> Option { + self.runtime.outbound_ring_id().map(|ring| ring.0) + } +} + +/// Command from the supervisor actor into the edge pump thread. The pump +/// thread solely owns the sessions: every mutation travels through this +/// channel, so the actor never touches pump-owned state (and the pump's +/// blocking connect handshakes can never stall the actor). +pub enum EdgePumpCmd { + /// Adopt a freshly created session (replaces any session for the same + /// attempt). + Establish(Box), + /// A node acked its inbound edge. + Ack(EdgeAck), + /// The current set of live node attempts; sessions for other attempts + /// are torn down. + LiveAttempts(Vec), + /// Tear everything down (supervisor shutdown). + DropAll, +} + +/// One pump→actor update: feed lines plus the full edge-state mirror for +/// the dashboard snapshot. +#[derive(Clone, Debug, Default)] +pub struct EdgePumpUpdate { + pub feed: Vec<(String, String)>, + pub states: Vec, +} + +/// Start the edge pump on the engine's blocking pool. Every tick: apply +/// pending commands, retry unacked provisions, poll each session's +/// runtime (the blocking connect handshakes belong on this thread — see +/// `control.rs` for the same constraint), and report an update to the +/// supervisor actor. +pub fn start_edge_pump( + engine: &swactor_engine::EngineHandle, + driver: Arc, + cmds: std::sync::mpsc::Receiver, + sender: swactor::runtime::ExternalSender, + supervisor: Arc>, +) { + let engine = engine.clone(); + engine.spawn_blocking(move || { + let mut sessions: Vec = Vec::new(); + let mut next_poll = std::time::Instant::now(); + loop { + // Drain commands (blocking with the remaining tick budget). + loop { + let timeout = next_poll.saturating_duration_since(std::time::Instant::now()); + match cmds.recv_timeout(timeout) { + Ok(EdgePumpCmd::Establish(session)) => { + let attempt = session.attempt; + sessions.retain(|existing| existing.attempt != attempt); + sessions.push(*session); + } + Ok(EdgePumpCmd::Ack(ack)) => { + if let Some(session) = sessions + .iter_mut() + .find(|session| session.edge_id.0 == ack.edge_id) + { + session.acked = Some(ack.outcome.clone()); + } + } + Ok(EdgePumpCmd::LiveAttempts(live)) => { + let mut i = 0; + while i < sessions.len() { + if live.contains(&sessions[i].attempt) { + i += 1; + } else { + let dead = sessions.remove(i); + // Dropping the session drops its writer, + // finishing the edge stream. + eprintln!( + "demo: edge {} torn down (node gone)", + dead.edge_id.0 + ); + } + } + } + Ok(EdgePumpCmd::DropAll) => sessions.clear(), + Err(std::sync::mpsc::RecvTimeoutError::Timeout) => break, + Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => return, + } + } + next_poll = std::time::Instant::now() + Duration::from_millis(250); + + let mut update = EdgePumpUpdate::default(); + for session in sessions.iter_mut() { + pump_one(session, &driver, &mut update); + } + update.states = sessions.iter().map(session_state_json).collect(); + if let Some(addr) = supervisor.get() { + let _ = sender.send_to( + addr.clone(), + crate::demo::feed::SupervisorMsg::EdgeUpdate(update), + ); + } + } + }); +} + +fn session_state_json(session: &EdgeSession) -> serde_json::Value { + serde_json::json!({ + "edge": session.edge_id.0, + "node": session.logical_node, + "state": session.state(), + "acked": session.acked.is_some(), + "ring": session.ring_id(), + }) +} + +fn pump_one( + session: &mut EdgeSession, + driver: &Arc, + update: &mut EdgePumpUpdate, +) { + // Retry the provision gossip until the node acks (the node re-acks + // duplicates, so retries are safe). + if session.acked.is_none() && !session.faulted { + let since_send = session + .last_send + .elapsed() + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + let since_start = session + .started + .elapsed() + .map(|d| d.as_millis() as u64) + .unwrap_or(0); + if since_start > PROVISION_ACK_TIMEOUT.as_millis() as u64 { + session.faulted = true; + update.feed.push(( + session.logical_node.clone(), + format!("edge {}: faulted (provision ack timeout)", session.edge_id.0), + )); + } else if since_send >= PROVISION_RETRY_PERIOD.as_millis() as u64 { + if let Ok(bytes) = serde_json::to_vec(&session.provision()) { + driver.send_tagged_gossip( + session.peer.clone(), + EDGE_PROVISION_TAG.as_bytes(), + bytes, + ); + } + session.last_send = SystemTime::now(); + } + } + let (observations, error) = session.poll(driver); + for observation in &observations { + match observation { + Observation::EdgeReady { .. } => update.feed.push(( + session.logical_node.clone(), + format!("edge {}: outbound ready", session.edge_id.0), + )), + Observation::EdgeFaulted { reason, .. } => update.feed.push(( + session.logical_node.clone(), + format!("edge {}: faulted ({reason:?})", session.edge_id.0), + )), + _ => {} + } + } + if let Some(error) = error { + update.feed.push(( + session.logical_node.clone(), + format!("edge {}: faulted ({error})", session.edge_id.0), + )); + } +} + +// ─── Node-side agent ──────────────────────────────────────────────────────── + +/// Messages into the node edge agent. +#[derive(Clone, Debug)] +pub enum NodeEdgeMsg { + /// Decoded `EdgeProvision` gossip from the supervisor. + Provision(EdgeProvision), + /// Poll tick from the engine interval. + Tick, +} + +/// Node-side owner of the (single) inbound edge. Provisions on gossip, +/// polls the real `EdgeRuntime` lifecycle, mirrors observations onto the +/// `node.edge` telemetry channel, and acks readiness/fault on the control +/// plane. A provision for a different edge id rebuilds the inbound edge +/// (the supervisor replaces sessions); a duplicate for the current edge id +/// only re-acks. +pub struct NodeEdgeAgent { + attempt: u64, + logical_node: String, + supervisor_addr: EndpointAddr, + /// Filled right after the driver is Arc-wrapped (the bridge setup + /// needs `&mut` on the driver first; same pattern as the supervisor + /// slot). + driver_slot: Arc>>, + producer: TelemetryProducer, + edge_channel: telemetry::ChannelId, + runtime: EdgeRuntime, + arena: ArenaManager, + worker: DemoWorkerPort, + edge_id: Option, + acked_ready_sent: bool, +} + +impl NodeEdgeAgent { + #[allow(clippy::too_many_arguments)] + pub fn new( + attempt: u64, + logical_node: String, + supervisor_addr: EndpointAddr, + driver_slot: Arc>>, + producer: TelemetryProducer, + edge_channel: telemetry::ChannelId, + ) -> Self { + Self { + attempt, + logical_node, + supervisor_addr, + driver_slot, + producer, + edge_channel, + runtime: EdgeRuntime::new(NodeId(attempt)), + arena: boot_arena(NodeId(attempt)), + worker: DemoWorkerPort::default(), + edge_id: None, + acked_ready_sent: false, + } + } + + fn send_ack(&self, outcome: &str) { + let Some(driver) = self.driver_slot.get() else { + return; + }; + let ack = EdgeAck { + attempt: self.attempt, + edge_id: self.edge_id.map(|edge| edge.0).unwrap_or(0), + outcome: outcome.to_owned(), + at_ms: unix_ms(SystemTime::now()), + }; + let Ok(bytes) = serde_json::to_vec(&ack) else { + return; + }; + driver.send_tagged_gossip( + self.supervisor_addr.clone(), + EDGE_ACK_TAG.as_bytes(), + bytes, + ); + } + + fn submit_observation(&self, observation: &Observation) { + let payload = self.observation_json(observation); + if let Ok(bytes) = serde_json::to_vec(&payload) { + self.producer.submit_bytes(self.edge_channel, bytes); + } + } + + fn observation_json(&self, observation: &Observation) -> serde_json::Value { + use serde_json::json; + let base = |kind: &str| { + json!({ + "at_ms": unix_ms(SystemTime::now()), + "node": self.logical_node, + "attempt": self.attempt, + "edge": self.edge_id.map(|edge| edge.0), + "kind": kind, + }) + }; + match observation { + Observation::StreamArrived { edge_id, stream_id } => { + let mut value = base("stream-arrived"); + value["edge"] = json!(edge_id.0); + value["stream"] = json!(stream_id.0); + value + } + Observation::BytesRead { + edge_id, + stream_id, + byte_count, + } => { + let mut value = base("bytes"); + value["edge"] = json!(edge_id.0); + value["stream"] = json!(stream_id.0); + value["bytes"] = json!(byte_count); + value + } + Observation::IngressRingWrite { + edge_id, + ring_id, + object_id, + sequence, + extent, + .. + } => { + let mut value = base("ingress-write"); + value["edge"] = json!(edge_id.0); + value["ring"] = json!(ring_id.0); + value["object"] = json!(object_id); + value["sequence"] = json!(sequence); + value["extent"] = json!(extent); + value + } + Observation::ObjectLoaded { + edge_id, + object: loaded, + .. + } => { + let mut value = base("object-loaded"); + value["edge"] = json!(edge_id.0); + value["object"] = json!(loaded.object_id); + value["sequence"] = json!(loaded.sequence); + value + } + Observation::ObjectFailed { edge_id, object_id } => { + let mut value = base("object-failed"); + value["edge"] = json!(edge_id.0); + if let Some(object) = object_id { + value["object"] = json!(object); + } + value + } + Observation::EdgeReady { edge_id, direction } => { + let mut value = base("edge-ready"); + value["edge"] = json!(edge_id.0); + value["direction"] = json!(format!("{direction:?}")); + value + } + Observation::EdgeFaulted { edge_id, reason } => { + let mut value = base("edge-faulted"); + value["edge"] = json!(edge_id.0); + value["reason"] = json!(format!("{reason:?}")); + value + } + Observation::EdgeStopped { edge_id } => { + let mut value = base("edge-stopped"); + value["edge"] = json!(edge_id.0); + value + } + } + } + + /// One poll pass: drive the runtime, mirror observations, ack + /// readiness/fault once (or again if the runtime was rebuilt). + fn pump(&mut self) { + let Some(driver) = self.driver_slot.get().cloned() else { + return; + }; + if self.edge_id.is_none() { + return; + } + let mut transport = DriverTransport(driver); + let result = self + .runtime + .poll(&mut transport, &mut self.arena, &mut self.worker); + let observations = self.runtime.take_observations(); + let mut ready = false; + let mut fault: Option = result.err(); + for observation in &observations { + self.submit_observation(observation); + match observation { + Observation::EdgeReady { .. } => ready = true, + Observation::EdgeFaulted { reason, .. } => { + fault = Some(format!("edge faulted: {reason:?}")); + } + _ => {} + } + } + if ready && !self.acked_ready_sent { + self.acked_ready_sent = true; + self.send_ack("ready"); + } + if let Some(fault) = fault { + self.send_ack(&format!("fault:{fault}")); + } + } +} + +impl swactor::actor::ActorInterface for NodeEdgeAgent { + type Incoming = NodeEdgeMsg; + type Response = (); + + fn handle(&mut self, _ctx: &swactor::actor::Ctx, msg: NodeEdgeMsg) { + match msg { + NodeEdgeMsg::Provision(provision) => { + if self.attempt != provision.attempt { + // Stale provision (replaced attempt): let it drop. + return; + } + match self.edge_id { + Some(current) if current.0 == provision.edge_id => { + // Duplicate retry: idempotent re-ack. + if self.acked_ready_sent { + self.send_ack("ready"); + } + } + Some(current) => { + // Supervisor replaced the session: rebuild inbound. + self.submit_observation(&Observation::EdgeStopped { + edge_id: current, + }); + self.edge_id = Some(EdgeId(provision.edge_id)); + self.runtime = EdgeRuntime::new(NodeId(self.attempt)); + self.arena = boot_arena(NodeId(self.attempt)); + self.worker = DemoWorkerPort::default(); + self.acked_ready_sent = false; + self.runtime.establish_inbound( + ProvisionRx { + run_id: RunId(1), + edge_id: EdgeId(provision.edge_id), + local_node_id: NodeId(self.attempt), + object_spec: demo_object_spec(), + ring_spec: demo_ring_spec(), + }, + demo_parse_spec(), + ); + self.pump(); + } + None => { + self.edge_id = Some(EdgeId(provision.edge_id)); + self.runtime.establish_inbound( + ProvisionRx { + run_id: RunId(1), + edge_id: EdgeId(provision.edge_id), + local_node_id: NodeId(self.attempt), + object_spec: demo_object_spec(), + ring_spec: demo_ring_spec(), + }, + demo_parse_spec(), + ); + self.pump(); + } + } + } + NodeEdgeMsg::Tick => self.pump(), + } + } +} + +// ─── Supervisor-side ack relay ────────────────────────────────────────────── + +/// Decodes `EdgeAck` gossip (via the supervisor's actor bridge) and +/// forwards it to the supervisor actor. Same pattern as the announce +/// relay. +pub struct EdgeAckRelay { + sender: swactor::runtime::ExternalSender, + supervisor: Arc>, +} + +impl EdgeAckRelay { + pub fn new( + sender: swactor::runtime::ExternalSender, + supervisor: Arc>, + ) -> Self { + Self { + sender, + supervisor, + } + } +} + +impl swactor::actor::ActorInterface for EdgeAckRelay { + type Incoming = EdgeAck; + type Response = (); + + fn handle(&mut self, _ctx: &swactor::actor::Ctx, ack: EdgeAck) { + if let Some(addr) = self.supervisor.get() { + let _ = self.sender.send_to( + addr.clone(), + crate::demo::feed::SupervisorMsg::EdgeAck(ack), + ); + } + } +} diff --git a/xtask/src/provisioning_demo/feed.rs b/xtask/src/demo/feed.rs similarity index 87% rename from xtask/src/provisioning_demo/feed.rs rename to xtask/src/demo/feed.rs index b8dd6e1..e023669 100644 --- a/xtask/src/provisioning_demo/feed.rs +++ b/xtask/src/demo/feed.rs @@ -26,7 +26,9 @@ use swactor::actor::{ActorInterface, Ctx}; use swactor_engine::EngineHandle; use telemetry::{ChannelContent, StreamDescriptor, TelemetryEndpoint, TelemetryProducer}; -use crate::provisioning_demo::provider::{ +use crate::demo::edge; +use crate::demo::edge::{EdgeAck, EdgePumpCmd, EdgeSession}; +use crate::demo::provider::{ DemoBackend, NodeManager, NodeTelemetry, register_node_channels, unix_ms, }; @@ -129,7 +131,7 @@ impl EffectExecutor for FeedExecutor<'_> { pub enum SupervisorMsg { Tick, Control(dashboard::control::ControlCommand), - Spawn(crate::provisioning_demo::provider::SpawnNodeRequest), + Spawn(crate::demo::provider::SpawnNodeRequest), /// Event from a per-node bootstrap actor. Bootstrap(provisioning::BootstrapEvent), /// A remote node's telemetry pull stream registered its header (stream @@ -140,7 +142,10 @@ pub enum SupervisorMsg { logical_node: String, attempt: u64, }, - /// Drain the cluster: desired → empty, stop every child, flag when done. + /// Control-plane ack from a node's edge agent (edge provisioning). + EdgeAck(EdgeAck), + /// State + feed update from the edge pump thread (sole session owner). + EdgeUpdate(crate::demo::edge::EdgePumpUpdate), Shutdown { drained: std::sync::Arc, }, @@ -167,7 +172,7 @@ pub struct SupervisorActor { pub driver: ClusterDriver, pub executor: IdempotentEffectExecutor, pub manager: NodeManager, - pub driver_handle: std::sync::Arc, + pub driver_handle: std::sync::Arc, pub telemetry: SupervisorTelemetry, pub events_channel: telemetry::ChannelId, pub snapshot_channel: telemetry::ChannelId, @@ -193,7 +198,12 @@ pub struct SupervisorActor { last_stages: BTreeMap)>, pub dashboard: dashboard::DashboardHandle, status_tick: u64, - launch: crate::provisioning_demo::LaunchStyle, + launch: crate::demo::LaunchStyle, + /// Command channel into the edge pump thread (sole session owner). + edge_cmd: std::sync::mpsc::Sender, + /// Last reported edge-state mirror from the pump (snapshot data). + edge_states: Vec, + next_edge_id: u64, } impl SupervisorActor { @@ -202,7 +212,7 @@ impl SupervisorActor { driver: ClusterDriver, executor: IdempotentEffectExecutor, manager: NodeManager, - driver_handle: std::sync::Arc, + driver_handle: std::sync::Arc, mut telemetry: SupervisorTelemetry, dashboard: dashboard::DashboardHandle, sender: swactor::runtime::ExternalSender, @@ -212,7 +222,8 @@ impl SupervisorActor { remote_sub: telemetry::TelemetrySubscription, initial_slots: Vec, run_id: RunId, - launch: crate::provisioning_demo::LaunchStyle, + launch: crate::demo::LaunchStyle, + edge_cmd: std::sync::mpsc::Sender, ) -> Self { let events_channel = telemetry.register("prov.reconciler.events"); let snapshot_channel = telemetry.register("prov.reconciler.snapshot"); @@ -240,6 +251,9 @@ impl SupervisorActor { dashboard, status_tick: 0, launch, + edge_cmd, + edge_states: Vec::new(), + next_edge_id: 0, } } @@ -270,7 +284,7 @@ impl SupervisorActor { fn spawn_node( &mut self, ctx: &Ctx, - request: crate::provisioning_demo::provider::SpawnNodeRequest, + request: crate::demo::provider::SpawnNodeRequest, ) { let attempt = request.attempt; @@ -281,11 +295,11 @@ impl SupervisorActor { let status_channel = register_node_channels(&telemetry.producer); let (kind, argv, mut env) = match &self.launch { - crate::provisioning_demo::LaunchStyle::Process { exe } => ( + crate::demo::LaunchStyle::Process { exe } => ( "process", vec![ exe.to_string_lossy().to_string(), - "provisioning-reconciler-demo".to_owned(), + "demo".to_owned(), "--demo-node".to_owned(), self.driver_handle.supervisor_addr_json.clone(), "--demo-attempt".to_owned(), @@ -293,26 +307,26 @@ impl SupervisorActor { ], Vec::new(), ), - crate::provisioning_demo::LaunchStyle::Docker(docker) => ( + crate::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), + crate::demo::docker::container_name(attempt), "--label".to_owned(), - format!("{}=1", crate::provisioning_demo::docker::SWEEP_LABEL), + format!("{}=1", crate::demo::docker::SWEEP_LABEL), "--label".to_owned(), format!( "{}={}", - crate::provisioning_demo::docker::RUN_LABEL, + crate::demo::docker::RUN_LABEL, docker.run_token ), "--network".to_owned(), docker.network.clone(), docker.image.clone(), - "provisioning-reconciler-demo".to_owned(), + "demo".to_owned(), "--demo-node".to_owned(), docker.supervisor_addr_json.clone(), "--demo-attempt".to_owned(), @@ -363,7 +377,7 @@ impl SupervisorActor { status_channel, }, ); - let runtime = crate::provisioning_demo::provider::NodeRuntime { + let runtime = crate::demo::provider::NodeRuntime { attempt, logical_node: request.logical_node.clone(), bootstrap, @@ -371,6 +385,7 @@ impl SupervisorActor { exited: None, spawn_failed: None, last_announce_ms: None, + endpoint_addr: None, }; let _ = request.reply.send(Ok(runtime)); } @@ -488,7 +503,8 @@ impl SupervisorActor { 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. + // death observation. Either way its edges are dead: the + // node process is gone. 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( @@ -837,6 +853,7 @@ impl SupervisorActor { "failure": managed.record.failed_reason, })); } + let edges_json = self.edge_states.clone(); let snapshot = json!({ "at_ms": unix_ms(now), @@ -845,6 +862,7 @@ impl SupervisorActor { "generation": self.driver.desired().generation, "converged": self.driver.is_converged(), "nodes": nodes_json, + "edges": edges_json, }); let bytes = serde_json::to_vec(&snapshot).expect("snapshot serializes"); self.telemetry @@ -935,6 +953,7 @@ impl ActorInterface for SupervisorActor { self.observe_world(now); self.replace_dead_ready_nodes(now); self.poll(now); + self.sweep_dead_edges(); self.emit_node_status(now); self.emit_feed(now); self.flush_telemetry(); @@ -952,12 +971,22 @@ impl ActorInterface for SupervisorActor { logical_node, attempt, } => self.register_remote_stream(header, logical_node, attempt), + SupervisorMsg::EdgeAck(ack) => { + let _ = self.edge_cmd.send(EdgePumpCmd::Ack(ack)); + } + SupervisorMsg::EdgeUpdate(update) => { + self.edge_states = update.states; + for (node, detail) in update.feed { + self.emit_event("edge", &node, detail); + } + } SupervisorMsg::Shutdown { drained } => { self.slots.clear(); let generation = self.driver.desired().generation.saturating_add(1); if let Err(error) = self.driver.update_desired(self.desired_shape(generation)) { eprintln!("demo: shutdown update_desired failed: {error}"); } + let _ = self.edge_cmd.send(EdgePumpCmd::DropAll); self.emit_event("control", "", "shutdown: desired → empty".to_owned()); drained.store(true, std::sync::atomic::Ordering::SeqCst); } @@ -1016,8 +1045,103 @@ impl SupervisorActor { eprintln!("demo: provision update_desired failed: {error}"); } } + dashboard::control::ControlCommand::EstablishEdge { node } => { + self.establish_edge(&node); + } } } + + // ─── Data-plane edges ────────────────────────────────────────────────── + + /// Establish (or replace) the supervisor→node edge from the dashboard. + /// Only control-plane facts gate this: the node must be registered, + /// alive, and have a fresh announce (endpoint addr + liveness). + fn establish_edge(&mut self, node: &str) { + let Some(runtime) = self.manager.find_by_stream_node(node) else { + self.emit_event("edge", node, "edge: unknown node".to_owned()); + return; + }; + if runtime.exited.is_some() { + self.emit_event("edge", node, "edge: node not running".to_owned()); + return; + } + let Some(addr_json) = runtime.endpoint_addr.clone() else { + self.emit_event( + "edge", + node, + "edge: node endpoint unknown (no announce yet)".to_owned(), + ); + return; + }; + // Announce freshness is control-plane liveness: a stale announce + // means the dial would target a dead endpoint. + if let Some(last) = runtime.last_announce_ms { + let age = unix_ms(SystemTime::now()).saturating_sub(last); + if age > (2 * crate::demo::HEARTBEAT_PERIOD).as_millis() as u64 + 2000 { + self.emit_event( + "edge", + node, + format!("edge: announce stale ({age}ms); refusing dial",), + ); + return; + } + } + let Ok(peer) = serde_json::from_str::(&addr_json) else { + self.emit_event("edge", node, "edge: node endpoint unparseable".to_owned()); + return; + }; + // One live session per node: a new edge replaces the old one (the + // pump owns the sessions; replacement happens on its thread). + self.next_edge_id += 1; + let edge_id = data_plane::ids::EdgeId(self.next_edge_id); + let session = EdgeSession::new( + edge_id, + runtime.attempt, + node.to_owned(), + peer.clone(), + ); + let provision = session.provision(); + if self + .edge_states + .iter() + .any(|state| state.get("node").and_then(|v| v.as_str()) == Some(node)) + { + self.emit_event("edge", node, "previous edge torn down (replaced)".to_owned()); + } + if self.edge_cmd.send(EdgePumpCmd::Establish(Box::new(session))).is_err() { + self.emit_event("edge", node, "edge: pump gone".to_owned()); + return; + } + if let Ok(bytes) = serde_json::to_vec(&provision) { + self.driver_handle + .driver + .send_tagged_gossip(peer, edge::EDGE_PROVISION_TAG.as_bytes(), bytes); + } + self.emit_event( + "edge", + node, + format!( + "edge {}: provision sent (outbound provisioning)", + edge_id.0 + ), + ); + } + + /// Tell the pump which node attempts are still live; it tears down + /// sessions for anything else (exit observed or registry entry gone). + fn sweep_dead_edges(&mut self) { + if self.edge_states.is_empty() { + return; + } + let live: Vec = self + .manager + .nodes() + .into_iter() + .filter(|runtime| runtime.exited.is_none()) + .map(|runtime| runtime.attempt) + .collect(); + let _ = self.edge_cmd.send(EdgePumpCmd::LiveAttempts(live)); + } } fn slot_group(slot: &str) -> provisioning::node::RunNodeGroupSpec { @@ -1051,7 +1175,7 @@ pub fn demo_group(id: &str, count: u32) -> provisioning::node::RunNodeGroupSpec boot: provisioning::node::BootSpec { ssh_user: "demo".to_owned(), verify_commands: vec!["true".to_owned()], - start_swactor_command: "xtask provisioning-reconciler-demo".to_owned(), + start_swactor_command: "xtask demo".to_owned(), stdout_sources: Vec::new(), stderr_sources: Vec::new(), env: Vec::new(), diff --git a/xtask/src/provisioning_demo/mod.rs b/xtask/src/demo/mod.rs similarity index 93% rename from xtask/src/provisioning_demo/mod.rs rename to xtask/src/demo/mod.rs index d53b8c6..87147c2 100644 --- a/xtask/src/provisioning_demo/mod.rs +++ b/xtask/src/demo/mod.rs @@ -1,4 +1,4 @@ -//! `cargo xtask provisioning-reconciler-demo` — a visual, human-checked E2E +//! `cargo xtask demo` — a visual, human-checked E2E //! sanity scenario for the provisioning reconciler. //! //! Supervisor role (default): a lightweight orchestrator — swactor engine + @@ -15,6 +15,7 @@ pub mod bootstrap; pub mod control; pub mod docker; +pub mod edge; pub mod feed; pub mod node; pub mod provider; @@ -99,7 +100,7 @@ fn resolve_exe() -> std::path::PathBuf { /// node roles) and the endpoint for outbound telemetry pulls. pub struct DemoDriverHandle { pub supervisor_addr_json: String, - pub(crate) driver: IrohDriver, + pub(crate) driver: std::sync::Arc, } impl DemoDriverHandle { @@ -169,7 +170,7 @@ impl provisioning::NodeTelemetryCollector for DemoTelemetryCollector { } } -/// Entry point: `provisioning-reconciler-demo [--port N] [--nodes N] +/// Entry point: `demo [--port N] [--nodes N] /// [--docker]` for the supervisor, or `--demo-node /// --demo-attempt ` for node children. pub fn run(args: &[String]) -> ExitCode { @@ -191,7 +192,7 @@ pub fn run(args: &[String]) -> ExitCode { match run_supervisor(args) { Ok(()) => ExitCode::SUCCESS, Err(error) => { - eprintln!("provisioning-reconciler-demo: {error}"); + eprintln!("demo: {error}"); ExitCode::FAILURE } } @@ -250,6 +251,11 @@ fn run_supervisor(args: &[String]) -> Result<(), String> { .map_err(|e| format!("iroh driver: {e}"))?; let supervisor_addr_json = serde_json::to_string(&driver.endpoint_addr()).map_err(|e| format!("addr: {e}"))?; + // 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()); { use distribution::transport_bridge::{Outbox, RelayMirror, RouteView}; use swactor_transport::CodecRegistry; @@ -266,13 +272,24 @@ fn run_supervisor(args: &[String]) -> Result<(), String> { sender.clone(), )) .map_err(|e| format!("spawn announce actor: {e}"))?; + let ack_relay = runtime + .spawn(edge::EdgeAckRelay::new( + sender.clone(), + supervisor_slot.clone(), + )) + .map_err(|e| format!("spawn edge ack relay: {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}"))) }); + codec.register_decoder::(edge::EDGE_ACK_TAG, |bytes| { + serde_json::from_slice(bytes) + .map_err(|e| swactor::Error::from(format!("edge ack decode: {e}"))) + }); let mut routes = std::collections::HashMap::new(); routes.insert(node::ANNOUNCE_TAG.to_owned(), announce); + routes.insert(edge::EDGE_ACK_TAG.to_owned(), ack_relay); let relay_mirror: RelayMirror = Arc::new(std::sync::RwLock::new(std::collections::HashMap::new())); let route_view: RouteView = @@ -292,7 +309,7 @@ fn run_supervisor(args: &[String]) -> Result<(), String> { let driver_handle = Arc::new(DemoDriverHandle { supervisor_addr_json: supervisor_addr_json.clone(), - driver, + driver: Arc::new(driver), }); // Dashboard. @@ -342,11 +359,6 @@ 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. @@ -377,6 +389,7 @@ fn run_supervisor(args: &[String]) -> Result<(), String> { )) as Box) }) }); + let edge_driver = std::sync::Arc::clone(&driver_handle.driver); let collector: Arc = Arc::new(DemoTelemetryCollector { engine: engine.handle(), @@ -386,6 +399,7 @@ fn run_supervisor(args: &[String]) -> Result<(), String> { supervisor_slot: supervisor_slot.clone(), }); + let (edge_cmd_tx, edge_cmd_rx) = std::sync::mpsc::channel::(); let supervisor = SupervisorActor::new( cluster_driver, executor, @@ -401,6 +415,7 @@ fn run_supervisor(args: &[String]) -> Result<(), String> { slots, RunId(1), launch.clone(), + edge_cmd_tx, ); // Control plane: dashboard → supervisor. @@ -444,9 +459,20 @@ fn run_supervisor(args: &[String]) -> Result<(), String> { .set(supervisor_addr.clone()) .expect("supervisor address slot set once"); - println!("provisioning-reconciler-demo: dashboard on http://localhost:{port}"); + // Edge pump on the blocking pool: it solely owns the edge sessions — + // edge runtime polls block their thread (connect handshakes), which + // must never run on a Tokio worker or hold a lock the actor needs. + edge::start_edge_pump( + &engine.handle(), + edge_driver, + edge_cmd_rx, + sender.clone(), + supervisor_slot.clone(), + ); + + println!("demo: dashboard on http://localhost:{port}"); println!(" /view/fleet — per-node cards (pid, lifecycle)"); - println!(" /view/demo-control — Fleet Control: stages, feeds, kill / provision"); + println!(" /view/demo-control — Fleet Control: stages, feeds, kill / provision / edge"); println!(" Ctrl-C to tear down."); // Block until Ctrl-C (synchronous signal flag — the wait must not depend diff --git a/xtask/src/provisioning_demo/node.rs b/xtask/src/demo/node.rs similarity index 77% rename from xtask/src/provisioning_demo/node.rs rename to xtask/src/demo/node.rs index fd6111e..dcb1fba 100644 --- a/xtask/src/provisioning_demo/node.rs +++ b/xtask/src/demo/node.rs @@ -33,7 +33,8 @@ use distribution::node::DistributedNodeConfig; use iroh_driver::{IrohDriver, IrohDriverConfig, TELEMETRY_ALPN, spawn_pull_server}; use telemetry::{ChannelContent, TelemetryEndpoint, TelemetryProducer}; -use crate::provisioning_demo::HEARTBEAT_PERIOD; +use crate::demo::edge; +use crate::demo::HEARTBEAT_PERIOD; /// Wire tag of the announce gossip frame (`CodecRegistry` decode key on the /// supervisor side; raw tag bytes on the node side). @@ -88,20 +89,22 @@ pub fn run_node_role(supervisor_addr_json: &str, attempt: u64) -> Result<(), Str // Bind the driver, then keep it alive for the process lifetime: dropping // 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}"))?, - ); + // the supervisor's pull connection can negotiate it, and EDGE_ALPN so + // the supervisor's data-plane edges can dial in. + let mut driver = 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(), + iroh_driver::EDGE_ALPN.to_vec(), + ], + }, + ) + .map_err(|error| format!("iroh driver: {error}"))?; let node_hex = swactor_transport::hex_encode(&driver.node_id().0); // Telemetry endpoint: stream identity is the transport node key, so @@ -140,6 +143,76 @@ pub fn run_node_role(supervisor_addr_json: &str, attempt: u64) -> Result<(), Str ); runtime.set_stats_hook(producer.stats_hook_on(actors_channel)); + // Data-plane edge agent: owns this node's (single) inbound edge. The + // supervisor provisions it over the control plane (tagged gossip + // decoded by the actor bridge below); observations are mirrored onto + // the `node.edge` telemetry channel (render-only), and readiness + // acks travel back as gossip — never through telemetry. + let edge_channel = endpoint.register_channel( + edge::NODE_EDGE_CHANNEL, + ChannelContent::JsonRecord { + schema: Some("demo.node.edge.v1".to_owned()), + }, + ); + let driver_slot: Arc>> = + Arc::new(std::sync::OnceLock::new()); + let edge_agent = runtime + .spawn(edge::NodeEdgeAgent::new( + attempt, + logical_node.clone(), + supervisor_addr.clone(), + driver_slot.clone(), + producer.clone(), + edge_channel, + )) + .map_err(|error| format!("spawn edge agent: {error}"))?; + { + use distribution::transport_bridge::{Outbox, RelayMirror, RouteView}; + use swactor_transport::CodecRegistry; + let mut codec = CodecRegistry::new(); + codec.register_decoder::(edge::EDGE_PROVISION_TAG, |bytes| { + let provision: edge::EdgeProvision = serde_json::from_slice(bytes) + .map_err(|e| swactor::Error::from(format!("edge provision decode: {e}")))?; + Ok(edge::NodeEdgeMsg::Provision(provision)) + }); + let mut routes = std::collections::HashMap::new(); + routes.insert(edge::EDGE_PROVISION_TAG.to_owned(), edge_agent); + 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(codec), + routes, + edge_agent, + relay_mirror, + route_view, + outbox, + ); + driver.install_actor_bridge_pump(Duration::from_millis(250)); + } + let driver = Arc::new(driver); + let _ = driver_slot.set(Arc::clone(&driver)); + // The node serves telemetry pulls itself (spawn_pull_server): keep + // TELEMETRY_ALPN connections out of the driver-owned ingress so the + // serve loop below can drain them. + driver.retain_telemetry_connections(); + { + // Edge agent tick: poll the edge runtime on the supervisor's + // session cadence. + let sender = runtime.create_sender(); + let edge_engine = engine.handle(); + edge_engine.clone().spawn(async move { + let mut interval = edge_engine.interval(Duration::from_millis(250)); + loop { + (&mut interval).await; + let _ = sender.send_to(edge_agent, edge::NodeEdgeMsg::Tick); + } + }); + } + // Join, then announce identity + advertised address to the supervisor's // bootstrap actor over the control plane (readiness + telemetry dial). driver.join(&[supervisor_addr.clone()]); diff --git a/xtask/src/provisioning_demo/provider.rs b/xtask/src/demo/provider.rs similarity index 95% rename from xtask/src/provisioning_demo/provider.rs rename to xtask/src/demo/provider.rs index a4580b8..012e5ac 100644 --- a/xtask/src/provisioning_demo/provider.rs +++ b/xtask/src/demo/provider.rs @@ -51,6 +51,9 @@ pub struct NodeRuntime { /// 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, + /// Serde-serialized `iroh::EndpointAddr` from the node's announce — + /// what the supervisor dials for telemetry pulls and demo edges. + pub endpoint_addr: Option, } /// Spawn request from the plugin (blocking thread) to the supervisor actor. @@ -122,6 +125,17 @@ impl NodeManager { .cloned() } + /// All registered runtimes (snapshot for liveness sweeps). + pub fn nodes(&self) -> Vec { + self.inner + .lock() + .expect("node manager") + .nodes + .values() + .cloned() + .collect() + } + pub fn find_by_stream_node(&self, node: &str) -> Option { self.inner .lock() @@ -159,6 +173,12 @@ impl NodeManager { pub fn set_announce(&self, attempt: u64, at_ms: u64) { self.update(attempt, |runtime| runtime.last_announce_ms = Some(at_ms)); } + + pub fn set_endpoint(&self, attempt: u64, endpoint_addr_json: String) { + self.update(attempt, |runtime| { + runtime.endpoint_addr = Some(endpoint_addr_json) + }); + } pub fn remove(&self, attempt: u64) { self.inner .lock() @@ -312,11 +332,13 @@ impl AnnounceActor { } impl ActorInterface for AnnounceActor { - type Incoming = crate::provisioning_demo::node::NodeAnnounce; + type Incoming = crate::demo::node::NodeAnnounce; type Response = (); - fn handle(&mut self, _ctx: &Ctx, announce: crate::provisioning_demo::node::NodeAnnounce) { + fn handle(&mut self, _ctx: &Ctx, announce: crate::demo::node::NodeAnnounce) { self.manager.set_announce(announce.attempt, announce.at_ms); + self.manager + .set_endpoint(announce.attempt, announce.endpoint_addr_json.clone()); let Some(runtime) = self.manager.get(announce.attempt) else { let key = &announce.key_hex; eprintln!( diff --git a/xtask/src/provisioning_demo/view.rs b/xtask/src/demo/view.rs similarity index 83% rename from xtask/src/provisioning_demo/view.rs rename to xtask/src/demo/view.rs index c8dafda..109ea41 100644 --- a/xtask/src/provisioning_demo/view.rs +++ b/xtask/src/demo/view.rs @@ -5,7 +5,7 @@ //! table. No standalone page: `html()` is `None` and the view stays out of //! the navbar while its API remains registered. -use std::collections::VecDeque; +use std::collections::{BTreeMap, VecDeque}; use std::time::{SystemTime, UNIX_EPOCH}; use parking_lot::Mutex; @@ -16,6 +16,8 @@ use telemetry::frame::{Frame, StreamId}; use dashboard::FrameEvent; use dashboard::view::DashboardView; +use crate::demo::edge::NODE_EDGE_CHANNEL; + const EVENTS_CHANNEL: &str = "prov.reconciler.events"; const SNAPSHOT_CHANNEL: &str = "prov.reconciler.snapshot"; const FEED_CAP: usize = 250; @@ -36,12 +38,17 @@ struct ReconcilerSnapshot { converged: bool, age_ms: u64, nodes: Vec, + edges: Vec, + /// Latest node-side edge record per logical node (render-only). + node_edges: Vec, feed: Vec, } struct ViewState { header: Option, nodes: Vec, + edges: Vec, + node_edges: BTreeMap, feed: VecDeque, snapshot_at_ms: u64, } @@ -65,6 +72,8 @@ impl Default for ReconcilerDashboardView { state: Mutex::new(ViewState { header: None, nodes: Vec::new(), + edges: Vec::new(), + node_edges: BTreeMap::new(), feed: VecDeque::new(), snapshot_at_ms: 0, }), @@ -92,7 +101,7 @@ impl DashboardView for ReconcilerDashboardView { } fn channels(&self) -> &'static [&'static str] { - &[EVENTS_CHANNEL, SNAPSHOT_CHANNEL] + &[EVENTS_CHANNEL, SNAPSHOT_CHANNEL, NODE_EDGE_CHANNEL] } fn show_in_nav(&self) -> bool { @@ -111,12 +120,22 @@ impl DashboardView for ReconcilerDashboardView { .and_then(Value::as_array) .cloned() .unwrap_or_default(); + state.edges = payload + .get("edges") + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); state.snapshot_at_ms = payload .get("at_ms") .and_then(Value::as_u64) .unwrap_or_else(|| unix_ms(SystemTime::now())); state.header = Some(payload); } + NODE_EDGE_CHANNEL => { + // Render-only: latest node-side edge record per node. + let node = event.stream.node.clone(); + state.node_edges.insert(node, payload); + } EVENTS_CHANNEL => { let line = FeedLine { at_ms: payload @@ -162,6 +181,8 @@ impl DashboardView for ReconcilerDashboardView { .unwrap_or(false), age_ms: now.saturating_sub(state.snapshot_at_ms), nodes: state.nodes.clone(), + edges: state.edges.clone(), + node_edges: state.node_edges.values().cloned().collect(), feed: state.feed.iter().cloned().collect(), }) .unwrap_or_else(|_| serde_json::json!({})) diff --git a/xtask/src/main.rs b/xtask/src/main.rs index 030c094..c97c9f5 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -266,7 +266,7 @@ 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] [--docker] + demo [--port n] [--nodes n] [--docker] Run the visual E2E provisioning reconciler sanity demo (supervisor + dashboard on localhost, node children join over iroh; --docker launches nodes as scratch @@ -6976,7 +6976,7 @@ fn collect_rs_files(dir: &str, out: &mut Vec) { } } -mod provisioning_demo; +mod demo; fn main() -> ExitCode { let mut args = std::env::args().skip(1); @@ -6986,8 +6986,8 @@ fn main() -> ExitCode { Some("myelin-chat-check") => run_myelin_chat_check(args.collect()), Some("myelin-chat-compare") => run_myelin_chat_compare(args.collect()), Some("myelin-chat") => run_myelin_chat(args.collect()), - Some("provisioning-reconciler-demo") => { - provisioning_demo::run(&args.collect::>()) + Some("demo") => { + demo::run(&args.collect::>()) } Some("help" | "--help" | "-h") | None => { print_usage();