From d97334004394a949391d24fecfa5a8370b794529 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Wed, 29 Jul 2026 12:14:52 +0400 Subject: [PATCH] refactor: actorization polish and stability --- crates/mvp-system/src/chat/runtime.rs | 116 ++- crates/mvp-system/src/node/actor.rs | 12 +- .../mvp-system/src/node/data_plane_bridge.rs | 7 +- .../src/node/worker_node_runtime.rs | 547 +++++++--- crates/mvp-system/src/orchestration/actor.rs | 5 + crates/mvp-system/src/orchestration/app.rs | 120 ++- .../provider_adapters/vastai/mod.rs | 961 +++++++++++------- crates/mvp-system/src/staging/gguf_shard.rs | 59 ++ .../src/tests/data_plane_bridge_guarantees.rs | 7 +- .../tests/vastai_provisioning_guarantees.rs | 9 +- xtask/src/main.rs | 13 +- 11 files changed, 1326 insertions(+), 530 deletions(-) diff --git a/crates/mvp-system/src/chat/runtime.rs b/crates/mvp-system/src/chat/runtime.rs index 379e38c..fd8a661 100644 --- a/crates/mvp-system/src/chat/runtime.rs +++ b/crates/mvp-system/src/chat/runtime.rs @@ -30,6 +30,10 @@ use crate::chat::node_image::{ use crate::observability::{benchmark, frame_archive::FrameArchive}; use crate::orchestration::node_provisioning::{ProviderKind, provider_kind}; use crate::orchestration::provider_adapters::vastai::config::ResolvedVastAiConfig; +use crate::orchestration::{ + DEFAULT_PIPELINE_CACHED_MODEL_FILE, DEFAULT_PIPELINE_CACHED_MODEL_ID, + DEFAULT_PIPELINE_CACHED_MODEL_MAX_CONTEXT, DEFAULT_PIPELINE_CACHED_MODEL_REPO, +}; use crate::prompt::rpc::{PromptEvent, SubmitPrompt, write_json_line}; use crate::transport::endpoint_advertisement::EndpointAddrMask; @@ -814,6 +818,16 @@ impl Config { let cached_model = cached_model_source .map(CachedModelConfig::from_source) .transpose()?; + let model = if provider == provider_kind::vastai() { + match &cached_model { + Some(cached_model) => { + vastai_model_config_for_cached_model(toml.model.clone(), cached_model)? + } + None => toml.model.clone(), + } + } else { + toml.model.clone() + }; let datastream_frame_log = if args.dump_logs { Some( args.dump_log_path @@ -847,7 +861,7 @@ impl Config { vastai_yes: args.vastai_yes, pipeline_stages, max_tokens, - model: toml.model, + model, vastai, skip_rebuild: args.skip_rebuild, gpu_run, @@ -1097,6 +1111,40 @@ impl ParsedArgs { } } +fn vastai_model_config_for_cached_model( + mut model: ChatModelConfig, + cached_model: &CachedModelConfig, +) -> Result { + let file_name = cached_model + .host_path + .file_name() + .and_then(|name| name.to_str()) + .ok_or_else(|| { + format!( + "cached model path {} does not have a UTF-8 file name", + cached_model.host_path.display() + ) + })?; + if file_name == DEFAULT_PIPELINE_CACHED_MODEL_FILE { + model.id = Some(DEFAULT_PIPELINE_CACHED_MODEL_ID.to_owned()); + model.gguf_local_path = None; + model.gguf_repo = Some(DEFAULT_PIPELINE_CACHED_MODEL_REPO.to_owned()); + model.gguf_file = Some(DEFAULT_PIPELINE_CACHED_MODEL_FILE.to_owned()); + model.gguf_revision = None; + model.max_context = Some(DEFAULT_PIPELINE_CACHED_MODEL_MAX_CONTEXT); + return Ok(model); + } + if model.gguf_file.as_deref() == Some(file_name) { + model.gguf_local_path = None; + return Ok(model); + } + Err(format!( + "VastAI cached model {} does not match configured remote GGUF {}; use --cached-model= or configure [model].gguf_repo and [model].gguf_file for that cache", + cached_model.host_path.display(), + model.gguf_file.as_deref().unwrap_or("") + )) +} + fn resolve_vastai_config( file: &ChatVastAiConfig, node_image: &str, @@ -3222,6 +3270,72 @@ relay_url = "https://relay.example" }); } + #[test] + fn vastai_cached_pipeline_model_selects_matching_remote_gguf() { + let temp = TempDir::new("vastai-cached-pipeline-model"); + let cached_path = temp.path().join(DEFAULT_PIPELINE_CACHED_MODEL_FILE); + fs::write(&cached_path, b"cached model").expect("write cached model"); + let config_path = write_config( + &temp, + "chat.toml", + r#" +[provider] +kind = "vastai" + +[image] +node = "docker.io/acme/node:latest" + +[model] +id = "qwen2.5-7b-instruct-q4-k-m" +gguf_repo = "bartowski/Qwen2.5-7B-Instruct-GGUF" +gguf_file = "Qwen2.5-7B-Instruct-Q4_K_M.gguf" +max_context = 512 + +[vastai] +relay_url = "https://relay.example" +bootstrap_command = "boot" +"#, + ); + + let cached_arg = format!("--cached-model={}", cached_path.display()); + with_process_state(&[("VAST_API_KEY", Some("secret"))], None, || { + let config_arg = config_path.to_string_lossy().into_owned(); + let config = Config::from_args(strings(&[ + "--config", + config_arg.as_str(), + cached_arg.as_str(), + "--yes", + ])) + .expect("VastAI cached pipeline model resolves"); + let args = config.orchestrator_cli_args("docker.io/acme/node:prepared"); + + assert_eq!( + config + .cached_model + .as_ref() + .and_then(|model| model.host_path.file_name()), + Some(OsStr::new(DEFAULT_PIPELINE_CACHED_MODEL_FILE)) + ); + assert!( + args.windows(2) + .any(|pair| pair == ["--model-id", DEFAULT_PIPELINE_CACHED_MODEL_ID]) + ); + assert!( + args.windows(2) + .any(|pair| pair == ["--gguf-repo", DEFAULT_PIPELINE_CACHED_MODEL_REPO]) + ); + assert!( + args.windows(2) + .any(|pair| pair == ["--gguf-file", DEFAULT_PIPELINE_CACHED_MODEL_FILE]) + ); + let expected_context = DEFAULT_PIPELINE_CACHED_MODEL_MAX_CONTEXT.to_string(); + assert!( + args.windows(2) + .any(|pair| pair == ["--max-context", expected_context.as_str()]) + ); + }); + } + fn panic_prepare_node_image(_: NodeImageRequest) -> Result { panic!("image preparer must not be called when --skip-rebuild is set") } diff --git a/crates/mvp-system/src/node/actor.rs b/crates/mvp-system/src/node/actor.rs index d0a3863..893c3e2 100644 --- a/crates/mvp-system/src/node/actor.rs +++ b/crates/mvp-system/src/node/actor.rs @@ -166,7 +166,9 @@ pub enum NodeAgentMsg { StepCompleted { step_id: u64, }, - WorkerCrashed, + WorkerCrashed { + reason: Option, + }, StopRun { run_id: u64, }, @@ -323,6 +325,7 @@ pub struct NodeAgentActor { outbound_edge: Option, command_cursor: usize, event_cursor: usize, + last_worker_crash: Option, } impl NodeAgentActor { @@ -339,6 +342,7 @@ impl NodeAgentActor { outbound_edge: None, command_cursor: 0, event_cursor: 0, + last_worker_crash: None, } } @@ -448,7 +452,10 @@ impl NodeAgentActor { step_id: stage::StepId(step_id), }) } - NodeAgentMsg::WorkerCrashed => self.core.observe(stage::StageEvent::WorkerCrashed), + NodeAgentMsg::WorkerCrashed { reason } => { + self.last_worker_crash = reason; + self.core.observe(stage::StageEvent::WorkerCrashed) + } NodeAgentMsg::StopRun { run_id } => self.core.observe(stage::StageEvent::StopRun { run_id: stage::RunId(run_id), }), @@ -584,6 +591,7 @@ impl NodeAgentActor { OrchestratorMsg::ObserveStageFault { run_id: run_id.0, stage_index: *stage_index, + reason: self.last_worker_crash.clone(), }, ); } diff --git a/crates/mvp-system/src/node/data_plane_bridge.rs b/crates/mvp-system/src/node/data_plane_bridge.rs index bf9d9c4..4164ecd 100644 --- a/crates/mvp-system/src/node/data_plane_bridge.rs +++ b/crates/mvp-system/src/node/data_plane_bridge.rs @@ -55,7 +55,12 @@ impl ActorInterface for MvpDataPlaneReportSinkActor { dp::DataPlaneReportMsg::ObjectProduced { .. } => {} dp::DataPlaneReportMsg::EdgeFaulted { .. } | dp::DataPlaneReportMsg::WorkerDataPlaneFaulted { .. } => { - let _ = ctx.send(self.node_agent, NodeAgentMsg::WorkerCrashed); + let _ = ctx.send( + self.node_agent, + NodeAgentMsg::WorkerCrashed { + reason: Some("data plane faulted".to_owned()), + }, + ); } dp::DataPlaneReportMsg::EdgeStopped { .. } => {} dp::DataPlaneReportMsg::LocalEdgesStopped { run_id } => { diff --git a/crates/mvp-system/src/node/worker_node_runtime.rs b/crates/mvp-system/src/node/worker_node_runtime.rs index ec37e82..9e31126 100644 --- a/crates/mvp-system/src/node/worker_node_runtime.rs +++ b/crates/mvp-system/src/node/worker_node_runtime.rs @@ -32,7 +32,9 @@ use crate::orchestration::provider_adapters::relay::relay_runtime_config_from_en use crate::orchestration::run_plan::{GgufSource, TokenizerSource}; use crate::prompt::rpc::{PromptEvent, TokenizerEvent}; use crate::staging::control as stage; -use crate::staging::gguf_shard::{StageShardPlan, materialize_stage_shard_http}; +use crate::staging::gguf_shard::{ + StageShardPlan, materialize_stage_shard_http, validate_stage_shard_cache, +}; use crate::transport::codec_registry::register_mvp_actor_codecs; use crate::transport::driver_pumps as driver_model; use crate::transport::endpoint_advertisement::{ @@ -52,7 +54,8 @@ use iroh_driver::{ }; use parking_lot::Mutex; use serde_json::{Value, json}; -use swactor::actor::ActorAddress; +use swactor::actor::{ActorAddress, ActorInterface}; +use swactor::runtime::{Ctx, ExternalSender}; use tokio::io::{AsyncBufReadExt, AsyncWriteExt}; const DEFAULT_WORKER_SCRIPT: &str = "/usr/local/share/mvp/tinygrad_worker.py"; @@ -1530,7 +1533,12 @@ impl WorkerEdgeRuntime { edge::EdgeLifecycleEvent::EdgeFaulted { edge_id, reason } => { stack .runtime - .send_to(node_actor, NodeAgentMsg::WorkerCrashed) + .send_to( + node_actor, + NodeAgentMsg::WorkerCrashed { + reason: Some(format!("edge {} faulted: {reason:?}", edge_id.0)), + }, + ) .map_err(|e| format!("mark worker crashed after edge fault: {e}"))?; return Err(format!("edge {} faulted: {reason:?}", edge_id.0)); } @@ -2279,9 +2287,12 @@ fn run() -> Result<(), String> { "failed", json!({"exit_status":status.to_string()}), ); - let _ = stack - .runtime - .send_to(node_actor, NodeAgentMsg::WorkerCrashed); + let _ = stack.runtime.send_to( + node_actor, + NodeAgentMsg::WorkerCrashed { + reason: Some(format!("tinygrad helper exited with {status}")), + }, + ); pump_network(&mut driver, &stack); return Err(format!("tinygrad helper exited with {status}")); } @@ -3101,15 +3112,290 @@ fn handle_decode_tokens_request( .map_err(|e| format!("send tokenizer decode response: {e}")) } -#[derive(Clone, Copy)] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] enum StageShardProcessStream { Stdout, Stderr, } -struct StageShardProcessLine { - stream: StageShardProcessStream, - line: String, +#[derive(Clone)] +enum StageShardFetchMsg { + Start, + PollChild, + ProcessLine { + stream: StageShardProcessStream, + line: String, + }, + ReaderError { + stream: StageShardProcessStream, + error: String, + }, + ReaderClosed { + stream: StageShardProcessStream, + }, +} + +#[derive(Clone, Debug)] +enum StageShardFetchReport { + Progress(Value), + Done(PathBuf), + Failed(String), +} + +struct StageShardFetchActor { + request_json: Vec, + output_path: PathBuf, + report_to: ActorAddress, + sender: ExternalSender, + child: Option, + stdout_reader: Option>, + stderr_reader: Option>, + stdout_closed: bool, + stderr_closed: bool, + ready_path: Option, + exit_status: Option, + finished: bool, +} + +impl StageShardFetchActor { + fn new( + request_json: Vec, + output_path: PathBuf, + report_to: ActorAddress, + sender: ExternalSender, + ) -> Self { + Self { + request_json, + output_path, + report_to, + sender, + child: None, + stdout_reader: None, + stderr_reader: None, + stdout_closed: true, + stderr_closed: true, + ready_path: None, + exit_status: None, + finished: false, + } + } + + fn start_fetch(&mut self, ctx: &Ctx) { + if self.finished { + return; + } + let exe = match std::env::current_exe() { + Ok(exe) => exe, + Err(error) => { + self.fail(ctx, format!("locate worker node executable: {error}")); + return; + } + }; + let mut child = match Command::new(exe) + .arg("stage-shard-fetcher") + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::piped()) + .spawn() + { + Ok(child) => child, + Err(error) => { + self.fail(ctx, format!("spawn stage shard fetcher: {error}")); + return; + } + }; + match child.stdin.take() { + Some(mut stdin) => { + if let Err(error) = stdin.write_all(&self.request_json) { + let mut child = Some(child); + stop_stage_shard_child(&mut child); + self.fail(ctx, format!("write stage shard fetch request: {error}")); + return; + } + } + None => { + let mut child = Some(child); + stop_stage_shard_child(&mut child); + self.fail(ctx, "stage shard fetcher stdin missing".to_owned()); + return; + } + } + + self.stdout_closed = false; + self.stderr_closed = false; + self.ready_path = None; + self.exit_status = None; + if let Some(stdout) = child.stdout.take() { + self.stdout_reader = Some(spawn_stage_shard_reader( + StageShardProcessStream::Stdout, + stdout, + self.sender.clone(), + ctx.self_addr(), + )); + } else { + self.stdout_closed = true; + } + if let Some(stderr) = child.stderr.take() { + self.stderr_reader = Some(spawn_stage_shard_reader( + StageShardProcessStream::Stderr, + stderr, + self.sender.clone(), + ctx.self_addr(), + )); + } else { + self.stderr_closed = true; + } + self.child = Some(child); + schedule_stage_shard_message( + self.sender.clone(), + ctx.self_addr(), + StageShardFetchMsg::PollChild, + PUMP_INTERVAL, + ); + } + + fn poll_child(&mut self, ctx: &Ctx) { + if self.finished { + return; + } + let Some(child) = self.child.as_mut() else { + return; + }; + match child.try_wait() { + Ok(Some(status)) => { + self.child = None; + self.exit_status = Some(status); + self.maybe_finish(ctx); + } + Ok(None) => schedule_stage_shard_message( + self.sender.clone(), + ctx.self_addr(), + StageShardFetchMsg::PollChild, + PUMP_INTERVAL, + ), + Err(error) => { + self.child = None; + self.fail(ctx, format!("poll stage shard fetcher: {error}")); + } + } + } + + fn handle_line(&mut self, ctx: &Ctx, stream: StageShardProcessStream, line: String) { + if line.is_empty() || self.finished { + return; + } + let event = match stream { + StageShardProcessStream::Stdout => match serde_json::from_str::(&line) { + Ok(value) => value, + Err(error) => { + json!({"type":"StageShardFetchOutputParseFailed","line":line,"error":error.to_string()}) + } + }, + StageShardProcessStream::Stderr => json!({"type":"StageShardFetchStderr","line":line}), + }; + if event.get("type").and_then(Value::as_str) == Some("StageShardReady") { + self.ready_path = event + .get("path") + .and_then(Value::as_str) + .map(PathBuf::from) + .or_else(|| Some(self.output_path.clone())); + } + let _ = ctx.send(self.report_to, StageShardFetchReport::Progress(event)); + } + + fn handle_reader_error(&mut self, ctx: &Ctx, stream: StageShardProcessStream, error: String) { + self.handle_line(ctx, stream, format!("reader error: {error}")); + } + + fn handle_reader_closed(&mut self, ctx: &Ctx, stream: StageShardProcessStream) { + match stream { + StageShardProcessStream::Stdout => self.stdout_closed = true, + StageShardProcessStream::Stderr => self.stderr_closed = true, + } + self.maybe_finish(ctx); + } + + fn maybe_finish(&mut self, ctx: &Ctx) { + if self.finished || self.exit_status.is_none() || !self.stdout_closed || !self.stderr_closed + { + return; + } + self.join_readers(); + let status = self.exit_status.take().expect("exit status checked"); + if status.success() { + let path = self + .ready_path + .clone() + .unwrap_or_else(|| self.output_path.clone()); + if path.is_file() { + self.finished = true; + let _ = ctx.send(self.report_to, StageShardFetchReport::Done(path)); + ctx.stop_self(); + return; + } + self.fail( + ctx, + format!( + "stage shard fetcher exited successfully but {} is missing", + path.display() + ), + ); + return; + } + self.fail(ctx, format!("stage shard fetcher exited with {status}")); + } + + fn fail(&mut self, ctx: &Ctx, error: String) { + if self.finished { + return; + } + self.finished = true; + stop_stage_shard_child(&mut self.child); + self.join_readers(); + let _ = ctx.send(self.report_to, StageShardFetchReport::Failed(error)); + ctx.stop_self(); + } + + fn join_readers(&mut self) { + if let Some(reader) = self.stdout_reader.take() { + let _ = reader.join(); + } + if let Some(reader) = self.stderr_reader.take() { + let _ = reader.join(); + } + self.stdout_closed = true; + self.stderr_closed = true; + } +} + +impl ActorInterface for StageShardFetchActor { + type Incoming = StageShardFetchMsg; + type Response = (); + + fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming) { + match msg { + StageShardFetchMsg::Start => self.start_fetch(ctx), + StageShardFetchMsg::PollChild => self.poll_child(ctx), + StageShardFetchMsg::ProcessLine { stream, line } => self.handle_line(ctx, stream, line), + StageShardFetchMsg::ReaderError { stream, error } => { + self.handle_reader_error(ctx, stream, error) + } + StageShardFetchMsg::ReaderClosed { stream } => self.handle_reader_closed(ctx, stream), + } + } + + fn on_stop(&mut self, _ctx: &Ctx) { + stop_stage_shard_child(&mut self.child); + self.join_readers(); + } +} + +fn stop_stage_shard_child(child: &mut Option) { + let Some(mut child) = child.take() else { + return; + }; + let _ = child.kill(); + let _ = child.wait(); } fn stage_shard_cache_path(plan: &StageShardPlan) -> PathBuf { @@ -3124,8 +3410,9 @@ fn stage_shard_cache_path(plan: &StageShardPlan) -> PathBuf { fn spawn_stage_shard_reader( stream: StageShardProcessStream, reader: R, - tx: mpsc::Sender, -) { + sender: ExternalSender, + actor: ActorAddress, +) -> thread::JoinHandle<()> { thread::spawn(move || { let mut reader = BufReader::new(reader); let mut line = String::new(); @@ -3134,20 +3421,39 @@ fn spawn_stage_shard_reader( match reader.read_line(&mut line) { Ok(0) => break, Ok(_) => { - let _ = tx.send(StageShardProcessLine { - stream, - line: line.trim_end_matches(['\r', '\n']).to_owned(), - }); + let _ = sender.send_to( + actor, + StageShardFetchMsg::ProcessLine { + stream, + line: line.trim_end_matches(['\r', '\n']).to_owned(), + }, + ); } Err(error) => { - let _ = tx.send(StageShardProcessLine { - stream, - line: format!("reader error: {error}"), - }); + let _ = sender.send_to( + actor, + StageShardFetchMsg::ReaderError { + stream, + error: error.to_string(), + }, + ); break; } } } + let _ = sender.send_to(actor, StageShardFetchMsg::ReaderClosed { stream }); + }) +} + +fn schedule_stage_shard_message( + sender: ExternalSender, + actor: ActorAddress, + msg: StageShardFetchMsg, + delay: Duration, +) { + thread::spawn(move || { + thread::sleep(delay); + let _ = sender.send_to(actor, msg); }); } @@ -3174,14 +3480,34 @@ fn materialize_stage_shard_with_process( ) -> Result { let output_path = stage_shard_cache_path(plan); if output_path.is_file() { - let event = json!({ - "type":"StageShardCacheReady", - "stage_index":plan.stage_index, - "path":output_path, - "cache_hit":true, - }); - publish_stage_shard_fetch_event(datastream, config, &event)?; - return Ok(output_path); + match validate_stage_shard_cache(&output_path, plan) { + Ok(()) => { + let event = json!({ + "type":"StageShardCacheReady", + "stage_index":plan.stage_index, + "path":output_path, + "cache_hit":true, + }); + publish_stage_shard_fetch_event(datastream, config, &event)?; + return Ok(output_path); + } + Err(error) => { + let event = json!({ + "type":"StageShardCacheInvalid", + "stage_index":plan.stage_index, + "path":output_path, + "cache_hit":false, + "error":error, + }); + publish_stage_shard_fetch_event(datastream, config, &event)?; + std::fs::remove_file(&output_path).map_err(|remove_error| { + format!( + "remove invalid stage shard cache {}: {remove_error}", + output_path.display() + ) + })?; + } + } } let request = StageShardFetchRequest { @@ -3190,95 +3516,39 @@ fn materialize_stage_shard_with_process( }; let request_json = serde_json::to_vec(&request) .map_err(|e| format!("serialize stage shard fetch request: {e}"))?; - let exe = std::env::current_exe().map_err(|e| format!("locate worker node executable: {e}"))?; - let mut child = Command::new(exe) - .arg("stage-shard-fetcher") - .stdin(Stdio::piped()) - .stdout(Stdio::piped()) - .stderr(Stdio::piped()) - .spawn() - .map_err(|e| format!("spawn stage shard fetcher: {e}"))?; - if let Some(mut stdin) = child.stdin.take() { - stdin - .write_all(&request_json) - .map_err(|e| format!("write stage shard fetch request: {e}"))?; - } - let (tx, rx) = mpsc::channel::(); - if let Some(stdout) = child.stdout.take() { - spawn_stage_shard_reader(StageShardProcessStream::Stdout, stdout, tx.clone()); - } - if let Some(stderr) = child.stderr.take() { - spawn_stage_shard_reader(StageShardProcessStream::Stderr, stderr, tx); - } - - let mut ready_path = None; + let reports = stack + .runtime + .new_inbox::() + .map_err(|e| format!("stage shard fetch report inbox: {e}"))?; + let actor = stack + .runtime + .spawn(StageShardFetchActor::new( + request_json, + output_path, + *reports.addr(), + stack.runtime.create_sender(), + )) + .map_err(|e| format!("spawn stage shard fetch actor: {e}"))?; + stack + .runtime + .send_to(actor, StageShardFetchMsg::Start) + .map_err(|e| format!("start stage shard fetch actor: {e}"))?; loop { - while let Ok(line) = rx.try_recv() { - if line.line.is_empty() { - continue; - } - let event = match line.stream { - StageShardProcessStream::Stdout => { - match serde_json::from_str::(&line.line) { - Ok(value) => value, - Err(error) => { - json!({"type":"StageShardFetchOutputParseFailed","line":line.line,"error":error.to_string()}) - } - } - } - StageShardProcessStream::Stderr => { - json!({"type":"StageShardFetchStderr","line":line.line}) - } - }; - if event.get("type").and_then(Value::as_str) == Some("StageShardReady") { - ready_path = event - .get("path") - .and_then(Value::as_str) - .map(PathBuf::from) - .or_else(|| Some(output_path.clone())); - } - publish_stage_shard_fetch_event(datastream, config, &event)?; - } - if let Some(status) = child - .try_wait() - .map_err(|e| format!("poll stage shard fetcher: {e}"))? - { - while let Ok(line) = rx.try_recv() { - if line.line.is_empty() { - continue; - } - let event = match line.stream { - StageShardProcessStream::Stdout => serde_json::from_str::(&line.line) - .unwrap_or_else(|error| { - json!({"type":"StageShardFetchOutputParseFailed","line":line.line,"error":error.to_string()}) - }), - StageShardProcessStream::Stderr => { - json!({"type":"StageShardFetchStderr","line":line.line}) - } - }; - if event.get("type").and_then(Value::as_str) == Some("StageShardReady") { - ready_path = event - .get("path") - .and_then(Value::as_str) - .map(PathBuf::from) - .or_else(|| Some(output_path.clone())); - } - publish_stage_shard_fetch_event(datastream, config, &event)?; - } - if status.success() { - let path = ready_path.unwrap_or_else(|| output_path.clone()); - if path.is_file() { - return Ok(path); - } - return Err(format!( - "stage shard fetcher exited successfully but {} is missing", - path.display() - )); - } - return Err(format!("stage shard fetcher exited with {status}")); - } pump_network(driver, stack); + while let Some(report) = reports.try_recv() { + match report { + StageShardFetchReport::Progress(event) => { + if let Err(error) = publish_stage_shard_fetch_event(datastream, config, &event) + { + let _ = stack.runtime.stop_actor(actor); + return Err(error); + } + } + StageShardFetchReport::Done(path) => return Ok(path), + StageShardFetchReport::Failed(error) => return Err(error), + } + } datastream.tick(); thread::sleep(PUMP_INTERVAL); } @@ -3337,9 +3607,12 @@ fn handle_stage_command( "failed", json!({"error":error}), ); - let _ = stack - .runtime - .send_to(node_actor, NodeAgentMsg::WorkerCrashed); + let _ = stack.runtime.send_to( + node_actor, + NodeAgentMsg::WorkerCrashed { + reason: Some(error.clone()), + }, + ); pump(); return Err(error); } @@ -3367,17 +3640,36 @@ fn handle_stage_command( }; let (resolved_gguf_source, using_stage_shard) = if let Some(stage_plan) = stage_shard_plan { - let local_path = materialize_stage_shard_with_process( + match materialize_stage_shard_with_process( &stage_plan, config, datastream, driver, stack, - )?; - ( - GgufSource::LocalPath(local_path.to_string_lossy().into_owned()), - true, - ) + ) { + Ok(local_path) => ( + GgufSource::LocalPath(local_path.to_string_lossy().into_owned()), + true, + ), + Err(error) => { + emit_node_event( + datastream, + config, + NODE_STAGE_CHANNEL, + "load_weights", + "failed", + json!({"error":error,"stage_shard":true}), + ); + let _ = stack.runtime.send_to( + node_actor, + NodeAgentMsg::WorkerCrashed { + reason: Some(error.clone()), + }, + ); + pump_network(driver, stack); + return Err(error); + } + } } else { (gguf_source, false) }; @@ -3417,9 +3709,12 @@ fn handle_stage_command( "failed", json!({"error":error}), ); - let _ = stack - .runtime - .send_to(node_actor, NodeAgentMsg::WorkerCrashed); + let _ = stack.runtime.send_to( + node_actor, + NodeAgentMsg::WorkerCrashed { + reason: Some(error.clone()), + }, + ); pump(); return Err(error); } diff --git a/crates/mvp-system/src/orchestration/actor.rs b/crates/mvp-system/src/orchestration/actor.rs index c57002d..65d18f0 100644 --- a/crates/mvp-system/src/orchestration/actor.rs +++ b/crates/mvp-system/src/orchestration/actor.rs @@ -57,6 +57,7 @@ pub enum OrchestratorMsg { ObserveStageFault { run_id: u64, stage_index: u32, + reason: Option, }, ObserveEndpointFault { run_id: u64, @@ -168,6 +169,7 @@ pub enum OrchestratorReport { StageFault { run_id: u64, stage_index: u32, + reason: Option, }, Snapshot { commands: Vec, @@ -247,6 +249,7 @@ impl OrchestratorActor { OrchestratorMsg::ObserveStageFault { run_id, stage_index, + reason: _, } => self.core.observe(core::RunEvent::StageFault { run_id: core::RunId(run_id), stage_index, @@ -396,9 +399,11 @@ impl ActorInterface for OrchestratorActor { OrchestratorMsg::ObserveStageFault { run_id, stage_index, + reason, } => Some(OrchestratorReport::StageFault { run_id, stage_index, + reason, }), _ => None, }; diff --git a/crates/mvp-system/src/orchestration/app.rs b/crates/mvp-system/src/orchestration/app.rs index 563795f..9d9ed38 100644 --- a/crates/mvp-system/src/orchestration/app.rs +++ b/crates/mvp-system/src/orchestration/app.rs @@ -77,7 +77,11 @@ const MVP_RUNTIME_CONFIG_ENV: &str = "MVP_RUNTIME_CONFIG"; const CACHED_MODEL_HOST_ENV: &str = "MVP_CACHED_MODEL_HOST_PATH"; const MVP_WORKER_BIN_ENV: &str = "MVP_WORKER_BIN"; const CACHED_MODEL_CONTAINER_DIR: &str = "/models/cached"; -const DEFAULT_PIPELINE_CACHED_MODEL_FILE: &str = "SmolLM2-135M-Instruct.Q4_0.gguf"; +pub(crate) const DEFAULT_PIPELINE_CACHED_MODEL_FILE: &str = "SmolLM2-135M-Instruct.Q4_0.gguf"; +pub(crate) const DEFAULT_PIPELINE_CACHED_MODEL_REPO: &str = + "QuantFactory/SmolLM2-135M-Instruct-GGUF"; +pub(crate) const DEFAULT_PIPELINE_CACHED_MODEL_ID: &str = "smollm2-135m-instruct-q4"; +pub(crate) const DEFAULT_PIPELINE_CACHED_MODEL_MAX_CONTEXT: u32 = 256; const DEFAULT_PIPELINE_MODEL_CACHE_DIR: &str = ".model-cache"; const DEFAULT_RPC_BIND: &str = "127.0.0.1:19777"; const DEFAULT_HF_REPO: &str = "bartowski/Llama-3.2-1B-Instruct-GGUF"; @@ -2365,6 +2369,21 @@ impl ProvisionedClusterGuard { None => Ok(()), } } + + fn complete_bootstrap(&mut self) -> Result<(), String> { + let mut first_error = None; + for handle in &self.handles { + if let Err(error) = self.provisioner.complete_bootstrap(handle) + && first_error.is_none() + { + first_error = Some(error); + } + } + match first_error { + Some(error) => Err(error), + None => Ok(()), + } + } } impl Drop for ProvisionedClusterGuard { @@ -2591,7 +2610,7 @@ fn start_and_provision_workers( } pending_specs = start_outcome.failed_specs; } - let provisioned_nodes = ProvisionedClusterGuard::new(provisioner, handles); + let mut provisioned_nodes = ProvisionedClusterGuard::new(provisioner, handles); drain_orch_stdio_capture( orch_stdio_rx, orch_datastream, @@ -2702,6 +2721,9 @@ fn start_and_provision_workers( &ack_targets, &pipeline_coordinator, )?; + provisioned_nodes + .complete_bootstrap() + .map_err(|e| format!("complete provider bootstrap after runtime-ready: {e}"))?; orch_datastream.emit_bootstrap( dashboard, @@ -3347,10 +3369,15 @@ fn wait_for_weights_loaded_count( OrchestratorReport::StageFault { run_id: report_run_id, stage_index, + reason, } if report_run_id == run_id && expected_stages.contains(&stage_index) => { - return Err(format!( - "stage {stage_index} faulted while loading pipeline weights" - )); + let mut error = + format!("stage {stage_index} faulted while loading pipeline weights"); + if let Some(reason) = reason { + error.push_str(": "); + error.push_str(&reason); + } + return Err(error); } _ => {} } @@ -4586,10 +4613,15 @@ fn wait_for_weights_loaded( OrchestratorReport::StageFault { run_id: report_run_id, stage_index: report_stage_index, + reason, } if report_run_id == run_id && report_stage_index == stage_index => { - return Err(format!( - "stage {report_stage_index} faulted while loading weights" - )); + let mut error = + format!("stage {report_stage_index} faulted while loading weights"); + if let Some(reason) = reason { + error.push_str(": "); + error.push_str(&reason); + } + return Err(error); } _ => {} } @@ -8958,9 +8990,16 @@ bootstrap_command = "/run" } } + #[derive(Default)] + struct FakeProvisionState { + stopped: Vec, + completed: Vec, + } + #[derive(Default)] struct FakeProvisionPlugin { stopped: Vec, + shared: Option>>, } impl ProvisionPlugin for FakeProvisionPlugin { @@ -8974,8 +9013,15 @@ bootstrap_command = "/run" fn complete_bootstrap( &mut self, - _handle: &crate::orchestration::provisioning::PluginNodeHandle, + handle: &crate::orchestration::provisioning::PluginNodeHandle, ) -> Result<(), String> { + if let Some(shared) = &self.shared { + shared + .lock() + .expect("fake provision state") + .completed + .push(handle.id); + } Ok(()) } @@ -8984,6 +9030,13 @@ bootstrap_command = "/run" handle: &crate::orchestration::provisioning::PluginNodeHandle, ) -> Result<(), String> { self.stopped.push(handle.id); + if let Some(shared) = &self.shared { + shared + .lock() + .expect("fake provision state") + .stopped + .push(handle.id); + } Ok(()) } } @@ -9032,6 +9085,55 @@ bootstrap_command = "/run" assert_eq!(outcome.results.len(), 2); } + #[test] + fn provisioned_cluster_guard_completes_bootstrap_without_dropping_stop_handles() { + let state = Arc::new(std::sync::Mutex::new(FakeProvisionState::default())); + let plugin = FakeProvisionPlugin { + shared: Some(Arc::clone(&state)), + ..FakeProvisionPlugin::default() + }; + let mut guard = ProvisionedClusterGuard::new( + Box::new(plugin), + vec![ + crate::orchestration::provisioning::PluginNodeHandle { + id: 7, + provider_process_id: None, + }, + crate::orchestration::provisioning::PluginNodeHandle { + id: 8, + provider_process_id: None, + }, + ], + ); + + guard + .complete_bootstrap() + .expect("runtime-ready bootstrap completion succeeds"); + assert_eq!( + state + .lock() + .expect("fake provision state") + .completed + .clone(), + vec![7, 8] + ); + assert!( + state + .lock() + .expect("fake provision state") + .stopped + .is_empty() + ); + + guard + .stop() + .expect("node stop still succeeds after completion"); + assert_eq!( + state.lock().expect("fake provision state").stopped.clone(), + vec![8, 7] + ); + } + #[test] fn provisioned_node_guard_stops_node_on_drop() { let mut plugin = FakeProvisionPlugin::default(); diff --git a/crates/mvp-system/src/orchestration/provider_adapters/vastai/mod.rs b/crates/mvp-system/src/orchestration/provider_adapters/vastai/mod.rs index 2c5b2f0..8bb3974 100644 --- a/crates/mvp-system/src/orchestration/provider_adapters/vastai/mod.rs +++ b/crates/mvp-system/src/orchestration/provider_adapters/vastai/mod.rs @@ -5,18 +5,14 @@ use std::collections::{BTreeMap, BTreeSet, HashSet}; use std::io::{BufRead, BufReader, Read}; use std::path::{Path, PathBuf}; use std::process::{Child, Command, Stdio}; -use std::sync::{ - Arc, - atomic::{AtomicBool, Ordering}, - mpsc, -}; -use std::thread::JoinHandle; +use std::sync::{Arc, mpsc}; +use std::thread::{self, JoinHandle}; use std::time::{Duration, Instant}; use datastream::DatastreamProducer; use serde::{Deserialize, Serialize}; use swactor::actor::{ActorAddress, ActorInterface}; -use swactor::runtime::{Ctx, Runtime}; +use swactor::runtime::{Ctx, ExternalSender, Runtime, RuntimeConfig, RuntimeHandle}; use swactor_vastai::{ CreateInstanceRequest, LifecyclePolicy, Offer, ProvisionRequest, ProvisionedInstance, SelectionPolicy, classify_vastai_error, create_instance, @@ -66,23 +62,27 @@ pub struct VastAiSshEndpoint { } pub struct VastAiProviderMonitor { - stopping: Arc, - join: Option>, + runtime: Option, + actor: ActorAddress, } impl VastAiProviderMonitor { - fn new(stopping: Arc, join: JoinHandle<()>) -> Self { + fn new(runtime: RuntimeHandle, actor: ActorAddress) -> Self { Self { - stopping, - join: Some(join), + runtime: Some(runtime), + actor, } } fn stop(&mut self) { - self.stopping.store(true, Ordering::SeqCst); - if let Some(join) = self.join.take() { - let _ = join.join(); - } + let Some(runtime) = self.runtime.take() else { + return; + }; + let _ = runtime + .runtime + .send_to(self.actor, VastAiProviderMonitorMsg::Stop); + runtime.shutdown(); + runtime.join(); } } @@ -207,140 +207,210 @@ impl ToolsVastAiLeaseClient { dph_total: offer.dph_total, }) } +} - fn monitor_provider_status( - &mut self, +#[derive(Clone)] +enum VastAiProviderMonitorMsg { + Poll, + Stop, +} + +struct VastAiProviderMonitorActor { + client: ToolsVastAiLeaseClient, + contract_id: u64, + label: String, + lifecycle: LifecyclePolicy, + spec: NodeProvisionSpec, + sink: PluginSink, + sender: ExternalSender, + last_state: Option, + state_since: Instant, + poll: u64, + stopped: bool, +} + +impl VastAiProviderMonitorActor { + fn new( + client: ToolsVastAiLeaseClient, contract_id: u64, label: String, lifecycle: LifecyclePolicy, spec: NodeProvisionSpec, sink: PluginSink, - stopping: Arc, - ) { - let mut last_state: Option = None; - let mut state_since = Instant::now(); - let mut poll = 0_u64; - while !stopping.load(Ordering::SeqCst) { - poll = poll.saturating_add(1); - let status = match self - .runtime - .block_on(self.client.instance_status(contract_id)) + sender: ExternalSender, + ) -> Self { + Self { + client, + contract_id, + label, + lifecycle, + spec, + sink, + sender, + last_state: None, + state_since: Instant::now(), + poll: 0, + stopped: false, + } + } + + fn observe_provider_line(&self, line: impl Into) { + self.sink.observe(PluginObservation::ProviderLine { + run_id: self.spec.run_id, + node_id: self.spec.node_id, + line: line.into(), + }); + } + + fn observe_failed(&self, reason: String) { + self.sink.observe(PluginObservation::Failed { + run_id: self.spec.run_id, + node_id: self.spec.node_id, + reason, + }); + } + + fn schedule_next_poll(&self, ctx: &Ctx) { + schedule_provider_monitor_poll( + self.sender.clone(), + ctx.self_addr(), + self.lifecycle.poll_interval, + ); + } + + fn poll_provider(&mut self, ctx: &Ctx) { + if self.stopped { + return; + } + self.poll = self.poll.saturating_add(1); + let poll = self.poll; + let status = match self + .client + .runtime + .block_on(self.client.client.instance_status(self.contract_id)) + { + Ok(status) => status, + Err(error) + if error.contains("not found while fetching provider status") + || error.contains("parse failed") => { - Ok(status) => status, - Err(error) - if error.contains("not found while fetching provider status") - || error.contains("parse failed") => - { - let reason = classified_start_error(format!( - "vastai provider monitor node {} contract {contract_id}: {error}", - spec.node_id - )); - sink.observe(PluginObservation::ProviderLine { - run_id: spec.run_id, - node_id: spec.node_id, - line: serde_json::json!({ - "type": "VastAiProviderStatusFailure", - "run_id": spec.run_id, - "node_id": spec.node_id, - "label": &label, - "contract_id": contract_id, - "poll": poll, - "reason": &reason, - }) - .to_string(), - }); - sink.observe(PluginObservation::Failed { - run_id: spec.run_id, - node_id: spec.node_id, - reason, - }); - return; - } - Err(error) => { - sink.observe(PluginObservation::ProviderLine { - run_id: spec.run_id, - node_id: spec.node_id, - line: serde_json::json!({ - "type": "VastAiProviderStatusPollRetry", - "run_id": spec.run_id, - "node_id": spec.node_id, - "label": &label, - "contract_id": contract_id, - "poll": poll, - "reason": error, - }) - .to_string(), - }); - if !sleep_provider_monitor(lifecycle.poll_interval, &stopping) { - return; - } - continue; - } - }; - - let actual = status.actual_status.as_str(); - if last_state.as_deref() != Some(actual) { - state_since = Instant::now(); - last_state = Some(actual.to_owned()); - } - let in_state_ms = state_since.elapsed().as_millis(); - sink.observe(PluginObservation::ProviderLine { - run_id: spec.run_id, - node_id: spec.node_id, - line: serde_json::json!({ - "type": "VastAiProviderStatusObserved", - "run_id": spec.run_id, - "node_id": spec.node_id, - "label": &label, - "contract_id": contract_id, - "poll": poll, - "actual_status": &status.actual_status, - "intended_status": &status.intended_status, - "status_msg": &status.status_msg, - "disk_usage": status.disk_usage, - "in_state_ms": in_state_ms, - }) - .to_string(), - }); - - if let Some(error) = provider_terminal_start_error( - contract_id, - &status.actual_status, - &status.intended_status, - status.status_msg.as_deref(), - ) { let reason = classified_start_error(format!( - "vastai provider monitor node {}: {error}", - spec.node_id + "vastai provider monitor node {} contract {}: {error}", + self.spec.node_id, self.contract_id )); - sink.observe(PluginObservation::ProviderLine { - run_id: spec.run_id, - node_id: spec.node_id, - line: serde_json::json!({ - "type": "VastAiProviderTerminalBeforeRuntimeReady", - "run_id": spec.run_id, - "node_id": spec.node_id, - "label": &label, - "contract_id": contract_id, + self.observe_provider_line( + serde_json::json!({ + "type": "VastAiProviderStatusFailure", + "run_id": self.spec.run_id, + "node_id": self.spec.node_id, + "label": &self.label, + "contract_id": self.contract_id, + "poll": poll, "reason": &reason, }) .to_string(), - }); - sink.observe(PluginObservation::Failed { - run_id: spec.run_id, - node_id: spec.node_id, - reason, - }); + ); + self.observe_failed(reason); + ctx.stop_self(); return; } - - if !sleep_provider_monitor(lifecycle.poll_interval, &stopping) { + Err(error) => { + self.observe_provider_line( + serde_json::json!({ + "type": "VastAiProviderStatusPollRetry", + "run_id": self.spec.run_id, + "node_id": self.spec.node_id, + "label": &self.label, + "contract_id": self.contract_id, + "poll": poll, + "reason": error, + }) + .to_string(), + ); + self.schedule_next_poll(ctx); return; } + }; + + let actual = status.actual_status.as_str(); + if self.last_state.as_deref() != Some(actual) { + self.state_since = Instant::now(); + self.last_state = Some(actual.to_owned()); + } + let in_state_ms = self.state_since.elapsed().as_millis(); + self.observe_provider_line( + serde_json::json!({ + "type": "VastAiProviderStatusObserved", + "run_id": self.spec.run_id, + "node_id": self.spec.node_id, + "label": &self.label, + "contract_id": self.contract_id, + "poll": poll, + "actual_status": &status.actual_status, + "intended_status": &status.intended_status, + "status_msg": &status.status_msg, + "disk_usage": status.disk_usage, + "in_state_ms": in_state_ms, + }) + .to_string(), + ); + + if let Some(error) = provider_terminal_start_error( + self.contract_id, + &status.actual_status, + &status.intended_status, + status.status_msg.as_deref(), + ) { + let reason = classified_start_error(format!( + "vastai provider monitor node {}: {error}", + self.spec.node_id + )); + self.observe_provider_line( + serde_json::json!({ + "type": "VastAiProviderTerminalBeforeRuntimeReady", + "run_id": self.spec.run_id, + "node_id": self.spec.node_id, + "label": &self.label, + "contract_id": self.contract_id, + "reason": &reason, + }) + .to_string(), + ); + self.observe_failed(reason); + ctx.stop_self(); + return; + } + + self.schedule_next_poll(ctx); + } +} + +impl ActorInterface for VastAiProviderMonitorActor { + type Incoming = VastAiProviderMonitorMsg; + type Response = (); + + fn on_start(&mut self, ctx: &Ctx) { + let _ = ctx.send(ctx.self_addr(), VastAiProviderMonitorMsg::Poll); + } + + fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming) { + match msg { + VastAiProviderMonitorMsg::Poll => self.poll_provider(ctx), + VastAiProviderMonitorMsg::Stop => { + self.stopped = true; + ctx.stop_self(); + } } } } +fn schedule_provider_monitor_poll(sender: ExternalSender, actor: ActorAddress, delay: Duration) { + thread::spawn(move || { + thread::sleep(delay); + let _ = sender.send_to(actor, VastAiProviderMonitorMsg::Poll); + }); +} + impl Clone for ToolsVastAiLeaseClient { fn clone(&self) -> Self { Self { @@ -468,20 +538,21 @@ impl VastAiLeaseClient for ToolsVastAiLeaseClient { spec: NodeProvisionSpec, sink: PluginSink, ) -> Option { - let stopping = Arc::new(AtomicBool::new(false)); - let thread_stopping = Arc::clone(&stopping); - let mut client = self.clone(); - let join = std::thread::spawn(move || { - client.monitor_provider_status( + let runtime = Runtime::new(RuntimeConfig::default()); + let sender = runtime.create_sender(); + let actor = runtime + .spawn(VastAiProviderMonitorActor::new( + self.clone(), contract_id, label, lifecycle, spec, sink, - thread_stopping, - ); - }); - Some(VastAiProviderMonitor::new(stopping, join)) + sender, + )) + .ok()?; + let runtime = runtime.run().ok()?; + Some(VastAiProviderMonitor::new(runtime, actor)) } fn destroy_contract(&mut self, contract_id: u64) -> Result<(), String> { @@ -520,11 +591,10 @@ fn provider_terminal_start_error( intended: &str, msg: Option<&str>, ) -> Option { - if let Some(message) = msg { - let lower = message.to_ascii_lowercase(); - if lower.contains("error") || lower.contains("failed") { - return Some(format!("instance {contract_id} error: {message}")); - } + if let Some(message) = msg + && provider_status_message_has_terminal_failure(message) + { + return Some(format!("instance {contract_id} error: {message}")); } if intended == "stopped" && actual != "running" { return Some(format!( @@ -540,16 +610,15 @@ fn provider_terminal_start_error( } } -fn sleep_provider_monitor(duration: Duration, stopping: &AtomicBool) -> bool { - let deadline = Instant::now() + duration; - while Instant::now() < deadline { - if stopping.load(Ordering::SeqCst) { - return false; - } - let remaining = deadline.saturating_duration_since(Instant::now()); - std::thread::sleep(std::cmp::min(remaining, Duration::from_millis(100))); - } - !stopping.load(Ordering::SeqCst) +fn provider_status_message_has_terminal_failure(message: &str) -> bool { + message + .split(|ch: char| !ch.is_ascii_alphanumeric()) + .any(|token| { + token.eq_ignore_ascii_case("error") + || token.eq_ignore_ascii_case("failed") + || token.eq_ignore_ascii_case("failure") + || token.eq_ignore_ascii_case("fatal") + }) } #[derive(Clone, Debug)] @@ -776,19 +845,271 @@ pub trait VastAiBootstrapLauncher: Send { fn stop_bootstrap(&mut self, handle: &mut Self::Handle, reason: BootstrapStopReason); } +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum SshBootstrapStream { + Stdout, + Stderr, +} + #[derive(Clone)] enum SshBootstrapMsg { + StartAttempt, + PollChild, + OutputLine { + stream: SshBootstrapStream, + line: String, + }, + ReaderError { + stream: SshBootstrapStream, + error: String, + }, + ReaderClosed { + stream: SshBootstrapStream, + }, Stop, } struct SshBootstrapActor { - child: Arc>>, - stopping: Arc, + bridge: BootstrapDatastreamBridge, + endpoint: VastAiSshEndpoint, + ssh_identity: Option, + sender: ExternalSender, + child: Option, + stdout_reader: Option>, + stderr_reader: Option>, + stdout_closed: bool, + stderr_closed: bool, + pending_status: Option, + pending_wait_error: Option, + attempt: u64, + backoff: Duration, + observation_class: Option<&'static str>, + stopped: bool, + start_on_boot: bool, } impl SshBootstrapActor { - fn new(child: Arc>>, stopping: Arc) -> Self { - Self { child, stopping } + fn new( + bridge: BootstrapDatastreamBridge, + endpoint: VastAiSshEndpoint, + ssh_identity: Option, + sender: ExternalSender, + ) -> Self { + Self { + bridge, + endpoint, + ssh_identity, + sender, + child: None, + stdout_reader: None, + stderr_reader: None, + stdout_closed: true, + stderr_closed: true, + pending_status: None, + pending_wait_error: None, + attempt: 1, + backoff: Duration::from_secs(1), + observation_class: None, + stopped: false, + start_on_boot: true, + } + } + + fn run_id(&self) -> u64 { + self.bridge.spec().run_id + } + + fn node_id(&self) -> u64 { + self.bridge.spec().node_id + } + + fn start_attempt(&mut self, ctx: &Ctx) { + if self.stopped { + return; + } + self.bridge.observe_provider_line(format!( + "VastAI SSH bootstrap attempt {} to {}@{}:{}", + self.attempt, self.endpoint.user, self.endpoint.host, self.endpoint.port + )); + + match spawn_ssh_bootstrap_attempt( + self.bridge.spec(), + &self.endpoint, + self.ssh_identity.as_deref(), + ) { + Ok((child, stdout, stderr)) => { + self.child = Some(child); + self.stdout_closed = false; + self.stderr_closed = false; + self.observation_class = None; + self.pending_status = None; + self.pending_wait_error = None; + self.stdout_reader = Some(spawn_ssh_output_reader( + SshBootstrapStream::Stdout, + stdout, + self.sender.clone(), + ctx.self_addr(), + )); + self.stderr_reader = Some(spawn_ssh_output_reader( + SshBootstrapStream::Stderr, + stderr, + self.sender.clone(), + ctx.self_addr(), + )); + schedule_ssh_message( + self.sender.clone(), + ctx.self_addr(), + SshBootstrapMsg::PollChild, + Duration::from_millis(100), + ); + } + Err(error) => { + self.bridge.observe_provider_line(format!( + "spawn VastAI SSH bootstrap attempt {} failed: {error}; retrying", + self.attempt + )); + self.schedule_retry(ctx); + } + } + } + + fn poll_child(&mut self, ctx: &Ctx) { + if self.stopped { + return; + } + let Some(child) = self.child.as_mut() else { + return; + }; + match child.try_wait() { + Ok(Some(status)) => { + self.child = None; + self.pending_status = Some(status); + self.maybe_finish_attempt(ctx); + } + Ok(None) => schedule_ssh_message( + self.sender.clone(), + ctx.self_addr(), + SshBootstrapMsg::PollChild, + Duration::from_millis(100), + ), + Err(error) => { + self.child = None; + self.pending_wait_error = Some(error.to_string()); + self.maybe_finish_attempt(ctx); + } + } + } + + fn handle_output_line(&mut self, stream: SshBootstrapStream, line: String) { + match stream { + SshBootstrapStream::Stdout => self.bridge.observe_stdout_line(line), + SshBootstrapStream::Stderr => { + if let Some(class) = classify_ssh_observation(&line) { + self.observation_class = Some(class); + self.bridge.observe_provider_line( + serde_json::json!({ + "type": "VastAiBootstrapObservationClass", + "run_id": self.run_id(), + "node_id": self.node_id(), + "class": class, + }) + .to_string(), + ); + } + self.bridge.observe_stderr_line(line); + } + } + } + + fn handle_reader_error(&self, stream: SshBootstrapStream, error: String) { + let stream = match stream { + SshBootstrapStream::Stdout => "stdout", + SshBootstrapStream::Stderr => "stderr", + }; + self.bridge + .observe_provider_line(format!("read VastAI SSH {stream}: {error}")); + } + + fn handle_reader_closed(&mut self, ctx: &Ctx, stream: SshBootstrapStream) { + match stream { + SshBootstrapStream::Stdout => self.stdout_closed = true, + SshBootstrapStream::Stderr => self.stderr_closed = true, + } + self.maybe_finish_attempt(ctx); + } + + fn maybe_finish_attempt(&mut self, ctx: &Ctx) { + if self.stopped || !self.stdout_closed || !self.stderr_closed { + return; + } + if let Some(error) = self.pending_wait_error.take() { + self.join_readers(); + self.bridge.observe_provider_line(format!( + "wait VastAI SSH bootstrap attempt {}: {error}; retrying", + self.attempt + )); + self.schedule_retry(ctx); + return; + } + let Some(status) = self.pending_status.take() else { + return; + }; + self.join_readers(); + let readiness = if status.success() { + "exited before runtime ready" + } else { + "not ready before runtime ready" + }; + let observation_class = self.observation_class.unwrap_or("process_exit"); + self.bridge.observe_provider_line( + serde_json::json!({ + "type": "VastAiBootstrapAttemptCompleted", + "run_id": self.run_id(), + "node_id": self.node_id(), + "attempt": self.attempt, + "status": status.to_string(), + "class": observation_class, + "classification": readiness, + }) + .to_string(), + ); + self.schedule_retry(ctx); + } + + fn schedule_retry(&mut self, ctx: &Ctx) { + if self.stopped { + return; + } + let delay = self.backoff; + self.bridge.observe_provider_line(format!( + "VastAI SSH bootstrap retrying in {}s after attempt {}", + delay.as_secs(), + self.attempt + )); + self.backoff = next_ssh_backoff(self.backoff); + self.attempt = self.attempt.saturating_add(1); + schedule_ssh_message( + self.sender.clone(), + ctx.self_addr(), + SshBootstrapMsg::StartAttempt, + delay, + ); + } + + fn join_readers(&mut self) { + if let Some(reader) = self.stdout_reader.take() { + let _ = reader.join(); + } + if let Some(reader) = self.stderr_reader.take() { + let _ = reader.join(); + } + self.stdout_closed = true; + self.stderr_closed = true; + } + + fn stop_child(&mut self) { + stop_ssh_child(&mut self.child); + self.join_readers(); } } @@ -796,18 +1117,37 @@ impl ActorInterface for SshBootstrapActor { type Incoming = SshBootstrapMsg; type Response = (); - fn handle(&mut self, _ctx: &Ctx, msg: Self::Incoming) { + fn on_start(&mut self, ctx: &Ctx) { + if self.start_on_boot { + let _ = ctx.send(ctx.self_addr(), SshBootstrapMsg::StartAttempt); + } + } + + fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming) { match msg { + SshBootstrapMsg::StartAttempt => self.start_attempt(ctx), + SshBootstrapMsg::PollChild => self.poll_child(ctx), + SshBootstrapMsg::OutputLine { stream, line } => self.handle_output_line(stream, line), + SshBootstrapMsg::ReaderError { stream, error } => { + self.handle_reader_error(stream, error) + } + SshBootstrapMsg::ReaderClosed { stream } => self.handle_reader_closed(ctx, stream), SshBootstrapMsg::Stop => { - self.stopping.store(true, Ordering::SeqCst); - stop_ssh_child(&self.child); + self.stopped = true; + self.stop_child(); + ctx.stop_self(); } } } + + fn on_stop(&mut self, _ctx: &Ctx) { + self.stopped = true; + self.stop_child(); + } } -fn stop_ssh_child(child_slot: &Arc>>) { - let Some(mut child) = child_slot.lock().take() else { +fn stop_ssh_child(child: &mut Option) { + let Some(mut child) = child.take() else { return; }; let _ = child.kill(); @@ -851,21 +1191,17 @@ impl VastAiBootstrapLauncher for SshCommandBootstrapLauncher { )); } - let child = Arc::new(Mutex::new(None)); - let stopping = Arc::new(AtomicBool::new(false)); + let sender = self.runtime.create_sender(); + let bridge = BootstrapDatastreamBridge::new(spec, sink, producer); let actor = self .runtime - .spawn(SshBootstrapActor::new(child.clone(), stopping.clone())) + .spawn(SshBootstrapActor::new( + bridge, + endpoint, + self.ssh_identity.clone(), + sender, + )) .map_err(|e| format!("spawn VastAI SSH bootstrap actor: {e}"))?; - spawn_retrying_ssh_bootstrap( - spec, - endpoint, - sink, - producer, - self.ssh_identity.clone(), - child, - stopping, - ); Ok(SshCommandBootstrapHandle { actor, @@ -902,196 +1238,50 @@ fn classify_ssh_observation(line: &str) -> Option<&'static str> { None } -fn spawn_classifying_stderr_reader( - stderr: R, - bridge: BootstrapDatastreamBridge, - observed_class: Arc>>, +fn spawn_ssh_output_reader( + stream: SshBootstrapStream, + reader: R, + sender: ExternalSender, + actor: ActorAddress, ) -> JoinHandle<()> where R: Read + Send + 'static, { - std::thread::spawn(move || { - let reader = BufReader::new(stderr); + thread::spawn(move || { + let reader = BufReader::new(reader); for next in reader.lines() { match next { Ok(line) => { - if let Some(class) = classify_ssh_observation(&line) { - *observed_class.lock() = Some(class); - bridge.observe_provider_line( - serde_json::json!({ - "type": "VastAiBootstrapObservationClass", - "run_id": bridge.spec().run_id, - "node_id": bridge.spec().node_id, - "class": class, - }) - .to_string(), - ); - } - bridge.observe_stderr_line(line); + let _ = sender.send_to(actor, SshBootstrapMsg::OutputLine { stream, line }); } Err(error) => { - bridge.observe_provider_line(format!("read VastAI SSH stderr: {error}")); + let _ = sender.send_to( + actor, + SshBootstrapMsg::ReaderError { + stream, + error: error.to_string(), + }, + ); break; } } } + let _ = sender.send_to(actor, SshBootstrapMsg::ReaderClosed { stream }); }) } -fn spawn_retrying_ssh_bootstrap( - spec: NodeProvisionSpec, - endpoint: VastAiSshEndpoint, - sink: PluginSink, - producer: Option, - ssh_identity: Option, - child_slot: Arc>>, - stopping: Arc, + +fn schedule_ssh_message( + sender: ExternalSender, + actor: ActorAddress, + msg: SshBootstrapMsg, + delay: Duration, ) { - std::thread::spawn(move || { - let run_id = spec.run_id; - let node_id = spec.node_id; - let mut attempt = 1u64; - let mut backoff = Duration::from_secs(1); - - while !stopping.load(Ordering::SeqCst) { - sink.observe(PluginObservation::ProviderLine { - run_id, - node_id, - line: format!( - "VastAI SSH bootstrap attempt {attempt} to {}@{}:{}", - endpoint.user, endpoint.host, endpoint.port - ), - }); - - match spawn_ssh_bootstrap_attempt(&spec, &endpoint, ssh_identity.as_deref()) { - Ok((child, stdout, stderr)) => { - *child_slot.lock() = Some(child); - let bridge = BootstrapDatastreamBridge::new( - spec.clone(), - sink.clone(), - producer.clone(), - ); - bridge.spawn_stdout_reader(stdout); - let observed_class = Arc::new(Mutex::new(None)); - let mut stderr_reader = Some(spawn_classifying_stderr_reader( - stderr, - bridge.clone(), - Arc::clone(&observed_class), - )); - - loop { - if stopping.load(Ordering::SeqCst) { - return; - } - - let wait_result = { - let mut guard = child_slot.lock(); - match guard.as_mut() { - Some(child) => match child.try_wait() { - Ok(Some(status)) => { - *guard = None; - Some(Ok(status)) - } - Ok(None) => None, - Err(error) => { - *guard = None; - Some(Err(error)) - } - }, - None => Some(Err(std::io::Error::new( - std::io::ErrorKind::Other, - "ssh child missing", - ))), - } - }; - - match wait_result { - Some(Ok(status)) => { - let readiness = if status.success() { - "exited before runtime ready" - } else { - "not ready before runtime ready" - }; - if let Some(reader) = stderr_reader.take() { - let _ = reader.join(); - } - let observation_class = - (*observed_class.lock()).unwrap_or("process_exit"); - sink.observe(PluginObservation::ProviderLine { - run_id, - node_id, - line: serde_json::json!({ - "type": "VastAiBootstrapAttemptCompleted", - "run_id": run_id, - "node_id": node_id, - "attempt": attempt, - "status": status.to_string(), - "class": observation_class, - "classification": readiness, - }) - .to_string(), - }); - break; - } - Some(Err(error)) => { - if let Some(reader) = stderr_reader.take() { - let _ = reader.join(); - } - sink.observe(PluginObservation::ProviderLine { - run_id, - node_id, - line: format!( - "wait VastAI SSH bootstrap attempt {attempt}: {error}; retrying" - ), - }); - break; - } - None => std::thread::sleep(Duration::from_millis(100)), - } - } - } - Err(error) => { - sink.observe(PluginObservation::ProviderLine { - run_id, - node_id, - line: format!( - "spawn VastAI SSH bootstrap attempt {attempt} failed: {error}; retrying" - ), - }); - } - } - - if stopping.load(Ordering::SeqCst) { - return; - } - sink.observe(PluginObservation::ProviderLine { - run_id, - node_id, - line: format!( - "VastAI SSH bootstrap retrying in {}s after attempt {attempt}", - backoff.as_secs() - ), - }); - if !sleep_ssh_backoff(backoff, &stopping) { - return; - } - backoff = next_ssh_backoff(backoff); - attempt += 1; - } + thread::spawn(move || { + thread::sleep(delay); + let _ = sender.send_to(actor, msg); }); } -fn sleep_ssh_backoff(backoff: Duration, stopping: &AtomicBool) -> bool { - let deadline = std::time::Instant::now() + backoff; - while std::time::Instant::now() < deadline { - if stopping.load(Ordering::SeqCst) { - return false; - } - let remaining = deadline.saturating_duration_since(std::time::Instant::now()); - std::thread::sleep(std::cmp::min(remaining, Duration::from_millis(50))); - } - !stopping.load(Ordering::SeqCst) -} - fn spawn_ssh_bootstrap_attempt( spec: &NodeProvisionSpec, endpoint: &VastAiSshEndpoint, @@ -1747,10 +1937,6 @@ where let Some(node) = self.nodes.get_mut(&handle.id) else { return Ok(()); }; - if let Some(monitor) = node.provider_monitor.as_mut() { - monitor.stop(); - } - node.provider_monitor = None; node.sink.observe(PluginObservation::ProviderLine { run_id: node.run_id, node_id: node.node_id, @@ -1764,10 +1950,6 @@ where }) .to_string(), }); - if let Some(mut bootstrap) = node.bootstrap.take() { - self.bootstrap - .stop_bootstrap(&mut bootstrap, BootstrapStopReason::RuntimeReady); - } Ok(()) } @@ -1973,7 +2155,7 @@ mod tests { } #[test] - fn vastai_complete_bootstrap_stops_optional_log_tail_before_node_stop() { + fn vastai_complete_bootstrap_keeps_optional_log_tail_until_node_stop() { let destroyed_contracts = Arc::new(Mutex::new(Vec::new())); let stop_reasons = Arc::new(Mutex::new(Vec::new())); let sink = PluginSink::new(Arc::new(ObservationSink::default())); @@ -1994,16 +2176,13 @@ mod tests { plugin .complete_bootstrap(&handle) .expect("runtime-ready bootstrap completion succeeds"); - assert_eq!( - *stop_reasons.lock(), - vec![BootstrapStopReason::RuntimeReady] + assert!( + stop_reasons.lock().is_empty(), + "runtime-ready keeps the SSH log tail alive for post-bootstrap diagnostics" ); plugin.stop_node(&handle).expect("VastAI node stops"); - assert_eq!( - *stop_reasons.lock(), - vec![BootstrapStopReason::RuntimeReady] - ); + assert_eq!(*stop_reasons.lock(), vec![BootstrapStopReason::NodeStop]); assert_eq!(*destroyed_contracts.lock(), vec![42]); } @@ -2098,14 +2277,25 @@ mod tests { } #[test] - fn ssh_bootstrap_backoff_sleep_observes_stop_without_waiting_full_backoff() { - let stopping = AtomicBool::new(true); - let started = std::time::Instant::now(); + fn provider_terminal_start_error_ignores_package_names_while_loading() { + let package_log = "#7 1.745 libevent-core-2.1-7t64 liberror-perl libglib2.0-data"; - assert!(!sleep_ssh_backoff(Duration::from_secs(5), &stopping)); - assert!( - started.elapsed() < Duration::from_millis(250), - "stopped bootstrap backoff should not wait for the full retry delay" + assert_eq!( + provider_terminal_start_error(46132050, "loading", "running", Some(package_log)), + None + ); + } + + #[test] + fn provider_terminal_start_error_reports_standalone_failure_words() { + assert_eq!( + provider_terminal_start_error( + 46132050, + "loading", + "running", + Some("ERROR: build failed") + ), + Some("instance 46132050 error: ERROR: build failed".to_owned()) ); } @@ -2119,19 +2309,40 @@ mod tests { .spawn() .expect("spawn sleep child"); let pid = child.id(); - let child_slot = Arc::new(Mutex::new(Some(child))); - let stopping = Arc::new(AtomicBool::new(false)); + let bridge = BootstrapDatastreamBridge::new( + node_spec_with_bootstrap_args(), + PluginSink::new(Arc::new(ObservationSink::default())), + None, + ); let actor = runtime - .spawn(SshBootstrapActor::new(child_slot.clone(), stopping.clone())) + .spawn(SshBootstrapActor { + bridge, + endpoint: VastAiSshEndpoint { + host: "127.0.0.1".to_owned(), + port: 22, + user: "ubuntu".to_owned(), + }, + ssh_identity: None, + sender: runtime.create_sender(), + child: Some(child), + stdout_reader: None, + stderr_reader: None, + stdout_closed: true, + stderr_closed: true, + pending_status: None, + pending_wait_error: None, + attempt: 1, + backoff: Duration::from_secs(1), + observation_class: None, + stopped: false, + start_on_boot: false, + }) .expect("spawn ssh bootstrap actor"); runtime .send_to(actor, SshBootstrapMsg::Stop) .expect("send stop"); runtime.tick(); - - assert!(stopping.load(Ordering::SeqCst)); - assert!(child_slot.lock().is_none()); #[cfg(target_os = "linux")] assert!( !std::path::Path::new(&format!("/proc/{pid}")).exists(), diff --git a/crates/mvp-system/src/staging/gguf_shard.rs b/crates/mvp-system/src/staging/gguf_shard.rs index efd1142..9a51ba3 100644 --- a/crates/mvp-system/src/staging/gguf_shard.rs +++ b/crates/mvp-system/src/staging/gguf_shard.rs @@ -10,6 +10,7 @@ const GGUF_MAGIC: &[u8; 4] = b"GGUF"; const SUPPORTED_GGUF_VERSION: u32 = 3; const DEFAULT_ALIGNMENT: u64 = 32; const MAX_STRING_BYTES: u64 = 64 * 1024 * 1024; +const STAGE_SHARD_CACHE_FORMAT_VERSION: &str = "stage-shard-cache-v2"; #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] pub struct ByteRange { @@ -166,6 +167,45 @@ pub fn plan_stage_shard( }) } +pub fn validate_stage_shard_cache(path: &Path, plan: &StageShardPlan) -> Result<(), String> { + let directory = read_gguf_directory(path) + .map_err(|error| format!("invalid cached stage shard {}: {error}", path.display()))?; + if directory.tensors.len() != plan.tensors.len() { + return Err(format!( + "cached stage shard {} has {} tensors; expected {}", + path.display(), + directory.tensors.len(), + plan.tensors.len() + )); + } + if directory.alignment != plan.alignment { + return Err(format!( + "cached stage shard {} has alignment {}; expected {}", + path.display(), + directory.alignment, + plan.alignment + )); + } + for (index, (actual, expected)) in directory.tensors.iter().zip(&plan.tensors).enumerate() { + if actual.name != expected.name + || actual.dims != expected.dims + || actual.ggml_type != expected.ggml_type + { + return Err(format!( + "cached stage shard {} tensor {index} is {} {:?} type {}; expected {} {:?} type {}", + path.display(), + actual.name, + actual.dims, + actual.ggml_type, + expected.name, + expected.dims, + expected.ggml_type + )); + } + } + Ok(()) +} + pub fn source_url(source: &GgufSource) -> Result { match source { GgufSource::HuggingFaceGguf { @@ -405,6 +445,7 @@ fn shard_cache_key( tensors: &[GgufTensorEntry], ) -> String { let mut hasher = blake3::Hasher::new(); + hasher.update(format!("format:{STAGE_SHARD_CACHE_FORMAT_VERSION}\n").as_bytes()); hasher.update(format!("source:{source:?}\n").as_bytes()); hasher.update( format!("stage:{stage_index}/{stage_count}:{layer_start}-{layer_end_exclusive}\n") @@ -1128,6 +1169,24 @@ mod tests { .sum::(); assert_eq!(total_requested, plan.planned_fetch_bytes()); assert!(total_requested < fixture.bytes_len); + validate_stage_shard_cache(&output_path, &plan).unwrap(); + } + + #[test] + fn corrupted_stage_shard_cache_is_rejected_before_reuse() { + let fixture = SyntheticGguf::new(4); + let source = GgufSource::HuggingFaceGguf { + repo: "org/repo".to_owned(), + file: "model.gguf".to_owned(), + revision: None, + }; + let plan = plan_stage_shard(&fixture.path, source, 1, 2, 2, 4).unwrap(); + let output_path = fixture.path.with_file_name(plan.cache_file_name()); + + std::fs::write(&output_path, b"not a gguf").unwrap(); + + let error = validate_stage_shard_cache(&output_path, &plan).unwrap_err(); + assert!(error.contains("invalid cached stage shard")); } fn names(plan: &StageShardPlan) -> Vec<&str> { diff --git a/crates/mvp-system/src/tests/data_plane_bridge_guarantees.rs b/crates/mvp-system/src/tests/data_plane_bridge_guarantees.rs index 26d7763..d70a5af 100644 --- a/crates/mvp-system/src/tests/data_plane_bridge_guarantees.rs +++ b/crates/mvp-system/src/tests/data_plane_bridge_guarantees.rs @@ -88,7 +88,12 @@ fn data_plane_fault_and_stop_reports_map_to_mvp_lifecycle_messages() { .expect("send stopped"); runtime.tick(); - assert_eq!(node_messages.try_recv(), Some(NodeAgentMsg::WorkerCrashed)); + assert_eq!( + node_messages.try_recv(), + Some(NodeAgentMsg::WorkerCrashed { + reason: Some("data plane faulted".to_owned()) + }) + ); assert_eq!( node_messages.try_recv(), Some(NodeAgentMsg::LocalEdgesStopped { run_id: 55 }) diff --git a/crates/mvp-system/src/tests/vastai_provisioning_guarantees.rs b/crates/mvp-system/src/tests/vastai_provisioning_guarantees.rs index ee691c0..dbaffc8 100644 --- a/crates/mvp-system/src/tests/vastai_provisioning_guarantees.rs +++ b/crates/mvp-system/src/tests/vastai_provisioning_guarantees.rs @@ -673,7 +673,7 @@ fn stop_destroys_known_vastai_contract_exactly_once() { } #[test] -fn vastai_complete_bootstrap_stops_optional_log_tail_before_node_stop() { +fn vastai_complete_bootstrap_keeps_optional_log_tail_until_node_stop() { let mut plugin = VastAiProvisioningPlugin::new( FakeLeaseClient::default().with_contract(100), FakeBootstrap::default(), @@ -683,10 +683,7 @@ fn vastai_complete_bootstrap_stops_optional_log_tail_before_node_stop() { plugin.complete_bootstrap(&handle).unwrap(); - assert_eq!( - plugin.bootstrap().stops, - vec![(1, BootstrapStopReason::RuntimeReady)] - ); + assert!(plugin.bootstrap().stops.is_empty()); assert_eq!(plugin.client().destroyed, Vec::::new()); assert_eq!(plugin.active_contract_count(), 1); @@ -695,7 +692,7 @@ fn vastai_complete_bootstrap_stops_optional_log_tail_before_node_stop() { assert_eq!(plugin.client().destroyed, vec![100]); assert_eq!( plugin.bootstrap().stops, - vec![(1, BootstrapStopReason::RuntimeReady)] + vec![(1, BootstrapStopReason::NodeStop)] ); assert_eq!(plugin.active_contract_count(), 0); } diff --git a/xtask/src/main.rs b/xtask/src/main.rs index d26c0ae..7413882 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -4689,20 +4689,16 @@ fn require_vastai_data_path_facts(facts: &DumpLogFacts) -> Result<(), String> { "worker ingress ring installed", )?; require_dump_log_fact(facts.ring_installed_egress, "worker egress ring installed")?; + let downstream_activation = facts.activation_downstream_object_loaded + && facts.max_activation_record_bytes >= DATA_PATH_MIN_PAYLOAD_BYTES; require_dump_log_fact( - facts.activation_object_loaded || facts.worker_ingress_object_loaded, - "worker object loaded from ingress ring", - )?; - require_dump_log_fact( - facts.activation_step_executed, - "activation-producing worker step executed", + facts.activation_step_executed || downstream_activation, + "activation-producing worker step executed or downstream activation loaded", )?; let explicit_transport = facts.activation_egress_ring_read && facts.activation_iroh_edge_sent && facts.activation_iroh_edge_read && facts.activation_ingress_ring_write; - let downstream_activation = facts.activation_downstream_object_loaded - && facts.max_activation_record_bytes >= DATA_PATH_MIN_PAYLOAD_BYTES; let downstream_prompt_output = facts.activation_egress_record_written && facts.pipeline_token_out_requests.len() >= 2 && facts.vastai_node_runtime_ready_nodes.len() >= 2; @@ -6100,7 +6096,6 @@ mod tests { ring_installed_ingress: true, ring_installed_egress: true, activation_object_loaded: true, - activation_step_executed: true, activation_downstream_object_loaded: true, max_activation_record_bytes: DATA_PATH_MIN_PAYLOAD_BYTES, ..DumpLogFacts::default()