diff --git a/apps/myelin/src/codecs.rs b/apps/myelin/src/codecs.rs index 9182462..4c77eae 100644 --- a/apps/myelin/src/codecs.rs +++ b/apps/myelin/src/codecs.rs @@ -2,12 +2,10 @@ //! //! The runtime's [`CodecRegistry`](swactor_transport::CodecRegistry) needs an //! encoder/decoder entry for each inter-node message type. This is the single -//! aggregator that wires up the node, orchestrator, prompt, and telemetry -//! publisher codecs. +//! aggregator that wires up the node, orchestrator, and prompt codecs. pub(crate) fn register_myelin_actor_codecs(registry: &mut swactor_transport::CodecRegistry) { crate::node_actor::register_codecs(registry); crate::orchestration::actor::register_codecs(registry); - telemetry::register_telemetry_publisher_codec(registry); swactor_job_runner::register_job_codecs(registry); } diff --git a/apps/myelin/src/node/actor.rs b/apps/myelin/src/node/actor.rs index 1676a72..949879a 100644 --- a/apps/myelin/src/node/actor.rs +++ b/apps/myelin/src/node/actor.rs @@ -137,7 +137,6 @@ pub(crate) enum NodeAgentMsg { stage_index: u32, endpoint: EndpointAddr, node_actor: ActorAddress, - telemetry_publisher: ActorAddress, readiness_id: u64, }, RuntimeReadyAck { @@ -450,7 +449,6 @@ impl NodeAgentActor { stage_index, endpoint, node_actor, - telemetry_publisher, readiness_id, } => { self.core.observe(stage::StageEvent::WorkerReady); @@ -462,7 +460,6 @@ impl NodeAgentActor { stage_index, endpoint, node_actor, - telemetry_publisher, readiness_id, }, ); diff --git a/apps/myelin/src/node/prompt_wire.rs b/apps/myelin/src/node/prompt_wire.rs index ffc29f9..ccd51a1 100644 --- a/apps/myelin/src/node/prompt_wire.rs +++ b/apps/myelin/src/node/prompt_wire.rs @@ -26,20 +26,6 @@ pub(crate) enum PromptEvent { }, } -impl PromptEvent { - pub(crate) fn request_id(&self) -> u64 { - match self { - Self::TextDelta { request_id, .. } - | Self::Done { request_id, .. } - | Self::Fault { request_id, .. } => *request_id, - } - } - - pub(crate) fn is_terminal(&self) -> bool { - matches!(self, Self::Done { .. } | Self::Fault { .. }) - } -} - impl NetworkMessage for PromptEvent { fn type_tag() -> &'static str { "myelin::PromptEvent" diff --git a/apps/myelin/src/node/worker_node_runtime.rs b/apps/myelin/src/node/worker_node_runtime.rs index f4c1402..75c73d3 100644 --- a/apps/myelin/src/node/worker_node_runtime.rs +++ b/apps/myelin/src/node/worker_node_runtime.rs @@ -19,8 +19,7 @@ use std::time::{Duration, Instant}; use telemetry::frame::TelemetryEvent; use telemetry::{ ChannelContent, ChannelId, Lifetime, NodeId, Record, StreamDescriptor, StreamId, StreamOrigin, - TELEMETRY_PUBLISHER_NAME, TelemetryEndpoint, TelemetryProducer, TelemetryPublisherActor, - TelemetrySubscribe, TelemetrySubscription, + TelemetryEndpoint, TelemetryProducer, TelemetrySubscription, }; use crate::codecs::register_myelin_actor_codecs; @@ -43,10 +42,7 @@ use distribution::node::DistributedNodeConfig; use distribution::telemetry::{MembershipTransition, SwimProbeEvent}; use distribution::types::{MemberState, NodeId as DistNodeId}; use iroh::EndpointAddr; -use iroh_driver::{ - EDGE_ALPN, IrohDriver, IrohDriverConfig, TELEMETRY_ALPN, TelemetryPublishHandle, - TelemetryQuicHeader, spawn_pull_server, -}; +use iroh_driver::{EDGE_ALPN, IrohDriver, IrohDriverConfig, TELEMETRY_ALPN, spawn_pull_server}; use iroh_driver::{EndpointAddrMask, MVP_IROH_ENDPOINT_ADDR_MASK_ENV, advertised_endpoint}; use parking_lot::Mutex; use serde_json::{Value, json}; @@ -1593,23 +1589,8 @@ fn run() -> Result<(), String> { let node_shutdown = |ds: &mut NodeTelemetry, phase: &str, status: &str, detail: Value| { emit_node_event(ds, &config, NODE_SHUTDOWN_CHANNEL, phase, status, detail) }; - let telemetry_transport = driver.telemetry_publish_handle(); - let telemetry_publisher = match stack - .runtime - .spawn(telemetry.publisher_actor(telemetry_transport)) - { - Ok(actor) => actor, - Err(error) => { - node_boot( - &mut telemetry, - "telemetry_publisher", - "failed", - json!({"error":error.to_string()}), - ); - return Err(format!("spawn telemetry publisher: {error}")); - } - }; - stack.register_local_actor(driver.register_actor(telemetry_publisher, 1)); + // Telemetry leaves this node exclusively through pull subscriptions served + // by `serve_telemetry_pulls` on `TELEMETRY_ALPN`; no publisher actor. let sampler_health_channel = telemetry.channel_by_name(NODE_SAMPLER_CHANNEL); let sampler_health_context = SamplerHealthContext::from_config(&config); spawn_host_gpu_sampler( @@ -1632,12 +1613,6 @@ fn run() -> Result<(), String> { telemetry.channels.arena, Arc::clone(&arena_manager), ); - node_boot( - &mut telemetry, - "telemetry_publisher", - "ready", - json!({"actor":telemetry_publisher,"name":TELEMETRY_PUBLISHER_NAME,"subscription_transport":"iroh"}), - ); let worker_synthetic_id = format!( "myelin-worker-{}-{}-telemetry-preflight", config.logical_node_id, config.stage_index @@ -1657,7 +1632,7 @@ fn run() -> Result<(), String> { "producer_class":"rust-worker-node", "synthetic_id":worker_synthetic_id, "telemetry_endpoint":{ - "role":"worker-node-iroh-publisher", + "role":"worker-node-iroh-pull-server", "transport":"iroh-telemetry", "endpoint_addr_mask":config.endpoint_addr_mask.as_str(), "relay_mode":format!("{:?}", config.relay_mode), @@ -1756,18 +1731,13 @@ fn run() -> Result<(), String> { "ready", json!({"framework":"none","workloads":"external_jobs"}), )?; - let mut pending_runtime_ready = PendingRuntimeReady::new( - &config, - advertised_self_endpoint.clone(), - node_actor, - telemetry_publisher, - ); + let mut pending_runtime_ready = + PendingRuntimeReady::new(&config, advertised_self_endpoint.clone(), node_actor); let ready = json!({ "type":"ready", "role":"node", "endpoint":advertised_self_endpoint.clone(), "node_actor":node_actor, - "telemetry_publisher":telemetry_publisher, "logical_node_id":config.logical_node_id, "stage_index":config.stage_index, }); @@ -1907,19 +1877,14 @@ fn run() -> Result<(), String> { } } let mut edge_runtime = WorkerEdgeRuntime::new(config.logical_node_id); - let mut pending_runtime_ready = PendingRuntimeReady::new( - &config, - advertised_self_endpoint.clone(), - node_actor, - telemetry_publisher, - ); + let mut pending_runtime_ready = + PendingRuntimeReady::new(&config, advertised_self_endpoint.clone(), node_actor); let ready = json!({ "type":"ready", "role":"node", "endpoint": advertised_self_endpoint.clone(), "node_actor": node_actor, - "telemetry_publisher": telemetry_publisher, "logical_node_id": config.logical_node_id, "stage_index": config.stage_index, }); @@ -2262,28 +2227,8 @@ impl NodeTelemetry { archive.drain(&self.by_id); } } - - fn publisher_actor(&self, transport: TelemetryPublishHandle) -> TelemetryPublisherActor { - TelemetryPublisherActor::new( - Arc::clone(&self.endpoint), - move |subscribe: TelemetrySubscribe, subscription: TelemetrySubscription| { - let Ok(header) = TelemetryQuicHeader::from_snapshot( - subscribe.flow_id, - subscribe.token, - subscription.snapshot(), - ) else { - return; - }; - transport.publish_subscription( - subscribe.collector, - header, - subscription, - Duration::from_millis(10), - ); - }, - ) - } } + fn serve_telemetry_pulls( driver: &IrohDriver, engine: &EngineHandle, @@ -2378,7 +2323,6 @@ struct PendingRuntimeReady { stage_index: u32, endpoint: EndpointAddr, node_actor: ActorAddress, - telemetry_publisher: ActorAddress, coordinator: Option, readiness_id: u64, attempts: u32, @@ -2389,19 +2333,13 @@ struct PendingRuntimeReady { } impl PendingRuntimeReady { - fn new( - config: &DeploymentConfig, - endpoint: EndpointAddr, - node_actor: ActorAddress, - telemetry_publisher: ActorAddress, - ) -> Self { + fn new(config: &DeploymentConfig, endpoint: EndpointAddr, node_actor: ActorAddress) -> Self { Self { run_id: config.run_id, node_id: config.logical_node_id, stage_index: config.stage_index, endpoint, node_actor, - telemetry_publisher, coordinator: config .coordinator_endpoint .as_ref() @@ -2463,7 +2401,6 @@ impl PendingRuntimeReady { stage_index: self.stage_index, endpoint: self.endpoint.clone(), node_actor: self.node_actor, - telemetry_publisher: self.telemetry_publisher, readiness_id: self.readiness_id, }, ) diff --git a/apps/myelin/src/orchestration/actor.rs b/apps/myelin/src/orchestration/actor.rs index 209f3c9..5a4923d 100644 --- a/apps/myelin/src/orchestration/actor.rs +++ b/apps/myelin/src/orchestration/actor.rs @@ -33,7 +33,6 @@ pub(crate) enum OrchestratorMsg { stage_index: u32, endpoint: EndpointAddr, node_actor: ActorAddress, - telemetry_publisher: ActorAddress, readiness_id: u64, }, ObserveNodeRuntimeReadyAck { @@ -154,7 +153,6 @@ pub(crate) enum OrchestratorReport { stage_index: u32, endpoint: EndpointAddr, node_actor: ActorAddress, - telemetry_publisher: ActorAddress, readiness_id: u64, }, NodeRuntimeReadyAck { @@ -326,7 +324,6 @@ impl ActorInterface for OrchestratorActor { stage_index, endpoint, node_actor, - telemetry_publisher, readiness_id, } => { if let Some(report_to) = self.report_to { @@ -338,7 +335,6 @@ impl ActorInterface for OrchestratorActor { stage_index, endpoint, node_actor, - telemetry_publisher, readiness_id, }, ); diff --git a/apps/myelin/src/orchestration/app.rs b/apps/myelin/src/orchestration/app.rs index 7b01947..03c2238 100644 --- a/apps/myelin/src/orchestration/app.rs +++ b/apps/myelin/src/orchestration/app.rs @@ -5,6 +5,7 @@ use std::io::{BufRead, BufReader, Write}; use std::os::fd::FromRawFd; use std::path::{Path, PathBuf}; use std::process::Command; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, mpsc}; use std::thread; use std::time::{Duration, Instant, SystemTime}; @@ -391,7 +392,7 @@ where json!({"actor":orchestrator_actor}), ); - let stop_rx = stop_rx.unwrap_or_else(spawn_stop_listener); + let stop_signal = spawn_stop_listener(stop_rx); let provisioner = config.build_provisioner(stack.runtime.clone())?; bootstrap( @@ -437,7 +438,7 @@ where obs_rx: &obs_rx, collector: &collector, orchestrator_reports: &orchestrator_reports, - stop_rx: &stop_rx, + stop_signal: stop_signal.as_ref(), dashboard: dashboard.as_ref(), orch_telemetry: &mut orch_telemetry, orch_stdio_rx: orch_stdio_rx.as_ref(), @@ -1757,7 +1758,6 @@ impl Config { struct RuntimeReady { endpoint: EndpointAddr, node_actor: ActorAddress, - telemetry_publisher: ActorAddress, stage_index: u32, readiness_id: u64, swim_node_id: DistNodeId, @@ -1800,14 +1800,14 @@ struct RuntimeReadyAckLoop<'a> { obs_rx: &'a mpsc::Receiver, collector: &'a FrameCollector, orchestrator_reports: &'a swactor::runtime::Inbox, - stop_rx: &'a mpsc::Receiver<()>, + stop_signal: &'a AtomicBool, dashboard: Option<&'a DashboardSupport>, orch_telemetry: &'a mut OrchTelemetry, orch_stdio_rx: Option<&'a mpsc::Receiver>, run_id: u64, orchestrator_node_id: u64, provider: &'a ProviderKind, - orchestrator_actor: ActorAddress, + engine: EngineHandle, } // synchronous process-control/orchestration sequencing; the engine drives all background work (ENGINE_SPEC.md §2) @@ -1823,7 +1823,7 @@ fn wait_for_runtime_ready_acks( obs_rx, collector, orchestrator_reports, - stop_rx, + stop_signal, dashboard, orch_telemetry, orch_stdio_rx, @@ -1877,7 +1877,7 @@ fn wait_for_runtime_ready_acks( run_id, orchestrator_node_id, ); - if stop_requested(stop_rx) { + if stop_requested(stop_signal) { return Err( "shutdown requested while waiting for runtime-ready acknowledgements".to_owned(), ); @@ -2060,11 +2060,12 @@ fn wait_for_runtime_readies( obs_rx, collector, orchestrator_reports, - stop_rx, + stop_signal, dashboard, orch_telemetry, orch_stdio_rx, run_id, + engine, provider, .. } = ctx; @@ -2100,7 +2101,7 @@ fn wait_for_runtime_readies( run_id, expected_node_ids.first().copied().unwrap_or(0), ); - if stop_requested(stop_rx) { + if stop_requested(stop_signal) { return Err("shutdown requested while waiting for pipeline nodes ready".to_owned()); } while let Ok(observation) = obs_rx.try_recv() { @@ -2121,7 +2122,6 @@ fn wait_for_runtime_readies( stage_index, endpoint, node_actor, - telemetry_publisher, readiness_id, } = report && report_run_id == run_id @@ -2129,12 +2129,24 @@ fn wait_for_runtime_readies( && cluster.current_attempt(node_id) == Some(::provisioning::NodeAttemptId(readiness_id)) { + // Bootstrap-owned telemetry: the first runtime-ready report + // for a node dials its pull server and retains the live + // subscription for the node's lifetime — the same contract + // as the xtask demo's NodeTelemetryCollector. + if !pending.contains_key(&node_id) { + collector.subscribe_node( + &engine, + driver.endpoint(), + endpoint.clone(), + run_id, + node_id, + ); + } pending.insert( node_id, RuntimeReady { endpoint: endpoint.clone(), node_actor, - telemetry_publisher, stage_index, readiness_id, swim_node_id: DistNodeId(*endpoint.id.as_bytes()), @@ -2254,33 +2266,37 @@ impl PluginObservationSink for ChannelObservationSink { } } -fn stop_requested(stop_rx: &mpsc::Receiver<()>) -> bool { - stop_rx.try_recv().is_ok() +fn stop_requested(stop_signal: &AtomicBool) -> bool { + stop_signal.load(Ordering::Acquire) } // top-level OS signal handling is process control, out of scope (ENGINE_SPEC.md §2) #[allow(clippy::disallowed_methods)] -fn spawn_stop_listener() -> mpsc::Receiver<()> { - let (tx, rx) = mpsc::channel(); - #[cfg(target_os = "linux")] - { +fn spawn_stop_listener(external: Option>) -> Arc { + let requested = Arc::new(AtomicBool::new(false)); + let listener_requested = Arc::clone(&requested); + if let Some(external) = external { thread::spawn(move || { - let Ok(mut signals) = signal_hook::iterator::Signals::new([ - signal_hook::consts::signal::SIGINT, - signal_hook::consts::signal::SIGTERM, - ]) else { - return; - }; - if signals.forever().next().is_some() { - let _ = tx.send(()); + if external.recv().is_ok() { + listener_requested.store(true, Ordering::Release); } }); + return requested; } - #[cfg(not(target_os = "linux"))] - { - drop(tx); - } - rx + + #[cfg(target_os = "linux")] + thread::spawn(move || { + let Ok(mut signals) = signal_hook::iterator::Signals::new([ + signal_hook::consts::signal::SIGINT, + signal_hook::consts::signal::SIGTERM, + ]) else { + return; + }; + if signals.forever().next().is_some() { + listener_requested.store(true, Ordering::Release); + } + }); + requested } struct ServeCluster<'a> { @@ -2289,7 +2305,7 @@ struct ServeCluster<'a> { obs_rx: &'a mpsc::Receiver, collector: &'a FrameCollector, orchestrator_reports: &'a swactor::runtime::Inbox, - stop_rx: &'a mpsc::Receiver<()>, + stop_signal: &'a AtomicBool, dashboard: Option<&'a DashboardSupport>, orch_telemetry: &'a mut OrchTelemetry, orch_stdio_rx: Option<&'a mpsc::Receiver>, @@ -2345,14 +2361,14 @@ impl ServeCluster<'_> { obs_rx: self.obs_rx, collector: self.collector, orchestrator_reports: self.orchestrator_reports, - stop_rx: self.stop_rx, + stop_signal: self.stop_signal, dashboard: self.dashboard, orch_telemetry: self.orch_telemetry, orch_stdio_rx: self.orch_stdio_rx, run_id: self.run_id, orchestrator_node_id: self.orchestrator_node_id, provider: self.provider, - orchestrator_actor: self.orchestrator_actor, + engine: self.engine.clone(), } } @@ -2494,13 +2510,6 @@ impl ServeCluster<'_> { .get(&logical_node_id) .cloned() .ok_or_else(|| format!("node {logical_node_id} did not announce runtime ready"))?; - self.collector.subscribe_node( - &self.engine, - self.driver.endpoint(), - ready.endpoint.clone(), - self.run_id, - logical_node_id, - ); let acknowledged = wait_for_runtime_ready_acks( self.ack_context(), &[RuntimeReadyAckTarget { @@ -2541,7 +2550,6 @@ impl ServeCluster<'_> { endpoint: serde_json::to_string(&ready.endpoint) .map_err(|error| format!("serialize node endpoint: {error}"))?, node_actor: ready.node_actor, - telemetry_publisher: ready.telemetry_publisher, swim_node_id: ready.swim_node_id, stage_index: ready.stage_index, readiness_id: ready.readiness_id, @@ -2630,11 +2638,25 @@ impl ServeCluster<'_> { return; } let result = match command { + ControlCommand::Provision { count: 0, .. } => { + Err("provision count must be at least 1".to_owned()) + } + ControlCommand::Provision { count, .. } if count > 8 => { + Err(format!("provision count {count} exceeds maximum 8")) + } ControlCommand::Provision { count, .. } => { (0..count).try_for_each(|_| self.add_node().map(|_| ())) } - ControlCommand::Kill { node, .. } => { - parse_control_node_id(&node).and_then(|node_id| self.kill_node(node_id).map(|_| ())) + ControlCommand::Kill { node, .. } => parse_control_node_id(&node).and_then(|node_id| { + self.kill_node(node_id)?.then_some(()).ok_or_else(|| { + format!("node {node_id} is not tracked or has no provision intent") + }) + }), + ControlCommand::Remove { count: 0, .. } => { + Err("remove count must be at least 1".to_owned()) + } + ControlCommand::Remove { count, .. } if count > 8 => { + Err(format!("remove count {count} exceeds maximum 8")) } ControlCommand::Remove { count, .. } => { let mut ids = self @@ -2669,17 +2691,31 @@ impl ServeCluster<'_> { stage_index, endpoint, node_actor, - telemetry_publisher, readiness_id, } = report && run_id == self.run_id { + // Bootstrap-owned telemetry: a new runtime generation announces a + // fresh pull server, so re-dial and retain the live subscription. + let known_readiness = self + .snapshot + .node(node_id) + .and_then(|node| node.runtime.as_ref()) + .map(|runtime| runtime.readiness_id); + if known_readiness != Some(readiness_id) { + self.collector.subscribe_node( + &self.engine, + self.driver.endpoint(), + endpoint.clone(), + run_id, + node_id, + ); + } if let Some(node) = self.snapshot.node_mut(node_id) { node.status = daemon::NodeStatus::Running; node.runtime = Some(daemon::RuntimeFacts { endpoint: serde_json::to_string(&endpoint).unwrap_or_default(), node_actor, - telemetry_publisher, swim_node_id: DistNodeId(*endpoint.id.as_bytes()), stage_index, readiness_id, @@ -2717,9 +2753,10 @@ impl ServeCluster<'_> { } fn teardown_if_requested(&mut self) -> Result<(), String> { - // Local process children cannot be adopted, so detaching them on an - // interactive shutdown only creates invisible orphan processes. - if self.destroy_on_exit || self.provider.as_str() == "process" { + // Local process and Docker nodes are run-scoped development resources: + // Ctrl+C must not leave invisible processes or containers behind. + // Remote providers remain adoptable unless explicitly destroyed. + if self.destroy_on_exit || matches!(self.provider.as_str(), "process" | "docker") { for (_, mut cluster) in std::mem::take(&mut self.live_clusters) { cluster.stop()?; } @@ -2798,7 +2835,7 @@ fn serve_cluster(mut ctx: ServeCluster<'_>) -> Result<(), String> { while let Ok(command) = ctx.control_rx.try_recv() { ctx.handle_dashboard_command(command); } - if stop_requested(ctx.stop_rx) { + if stop_requested(ctx.stop_signal) { break; } thread::sleep(PUMP_INTERVAL); diff --git a/apps/myelin/src/orchestration/daemon.rs b/apps/myelin/src/orchestration/daemon.rs index 3b2c40f..d94feeb 100644 --- a/apps/myelin/src/orchestration/daemon.rs +++ b/apps/myelin/src/orchestration/daemon.rs @@ -7,7 +7,9 @@ //! intent and facts so a restarted daemon can adopt what still exists and //! never silently re-provisions. -use std::collections::{BTreeMap, BTreeSet}; +#[cfg(test)] +use std::collections::BTreeMap; +use std::collections::BTreeSet; use std::fs; use std::path::{Path, PathBuf}; use std::time::{SystemTime, UNIX_EPOCH}; @@ -39,7 +41,6 @@ pub(crate) enum NodeStatus { pub(crate) struct RuntimeFacts { pub endpoint: String, pub node_actor: ActorAddress, - pub telemetry_publisher: ActorAddress, pub swim_node_id: DistNodeId, pub stage_index: u32, pub readiness_id: u64, diff --git a/apps/myelin/src/tests/job_runner_iroh.rs b/apps/myelin/src/tests/job_runner_iroh.rs index 2db269a..0c90d9a 100644 --- a/apps/myelin/src/tests/job_runner_iroh.rs +++ b/apps/myelin/src/tests/job_runner_iroh.rs @@ -9,8 +9,6 @@ use std::collections::BTreeMap; use std::time::{Duration, Instant}; -use swactor::actor::Message; -use swactor::runtime::Inbox; use swactor_engine::{Engine, TokioBackend, TokioConfig}; use swactor_job_runner::{ Job, JobDone, JobState, NodeJobActor, OrchestratorJobActor, OrchestratorJobMsg, Workspace, @@ -79,10 +77,6 @@ fn job_runs_across_two_nodes_over_real_iroh() { // route to it through the converged directory. let (engine_b, driver_b, stack_b) = build_composition(); let sender_b = stack_b.runtime.create_sender(); - let done_b_echo = stack_b - .runtime - .new_inbox::() - .expect("worker echo inbox"); // Placeholder orchestrator address: the real one is on A; the node only // needs it once the orchestrator submits. We point the node at A's // orchestrator after it exists (address is fixed below), but the node actor @@ -112,17 +106,18 @@ fn job_runs_across_two_nodes_over_real_iroh() { .expect("spawn node job actor on B"); stack_b.register_local_actor(driver_b.register_actor(job_actor, 1)); - // Orchestrator (A) joins the worker (B) over iroh and waits for the directory - // to converge: A's route view must learn job_actor → B. + // Both control directions must be routable before Submit. A learning the + // worker route does not imply B has already learned the orchestrator route; + // submitting at that one-way boundary loses the first node event. driver_a_join(&stack_a, &_driver_a, &driver_b); stack_a.register_local_actor(_driver_a.register_actor(orch, 1)); let converged = wait_until(CONVERGE_DEADLINE, || { - stack_a.route_view.read().unwrap().contains_key(&job_actor) + stack_a.route_owner(job_actor).is_some() && stack_b.route_owner(orch).is_some() }); assert!( converged, - "directory did not converge: A never learned the worker's NodeJobActor" + "directory did not converge bidirectionally for orchestrator and worker actors" ); let job = Job { diff --git a/apps/myelin/src/tests/node_guarantees.rs b/apps/myelin/src/tests/node_guarantees.rs index 4563f46..db5350a 100644 --- a/apps/myelin/src/tests/node_guarantees.rs +++ b/apps/myelin/src/tests/node_guarantees.rs @@ -21,7 +21,6 @@ fn node_agent_runtime_loaded_reports_orchestrator() { .expect("orchestrator inbox"); let orchestrator = *orchestrator_inbox.addr(); let node_actor = ActorAddress::new_random(); - let telemetry_publisher = ActorAddress::new_random(); let endpoint = EndpointAddr::new(SecretKey::from_bytes(&[9; 32]).public()); let actor = runtime .spawn(NodeAgentActor::new(stage::NodeId(11), orchestrator, None)) @@ -36,7 +35,6 @@ fn node_agent_runtime_loaded_reports_orchestrator() { stage_index: 3, endpoint: endpoint.clone(), node_actor, - telemetry_publisher, readiness_id: 99, }, ) @@ -51,7 +49,6 @@ fn node_agent_runtime_loaded_reports_orchestrator() { stage_index: 3, endpoint, node_actor, - telemetry_publisher, readiness_id: 99, }) ); diff --git a/apps/myelin/src/tests/prompt_guarantees.rs b/apps/myelin/src/tests/prompt_guarantees.rs index 92ea423..20b533b 100644 --- a/apps/myelin/src/tests/prompt_guarantees.rs +++ b/apps/myelin/src/tests/prompt_guarantees.rs @@ -2,51 +2,24 @@ use myelin::node::prompt_wire::PromptEvent; -#[test] -fn event_terminal_state_is_explicit() { - assert!( - !PromptEvent::TextDelta { - request_id: 1, - text: "a".to_owned(), - } - .is_terminal() - ); - assert!( - PromptEvent::Done { - request_id: 1, - final_text: "a".to_owned(), - tokens_generated: 1, - elapsed_ms: 2, - } - .is_terminal() - ); - assert!( - PromptEvent::Fault { - request_id: 1, - error: "boom".to_owned(), - } - .is_terminal() - ); +fn is_terminal(event: &PromptEvent) -> bool { + matches!(event, PromptEvent::Done { .. } | PromptEvent::Fault { .. }) } #[test] -fn event_request_ids_are_always_available() { - let events = [ - PromptEvent::TextDelta { - request_id: 4, - text: String::new(), - }, - PromptEvent::Done { - request_id: 5, - final_text: String::new(), - tokens_generated: 0, - elapsed_ms: 0, - }, - PromptEvent::Fault { - request_id: 6, - error: String::new(), - }, - ]; - let ids: Vec = events.iter().map(PromptEvent::request_id).collect(); - assert_eq!(ids, vec![4, 5, 6]); +fn event_terminal_state_is_explicit() { + assert!(!is_terminal(&PromptEvent::TextDelta { + request_id: 1, + text: "a".to_owned(), + })); + assert!(is_terminal(&PromptEvent::Done { + request_id: 1, + final_text: "a".to_owned(), + tokens_generated: 1, + elapsed_ms: 2, + })); + assert!(is_terminal(&PromptEvent::Fault { + request_id: 1, + error: "boom".to_owned(), + })); } diff --git a/crates/dashboard/src/control_plane_page.html b/crates/dashboard/src/control_plane_page.html index f1adeb1..06babf7 100644 --- a/crates/dashboard/src/control_plane_page.html +++ b/crates/dashboard/src/control_plane_page.html @@ -59,7 +59,7 @@ #status { font-family: var(--mono); font-size: 13px; color: var(--amber); } a.node-card { color: inherit; text-decoration: none; display: block; } .node-card.orchestrator { grid-column: 1 / -1; border-color: var(--amber); background: var(--card-hover); } - .node-card .role { display: inline-flex; margin-left: 8px; padding: 2px 6px; border: 1px solid var(--amber); border-radius: var(--r); color: var(--amber); font: 600 10px/1.4 var(--mono); text-transform: uppercase; letter-spacing: .07em; vertical-align: 2px; } + .node-card.orchestrator .role { margin-bottom: 6px; color: var(--muted); font: 600 9px/1.2 var(--mono); text-transform: uppercase; letter-spacing: .12em; } :is(a, button, input, summary):focus-visible, tr[data-addr]:focus-visible { outline: 2px solid var(--cyan); outline-offset: 2px; } .grid { display: grid; grid-template-columns: repeat(auto-fill, minmax(300px, 1fr)); gap: 12px; } .node-card { background: var(--panel); border: 1px solid var(--border); border-radius: var(--r); padding: 14px; cursor: pointer; transition: border-color var(--t), background var(--t); } @@ -262,7 +262,8 @@ function nodeCard(node) { // supervisors; it renders as a pass-filled span so the card markup — and // with it the whole grid — stays byte-identical between polls. return ` -

${esc(node.stream.node)}${role} life ${fmt(node.stream.life)}

+ ${role} +

${esc(node.stream.node)} life ${fmt(node.stream.life)}

${proc}seen ago · ${label}
${summary.poisoned ? `
${fmt(summary.poisoned)} poisoned actor(s)
` : ''} `; diff --git a/crates/dashboard/src/demo_control_page.html b/crates/dashboard/src/demo_control_page.html index c55544d..0e4b6cc 100644 --- a/crates/dashboard/src/demo_control_page.html +++ b/crates/dashboard/src/demo_control_page.html @@ -62,6 +62,7 @@ .seg .seg-v { position: relative; } .seg-sm { font-size: 15px; } .blink { animation: seg-blink 1.1s steps(2, start) infinite; } + .counter[hidden] { display: none; } @keyframes seg-blink { to { visibility: hidden; } } .controls { display: flex; align-items: center; flex-wrap: wrap; gap: 8px; } .controls input { diff --git a/crates/iroh-driver/src/telemetry_transport.rs b/crates/iroh-driver/src/telemetry_transport.rs index 603b87d..21a8832 100644 --- a/crates/iroh-driver/src/telemetry_transport.rs +++ b/crates/iroh-driver/src/telemetry_transport.rs @@ -104,11 +104,11 @@ pub fn spawn_pull_server( }); } -/// 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. +/// Supervisor side: retain a pull subscription to a node on `TELEMETRY_ALPN`. +/// +/// A transport interruption reconnects with bounded backoff. Returning after +/// the first EOF leaves a healthy node permanently stale, which is especially +/// easy to trigger while several freshly-bootstrapped nodes answer at once. pub fn spawn_pull_collector( engine: &EngineHandle, endpoint: Endpoint, @@ -119,35 +119,74 @@ pub fn spawn_pull_collector( fanout: std::sync::Arc, on_header: std::sync::mpsc::Sender, ) { + let engine_handle = engine.clone(); 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); + let mut retry_delay = Duration::from_millis(250); + loop { + match collect_pull_once( + &endpoint, &peer, flow_id, &token, &request, &fanout, &on_header, + ) + .await + { + Ok(()) => return, + Err(error) => { + eprintln!( + "telemetry-pull: {peer_id}: {error}; retrying in {} ms", + retry_delay.as_millis() + ); + } + } + engine_handle.timer(retry_delay).await; + retry_delay = retry_delay + .checked_mul(2) + .unwrap_or(Duration::from_secs(5)) + .min(Duration::from_secs(5)); } }); } + +async fn collect_pull_once( + endpoint: &Endpoint, + peer: &EndpointAddr, + flow_id: [u8; 16], + token: &[u8], + request: &telemetry::SubscriptionRequest, + fanout: &telemetry::DeliveryFanout, + on_header: &std::sync::mpsc::Sender, +) -> Result<(), String> { + let conn = endpoint + .connect(peer.clone(), TELEMETRY_ALPN) + .await + .map_err(|error| format!("connect failed: {error}"))?; + let mut req = conn + .open_uni() + .await + .map_err(|error| format!("open request stream failed: {error}"))?; + write_pull_request(&mut req, flow_id, token, request) + .await + .map_err(|error| format!("write request failed: {error}"))?; + let mut recv = conn + .accept_uni() + .await + .map_err(|error| format!("no answer stream: {error}"))?; + let header = read_header(&mut recv) + .await + .map_err(|error| format!("answer header unreadable: {error}"))?; + if on_header.send(header.clone()).is_err() { + return Ok(()); + } + let stream = header.stream; + loop { + match read_next_event(&mut recv, &stream).await { + Ok(Some(event)) => { + fanout.publish(event); + } + Ok(None) => return Err("answer stream closed".to_owned()), + Err(error) => return Err(format!("read answer stream failed: {error}")), + } + } +} const MAX_RECORD_BYTES: usize = 16 * 1024 * 1024; type BoxError = Box;