From 915d548e413d752787fb77a6ce11513b9eec559c Mon Sep 17 00:00:00 2001 From: Zachery Aaron Shores-Chmielewski Date: Wed, 24 Jun 2026 19:23:09 +0400 Subject: [PATCH] stash more pruning mvp engine --- Cargo.toml | 3 - crates/mvp-system/src/bin/local_e2e.rs | 191 ++++--- .../mvp-system/src/engine_builder/engine.rs | 299 +++++++++++ crates/mvp-system/src/engine_builder/error.rs | 159 ++++++ .../mvp-system/src/engine_builder/events.rs | 16 + .../mvp-system/src/engine_builder/launcher.rs | 170 ++++++ crates/mvp-system/src/engine_builder/mod.rs | 39 ++ crates/mvp-system/src/engine_builder/model.rs | 99 ++++ .../src/engine_builder/node_image.rs | 40 ++ .../mvp-system/src/engine_builder/planner.rs | 138 +++++ crates/mvp-system/src/engine_builder/pool.rs | 154 ++++++ crates/mvp-system/src/engine_builder/roles.rs | 56 ++ .../src/engine_builder/runtime_stack.rs | 183 +++++++ .../mvp-system/src/engine_builder/workload.rs | 13 + crates/mvp-system/src/lib.rs | 1 + .../src/tests/engine_builder_guarantees.rs | 201 +++++++ .../src/tests/local_mock/environment.rs | 93 +++- .../tests/local_mock_pipeline_integration.rs | 41 ++ crates/mvp-system/src/tests/mod.rs | 1 + .../tests/local_unmocked_mvp_e2e.rs | 12 + src/guarantees/RUNTIME_GUARANTEES.md | 90 ---- src/guarantees/correspondence.rs | 235 --------- src/guarantees/g4_lifecycle.rs | 422 --------------- src/guarantees/g5_fault_isolation.rs | 366 ------------- src/guarantees/mod.rs | 23 - src/guarantees/model_checker.rs | 215 -------- src/guarantees/stateright_death_orphan.rs | 494 ------------------ src/guarantees/stateright_lifecycle.rs | 352 ------------- src/lib.rs | 3 - src/worker.rs | 165 +++++- tests/actor_lifecycle.rs | 369 ++++++++++++- 31 files changed, 2343 insertions(+), 2300 deletions(-) create mode 100644 crates/mvp-system/src/engine_builder/engine.rs create mode 100644 crates/mvp-system/src/engine_builder/error.rs create mode 100644 crates/mvp-system/src/engine_builder/events.rs create mode 100644 crates/mvp-system/src/engine_builder/launcher.rs create mode 100644 crates/mvp-system/src/engine_builder/mod.rs create mode 100644 crates/mvp-system/src/engine_builder/model.rs create mode 100644 crates/mvp-system/src/engine_builder/node_image.rs create mode 100644 crates/mvp-system/src/engine_builder/planner.rs create mode 100644 crates/mvp-system/src/engine_builder/pool.rs create mode 100644 crates/mvp-system/src/engine_builder/roles.rs create mode 100644 crates/mvp-system/src/engine_builder/runtime_stack.rs create mode 100644 crates/mvp-system/src/engine_builder/workload.rs create mode 100644 crates/mvp-system/src/tests/engine_builder_guarantees.rs delete mode 100644 src/guarantees/RUNTIME_GUARANTEES.md delete mode 100644 src/guarantees/correspondence.rs delete mode 100644 src/guarantees/g4_lifecycle.rs delete mode 100644 src/guarantees/g5_fault_isolation.rs delete mode 100644 src/guarantees/mod.rs delete mode 100644 src/guarantees/model_checker.rs delete mode 100644 src/guarantees/stateright_death_orphan.rs delete mode 100644 src/guarantees/stateright_lifecycle.rs diff --git a/Cargo.toml b/Cargo.toml index 146c537..fd2b0aa 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -48,9 +48,6 @@ crossbeam-queue = "0.3.12" crossbeam-utils = "0.8.21" parking_lot = "0.12" -[lints.rust] -unexpected_cfgs = { level = "allow", check-cfg = ['cfg(kani)'] } - [dev-dependencies] criterion = { version = "0.5", features = ["html_reports"] } proptest = "1" diff --git a/crates/mvp-system/src/bin/local_e2e.rs b/crates/mvp-system/src/bin/local_e2e.rs index 860d0fb..19b8a41 100644 --- a/crates/mvp-system/src/bin/local_e2e.rs +++ b/crates/mvp-system/src/bin/local_e2e.rs @@ -21,6 +21,7 @@ use mvp_system::actors::register_mvp_actor_codecs; use mvp_system::dashboard::MvpDashboard; use mvp_system::distribution_stack::DistributionRuntimeStack; use mvp_system::driver_pumps; +use mvp_system::engine_builder as engine; use mvp_system::observability_surface as obs; use mvp_system::orchestrator_run_fsm as fsm; use mvp_system::run_plan as plan; @@ -208,43 +209,52 @@ fn run_supervisor_once( orchestrator_addr, ); + let topology = build_local_engine_topology(run_id)?; + let run_plan = topology.role_plan.run_plan.clone(); + let stage0 = run_plan + .stages + .iter() + .find(|stage| stage.stage_index == 0) + .cloned() + .ok_or_else(|| "engine builder did not assign stage 0".to_owned())?; + let stage1 = run_plan + .stages + .iter() + .find(|stage| stage.stage_index == 1) + .cloned() + .ok_or_else(|| "engine builder did not assign stage 1".to_owned())?; + let self_endpoint_json = serde_json::to_string(&driver.endpoint_addr()) .map_err(|e| format!("serialize endpoint addr: {e}"))?; let orchestrator_actor_json = serde_json::to_string(&orchestrator_addr) .map_err(|e| format!("serialize orchestrator actor: {e}"))?; let mut node1 = spawn_node_process( - NODE1_LOGICAL_ID, - 1, - 7001, + stage1.node_id.0, + stage1.stage_index, + stage1.inbound_edge.0, token_out_addr, &self_endpoint_json, &orchestrator_actor_json, )?; let mut node0 = spawn_node_process( - NODE0_LOGICAL_ID, - 0, - 7000, + stage0.node_id.0, + stage0.stage_index, + stage0.inbound_edge.0, node1.ready.token_in_addr, &self_endpoint_json, &orchestrator_actor_json, )?; - record_dashboard_event( - &mut dashboard, - node_event(NODE1_LOGICAL_ID, obs::EventKind::NodeStarted), - ); - record_dashboard_event( - &mut dashboard, - node_event(NODE1_LOGICAL_ID, obs::EventKind::NodeAvailable), - ); - record_dashboard_event( - &mut dashboard, - node_event(NODE0_LOGICAL_ID, obs::EventKind::NodeStarted), - ); - record_dashboard_event( - &mut dashboard, - node_event(NODE0_LOGICAL_ID, obs::EventKind::NodeAvailable), - ); + for stage in [&stage1, &stage0] { + record_dashboard_event( + &mut dashboard, + node_event(stage.node_id.0, obs::EventKind::NodeStarted), + ); + record_dashboard_event( + &mut dashboard, + node_event(stage.node_id.0, obs::EventKind::NodeAvailable), + ); + } let started_at = Instant::now(); wait_for_routes( @@ -254,13 +264,16 @@ fn run_supervisor_once( Duration::from_secs(20), )?; - let run_plan = two_stage_plan(run_id)?; stack .runtime .send_to( orchestrator_addr, OrchestratorMsg::ObservePoolReady { - nodes: vec![NODE0_LOGICAL_ID, NODE1_LOGICAL_ID], + nodes: run_plan + .stages + .iter() + .map(|stage| stage.node_id.0) + .collect(), }, ) .map_err(|e| format!("observe pool ready: {e}"))?; @@ -271,16 +284,14 @@ fn run_supervisor_once( orchestrator_addr, OrchestratorMsg::ObservePlanAvailable { run_id, - stages: vec![ - StageRefWire { - stage_index: 0, - node_id: NODE0_LOGICAL_ID, - }, - StageRefWire { - stage_index: 1, - node_id: NODE1_LOGICAL_ID, - }, - ], + stages: run_plan + .stages + .iter() + .map(|stage| StageRefWire { + stage_index: stage.stage_index, + node_id: stage.node_id.0, + }) + .collect(), }, ) .map_err(|e| format!("observe plan: {e}"))?; @@ -432,6 +443,19 @@ fn run_supervisor_once( if injected && completed && torn_down && sent_stop_to_node0 && sent_stop_to_node1 { shutdown_node(&mut node0); shutdown_node(&mut node1); + let builder_stage_assignments = topology + .events + .iter() + .filter(|event| { + matches!( + event, + engine::EngineEvent::RoleAssigned { + role: engine::RoleKind::StageWorker { .. }, + .. + } + ) + }) + .count(); let summary = json!({ "ok": true, "actor_plane": "iroh-swactor", @@ -453,6 +477,10 @@ fn run_supervisor_once( "run_completed_observed": completed, "run_torn_down_observed": torn_down, "stop_sent_to_all_nodes": sent_stop_to_node0 && sent_stop_to_node1, + "engine_builder_pattern": "pool-first-static-launcher", + "engine_builder_event_count": topology.events.len(), + "engine_builder_node_count": topology.node_summaries.len(), + "engine_builder_stage_assignments": builder_stage_assignments, "node_route_count": stack.route_view.read().map(|view| view.len()).unwrap_or_default(), "stage_ready_stdout_count": stage_ready_count, }); @@ -668,39 +696,72 @@ fn wait_for_routes( Err("directory routes for node actors did not converge".to_owned()) } -fn two_stage_plan(run_id: u64) -> Result { - plan::plan_run(plan::PlannerInput { - run_id: plan::RunId(run_id), - orchestrator_node_id: plan::NodeId(ORCHESTRATOR_LOGICAL_NODE_ID), - model: plan::ModelFacts { - model_id: "local-e2e-fixture".to_owned(), - num_layers: 4, - hidden_dim: 8, - dtype_family: plan::DTypeFamily::BFloat, - dtype_width_bytes: 2, - max_seq_len: 8, - eos_token_id: 99, - }, - runtime: plan::RuntimeConfig { max_tokens: 1 }, - candidate_pool: vec![ - plan::NodeId(NODE0_LOGICAL_ID), - plan::NodeId(NODE1_LOGICAL_ID), - ], - stage_count: 2, - placement: plan::PlacementInput::FixedLinear(vec![ - plan::StagePlacement { - stage_index: 0, - node_id: plan::NodeId(NODE0_LOGICAL_ID), +struct LocalEngineTopology { + role_plan: engine::RoleAssignmentPlan, + events: Vec, + node_summaries: Vec, +} + +fn build_local_engine_topology(run_id: u64) -> Result { + let cluster = engine::ClusterBuilder::new( + "local-process-e2e", + engine::ModelSpec::pipelined_causal_llm( + "local-e2e-fixture", + engine::ModelArtifact::TestTinyLlm { + path: "local-process://local-e2e-fixture".to_owned(), }, - plan::StagePlacement { - stage_index: 1, - node_id: plan::NodeId(NODE1_LOGICAL_ID), - }, - ]), - activation_ring: plan::RingSpec::test_default_activation(), - token_ring: plan::RingSpec::test_default_token(), + 4, + 8, + engine::DTypeFamily::BFloat, + 2, + 8, + 99, + ), + ) + .run_id(run_id) + .image( + engine::NodeImageSpec::new("mvp-local-e2e") + .worker_runtime(engine::WorkerRuntimeSpec::DumbProcess), + ) + .pool_provider(engine::StaticPoolProvider::new(vec![ + engine::NodeLease::new( + "orchestrator", + engine::NodeId(ORCHESTRATOR_LOGICAL_NODE_ID), + [engine::NodeCapability::Coordinator], + ) + .resources(engine::ResourceFacts::cpu_only(2, 2 << 30)), + engine::NodeLease::new( + "node0", + engine::NodeId(NODE0_LOGICAL_ID), + [engine::NodeCapability::Worker], + ) + .resources(engine::ResourceFacts::cpu_only(2, 2 << 30)), + engine::NodeLease::new( + "node1", + engine::NodeId(NODE1_LOGICAL_ID), + [engine::NodeCapability::Worker], + ) + .resources(engine::ResourceFacts::cpu_only(2, 2 << 30)), + ])) + .launcher(engine::StaticNodeLauncher) + .planner( + engine::FixedLinearPipelinePlanner::new(2).runtime(plan::RuntimeConfig { + max_tokens: MAX_TOKENS as u32, + }), + ) + .launch() + .map_err(|e| format!("local engine builder launch: {e}"))?; + let role_plan = cluster.role_plan().clone(); + let events = cluster.events().to_vec(); + let node_summaries = cluster.node_summaries(); + cluster + .shutdown() + .map_err(|e| format!("local engine builder shutdown: {e}"))?; + Ok(LocalEngineTopology { + role_plan, + events, + node_summaries, }) - .map_err(|e| format!("plan rejected: {:?}", e.kind())) } fn stage_provision_wire( diff --git a/crates/mvp-system/src/engine_builder/engine.rs b/crates/mvp-system/src/engine_builder/engine.rs new file mode 100644 index 0000000..adede87 --- /dev/null +++ b/crates/mvp-system/src/engine_builder/engine.rs @@ -0,0 +1,299 @@ +use std::collections::BTreeMap; +use std::time::Duration; + +use crate::run_plan::RunId; + +use super::error::EngineBuildError; +use super::events::EngineEvent; +use super::launcher::{ + LaunchedNode, NodeControl, NodeFacts, NodeLaunchSpec, NodeLauncher, SeedSpec, +}; +use super::model::ModelSpec; +use super::node_image::NodeImageSpec; +use super::planner::{RoleAssignmentPlan, RolePlanner, RolePlannerInput}; +use super::pool::{PoolProvider, PoolRequest, ResourceRequest}; +use super::roles::{RoleAssignment, RoleKind}; + +pub struct ClusterBuilder { + cluster_id: String, + run_id: RunId, + model: ModelSpec, + image: Option, + pool_provider: Option>, + launcher: Option>, + planner: Option>, + required_resources: ResourceRequest, + boot_timeout: Duration, + convergence_timeout: Duration, +} + +impl ClusterBuilder { + pub fn new(cluster_id: impl Into, model: ModelSpec) -> Self { + Self { + cluster_id: cluster_id.into(), + run_id: RunId(1), + model, + image: None, + pool_provider: None, + launcher: None, + planner: None, + required_resources: ResourceRequest::default(), + boot_timeout: Duration::from_secs(30), + convergence_timeout: Duration::from_secs(60), + } + } + + pub fn run_id(mut self, run_id: impl Into) -> Self { + self.run_id = run_id.into(); + self + } + + pub fn image(mut self, image: NodeImageSpec) -> Self { + self.image = Some(image); + self + } + + pub fn pool_provider(mut self, provider: impl PoolProvider + 'static) -> Self { + self.pool_provider = Some(Box::new(provider)); + self + } + + pub fn launcher(mut self, launcher: impl NodeLauncher + 'static) -> Self { + self.launcher = Some(Box::new(launcher)); + self + } + + pub fn planner(mut self, planner: impl RolePlanner + 'static) -> Self { + self.planner = Some(Box::new(planner)); + self + } + + pub fn required_resources(mut self, required_resources: ResourceRequest) -> Self { + self.required_resources = required_resources; + self + } + + pub fn boot_timeout(mut self, timeout: Duration) -> Self { + self.boot_timeout = timeout; + self + } + + pub fn convergence_timeout(mut self, timeout: Duration) -> Self { + self.convergence_timeout = timeout; + self + } + + pub fn launch(mut self) -> Result { + let image = self + .image + .take() + .ok_or(EngineBuildError::MissingComponent("image"))?; + let pool_provider = self + .pool_provider + .take() + .ok_or(EngineBuildError::MissingComponent("pool_provider"))?; + let launcher = self + .launcher + .take() + .ok_or(EngineBuildError::MissingComponent("launcher"))?; + let planner = self + .planner + .take() + .ok_or(EngineBuildError::MissingComponent("planner"))?; + + let mut events = Vec::new(); + let leases = pool_provider.acquire_pool(PoolRequest { + cluster_id: self.cluster_id.clone(), + min_nodes: planner.required_node_count(), + image: image.clone(), + required_resources: self.required_resources.clone(), + })?; + if leases.is_empty() { + return Err(EngineBuildError::EmptyPool); + } + events.push(EngineEvent::PoolAcquired { + node_count: leases.len(), + }); + + let mut nodes = Vec::with_capacity(leases.len()); + let mut iter = leases.into_iter(); + let seed_lease = iter.next().ok_or(EngineBuildError::EmptyPool)?; + let mut seed = launcher.launch_node( + &seed_lease, + NodeLaunchSpec { + cluster_id: self.cluster_id.clone(), + image: image.clone(), + seed: None, + is_seed: true, + env: BTreeMap::new(), + }, + )?; + events.push(EngineEvent::NodeLaunched { + node_id: seed.lease.logical_node_id, + seed: true, + }); + let seed_facts = seed.control.wait_boot_ready(self.boot_timeout)?; + events.push(EngineEvent::NodeBootReady { + node_id: seed_facts.node_id, + }); + let seed_endpoint = + seed_facts + .seed_endpoint + .clone() + .ok_or(EngineBuildError::SeedEndpointMissing { + node_id: seed_facts.node_id.0, + })?; + nodes.push(EngineNode::new(seed, seed_facts)); + + for lease in iter { + let mut node = launcher.launch_node( + &lease, + NodeLaunchSpec { + cluster_id: self.cluster_id.clone(), + image: image.clone(), + seed: Some(SeedSpec { + endpoint: seed_endpoint.clone(), + }), + is_seed: false, + env: BTreeMap::new(), + }, + )?; + events.push(EngineEvent::NodeLaunched { + node_id: node.lease.logical_node_id, + seed: false, + }); + let facts = node.control.wait_boot_ready(self.boot_timeout)?; + events.push(EngineEvent::NodeBootReady { + node_id: facts.node_id, + }); + nodes.push(EngineNode::new(node, facts)); + } + + let expected_alive = nodes.len(); + for node in &mut nodes { + node.control + .wait_cluster_converged(expected_alive, self.convergence_timeout)?; + } + events.push(EngineEvent::ClusterConverged { + node_count: expected_alive, + }); + + let plan = planner.plan(RolePlannerInput { + cluster_id: self.cluster_id.clone(), + run_id: self.run_id, + model: self.model, + nodes: nodes.iter().map(|node| node.facts.clone()).collect(), + })?; + events.push(EngineEvent::RolesPlanned { + stage_count: plan.stages.len(), + }); + + assign_role( + &mut nodes, + RoleAssignment::Coordinator(plan.coordinator.clone()), + &mut events, + )?; + for stage in &plan.stages { + assign_role( + &mut nodes, + RoleAssignment::StageWorker(stage.clone()), + &mut events, + )?; + } + events.push(EngineEvent::EngineReady { + cluster_id: self.cluster_id.clone(), + }); + + Ok(ClusterHandle { + cluster_id: self.cluster_id, + nodes, + plan, + events, + }) + } +} + +pub struct ClusterHandle { + cluster_id: String, + nodes: Vec, + plan: RoleAssignmentPlan, + events: Vec, +} + +impl ClusterHandle { + pub fn cluster_id(&self) -> &str { + &self.cluster_id + } + + pub fn role_plan(&self) -> &RoleAssignmentPlan { + &self.plan + } + + pub fn events(&self) -> &[EngineEvent] { + &self.events + } + + pub fn node_summaries(&self) -> Vec { + self.nodes + .iter() + .map(|node| NodeSummary { + node_id: node.facts.node_id, + roles: node.roles.iter().map(RoleAssignment::kind).collect(), + facts: node.facts.clone(), + }) + .collect() + } + + pub fn shutdown(mut self) -> Result, EngineBuildError> { + for node in &mut self.nodes { + node.control.shutdown()?; + self.events.push(EngineEvent::NodeStopped { + node_id: node.facts.node_id, + }); + } + self.events.push(EngineEvent::ShutdownComplete { + cluster_id: self.cluster_id, + }); + Ok(self.events) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct NodeSummary { + pub node_id: crate::run_plan::NodeId, + pub roles: Vec, + pub facts: NodeFacts, +} + +struct EngineNode { + facts: NodeFacts, + roles: Vec, + control: Box, +} + +impl EngineNode { + fn new(launched: LaunchedNode, facts: NodeFacts) -> Self { + Self { + facts, + roles: Vec::new(), + control: launched.control, + } + } +} + +fn assign_role( + nodes: &mut [EngineNode], + assignment: RoleAssignment, + events: &mut Vec, +) -> Result<(), EngineBuildError> { + let node_id = assignment.node_id(); + let node = nodes + .iter_mut() + .find(|node| node.facts.node_id == node_id) + .ok_or(EngineBuildError::RoleTargetMissing { node_id: node_id.0 })?; + node.control.assign_role(assignment.clone())?; + let role = assignment.kind(); + node.roles.push(assignment); + events.push(EngineEvent::RoleAssigned { node_id, role }); + Ok(()) +} diff --git a/crates/mvp-system/src/engine_builder/error.rs b/crates/mvp-system/src/engine_builder/error.rs new file mode 100644 index 0000000..dff7f22 --- /dev/null +++ b/crates/mvp-system/src/engine_builder/error.rs @@ -0,0 +1,159 @@ +use std::error::Error; +use std::fmt; + +use crate::run_plan; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum EngineBuildError { + MissingComponent(&'static str), + EmptyPool, + SeedEndpointMissing { node_id: u64 }, + RoleTargetMissing { node_id: u64 }, + Pool(PoolError), + Launch(LaunchError), + Node(NodeControlError), + Planning(PlanningError), +} + +impl fmt::Display for EngineBuildError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::MissingComponent(name) => write!(f, "missing engine builder component: {name}"), + Self::EmptyPool => write!(f, "pool provider returned no nodes"), + Self::SeedEndpointMissing { node_id } => { + write!(f, "seed node {node_id} did not report a seed endpoint") + } + Self::RoleTargetMissing { node_id } => { + write!(f, "role assignment targeted unknown node {node_id}") + } + Self::Pool(err) => err.fmt(f), + Self::Launch(err) => err.fmt(f), + Self::Node(err) => err.fmt(f), + Self::Planning(err) => err.fmt(f), + } + } +} + +impl Error for EngineBuildError {} + +impl From for EngineBuildError { + fn from(value: PoolError) -> Self { + Self::Pool(value) + } +} + +impl From for EngineBuildError { + fn from(value: LaunchError) -> Self { + Self::Launch(value) + } +} + +impl From for EngineBuildError { + fn from(value: NodeControlError) -> Self { + Self::Node(value) + } +} + +impl From for EngineBuildError { + fn from(value: PlanningError) -> Self { + Self::Planning(value) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum PoolError { + InsufficientNodes { requested: usize, available: usize }, + Provider(String), +} + +impl fmt::Display for PoolError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::InsufficientNodes { + requested, + available, + } => write!( + f, + "pool has {available} matching nodes, but {requested} were requested" + ), + Self::Provider(message) => write!(f, "pool provider failed: {message}"), + } + } +} + +impl Error for PoolError {} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum LaunchError { + Backend(String), +} + +impl fmt::Display for LaunchError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::Backend(message) => write!(f, "node launcher failed: {message}"), + } + } +} + +impl Error for LaunchError {} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum NodeControlError { + NotBooted { node_id: u64 }, + Stopped { node_id: u64 }, + RoleNodeMismatch { node_id: u64, role_node_id: u64 }, + Backend(String), +} + +impl fmt::Display for NodeControlError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::NotBooted { node_id } => write!(f, "node {node_id} is not boot-ready"), + Self::Stopped { node_id } => write!(f, "node {node_id} is already stopped"), + Self::RoleNodeMismatch { + node_id, + role_node_id, + } => write!( + f, + "node {node_id} cannot accept role targeted at node {role_node_id}" + ), + Self::Backend(message) => write!(f, "node control failed: {message}"), + } + } +} + +impl Error for NodeControlError {} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum PlanningError { + DuplicateNodeId { node_id: u64 }, + NoCoordinatorCandidate, + InsufficientWorkers { required: usize, available: usize }, + ModelRejected(run_plan::PlanRejectionKind), + StageProjection(run_plan::ProjectionRejection), +} + +impl fmt::Display for PlanningError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::DuplicateNodeId { node_id } => { + write!(f, "planner input contained duplicate node id {node_id}") + } + Self::NoCoordinatorCandidate => write!(f, "no coordinator-capable node available"), + Self::InsufficientWorkers { + required, + available, + } => write!( + f, + "planner needs {required} worker nodes, but only {available} are available" + ), + Self::ModelRejected(kind) => write!(f, "model/run planner rejected input: {kind:?}"), + Self::StageProjection(kind) => { + write!(f, "stage provision projection failed: {kind:?}") + } + } + } +} + +impl Error for PlanningError {} diff --git a/crates/mvp-system/src/engine_builder/events.rs b/crates/mvp-system/src/engine_builder/events.rs new file mode 100644 index 0000000..d8c1510 --- /dev/null +++ b/crates/mvp-system/src/engine_builder/events.rs @@ -0,0 +1,16 @@ +use crate::run_plan::NodeId; + +use super::roles::RoleKind; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum EngineEvent { + PoolAcquired { node_count: usize }, + NodeLaunched { node_id: NodeId, seed: bool }, + NodeBootReady { node_id: NodeId }, + ClusterConverged { node_count: usize }, + RolesPlanned { stage_count: usize }, + RoleAssigned { node_id: NodeId, role: RoleKind }, + EngineReady { cluster_id: String }, + NodeStopped { node_id: NodeId }, + ShutdownComplete { cluster_id: String }, +} diff --git a/crates/mvp-system/src/engine_builder/launcher.rs b/crates/mvp-system/src/engine_builder/launcher.rs new file mode 100644 index 0000000..1d7cbfa --- /dev/null +++ b/crates/mvp-system/src/engine_builder/launcher.rs @@ -0,0 +1,170 @@ +use std::collections::{BTreeMap, BTreeSet}; +use std::time::Duration; + +use crate::run_plan::NodeId; + +use super::error::{LaunchError, NodeControlError}; +use super::node_image::NodeImageSpec; +use super::pool::{NodeCapability, NodeLease, ResourceFacts}; +use super::roles::RoleAssignment; + +pub trait NodeLauncher: Send + Sync { + fn launch_node( + &self, + lease: &NodeLease, + spec: NodeLaunchSpec, + ) -> Result; +} + +pub trait NodeControl: Send { + fn wait_boot_ready(&mut self, timeout: Duration) -> Result; + fn wait_cluster_converged( + &mut self, + expected_alive: usize, + timeout: Duration, + ) -> Result<(), NodeControlError>; + fn assign_role(&mut self, assignment: RoleAssignment) -> Result<(), NodeControlError>; + fn shutdown(&mut self) -> Result<(), NodeControlError>; +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct NodeLaunchSpec { + pub cluster_id: String, + pub image: NodeImageSpec, + pub seed: Option, + pub is_seed: bool, + pub env: BTreeMap, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct SeedSpec { + pub endpoint: String, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct NodeFacts { + pub node_id: NodeId, + pub seed_endpoint: Option, + pub resources: ResourceFacts, + pub capabilities: BTreeSet, +} + +impl NodeFacts { + pub fn from_lease(lease: &NodeLease, seed_endpoint: Option) -> Self { + Self { + node_id: lease.logical_node_id, + seed_endpoint, + resources: lease.expected_resources.clone(), + capabilities: lease.capabilities.clone(), + } + } +} + +pub struct LaunchedNode { + pub lease: NodeLease, + pub control: Box, +} + +#[derive(Clone, Debug, Default)] +pub struct StaticNodeLauncher; + +impl NodeLauncher for StaticNodeLauncher { + fn launch_node( + &self, + lease: &NodeLease, + spec: NodeLaunchSpec, + ) -> Result { + let endpoint = format!( + "static://{}/node/{}", + spec.cluster_id, lease.logical_node_id.0 + ); + let facts = NodeFacts::from_lease(lease, Some(endpoint)); + Ok(LaunchedNode { + lease: lease.clone(), + control: Box::new(StaticNodeControl { + facts, + booted: false, + stopped: false, + assigned_roles: Vec::new(), + }), + }) + } +} + +struct StaticNodeControl { + facts: NodeFacts, + booted: bool, + stopped: bool, + assigned_roles: Vec, +} + +impl StaticNodeControl { + fn node_id(&self) -> u64 { + self.facts.node_id.0 + } +} + +impl NodeControl for StaticNodeControl { + fn wait_boot_ready(&mut self, _timeout: Duration) -> Result { + if self.stopped { + return Err(NodeControlError::Stopped { + node_id: self.node_id(), + }); + } + self.booted = true; + Ok(self.facts.clone()) + } + + fn wait_cluster_converged( + &mut self, + expected_alive: usize, + _timeout: Duration, + ) -> Result<(), NodeControlError> { + if self.stopped { + return Err(NodeControlError::Stopped { + node_id: self.node_id(), + }); + } + if !self.booted { + return Err(NodeControlError::NotBooted { + node_id: self.node_id(), + }); + } + if expected_alive == 0 { + return Err(NodeControlError::Backend( + "expected_alive must be greater than zero".to_owned(), + )); + } + Ok(()) + } + + fn assign_role(&mut self, assignment: RoleAssignment) -> Result<(), NodeControlError> { + if self.stopped { + return Err(NodeControlError::Stopped { + node_id: self.node_id(), + }); + } + if !self.booted { + return Err(NodeControlError::NotBooted { + node_id: self.node_id(), + }); + } + let role_node_id = assignment.node_id().0; + if role_node_id != self.node_id() { + return Err(NodeControlError::RoleNodeMismatch { + node_id: self.node_id(), + role_node_id, + }); + } + self.assigned_roles.push(assignment); + Ok(()) + } + + fn shutdown(&mut self) -> Result<(), NodeControlError> { + if self.stopped { + return Ok(()); + } + self.stopped = true; + Ok(()) + } +} diff --git a/crates/mvp-system/src/engine_builder/mod.rs b/crates/mvp-system/src/engine_builder/mod.rs new file mode 100644 index 0000000..ab338f5 --- /dev/null +++ b/crates/mvp-system/src/engine_builder/mod.rs @@ -0,0 +1,39 @@ +//! Pool-based engine/node builder primitives. +//! +//! This module owns topology construction: acquire a role-neutral node pool, +//! launch the same node image everywhere, wait for node/cluster readiness, map a +//! model onto discovered nodes, assign roles, and return a live cluster handle. +//! Workload semantics stay outside this module; see [`WorkloadAdapter`]. + +pub mod engine; +pub mod error; +pub mod events; +pub mod launcher; +pub mod model; +pub mod node_image; +pub mod planner; +pub mod pool; +pub mod roles; +#[cfg(feature = "local-e2e")] +pub mod runtime_stack; +pub mod workload; + +pub use crate::run_plan::{NodeId, RunId}; +pub use engine::{ClusterBuilder, ClusterHandle, NodeSummary}; +pub use error::{EngineBuildError, LaunchError, NodeControlError, PlanningError, PoolError}; +pub use events::EngineEvent; +pub use launcher::{ + LaunchedNode, NodeControl, NodeFacts, NodeLaunchSpec, NodeLauncher, SeedSpec, + StaticNodeLauncher, +}; +pub use model::{DTypeFamily, ModelArchitecture, ModelArtifact, ModelSpec}; +pub use node_image::{NodeImageSpec, WorkerRuntimeSpec}; +pub use planner::{FixedLinearPipelinePlanner, RoleAssignmentPlan, RolePlanner, RolePlannerInput}; +pub use pool::{ + LaunchTarget, NodeCapability, NodeLease, PoolProvider, PoolRequest, ResourceFacts, + ResourceRequest, StaticPoolProvider, +}; +pub use roles::{CoordinatorAssignment, RoleAssignment, RoleKind, StageAssignment}; +#[cfg(feature = "local-e2e")] +pub use runtime_stack::{RuntimeNode, RuntimeNodeConfig, RuntimeNodeError}; +pub use workload::WorkloadAdapter; diff --git a/crates/mvp-system/src/engine_builder/model.rs b/crates/mvp-system/src/engine_builder/model.rs new file mode 100644 index 0000000..3ee89bb --- /dev/null +++ b/crates/mvp-system/src/engine_builder/model.rs @@ -0,0 +1,99 @@ +use crate::run_plan; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ModelSpec { + pub model_id: String, + pub architecture: ModelArchitecture, + pub artifact: ModelArtifact, + pub num_layers: u32, + pub hidden_dim: u64, + pub dtype_family: DTypeFamily, + pub dtype_width_bytes: u64, + pub max_seq_len: u64, + pub eos_token_id: u32, +} + +impl ModelSpec { + pub fn pipelined_causal_llm( + model_id: impl Into, + artifact: ModelArtifact, + num_layers: u32, + hidden_dim: u64, + dtype_family: DTypeFamily, + dtype_width_bytes: u64, + max_seq_len: u64, + eos_token_id: u32, + ) -> Self { + Self { + model_id: model_id.into(), + architecture: ModelArchitecture::PipelinedCausalLlm, + artifact, + num_layers, + hidden_dim, + dtype_family, + dtype_width_bytes, + max_seq_len, + eos_token_id, + } + } + + pub fn mvp_tiny_open_llm_fixture() -> Self { + Self::pipelined_causal_llm( + "mvp-tiny-open-llm-fixture", + ModelArtifact::ContainerPath { + path: "/models/mvp-tiny-open-llm.gguf".to_owned(), + }, + 4, + 8, + DTypeFamily::BFloat, + 2, + 8, + 99, + ) + } + + pub fn to_run_plan_facts(&self) -> run_plan::ModelFacts { + run_plan::ModelFacts { + model_id: self.model_id.clone(), + num_layers: self.num_layers, + hidden_dim: self.hidden_dim, + dtype_family: self.dtype_family.into(), + dtype_width_bytes: self.dtype_width_bytes, + max_seq_len: self.max_seq_len, + eos_token_id: self.eos_token_id, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ModelArchitecture { + PipelinedCausalLlm, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum ModelArtifact { + ContainerPath { + path: String, + }, + HuggingFaceGguf { + repo: String, + file: String, + revision: Option, + }, + TestTinyLlm { + path: String, + }, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum DTypeFamily { + BFloat, +} + +impl From for run_plan::DTypeFamily { + fn from(value: DTypeFamily) -> Self { + match value { + DTypeFamily::BFloat => Self::BFloat, + } + } +} diff --git a/crates/mvp-system/src/engine_builder/node_image.rs b/crates/mvp-system/src/engine_builder/node_image.rs new file mode 100644 index 0000000..eb62429 --- /dev/null +++ b/crates/mvp-system/src/engine_builder/node_image.rs @@ -0,0 +1,40 @@ +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct NodeImageSpec { + pub image: String, + pub binary: String, + pub worker_runtime: WorkerRuntimeSpec, +} + +impl NodeImageSpec { + pub fn new(image: impl Into) -> Self { + Self { + image: image.into(), + binary: "mvp-node".to_owned(), + worker_runtime: WorkerRuntimeSpec::External { + name: "node-image-default".to_owned(), + }, + } + } + + pub fn binary(mut self, binary: impl Into) -> Self { + self.binary = binary.into(); + self + } + + pub fn worker_runtime(mut self, worker_runtime: WorkerRuntimeSpec) -> Self { + self.worker_runtime = worker_runtime; + self + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum WorkerRuntimeSpec { + DumbProcess, + TinygradCuda { + worker_script: String, + device_env: String, + }, + External { + name: String, + }, +} diff --git a/crates/mvp-system/src/engine_builder/planner.rs b/crates/mvp-system/src/engine_builder/planner.rs new file mode 100644 index 0000000..fc266da --- /dev/null +++ b/crates/mvp-system/src/engine_builder/planner.rs @@ -0,0 +1,138 @@ +use std::collections::BTreeSet; + +use crate::run_plan::{self, NodeId, RunId}; + +use super::error::PlanningError; +use super::launcher::NodeFacts; +use super::model::ModelSpec; +use super::pool::NodeCapability; +use super::roles::{CoordinatorAssignment, StageAssignment}; + +pub trait RolePlanner: Send + Sync { + fn required_node_count(&self) -> usize; + fn plan(&self, input: RolePlannerInput) -> Result; +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RolePlannerInput { + pub cluster_id: String, + pub run_id: RunId, + pub model: ModelSpec, + pub nodes: Vec, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RoleAssignmentPlan { + pub coordinator: CoordinatorAssignment, + pub stages: Vec, + pub run_plan: run_plan::RunPlan, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct FixedLinearPipelinePlanner { + pub stage_count: u32, + pub runtime: run_plan::RuntimeConfig, + pub activation_ring: run_plan::RingSpec, + pub token_ring: run_plan::RingSpec, +} + +impl FixedLinearPipelinePlanner { + pub fn new(stage_count: u32) -> Self { + Self { + stage_count, + runtime: run_plan::RuntimeConfig::test_default(), + activation_ring: run_plan::RingSpec::test_default_activation(), + token_ring: run_plan::RingSpec::test_default_token(), + } + } + + pub fn runtime(mut self, runtime: run_plan::RuntimeConfig) -> Self { + self.runtime = runtime; + self + } +} + +impl RolePlanner for FixedLinearPipelinePlanner { + fn required_node_count(&self) -> usize { + self.stage_count as usize + 1 + } + + fn plan(&self, input: RolePlannerInput) -> Result { + reject_duplicate_nodes(&input.nodes)?; + let coordinator = input + .nodes + .iter() + .find(|node| node.capabilities.contains(&NodeCapability::Coordinator)) + .ok_or(PlanningError::NoCoordinatorCandidate)?; + let workers = input + .nodes + .iter() + .filter(|node| { + node.node_id != coordinator.node_id + && node.capabilities.contains(&NodeCapability::Worker) + }) + .collect::>(); + let required = self.stage_count as usize; + if workers.len() < required { + return Err(PlanningError::InsufficientWorkers { + required, + available: workers.len(), + }); + } + + let placements = workers + .iter() + .take(required) + .enumerate() + .map(|(stage_index, node)| run_plan::StagePlacement { + stage_index: stage_index as u32, + node_id: node.node_id, + }) + .collect::>(); + let candidate_pool = workers.iter().map(|node| node.node_id).collect::>(); + let run_plan = run_plan::plan_run(run_plan::PlannerInput { + run_id: input.run_id, + orchestrator_node_id: coordinator.node_id, + model: input.model.to_run_plan_facts(), + runtime: self.runtime.clone(), + candidate_pool, + stage_count: self.stage_count, + placement: run_plan::PlacementInput::FixedLinear(placements), + activation_ring: self.activation_ring, + token_ring: self.token_ring, + }) + .map_err(|err| PlanningError::ModelRejected(err.kind()))?; + + let mut stages = Vec::with_capacity(self.stage_count as usize); + for stage_index in 0..self.stage_count { + let provision = run_plan::derive_stage_provision(&run_plan, stage_index) + .map_err(PlanningError::StageProjection)?; + stages.push(StageAssignment { + cluster_id: input.cluster_id.clone(), + provision, + }); + } + + Ok(RoleAssignmentPlan { + coordinator: CoordinatorAssignment { + cluster_id: input.cluster_id, + node_id: coordinator.node_id, + model: input.model, + }, + stages, + run_plan, + }) + } +} + +fn reject_duplicate_nodes(nodes: &[NodeFacts]) -> Result<(), PlanningError> { + let mut seen = BTreeSet::::new(); + for node in nodes { + if !seen.insert(node.node_id) { + return Err(PlanningError::DuplicateNodeId { + node_id: node.node_id.0, + }); + } + } + Ok(()) +} diff --git a/crates/mvp-system/src/engine_builder/pool.rs b/crates/mvp-system/src/engine_builder/pool.rs new file mode 100644 index 0000000..fab56fa --- /dev/null +++ b/crates/mvp-system/src/engine_builder/pool.rs @@ -0,0 +1,154 @@ +use std::collections::BTreeSet; + +use crate::run_plan::NodeId; + +use super::error::PoolError; +use super::node_image::NodeImageSpec; + +pub trait PoolProvider: Send + Sync { + fn acquire_pool(&self, request: PoolRequest) -> Result, PoolError>; +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct PoolRequest { + pub cluster_id: String, + pub min_nodes: usize, + pub image: NodeImageSpec, + pub required_resources: ResourceRequest, +} + +#[derive(Clone, Debug, PartialEq, Eq, Default)] +pub struct ResourceRequest { + pub min_gpu_count: u32, + pub min_gpu_memory_bytes: u64, + pub require_cuda: bool, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct NodeLease { + pub lease_id: String, + pub logical_node_id: NodeId, + pub launch_target: LaunchTarget, + pub expected_resources: ResourceFacts, + pub capabilities: BTreeSet, +} + +impl NodeLease { + pub fn new( + lease_id: impl Into, + logical_node_id: NodeId, + capabilities: impl IntoIterator, + ) -> Self { + Self { + lease_id: lease_id.into(), + logical_node_id, + launch_target: LaunchTarget::InProcess, + expected_resources: ResourceFacts::default(), + capabilities: capabilities.into_iter().collect(), + } + } + + pub fn launch_target(mut self, launch_target: LaunchTarget) -> Self { + self.launch_target = launch_target; + self + } + + pub fn resources(mut self, expected_resources: ResourceFacts) -> Self { + self.expected_resources = expected_resources; + self + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum LaunchTarget { + InProcess, + LocalProcess { program: String, args: Vec }, + DockerContainer { name: String }, + RemoteHost { label: String }, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub enum NodeCapability { + Coordinator, + Worker, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct ResourceFacts { + pub gpu_count: u32, + pub gpu_memory_bytes: u64, + pub cpu_cores: u32, + pub ram_bytes: u64, + pub cuda_available: bool, +} + +impl ResourceFacts { + pub fn cpu_only(cpu_cores: u32, ram_bytes: u64) -> Self { + Self { + gpu_count: 0, + gpu_memory_bytes: 0, + cpu_cores, + ram_bytes, + cuda_available: false, + } + } + + pub fn cuda(gpu_count: u32, gpu_memory_bytes: u64, cpu_cores: u32, ram_bytes: u64) -> Self { + Self { + gpu_count, + gpu_memory_bytes, + cpu_cores, + ram_bytes, + cuda_available: true, + } + } + + fn satisfies(&self, request: &ResourceRequest) -> bool { + self.gpu_count >= request.min_gpu_count + && self.gpu_memory_bytes >= request.min_gpu_memory_bytes + && (!request.require_cuda || self.cuda_available) + } +} + +impl Default for ResourceFacts { + fn default() -> Self { + Self::cpu_only(1, 512 * 1024 * 1024) + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct StaticPoolProvider { + leases: Vec, +} + +impl StaticPoolProvider { + pub fn new(leases: Vec) -> Self { + Self { leases } + } + + pub fn leases(&self) -> &[NodeLease] { + &self.leases + } +} + +impl PoolProvider for StaticPoolProvider { + fn acquire_pool(&self, request: PoolRequest) -> Result, PoolError> { + let matching = self + .leases + .iter() + .filter(|lease| { + lease + .expected_resources + .satisfies(&request.required_resources) + }) + .cloned() + .collect::>(); + if matching.len() < request.min_nodes { + return Err(PoolError::InsufficientNodes { + requested: request.min_nodes, + available: matching.len(), + }); + } + Ok(matching) + } +} diff --git a/crates/mvp-system/src/engine_builder/roles.rs b/crates/mvp-system/src/engine_builder/roles.rs new file mode 100644 index 0000000..fd0c94a --- /dev/null +++ b/crates/mvp-system/src/engine_builder/roles.rs @@ -0,0 +1,56 @@ +use crate::run_plan::{self, NodeId}; + +use super::model::ModelSpec; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct CoordinatorAssignment { + pub cluster_id: String, + pub node_id: NodeId, + pub model: ModelSpec, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct StageAssignment { + pub cluster_id: String, + pub provision: run_plan::ProvisionStage, +} + +impl StageAssignment { + pub fn node_id(&self) -> NodeId { + self.provision.node_id + } + + pub fn stage_index(&self) -> u32 { + self.provision.stage_index + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum RoleAssignment { + Coordinator(CoordinatorAssignment), + StageWorker(StageAssignment), +} + +impl RoleAssignment { + pub fn node_id(&self) -> NodeId { + match self { + Self::Coordinator(assignment) => assignment.node_id, + Self::StageWorker(assignment) => assignment.node_id(), + } + } + + pub fn kind(&self) -> RoleKind { + match self { + Self::Coordinator(_) => RoleKind::Coordinator, + Self::StageWorker(assignment) => RoleKind::StageWorker { + stage_index: assignment.stage_index(), + }, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum RoleKind { + Coordinator, + StageWorker { stage_index: u32 }, +} diff --git a/crates/mvp-system/src/engine_builder/runtime_stack.rs b/crates/mvp-system/src/engine_builder/runtime_stack.rs new file mode 100644 index 0000000..447ba73 --- /dev/null +++ b/crates/mvp-system/src/engine_builder/runtime_stack.rs @@ -0,0 +1,183 @@ +use std::thread; +use std::time::{Duration, Instant}; + +use distribution::node::DistributedNodeConfig; +use distribution::types::{DirectoryEntry, NodeId}; +use iroh::EndpointAddr; +use iroh_driver::{IrohDriver, IrohDriverConfig}; +use swactor::actor::ActorAddress; + +use crate::actors::node_agent::NodeAgentActor; +use crate::actors::orchestrator::OrchestratorActor; +use crate::actors::register_mvp_actor_codecs; +use crate::distribution_stack::DistributionRuntimeStack; +use crate::orchestrator_run_fsm as orchestrator_core; +use crate::stage_controller as stage_core; + +pub struct RuntimeNodeConfig { + pub distributed: DistributedNodeConfig, + pub relay_mode: iroh::RelayMode, +} + +impl Default for RuntimeNodeConfig { + fn default() -> Self { + Self { + distributed: DistributedNodeConfig::default(), + relay_mode: iroh::RelayMode::Disabled, + } + } +} + +pub struct RuntimeNode { + _tokio: tokio::runtime::Runtime, + driver: IrohDriver, + stack: DistributionRuntimeStack, +} + +impl RuntimeNode { + pub fn start_default() -> Result { + Self::start_with_codecs(RuntimeNodeConfig::default(), |_| {}) + } + + pub fn start_with_codecs( + config: RuntimeNodeConfig, + extend_codecs: impl FnOnce(&mut swactor_transport::CodecRegistry), + ) -> Result { + let tokio = tokio::runtime::Runtime::new() + .map_err(|err| RuntimeNodeError::Start(format!("tokio runtime: {err}")))?; + let mut driver = IrohDriver::with_handle( + tokio.handle().clone(), + IrohDriverConfig { + secret_key: None, + relay_mode: config.relay_mode, + node: config.distributed.clone(), + peer_auth: None, + additional_alpns: vec![], + }, + ) + .map_err(|err| RuntimeNodeError::Start(format!("iroh driver: {err}")))?; + let stack = DistributionRuntimeStack::new_with_codecs( + driver.node_id(), + config.distributed, + |registry| { + register_mvp_actor_codecs(registry); + extend_codecs(registry); + }, + ); + driver.enable_actor_bridge( + stack.runtime.clone(), + stack.codec.clone(), + stack.actor_bridge_routes(), + stack.actors.swim, + stack.relay_mirror.clone(), + stack.route_view.clone(), + ); + Ok(Self { + _tokio: tokio, + driver, + stack, + }) + } + + pub fn node_id(&self) -> NodeId { + self.driver.node_id() + } + + pub fn endpoint_addr(&self) -> EndpointAddr { + self.driver.endpoint_addr() + } + + pub fn join(&mut self, seeds: &[EndpointAddr]) { + self.driver.join(seeds); + } + + pub fn register_actor_route(&mut self, actor_addr: ActorAddress, generation: u64) { + let entry = self.driver.register_actor(actor_addr, generation); + self.stack.register_local_actor(entry); + } + + pub fn register_directory_entry(&self, entry: DirectoryEntry) { + self.stack.register_local_actor(entry); + } + + pub fn spawn_orchestrator_actor( + &mut self, + config: orchestrator_core::RunConfig, + report_to: Option, + ) -> Result { + let actor = self + .stack + .runtime + .spawn(OrchestratorActor::new(config, report_to)) + .map_err(|err| RuntimeNodeError::Start(format!("spawn orchestrator actor: {err}")))?; + self.register_actor_route(actor, 1); + Ok(actor) + } + + pub fn spawn_node_agent_actor( + &mut self, + local_node_id: stage_core::NodeId, + orchestrator: ActorAddress, + report_to: Option, + ) -> Result { + let actor = self + .stack + .runtime + .spawn(NodeAgentActor::new(local_node_id, orchestrator, report_to)) + .map_err(|err| RuntimeNodeError::Start(format!("spawn node agent actor: {err}")))?; + self.register_actor_route(actor, 1); + Ok(actor) + } + + pub fn pump_once(&mut self) { + self.stack.tick_protocol_actors(Instant::now()); + self.driver.pump_inbound_to_actors(); + self.stack.pump_runtime_once(); + self.driver.drain_outbox(&self.stack.outbox); + } + + pub fn wait_for_routes( + &mut self, + actors: &[ActorAddress], + timeout: Duration, + ) -> Result<(), RuntimeNodeError> { + let started = Instant::now(); + while started.elapsed() < timeout { + self.pump_once(); + let ready = self + .stack + .route_view + .read() + .map(|view| actors.iter().all(|actor| view.contains_key(actor))) + .unwrap_or(false); + if ready { + return Ok(()); + } + thread::sleep(Duration::from_millis(20)); + } + Err(RuntimeNodeError::Convergence( + "directory routes did not converge".to_owned(), + )) + } + + pub fn alive_count(&self) -> usize { + self.stack.alive_count() + } +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum RuntimeNodeError { + Start(String), + Convergence(String), +} + +impl std::fmt::Display for RuntimeNodeError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Start(message) => write!(f, "runtime node start failed: {message}"), + Self::Convergence(message) => write!(f, "runtime node convergence failed: {message}"), + } + } +} + +impl std::error::Error for RuntimeNodeError {} diff --git a/crates/mvp-system/src/engine_builder/workload.rs b/crates/mvp-system/src/engine_builder/workload.rs new file mode 100644 index 0000000..fe36905 --- /dev/null +++ b/crates/mvp-system/src/engine_builder/workload.rs @@ -0,0 +1,13 @@ +use super::engine::ClusterHandle; + +pub trait WorkloadAdapter { + type Input; + type Output; + type Error; + + fn submit( + &self, + cluster: &mut ClusterHandle, + input: Self::Input, + ) -> Result; +} diff --git a/crates/mvp-system/src/lib.rs b/crates/mvp-system/src/lib.rs index 733a8b7..325f89d 100644 --- a/crates/mvp-system/src/lib.rs +++ b/crates/mvp-system/src/lib.rs @@ -11,6 +11,7 @@ pub mod device_bridge; pub mod distribution_stack; pub mod driver_pumps; pub mod edge_establisher; +pub mod engine_builder; pub mod gpu_worker_ctl; pub mod gpu_worker_egress_producer; pub mod gpu_worker_ingress_parser; diff --git a/crates/mvp-system/src/tests/engine_builder_guarantees.rs b/crates/mvp-system/src/tests/engine_builder_guarantees.rs new file mode 100644 index 0000000..ff1ff92 --- /dev/null +++ b/crates/mvp-system/src/tests/engine_builder_guarantees.rs @@ -0,0 +1,201 @@ +//! Black-box contract tests for the pool-based engine builder. +//! +//! The builder contract is topology/lifecycle only: it acquires a neutral node +//! pool, launches the same node image, waits for readiness/convergence, lets a +//! planner assign roles, and stays agnostic to workload input semantics. + +use mvp_system::engine_builder as engine; +use mvp_system::engine_builder::WorkloadAdapter; + +fn model() -> engine::ModelSpec { + engine::ModelSpec::mvp_tiny_open_llm_fixture() +} + +fn image() -> engine::NodeImageSpec { + engine::NodeImageSpec::new("mvp-node:cuda").worker_runtime( + engine::WorkerRuntimeSpec::TinygradCuda { + worker_script: "/opt/mvp/mvp_tinygrad_worker.py".to_owned(), + device_env: "CUDA".to_owned(), + }, + ) +} + +fn full_pool() -> engine::StaticPoolProvider { + engine::StaticPoolProvider::new(vec![ + engine::NodeLease::new( + "coordinator", + engine::NodeId(900), + [engine::NodeCapability::Coordinator], + ) + .resources(engine::ResourceFacts::cpu_only(2, 2 << 30)), + engine::NodeLease::new( + "worker-0", + engine::NodeId(11), + [engine::NodeCapability::Worker], + ) + .resources(engine::ResourceFacts::cuda(1, 8 << 30, 4, 8 << 30)), + engine::NodeLease::new( + "worker-1", + engine::NodeId(12), + [engine::NodeCapability::Worker], + ) + .resources(engine::ResourceFacts::cuda(1, 8 << 30, 4, 8 << 30)), + ]) +} + +#[test] +fn builder_launches_pool_converges_and_assigns_planned_roles() { + let cluster = engine::ClusterBuilder::new("cluster-a", model()) + .run_id(77) + .image(image()) + .pool_provider(full_pool()) + .launcher(engine::StaticNodeLauncher) + .planner(engine::FixedLinearPipelinePlanner::new(2)) + .launch() + .expect("launch cluster"); + + let summaries = cluster.node_summaries(); + let coordinator = summaries + .iter() + .find(|node| node.node_id == engine::NodeId(900)) + .expect("coordinator summary"); + assert_eq!(coordinator.roles, vec![engine::RoleKind::Coordinator]); + let worker0 = summaries + .iter() + .find(|node| node.node_id == engine::NodeId(11)) + .expect("worker 0 summary"); + assert_eq!( + worker0.roles, + vec![engine::RoleKind::StageWorker { stage_index: 0 }] + ); + let worker1 = summaries + .iter() + .find(|node| node.node_id == engine::NodeId(12)) + .expect("worker 1 summary"); + assert_eq!( + worker1.roles, + vec![engine::RoleKind::StageWorker { stage_index: 1 }] + ); + + let plan = cluster.role_plan(); + assert_eq!(plan.run_plan.stages.len(), 2); + assert_eq!(plan.run_plan.edges.len(), 3); + assert_eq!(plan.run_plan.stages[0].layer_start, 0); + assert_eq!(plan.run_plan.stages[0].layer_end_exclusive, 2); + assert_eq!(plan.run_plan.stages[1].layer_start, 2); + assert_eq!(plan.run_plan.stages[1].layer_end_exclusive, 4); + assert!( + cluster + .events() + .contains(&engine::EngineEvent::EngineReady { + cluster_id: "cluster-a".to_owned() + }) + ); + + let shutdown_events = cluster.shutdown().expect("shutdown cluster"); + assert!( + shutdown_events.contains(&engine::EngineEvent::ShutdownComplete { + cluster_id: "cluster-a".to_owned() + }) + ); +} + +#[test] +fn planner_rejects_when_pool_cannot_supply_stage_workers() { + let short_pool = engine::StaticPoolProvider::new(vec![ + engine::NodeLease::new( + "coordinator", + engine::NodeId(900), + [engine::NodeCapability::Coordinator], + ), + engine::NodeLease::new( + "worker-0", + engine::NodeId(11), + [engine::NodeCapability::Worker], + ), + engine::NodeLease::new( + "observer", + engine::NodeId(901), + [engine::NodeCapability::Coordinator], + ), + ]); + + let result = engine::ClusterBuilder::new("cluster-short", model()) + .run_id(77) + .image(image()) + .pool_provider(short_pool) + .launcher(engine::StaticNodeLauncher) + .planner(engine::FixedLinearPipelinePlanner::new(2)) + .launch(); + + match result { + Err(engine::EngineBuildError::Planning(engine::PlanningError::InsufficientWorkers { + required, + available, + })) => { + assert_eq!(required, 2); + assert_eq!(available, 1); + } + Ok(_) => panic!("expected insufficient worker planning error, got launched cluster"), + Err(other) => panic!("unexpected error: {other:?}"), + } +} + +struct ProbeWorkload; + +impl engine::WorkloadAdapter for ProbeWorkload { + type Input = Vec<&'static str>; + type Output = usize; + type Error = std::convert::Infallible; + + fn submit( + &self, + _cluster: &mut engine::ClusterHandle, + input: Self::Input, + ) -> Result { + Ok(input.len()) + } +} + +#[test] +fn workload_input_semantics_live_outside_the_cluster_builder() { + let mut cluster = engine::ClusterBuilder::new("cluster-opaque", model()) + .run_id(77) + .image(image()) + .pool_provider(full_pool()) + .launcher(engine::StaticNodeLauncher) + .planner(engine::FixedLinearPipelinePlanner::new(2)) + .launch() + .expect("launch cluster"); + + let observed = ProbeWorkload + .submit(&mut cluster, vec!["not", "tokens"]) + .expect("submit probe workload"); + assert_eq!(observed, 2); + + cluster.shutdown().expect("shutdown cluster"); +} + +#[cfg(feature = "local-e2e")] +#[test] +fn runtime_node_builds_the_reusable_iroh_swactor_stack() { + let mut node = engine::RuntimeNode::start_default().expect("start runtime node"); + + let orchestrator = node + .spawn_orchestrator_actor( + mvp_system::orchestrator_run_fsm::RunConfig { + run_id: mvp_system::orchestrator_run_fsm::RunId(77), + max_tokens: 1, + prompt: vec![1, 2, 3], + }, + None, + ) + .expect("spawn orchestrator actor"); + let worker = node + .spawn_node_agent_actor(mvp_system::stage_controller::NodeId(11), orchestrator, None) + .expect("spawn node agent actor"); + + node.wait_for_routes(&[orchestrator, worker], std::time::Duration::from_secs(1)) + .expect("local actor routes"); + assert_eq!(node.node_id().0, *node.endpoint_addr().id.as_bytes()); +} diff --git a/crates/mvp-system/src/tests/local_mock/environment.rs b/crates/mvp-system/src/tests/local_mock/environment.rs index f9d7cdb..463618b 100644 --- a/crates/mvp-system/src/tests/local_mock/environment.rs +++ b/crates/mvp-system/src/tests/local_mock/environment.rs @@ -1,5 +1,6 @@ use std::collections::{BTreeMap, BTreeSet}; +use mvp_system::engine_builder as engine; use mvp_system::observability_surface as obs; use mvp_system::orchestrator_run_fsm as fsm; use mvp_system::run_plan as plan; @@ -29,6 +30,7 @@ pub struct LocalMockCluster { run_id: plan::RunId, orchestrator_node_id: plan::NodeId, max_tokens: u32, + engine_events: Vec, plan: plan::RunPlan, nodes: BTreeMap, orchestrator: Option, @@ -44,6 +46,7 @@ pub struct LocalMockCluster { #[derive(Clone, Debug)] pub struct LocalMockOutcome { pub trace: Vec, + pub engine_events: Vec, pub injected_sequences: Vec, pub stage_count: usize, pub live_edges: usize, @@ -97,6 +100,29 @@ impl ResourceTracker { } } +fn mock_pool(orchestrator_node_id: plan::NodeId, stage_count: u32) -> engine::StaticPoolProvider { + let mut leases = Vec::with_capacity(stage_count as usize + 1); + leases.push( + engine::NodeLease::new( + "mock-coordinator", + engine::NodeId(orchestrator_node_id.0), + [engine::NodeCapability::Coordinator], + ) + .resources(engine::ResourceFacts::cpu_only(2, 2 << 30)), + ); + for stage_index in 0..stage_count { + leases.push( + engine::NodeLease::new( + format!("mock-worker-{stage_index}"), + engine::NodeId(11 + u64::from(stage_index)), + [engine::NodeCapability::Worker], + ) + .resources(engine::ResourceFacts::cpu_only(2, 2 << 30)), + ); + } + engine::StaticPoolProvider::new(leases) +} + impl LocalMockCluster { pub fn two_stage() -> Self { Self::with_config(LocalMockConfig::default()) @@ -111,34 +137,43 @@ impl LocalMockCluster { let run_id = plan::RunId(77); let orchestrator_node_id = plan::NodeId(900); - let stages = (0..config.stage_count) - .map(|stage_index| plan::StagePlacement { - stage_index, - node_id: plan::NodeId(11 + u64::from(stage_index)), - }) - .collect::>(); - let plan = plan::plan_run(plan::PlannerInput { - run_id, - orchestrator_node_id, - model: plan::ModelFacts { - model_id: "mock-gguf".to_owned(), - num_layers: config.stage_count * 2, - hidden_dim: 8, - dtype_family: plan::DTypeFamily::BFloat, - dtype_width_bytes: 2, - max_seq_len: 8, - eos_token_id: 99, - }, - runtime: plan::RuntimeConfig { - max_tokens: config.max_tokens, - }, - candidate_pool: stages.iter().map(|stage| stage.node_id).collect(), - stage_count: config.stage_count, - placement: plan::PlacementInput::FixedLinear(stages), - activation_ring: plan::RingSpec::test_default_activation(), - token_ring: plan::RingSpec::test_default_token(), - }) - .expect("mock plan must be valid"); + let engine_cluster = engine::ClusterBuilder::new( + "local-mock", + engine::ModelSpec::pipelined_causal_llm( + "mock-gguf", + engine::ModelArtifact::TestTinyLlm { + path: "local-mock://mock-gguf".to_owned(), + }, + config.stage_count * 2, + 8, + engine::DTypeFamily::BFloat, + 2, + 8, + 99, + ), + ) + .run_id(run_id.0) + .image( + engine::NodeImageSpec::new("local-mock-node") + .worker_runtime(engine::WorkerRuntimeSpec::DumbProcess), + ) + .pool_provider(mock_pool(orchestrator_node_id, config.stage_count)) + .launcher(engine::StaticNodeLauncher) + .planner( + engine::FixedLinearPipelinePlanner::new(config.stage_count).runtime( + plan::RuntimeConfig { + max_tokens: config.max_tokens, + }, + ), + ) + .launch() + .expect("local mock engine builder must launch"); + let engine_events = engine_cluster.events().to_vec(); + let role_plan = engine_cluster.role_plan().clone(); + engine_cluster + .shutdown() + .expect("local mock engine builder must shutdown"); + let plan = role_plan.run_plan; let nodes = plan .stages .iter() @@ -157,6 +192,7 @@ impl LocalMockCluster { Self { run_id, orchestrator_node_id, + engine_events, max_tokens: config.max_tokens, plan, nodes, @@ -752,6 +788,7 @@ impl LocalMockCluster { fn finish_outcome(&self) -> LocalMockOutcome { LocalMockOutcome { + engine_events: self.engine_events.clone(), trace: self.trace.clone(), injected_sequences: self .orchestrator diff --git a/crates/mvp-system/src/tests/local_mock_pipeline_integration.rs b/crates/mvp-system/src/tests/local_mock_pipeline_integration.rs index 991eb77..2f88ac2 100644 --- a/crates/mvp-system/src/tests/local_mock_pipeline_integration.rs +++ b/crates/mvp-system/src/tests/local_mock_pipeline_integration.rs @@ -5,6 +5,7 @@ //! and CUDA while still driving the run through planning, provisioning, //! readiness, prompt injection, stage execution, completion, and teardown. +use mvp_system::engine_builder as engine; use mvp_system::observability_surface as obs; use super::local_mock::{ @@ -19,6 +20,7 @@ fn local_mock_two_stage_pipeline_completes_and_tears_down() { assert_happy_path_lifecycle(&outcome); assert_topology_surface(&outcome); + assert_engine_builder_surface(&outcome); assert_terminal_success(&outcome); } @@ -57,6 +59,7 @@ fn local_mock_pipeline_topologies_complete() { assert_eq!(outcome.injected_sequences, expected_sequences); assert_topology_surface(&outcome); + assert_engine_builder_surface(&outcome); assert_terminal_success(&outcome); } } @@ -166,6 +169,44 @@ fn local_mock_rejects_invalid_cross_stage_events() { assert_terminal_fault(&sequence_violation); } +fn assert_engine_builder_surface(outcome: &LocalMockOutcome) { + assert!( + outcome + .engine_events + .iter() + .any(|event| matches!(event, engine::EngineEvent::PoolAcquired { .. })), + "local mock integration must be built from a neutral engine pool" + ); + assert!( + outcome + .engine_events + .iter() + .any(|event| matches!(event, engine::EngineEvent::ClusterConverged { .. })), + "local mock integration must pass through the builder convergence barrier" + ); + assert!( + outcome + .engine_events + .iter() + .any(|event| matches!(event, engine::EngineEvent::EngineReady { .. })), + "local mock integration must return an engine-ready handle before workload IO" + ); + let assigned_stages = outcome + .engine_events + .iter() + .filter(|event| { + matches!( + event, + engine::EngineEvent::RoleAssigned { + role: engine::RoleKind::StageWorker { .. }, + .. + } + ) + }) + .count(); + assert_eq!(assigned_stages, outcome.stage_count); +} + fn count_kind(outcome: &LocalMockOutcome, kind: obs::EventKind) -> usize { outcome .trace diff --git a/crates/mvp-system/src/tests/mod.rs b/crates/mvp-system/src/tests/mod.rs index 728b6eb..32fec57 100644 --- a/crates/mvp-system/src/tests/mod.rs +++ b/crates/mvp-system/src/tests/mod.rs @@ -1,6 +1,7 @@ mod arena_manager_guarantees; mod device_bridge_guarantees; mod edge_establisher_guarantees; +mod engine_builder_guarantees; mod gpu_worker_ctl_guarantees; mod gpu_worker_egress_producer_guarantees; mod gpu_worker_ingress_parser_guarantees; diff --git a/crates/mvp-system/tests/local_unmocked_mvp_e2e.rs b/crates/mvp-system/tests/local_unmocked_mvp_e2e.rs index a4e86ce..a23232b 100644 --- a/crates/mvp-system/tests/local_unmocked_mvp_e2e.rs +++ b/crates/mvp-system/tests/local_unmocked_mvp_e2e.rs @@ -19,6 +19,18 @@ fn local_e2e_binary_drives_real_local_process_deployment() { assert_eq!(value["actor_plane"], "iroh-swactor"); assert_eq!(value["data_plane"], "tcp-loopback-streams"); assert_eq!(value["worker_processes"], "mvp-dumb-worker-per-node"); + assert_eq!( + value["engine_builder_pattern"], + "pool-first-static-launcher" + ); + assert_eq!(value["engine_builder_node_count"], 3); + assert_eq!(value["engine_builder_stage_assignments"], 2); + assert!( + value["engine_builder_event_count"] + .as_u64() + .is_some_and(|count| count >= 10), + "{value}" + ); assert_eq!(value["injected_prompt_observed"], true); assert_eq!(value["token_received_observed"], true); assert_eq!(value["run_completed_observed"], true); diff --git a/src/guarantees/RUNTIME_GUARANTEES.md b/src/guarantees/RUNTIME_GUARANTEES.md deleted file mode 100644 index 459bdc1..0000000 --- a/src/guarantees/RUNTIME_GUARANTEES.md +++ /dev/null @@ -1,90 +0,0 @@ -# Runtime Guarantees - -This document describes the beta core-runtime guarantees that remain after the -alpha-only `std` features were pruned. Tests now separate core correctness from -optional library patterns. - -## Verification layers - -1. Unit/integration tests exercise observable runtime behavior. -2. Kani/model-check modules cover core finite-state lifecycle decisions where - enabled. -3. Exhaustive correspondence tests enumerate production decision-function truth - tables and compare them with runtime behavior. -4. `tests/core_extension_seams.rs` verifies the generic extension seam without - depending on unused std features. - -## Core decision functions - -| Function | Defined in | Runtime use | -| --- | --- | --- | -| `should_skip_actor(poisoned, stopping, suspended) -> bool` | `worker.rs` | `tick_all` skips actors that must not process mailbox messages. | -| `is_on_stop_eligible(stopping, poisoned) -> bool` | `worker.rs` | Cleanup decides whether `on_stop` should run. | -| `determine_stop_reason(poisoned, has_exit_value) -> StopReason` | `worker.rs` | Cleanup reports normal, panic, or completed exits. | - -## Retained core guarantees - -### G4: Lifecycle ordering - -Actors process messages only while eligible. Stopping actors do not handle later -mailbox messages, poisoned actors do not run `on_stop`, and graceful stops run -`on_stop` exactly once. - -Evidence: -- `src/guarantees/correspondence.rs` -- `src/guarantees/g4_lifecycle.rs` when Kani is enabled -- `src/guarantees/stateright_lifecycle.rs` - -### G5: Fault isolation - -A panicking actor is removed/poisoned without preventing unrelated actors from -continuing to process messages. - -Evidence: -- `src/guarantees/g5_fault_isolation.rs` -- `tests/actor_lifecycle.rs` - -### Core extension seam correctness - -The core runtime correctly invokes extension hooks independent of any specific -std feature: - -- `RuntimeExtension::on_spawn` can mutate the spawned actor environment. -- `RuntimeExtension::on_actor_death` can return messages, and core routes them - normally. -- `RuntimeExtension::cleanup_dead` receives dead actor batches. -- `RuntimeExtension::create_worker_extension` installs a per-worker extension. -- `WorkerExtension::handle_request`, `on_tick`, `gc_dead`, and - `has_pending_work` participate in worker progress and message routing. - -Evidence: -- `tests/core_extension_seams.rs` - -## Retained std beta guarantees - -Only std code used by production crates remains in the beta surface: - -- runtime naming registration, lookup, unregister, listing, and dead-actor cleanup -- runtime groups join, leave, publish, membership listing, and dead-actor cleanup -- actor-side `ctx.watch(target)` death notifications via `ActorExited` -- actor-side `ctx.join_group(group)` membership - -Evidence: -- `tests/std_extension.rs` - -## Removed alpha-only guarantees - -The following were tied to unused `std` features and are no longer part of the -beta guarantee set: - -- monitor/`Down` delivery and demonitor cancellation -- tick timers and interval timers -- supervisor restart policies and strategies -- router distribution/replacement/meltdown behavior -- service/resource injection and typed resource handles -- std wrapper traits for lifecycle, lineage, capabilities, system info, - self-stats, and environment access -- supervised-orphan distinction - -If one of these features becomes production-used again, reintroduce it with a -focused beta API and fresh correctness tests for that feature. diff --git a/src/guarantees/correspondence.rs b/src/guarantees/correspondence.rs deleted file mode 100644 index 5ee3ba9..0000000 --- a/src/guarantees/correspondence.rs +++ /dev/null @@ -1,235 +0,0 @@ -//! Exhaustive correspondence tests for core runtime lifecycle decisions. -//! -//! These tests enumerate finite input spaces for production decision functions and -//! compare them to observable runtime behavior. Pruned alpha std features are not -//! part of the beta guarantee set. - -use crate::actor::{ActorAddress, ActorInterface, StopReason}; -use crate::config::RuntimeConfig; -use crate::runtime::{Ctx, Runtime}; -use crate::worker::{is_on_stop_eligible, should_skip_actor}; - -fn tick_many(rt: &Runtime, n: usize) { - for _ in 0..n { - rt.tick(); - } -} - -#[derive(Clone, Debug)] -struct Ping; - -#[derive(Clone, Debug)] -struct HandleCalled(#[allow(dead_code)] ActorAddress); - -#[derive(Clone, Debug)] -struct OnStopCalled(#[allow(dead_code)] ActorAddress); - -struct DualReporter { - handle_to: ActorAddress, - stop_to: ActorAddress, -} - -impl ActorInterface for DualReporter { - type Incoming = Ping; - type Response = (); - - fn handle(&mut self, ctx: &Ctx, _msg: Ping) { - let _ = ctx.send(self.handle_to, HandleCalled(ctx.self_addr())); - } - - fn on_stop(&mut self, ctx: &Ctx) { - let _ = ctx.send(self.stop_to, OnStopCalled(ctx.self_addr())); - } -} - -struct OnStartPanicker { - handle_to: ActorAddress, - stop_to: ActorAddress, -} - -impl ActorInterface for OnStartPanicker { - type Incoming = Ping; - type Response = (); - - fn on_start(&mut self, _ctx: &Ctx) { - panic!("intentional on_start panic"); - } - - fn handle(&mut self, ctx: &Ctx, _msg: Ping) { - let _ = ctx.send(self.handle_to, HandleCalled(ctx.self_addr())); - } - - fn on_stop(&mut self, ctx: &Ctx) { - let _ = ctx.send(self.stop_to, OnStopCalled(ctx.self_addr())); - } -} - -struct HandlePanicker { - stop_to: ActorAddress, -} - -impl ActorInterface for HandlePanicker { - type Incoming = Ping; - type Response = (); - - fn handle(&mut self, _ctx: &Ctx, _msg: Ping) { - panic!("intentional handle panic"); - } - - fn on_stop(&mut self, ctx: &Ctx) { - let _ = ctx.send(self.stop_to, OnStopCalled(ctx.self_addr())); - } -} - -/// Exhaustive G4: for a healthy actor, the production `should_skip_actor` -/// predicts handle will be called — the real runtime agrees. -#[test] -fn g4_healthy_actor_handle_called() { - for msg_count in 1..=5 { - let rt = Runtime::new(RuntimeConfig::default()); - let h_inbox = rt.new_inbox::().unwrap(); - let report_addr = *h_inbox.addr(); - - let addr = rt - .spawn(DualReporter { - handle_to: report_addr, - stop_to: report_addr, - }) - .unwrap(); - rt.tick(); - - assert!(!should_skip_actor(false, false, false)); - - for _ in 0..msg_count { - rt.send_to(addr, Ping).unwrap(); - } - tick_many(&rt, msg_count + 5); - - let handle_count = std::iter::from_fn(|| h_inbox.try_recv()).count(); - assert_eq!(handle_count, msg_count); - } -} - -/// Exhaustive G4: a poisoned actor must not have handle or on_stop called. -#[test] -fn g4_poisoned_actor_no_handle_no_on_stop() { - for msg_count in 1..=5 { - let rt = Runtime::new(RuntimeConfig::default()); - let h_inbox = rt.new_inbox::().unwrap(); - let s_inbox = rt.new_inbox::().unwrap(); - - assert!(should_skip_actor(true, false, false)); - assert!(!is_on_stop_eligible(false, true)); - - let addr = rt - .spawn(OnStartPanicker { - handle_to: *h_inbox.addr(), - stop_to: *s_inbox.addr(), - }) - .unwrap(); - rt.tick(); - - for _ in 0..msg_count { - let _ = rt.send_to(addr, Ping); - } - tick_many(&rt, msg_count + 5); - - assert_eq!(std::iter::from_fn(|| h_inbox.try_recv()).count(), 0); - assert_eq!(std::iter::from_fn(|| s_inbox.try_recv()).count(), 0); - } -} - -/// Exhaustive G4: a stopping actor skips later messages and calls on_stop once. -#[test] -fn g4_stopping_actor_no_handle_yes_on_stop() { - for msg_count in 1..=5 { - assert!(should_skip_actor(false, true, false)); - assert!(is_on_stop_eligible(true, false)); - - let rt = Runtime::new(RuntimeConfig::default()); - let h_inbox = rt.new_inbox::().unwrap(); - let s_inbox = rt.new_inbox::().unwrap(); - - let addr = rt - .spawn(DualReporter { - handle_to: *h_inbox.addr(), - stop_to: *s_inbox.addr(), - }) - .unwrap(); - rt.tick(); - - rt.stop_actor(addr).unwrap(); - for _ in 0..msg_count { - let _ = rt.send_to(addr, Ping); - } - tick_many(&rt, 5); - - assert_eq!(std::iter::from_fn(|| h_inbox.try_recv()).count(), 0); - assert_eq!(std::iter::from_fn(|| s_inbox.try_recv()).count(), 1); - } -} - -/// Exhaustive G4: a handle-panicked actor must not call on_stop. -#[test] -fn g4_handle_panic_poisons_no_on_stop() { - let rt = Runtime::new(RuntimeConfig::default()); - let s_inbox = rt.new_inbox::().unwrap(); - - let addr = rt - .spawn(HandlePanicker { - stop_to: *s_inbox.addr(), - }) - .unwrap(); - rt.tick(); - - rt.send_to(addr, Ping).unwrap(); - tick_many(&rt, 5); - - assert!(!is_on_stop_eligible(false, true)); - assert_eq!(std::iter::from_fn(|| s_inbox.try_recv()).count(), 0); -} - -/// Exhaustively enumerate core lifecycle decision function truth tables. -#[test] -fn g4_exhaustive_decision_function_truth_table() { - use crate::worker::determine_stop_reason; - - let mut skip_combinations = 0; - for poisoned in [false, true] { - for stopping in [false, true] { - for suspended in [false, true] { - assert_eq!( - should_skip_actor(poisoned, stopping, suspended), - poisoned || stopping || suspended - ); - skip_combinations += 1; - } - } - } - assert_eq!(skip_combinations, 8); - - let mut on_stop_combinations = 0; - for stopping in [false, true] { - for poisoned in [false, true] { - assert_eq!(is_on_stop_eligible(stopping, poisoned), stopping && !poisoned); - on_stop_combinations += 1; - } - } - assert_eq!(on_stop_combinations, 4); - - let mut reason_combinations = 0; - for poisoned in [false, true] { - for has_exit_value in [false, true] { - let expected = if poisoned { - StopReason::Panicked - } else if has_exit_value { - StopReason::Completed - } else { - StopReason::Normal - }; - assert_eq!(determine_stop_reason(poisoned, has_exit_value), expected); - reason_combinations += 1; - } - } - assert_eq!(reason_combinations, 4); -} diff --git a/src/guarantees/g4_lifecycle.rs b/src/guarantees/g4_lifecycle.rs deleted file mode 100644 index afb42f2..0000000 --- a/src/guarantees/g4_lifecycle.rs +++ /dev/null @@ -1,422 +0,0 @@ -//! Kani proof harnesses for G4 — Actor Lifecycle Ordering. -//! -//! Bounded mirror of the actor lifecycle FSM from `worker.rs`. Models -//! the four boolean flags (`started`, `stopping`, `poisoned`, `suspended`) -//! and the transitions that `tick_all` and `cleanup_dead` apply. -//! -//! The mirror's decision points call the **production** pure functions -//! (`should_skip_actor`, `is_on_stop_eligible`) from `worker.rs`, so Kani -//! is proving properties of the real code, not a test-only re-implementation. -//! -//! Properties proven: -//! - **G4a**: `on_start` fires exactly once, before any `handle`. -//! - **G4b**: `handle` is never called when `stopping || poisoned`. -//! - **G4c**: `on_stop` fires at most once, only when `stopping && !poisoned`. -//! - **G4d**: No transition sequence reaches `handle` after `on_stop`. -//! - **G4e**: Suspension pauses message processing; resume restores it. - -use crate::worker::{is_on_stop_eligible, should_skip_actor}; - -// ─── Bounded mirror ───────────────────────────────────────────────────────── - -/// Events that can occur during a tick, mirroring the control flow in -/// `ActorPool::tick_all` and `ActorPool::deliver`. -#[derive(Clone, Copy, PartialEq, Eq)] -enum Event { - /// A regular message is delivered and processed. - Message, - /// The actor's handler (or on_start) panics. - Panic, - /// A stop request arrives (StopSignal or ctx.stop()). - Stop, - /// A suspend request arrives (ctx.suspend()). - Suspend, - /// A resume signal is delivered. - Resume, -} - -/// Bounded mirror of `ActorSlot`'s lifecycle state. Tracks the four -/// boolean flags and lifecycle callback invocations. -struct KaniActorState { - started: bool, - stopping: bool, - poisoned: bool, - suspended: bool, - - // Counters for property assertions - on_start_count: u32, - handle_count: u32, - on_stop_count: u32, - cleanup_done: bool, -} - -impl KaniActorState { - fn new() -> Self { - Self { - started: false, - stopping: false, - poisoned: false, - suspended: false, - on_start_count: 0, - handle_count: 0, - on_stop_count: 0, - cleanup_done: false, - } - } - - /// Mirror of the per-actor logic inside `tick_all`. - /// Returns true if this actor was processed (not skipped). - fn tick(&mut self, events: &[Event], event_count: usize) { - // Use production decision function for skip check - if should_skip_actor(self.poisoned, self.stopping, self.suspended) { - return; - } - - // on_start phase (worker.rs:519-571) - if !self.started { - self.on_start_count += 1; - self.started = true; - - // Check if on_start triggered a panic - if event_count > 0 && events[0] == Event::Panic { - self.poisoned = true; - return; - } - - // Check if on_start requested stop - if event_count > 0 && events[0] == Event::Stop { - self.stopping = true; - return; - } - - // Check if on_start requested suspend - if event_count > 0 && events[0] == Event::Suspend { - self.suspended = true; - return; - } - - // If the first event was consumed by on_start, we'd need - // to handle that — but in the real code, on_start doesn't - // consume a mailbox message; it's a separate phase. The - // events model control-flow outcomes. For on_start we only - // consume event[0] if it's Panic/Stop/Suspend (side effects - // of on_start). Message events start from the next index. - } - - // Message processing loop (worker.rs:573-661) - let start_idx = if self.on_start_count > 0 - && !self.poisoned - && !self.stopping - && !self.suspended - && event_count > 0 - && matches!(events[0], Event::Panic | Event::Stop | Event::Suspend) - { - // Event[0] was consumed by on_start outcome check above - // But wait — if we already returned above for those cases, we - // won't reach here. So start_idx is always 0 for message events - // when on_start succeeded without side effects. - 0 - } else { - 0 - }; - - let mut i = start_idx; - while i < event_count { - let event = events[i]; - i += 1; - - match event { - Event::Message => { - // handle_any called (worker.rs:596-621) - self.handle_count += 1; - } - Event::Panic => { - // Panic during handle (worker.rs:603-611) - // The handle call itself panicked — we count it as a - // handle attempt that failed, but the key point is - // poisoned is set. - self.handle_count += 1; - self.poisoned = true; - return; - } - Event::Stop => { - // StopSignal in mailbox or ctx.stop() after handle - // (worker.rs:576-583, 626-644) - self.stopping = true; - return; - } - Event::Suspend => { - // ctx.suspend() after handle (worker.rs:649-655) - self.suspended = true; - return; - } - Event::Resume => { - // Resume signals are handled in deliver(), not in - // tick_all. In tick_all, a ResumeSignal in the mailbox - // would be processed as a regular message (type mismatch). - // For the FSM model, resume only matters when delivered - // to a suspended actor via deliver(). We treat it as a - // no-op message here. - self.handle_count += 1; - } - } - } - } - - /// Mirror of `deliver` for suspended actors (worker.rs:450-478). - fn deliver(&mut self, event: Event) { - if self.suspended { - match event { - Event::Resume => { - self.suspended = false; - } - Event::Stop => { - self.stopping = true; - } - _ => { - // Message queued but not processed - } - } - } - // Non-suspended: message is just pushed to mailbox (handled in tick) - } - - /// Mirror of `cleanup_dead` (worker.rs:684-721). - fn cleanup(&mut self) { - if !self.poisoned && !self.stopping { - return; - } - // Use production decision function for on_stop eligibility - if is_on_stop_eligible(self.stopping, self.poisoned) { - self.on_stop_count += 1; - } - self.cleanup_done = true; - } -} - -// ─── Proof harnesses ──────────────────────────────────────────────────────── - -const MAX_EVENTS: usize = 6; - -/// Helper: generate a bounded event sequence from symbolic inputs. -fn symbolic_events(events: &mut [Event; MAX_EVENTS]) -> usize { - let len: usize = kani::any(); - kani::assume(len <= MAX_EVENTS); - - let mut i = 0; - while i < len { - let e: u8 = kani::any(); - kani::assume(e < 5); - events[i] = match e { - 0 => Event::Message, - 1 => Event::Panic, - 2 => Event::Stop, - 3 => Event::Suspend, - _ => Event::Resume, - }; - i += 1; - } - len -} - -/// **G4a**: `on_start` fires exactly once, before any `handle`. -#[kani::proof] -#[kani::unwind(8)] -fn proof_g4a_on_start_exactly_once() { - let mut actor = KaniActorState::new(); - - // Run multiple ticks with symbolic events - const MAX_TICKS: usize = 3; - let num_ticks: usize = kani::any(); - kani::assume(num_ticks <= MAX_TICKS); - - let mut total_on_start = 0u32; - let mut any_handle_before_start = false; - let mut t = 0; - - while t < num_ticks { - let prev_on_start = actor.on_start_count; - let prev_handle = actor.handle_count; - - let mut events = [Event::Message; MAX_EVENTS]; - let len = symbolic_events(&mut events); - - // Optionally deliver a resume between ticks - let do_resume: bool = kani::any(); - if do_resume { - actor.deliver(Event::Resume); - } - - actor.tick(&events, len); - - // Check: if handle increased but on_start hadn't fired yet, that's a violation - if actor.handle_count > prev_handle && prev_on_start == 0 { - any_handle_before_start = true; - } - - t += 1; - } - - actor.cleanup(); - - // on_start fires at most once - assert!(actor.on_start_count <= 1); - - // If the actor was ever ticked (not always skipped), on_start fired - // exactly once — unless it was already poisoned/stopping before first tick. - // (An actor that is never ticked never gets on_start, which is correct.) - - // No handle before on_start - assert!(!any_handle_before_start); -} - -/// **G4b**: `handle` is never called when `stopping || poisoned`. -#[kani::proof] -#[kani::unwind(8)] -fn proof_g4b_no_handle_when_stopping_or_poisoned() { - let mut actor = KaniActorState::new(); - - const MAX_TICKS: usize = 3; - let num_ticks: usize = kani::any(); - kani::assume(num_ticks <= MAX_TICKS); - - let mut t = 0; - while t < num_ticks { - let was_stopping = actor.stopping; - let was_poisoned = actor.poisoned; - let prev_handle = actor.handle_count; - - let mut events = [Event::Message; MAX_EVENTS]; - let len = symbolic_events(&mut events); - - let do_resume: bool = kani::any(); - if do_resume { - actor.deliver(Event::Resume); - } - - actor.tick(&events, len); - - // If actor was stopping or poisoned before this tick, handle must not increase - if was_stopping || was_poisoned { - assert!(actor.handle_count == prev_handle); - } - - t += 1; - } -} - -/// **G4c**: `on_stop` fires at most once, only when `stopping && !poisoned`. -#[kani::proof] -#[kani::unwind(8)] -fn proof_g4c_on_stop_conditions() { - let mut actor = KaniActorState::new(); - - let mut events = [Event::Message; MAX_EVENTS]; - let len = symbolic_events(&mut events); - actor.tick(&events, len); - - // Possibly deliver more events and tick again - let do_second_tick: bool = kani::any(); - if do_second_tick { - let do_resume: bool = kani::any(); - if do_resume { - actor.deliver(Event::Resume); - } - let mut events2 = [Event::Message; MAX_EVENTS]; - let len2 = symbolic_events(&mut events2); - actor.tick(&events2, len2); - } - - let was_stopping = actor.stopping; - let was_poisoned = actor.poisoned; - - actor.cleanup(); - - // on_stop fires at most once - assert!(actor.on_stop_count <= 1); - - // on_stop fires only if stopping && !poisoned - if actor.on_stop_count == 1 { - assert!(was_stopping && !was_poisoned); - } - - // If poisoned, on_stop must NOT fire - if was_poisoned { - assert!(actor.on_stop_count == 0); - } -} - -/// **G4d**: No `handle` after `on_stop`. Since `on_stop` only fires in -/// `cleanup_dead` which removes the actor from the pool, no further ticks -/// are possible. We verify: once cleanup is done, no further ticks can -/// increase handle_count. -#[kani::proof] -#[kani::unwind(8)] -fn proof_g4d_no_handle_after_on_stop() { - let mut actor = KaniActorState::new(); - - // First tick - let mut events = [Event::Message; MAX_EVENTS]; - let len = symbolic_events(&mut events); - actor.tick(&events, len); - - // Cleanup (on_stop fires here if applicable) - actor.cleanup(); - let handle_at_cleanup = actor.handle_count; - let on_stop_fired = actor.on_stop_count > 0; - - // Attempt another tick after cleanup - let mut events2 = [Event::Message; MAX_EVENTS]; - let len2 = symbolic_events(&mut events2); - actor.tick(&events2, len2); - - // If on_stop fired, actor must be stopping (or poisoned), so tick is a no-op - if on_stop_fired { - assert!(actor.handle_count == handle_at_cleanup); - } -} - -/// **G4e**: Suspension pauses message processing; resume restores it. -/// No `handle` calls occur while suspended. -#[kani::proof] -#[kani::unwind(8)] -fn proof_g4e_suspension_pauses_handle() { - let mut actor = KaniActorState::new(); - - // First tick — may suspend - let mut events1 = [Event::Message; MAX_EVENTS]; - let len1 = symbolic_events(&mut events1); - actor.tick(&events1, len1); - - let handle_after_first = actor.handle_count; - - // If suspended, a tick should not increase handle_count - if actor.suspended { - let mut events2 = [Event::Message; MAX_EVENTS]; - let len2 = symbolic_events(&mut events2); - actor.tick(&events2, len2); - assert!(actor.handle_count == handle_after_first); - - // Resume via deliver - actor.deliver(Event::Resume); - assert!(!actor.suspended); - - // Now tick should be able to process messages again - let mut events3 = [Event::Message; MAX_EVENTS]; - let len3 = symbolic_events(&mut events3); - - // Only assert handle can increase if there are Message events - // and actor isn't stopping/poisoned - let handle_before_resume_tick = actor.handle_count; - actor.tick(&events3, len3); - - // After resume, if we had Message events and actor is healthy, - // handle_count should have increased (unless len3 == 0 or all - // events were non-Message). The key property is simply that - // the tick was NOT skipped — the suspended check didn't block it. - // We verify this indirectly: actor is no longer suspended. - if actor.handle_count > handle_before_resume_tick { - assert!(!actor.suspended || actor.poisoned || actor.stopping); - } - } -} diff --git a/src/guarantees/g5_fault_isolation.rs b/src/guarantees/g5_fault_isolation.rs deleted file mode 100644 index 8e89e49..0000000 --- a/src/guarantees/g5_fault_isolation.rs +++ /dev/null @@ -1,366 +0,0 @@ -//! Property-based tests for G5: Fault Isolation. -//! -//! An actor panic must never affect any other actor. Sibling actors must -//! continue to process messages and complete their lifecycle normally. - -use proptest::prelude::*; - -use crate::actor::{ActorAddress, ActorInterface}; -use crate::config::RuntimeConfig; -use crate::runtime::{Ctx, Runtime}; - -// ─── Message Types ────────────────────────────────────────────────────────── - -#[derive(Clone, Debug)] -struct Count(u64); - -#[derive(Clone, Debug)] -struct Started(ActorAddress); - -#[derive(Clone, Debug)] -struct Stopped(ActorAddress); - -// ─── Actor Types ──────────────────────────────────────────────────────────── - -/// Counts messages received, sends final count to inbox on stop. -struct CountingActor { - count: u64, - report_to: ActorAddress, -} - -impl ActorInterface for CountingActor { - type Incoming = Count; - type Response = (); - - fn handle(&mut self, _ctx: &Ctx, _msg: Count) { - self.count += 1; - } - - fn on_stop(&mut self, ctx: &Ctx) { - let _ = ctx.send(self.report_to, Count(self.count)); - } -} - -/// Panics after receiving exactly `panic_at` messages. -struct DelayedPanicActor { - count: u64, - panic_at: u64, -} - -impl ActorInterface for DelayedPanicActor { - type Incoming = Count; - type Response = (); - - fn handle(&mut self, _ctx: &Ctx, _msg: Count) { - self.count += 1; - if self.count == self.panic_at { - panic!("intentional panic at message {}", self.panic_at); - } - } -} - -/// Panics in on_start. -struct StartPanicActor; - -impl ActorInterface for StartPanicActor { - type Incoming = Count; - type Response = (); - - fn handle(&mut self, _ctx: &Ctx, _msg: Count) {} - - fn on_start(&mut self, _ctx: &Ctx) { - panic!("intentional panic in on_start"); - } -} - -/// Reports lifecycle events to an inbox so we can verify them externally. -struct LifecycleActor { - start_report: ActorAddress, - stop_report: ActorAddress, - count: u64, - count_report: ActorAddress, -} - -impl ActorInterface for LifecycleActor { - type Incoming = Count; - type Response = (); - - fn on_start(&mut self, ctx: &Ctx) { - let _ = ctx.send(self.start_report, Started(ctx.self_addr())); - } - - fn handle(&mut self, _ctx: &Ctx, _msg: Count) { - self.count += 1; - } - - fn on_stop(&mut self, ctx: &Ctx) { - let _ = ctx.send(self.count_report, Count(self.count)); - let _ = ctx.send(self.stop_report, Stopped(ctx.self_addr())); - } -} - -// ─── Helpers ──────────────────────────────────────────────────────────────── - -fn tick_many(rt: &Runtime, n: usize) { - for _ in 0..n { - rt.tick(); - } -} - -// ─── Property Tests ───────────────────────────────────────────────────────── - -proptest! { - #![proptest_config(ProptestConfig::with_cases(80))] - - /// Spawn N healthy actors and 1 that panics at a random message index. - /// Send `msg_count` messages to every actor. Assert all healthy actors - /// receive exactly `msg_count` messages and call on_stop normally. - #[test] - fn panic_at_random_index_isolates( - n in 2usize..=12, - msg_count in 1u64..=200, - panic_at in 1u64..=200, - ) { - let rt = Runtime::new(RuntimeConfig::default()); - let report_inbox = rt.new_inbox::().unwrap(); - let report_addr = *report_inbox.addr(); - - // Spawn N healthy counting actors - let mut healthy: Vec = Vec::with_capacity(n); - for _ in 0..n { - let addr = rt.spawn(CountingActor { - count: 0, - report_to: report_addr, - }).unwrap(); - healthy.push(addr); - } - - // Spawn the panicking actor (clamp panic_at to msg_count so it fires) - let actual_panic_at = (panic_at % msg_count) + 1; - let panic_addr = rt.spawn(DelayedPanicActor { - count: 0, - panic_at: actual_panic_at, - }).unwrap(); - - // Deliver on_start - rt.tick(); - - // Send msg_count messages to every actor (healthy + panicker) - let all_addrs: Vec = healthy.iter() - .copied() - .chain(std::iter::once(panic_addr)) - .collect(); - - for i in 0..msg_count { - for &addr in &all_addrs { - rt.send_to(addr, Count(i)).unwrap(); - } - } - - // Tick enough to drain everything (budget=64 default) - let ticks = (msg_count as usize / 64) + 10; - tick_many(&rt, ticks); - - // Stop healthy actors so they report - for &addr in &healthy { - rt.stop_actor(addr).unwrap(); - } - tick_many(&rt, 3); - - // Collect reports - let mut reports = Vec::new(); - while let Some(msg) = report_inbox.try_recv() { - reports.push(msg.0); - } - - // Every healthy actor must have processed exactly msg_count messages - prop_assert_eq!( - reports.len(), n, - "expected {} reports, got {}", n, reports.len() - ); - for (i, &count) in reports.iter().enumerate() { - prop_assert_eq!( - count, msg_count, - "actor {} processed {} messages, expected {}", - i, count, msg_count - ); - } - } - - /// An actor that panics in on_start must not affect siblings spawned - /// before or after it. - #[test] - fn on_start_panic_isolates_siblings( - before_count in 1usize..=8, - after_count in 1usize..=8, - msgs_each in 1u64..=100, - ) { - let rt = Runtime::new(RuntimeConfig::default()); - let start_inbox = rt.new_inbox::().unwrap(); - let stop_inbox = rt.new_inbox::().unwrap(); - let count_inbox = rt.new_inbox::().unwrap(); - - let start_addr = *start_inbox.addr(); - let stop_addr = *stop_inbox.addr(); - let count_addr = *count_inbox.addr(); - - // Spawn "before" siblings - let mut before_addrs = Vec::new(); - for _ in 0..before_count { - let addr = rt.spawn(LifecycleActor { - start_report: start_addr, - stop_report: stop_addr, - count: 0, - count_report: count_addr, - }).unwrap(); - before_addrs.push(addr); - } - - // Spawn the on_start panicker - let _panic_addr = rt.spawn(StartPanicActor).unwrap(); - - // Spawn "after" siblings - let mut after_addrs = Vec::new(); - for _ in 0..after_count { - let addr = rt.spawn(LifecycleActor { - start_report: start_addr, - stop_report: stop_addr, - count: 0, - count_report: count_addr, - }).unwrap(); - after_addrs.push(addr); - } - - // Tick to run on_start for everyone - tick_many(&rt, 3); - - // Verify all siblings started successfully - let mut started = Vec::new(); - while let Some(Started(addr)) = start_inbox.try_recv() { - started.push(addr); - } - let total_siblings = before_count + after_count; - prop_assert_eq!( - started.len(), total_siblings, - "expected {} on_start reports, got {}", total_siblings, started.len() - ); - - // Send messages to all siblings - let all_siblings: Vec = before_addrs.iter() - .chain(after_addrs.iter()) - .copied() - .collect(); - for i in 0..msgs_each { - for &addr in &all_siblings { - rt.send_to(addr, Count(i)).unwrap(); - } - } - - let ticks = (msgs_each as usize / 64) + 5; - tick_many(&rt, ticks); - - // Stop all siblings - for &addr in &all_siblings { - rt.stop_actor(addr).unwrap(); - } - tick_many(&rt, 3); - - // Check stop reports - let mut stopped = Vec::new(); - while let Some(Stopped(addr)) = stop_inbox.try_recv() { - stopped.push(addr); - } - prop_assert_eq!( - stopped.len(), total_siblings, - "expected {} on_stop reports, got {}", total_siblings, stopped.len() - ); - - // Check message counts - let mut counts = Vec::new(); - while let Some(Count(c)) = count_inbox.try_recv() { - counts.push(c); - } - prop_assert_eq!(counts.len(), total_siblings); - for (i, &c) in counts.iter().enumerate() { - prop_assert_eq!( - c, msgs_each, - "sibling {} processed {} messages, expected {}", i, c, msgs_each - ); - } - } - - /// Multiple actors panic in the same tick. Non-panicking actors must be - /// completely unaffected. - #[test] - fn multiple_panics_same_tick_isolates( - healthy_count in 2usize..=10, - panic_count in 2usize..=6, - msgs_each in 1u64..=150, - ) { - let rt = Runtime::new(RuntimeConfig::default()); - let report_inbox = rt.new_inbox::().unwrap(); - let report_addr = *report_inbox.addr(); - - // Spawn healthy actors - let mut healthy = Vec::new(); - for _ in 0..healthy_count { - let addr = rt.spawn(CountingActor { - count: 0, - report_to: report_addr, - }).unwrap(); - healthy.push(addr); - } - - // Spawn panicking actors — they all panic on message 1 - let mut panickers = Vec::new(); - for _ in 0..panic_count { - let addr = rt.spawn(DelayedPanicActor { - count: 0, - panic_at: 1, - }).unwrap(); - panickers.push(addr); - } - - // Tick to process on_start - rt.tick(); - - // Send messages to everyone — panickers get at least 1 so they - // all panic during the same tick - let all: Vec = healthy.iter() - .chain(panickers.iter()) - .copied() - .collect(); - for i in 0..msgs_each { - for &addr in &all { - rt.send_to(addr, Count(i)).unwrap(); - } - } - - // Tick enough to drain all messages - let ticks = (msgs_each as usize / 64) + 10; - tick_many(&rt, ticks); - - // Stop healthy actors to trigger on_stop reports - for &addr in &healthy { - rt.stop_actor(addr).unwrap(); - } - tick_many(&rt, 3); - - // Every healthy actor must have processed all messages - let mut reports = Vec::new(); - while let Some(Count(c)) = report_inbox.try_recv() { - reports.push(c); - } - prop_assert_eq!( - reports.len(), healthy_count, - "expected {} reports, got {}", healthy_count, reports.len() - ); - for (i, &c) in reports.iter().enumerate() { - prop_assert_eq!( - c, msgs_each, - "healthy actor {} processed {} messages, expected {}", - i, c, msgs_each - ); - } - } -} diff --git a/src/guarantees/mod.rs b/src/guarantees/mod.rs deleted file mode 100644 index 161aaf2..0000000 --- a/src/guarantees/mod.rs +++ /dev/null @@ -1,23 +0,0 @@ -//! Runtime guarantee verification modules. -//! -//! Formal and exhaustive checks now cover core runtime behavior only. Alpha std -//! features that are not used by production crates were pruned from the beta -//! surface, so their feature-specific guarantee modules are no longer compiled. - -#[cfg(kani)] -mod g4_lifecycle; - -#[cfg(test)] -mod g5_fault_isolation; - -#[cfg(test)] -mod correspondence; - -#[cfg(test)] -mod model_checker; - -#[cfg(test)] -mod stateright_lifecycle; - -#[cfg(test)] -mod stateright_death_orphan; diff --git a/src/guarantees/model_checker.rs b/src/guarantees/model_checker.rs deleted file mode 100644 index 3b5ac7d..0000000 --- a/src/guarantees/model_checker.rs +++ /dev/null @@ -1,215 +0,0 @@ -//! Minimal exhaustive DFS model checker. -//! -//! Provides the same core API surface as `stateright` — [`Model`] trait, -//! [`Property`] (always/sometimes), and a [`Checker`] with DFS exploration — -//! without requiring an external crate dependency. -//! -//! This keeps Cargo.lock unchanged while still providing genuine exhaustive -//! state-space exploration for the runtime guarantee proofs. - -use std::collections::HashSet; -use std::fmt::Debug; -use std::hash::Hash; -use std::marker::PhantomData; - -// ── Model trait ────────────────────────────────────────────────────────────── - -/// A finite-state model suitable for exhaustive exploration. -pub trait Model: Sized { - type State: Clone + Debug + Hash + Eq; - type Action: Clone + Debug + Hash + Eq; - - /// Initial states to begin exploration from. - fn init_states(&self) -> Vec; - - /// Enumerate all enabled actions in `state`, pushing them into `actions`. - fn actions(&self, state: &Self::State, actions: &mut Vec); - - /// Compute the successor state after applying `action`. Return `None` if - /// the action is a no-op (state unchanged). - fn next_state(&self, state: &Self::State, action: Self::Action) -> Option; - - /// Properties to verify across all reachable states. - fn properties(&self) -> Vec>; - - /// Create a checker for this model. - fn checker(&self) -> Checker<'_, Self> { - Checker { model: self } - } -} - -// ── Property ───────────────────────────────────────────────────────────────── - -/// The kind of temporal property. -enum PropertyKind { - /// Must hold in every reachable state. - Always, - /// Must hold in at least one reachable state (liveness canary). - Sometimes, -} - -/// A named property checked during model exploration. -pub struct Property { - name: String, - kind: PropertyKind, - checker_fn: Box bool>, -} - -impl Property { - /// Safety invariant: must hold in every reachable state. - pub fn always(name: &str, f: impl Fn(&M, &M::State) -> bool + 'static) -> Self { - Self { - name: name.to_string(), - kind: PropertyKind::Always, - checker_fn: Box::new(f), - } - } - - /// Liveness canary: must hold in at least one reachable state. - pub fn sometimes(name: &str, f: impl Fn(&M, &M::State) -> bool + 'static) -> Self { - Self { - name: name.to_string(), - kind: PropertyKind::Sometimes, - checker_fn: Box::new(f), - } - } -} - -// ── Checker & DFS ──────────────────────────────────────────────────────────── - -/// Builder that holds a reference to the model. -pub struct Checker<'a, M: Model> { - model: &'a M, -} - -impl<'a, M: Model> Checker<'a, M> { - /// Run exhaustive DFS. Named `spawn_dfs` for API compatibility, but runs - /// synchronously (no threads needed for our bounded models). - pub fn spawn_dfs(self) -> DfsHandle { - let properties = self.model.properties(); - let mut visited: HashSet = HashSet::new(); - let mut stack: Vec<(M::State, usize)> = Vec::new(); // (state, depth) - let mut max_depth: usize = 0; - let mut actions_buf: Vec = Vec::new(); - - // Track property results - let mut always_violated: Vec> = properties.iter().map(|_| None).collect(); - let mut sometimes_satisfied: Vec = properties.iter().map(|_| false).collect(); - - // Seed with init states - for s in self.model.init_states() { - if visited.insert(s.clone()) { - stack.push((s, 0)); - } - } - - while let Some((state, depth)) = stack.pop() { - if depth > max_depth { - max_depth = depth; - } - - // Check all properties against this state - for (i, prop) in properties.iter().enumerate() { - let holds = (prop.checker_fn)(self.model, &state); - match prop.kind { - PropertyKind::Always => { - if !holds && always_violated[i].is_none() { - always_violated[i] = Some(format!( - "ALWAYS property {:?} violated in state: {:?}", - prop.name, state - )); - } - } - PropertyKind::Sometimes => { - if holds { - sometimes_satisfied[i] = true; - } - } - } - } - - // Expand successors - actions_buf.clear(); - self.model.actions(&state, &mut actions_buf); - - for action in actions_buf.drain(..) { - if let Some(next) = self.model.next_state(&state, action) { - if visited.insert(next.clone()) { - stack.push((next, depth + 1)); - } - } - } - } - - // Build failures list - let mut failures = Vec::new(); - for (i, prop) in properties.iter().enumerate() { - match prop.kind { - PropertyKind::Always => { - if let Some(msg) = &always_violated[i] { - failures.push(msg.clone()); - } - } - PropertyKind::Sometimes => { - if !sometimes_satisfied[i] { - failures.push(format!( - "SOMETIMES property {:?} was never satisfied across {} states", - prop.name, - visited.len() - )); - } - } - } - } - - DfsHandle { - result: CheckResult { - unique_states: visited.len(), - max_depth, - failures, - }, - _phantom: PhantomData, - } - } -} - -/// Handle returned by `spawn_dfs`. Call `.join()` to get the result. -pub struct DfsHandle { - result: CheckResult, - _phantom: PhantomData, -} - -// Suppress unused type parameter warning -impl DfsHandle { - /// Consume the handle and return the exploration result. - pub fn join(self) -> CheckResult { - self.result - } -} - -/// Result of an exhaustive DFS exploration. -pub struct CheckResult { - unique_states: usize, - max_depth: usize, - failures: Vec, -} - -impl CheckResult { - /// Number of unique states explored. - pub fn unique_state_count(&self) -> usize { - self.unique_states - } - - /// Maximum DFS depth reached. - pub fn max_depth(&self) -> usize { - self.max_depth - } - - /// Panic if any property was violated. - pub fn assert_properties(&self) { - if !self.failures.is_empty() { - let msg = self.failures.join("\n"); - panic!("Property violations:\n{msg}"); - } - } -} diff --git a/src/guarantees/stateright_death_orphan.rs b/src/guarantees/stateright_death_orphan.rs deleted file mode 100644 index 0af4dbd..0000000 --- a/src/guarantees/stateright_death_orphan.rs +++ /dev/null @@ -1,494 +0,0 @@ -//! Stateright model-checking of death notifications (G6) and orphan cleanup (G7). -//! -//! Two focused models keep the state space tractable: -//! -//! 1. **MonitorModel** (G6): 3 actors with monitor/demonitor/kill. Proves -//! notification exactness, no-notification-for-alive, demonitor suppression. -//! -//! 2. **OrphanModel** (G7): 4 actors with parent-child/kill/orphan-cleanup. -//! Proves unsupervised children stop, supervised survive, cascading cleanup. - -use super::model_checker::{Model, Property}; - -// ═══════════════════════════════════════════════════════════════════════════ -// G6: Monitor Model -// ═══════════════════════════════════════════════════════════════════════════ - -const MON_N: usize = 3; -const MON_PAIRS: usize = MON_N * (MON_N - 1); // 6 - -fn mon_pair(i: usize, j: usize) -> usize { - debug_assert!(i < MON_N && j < MON_N && i != j); - if j < i { - i * (MON_N - 1) + j - } else { - i * (MON_N - 1) + j - 1 - } -} - -#[derive(Clone, Debug, Hash, PartialEq, Eq)] -struct MonitorState { - alive: [bool; MON_N], - /// Active monitor from i to j. - active: [bool; MON_PAIRS], - /// Demonitored while target was still alive (the meaningful demonitor case). - deactivated_while_alive: [bool; MON_PAIRS], - /// Notification count (capped at 2 to detect duplicates). - notif: [u8; MON_PAIRS], -} - -impl MonitorState { - fn init() -> Self { - Self { - alive: [true; MON_N], - active: [false; MON_PAIRS], - deactivated_while_alive: [false; MON_PAIRS], - notif: [0; MON_PAIRS], - } - } -} - -#[derive(Clone, Debug, Hash, PartialEq, Eq)] -enum MonAction { - Monitor(usize, usize), - Demonitor(usize, usize), - Kill(usize), -} - -#[derive(Clone)] -struct MonitorModel; - -impl Model for MonitorModel { - type State = MonitorState; - type Action = MonAction; - - fn init_states(&self) -> Vec { - vec![MonitorState::init()] - } - - fn actions(&self, s: &Self::State, actions: &mut Vec) { - for i in 0..MON_N { - if !s.alive[i] { - continue; - } - - for j in 0..MON_N { - if i == j { - continue; - } - let idx = mon_pair(i, j); - - if s.alive[j] && !s.active[idx] { - actions.push(MonAction::Monitor(i, j)); - } - if s.active[idx] { - actions.push(MonAction::Demonitor(i, j)); - } - } - - actions.push(MonAction::Kill(i)); - } - } - - fn next_state(&self, s: &Self::State, action: Self::Action) -> Option { - let mut n = s.clone(); - match action { - MonAction::Monitor(w, t) => { - let idx = mon_pair(w, t); - n.active[idx] = true; - n.deactivated_while_alive[idx] = false; - } - MonAction::Demonitor(w, t) => { - let idx = mon_pair(w, t); - n.active[idx] = false; - if n.alive[t] { - n.deactivated_while_alive[idx] = true; - } - } - MonAction::Kill(t) => { - if !n.alive[t] { - return None; - } - n.alive[t] = false; - for w in 0..MON_N { - if w == t { - continue; - } - let idx = mon_pair(w, t); - if n.alive[w] && n.active[idx] && n.notif[idx] < 2 { - n.notif[idx] += 1; - } - } - } - } - if n == *s { None } else { Some(n) } - } - - fn properties(&self) -> Vec> { - vec![ - Property::::always( - "G6a: exactly one notification per active monitor on dead target", - |_, s| { - for w in 0..MON_N { - if !s.alive[w] { - continue; - } - for t in 0..MON_N { - if w == t { - continue; - } - let idx = mon_pair(w, t); - if s.active[idx] && !s.alive[t] && s.notif[idx] != 1 { - return false; - } - } - } - true - }, - ), - Property::::always("G6b: no notification for alive targets", |_, s| { - for w in 0..MON_N { - for t in 0..MON_N { - if w == t { - continue; - } - if s.alive[t] && s.notif[mon_pair(w, t)] > 0 { - return false; - } - } - } - true - }), - Property::::always( - "G6c: demonitor before death suppresses notification", - |_, s| { - for w in 0..MON_N { - for t in 0..MON_N { - if w == t { - continue; - } - let idx = mon_pair(w, t); - // If demonitored while target was alive, no notification should exist - if s.deactivated_while_alive[idx] && s.notif[idx] > 0 { - return false; - } - } - } - true - }, - ), - // Liveness - Property::::sometimes("L1: monitor fires", |_, s| s.notif.iter().any(|&c| c > 0)), - Property::::sometimes("L2: demonitor suppression reachable", |_, s| { - (0..MON_N).any(|w| { - (0..MON_N).any(|t| { - w != t && { - let idx = mon_pair(w, t); - s.deactivated_while_alive[idx] && !s.alive[t] && s.notif[idx] == 0 - } - }) - }) - }), - Property::::sometimes("L3: multiple monitors on same target", |_, s| { - (0..MON_N).any(|t| { - let watchers: usize = (0..MON_N) - .filter(|&w| w != t && s.notif[mon_pair(w, t)] > 0) - .count(); - watchers >= 2 - }) - }), - ] - } -} - -// ═══════════════════════════════════════════════════════════════════════════ -// G7: Orphan Model -// ═══════════════════════════════════════════════════════════════════════════ - -/// 4 actors: enough for parent → child → grandchild chains (3 deep) plus -/// a sibling to test supervised vs unsupervised. -const ORP_N: usize = 4; -const ORP_PAIRS: usize = ORP_N * (ORP_N - 1); // 12 - -fn orp_pair(i: usize, j: usize) -> usize { - debug_assert!(i < ORP_N && j < ORP_N && i != j); - if j < i { - i * (ORP_N - 1) + j - } else { - i * (ORP_N - 1) + j - 1 - } -} - -#[derive(Clone, Debug, Hash, PartialEq, Eq)] -struct OrphanState { - alive: [bool; ORP_N], - /// Unsupervised parent→child. - parent_unsup: [bool; ORP_PAIRS], - /// Supervised parent→child. - parent_sup: [bool; ORP_PAIRS], - /// Orphan cleanup propagated for actor i. - orphan_cleaned: [bool; ORP_N], - /// Whether actor was killed by OrphanCleanup (not by explicit Kill). - orphan_killed: [bool; ORP_N], - /// How many parent-child links have been set (cap to limit state space). - link_count: u8, -} - -/// Max parent-child links to prevent state explosion. -const MAX_LINKS: u8 = 3; - -impl OrphanState { - fn init() -> Self { - Self { - alive: [true; ORP_N], - parent_unsup: [false; ORP_PAIRS], - parent_sup: [false; ORP_PAIRS], - orphan_cleaned: [false; ORP_N], - orphan_killed: [false; ORP_N], - link_count: 0, - } - } - - fn has_parent(&self, c: usize) -> bool { - (0..ORP_N).any(|p| { - p != c && { - let idx = orp_pair(p, c); - self.parent_unsup[idx] || self.parent_sup[idx] - } - }) - } - - fn has_children(&self, p: usize) -> bool { - (0..ORP_N).any(|c| { - c != p && { - let idx = orp_pair(p, c); - self.parent_unsup[idx] || self.parent_sup[idx] - } - }) - } -} - -#[derive(Clone, Debug, Hash, PartialEq, Eq)] -enum OrpAction { - SetParentUnsup(usize, usize), - SetParentSup(usize, usize), - Kill(usize), - OrphanCleanup(usize), -} - -#[derive(Clone)] -struct OrphanModel; - -impl Model for OrphanModel { - type State = OrphanState; - type Action = OrpAction; - - fn init_states(&self) -> Vec { - vec![OrphanState::init()] - } - - fn actions(&self, s: &Self::State, actions: &mut Vec) { - // SetParent actions only if under link cap - if s.link_count < MAX_LINKS { - for p in 0..ORP_N { - if !s.alive[p] { - continue; - } - for c in 0..ORP_N { - if p == c || !s.alive[c] { - continue; - } - if s.has_parent(c) { - continue; - } - // Prevent cycles: c must not be an ancestor of p - if is_ancestor(s, c, p) { - continue; - } - actions.push(OrpAction::SetParentUnsup(p, c)); - actions.push(OrpAction::SetParentSup(p, c)); - } - } - } - - for i in 0..ORP_N { - if s.alive[i] { - actions.push(OrpAction::Kill(i)); - } - if !s.alive[i] && !s.orphan_cleaned[i] && s.has_children(i) { - actions.push(OrpAction::OrphanCleanup(i)); - } - } - } - - fn next_state(&self, s: &Self::State, action: Self::Action) -> Option { - let mut n = s.clone(); - match action { - OrpAction::SetParentUnsup(p, c) => { - n.parent_unsup[orp_pair(p, c)] = true; - n.link_count += 1; - } - OrpAction::SetParentSup(p, c) => { - n.parent_sup[orp_pair(p, c)] = true; - n.link_count += 1; - } - OrpAction::Kill(i) => { - if !n.alive[i] { - return None; - } - n.alive[i] = false; - } - OrpAction::OrphanCleanup(dead_parent) => { - n.orphan_cleaned[dead_parent] = true; - // Kill unsupervised children - for c in 0..ORP_N { - if c == dead_parent { - continue; - } - if n.parent_unsup[orp_pair(dead_parent, c)] && n.alive[c] { - n.alive[c] = false; - n.orphan_killed[c] = true; - } - } - } - } - if n == *s { None } else { Some(n) } - } - - fn properties(&self) -> Vec> { - vec![ - // G7a: After orphan cleanup, all unsupervised children are dead - Property::::always("G7a: orphan cleanup stops unsupervised children", |_, s| { - for p in 0..ORP_N { - if !s.alive[p] && s.orphan_cleaned[p] { - for c in 0..ORP_N { - if c == p { - continue; - } - if s.parent_unsup[orp_pair(p, c)] && s.alive[c] { - return false; - } - } - } - } - true - }), - // G7b: OrphanCleanup never kills supervised-only children. - // If orphan_killed[c] is true, there must be a parent p with an - // unsupervised link (parent_unsup[p→c]) that was orphan-cleaned. - // A supervised-only child has no parent_unsup link, so orphan_killed - // being true for it would violate this property. - Property::::always("G7b: supervised children survive orphan cleanup", |_, s| { - for c in 0..ORP_N { - if s.orphan_killed[c] { - // There must exist a dead, cleaned parent with unsup link to c - let has_unsup_cleaned_parent = (0..ORP_N).any(|p| { - p != c && s.parent_unsup[orp_pair(p, c)] && s.orphan_cleaned[p] - }); - if !has_unsup_cleaned_parent { - return false; - } - } - } - true - }), - // G7c: Cascading — if orphan-cleaned parent's child also died and was - // orphan-cleaned, its unsupervised children are dead too - Property::::always("G7c: cascading orphan cleanup", |_, s| { - for p in 0..ORP_N { - if s.orphan_cleaned[p] { - for c in 0..ORP_N { - if c == p { - continue; - } - if s.parent_unsup[orp_pair(p, c)] && !s.alive[c] && s.orphan_cleaned[c] - { - for gc in 0..ORP_N { - if gc == c { - continue; - } - if s.parent_unsup[orp_pair(c, gc)] && s.alive[gc] { - return false; - } - } - } - } - } - } - true - }), - // Liveness - Property::::sometimes("L1: orphan cleanup triggers", |_, s| { - s.orphan_cleaned.iter().any(|&c| c) - }), - Property::::sometimes("L2: cascading cleanup reachable", |_, s| { - // Parent cleaned → child died → child cleaned - (0..ORP_N).any(|p| { - s.orphan_cleaned[p] - && (0..ORP_N).any(|c| { - c != p - && s.parent_unsup[orp_pair(p, c)] - && !s.alive[c] - && s.orphan_cleaned[c] - }) - }) - }), - Property::::sometimes("L3: supervised child survives cleanup", |_, s| { - (0..ORP_N).any(|p| { - s.orphan_cleaned[p] - && (0..ORP_N).any(|c| c != p && s.parent_sup[orp_pair(p, c)] && s.alive[c]) - }) - }), - ] - } -} - -/// Check if `ancestor` is an ancestor of `descendant` via parent links. -fn is_ancestor(s: &OrphanState, ancestor: usize, descendant: usize) -> bool { - // Walk up from descendant - let mut current = descendant; - for _ in 0..ORP_N { - let parent = (0..ORP_N).find(|&p| { - p != current && { - let idx = orp_pair(p, current); - s.parent_unsup[idx] || s.parent_sup[idx] - } - }); - match parent { - Some(p) if p == ancestor => return true, - Some(p) => current = p, - None => return false, - } - } - false -} - -// ── Tests ──────────────────────────────────────────────────────────────────── - -#[test] -#[ignore] // Exhaustive proof — run deliberately with `cargo test -- --ignored` -fn g6_monitor_model_check() { - let result = MonitorModel.checker().spawn_dfs().join(); - let unique = result.unique_state_count(); - let depth = result.max_depth(); - println!( - "Stateright G6 (Monitor): {} unique states, max depth {}", - unique, depth - ); - result.assert_properties(); - assert!(unique > 100, "Too few states ({unique})"); -} - -#[test] -#[ignore] // Exhaustive proof — run deliberately with `cargo test -- --ignored` -fn g7_orphan_model_check() { - let result = OrphanModel.checker().spawn_dfs().join(); - let unique = result.unique_state_count(); - let depth = result.max_depth(); - println!( - "Stateright G7 (Orphan): {} unique states, max depth {}", - unique, depth - ); - result.assert_properties(); - assert!(unique > 100, "Too few states ({unique})"); -} diff --git a/src/guarantees/stateright_lifecycle.rs b/src/guarantees/stateright_lifecycle.rs deleted file mode 100644 index 97aa9bd..0000000 --- a/src/guarantees/stateright_lifecycle.rs +++ /dev/null @@ -1,352 +0,0 @@ -//! Stateright model-checking of the actor lifecycle state machine (G4, G5). -//! -//! Exhaustively explores all interleavings of lifecycle transitions across a -//! bounded runtime with 3 actors. Each action directly transitions one actor's -//! lifecycle state (no explicit mailbox — events are modeled as actions). -//! -//! Verifies: -//! -//! - **G4**: Lifecycle ordering (on_start once, no handle when stopping/poisoned, -//! on_stop conditions, no handle after on_stop, suspension pauses processing). -//! - **G5**: Fault isolation (a panic in one actor never affects another's flags -//! or counters; healthy actors process messages despite sibling panics). -//! -//! Uses production decision functions (`should_skip_actor`, `is_on_stop_eligible`) -//! from `worker.rs` so the model checks real code, not test-only mirrors. - -use super::model_checker::{Model, Property}; -use crate::worker::{is_on_stop_eligible, should_skip_actor}; - -// ── Bounded constants ──────────────────────────────────────────────────────── - -/// Number of actors. 3 is the minimum to exercise isolation between a poisoned -/// actor and multiple healthy siblings. -const NUM_ACTORS: usize = 3; - -/// Cap on handle_count. 2 is sufficient to verify "handle fires" and -/// "handle does not fire after stop/poison" without state explosion. -const MAX_HANDLE: u8 = 2; - -// ── Per-actor state ────────────────────────────────────────────────────────── - -/// Lifecycle state for one actor. No mailbox — events are modeled as actions. -#[derive(Clone, Debug, Hash, PartialEq, Eq)] -struct ActorState { - started: bool, - stopping: bool, - poisoned: bool, - suspended: bool, - alive: bool, - - on_start_count: u8, - handle_count: u8, - on_stop_count: u8, -} - -impl ActorState { - fn new() -> Self { - Self { - started: false, - stopping: false, - poisoned: false, - suspended: false, - alive: true, - on_start_count: 0, - handle_count: 0, - on_stop_count: 0, - } - } -} - -// ── Runtime state ──────────────────────────────────────────────────────────── - -#[derive(Clone, Debug, Hash, PartialEq, Eq)] -struct RuntimeState { - actors: [ActorState; NUM_ACTORS], -} - -impl RuntimeState { - fn init() -> Self { - Self { - actors: std::array::from_fn(|_| ActorState::new()), - } - } -} - -// ── Actions ────────────────────────────────────────────────────────────────── - -#[derive(Clone, Debug, Hash, PartialEq, Eq)] -enum RuntimeAction { - /// Fire on_start for an actor (first tick). - Start(usize), - /// Deliver a message — increments handle_count. - Handle(usize), - /// Actor's handler panics — sets poisoned. - PanicHandle(usize), - /// Actor's on_start panics — sets started + poisoned, handle_count stays 0. - PanicStart(usize), - /// Stop signal arrives — sets stopping. - Stop(usize), - /// Suspend signal — sets suspended. - Suspend(usize), - /// Resume signal delivered to suspended actor. - Resume(usize), - /// Run cleanup_dead for an actor (fires on_stop if eligible, marks not alive). - Cleanup(usize), -} - -// ── Stateright Model ───────────────────────────────────────────────────────── - -#[derive(Clone)] -struct LifecycleModel; - -impl Model for LifecycleModel { - type State = RuntimeState; - type Action = RuntimeAction; - - fn init_states(&self) -> Vec { - vec![RuntimeState::init()] - } - - fn actions(&self, state: &Self::State, actions: &mut Vec) { - for idx in 0..NUM_ACTORS { - let a = &state.actors[idx]; - - if !a.alive { - continue; - } - - // Start: only if not yet started and not skipped by production logic - if !a.started && !should_skip_actor(a.poisoned, a.stopping, a.suspended) { - actions.push(RuntimeAction::Start(idx)); - actions.push(RuntimeAction::PanicStart(idx)); - } - - // Handle/PanicHandle: only if started, not skipped, handle under cap - if a.started && !should_skip_actor(a.poisoned, a.stopping, a.suspended) { - if a.handle_count < MAX_HANDLE { - actions.push(RuntimeAction::Handle(idx)); - actions.push(RuntimeAction::PanicHandle(idx)); - } - } - - // Stop: only if started and not already stopping. - // Production justification: StopSignal goes through mailbox, - // processed AFTER on_start; ctx.stop_self() requires started. - if a.started && !a.stopping { - actions.push(RuntimeAction::Stop(idx)); - } - - // Suspend: only if started, not suspended, not stopping/poisoned - if a.started && !a.suspended && !a.stopping && !a.poisoned { - actions.push(RuntimeAction::Suspend(idx)); - } - - // Resume: only if suspended - if a.suspended { - actions.push(RuntimeAction::Resume(idx)); - } - - // Cleanup: only if stopping or poisoned - if a.stopping || a.poisoned { - actions.push(RuntimeAction::Cleanup(idx)); - } - } - } - - fn next_state(&self, state: &Self::State, action: Self::Action) -> Option { - let mut next = state.clone(); - - match action { - RuntimeAction::Start(idx) => { - let a = &mut next.actors[idx]; - a.on_start_count += 1; - a.started = true; - } - RuntimeAction::Handle(idx) => { - let a = &mut next.actors[idx]; - a.handle_count += 1; - } - RuntimeAction::PanicHandle(idx) => { - let a = &mut next.actors[idx]; - a.handle_count += 1; - a.poisoned = true; - } - RuntimeAction::PanicStart(idx) => { - let a = &mut next.actors[idx]; - a.on_start_count += 1; - a.started = true; - a.poisoned = true; - } - RuntimeAction::Stop(idx) => { - next.actors[idx].stopping = true; - } - RuntimeAction::Suspend(idx) => { - next.actors[idx].suspended = true; - } - RuntimeAction::Resume(idx) => { - next.actors[idx].suspended = false; - } - RuntimeAction::Cleanup(idx) => { - let a = &mut next.actors[idx]; - // Use production decision function - if is_on_stop_eligible(a.stopping, a.poisoned) { - a.on_stop_count += 1; - } - a.alive = false; - } - } - - // Prune no-change transitions - if next == *state { - return None; - } - - Some(next) - } - - fn properties(&self) -> Vec> { - vec![ - // ── G4 Safety Properties ────────────────────────────────────── - - // G4a: on_start fires at most once per actor - Property::::always("G4a: on_start_count <= 1", |_, state| { - state.actors.iter().all(|a| a.on_start_count <= 1) - }), - // G4a: no handle before on_start - Property::::always("G4a: no handle before start", |_, state| { - state - .actors - .iter() - .all(|a| !(a.handle_count > 0 && a.on_start_count == 0)) - }), - // G4b: if poisoned or stopping, no further handle calls - // (encoded in action generation, verified here as invariant) - Property::::always("G4b: poisoned implies no on_stop", |_, state| { - state - .actors - .iter() - .all(|a| !(a.poisoned && a.on_stop_count > 0)) - }), - // G4c: on_stop fires at most once - Property::::always("G4c: on_stop_count <= 1", |_, state| { - state.actors.iter().all(|a| a.on_stop_count <= 1) - }), - // G4c: on_stop only fires when stopping && !poisoned - Property::::always("G4c: on_stop implies stopping && !poisoned", |_, state| { - state.actors.iter().all(|a| { - if a.on_stop_count == 1 { - a.stopping && !a.poisoned - } else { - true - } - }) - }), - // G4d: on_stop implies actor is removed (no further handle possible) - Property::::always("G4d: on_stop implies not alive", |_, state| { - state - .actors - .iter() - .all(|a| if a.on_stop_count > 0 { !a.alive } else { true }) - }), - // G4e: on_stop implies the actor was started (no cleanup of - // never-initialized actors). Enabled by the Stop guard requiring - // `started`, which mirrors production: StopSignal goes through - // the mailbox and is processed after on_start. - Property::::always("G4e: on_stop implies started", |_, state| { - state - .actors - .iter() - .all(|a| if a.on_stop_count > 0 { a.started } else { true }) - }), - // ── G5 Safety Properties ────────────────────────────────────── - - // G5: every actor's lifecycle invariants hold independently, - // regardless of what happened to other actors. - Property::::always("G5: per-actor invariants hold", |_, state| { - for a in &state.actors { - // Each actor's lifecycle is self-consistent - if a.on_start_count > 1 || a.on_stop_count > 1 { - return false; - } - if a.handle_count > 0 && a.on_start_count == 0 { - return false; - } - if a.alive && a.on_stop_count > 0 { - return false; - } - if a.poisoned && a.on_stop_count > 0 { - return false; - } - } - true - }), - // G5: a panic on one actor doesn't corrupt another's started flag - Property::::always("G5: panic isolation on started", |_, state| { - for i in 0..NUM_ACTORS { - if state.actors[i].poisoned { - for j in 0..NUM_ACTORS { - if i != j { - let other = &state.actors[j]; - // Other actor's lifecycle must be internally consistent - if other.handle_count > 0 && !other.started { - return false; - } - } - } - } - } - true - }), - // ── Liveness Canaries ───────────────────────────────────────── - - // L1: handle_count > 0 is reachable - Property::::sometimes("L1: handle reachable", |_, state| { - state.actors.iter().any(|a| a.handle_count > 0) - }), - // L2: on_stop_count == 1 is reachable - Property::::sometimes("L2: on_stop reachable", |_, state| { - state.actors.iter().any(|a| a.on_stop_count == 1) - }), - // L3: one actor poisoned while another has handle_count > 0 - Property::::sometimes("L3: poison + sibling handle", |_, state| { - let any_poisoned = state.actors.iter().any(|a| a.poisoned); - let any_handled = state - .actors - .iter() - .any(|a| a.handle_count > 0 && !a.poisoned); - any_poisoned && any_handled - }), - // L4: on_start panic reachable (poisoned with handle_count == 0) - Property::::sometimes("L4: on_start panic reachable", |_, state| { - state - .actors - .iter() - .any(|a| a.poisoned && a.handle_count == 0 && a.on_start_count > 0) - }), - ] - } -} - -// ── Test ───────────────────────────────────────────────────────────────────── - -#[test] -#[ignore] // Exhaustive proof — run deliberately with `cargo test -- --ignored` -fn lifecycle_fault_isolation_model_check() { - let result = LifecycleModel.checker().spawn_dfs().join(); - - let unique_states = result.unique_state_count(); - let max_depth = result.max_depth(); - println!( - "Stateright G4/G5: explored {} unique states, max depth {}", - unique_states, max_depth, - ); - - result.assert_properties(); - - // Sanity: the model explored a meaningful state space. - assert!( - unique_states > 100, - "Model explored too few states ({unique_states}); bounds may be too tight", - ); -} diff --git a/src/lib.rs b/src/lib.rs index 6512853..9396d65 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -36,9 +36,6 @@ pub(crate) fn get_random(buf: &mut [u8]) { getrandom::getrandom(buf).unwrap() } -#[cfg(any(kani, test))] -mod guarantees; - #[cfg(all(feature = "no_random", not(feature = "getrandom")))] pub(crate) fn get_random(buf: &mut [u8]) { use core::sync::atomic::{AtomicUsize, Ordering}; diff --git a/src/worker.rs b/src/worker.rs index 22ebb2b..12e4420 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -17,7 +17,6 @@ use crate::stats::{ActorSnapshot, TickTiming, WorkerStats}; use crate::extension::WorkerExtension; -// Extracted pure functions for use in kani to prove guarantees /// Whether an actor should be skipped during `tick_all`. pub(crate) fn should_skip_actor(poisoned: bool, stopping: bool, suspended: bool) -> bool { @@ -826,3 +825,167 @@ impl ActorPool { })); } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::actor::StopReason; + + #[test] + fn lifecycle_decision_helpers_cover_all_inputs() { + let mut skip_combinations = 0; + for poisoned in [false, true] { + for stopping in [false, true] { + for suspended in [false, true] { + assert_eq!( + should_skip_actor(poisoned, stopping, suspended), + poisoned || stopping || suspended + ); + skip_combinations += 1; + } + } + } + assert_eq!(skip_combinations, 8); + + let mut on_stop_combinations = 0; + for stopping in [false, true] { + for poisoned in [false, true] { + assert_eq!( + is_on_stop_eligible(stopping, poisoned), + stopping && !poisoned + ); + on_stop_combinations += 1; + } + } + assert_eq!(on_stop_combinations, 4); + + let mut reason_combinations = 0; + for poisoned in [false, true] { + for has_exit_value in [false, true] { + let expected = if poisoned { + StopReason::Panicked + } else if has_exit_value { + StopReason::Completed + } else { + StopReason::Normal + }; + assert_eq!(determine_stop_reason(poisoned, has_exit_value), expected); + reason_combinations += 1; + } + } + assert_eq!(reason_combinations, 4); + } + + #[test] + fn bounded_lifecycle_model_preserves_callback_invariants() { + #[derive(Clone, Copy)] + enum Step { + Tick, + RequestStop, + Poison, + Suspend, + Resume, + Cleanup, + } + + #[derive(Clone, Copy, Default)] + struct ActorProbe { + started: bool, + stopping: bool, + poisoned: bool, + suspended: bool, + removed: bool, + on_start_count: u8, + handle_count: u8, + on_stop_count: u8, + handled_after_on_stop: bool, + } + + impl ActorProbe { + fn apply(&mut self, step: Step) { + if self.removed { + return; + } + + match step { + Step::Tick => { + let eligible = + !should_skip_actor(self.poisoned, self.stopping, self.suspended); + if !self.started && eligible { + self.started = true; + self.on_start_count += 1; + } + if self.started && eligible { + if self.on_stop_count > 0 { + self.handled_after_on_stop = true; + } + self.handle_count += 1; + } + } + Step::RequestStop => { + self.stopping = true; + } + Step::Poison => { + self.poisoned = true; + } + Step::Suspend => { + self.suspended = true; + } + Step::Resume => { + self.suspended = false; + } + Step::Cleanup => { + if self.stopping || self.poisoned { + if is_on_stop_eligible(self.stopping, self.poisoned) { + self.on_stop_count += 1; + } + self.removed = true; + } + } + } + } + + fn assert_invariants(self) { + assert!(self.on_start_count <= 1, "on_start fired more than once"); + assert!(self.on_stop_count <= 1, "on_stop fired more than once"); + assert!( + !self.handled_after_on_stop, + "handle fired after on_stop cleanup" + ); + if self.handle_count > 0 { + assert!(self.started, "handle fired before on_start"); + } + if self.on_stop_count > 0 { + assert!(self.removed, "on_stop fired without cleanup"); + assert!(!self.poisoned, "poisoned actor ran on_stop"); + } + } + } + + fn walk(depth: usize, state: ActorProbe, checked: &mut usize) { + state.assert_invariants(); + *checked += 1; + + if depth == 0 { + return; + } + + for step in [ + Step::Tick, + Step::RequestStop, + Step::Poison, + Step::Suspend, + Step::Resume, + Step::Cleanup, + ] { + let mut next = state; + next.apply(step); + walk(depth - 1, next, checked); + } + } + + let mut checked = 0; + walk(6, ActorProbe::default(), &mut checked); + assert_eq!(checked, 55_987); + } +} diff --git a/tests/actor_lifecycle.rs b/tests/actor_lifecycle.rs index 9107ea3..28ca04d 100644 --- a/tests/actor_lifecycle.rs +++ b/tests/actor_lifecycle.rs @@ -1,13 +1,13 @@ //! Actor Lifecycle Tests — birth, life, death of individual actors. //! -//! Covers: spawning, on_start, parent-child delegation, graceful stop, -//! panic isolation, dead actor cleanup, and watching (ActorExited). +//! Covers: spawning, on_start, lifecycle decision paths, parent-child delegation, +//! graceful stop, panic isolation, dead actor cleanup, and watching (ActorExited). mod common; use common::*; -use std::sync::Arc; use parking_lot::Mutex; +use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; // ── Local actors ──────────────────────────────────────────────────────────── @@ -224,6 +224,115 @@ impl ActorInterface for Sleeper { fn handle(&mut self, _ctx: &Ctx, _msg: Noop) {} } +#[derive(Clone)] +struct Work; + +#[derive(Clone, Debug, PartialEq, Eq)] +struct WorkCount(usize); + +struct StartStopCountingActor { + started: Arc, + stopped: Arc, + handled: Arc, +} + +impl ActorInterface for StartStopCountingActor { + type Incoming = Work; + type Response = (); + + fn on_start(&mut self, _ctx: &Ctx) { + self.started.fetch_add(1, Ordering::SeqCst); + } + + fn handle(&mut self, _ctx: &Ctx, _msg: Work) { + self.handled.fetch_add(1, Ordering::SeqCst); + } + + fn on_stop(&mut self, _ctx: &Ctx) { + self.stopped.fetch_add(1, Ordering::SeqCst); + } +} + +struct StopReportingCounter { + handled: usize, + report_to: ActorAddress, +} + +impl ActorInterface for StopReportingCounter { + type Incoming = Work; + type Response = (); + + fn handle(&mut self, _ctx: &Ctx, _msg: Work) { + self.handled += 1; + } + + fn on_stop(&mut self, ctx: &Ctx) { + let _ = ctx.send(self.report_to, WorkCount(self.handled)); + } +} + +struct PanicOnWorkNumber { + handled: usize, + panic_at: usize, +} + +impl ActorInterface for PanicOnWorkNumber { + type Incoming = Work; + type Response = (); + + fn handle(&mut self, _ctx: &Ctx, _msg: Work) { + self.handled += 1; + if self.handled == self.panic_at { + panic!("intentional panic at work item {}", self.panic_at); + } + } +} + +struct PanicOnStartWithStopReport { + handled: Arc, + stopped: Arc, +} + +impl ActorInterface for PanicOnStartWithStopReport { + type Incoming = Work; + type Response = (); + + fn on_start(&mut self, _ctx: &Ctx) { + panic!("intentional on_start panic"); + } + + fn handle(&mut self, _ctx: &Ctx, _msg: Work) { + self.handled.fetch_add(1, Ordering::SeqCst); + } + + fn on_stop(&mut self, _ctx: &Ctx) { + self.stopped.fetch_add(1, Ordering::SeqCst); + } +} + +struct PanicOnHandleWithStopReport { + stopped: Arc, +} + +impl ActorInterface for PanicOnHandleWithStopReport { + type Incoming = Work; + type Response = (); + + fn handle(&mut self, _ctx: &Ctx, _msg: Work) { + panic!("intentional handle panic"); + } + + fn on_stop(&mut self, _ctx: &Ctx) { + self.stopped.fetch_add(1, Ordering::SeqCst); + } +} + +fn drain_work_counts(inbox: &Inbox) -> Vec { + std::iter::from_fn(|| inbox.try_recv()) + .map(|WorkCount(count)| count) + .collect() +} + // ═══════════════════════════════════════════════════════════════════════════ // Tests // ═══════════════════════════════════════════════════════════════════════════ @@ -731,3 +840,257 @@ fn watch_edge_cases() { ); } +/// Core lifecycle decision paths are observable without the old guarantee module: +/// healthy actors handle work, stopping actors skip later work and run `on_stop` +/// once, and poisoned actors never run `handle` or `on_stop` after poisoning. +#[test] +fn lifecycle_decision_paths_match_runtime_behavior() { + for msg_count in 1..=5 { + let rt = Runtime::new(RuntimeConfig::default()); + let started = Arc::new(AtomicUsize::new(0)); + let stopped = Arc::new(AtomicUsize::new(0)); + let handled = Arc::new(AtomicUsize::new(0)); + + let addr = rt + .spawn(StartStopCountingActor { + started: started.clone(), + stopped: stopped.clone(), + handled: handled.clone(), + }) + .unwrap(); + rt.tick(); + + for _ in 0..msg_count { + rt.send_to(addr, Work).unwrap(); + } + tick_n(&rt, msg_count + 3); + assert_eq!(started.load(Ordering::SeqCst), 1); + assert_eq!(handled.load(Ordering::SeqCst), msg_count); + + rt.stop_actor(addr).unwrap(); + tick_n(&rt, 3); + assert_eq!(stopped.load(Ordering::SeqCst), 1); + } + + for msg_count in 1..=5 { + let rt = Runtime::new(RuntimeConfig::default()); + let started = Arc::new(AtomicUsize::new(0)); + let stopped = Arc::new(AtomicUsize::new(0)); + let handled = Arc::new(AtomicUsize::new(0)); + + let addr = rt + .spawn(StartStopCountingActor { + started, + stopped: stopped.clone(), + handled: handled.clone(), + }) + .unwrap(); + rt.tick(); + + rt.stop_actor(addr).unwrap(); + for _ in 0..msg_count { + let _ = rt.send_to(addr, Work); + } + tick_n(&rt, 5); + assert_eq!(handled.load(Ordering::SeqCst), 0); + assert_eq!(stopped.load(Ordering::SeqCst), 1); + } + + for msg_count in 1..=5 { + let rt = Runtime::new(RuntimeConfig::default()); + let handled = Arc::new(AtomicUsize::new(0)); + let stopped = Arc::new(AtomicUsize::new(0)); + + let addr = rt + .spawn(PanicOnStartWithStopReport { + handled: handled.clone(), + stopped: stopped.clone(), + }) + .unwrap(); + rt.tick(); + + for _ in 0..msg_count { + let _ = rt.send_to(addr, Work); + } + tick_n(&rt, msg_count + 3); + assert_eq!(handled.load(Ordering::SeqCst), 0); + assert_eq!(stopped.load(Ordering::SeqCst), 0); + } + + let rt = Runtime::new(RuntimeConfig::default()); + let stopped = Arc::new(AtomicUsize::new(0)); + let addr = rt + .spawn(PanicOnHandleWithStopReport { + stopped: stopped.clone(), + }) + .unwrap(); + rt.tick(); + + rt.send_to(addr, Work).unwrap(); + tick_n(&rt, 5); + assert_eq!(stopped.load(Ordering::SeqCst), 0); +} + +/// Deterministic replacements for the old property guarantee: delayed panics, +/// start panics, and same-tick multi-panics must not reduce sibling progress. +#[test] +fn panicking_actors_do_not_affect_sibling_progress() { + for (healthy_count, msg_count, panic_at) in [(2, 1, 1), (4, 65, 17), (8, 130, 1)] { + let rt = Runtime::new(RuntimeConfig::default()); + let report_inbox = rt.new_inbox::().unwrap(); + let report_to = *report_inbox.addr(); + + let mut healthy = Vec::with_capacity(healthy_count); + for _ in 0..healthy_count { + healthy.push( + rt.spawn(StopReportingCounter { + handled: 0, + report_to, + }) + .unwrap(), + ); + } + let panicker = rt + .spawn(PanicOnWorkNumber { + handled: 0, + panic_at, + }) + .unwrap(); + rt.tick(); + + for _ in 0..msg_count { + for &addr in &healthy { + rt.send_to(addr, Work).unwrap(); + } + rt.send_to(panicker, Work).unwrap(); + } + tick_n(&rt, (msg_count / 64) + 10); + + for &addr in &healthy { + rt.stop_actor(addr).unwrap(); + } + tick_n(&rt, 3); + + let reports = drain_work_counts(&report_inbox); + assert_eq!(reports.len(), healthy_count); + assert!( + reports.iter().all(|&count| count == msg_count), + "healthy reports were {reports:?}, expected every actor to process {msg_count}" + ); + } +} + +#[test] +fn on_start_panic_does_not_block_siblings() { + for (before_count, after_count, msgs_each) in [(1, 1, 1), (4, 4, 25), (8, 3, 70)] { + let rt = Runtime::new(RuntimeConfig::default()); + let started = Arc::new(AtomicUsize::new(0)); + let stopped = Arc::new(AtomicUsize::new(0)); + let handled = Arc::new(AtomicUsize::new(0)); + let panic_handled = Arc::new(AtomicUsize::new(0)); + let panic_stopped = Arc::new(AtomicUsize::new(0)); + + let mut siblings = Vec::with_capacity(before_count + after_count); + for _ in 0..before_count { + siblings.push( + rt.spawn(StartStopCountingActor { + started: started.clone(), + stopped: stopped.clone(), + handled: handled.clone(), + }) + .unwrap(), + ); + } + + let _panic_addr = rt + .spawn(PanicOnStartWithStopReport { + handled: panic_handled.clone(), + stopped: panic_stopped.clone(), + }) + .unwrap(); + + for _ in 0..after_count { + siblings.push( + rt.spawn(StartStopCountingActor { + started: started.clone(), + stopped: stopped.clone(), + handled: handled.clone(), + }) + .unwrap(), + ); + } + tick_n(&rt, 3); + + assert_eq!(started.load(Ordering::SeqCst), siblings.len()); + assert_eq!(panic_handled.load(Ordering::SeqCst), 0); + assert_eq!(panic_stopped.load(Ordering::SeqCst), 0); + + for _ in 0..msgs_each { + for &addr in &siblings { + rt.send_to(addr, Work).unwrap(); + } + } + tick_n(&rt, (msgs_each / 64) + 5); + assert_eq!(handled.load(Ordering::SeqCst), siblings.len() * msgs_each); + + for &addr in &siblings { + rt.stop_actor(addr).unwrap(); + } + tick_n(&rt, 3); + assert_eq!(stopped.load(Ordering::SeqCst), siblings.len()); + } +} + +#[test] +fn multiple_panics_in_same_tick_preserve_healthy_actors() { + for (healthy_count, panic_count, msgs_each) in [(2, 2, 1), (6, 4, 70)] { + let rt = Runtime::new(RuntimeConfig::default()); + let report_inbox = rt.new_inbox::().unwrap(); + let report_to = *report_inbox.addr(); + + let mut healthy = Vec::with_capacity(healthy_count); + for _ in 0..healthy_count { + healthy.push( + rt.spawn(StopReportingCounter { + handled: 0, + report_to, + }) + .unwrap(), + ); + } + + let mut panickers = Vec::with_capacity(panic_count); + for _ in 0..panic_count { + panickers.push( + rt.spawn(PanicOnWorkNumber { + handled: 0, + panic_at: 1, + }) + .unwrap(), + ); + } + rt.tick(); + + for _ in 0..msgs_each { + for &addr in &healthy { + rt.send_to(addr, Work).unwrap(); + } + for &addr in &panickers { + rt.send_to(addr, Work).unwrap(); + } + } + tick_n(&rt, (msgs_each / 64) + 10); + + for &addr in &healthy { + rt.stop_actor(addr).unwrap(); + } + tick_n(&rt, 3); + + let reports = drain_work_counts(&report_inbox); + assert_eq!(reports.len(), healthy_count); + assert!( + reports.iter().all(|&count| count == msgs_each), + "healthy reports were {reports:?}, expected every actor to process {msgs_each}" + ); + } +}