From 4605700af618c55e3c4d81a1b7cef59169030386 Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Sun, 26 Jul 2026 00:05:44 +0400 Subject: [PATCH] feat: add swim probe telemetry, vastai planning cache - Add SWIM probe telemetry: SwimProbeEvent / swim.probes channel and SwimTelemetry observer tracking probe events, consecutive timeouts, and last-ack age. - Enrich MembershipTransition; thread telemetry through worker_node/config/ distribution_stack. - Expand orchestrator_app VastAI planning-cache handling; add vastai teardown logging; xtask updates. Signed-off-by: Zachery Aaron Shores-Chmielewski --- crates/distribution/src/swim/telemetry.rs | 142 ++++++++- crates/distribution/src/telemetry.rs | 56 ++++ crates/distribution/tests/telemetry.rs | 113 ++++++- crates/mvp-system/src/bin/mvp_chat.rs | 43 ++- crates/mvp-system/src/bin/worker_node.rs | 138 ++++++++ crates/mvp-system/src/config.rs | 10 + crates/mvp-system/src/distribution_stack.rs | 12 +- crates/mvp-system/src/orchestrator_app.rs | 329 ++++++++++++++++++-- tools/vastai/src/teardown.rs | 92 +++++- xtask/src/main.rs | 32 +- 10 files changed, 909 insertions(+), 58 deletions(-) diff --git a/crates/distribution/src/swim/telemetry.rs b/crates/distribution/src/swim/telemetry.rs index ffe804d..00e98fa 100644 --- a/crates/distribution/src/swim/telemetry.rs +++ b/crates/distribution/src/swim/telemetry.rs @@ -28,19 +28,45 @@ const TARGET_RING: usize = 16; /// this without bound. Oldest are dropped first (a lost transition is a gap, not /// a renumber — the same tolerance as the mux). const TRANSITION_CAP: usize = 256; +/// Cap on undrained probe events. Probe events are diagnostics; dropping old +/// ones is better than letting a wedged consumer grow this queue forever. +const PROBE_EVENT_CAP: usize = 512; /// In-flight probes older than this are pruned defensively. The probe state /// machine resolves every probe (ack or timeout), so this only guards against a /// dropped observation leaking an entry forever. const IN_FLIGHT_TTL: Duration = Duration::from_secs(30); /// One captured membership transition, carrying the real cause string the -/// state-diff path could never know. +/// state-diff path could never know, plus the probe state visible at the moment +/// of transition. #[derive(Debug, Clone)] pub struct ObservedTransition { pub peer: NodeId, pub from: Option, pub to: MemberState, pub reason: &'static str, + pub last_ack_age: Option, + pub consecutive_timeouts: u32, +} + +/// One captured SWIM probe lifecycle event. +#[derive(Debug, Clone)] +pub struct ObservedProbeEvent { + pub event: &'static str, + pub target: NodeId, + pub sequence: u64, + pub kind: &'static str, + pub rtt_ms: Option, + pub budget_ms: Option, + pub last_ack_age: Option, + pub consecutive_timeouts: u32, +} + +/// Current probe state for one peer. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct PeerProbeState { + pub last_ack_age: Option, + pub consecutive_timeouts: u32, } #[derive(Default)] @@ -51,6 +77,12 @@ struct Inner { rtts: VecDeque, /// Recent probe targets, newest last. targets: VecDeque, + /// Last successful probe ack per peer. + last_ack: HashMap, + /// Consecutive direct/indirect timeouts per peer since the last ack. + consecutive_timeouts: HashMap, + /// Probe events awaiting drain by telemetry consumers. + probe_events: VecDeque, /// Transitions awaiting drain by the node's telemetry tick. transitions: VecDeque, } @@ -104,6 +136,23 @@ impl SwimTelemetry { .collect() } + /// Take probe lifecycle events captured since the last call (FIFO, then + /// cleared). + pub fn drain_probe_events(&self) -> Vec { + self.inner + .lock() + .expect("swim telemetry poisoned") + .probe_events + .drain(..) + .collect() + } + + /// Snapshot the probe state for one peer without draining events. + pub fn peer_probe_state(&self, peer: NodeId) -> PeerProbeState { + let inner = self.inner.lock().expect("swim telemetry poisoned"); + Self::peer_probe_state_locked(&inner, peer) + } + /// The most recent transition cause per peer, **without** draining the queue. /// A snapshot reader (e.g. the orchestrator's Distribution view) uses this to /// label each member with *why* it last changed state; draining stays @@ -122,7 +171,9 @@ impl SwimTelemetry { let mut inner = self.inner.lock().expect("swim telemetry poisoned"); match observation { SwimObservation::ProbeSent { - target, sequence, .. + target, + sequence, + kind, } => { // Drop any leaked in-flight entries before tracking a new probe. inner @@ -133,23 +184,81 @@ impl SwimTelemetry { inner.targets.pop_front(); } inner.targets.push_back(target); + let state = Self::peer_probe_state_locked(&inner, target); + Self::push_probe_event( + &mut inner, + ObservedProbeEvent { + event: "sent", + target, + sequence, + kind, + rtt_ms: None, + budget_ms: None, + last_ack_age: state.last_ack_age, + consecutive_timeouts: state.consecutive_timeouts, + }, + ); } SwimObservation::ProbeAcked { - target, sequence, .. + target, + sequence, + kind, } => { - if let Some(sent) = inner.in_flight.remove(&(target, sequence)) { - let rtt = sent.elapsed().as_millis().min(u32::MAX as u128) as u32; + let state = Self::peer_probe_state_locked(&inner, target); + let rtt_ms = inner + .in_flight + .remove(&(target, sequence)) + .map(|sent| sent.elapsed().as_millis().min(u32::MAX as u128) as u32); + if let Some(rtt) = rtt_ms { if inner.rtts.len() >= RTT_RING { inner.rtts.pop_front(); } inner.rtts.push_back(rtt); } + inner.last_ack.insert(target, Instant::now()); + inner.consecutive_timeouts.remove(&target); + Self::push_probe_event( + &mut inner, + ObservedProbeEvent { + event: "acked", + target, + sequence, + kind, + rtt_ms, + budget_ms: None, + last_ack_age: state.last_ack_age, + consecutive_timeouts: state.consecutive_timeouts, + }, + ); } SwimObservation::ProbeTimedOut { - target, sequence, .. + target, + sequence, + kind, + budget_ticks, } => { // A timeout is not a round-trip — drop the in-flight entry, no sample. inner.in_flight.remove(&(target, sequence)); + let timeouts = inner + .consecutive_timeouts + .entry(target) + .and_modify(|count| *count = count.saturating_add(1)) + .or_insert(1); + let consecutive_timeouts = *timeouts; + let last_ack_age = inner.last_ack.get(&target).map(Instant::elapsed); + Self::push_probe_event( + &mut inner, + ObservedProbeEvent { + event: "timed_out", + target, + sequence, + kind, + rtt_ms: None, + budget_ms: Some(budget_ticks), + last_ack_age, + consecutive_timeouts, + }, + ); } SwimObservation::Transition { peer, @@ -160,15 +269,36 @@ impl SwimTelemetry { if inner.transitions.len() >= TRANSITION_CAP { inner.transitions.pop_front(); } + let state = Self::peer_probe_state_locked(&inner, peer); inner.transitions.push_back(ObservedTransition { peer, from, to, reason, + last_ack_age: state.last_ack_age, + consecutive_timeouts: state.consecutive_timeouts, }); } } } + + fn peer_probe_state_locked(inner: &Inner, peer: NodeId) -> PeerProbeState { + PeerProbeState { + last_ack_age: inner.last_ack.get(&peer).map(Instant::elapsed), + consecutive_timeouts: inner + .consecutive_timeouts + .get(&peer) + .copied() + .unwrap_or_default(), + } + } + + fn push_probe_event(inner: &mut Inner, event: ObservedProbeEvent) { + if inner.probe_events.len() >= PROBE_EVENT_CAP { + inner.probe_events.pop_front(); + } + inner.probe_events.push_back(event); + } } impl SwimObserver for Arc { diff --git a/crates/distribution/src/telemetry.rs b/crates/distribution/src/telemetry.rs index 1775170..a1f2111 100644 --- a/crates/distribution/src/telemetry.rs +++ b/crates/distribution/src/telemetry.rs @@ -9,6 +9,8 @@ pub const TRANSPORT_INTERNALS: &str = "transport.internals"; pub const MEMBERSHIP: &str = "membership"; /// Distribution-subsystem state: cache, directory, registry, probes, peer auth. pub const DIST_STATE: &str = "dist.state"; +/// Probe lifecycle events emitted by the SWIM observer. +pub const SWIM_PROBES: &str = "swim.probes"; /// Transport-internals record. #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] @@ -31,6 +33,57 @@ pub struct MembershipTransition { pub to: String, #[serde(default)] pub reason: String, + #[serde(default)] + pub last_ack_age_ms: Option, + #[serde(default)] + pub consecutive_timeouts: u32, + #[serde(default)] + pub recent_probe_targets: Vec, + #[serde(default)] + pub member_state: Option, +} + +/// One SWIM probe lifecycle event. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct SwimProbeEvent { + /// `"sent"`, `"acked"`, or `"timed_out"`. + pub event: String, + pub target: String, + pub sequence: u64, + /// `"direct"` or `"indirect"`. + pub kind: String, + #[serde(default)] + pub rtt_ms: Option, + #[serde(default)] + pub budget_ms: Option, + /// Retained for compatibility with the existing diagnostic field name. The + /// value is milliseconds in the wall-clock implementation. + #[serde(default)] + pub budget_ticks: Option, + #[serde(default)] + pub last_ack_age_ms: Option, + #[serde(default)] + pub consecutive_timeouts: u32, + #[serde(default)] + pub recent_probe_targets: Vec, + #[serde(default)] + pub member_state: Option, + #[serde(default)] + pub local_phase: String, + #[serde(default)] + pub probe_interval_ms: u64, + #[serde(default)] + pub probe_timeout_ms: u64, + #[serde(default)] + pub indirect_probes: u32, + #[serde(default)] + pub suspicion_timeout_ms: u64, + #[serde(default)] + pub dead_reprobe_interval_ms: u64, + #[serde(default)] + pub probe_mode: String, + #[serde(default)] + pub lifeguard_enabled: bool, } /// Consolidated distribution-subsystem state. @@ -86,6 +139,9 @@ impl Record for TransportInternals { impl Record for MembershipTransition { const CHANNEL: &'static str = MEMBERSHIP; } +impl Record for SwimProbeEvent { + const CHANNEL: &'static str = SWIM_PROBES; +} impl Record for DistributionState { const CHANNEL: &'static str = DIST_STATE; } diff --git a/crates/distribution/tests/telemetry.rs b/crates/distribution/tests/telemetry.rs index 6ede5a9..b004608 100644 --- a/crates/distribution/tests/telemetry.rs +++ b/crates/distribution/tests/telemetry.rs @@ -16,7 +16,8 @@ mod datastream_records { use datastream::frame::{Lifetime, NodeId, StreamId}; use datastream::{ChannelId, Mux, Position, Record}; use distribution::telemetry::{ - CacheEntryRec, DIST_STATE, DistributionState, MembershipTransition, RegistryEntryRec, + CacheEntryRec, DIST_STATE, DistributionState, MEMBERSHIP, MembershipTransition, + RegistryEntryRec, SWIM_PROBES, SwimProbeEvent, TRANSPORT_INTERNALS, TransportInternals, }; #[test] @@ -69,6 +70,22 @@ mod datastream_records { ); } + #[test] + fn transport_internals_record_round_trips_probe_rtt() { + let internals = TransportInternals { + relay_connected: true, + direct_peers: 2, + relay_peers: 1, + rtt_ms_p50: 405, + }; + + assert_eq!(TransportInternals::CHANNEL, TRANSPORT_INTERNALS); + assert_eq!( + TransportInternals::decode(&internals.encode()).unwrap(), + internals + ); + } + #[test] fn membership_transition_record_round_trips_from_owner_crate() { let transition = MembershipTransition { @@ -76,13 +93,46 @@ mod datastream_records { from: "alive".into(), to: "suspect".into(), reason: "probe timeout".into(), + last_ack_age_ms: Some(15_000), + consecutive_timeouts: 2, + recent_probe_targets: vec!["peer-a".into(), "peer-b".into()], + member_state: Some("Suspect".into()), }; + assert_eq!(MembershipTransition::CHANNEL, MEMBERSHIP); assert_eq!( MembershipTransition::decode(&transition.encode()).unwrap(), transition ); } + + #[test] + fn swim_probe_event_record_round_trips_from_owner_crate() { + let event = SwimProbeEvent { + event: "timed_out".into(), + target: "peer-a".into(), + sequence: 9, + kind: "direct".into(), + rtt_ms: None, + budget_ms: Some(15_000), + budget_ticks: Some(15_000), + last_ack_age_ms: Some(45_000), + consecutive_timeouts: 3, + recent_probe_targets: vec!["peer-a".into()], + member_state: Some("Suspect".into()), + local_phase: "weights_loaded_wait".into(), + probe_interval_ms: 200, + probe_timeout_ms: 15_000, + indirect_probes: 2, + suspicion_timeout_ms: 45_000, + dead_reprobe_interval_ms: 1_000, + probe_mode: "Periodic".into(), + lifeguard_enabled: false, + }; + + assert_eq!(SwimProbeEvent::CHANNEL, SWIM_PROBES); + assert_eq!(SwimProbeEvent::decode(&event.encode()).unwrap(), event); + } } mod snapshot_and_swim_telemetry { @@ -93,6 +143,7 @@ mod snapshot_and_swim_telemetry { use distribution::swim::node::{SwimObservation, SwimObserver}; use distribution::swim::telemetry::SwimTelemetry; use distribution::types::{MemberState, NodeId}; + use std::time::Duration; fn id(byte: u8) -> NodeId { NodeId([byte; 32]) @@ -147,6 +198,66 @@ mod snapshot_and_swim_telemetry { assert_eq!(drained[0].from, Some(MemberState::Alive)); assert_eq!(drained[0].to, MemberState::Suspect); assert_eq!(drained[0].reason, "probe-timeout"); + assert_eq!(drained[0].consecutive_timeouts, 0); + assert!(drained[0].last_ack_age.is_none()); assert!(telemetry.drain_transitions().is_empty()); + assert_eq!(telemetry.recent_targets(), vec![peer]); + } + + #[test] + fn swim_telemetry_keeps_recent_probe_targets_bounded_and_ordered() { + let telemetry = SwimTelemetry::new(); + + for byte in 0..17 { + telemetry.observe(SwimObservation::ProbeSent { + target: id(byte), + sequence: byte as u64, + kind: "direct", + }); + } + + let expected = (1..17).map(id).collect::>(); + assert_eq!(telemetry.recent_targets(), expected); + } + + #[test] + fn swim_telemetry_records_probe_events_and_timeout_state_without_fabricating_rtt() { + let telemetry = SwimTelemetry::new(); + let peer = id(8); + + telemetry.observe(SwimObservation::ProbeSent { + target: peer, + sequence: 1, + kind: "direct", + }); + std::thread::sleep(Duration::from_millis(1)); + telemetry.observe(SwimObservation::ProbeAcked { + target: peer, + sequence: 1, + kind: "direct", + }); + telemetry.observe(SwimObservation::ProbeSent { + target: peer, + sequence: 2, + kind: "direct", + }); + telemetry.observe(SwimObservation::ProbeTimedOut { + target: peer, + sequence: 2, + kind: "direct", + budget_ticks: 15_000, + }); + + let events = telemetry.drain_probe_events(); + assert_eq!(events.len(), 4); + assert_eq!(events[0].event, "sent"); + assert_eq!(events[1].event, "acked"); + assert!(events[1].rtt_ms.is_some()); + assert_eq!(events[2].event, "sent"); + assert_eq!(events[3].event, "timed_out"); + assert_eq!(events[3].budget_ms, Some(15_000)); + assert_eq!(events[3].rtt_ms, None); + assert_eq!(events[3].consecutive_timeouts, 1); + assert!(telemetry.drain_probe_events().is_empty()); } } diff --git a/crates/mvp-system/src/bin/mvp_chat.rs b/crates/mvp-system/src/bin/mvp_chat.rs index 26b7ce0..eaacbcd 100644 --- a/crates/mvp-system/src/bin/mvp_chat.rs +++ b/crates/mvp-system/src/bin/mvp_chat.rs @@ -638,6 +638,7 @@ struct ChatVastAiConfig { max_dph_total: Option, min_reliability: Option, require_verified: Option, + blacklist_hosts: Vec, disk_gb: Option, onstart: Option, ssh_identity: Option, @@ -793,7 +794,7 @@ impl Config { self.run_id.to_string(), "--pipeline-stages".to_owned(), self.pipeline_stages.to_string(), - "--no-dashboard".to_owned(), + "--dashboard".to_owned(), ]; if let Some(model_id) = &self.model.id { args.extend(["--model-id".to_owned(), model_id.clone()]); @@ -1028,6 +1029,7 @@ fn resolve_vastai_config( max_dph_total: file.max_dph_total, min_reliability: file.min_reliability, require_verified: file.require_verified, + blacklist_hosts: file.blacklist_hosts.clone(), onstart: first_non_empty([file.onstart.clone()]), ssh_identity: first_non_empty([file.ssh_identity.clone()]), } @@ -1964,20 +1966,26 @@ fn cargo_command() -> &'static str { "cargo" } +fn mvp_orchestrator_build_args() -> &'static [&'static str] { + &[ + "build", + "--quiet", + "-p", + "mvp-system", + "--features", + "dashboard", + "--bin", + "mvp-orchestrator", + ] +} + fn ensure_orch_binary(config: &Config) -> Result<(), String> { if config.skip_rebuild { return ensure_existing_artifact(&config.orch_bin, "mvp-orchestrator"); } run_status( cargo_command(), - &[ - "build", - "--quiet", - "-p", - "mvp-system", - "--bin", - "mvp-orchestrator", - ], + mvp_orchestrator_build_args(), "build mvp-orchestrator", ) } @@ -2333,6 +2341,22 @@ mod tests { } } + #[test] + fn observability_server_launch_contract_enables_orchestrator_dashboard() { + let build_args = mvp_orchestrator_build_args(); + assert!( + build_args + .windows(2) + .any(|pair| pair[0] == "--features" && pair[1] == "dashboard"), + "{build_args:?}" + ); + + let config = base_config(ProviderKind::Process); + let args = config.orchestrator_cli_args("resolved-image"); + assert!(args.iter().any(|arg| arg == "--dashboard"), "{args:?}"); + assert!(!args.iter().any(|arg| arg == "--no-dashboard"), "{args:?}"); + } + fn valid_vastai() -> ResolvedVastAiConfig { ResolvedVastAiConfig { api_key: "secret".to_owned(), @@ -2347,6 +2371,7 @@ mod tests { max_dph_total: None, min_reliability: None, require_verified: None, + blacklist_hosts: Vec::new(), onstart: None, ssh_identity: None, } diff --git a/crates/mvp-system/src/bin/worker_node.rs b/crates/mvp-system/src/bin/worker_node.rs index 828bca0..a3e5585 100644 --- a/crates/mvp-system/src/bin/worker_node.rs +++ b/crates/mvp-system/src/bin/worker_node.rs @@ -17,6 +17,8 @@ use datastream::{ }; use distribution::node::DistributedNodeConfig; +use distribution::swim::telemetry::ObservedProbeEvent; +use distribution::telemetry::{MembershipTransition, SwimProbeEvent}; use distribution::types::{MemberState, NodeId as DistNodeId}; use iroh::EndpointAddr; use iroh_driver::{ @@ -1843,6 +1845,7 @@ fn run() -> Result<(), String> { ); loop { pump_network(&mut driver, &stack); + emit_swim_telemetry(&mut datastream, &stack, "main_loop"); drain_debug_join_commands(&mut debug_join_rx, &mut driver, &config, &mut datastream); datastream.tick(); worker.drain_stderr(&config, &mut datastream); @@ -1999,6 +2002,85 @@ fn pump_network(driver: &mut IrohDriver, stack: &DistributionRuntimeStack) { driver.drain_outbox(&stack.outbox); } +fn emit_swim_telemetry( + datastream: &mut NodeDatastream, + stack: &DistributionRuntimeStack, + local_phase: &str, +) { + for transition in stack.drain_swim_transitions() { + let peer = format_dist_node_id(transition.peer); + let from = transition.from.map(|state| format!("{:?}", state)); + let to = format!("{:?}", transition.to); + let member_state = stack + .member_state(transition.peer) + .map(|state| format!("{:?}", state)); + let record = MembershipTransition { + peer, + from: from.unwrap_or_default(), + to, + reason: transition.reason.to_owned(), + last_ack_age_ms: transition.last_ack_age.map(duration_ms_u64), + consecutive_timeouts: transition.consecutive_timeouts, + recent_probe_targets: swim_recent_probe_targets(stack), + member_state, + }; + datastream + .producer + .submit_record(datastream.channels.membership, &record); + } + for event in stack.drain_swim_probe_events() { + let record = swim_probe_event_record(stack, event, local_phase); + datastream + .producer + .submit_record(datastream.channels.swim_probes, &record); + } +} + +fn swim_probe_event_record( + stack: &DistributionRuntimeStack, + event: ObservedProbeEvent, + local_phase: &str, +) -> SwimProbeEvent { + let config = &stack.swim_config; + let budget_ms = event.budget_ms; + SwimProbeEvent { + event: event.event.to_owned(), + target: format_dist_node_id(event.target), + sequence: event.sequence, + kind: event.kind.to_owned(), + rtt_ms: event.rtt_ms, + budget_ms, + budget_ticks: budget_ms, + last_ack_age_ms: event.last_ack_age.map(duration_ms_u64), + consecutive_timeouts: event.consecutive_timeouts, + recent_probe_targets: swim_recent_probe_targets(stack), + member_state: stack + .member_state(event.target) + .map(|state| format!("{:?}", state)), + local_phase: local_phase.to_owned(), + probe_interval_ms: duration_ms_u64(config.probe_interval), + probe_timeout_ms: duration_ms_u64(config.probe_timeout), + indirect_probes: u32::try_from(config.indirect_probes).unwrap_or(u32::MAX), + suspicion_timeout_ms: duration_ms_u64(config.suspicion_timeout), + dead_reprobe_interval_ms: duration_ms_u64(config.dead_reprobe_interval), + probe_mode: format!("{:?}", config.probe_mode), + lifeguard_enabled: config.lifeguard.is_some(), + } +} + +fn swim_recent_probe_targets(stack: &DistributionRuntimeStack) -> Vec { + stack + .swim_telemetry + .recent_targets() + .into_iter() + .map(format_dist_node_id) + .collect() +} + +fn format_dist_node_id(node_id: DistNodeId) -> String { + format!("{:?}", node_id) +} + fn node_datastream(config: &DeploymentConfig) -> NodeDatastream { NodeDatastream::new(config) } @@ -2011,6 +2093,8 @@ struct DatastreamChannelSet { worker_stderr: ChannelId, host_cpu: ChannelId, host_gpu: ChannelId, + membership: ChannelId, + swim_probes: ChannelId, host_net: ChannelId, arena: ChannelId, } @@ -2109,6 +2193,16 @@ impl NodeDatastream { &mut by_name, &mut by_id, ), + membership: register_record_channel::( + &producer, + &mut by_name, + &mut by_id, + ), + swim_probes: register_record_channel::( + &producer, + &mut by_name, + &mut by_id, + ), }; let archive = config.datastream_frame_log.as_deref().and_then(|path| { DatastreamArchive::open(path, endpoint.subscribe_all("frame-log")).ok() @@ -3773,6 +3867,7 @@ fn tokenizer_from_env() -> TokenizerSource { mod tests { use super::*; use distribution::swim::actor::MembershipChanged; + use distribution::swim::node::{SwimObservation, SwimObserver}; use mvp_system::actors::orchestrator::OrchestratorMsg; fn endpoint(seed: u8) -> EndpointAddr { @@ -3824,6 +3919,49 @@ mod tests { ); } + #[test] + fn worker_swim_telemetry_emits_probe_records_to_datastream() { + let stack = test_stack(); + let peer = DistNodeId([9; 32]); + stack.swim_telemetry.observe(SwimObservation::ProbeSent { + target: peer, + sequence: 41, + kind: "direct", + }); + stack + .swim_telemetry + .observe(SwimObservation::ProbeTimedOut { + target: peer, + sequence: 41, + kind: "direct", + budget_ticks: 15_000, + }); + let mut datastream = NodeDatastream::new(&test_config(None)); + + emit_swim_telemetry(&mut datastream, &stack, "unit_phase"); + + let frames = datastream.endpoint.mux().drain(); + let probe_records = frames + .iter() + .filter(|frame| { + datastream + .by_id + .get(&frame.channel) + .is_some_and(|channel| channel == SwimProbeEvent::CHANNEL) + }) + .map(|frame| SwimProbeEvent::decode(&frame.payload).expect("probe record decodes")) + .collect::>(); + + assert_eq!(probe_records.len(), 2); + assert_eq!(probe_records[0].event, "sent"); + assert_eq!(probe_records[0].sequence, 41); + assert_eq!(probe_records[0].local_phase, "unit_phase"); + assert_eq!(probe_records[0].probe_timeout_ms, 15_000); + assert_eq!(probe_records[1].event, "timed_out"); + assert_eq!(probe_records[1].budget_ms, Some(15_000)); + assert_eq!(probe_records[1].consecutive_timeouts, 1); + } + #[test] fn debug_join_client_serializes_endpoint_from_stdin() { let secret = iroh::SecretKey::from_bytes(&[7; 32]); diff --git a/crates/mvp-system/src/config.rs b/crates/mvp-system/src/config.rs index adbb5b0..c4be47f 100644 --- a/crates/mvp-system/src/config.rs +++ b/crates/mvp-system/src/config.rs @@ -109,6 +109,7 @@ pub struct VastAiConfig { pub max_dph_total: Option, pub min_reliability: Option, pub require_verified: Option, + pub blacklist_hosts: Vec, pub poll_interval_secs: Option, pub onstart: Option, pub ssh_identity: Option, @@ -134,6 +135,7 @@ pub struct ResolvedVastAiConfig { pub max_dph_total: Option, pub min_reliability: Option, pub require_verified: Option, + pub blacklist_hosts: Vec, pub onstart: Option, pub ssh_identity: Option, } @@ -215,6 +217,11 @@ impl ResolvedVastAiConfig { if let Some(require_verified) = self.require_verified { policy.require_verified = require_verified; } + for host_id in &self.blacklist_hosts { + if !policy.blacklist_hosts.contains(host_id) { + policy.blacklist_hosts.push(*host_id); + } + } policy } } @@ -267,6 +274,7 @@ mod tests { max_dph_total: Some(0.10), min_reliability: Some(0.98), require_verified: Some(true), + blacklist_hosts: vec![155385], onstart: None, ssh_identity: Some("~/.ssh/swactor_vastai_ed25519".to_owned()), } @@ -335,6 +343,7 @@ min_up_mbps = 50.25 max_dph_total = 0.10 min_reliability = 0.99 require_verified = true +blacklist_hosts = [155385, 59017] onstart = "echo preparing" ssh_identity = "~/.ssh/swactor_vastai_ed25519" poll_interval_secs = 30 @@ -412,6 +421,7 @@ poll_interval_secs = 30 assert_eq!(config.vastai.max_dph_total, Some(0.10)); assert_eq!(config.vastai.min_reliability, Some(0.99)); assert_eq!(config.vastai.require_verified, Some(true)); + assert_eq!(config.vastai.blacklist_hosts, vec![155385, 59017]); assert_eq!(config.vastai.poll_interval_secs, Some(30)); assert_eq!(config.vastai.onstart.as_deref(), Some("echo preparing")); assert_eq!( diff --git a/crates/mvp-system/src/distribution_stack.rs b/crates/mvp-system/src/distribution_stack.rs index 71f2e0f..e4bd10e 100644 --- a/crates/mvp-system/src/distribution_stack.rs +++ b/crates/mvp-system/src/distribution_stack.rs @@ -24,7 +24,8 @@ use distribution::node_metadata_actor::{MetadataActor, MetadataIn}; use distribution::registry_actor::{RegistryActor, RegistryIn}; use distribution::swim::actor::{MembershipChanged, SwimActor, SwimIn}; use distribution::swim::member_list::MemberList; -use distribution::swim::telemetry::{ObservedTransition, SwimTelemetry}; +use distribution::swim::probe::SwimConfig; +use distribution::swim::telemetry::{ObservedProbeEvent, ObservedTransition, SwimTelemetry}; use distribution::transport_bridge::{ Outbox, OutboxPeerDirectory, OutboxRouteBinder, RelayMirror, RouteView, RouteViewTransport, }; @@ -48,6 +49,7 @@ pub struct DistributionRuntimeStack { pub route_view: RouteView, pub membership_mirror: Arc>, pub swim_telemetry: Arc, + pub swim_config: SwimConfig, pub actors: DistributionActorAddrs, } @@ -80,13 +82,14 @@ impl DistributionRuntimeStack { Arc::clone(&transport_router), Arc::clone(&outbox), )); + let swim_config = config.swim.clone(); let swim_telemetry = SwimTelemetry::new(); let swim_addr = runtime .spawn( SwimActor::new( node_id, - config.swim.clone(), + swim_config.clone(), Instant::now(), peer_directory.clone(), ) @@ -152,6 +155,7 @@ impl DistributionRuntimeStack { route_view, membership_mirror, swim_telemetry, + swim_config, actors: DistributionActorAddrs { swim: swim_addr, registry: registry_addr, @@ -227,6 +231,10 @@ impl DistributionRuntimeStack { pub fn drain_swim_transitions(&self) -> Vec { self.swim_telemetry.drain_transitions() } + + pub fn drain_swim_probe_events(&self) -> Vec { + self.swim_telemetry.drain_probe_events() + } } struct MembershipFanout { diff --git a/crates/mvp-system/src/orchestrator_app.rs b/crates/mvp-system/src/orchestrator_app.rs index 64632b6..f0d67fc 100644 --- a/crates/mvp-system/src/orchestrator_app.rs +++ b/crates/mvp-system/src/orchestrator_app.rs @@ -52,10 +52,12 @@ use crate::vastai_provisioning::{ }; use datastream::{ ChannelContent, ChannelId, ChannelRef, DatastreamEndpoint, DatastreamEvent, DatastreamProducer, - DatastreamPublisherMsg, DatastreamSubscribe, Frame, Lifetime, NodeId, StreamDescriptor, + DatastreamPublisherMsg, DatastreamSubscribe, Frame, Lifetime, NodeId, Record, StreamDescriptor, StreamId, StreamOrigin, SubscriptionRequest, }; use distribution::node::DistributedNodeConfig; +use distribution::swim::telemetry::ObservedProbeEvent; +use distribution::telemetry::{MembershipTransition, SwimProbeEvent}; use distribution::types::{MemberState, NodeId as DistNodeId}; use iroh::EndpointAddr; use iroh_driver::{ @@ -759,9 +761,12 @@ struct CachedModelConfig { impl CachedModelConfig { fn from_host_path(provider: ProviderKind, requested: PathBuf) -> Result { - if !matches!(provider, ProviderKind::Process | ProviderKind::Docker) { + if !matches!( + provider, + ProviderKind::Process | ProviderKind::Docker | ProviderKind::VastAi + ) { return Err(format!( - "{CACHED_MODEL_HOST_ENV} is a host-local cache path and is only supported by provider=process or provider=docker" + "{CACHED_MODEL_HOST_ENV} is a host-local cache path and is only supported by provider=process, provider=docker, or provider=vastai planning" )); } let host_path = requested.canonicalize().map_err(|e| { @@ -1421,7 +1426,9 @@ impl ConfigBuilder { .transpose()?; let mut gguf_source = self.gguf_source.clone(); if let Some(cached_model) = &cached_model { - gguf_source = GgufSource::LocalPath(cached_model.worker_path(provider)); + if provider != ProviderKind::VastAi { + gguf_source = GgufSource::LocalPath(cached_model.worker_path(provider)); + } } let relay = relay_runtime_config_from_settings( self.run_id, @@ -2870,6 +2877,7 @@ fn wait_for_runtime_readies( expected_node_ids.first().copied().unwrap_or(0), stack, ); + emit_swim_probe_events(orch_datastream, dashboard, stack, "runtime_ready_wait"); drain_frames(frame_rx, dashboard, orch_datastream); drain_orch_stdio_capture( orch_stdio_rx, @@ -2981,6 +2989,7 @@ fn wait_for_weights_loaded_count( loop { pump(driver, stack, frame_tx); emit_swim_transitions(orch_datastream, dashboard, run_id, node_id, stack); + emit_swim_probe_events(orch_datastream, dashboard, stack, "weights_loaded_wait"); drain_orch_stdio_capture(orch_stdio_rx, orch_datastream, dashboard, run_id, node_id); if stop_requested(stop_rx) { return Err("shutdown requested while waiting for pipeline weights loaded".to_owned()); @@ -3029,7 +3038,9 @@ fn wait_for_weights_loaded_count( }), ); let route_owner = stack.route_owner(ready.node_actor); + let datastream_route_owner = stack.route_owner(ready.datastream_publisher); let member_state = stack.member_state(ready.swim_node_id); + let route_matches_ready = route_owner == Some(ready.swim_node_id); orch_datastream.emit_bootstrap_to_channel( dashboard, MVP_STAGE_ROUTE, @@ -3046,10 +3057,36 @@ fn wait_for_weights_loaded_count( "swim_node_id":format!("{:?}", ready.swim_node_id), "member_state":member_state.map(|state| format!("{:?}", state)), "route_owner":route_owner.map(|owner| format!("{:?}", owner)), - "datastream_route_owner":stack.route_owner(ready.datastream_publisher).map(|owner| format!("{:?}", owner)), - "route_matches_ready":route_owner == Some(ready.swim_node_id), + "datastream_route_owner":datastream_route_owner.map(|owner| format!("{:?}", owner)), + "route_matches_ready":route_matches_ready, }), ); + if member_state == Some(MemberState::Dead) { + let reason = format!( + "stage {} node {} is dead while loading pipeline weights", + stage.stage_index, stage_node_id + ); + orch_datastream.emit_bootstrap( + dashboard, + run_id, + node_id, + "stage_provision_wait", + "failed", + json!({ + "attempt":resend_attempt, + "stage_count":pipeline_plan.stages.len(), + "stage_index":stage.stage_index, + "stage_node_id":stage_node_id, + "stage_send_count":stage_send_count, + "loaded_stage_count":loaded_stages.len(), + "member_state":"Dead", + "route_owner":route_owner.map(|owner| format!("{:?}", owner)), + "datastream_route_owner":datastream_route_owner.map(|owner| format!("{:?}", owner)), + "reason":reason, + }), + ); + return Err(reason); + } if emit_wait_headline { orch_datastream.emit_bootstrap( dashboard, @@ -3337,6 +3374,8 @@ impl OrchDatastream { ] { out.channel_by_name(name); } + out.record_channel::(); + out.record_channel::(); Ok(out) } @@ -3355,6 +3394,16 @@ impl OrchDatastream { id } + fn record_channel(&mut self) -> ChannelId { + if let Some(id) = self.channels.get(R::CHANNEL).copied() { + return id; + } + let id = self.producer.register_record::(); + self.channels.insert(R::CHANNEL.to_owned(), id); + self.channel_names.insert(id, R::CHANNEL.to_owned()); + id + } + fn emit_event(&mut self, dashboard: Option<&DashboardSupport>, event: ProvisionEvent) { let payload = serde_json::to_vec(&MvpProvisionEventRecord::new(event)) .expect("serialize provisioning event"); @@ -3435,6 +3484,12 @@ impl OrchDatastream { self.emit_bytes(dashboard, MVP_ORCH_PROMPT, payload); } + fn emit_record(&mut self, dashboard: Option<&DashboardSupport>, record: &R) { + let id = self.record_channel::(); + self.producer.submit_record(id, record); + self.flush(dashboard, "orchestrator"); + } + fn emit_bytes( &mut self, dashboard: Option<&DashboardSupport>, @@ -4902,7 +4957,7 @@ fn drain_frames( &collected.frame, ); orch_datastream.archive_frame( - "node_cluster", + "node", &collected.stream, &collected.channel_name, &collected.frame, @@ -4929,6 +4984,15 @@ fn emit_swim_transitions( stack: &DistributionRuntimeStack, ) { for transition in stack.drain_swim_transitions() { + let peer = format_dist_node_id(transition.peer); + let from = transition.from.map(|state| format!("{:?}", state)); + let to = format!("{:?}", transition.to); + let member_state = stack + .member_state(transition.peer) + .map(|state| format!("{:?}", state)); + let last_ack_age_ms = transition.last_ack_age.map(duration_ms_u64); + let consecutive_timeouts = transition.consecutive_timeouts; + let recent_probe_targets = swim_recent_probe_targets(stack); orch_datastream.emit_bootstrap_to_channel( dashboard, MVP_SWIM_MEMBERSHIP, @@ -4937,15 +5001,93 @@ fn emit_swim_transitions( "membership_transition", "observed", json!({ - "peer":format!("{:?}", transition.peer), - "from":transition.from.map(|state| format!("{:?}", state)), - "to":format!("{:?}", transition.to), + "peer":peer.clone(), + "from":from.clone(), + "to":to.clone(), "reason":transition.reason, + "last_ack_age_ms":last_ack_age_ms, + "consecutive_timeouts":consecutive_timeouts, + "recent_probe_targets":recent_probe_targets.clone(), + "member_state":member_state.clone(), }), ); + orch_datastream.emit_record( + dashboard, + &MembershipTransition { + peer, + from: from.unwrap_or_default(), + to, + reason: transition.reason.to_owned(), + last_ack_age_ms, + consecutive_timeouts, + recent_probe_targets, + member_state, + }, + ); } } +fn emit_swim_probe_events( + orch_datastream: &mut OrchDatastream, + dashboard: Option<&DashboardSupport>, + stack: &DistributionRuntimeStack, + local_phase: &str, +) { + for event in stack.drain_swim_probe_events() { + let record = swim_probe_event_record(stack, event, local_phase); + orch_datastream.emit_record(dashboard, &record); + } +} + +fn swim_probe_event_record( + stack: &DistributionRuntimeStack, + event: ObservedProbeEvent, + local_phase: &str, +) -> SwimProbeEvent { + let config = &stack.swim_config; + let budget_ms = event.budget_ms; + SwimProbeEvent { + event: event.event.to_owned(), + target: format_dist_node_id(event.target), + sequence: event.sequence, + kind: event.kind.to_owned(), + rtt_ms: event.rtt_ms, + budget_ms, + budget_ticks: budget_ms, + last_ack_age_ms: event.last_ack_age.map(duration_ms_u64), + consecutive_timeouts: event.consecutive_timeouts, + recent_probe_targets: swim_recent_probe_targets(stack), + member_state: stack + .member_state(event.target) + .map(|state| format!("{:?}", state)), + local_phase: local_phase.to_owned(), + probe_interval_ms: duration_ms_u64(config.probe_interval), + probe_timeout_ms: duration_ms_u64(config.probe_timeout), + indirect_probes: u32::try_from(config.indirect_probes).unwrap_or(u32::MAX), + suspicion_timeout_ms: duration_ms_u64(config.suspicion_timeout), + dead_reprobe_interval_ms: duration_ms_u64(config.dead_reprobe_interval), + probe_mode: format!("{:?}", config.probe_mode), + lifeguard_enabled: config.lifeguard.is_some(), + } +} + +fn swim_recent_probe_targets(stack: &DistributionRuntimeStack) -> Vec { + stack + .swim_telemetry + .recent_targets() + .into_iter() + .map(format_dist_node_id) + .collect() +} + +fn format_dist_node_id(node_id: DistNodeId) -> String { + format!("{:?}", node_id) +} + +fn duration_ms_u64(duration: Duration) -> u64 { + u64::try_from(duration.as_millis()).unwrap_or(u64::MAX) +} + fn pump( driver: &mut IrohDriver, stack: &DistributionRuntimeStack, @@ -5187,6 +5329,7 @@ where #[cfg(test)] mod tests { use super::*; + use distribution::swim::node::{SwimObservation, SwimObserver}; use std::{ffi::OsString, path::PathBuf}; static ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); @@ -6453,6 +6596,69 @@ mod tests { ); } + #[test] + fn emit_swim_probe_events_archives_probe_lifecycle_once_with_config() { + static NEXT_TEMP_FILE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + + let suffix = NEXT_TEMP_FILE.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let path = std::env::temp_dir().join(format!( + "mvp-swim-probe-archive-test-{}-{suffix}.jsonl", + std::process::id() + )); + let _ = std::fs::remove_file(&path); + let stack = + DistributionRuntimeStack::new(DistNodeId([7; 32]), DistributedNodeConfig::default()); + let peer = DistNodeId([8; 32]); + stack.swim_telemetry.observe(SwimObservation::ProbeSent { + target: peer, + sequence: 99, + kind: "direct", + }); + stack + .swim_telemetry + .observe(SwimObservation::ProbeTimedOut { + target: peer, + sequence: 99, + kind: "direct", + budget_ticks: 15_000, + }); + let mut datastream = OrchDatastream::new(77, Some(&path)).expect("datastream opens"); + + emit_swim_probe_events(&mut datastream, None, &stack, "weights_loaded_wait"); + emit_swim_probe_events(&mut datastream, None, &stack, "weights_loaded_wait"); + drop(datastream); + + let contents = std::fs::read_to_string(&path).expect("read frame archive jsonl"); + let records = contents + .lines() + .map(|line| serde_json::from_str::(line).expect("archive line is json")) + .collect::>(); + let _ = std::fs::remove_file(&path); + let probe_payloads = records + .iter() + .filter(|record| record["channel"] == json!(SwimProbeEvent::CHANNEL)) + .map(|record| { + let value = record["payload"]["value"] + .as_str() + .expect("probe payload archived as utf8 json"); + serde_json::from_str::(value).expect("probe payload parses") + }) + .collect::>(); + + assert_eq!(probe_payloads.len(), 2); + assert_eq!(probe_payloads[0]["event"], json!("sent")); + assert_eq!(probe_payloads[0]["sequence"], json!(99)); + assert_eq!( + probe_payloads[0]["local_phase"], + json!("weights_loaded_wait") + ); + assert_eq!(probe_payloads[0]["probe_timeout_ms"], json!(15_000)); + assert_eq!(probe_payloads[1]["event"], json!("timed_out")); + assert_eq!(probe_payloads[1]["budget_ms"], json!(15_000)); + assert_eq!(probe_payloads[1]["budget_ticks"], json!(15_000)); + assert_eq!(probe_payloads[1]["consecutive_timeouts"], json!(1)); + } + #[test] fn orchestrator_stdio_drain_archives_stdout_and_stderr_as_provision_logs() { static NEXT_TEMP_FILE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); @@ -7466,34 +7672,91 @@ bootstrap_command = "/run" } #[test] - fn cached_model_with_deploy_provider_is_rejected_before_vastai_env_is_parsed() { - let model = TempModelFile::new("deploy-rejected.gguf"); - let error = with_clean_env_os( - &[ - ("MVP_RUNTIME_CONFIG", OsString::from("deploy")), - (CACHED_MODEL_HOST_ENV, model.raw_path.as_os_str().to_owned()), - ( - "MVP_VASTAI_CONFIRM_LEASE", - OsString::from("definitely-not-a-bool"), - ), - ("MVP_VASTAI_DISK_GB", OsString::from("not-a-u32")), - ], - || match Config::from_layers_with_path_and_args(None, std::iter::empty::()) { - Ok(_) => panic!("deploy cached model must be rejected"), - Err(error) => error, + fn vastai_cached_model_host_path_is_planning_only_and_keeps_remote_worker_source() { + let model = TempModelFile::with_metadata( + "Qwen2.5-7B-Instruct-Q4_K_M.gguf", + TestGgufMetadata { + num_layers: 28, + hidden_dim: 3584, + context_length: 32_768, + eos_token_id: 151_645, }, ); - assert!( - error.contains( - "MVP_CACHED_MODEL_HOST_PATH is a host-local cache path and is only supported by provider=process or provider=docker" - ), - "unexpected error: {error}" + let config = with_clean_env_os( + &[ + ("MVP_RUNTIME_CONFIG", OsString::from("deploy")), + (CACHED_MODEL_HOST_ENV, model.raw_path.as_os_str().to_owned()), + ("MVP_VASTAI_API_KEY", OsString::from("test-key")), + ], + || { + Config::from_layers_with_path_and_args( + None, + [ + "--provider", + "vastai", + "--pipeline-stages", + "4", + "--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-bootstrap-command", + "boot", + ] + .into_iter() + .map(str::to_owned), + ) + .expect("VastAI planning-cache config parses") + }, ); - assert!( - !error.contains("MVP_VASTAI_CONFIRM_LEASE") && !error.contains("MVP_VASTAI_DISK_GB"), - "cached-model rejection should not require valid VastAI env, got: {error}" + + assert_eq!(config.provider, ProviderKind::VastAi); + assert_eq!( + config + .cached_model + .as_ref() + .expect("VastAI planning cache retained") + .host_path, + model.canonical_path ); + assert_eq!( + config.gguf_source, + GgufSource::HuggingFaceGguf { + repo: "bartowski/Qwen2.5-7B-Instruct-GGUF".to_owned(), + file: "Qwen2.5-7B-Instruct-Q4_K_M.gguf".to_owned(), + revision: None, + } + ); + + let plan = config + .build_run_plan() + .expect("VastAI planning cache supplies local GGUF metadata"); + assert_eq!(plan.model.num_layers, 28); + assert_eq!(plan.stages.len(), 4); + + let coordinator = EndpointAddr::new(iroh::SecretKey::from_bytes(&[55; 32]).public()); + let orchestrator_actor = ActorAddress([56; 32]); + let specs = stage_node_specs(&config, Some(&plan), coordinator, orchestrator_actor) + .expect("VastAI stage specs build with planning cache"); + assert_eq!(specs.len(), 4); + for spec in specs { + assert!(spec.mounts.is_empty(), "VastAI stage specs must not mount"); + assert_eq!( + env_value(&spec.env, "MVP_GGUF_REPO"), + Some("bartowski/Qwen2.5-7B-Instruct-GGUF") + ); + assert_eq!( + env_value(&spec.env, "MVP_GGUF_FILE"), + Some("Qwen2.5-7B-Instruct-Q4_K_M.gguf") + ); + assert_eq!(env_value(&spec.env, "MVP_GGUF_LOCAL_PATH"), None); + assert_eq!(env_value(&spec.env, "MVP_PIPELINE_STAGES"), Some("4")); + } } #[derive(Default)] diff --git a/tools/vastai/src/teardown.rs b/tools/vastai/src/teardown.rs index 1e40140..a502ba5 100644 --- a/tools/vastai/src/teardown.rs +++ b/tools/vastai/src/teardown.rs @@ -1,7 +1,11 @@ use std::time::Duration; +use reqwest::StatusCode; + use crate::types::{InstanceInfo, InstanceListResponse, LabeledInstance}; +const DESTROY_RETRY_ATTEMPTS: u64 = 10; + /// Destroy one vast.ai instance by contract id. pub async fn destroy_instance( client: &reqwest::Client, @@ -19,6 +23,9 @@ pub async fn destroy_instance( .send() .await .map_err(|e| format!("destroy_instance request failed: {e}"))?; + if resp.status() == StatusCode::NOT_FOUND { + return Ok(()); + } if !resp.status().is_success() { let status = resp.status(); let body = resp.text().await.unwrap_or_default(); @@ -50,15 +57,40 @@ pub async fn destroy_instance_with_retry( api_key: &str, contract_id: u64, ) -> Result<(), String> { + destroy_instance_with_retry_policy( + client, + base_url, + api_key, + contract_id, + DESTROY_RETRY_ATTEMPTS, + Duration::from_millis(500), + Duration::from_secs(30), + ) + .await +} + +async fn destroy_instance_with_retry_policy( + client: &reqwest::Client, + base_url: &str, + api_key: &str, + contract_id: u64, + max_attempts: u64, + initial_backoff: Duration, + max_backoff: Duration, +) -> Result<(), String> { + let max_attempts = max_attempts.max(1); let mut attempt = 1_u64; loop { match destroy_instance(client, base_url, api_key, contract_id).await { Ok(()) => return Ok(()), + Err(error) if attempt >= max_attempts => { + return Err(format!( + "destroy_instance {contract_id} failed after {attempt} attempts: {error}" + )); + } Err(_) => { - let backoff = std::cmp::min( - Duration::from_millis(500_u64.saturating_mul(attempt)), - Duration::from_secs(30), - ); + let backoff = + std::cmp::min(initial_backoff.saturating_mul(attempt as u32), max_backoff); tokio::time::sleep(backoff).await; attempt = attempt.saturating_add(1); } @@ -116,3 +148,55 @@ pub async fn list_instances_by_label( out.sort_by_key(|i| i.contract_id); Ok(out) } + +#[cfg(test)] +mod tests { + use super::*; + use wiremock::matchers::{method, path}; + use wiremock::{Mock, MockServer, ResponseTemplate}; + + #[tokio::test] + async fn destroy_missing_contract_is_success() { + let server = MockServer::start().await; + Mock::given(method("DELETE")) + .and(path("/api/v0/instances/123/")) + .respond_with(ResponseTemplate::new(404).set_body_string("not found")) + .mount(&server) + .await; + + destroy_instance(&reqwest::Client::new(), &server.uri(), "secret", 123) + .await + .expect("destroy should be idempotent when the contract is already gone"); + } + + #[tokio::test] + async fn destroy_retry_returns_last_error_after_policy_exhausted() { + let server = MockServer::start().await; + Mock::given(method("DELETE")) + .and(path("/api/v0/instances/123/")) + .respond_with(ResponseTemplate::new(500).set_body_string("try later")) + .mount(&server) + .await; + + let error = destroy_instance_with_retry_policy( + &reqwest::Client::new(), + &server.uri(), + "secret", + 123, + 3, + Duration::from_millis(1), + Duration::from_millis(1), + ) + .await + .expect_err("persistent destroy failure should not retry forever"); + + assert!( + error.contains("failed after 3 attempts"), + "error should report retry exhaustion: {error}" + ); + assert!( + error.contains("HTTP 500"), + "error should preserve provider failure details: {error}" + ); + } +} diff --git a/xtask/src/main.rs b/xtask/src/main.rs index f236c5b..c4d29c3 100644 --- a/xtask/src/main.rs +++ b/xtask/src/main.rs @@ -18,11 +18,21 @@ struct TestStep { args: &'static [&'static str], } -const MVP_CHAT_CHECK_TIMEOUT_SECS: u64 = 1_800; +const MVP_CHAT_CHECK_TIMEOUT_SECS: u64 = 3_600; const MVP_CHAT_CHECK_POLL_MS: u64 = 100; const MVP_CHAT_CHECK_TERM_GRACE_MS: u64 = 30_000; const MVP_CHAT_CHECK_PROMPTS: &[u8] = b"ping\nsecond prompt\n"; const DATA_PATH_MIN_PAYLOAD_BYTES: u64 = 512; +const MVP_CHAT_CARGO_RUN_ARGS: &[&str] = &[ + "run", + "--package", + "mvp-system", + "--features", + "dashboard", + "--bin", + "mvp-chat", + "--", +]; struct MvpChatCheckPaths { root: PathBuf, @@ -337,7 +347,7 @@ fn run_mvp_chat(args: Vec) -> ExitCode { return ExitCode::SUCCESS; } let mut command = Command::new(cargo_bin()); - command.args(["run", "--package", "mvp-system", "--bin", "mvp-chat", "--"]); + command.args(MVP_CHAT_CARGO_RUN_ARGS); let dump_log_path = explicit_dump_log_path_from_mvp_chat_args(&forwarded); let run_id = run_id_from_mvp_chat_args(&forwarded); let benchmark_target = dump_log_path.as_deref().zip(run_id); @@ -347,7 +357,7 @@ fn run_mvp_chat(args: Vec) -> ExitCode { "started", json!({ "program": "cargo", - "args": ["run", "--package", "mvp-system", "--bin", "mvp-chat", "--"], + "args": MVP_CHAT_CARGO_RUN_ARGS, }), ); if let Err(error) = @@ -3686,6 +3696,22 @@ mod tests { values.iter().map(|value| (*value).to_owned()).collect() } + #[test] + fn mvp_chat_launcher_builds_dashboard_feature() { + assert!( + MVP_CHAT_CARGO_RUN_ARGS + .windows(2) + .any(|pair| pair[0] == "--features" && pair[1] == "dashboard"), + "{MVP_CHAT_CARGO_RUN_ARGS:?}" + ); + assert!( + MVP_CHAT_CARGO_RUN_ARGS + .windows(2) + .any(|pair| pair[0] == "--bin" && pair[1] == "mvp-chat"), + "{MVP_CHAT_CARGO_RUN_ARGS:?}" + ); + } + #[test] fn scenario_flags_select_expected_launch_contract() { let dump_log = Path::new("/tmp/mvp-chat-check.ndjson");