stash: another failed run, more telemetry
This commit is contained in:
parent
d3a67d021f
commit
545e39d2aa
10 changed files with 909 additions and 58 deletions
|
|
@ -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<MemberState>,
|
||||
pub to: MemberState,
|
||||
pub reason: &'static str,
|
||||
pub last_ack_age: Option<Duration>,
|
||||
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<u32>,
|
||||
pub budget_ms: Option<u64>,
|
||||
pub last_ack_age: Option<Duration>,
|
||||
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<Duration>,
|
||||
pub consecutive_timeouts: u32,
|
||||
}
|
||||
|
||||
#[derive(Default)]
|
||||
|
|
@ -51,6 +77,12 @@ struct Inner {
|
|||
rtts: VecDeque<u32>,
|
||||
/// Recent probe targets, newest last.
|
||||
targets: VecDeque<NodeId>,
|
||||
/// Last successful probe ack per peer.
|
||||
last_ack: HashMap<NodeId, Instant>,
|
||||
/// Consecutive direct/indirect timeouts per peer since the last ack.
|
||||
consecutive_timeouts: HashMap<NodeId, u32>,
|
||||
/// Probe events awaiting drain by telemetry consumers.
|
||||
probe_events: VecDeque<ObservedProbeEvent>,
|
||||
/// Transitions awaiting drain by the node's telemetry tick.
|
||||
transitions: VecDeque<ObservedTransition>,
|
||||
}
|
||||
|
|
@ -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<ObservedProbeEvent> {
|
||||
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<SwimTelemetry> {
|
||||
|
|
|
|||
|
|
@ -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<u64>,
|
||||
#[serde(default)]
|
||||
pub consecutive_timeouts: u32,
|
||||
#[serde(default)]
|
||||
pub recent_probe_targets: Vec<String>,
|
||||
#[serde(default)]
|
||||
pub member_state: Option<String>,
|
||||
}
|
||||
|
||||
/// 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<u32>,
|
||||
#[serde(default)]
|
||||
pub budget_ms: Option<u64>,
|
||||
/// Retained for compatibility with the existing diagnostic field name. The
|
||||
/// value is milliseconds in the wall-clock implementation.
|
||||
#[serde(default)]
|
||||
pub budget_ticks: Option<u64>,
|
||||
#[serde(default)]
|
||||
pub last_ack_age_ms: Option<u64>,
|
||||
#[serde(default)]
|
||||
pub consecutive_timeouts: u32,
|
||||
#[serde(default)]
|
||||
pub recent_probe_targets: Vec<String>,
|
||||
#[serde(default)]
|
||||
pub member_state: Option<String>,
|
||||
#[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;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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::<Vec<_>>();
|
||||
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());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -638,6 +638,7 @@ struct ChatVastAiConfig {
|
|||
max_dph_total: Option<f64>,
|
||||
min_reliability: Option<f64>,
|
||||
require_verified: Option<bool>,
|
||||
blacklist_hosts: Vec<u64>,
|
||||
disk_gb: Option<u32>,
|
||||
onstart: Option<String>,
|
||||
ssh_identity: Option<String>,
|
||||
|
|
@ -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,
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<String> {
|
||||
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::<MembershipTransition>(
|
||||
&producer,
|
||||
&mut by_name,
|
||||
&mut by_id,
|
||||
),
|
||||
swim_probes: register_record_channel::<SwimProbeEvent>(
|
||||
&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::<Vec<_>>();
|
||||
|
||||
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]);
|
||||
|
|
|
|||
|
|
@ -109,6 +109,7 @@ pub struct VastAiConfig {
|
|||
pub max_dph_total: Option<f64>,
|
||||
pub min_reliability: Option<f64>,
|
||||
pub require_verified: Option<bool>,
|
||||
pub blacklist_hosts: Vec<u64>,
|
||||
pub poll_interval_secs: Option<u64>,
|
||||
pub onstart: Option<String>,
|
||||
pub ssh_identity: Option<String>,
|
||||
|
|
@ -134,6 +135,7 @@ pub struct ResolvedVastAiConfig {
|
|||
pub max_dph_total: Option<f64>,
|
||||
pub min_reliability: Option<f64>,
|
||||
pub require_verified: Option<bool>,
|
||||
pub blacklist_hosts: Vec<u64>,
|
||||
pub onstart: Option<String>,
|
||||
pub ssh_identity: Option<String>,
|
||||
}
|
||||
|
|
@ -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!(
|
||||
|
|
|
|||
|
|
@ -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<Mutex<MemberList>>,
|
||||
pub swim_telemetry: Arc<SwimTelemetry>,
|
||||
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<ObservedTransition> {
|
||||
self.swim_telemetry.drain_transitions()
|
||||
}
|
||||
|
||||
pub fn drain_swim_probe_events(&self) -> Vec<ObservedProbeEvent> {
|
||||
self.swim_telemetry.drain_probe_events()
|
||||
}
|
||||
}
|
||||
|
||||
struct MembershipFanout {
|
||||
|
|
|
|||
|
|
@ -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<Self, String> {
|
||||
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::<MembershipTransition>();
|
||||
out.record_channel::<SwimProbeEvent>();
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
|
|
@ -3355,6 +3394,16 @@ impl OrchDatastream {
|
|||
id
|
||||
}
|
||||
|
||||
fn record_channel<R: Record>(&mut self) -> ChannelId {
|
||||
if let Some(id) = self.channels.get(R::CHANNEL).copied() {
|
||||
return id;
|
||||
}
|
||||
let id = self.producer.register_record::<R>();
|
||||
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<R: Record>(&mut self, dashboard: Option<&DashboardSupport>, record: &R) {
|
||||
let id = self.record_channel::<R>();
|
||||
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<String> {
|
||||
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::<Value>(line).expect("archive line is json"))
|
||||
.collect::<Vec<_>>();
|
||||
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>(value).expect("probe payload parses")
|
||||
})
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
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::<String>()) {
|
||||
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)]
|
||||
|
|
|
|||
|
|
@ -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}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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<String>) -> 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<String>) -> 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");
|
||||
|
|
|
|||
Loading…
Reference in a new issue