zacheryasc/swactor
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(engine): substrate-neutral execution engine abstraction

Browse source 
Introduce the swactor engine: a swactor-owned composite that retains a
selected execution substrate, drives the core runtime, and hosts the
async/blocking/timer work that backs actors. Integrations receive one
cloneable EngineHandle and never construct or borrow a raw Tokio
runtime/handle.

Engine crate (crates/engine):
- The contract: spawn / spawn_blocking / timer / interval / now, a
  per-implementation capability model with construction-time binding
  (require()), and engine-owned time. The engine owns all progression;
  actor handlers stay synchronous and never .await.
- TokioBackend owns the Tokio runtime and schedules core ticks and
  supporting futures on it; SteppingBackend is a single-threaded
  deterministic scheduler with virtual time (the non-Tokio portability
  proof). Core is driven through its existing tick() surface; a
  self-rescheduling CoreDriver is installed at construction and is the
  sole place permitted to call try_tick.

iroh-driver:
- Receives an EngineHandle instead of a raw Tokio Handle. Accepts,
  reads, dials, writes, endpoint construction, and teardown schedule
  through it; required capabilities (tasks/timers/io) are validated
  before the endpoint binds. Engine-hosted interval pumps drive
  actor-bridge, datastream, and edge ingress.

myelin:
- One node/orchestrator engine owns core, protocol tick injection, and
  transport progression; the application loop only drains
  integration-owned queues. Stage-shard process readers, delayed actor
  messages, helper stdout/stderr, prompt RPC, and CPU sampling all
  schedule through the engine (spawn_blocking / engine tasks / timers).
- Removed the split-engine APIs: install_actor_bridge_pump(period) and
  spawn_protocol_ticker(period) use each component's stored engine;
  deleted the no-op pump_network callback and its plumbing; deleted the
  dashboard raw-Tokio/standalone-runtime conveniences.

Enforcement:
- A clippy disallowed-methods boundary forbids direct runtime/scheduling/
  time/core-driving bypasses, denied in swactor-engine, iroh-driver, and
  myelin. Retained excluded uses (VastAI provider, provider process
  supervision/log capture, OS-signal/stdin/process-control sequencing)
  carry narrow allowances with reasons.

Verification:
- Engine contract + unit tests (incl. the SteppingBackend portability
  proof), iroh integration tests (capability rejection before binding,
  multi-node actor behavior), and a production execution-composition
  smoke test that observes engine-driven actor progress with no ambient
  Tokio runtime and no manual tick/pump. Workspace all-target/all-feature
  clippy and tests are green.

Specs co-located with their crates: ENGINE_SPEC.md in crates/engine,
IROH_DRIVER_SPEC.md in crates/iroh-driver. VastAI remains explicitly out
of scope pending its separate redesign.
This commit is contained in:
Zachery Aaron Shores-Chmielewski 2026-08-11 00:23:03 +04:00
parent f8fc594b95
commit b887e941cb
41 changed files with 3518 additions and 980 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
Cargo.lock generated
View file

@ -2149,6 +2149,7 @@ dependencies = [
"serde",
"serde_json",
"swactor",
"swactor-engine",
"swactor-transport",
"tokio",
]
@ -2534,6 +2535,7 @@ dependencies = [
"serde_json",
"signal-hook",
"swactor",
"swactor-engine",
"swactor-process",
"swactor-transport",
"swactor-vastai",
@ -4401,6 +4403,15 @@ dependencies = [
"web-time 0.2.4",
]
[[package]]
name = "swactor-engine"
version = "0.1.0"
dependencies = [
"parking_lot",
"swactor",
"tokio",
]
[[package]]
name = "swactor-process"
version = "0.1.0"

2
Cargo.toml
View file

@ -1,6 +1,7 @@
[workspace]
members = [
".",
"crates/engine",
"crates/bindings/python",
"crates/bindings/wasm-runtime",
"crates/process",
@ -17,6 +18,7 @@ members = [
]
default-members = [
".",
"crates/engine",
"crates/process",
"crates/provisioning",
"crates/transport",

1
apps/myelin/Cargo.toml
View file

@ -18,6 +18,7 @@ dashboard = { path = "../../crates/dashboard" }
serde_json = "1"
serde = { version = "1", features = ["derive"] }
swactor = { path = "../..", features = ["serde", "transport"] }
swactor-engine = { path = "../../crates/engine" }
swactor-transport = { path = "../../crates/transport" }
swactor-process = { path = "../../crates/process" }
distribution = { path = "../../crates/distribution" }

4
apps/myelin/src/chat/node_image.rs
View file

@ -621,6 +621,8 @@ enum CommandOutputLine {
Stderr(String),
}
// container image build is provisioning infrastructure, out of scope (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn spawn_line_reader<R>(
reader: R,
to_line: fn(String) -> CommandOutputLine,
@ -660,6 +662,8 @@ fn drain_command_lines(
}
}
// container image build is provisioning infrastructure, out of scope (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn run_status_command(
root: &Path,
program: &str,

12
apps/myelin/src/chat/runtime.rs
View file

@ -120,6 +120,8 @@ impl Drop for RuntimeEnvGuard {
}
}
// blocking user-stdin thread is process control, out of scope (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn run<I>(args: I) -> Result<(), String>
where
I: IntoIterator<Item = String>,
@ -1356,6 +1358,8 @@ struct InProcessOrch {
cleaned: bool,
}
// synchronous process-control readiness sequencing; the engine drives all background work (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn wait_for_rpc_ready(
rpc_addr: &str,
mut check_dead: impl FnMut() -> Result<(), String>,
@ -1384,6 +1388,8 @@ fn wait_for_rpc_ready(
}
impl InProcessOrch {
// spawns the orchestrator process; process control, out of scope (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn spawn(config: &Config, image_ref: &str) -> Result<Self, String> {
let args = config.orchestrator_cli_args(image_ref);
let (stop_tx, stop_rx) = mpsc::channel();
@ -1412,6 +1418,8 @@ impl InProcessOrch {
})
}
// synchronous process-control readiness sequencing; the engine drives all background work (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn shutdown(&mut self) {
if self.cleaned {
return;
@ -1509,6 +1517,8 @@ impl OrchChild {
// The orchestrator shutdown spec is still pending. Replace this with the approved
// shutdown contract when it is finalized; do not add private stdin commands here.
// synchronous process-control readiness sequencing; the engine drives all background work (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn shutdown(&mut self) {
if self.cleaned {
return;
@ -2102,6 +2112,8 @@ fn ensure_runtime_binary(
}
}
// top-level OS signal handling is process control, out of scope (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn install_signal_handlers() -> Result<(), String> {
#[cfg(target_os = "linux")]
{

6
apps/myelin/src/lib.rs
View file

@ -1,3 +1,9 @@
// Engine boundary enforcement: disallowed scheduling/time/core-driving methods
// are hard errors in this crate (ENGINE_SPEC.md §2). The VastAI
// provider adapter carries a module-level `#![allow]` pending its separate
// redesign; unit tests that drive a raw Runtime in isolation are exempted
// locally.
#![deny(clippy::disallowed_methods)]
#![recursion_limit = "256"]
#[cfg(test)]

377
apps/myelin/src/node/worker_node_runtime.rs
View file

File diff suppressed because it is too large Load diff

4
apps/myelin/src/observability/provisioning_logs.rs
View file

@ -75,6 +75,8 @@ impl BootstrapDatastreamBridge {
});
}
// provider log capture is out of scope (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
pub(crate) fn spawn_stdout_reader<R>(&self, stdout: R) -> JoinHandle<()>
where
R: Read + Send + 'static,
@ -83,6 +85,8 @@ impl BootstrapDatastreamBridge {
thread::spawn(move || bridge.read_stdout(stdout))
}
// provider log capture is out of scope (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
pub(crate) fn spawn_stderr_reader<R>(&self, stderr: R) -> JoinHandle<()>
where
R: Read + Send + 'static,

168
apps/myelin/src/orchestration/app.rs
View file

@ -66,7 +66,7 @@ use iroh_driver::{EndpointAddrMask, MVP_IROH_ENDPOINT_ADDR_MASK_ENV, advertised_
use parking_lot::Mutex;
use serde_json::{Value, json};
use swactor::actor::ActorAddress;
use swactor_engine::{Engine, EngineHandle, TokioBackend, TokioConfig};
const DEFAULT_IMAGE: &str = "myelin-node:latest";
const MYELIN_RUNTIME_CONFIG_ENV: &str = "MYELIN_RUNTIME_CONFIG";
const CACHED_MODEL_HOST_ENV: &str = "MYELIN_CACHED_MODEL_HOST_PATH";
@ -216,30 +216,47 @@ where
None
};
let tokio = match tokio::runtime::Runtime::new() {
Ok(runtime) => {
let actors_channel = orch_datastream.channel_by_name("runtime.actors");
let orch_stats_hook = orch_datastream.producer.stats_hook_on(actors_channel);
// Build the core swactor runtime, then hand it to the engine. The engine
// owns both the runtime (it drives actor progression) and the Tokio
// substrate (it schedules all background work). After this point the engine
// is the sole owner of Tokio and core progression — no raw handles are
// passed to components (ENGINE_SPEC.md).
let (runtime, codec, transport_router) = DistributionRuntimeStack::build_runtime(
|registry| {
register_myelin_actor_codecs(registry);
datastream::wire::register_datastream_codec(registry);
},
Some(orch_stats_hook),
);
let engine = match TokioBackend::new(TokioConfig::default())
.and_then(|backend| Engine::new(runtime.clone(), backend))
{
Ok(engine) => {
bootstrap(
&mut orch_datastream,
None,
"tokio_runtime",
"engine",
"ready",
json!({"runtime":"tokio"}),
json!({"backend":"tokio","owns":"core+substrate"}),
);
runtime
engine
}
Err(error) => {
bootstrap(
&mut orch_datastream,
None,
"tokio_runtime",
"engine",
"failed",
json!({"error":error.to_string()}),
);
return Err(format!("tokio runtime: {error}"));
return Err(format!("create engine: {error}"));
}
};
let mut driver = match IrohDriver::with_handle(
tokio.handle().clone(),
let mut driver = match IrohDriver::with_engine(
engine.handle(),
IrohDriverConfig {
secret_key: None,
relay_mode: config.relay.mode.clone(),
@ -284,16 +301,13 @@ where
"connectivity_preflight":"ready",
}),
);
let actors_channel = orch_datastream.channel_by_name("runtime.actors");
let orch_stats_hook = orch_datastream.producer.stats_hook_on(actors_channel);
let stack = DistributionRuntimeStack::new_with_codecs(
let stack = DistributionRuntimeStack::new_from_runtime(
runtime,
codec,
transport_router,
driver.node_id(),
DistributedNodeConfig::default(),
|registry| {
register_myelin_actor_codecs(registry);
datastream::wire::register_datastream_codec(registry);
},
Some(orch_stats_hook),
engine.handle(),
);
bootstrap(
&mut orch_datastream,
@ -316,13 +330,18 @@ where
stack.actors.swim,
stack.relay_mirror.clone(),
stack.route_view.clone(),
stack.outbox.clone(),
);
// Engine owns protocol tick injection and core progression; the application
// loop only drains integration-owned queues (ENGINE_SPEC.md).
stack.spawn_protocol_ticker(PUMP_INTERVAL);
driver.install_actor_bridge_pump(PUMP_INTERVAL);
bootstrap(
&mut orch_datastream,
None,
"actor_bridge",
"ready",
json!({"transport":"iroh","routes":"attached"}),
json!({"transport":"iroh","routes":"attached","protocol_ticker":"engine-hosted"}),
);
let (frame_tx, frame_rx) = mpsc::channel::<CollectedDatastreamFrame>();
@ -333,7 +352,7 @@ where
"ready",
json!({"alpn":String::from_utf8_lossy(DATASTREAM_ALPN)}),
);
let dashboard = DashboardSupport::start(config.dashboard)?;
let dashboard = DashboardSupport::start(config.dashboard, &engine.handle())?;
bootstrap(
&mut orch_datastream,
dashboard.as_ref(),
@ -496,7 +515,7 @@ where
}),
);
let rpc_addr = match spawn_prompt_rpc(config.rpc_bind, work_tx, config.default_max_tokens) {
let rpc_addr = match spawn_prompt_rpc(&engine.handle(), config.rpc_bind, work_tx, config.default_max_tokens) {
Ok(addr) => {
bootstrap(
&mut orch_datastream,
@ -2034,6 +2053,8 @@ struct RuntimeReadyAckLoop<'a> {
orchestrator_actor: ActorAddress,
}
// synchronous process-control/orchestration sequencing; the engine drives all background work (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn wait_for_runtime_ready_acks(
ctx: RuntimeReadyAckLoop<'_>,
targets: &[RuntimeReadyAckTarget],
@ -2083,7 +2104,7 @@ fn wait_for_runtime_ready_acks(
let mut last_send = None::<Instant>;
while !pending.is_empty() {
pump(driver, stack, frame_tx);
pump(driver, frame_tx);
drain_orch_stdio_capture(
orch_stdio_rx,
orch_datastream,
@ -2171,7 +2192,6 @@ fn wait_for_runtime_ready_acks(
}),
);
}
driver.drain_outbox(&stack.outbox);
last_send = Some(Instant::now());
}
thread::sleep(PUMP_INTERVAL);
@ -2232,6 +2252,8 @@ impl Drop for ProvisionedClusterGuard {
}
}
// provider lifecycle/provisioning is out of scope (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn start_and_provision_workers(
mut provisioner: Box<dyn ProvisionPlugin>,
config: &Config,
@ -2901,6 +2923,8 @@ fn stage_ring_spec_wire(spec: run_plan::RingSpec) -> StageRingSpecWire {
}
}
// synchronous process-control/orchestration sequencing; the engine drives all background work (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn wait_for_runtime_readies(
ctx: RuntimeReadyAckLoop<'_>,
expected_node_ids: &[u64],
@ -2923,7 +2947,7 @@ fn wait_for_runtime_readies(
let expected = expected_node_ids.iter().copied().collect::<BTreeSet<_>>();
let mut pending = BTreeMap::<u64, RuntimeReady>::new();
loop {
pump(driver, stack, frame_tx);
pump(driver, frame_tx);
emit_swim_transitions(
orch_datastream,
dashboard,
@ -2995,6 +3019,8 @@ fn wait_for_runtime_readies(
}
}
// synchronous process-control/orchestration sequencing; the engine drives all background work (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn wait_for_weights_loaded_count(
ctx: RuntimeReadyAckLoop<'_>,
expected_count: usize,
@ -3031,7 +3057,7 @@ fn wait_for_weights_loaded_count(
let mut stage_last_sends = BTreeMap::<u32, Instant>::new();
let mut load_progress = BTreeMap::<u64, StageLoadProgress>::new();
loop {
pump(driver, stack, frame_tx);
pump(driver, frame_tx);
emit_swim_transitions(orch_datastream, dashboard, run_id, node_id, stack);
emit_swim_probe_events(orch_datastream, dashboard, stack, "weights_loaded_wait");
drain_orch_stdio_capture(orch_stdio_rx, orch_datastream, dashboard, run_id, node_id);
@ -3363,7 +3389,7 @@ fn send_pipeline_stage_provision(
ctx.pipeline_coordinator,
ctx.stage_shard_plans,
)?;
pump(ctx.driver, ctx.stack, ctx.frame_tx);
pump(ctx.driver, ctx.frame_tx);
Ok(())
}
@ -3630,10 +3656,9 @@ fn load_phase_for_worker_event(event_type: &str) -> Option<&'static str> {
}
fn drain_datastream_connections(
driver: &mut IrohDriver,
driver: &IrohDriver,
frame_tx: &mpsc::Sender<CollectedDatastreamFrame>,
) {
driver.pump_datastream_ingress();
for read in driver.drain_datastream_reads() {
let mut channels = read
.header
@ -3953,6 +3978,8 @@ impl OrchStdioCapture {
Ok(unsafe { File::from_raw_fd(pipe_fds[0]) })
}
// provider log capture is out of scope (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn spawn_reader(file: File, stream: ProvisionLogStream, tx: mpsc::Sender<OrchStdioLine>) {
thread::spawn(move || {
let reader = BufReader::new(file);
@ -4001,12 +4028,11 @@ fn drain_orch_stdio_capture(
#[cfg(feature = "dashboard")]
struct DashboardSupport {
handle: dashboard::DashboardHandle,
_runtime: tokio::runtime::Runtime,
}
#[cfg(feature = "dashboard")]
impl DashboardSupport {
fn start(enabled: bool) -> Result<Option<Self>, String> {
fn start(enabled: bool, engine: &EngineHandle) -> Result<Option<Self>, String> {
if !enabled {
return Ok(None);
}
@ -4016,17 +4042,10 @@ impl DashboardSupport {
.parse::<u16>()
.map_err(|e| format!("invalid MYELIN_DASHBOARD_PORT={port:?}: {e}"))?;
}
let runtime = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.map_err(|e| format!("dashboard runtime: {e}"))?;
let handle = dashboard::DashboardHandle::new(config);
handle.register_view(Arc::new(MyelinClusterDashboardView::new()));
handle.spawn_http(runtime.handle());
Ok(Some(Self {
handle,
_runtime: runtime,
}))
engine.spawn(handle.http_server());
Ok(Some(Self { handle }))
}
fn publish_frame(&self, stream: &StreamId, channel: &str, frame: &Frame) {
@ -4047,7 +4066,7 @@ struct DashboardSupport;
#[cfg(not(feature = "dashboard"))]
impl DashboardSupport {
fn start(enabled: bool) -> Result<Option<Self>, String> {
fn start(enabled: bool, _engine: &EngineHandle) -> Result<Option<Self>, String> {
if enabled {
return Err(
"MYELIN_DASHBOARD requires building myelin-system with feature dashboard".to_owned(),
@ -4059,6 +4078,7 @@ impl DashboardSupport {
fn publish_frame(&self, _stream: &StreamId, _channel: &str, _frame: &Frame) {}
}
struct ChannelObservationSink {
tx: Mutex<mpsc::Sender<PluginObservation>>,
}
@ -4070,21 +4090,43 @@ impl PluginObservationSink for ChannelObservationSink {
}
fn spawn_prompt_rpc(
engine: &EngineHandle,
bind: SocketAddr,
work_tx: mpsc::Sender<PromptWork>,
default_max_tokens: u32,
) -> Result<SocketAddr, String> {
let listener = TcpListener::bind(bind).map_err(|e| format!("bind prompt RPC {bind}: {e}"))?;
let addr = listener
// Bind synchronously (there is no ambient runtime at orchestrator startup)
// and report the bound address, then drive accept on the orchestrator
// engine. Each accepted connection runs on the engine's blocking pool,
// reusing the synchronous request/response parser unchanged. There is no
// listener thread and no per-connection std thread (ENGINE_SPEC.md);
// no raw Tokio handle or second runtime is introduced.
let std_listener = TcpListener::bind(bind)
.map_err(|e| format!("bind prompt RPC {bind}: {e}"))?;
let addr = std_listener
.local_addr()
.map_err(|e| format!("read prompt RPC addr: {e}"))?;
thread::spawn(move || {
for accepted in listener.incoming() {
match accepted {
Ok(stream) => {
std_listener
.set_nonblocking(true)
.map_err(|e| format!("set prompt RPC nonblocking: {e}"))?;
let engine = engine.clone();
engine.clone().spawn(async move {
let listener = match tokio::net::TcpListener::from_std(std_listener) {
Ok(listener) => listener,
Err(_) => return,
};
loop {
match listener.accept().await {
Ok((stream, _)) => {
let tx = work_tx.clone();
thread::spawn(move || {
let _ = handle_prompt_connection(stream, tx, default_max_tokens);
let engine = engine.clone();
engine.spawn_blocking(move || {
if let Ok(std_stream) = stream.into_std() {
// The synchronous parser uses blocking I/O.
let _ = std_stream.set_nonblocking(false);
let _ =
handle_prompt_connection(std_stream, tx, default_max_tokens);
}
});
}
Err(_) => break,
@ -4130,6 +4172,8 @@ fn handle_prompt_connection(
Ok(())
}
// synchronous process-control/orchestration sequencing; the engine drives all background work (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn wait_for_runtime_ready(ctx: RuntimeReadyAckLoop<'_>) -> Result<RuntimeReady, String> {
let RuntimeReadyAckLoop {
driver,
@ -4152,7 +4196,7 @@ fn wait_for_runtime_ready(ctx: RuntimeReadyAckLoop<'_>) -> Result<RuntimeReady,
let mut node_swim_ready = false;
let mut node_route_started = false;
loop {
pump(driver, stack, frame_tx);
pump(driver, frame_tx);
drain_frames(frame_rx, dashboard, orch_datastream);
drain_orch_stdio_capture(orch_stdio_rx, orch_datastream, dashboard, run_id, node_id);
if stop_requested(stop_rx) {
@ -4276,10 +4320,12 @@ fn provision_stage(
.map_err(|e| format!("send stage provision: {e}"))
}
// synchronous process-control/orchestration sequencing; the engine drives all background work (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn wait_for_weights_loaded(ctx: RuntimeReadyAckLoop<'_>, stage_index: u32) -> Result<(), String> {
let RuntimeReadyAckLoop {
driver,
stack,
stack: _,
obs_rx,
frame_rx,
frame_tx,
@ -4294,7 +4340,7 @@ fn wait_for_weights_loaded(ctx: RuntimeReadyAckLoop<'_>, stage_index: u32) -> Re
..
} = ctx;
loop {
pump(driver, stack, frame_tx);
pump(driver, frame_tx);
drain_orch_stdio_capture(orch_stdio_rx, orch_datastream, dashboard, run_id, node_id);
if stop_requested(stop_rx) {
return Err("shutdown requested while waiting for weights loaded".to_owned());
@ -4788,7 +4834,6 @@ impl PipelinePromptRuntime {
}
fn poll_driver(&mut self, driver: &mut IrohDriver) {
driver.pump_edge_ingress();
for event in driver.drain_edge_events() {
match event {
EdgeTransportEvent::BytesRead { edge_id, bytes, .. }
@ -4922,6 +4967,8 @@ fn take_pipeline_token_record(
Ok(Some(out))
}
// synchronous process-control/orchestration sequencing; the engine drives all background work (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn serve_prompts(
ctx: RuntimeReadyAckLoop<'_>,
work_rx: &mpsc::Receiver<PromptWork>,
@ -4971,7 +5018,7 @@ fn serve_prompts(
};
let mut active: Option<ActivePrompt> = None;
loop {
pump(driver, stack, frame_tx);
pump(driver, frame_tx);
orch_datastream.flush(dashboard, "orchestrator");
if let Some(pipeline) = pipeline_runtime.as_mut() {
pipeline.poll_driver(driver);
@ -5022,7 +5069,7 @@ fn serve_prompts(
let _ = stack
.runtime
.send_to(orchestrator_actor, OrchestratorMsg::ObserveOperatorStop { run_id });
pump(driver, stack, frame_tx);
pump(driver, frame_tx);
return Ok(());
}
@ -5191,6 +5238,8 @@ fn stop_requested(stop_rx: &mpsc::Receiver<()>) -> bool {
stop_rx.try_recv().is_ok()
}
// top-level OS signal handling is process control, out of scope (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn spawn_stop_listener() -> mpsc::Receiver<()> {
let (tx, rx) = mpsc::channel();
#[cfg(target_os = "linux")]
@ -5427,15 +5476,14 @@ fn emit_swim_probe_events(
}
}
/// Drain iroh ingress/egress queues and datastream connections. Core
/// progression and protocol tick injection are owned by the engine (see
/// `spawn_protocol_ticker`); this only drains integration-owned queues
/// (ENGINE_SPEC.md).
fn pump(
driver: &mut IrohDriver,
stack: &DistributionRuntimeStack,
driver: &IrohDriver,
frame_tx: &mpsc::Sender<CollectedDatastreamFrame>,
) {
stack.tick_protocol_actors(Instant::now());
driver.pump_inbound_to_actors();
stack.pump_runtime_once();
driver.drain_outbox(&stack.outbox);
drain_datastream_connections(driver, frame_tx);
}

88
apps/myelin/src/orchestration/distribution_stack.rs
View file

@ -3,8 +3,10 @@
//!
//! This is the production version of the actor-stack setup that integration
//! tests used to copy by hand: a swactor runtime, the four distribution protocol
//! actors, codec/transport routing, the actor-directory mirrors, and one pumpable
//! tick seam for concrete network drivers such as `iroh-driver`.
//! actors, codec/transport routing, and the actor-directory mirrors. Protocol
//! tick injection is owned by the swactor engine (see
//! [`DistributionRuntimeStack::spawn_protocol_ticker`]); the application loop
//! no longer manually ticks core.
use std::collections::HashMap;
use std::sync::{Arc, Mutex, RwLock};
@ -13,6 +15,7 @@ use std::time::{Duration, Instant};
use swactor::actor::{ActorAddress, ActorInterface};
use swactor::config::RuntimeConfig;
use swactor::runtime::{Ctx, Runtime};
use swactor_engine::EngineHandle;
use swactor::stats::StatsHook;
use swactor::std::StdExtension;
use swactor_transport::{CodecRegistry, CodecRemoteSink, NetworkMessage, TransportRouter};
@ -46,6 +49,10 @@ pub(crate) struct DistributionActorAddrs {
pub(crate) struct DistributionRuntimeStack {
pub runtime: Arc<Runtime>,
/// The node engine this stack is bound to. Protocol ticking and all
/// supporting work schedule on this stored handle; the stack does not
/// accept an unrelated engine at each call (ENGINE_SPEC.md).
pub engine: EngineHandle,
pub codec: Arc<CodecRegistry>,
pub outbox: Outbox,
pub relay_mirror: RelayMirror,
@ -57,12 +64,19 @@ pub(crate) struct DistributionRuntimeStack {
}
impl DistributionRuntimeStack {
pub(crate) fn new_with_codecs(
node_id: NodeId,
config: DistributedNodeConfig,
/// Build and configure the core swactor runtime + codec, returning the
/// shared transport router needed by [`new_from_runtime`]. The runtime is
/// fully configured — extension, remote sink, statistics hook — but no
/// actors are spawned yet.
///
/// This split lets the engine own the runtime before the driver exists:
/// construct the runtime, hand it to [`Engine::new`](swactor_engine::Engine),
/// create the driver (which needs the engine handle), then spawn actors via
/// [`new_from_runtime`] using `driver.node_id()`.
pub(crate) fn build_runtime(
extend_codecs: impl FnOnce(&mut CodecRegistry),
stats_hook: Option<Arc<dyn StatsHook>>,
) -> Self {
) -> (Arc<Runtime>, Arc<CodecRegistry>, Arc<TransportRouter>) {
let mut runtime =
Runtime::new(RuntimeConfig::default()).with_extension(Arc::new(StdExtension::new()));
let mut codec = actor_codec_registry();
@ -76,8 +90,19 @@ impl DistributionRuntimeStack {
if let Some(hook) = stats_hook {
runtime.set_stats_hook(hook);
}
let runtime = Arc::new(runtime);
(Arc::new(runtime), codec, transport_router)
}
/// Spawn the four distribution protocol actors on a pre-built runtime.
/// Used after [`build_runtime`] when the engine already owns the runtime.
pub(crate) fn new_from_runtime(
runtime: Arc<Runtime>,
codec: Arc<CodecRegistry>,
transport_router: Arc<TransportRouter>,
node_id: NodeId,
config: DistributedNodeConfig,
engine: EngineHandle,
) -> Self {
let outbox: Outbox = Arc::new(Mutex::new(Vec::new()));
let relay_mirror: RelayMirror = Arc::new(RwLock::new(HashMap::new()));
let route_view: RouteView = Arc::new(RwLock::new(HashMap::new()));
@ -151,6 +176,7 @@ impl DistributionRuntimeStack {
Self {
runtime,
engine,
codec,
outbox,
relay_mirror,
@ -168,6 +194,20 @@ impl DistributionRuntimeStack {
}
}
/// Convenience: build the runtime and spawn actors in one step. Use
/// [`build_runtime`] + [`new_from_runtime`] when the engine must own the
/// runtime before the driver is constructed.
pub(crate) fn new_with_codecs(
node_id: NodeId,
config: DistributedNodeConfig,
extend_codecs: impl FnOnce(&mut CodecRegistry),
stats_hook: Option<Arc<dyn StatsHook>>,
engine: EngineHandle,
) -> Self {
let (runtime, codec, transport_router) = Self::build_runtime(extend_codecs, stats_hook);
Self::new_from_runtime(runtime, codec, transport_router, node_id, config, engine)
}
pub(crate) fn actor_bridge_routes(&self) -> HashMap<String, ActorAddress> {
let mut routes = HashMap::new();
for tag in [
@ -189,17 +229,29 @@ impl DistributionRuntimeStack {
routes
}
pub(crate) fn tick_protocol_actors(&self, now: Instant) {
let _ = self.runtime.send_to(self.actors.swim, SwimIn::Tick { now });
let _ = self.runtime.send_to(self.actors.registry, RegistryIn::Tick);
let _ = self.runtime.send_to(self.actors.metadata, MetadataIn::Tick);
let _ = self
.runtime
.send_to(self.actors.directory, DirectoryIn::Tick);
}
pub(crate) fn pump_runtime_once(&self) {
self.runtime.tick();
/// Spawn an engine-hosted interval task that injects protocol Tick messages
/// (SWIM, registry, metadata, directory), replacing the manual tick
/// injection previously done by the application pump loop
/// (ENGINE_SPEC.md). The engine owns protocol progression; the
/// application loop no longer calls tick or core-driving methods.
pub(crate) fn spawn_protocol_ticker(&self, period: Duration) {
let runtime = self.runtime.clone();
let swim = self.actors.swim;
let registry = self.actors.registry;
let metadata = self.actors.metadata;
let directory = self.actors.directory;
let engine = self.engine.clone();
engine.clone().spawn(async move {
let mut interval = engine.interval(period);
loop {
(&mut interval).await;
let now = engine.now().to_instant();
let _ = runtime.send_to(swim, SwimIn::Tick { now });
let _ = runtime.send_to(registry, RegistryIn::Tick);
let _ = runtime.send_to(metadata, MetadataIn::Tick);
let _ = runtime.send_to(directory, DirectoryIn::Tick);
}
});
}
pub(crate) fn register_local_actor(&self, entry: DirectoryEntry) {

6
apps/myelin/src/orchestration/provider_adapters/vastai/mod.rs
View file

@ -1,3 +1,9 @@
// VastAI provider adapter: temporarily exempt from the engine disallowed-methods
// policy. This module owns private Tokio runtimes, blocking facades, and a
// polling thread because it predates the engine and is explicitly OUT of engine
// scope (ENGINE_SPEC.md §2). It will be redesigned independently; until then it
// carries this narrow allowance rather than being migrated piecemeal.
#![allow(clippy::disallowed_methods)]
use parking_lot::Mutex;
use std::collections::{BTreeMap, BTreeSet, HashSet};
use std::io::{BufRead, BufReader, Read};

6
apps/myelin/src/orchestration/provisioning.rs
View file

@ -235,6 +235,8 @@ fn lock_process_child(
}
impl ProvisionPlugin for LocalProcessPlugin {
// provider process supervision/lifecycle is out of scope (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn start_node(
&mut self,
spec: NodeProvisionSpec,
@ -341,6 +343,8 @@ impl ProvisionPlugin for LocalProcessPlugin {
Ok(())
}
// provider process supervision/lifecycle is out of scope (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn stop_node(&mut self, handle: &PluginNodeHandle) -> Result<(), String> {
let Some(mut node) = self.nodes.remove(&handle.id) else {
return Ok(());
@ -434,6 +438,8 @@ impl Drop for LocalProcessPlugin {
}
impl ProvisionPlugin for LocalDockerPlugin {
// provider process supervision/lifecycle is out of scope (ENGINE_SPEC.md §2)
#[allow(clippy::disallowed_methods)]
fn start_node(
&mut self,
spec: NodeProvisionSpec,

187
apps/myelin/src/tests/engine_composition.rs Normal file
View file

@ -0,0 +1,187 @@
//! Production execution-composition smoke test (ENGINE_SPEC.md).
//!
//! Verifies the real process-local execution composition for one Myelin node:
//! a swactor `Engine` over a Tokio substrate owns the core runtime and drives
//! it; the production distribution runtime/actors are constructed on that
//! engine; a real `IrohDriver` is bound through `EngineHandle` with
//! relay-disabled networking; actor-bridge and protocol-ticker progression are
//! installed on that same engine; and observable actor progress happens with no
//! ambient Tokio runtime and no application call to `tick`, `try_tick`,
//! `has_work`, or a manual network pump.
//!
//! The composition shares the production wiring (`DistributionRuntimeStack` +
//! `IrohDriver`); it does not duplicate a fake version of it. The only blocking
//! here is test-side observation polling — never engine work.
use std::time::{Duration, Instant};
use swactor::actor::{ActorAddress, ActorInterface, Ctx, Message};
use swactor::runtime::Inbox;
use swactor_engine::{Engine, TokioBackend, TokioConfig};
use distribution::node::DistributedNodeConfig;
use iroh::RelayMode;
use iroh_driver::{IrohDriver, IrohDriverConfig};
use crate::orchestration::distribution_stack::DistributionRuntimeStack;
const PROBE_TICK: Duration = Duration::from_millis(10);
const PROBE_DEADLINE: Duration = Duration::from_secs(8);
// ── Local probe actor ──────────────────────────────────────────────────────
/// Probe message: replies `ProbePong` to a captured address.
#[derive(Clone)]
struct ProbePing;
#[derive(Clone, PartialEq, Eq, Debug)]
struct ProbePong;
struct EchoProbe {
reply_to: ActorAddress,
}
impl ActorInterface for EchoProbe {
type Incoming = ProbePing;
type Response = ();
fn handle(&mut self, ctx: &Ctx, _: ProbePing) {
let _ = ctx.send(self.reply_to, ProbePong);
}
}
// ── Production composition ─────────────────────────────────────────────────
/// Build the production composition for one node, mirroring the node boot
/// sequence: build the core runtime → the engine owns and drives it → the iroh
/// driver is bound through the engine handle → the distribution stack is built
/// on the same runtime → actor bridge, protocol ticker, and adapter pump are
/// installed on that one engine.
fn build_composition() -> (Engine, IrohDriver, DistributionRuntimeStack) {
let (runtime, codec, transport_router) =
DistributionRuntimeStack::build_runtime(|_| {}, None);
let engine = Engine::new(
runtime.clone(),
TokioBackend::new(TokioConfig::default()).expect("build tokio backend"),
)
.expect("build engine");
let mut driver = IrohDriver::with_engine(
engine.handle(),
IrohDriverConfig {
secret_key: None,
relay_mode: RelayMode::Disabled,
node: DistributedNodeConfig::default(),
peer_auth: None,
additional_alpns: vec![],
},
)
.expect("build iroh driver with engine handle");
let stack = DistributionRuntimeStack::new_from_runtime(
runtime,
codec,
transport_router,
driver.node_id(),
DistributedNodeConfig::default(),
engine.handle(),
);
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(),
stack.outbox.clone(),
);
// Engine-hosted protocol tick injection + adapter progression — no manual
// pump is wired anywhere.
stack.spawn_protocol_ticker(PROBE_TICK);
driver.install_actor_bridge_pump(PROBE_TICK);
(engine, driver, stack)
}
/// Poll an inbox until a value arrives or the deadline elapses. The only
/// `thread::sleep` in this module: test observation, not engine work.
#[allow(clippy::disallowed_methods)]
fn recv_within<T: Message>(inbox: &Inbox<T>, deadline: Duration) -> Option<T> {
let started = Instant::now();
loop {
if let Some(value) = inbox.try_recv() {
return Some(value);
}
if started.elapsed() >= deadline {
return None;
}
std::thread::sleep(PROBE_TICK);
}
}
#[test]
fn engine_drives_actor_progress_without_manual_tick() {
let (engine, _driver, stack) = build_composition();
// A probe actor plus an external inbox observe its reply. Delivery and the
// reply are processed entirely by engine-driven core progression — this
// test never calls tick / try_tick / has_work and pumps no network queue.
let pong_inbox = stack
.runtime
.new_inbox::<ProbePong>()
.expect("create pong inbox");
let echo = stack
.runtime
.spawn(EchoProbe {
reply_to: *pong_inbox.addr(),
})
.expect("spawn echo probe");
stack
.runtime
.send_to(echo, ProbePing)
.expect("send probe ping");
let pong = recv_within(&pong_inbox, PROBE_DEADLINE);
// Keep the engine alive until the observation completes.
drop(engine);
assert_eq!(pong, Some(ProbePong), "engine did not drive actor progress");
}
#[cfg(feature = "dashboard")]
#[test]
fn dashboard_server_is_scheduled_through_the_engine() {
// The dashboard server future is scheduled through the Myelin engine path
// (engine.spawn(handle.http_server())), exactly as in production, without
// constructing another runtime (ENGINE_SPEC.md).
let free_port = std::net::TcpListener::bind(("127.0.0.1", 0))
.expect("probe bind for free port")
.local_addr()
.expect("probe local addr")
.port();
let (engine, _driver, _stack) = build_composition();
let mut config = dashboard::DashboardConfig::default();
config.port = free_port;
let handle = dashboard::DashboardHandle::new(config);
engine.handle().spawn(handle.http_server());
// Behavioral proof the server future is actually running on the engine:
// the bound port accepts a TCP connection. No second runtime is involved.
let connected = poll_connect(("127.0.0.1", free_port), PROBE_DEADLINE);
drop(engine);
assert!(connected, "dashboard server did not accept connections");
}
#[cfg(feature = "dashboard")]
#[allow(clippy::disallowed_methods)]
fn poll_connect(addr: (&str, u16), deadline: Duration) -> bool {
use std::net::TcpStream;
let started = Instant::now();
loop {
if TcpStream::connect(addr).is_ok() {
return true;
}
if started.elapsed() >= deadline {
return false;
}
std::thread::sleep(PROBE_TICK);
}
}

1
apps/myelin/src/tests/mod.rs
View file

@ -1,3 +1,4 @@
mod engine_composition;
mod harness;
mod local_e2e_guarantees;
mod local_mock;

5
apps/myelin/src/tests/node_guarantees.rs
View file

@ -1,4 +1,9 @@
//! Behavior guarantees for the `node` module.
//!
//! These unit tests drive a raw `Runtime` in isolation to verify actor message
//! routing — they are not engine integration tests and are exempt from the
//! disallowed-methods policy (ENGINE_SPEC.md §2).
#![allow(clippy::disallowed_methods)]
use crate::node_actor::{NodeAgentActor, NodeAgentMsg, NodeAgentReport};
use crate::orchestration::actor::OrchestratorMsg;

52
clippy.toml Normal file
View file

@ -0,0 +1,52 @@
# Clippy enforcement policy for the swactor engine boundary
# (ENGINE_SPEC.md §2 / §3.1).
#
# These direct runtime / scheduling / time / core-driving operations are
# disallowed outside the engine's own substrate implementation. Integrations
# (iroh-driver, myelin, ...) must go through `EngineHandle`. The
# `swactor-engine` Tokio backend and the core driver carry narrow
# `#[allow(clippy::disallowed_methods)]` exemptions because they ARE the
# substrate implementor; the VastAI provider module carries a temporary
# module-level exemption pending its separate redesign (out of scope per §2).
#
# In-scope work that backs actors, transport, RPC, sampling, or node/orchestrator
# progression must schedule through `EngineHandle`. The only retained direct
# uses are narrow exclusions (§2): provider adapters/lifecycle (VastAI),
# provider-specific process supervision and log capture, and top-level OS-signal
# / blocking user-stdin / synchronous process-control sequencing. Each retained
# use carries a local `#[allow]` with its exclusion reason.
#
# Workspace-wide enforcement: `swactor-engine`, `iroh-driver`, and in-scope
# `myelin` carry `#![deny(clippy::disallowed_methods)]` and pass clean.
disallowed-methods = [
{ path = "tokio::runtime::Runtime::new", reason = "runtime ownership belongs to the engine; construct an engine-owned substrate instead" },
{ path = "tokio::runtime::Builder::new_current_thread", reason = "runtime ownership belongs to the engine; use EngineHandle" },
{ path = "tokio::runtime::Builder::new_multi_thread", reason = "runtime ownership belongs to the engine; use EngineHandle" },
{ path = "tokio::runtime::Handle::current", reason = "ambient runtime detection is forbidden; construct an engine-owned substrate instead" },
{ path = "tokio::runtime::Handle::try_current", reason = "ambient runtime detection is forbidden; construct an engine-owned substrate instead" },
{ path = "tokio::runtime::Runtime::block_on", reason = "blocking on a runtime is forbidden; schedule through EngineHandle" },
{ path = "tokio::runtime::Handle::block_on", reason = "blocking on a runtime is forbidden; schedule through EngineHandle" },
{ path = "tokio::spawn", reason = "direct scheduling is forbidden; use EngineHandle::spawn" },
{ path = "tokio::task::spawn", reason = "direct scheduling is forbidden; use EngineHandle::spawn" },
{ path = "tokio::task::spawn_blocking", reason = "use EngineHandle::spawn_blocking" },
{ path = "tokio::runtime::Runtime::spawn", reason = "direct scheduling is forbidden; use EngineHandle::spawn" },
{ path = "tokio::runtime::Handle::spawn", reason = "direct scheduling is forbidden; use EngineHandle::spawn" },
{ path = "tokio::runtime::Runtime::spawn_blocking", reason = "use EngineHandle::spawn_blocking" },
{ path = "tokio::runtime::Handle::spawn_blocking", reason = "use EngineHandle::spawn_blocking" },
{ path = "tokio::time::sleep", reason = "use EngineHandle::timer" },
{ path = "tokio::time::sleep_until", reason = "use EngineHandle::timer" },
{ path = "tokio::time::interval", reason = "use EngineHandle::interval" },
{ path = "tokio::time::interval_at", reason = "use EngineHandle::interval" },
{ path = "tokio::time::timeout", reason = "use an engine-derived timeout" },
{ path = "tokio::time::timeout_at", reason = "use an engine-derived timeout" },
{ path = "std::thread::spawn", reason = "direct thread scheduling is forbidden; schedule through EngineHandle" },
{ path = "std::thread::sleep", reason = "use EngineHandle::timer; retained only for narrow process-control exclusions (ENGINE_SPEC.md §2)" },
{ path = "swactor::runtime::Runtime::tick", reason = "manual core driving is forbidden; the engine owns core progression" },
{ path = "swactor::runtime::Runtime::try_tick", reason = "manual core driving is forbidden; the engine owns core progression" },
{ path = "swactor::runtime::Runtime::has_work", reason = "manual core driving is forbidden; the engine owns core progression" },
]

24
crates/dashboard/src/lib.rs
View file

@ -90,8 +90,8 @@ pub struct DashboardHandle {
impl DashboardHandle {
/// Create the datastream dashboard state.
///
/// The HTTP server is not started until `spawn_http` or
/// `start_http_standalone` is called.
/// The HTTP server future is obtained from [`DashboardHandle::http_server`]
/// and scheduled by the owning swactor engine.
pub fn new(config: DashboardConfig) -> Self {
let views = Arc::new(ViewRegistry::new());
views.register(Arc::new(live_explorer::LiveDatastreamExplorer::default()));
@ -150,28 +150,12 @@ impl DashboardHandle {
}
}
/// Spawn the HTTP server on an existing Tokio runtime and return its task handle.
pub fn spawn_http(&self, handle: &tokio::runtime::Handle) -> tokio::task::JoinHandle<()> {
handle.spawn(self.http_server())
}
/// Spawn the HTTP server on a dedicated Tokio runtime in a background thread.
pub fn start_http_standalone(&self) {
let server = self.http_server();
std::thread::spawn(move || {
let runtime = tokio::runtime::Builder::new_multi_thread()
.enable_io()
.build()
.expect("dashboard standalone HTTP runtime");
runtime.block_on(server);
});
}
}
/// Create the datastream dashboard state.
///
/// The HTTP server is not started until `DashboardHandle::start_http_standalone`
/// or `DashboardHandle::spawn_http` is called.
/// The HTTP server future is obtained from [`DashboardHandle::http_server`] and
/// scheduled by the owning swactor engine.
pub fn start_dashboard(config: DashboardConfig) -> DashboardHandle {
DashboardHandle::new(config)
}

6
crates/distribution/tests/routing.rs
View file

@ -760,9 +760,9 @@ mod directory_route_path {
c
}
/// Move every queued frame to its destination, mirroring the live driver:
/// `drain_outbox` (sender side) then `pump_inbound_to_actors` (receiver side,
/// dest-first). A frame addressed to a node's peer-mailbox is gossip and is
/// Move every queued frame to its destination, mirroring the
/// engine-hosted adapter pump (sender-side outbox drain, then dest-first
/// delivery). A frame addressed to a node's peer-mailbox is gossip and is
/// routed by tag; anything else is an app message delivered to its `dest`.
fn deliver_wire(&self) {
let mut frames: Vec<OutFrame> = Vec::new();

14
crates/engine/Cargo.toml Normal file
View file

@ -0,0 +1,14 @@
[package]
name = "swactor-engine"
version = "0.1.0"
edition = "2024"
license = "AGPL-3.0-only"
[features]
default = ["tokio"]
tokio = ["dep:tokio"]
[dependencies]
swactor = { path = "../.." }
parking_lot = "0.12"
tokio = { workspace = true, optional = true }

144
crates/engine/ENGINE_SPEC.md Normal file
View file

@ -0,0 +1,144 @@
# swactor engine — specification
Id: 1
Last modified: f8fc594b95871813a890b5d60f60dee505ef93bc
Last reviewed: f8fc594b95871813a890b5d60f60dee505ef93bc
**Scope:** the execution substrate that drives swactor workers and hosts its async side-work, defined as an interface implemented per environment.
**Status: implemented** for the default native Tokio engine. The substrate-neutral contract, capability model, and engine time are in place, verified by the native behavioral contract and a non-Tokio portability proof. The VastAI provider adapter remains explicitly out of scope pending its separate redesign.
## Motivation
This abstraction was motivated by the original `IrohDriver` and `apps/myelin` integration. Before the engine, `IrohDriver` accepted and stored a Tokio `Handle`, called `spawn` for accepts, reads, dials, and writes, and used `block_on` during synchronous construction and shutdown, with a legacy constructor that detected an ambient Tokio runtime or silently created one. Separately, `apps/myelin` constructed the Tokio runtime, constructed the swactor runtime, and manually sequenced protocol tick injection, Iroh ingress, `Runtime::tick()`, Iroh egress, and polling sleeps. The execution substrate was thus hardcoded and reinvented per crate (ambient `Handle::try_current()`, silently-owned runtimes, ad-hoc `block_on` sync facades, a mix of tokio tasks and std threads).
The engine resolves this. swactor consumes and retains the selected execution substrate—Tokio, std threads, JS, or another implementation—and exposes one explicit engine contract to integrations such as Iroh and Myelin. That engine hosts actor progression and the asynchronous or blocking side-work supporting actors as one system, with one lifecycle and one place controlling execution semantics. `IrohDriver` now receives a swactor engine handle rather than a raw Tokio handle, and application code no longer assembles an independent actor driver beside an independent task runtime. Core remains engine-independent and synchronous.
This process-local execution engine is distinct from Myelin's cluster-level `orchestration::engine_builder`, which acquires nodes, waits for convergence, assigns roles, and returns a cluster handle. A Myelin node owns a swactor execution engine; the two abstractions operate at different layers.
## 1. Purpose
swactor actors are synchronous, single-writer message handlers. Real systems need work actors cannot do inline: draining byte streams, running retry backoffs, polling on an interval, blocking GPU calls. That work lives in *tasks* on an execution substrate. Without an engine, each integration reinvents that substrate independently (see Motivation); the engine gives those tasks one swactor-owned home.
This spec defines the **engine**: a swactor-owned composite that retains a selected execution substrate, drives the core runtime, and hosts the supporting work that backs actors. Tokio/native, WASM/event-loop, embedded/cooperative, minimal std-thread, Go, JS-worker, and deterministic test schedulers are substrate implementations behind the same swactor engine contract. Authoring that contract from swactor's needs keeps execution ownership and semantics in swactor while allowing core and each substrate implementation to remain independently optimized.
## 2. Scope
**In scope**
- The engine interface: what swactor requires of an engine, and what an engine provides.
- The responsibility split between actor-workers and the engine.
- How the engine drives core through its existing synchronous tick semantics.
- The capability surface: tasks, timers, I/O, blocking, time.
- How engine-hosted work may communicate through integration-owned boundaries (non-normative).
- The invariants an engine must uphold.
- Reference instantiations (non-normative).
**Out of scope**
- Actor execution semantics — single-writer, per-(sender,target) FIFO, fairness, panic isolation. Those belong to the actor-worker / core.
- Backpressure policy. Producers and consumers share one engine; pressure handling is the application's decision, not swactor's.
- Cancellation and shutdown lifecycle (deferred; nice-to-have).
- Failure / observability propagation from engine-hosted work.
- Cross-process / cross-isolation delivery and serialization.
- Specific protocols and codecs (iroh/QUIC, datastream framing). Those are crate logic built *on* the engine.
- Provider adapters, including the VastAI provider adapter. Their private runtimes, manually driven swactor runtimes, blocking facades, polling threads, and provider lifecycle are crate-level concerns built *beside* the engine, not on it; they require their own redesign rather than incremental engine migration.
## 3. Model
- The **core runtime** owns its actor-workers, pools, mailboxes, and routing. Actor execution remains synchronous and single-writer; core exposes tick semantics that advance those state machines and return immediately.
- The **engine** is a swactor-owned composite. It retains the selected execution substrate and the core runtime, and it does exactly two things:
1. **Drives actor execution** — schedules core ticks without application involvement.
2. **Runs supporting work** — schedules the I/O, blocking calls, retries, and other long-lived flows that back actors.
- **The engine owns all progression.** Actor handlers never `.await`. Every handler is a synchronous transition that returns control immediately; the engine carries control flow across time.
- Actor execution and supporting work are not independently driven systems. They share one engine, one execution policy, and one lifecycle. An engine may use multiple internal pools, threads, scheduler domains, or substrate-native facilities to meet its progression and performance requirements.
- Integrations receive a cloneable engine handle through which they schedule supporting work. They do not receive or own the underlying Tokio, thread-pool, or host-runtime handle.
## 4. The engine interface
The interface is authored from swactor's needs. It is a **contract** — operations plus their semantics and invariants. A Rust trait is its canonical Rust binding; Go, JS, and other hosts implement the same contract natively. This spec defines the contract, not the Rust signature.
Constructing a swactor engine consumes or retains the selected execution substrate and establishes core driving for the engine's lifetime. Worker installation is internal engine behavior: applications and integrations do not register workers or receive core routing handles.
**The engine handle provides:**
| operation | meaning |
|---|---|
| `spawn(task)` | Schedule an async unit of work on the substrate. |
| `spawn_blocking(work)` | Schedule blocking CPU / syscall work off the async path. |
| `timer(delay)` / `interval(period)` | Schedule future or recurring work. |
| `now()` | The engine's monotonic clock. |
Time belongs to the engine rather than actor core. Engine-hosted work needs delays, intervals, retry deadlines, and timeouts; leaving those operations outside the contract would keep integrations such as Iroh and Myelin coupled to `tokio::time` or `std::thread::sleep`. An engine-owned monotonic clock also gives all hosted work one time source and allows a deterministic engine to substitute virtual time without changing integration code.
**Existing core integration.** The engine wraps and drives core without redefining it. Actor progression uses the existing `Runtime::tick()` / `Runtime::try_tick()` surface, and message delivery continues through existing runtime and sender APIs. Core implements no engine trait, exposes no worker callback, and receives no engine-specific routing handle. Core owns actor logic, routing, inboxes, and delivery; the engine owns when core transitions run and schedules all supporting work on the same substrate. **Core only transitions. The engine drives.**
## 5. Driving workers
- Driving core is intrinsic to the engine and is established during engine construction. Application code and integrations never register or manually drive workers.
- The engine advances core through its existing synchronous tick semantics. Each tick runs to completion and returns control to the engine scheduler.
- **Non-reentrancy.** The engine must never invoke the same worker concurrently. Actor state is live only for the duration of a synchronous tick.
- **Scheduling strategy is the engine's choice.** Tick cadence, batching, thread placement, and cooperative scheduling are implementation decisions, subject to the progress guarantees in §8.
## 6. Capability surface
The primitives an engine may provide. Capabilities are **per-implementation and discoverable**: each engine reports which it supports, and binding an engine that lacks a required capability fails at construction, never at runtime.
- **Tasks** — `spawn` of an async unit of work; the substrate's unit of concurrency.
- **Timers** — one-shot delay and recurring interval.
- **I/O** — streams, sockets, files, and protocol endpoints used by engine-hosted work. An engine may implement I/O through asynchronous operations, blocking operations on managed threads, callbacks, or host-native facilities. Integrations declare the I/O capabilities they require, and binding fails at construction when the selected engine cannot provide them.
- **Blocking** — `spawn_blocking` for CPU-bound or syscall work that must not stall the executor.
- **Time** — `now()`. In a test engine this is virtual, advanced by the test; this is what makes deterministic testing possible.
An engine that provides only tasks + time is still valid. Blocking and I/O are additional capabilities declared by integrations that require them.
### Native Rust binding scope
The implemented Rust SPI is the **native, sendable** binding: its task representation erases a task once when installed and requires `Send + Sync`. That `Send + Sync` model is the native binding — it is not a claim that this SPI is the Rust/WASM-local binding. The deterministic stepping backend proves executor and time independence on native Rust; it does not prove support for non-`Send` browser futures. A separate local-task binding for non-`Send` futures may be introduced when a real WASM implementation exists; until then the contract deliberately avoids conditional trait hierarchies, associated-future abstractions, or target-specific generic complexity.
## 7. Integration boundary (non-normative)
The engine executes opaque supporting work. It does not define how the results of that work become actor messages.
- Integrations own the handles and buffers through which their supporting work communicates. Valid patterns include capturing an existing core sender, writing to an integration-owned queue drained by actor-facing code, completing a callback or result channel, or producing no actor message at all.
- The engine does not define actor addressing, message delivery, mailbox ordering, or delivery guarantees. Those remain core and integration concerns.
- Actor state remains synchronous and single-writer. Supporting work must not retain mutable actor or worker state across engine scheduling points.
The current Iroh integration illustrates the queued pattern: background network readers write wire frames to an Iroh-owned ingress queue; an Iroh actor adapter drains and decodes those frames and calls the existing `Runtime::deliver_raw()` surface. Outbound actor frames pass through the integration-owned outbox to engine-hosted network writers. Another integration may instead capture an `ExternalSender` and deliver directly. Both patterns use the same engine without making actor delivery part of the engine contract.
## 8. Invariants
An engine must uphold:
- **Non-reentrant ticks.** At most one tick per worker at any instant.
- **Progress independence.** A long-running or blocked piece of supporting work must not stall actor ticks, and actor execution must not stall unrelated supporting work. The engine provides enough concurrency or cooperative scheduling for both to progress.
- **Actors never await.** No `.await` reaches actor code; the engine owns every long-lived flow.
- **Hot-path transparency.** The engine contract imposes no required per-item allocation, copy, serialization, actor hop, dynamic dispatch, or scheduler transition. Long-lived engine-hosted work may retain substrate-native I/O resources and transfer data directly through integration-owned buffers.
## 9. Reference instantiations (non-normative)
Illustrations of how each environment satisfies the contract — not prescription.
- **tokio.** A swactor engine owns a Tokio runtime and uses it to schedule both core ticks and supporting futures. I/O, blocking work, and actor progression share that runtime; integrations receive a swactor engine handle rather than a raw Tokio `Handle`.
- **std-thread.** A swactor engine owns its threads or small pool and schedules both core ticks and blocking supporting work there. It offers no native async I/O, but preserves the same ownership and progression contract.
- **deterministic test engine.** A single-threaded stepping scheduler advances core and supporting work under test control. It implements the same contract with no real network or threads.
- **Go / JS-worker (illustrative).** Core ticks and supporting work share goroutines plus the Go scheduler, or the JS event loop plus workers. Each implementation exposes only a swactor engine handle to integrations.
### Iroh / Myelin integration
1. `apps/myelin` constructs a swactor engine with a Tokio substrate.
2. The swactor engine retains Tokio, owns the core runtime, and drives actor execution.
3. `IrohDriver` receives a swactor engine handle rather than a raw Tokio `Handle`; accepts, reads, dials, writes, and blocking work are scheduled through that handle.
4. Actor execution and Iroh work therefore share one engine and lifecycle. Myelin does not assemble an independent actor driver beside an independent task runtime.
## 10. What this spec does not define
The boundary, stated plainly:
- Actor execution semantics (single-writer, FIFO, fairness, panic isolation).
- Backpressure.
- Cancellation and shutdown.
- Failure / observability propagation across integration boundaries.
- Cross-process / cross-isolation delivery and serialization.
- Specific protocols and codecs.
- Provider adapters (e.g. the VastAI provider adapter) and their private runtimes / provider lifecycle.

100
crates/engine/src/backend.rs Normal file
View file

@ -0,0 +1,100 @@
//! Execution backend SPI.
//!
//! The [`ExecutionBackend`] trait is an implementation seam the engine uses to
//! schedule work and read engine time. It is not the interface applications
//! consume; that is [`crate::EngineHandle`]. A native implementation erases
//! tasks once when they are installed; this representation is not a
//! cross-target requirement (see `ENGINE_SPEC.md`).
use std::pin::Pin;
use std::time::Duration;
use crate::time::EngineInstant;
/// A boxed, sendable future returned by the engine substrate.
pub type BoxTask = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
/// A boxed, sendable timer future produced by the substrate.
pub type BoxTimer = Pin<Box<dyn Future<Output = ()> + Send + 'static>>;
/// A boxed, sendable one-shot blocking workload.
pub type BoxWork = Box<dyn FnOnce() + Send + 'static>;
/// The substrate-specific execution surface an engine schedules onto.
///
/// Object-safe so the composite [`Engine`](crate::Engine) can store it as
/// `Arc<dyn ExecutionBackend>` without being generic over the backend.
pub trait ExecutionBackend: Send + Sync + 'static {
/// Schedule `task` to run as cooperative engine work.
fn spawn(&self, task: BoxTask);
/// Schedule `work` on a dedicated blocking thread.
fn spawn_blocking(&self, work: BoxWork);
/// Produce a future that completes after `delay`.
fn timer(&self, delay: Duration) -> BoxTimer;
/// Read the engine's monotonic clock.
fn now(&self) -> EngineInstant;
/// Report the substrate's advertised capabilities.
fn capabilities(&self) -> Capabilities;
}
/// Capabilities an execution backend advertises.
///
/// - `tasks`: cooperative task scheduling.
/// - `timers`: timer and interval support.
/// - `blocking`: dedicated blocking-thread pools.
/// - `io`: asynchronous I/O reactor (e.g. Tokio's I/O driver from `enable_all`).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Capabilities {
pub tasks: bool,
pub timers: bool,
pub blocking: bool,
pub io: bool,
}
impl Capabilities {
/// Convenience: only baseline task execution.
pub const TASKS_ONLY: Self = Self { tasks: true, timers: false, blocking: false, io: false };
/// Convenience: every capability.
pub const ALL: Self = Self { tasks: true, timers: true, blocking: true, io: true };
/// Convenience: no capabilities. Reported by an [`EngineHandle`](crate::EngineHandle)
/// whose owning engine has been dropped (ENGINE_SPEC.md).
pub const NONE: Self = Self { tasks: false, timers: false, blocking: false, io: false };
/// Whether `self` satisfies every capability marked `true` in `required`.
///
/// A `required` field set to `false` is treated as "not required" — the
/// backend may or may not provide it. Used by
/// [`EngineHandle::require`](crate::EngineHandle::require) to validate
/// integration requirements (ENGINE_SPEC.md).
pub fn satisfies(&self, required: Capabilities) -> bool {
(!required.tasks || self.tasks)
&& (!required.timers || self.timers)
&& (!required.blocking || self.blocking)
&& (!required.io || self.io)
}
}
/// Errors that can arise while constructing an engine.
#[derive(Debug, Clone)]
pub enum EngineError {
/// A capability required by the engine is not advertised by the backend.
MissingRequiredCapability,
/// A backend-owned substrate could not be constructed (e.g. a Tokio
/// runtime failed to build).
BackendSetup(String),
}
impl std::fmt::Display for EngineError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
EngineError::MissingRequiredCapability => {
write!(f, "backend is missing a capability required by the engine")
}
EngineError::BackendSetup(msg) => {
write!(f, "backend substrate setup failed: {msg}")
}
}
}
}
impl std::error::Error for EngineError {}

66
crates/engine/src/core_driver.rs Normal file
View file

@ -0,0 +1,66 @@
//! Core driving loop.
//!
//! Substrate-neutral: the driver is one allocated task that runs one
//! [`Runtime::try_tick`] per poll — synchronous, returns immediately — then
//! re-schedules itself by waking its own waker. There is no backend reference
//! on the hot path, no per-turn boxed yield, no inbox-wake or readiness
//! mechanism, and no `has_work()` gate; later polls observe newly delivered
//! messages (ENGINE_SPEC.md).
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll};
use crate::backend::{BoxTask, ExecutionBackend};
/// The sole core-driving loop, installed once per engine.
///
/// Each poll runs one [`Runtime::try_tick`] — synchronous, returns immediately
/// — then re-arms itself via `cx.waker().wake_by_ref()` and returns `Pending`.
/// Rescheduling through the waker hands control back to the substrate
/// scheduler between ticks, so other engine work progresses. The driver holds
/// no backend reference and allocates nothing per turn (ENGINE_SPEC.md).
///
/// On a real executor (e.g. Tokio) `wake_by_ref` re-enqueues the task rather
/// than re-polling inline, so other ready tasks run between ticks. On the
/// stepping test backend the waker is a no-op and each `step` re-polls every
/// task, so one `step` still advances the driver by exactly one tick.
///
/// # Non-reentrancy
/// Only one driver is installed per engine, and `try_tick` runs to completion
/// within a poll, so the runtime's worker is never borrowed concurrently.
/// This is what makes `unsafe impl Sync` on [`Runtime`] sound under a single
/// driver (see ENGINE_SPEC.md §5/§8).
///
/// [`Runtime::try_tick`]: swactor::runtime::Runtime::try_tick
/// [`Runtime`]: swactor::runtime::Runtime
struct CoreDriver {
runtime: Arc<swactor::runtime::Runtime>,
}
// The core driver is the engine's sole core-progression path; it is the one
// place permitted to call `Runtime::try_tick` (ENGINE_SPEC.md §2).
#[allow(clippy::disallowed_methods)]
impl Future for CoreDriver {
type Output = ();
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
// `CoreDriver` is `Unpin` (`Arc<…>` is `Unpin`), so field access
// through `Pin<&mut Self>` is sound without projection.
self.runtime.try_tick();
// Re-arm immediately: the substrate scheduler redispatches this task,
// yielding to other engine work between ticks. No boxed yield is
// allocated per turn and no backend reference is retained.
cx.waker().wake_by_ref();
Poll::Pending
}
}
/// Install the sole core-driving loop for `runtime` onto `backend`.
///
/// One task is allocated at engine construction and runs for the engine's
/// lifetime; the substrate cancels it when the backend is dropped.
pub(crate) fn install(runtime: Arc<swactor::runtime::Runtime>, backend: &Arc<dyn ExecutionBackend>) {
let driver: BoxTask = Box::pin(CoreDriver { runtime });
backend.spawn(driver);
}

163
crates/engine/src/engine.rs Normal file
View file

@ -0,0 +1,163 @@
//! Composite engine and cloneable scheduler handle.
use std::sync::{Arc, Weak};
use std::time::Duration;
use crate::backend::{Capabilities, EngineError, ExecutionBackend};
use crate::time::{EngineInstant, Interval, Timer, Timeout};
/// The composite engine: retains a configured core runtime and its execution
/// backend, and owns the sole core-driving loop for that runtime.
///
/// Construct with [`Engine::new`]; obtain a scheduler handle with
/// [`Engine::handle`].
pub struct Engine {
/// Retained so the engine owns the runtime it drives for its full lifetime.
/// The core driver holds its own clone; this field anchors ownership (and
/// future admin/shutdown surfaces) even though it is not read directly.
#[allow(dead_code)]
runtime: Arc<swactor::runtime::Runtime>,
backend: Arc<dyn ExecutionBackend>,
}
impl Engine {
/// Construct an engine over `runtime` driven by `backend`.
///
/// The runtime must be fully configured beforehand; after construction the
/// engine is its sole driver. Construction fails if `backend` does not
/// advertise a capability the engine requires (at minimum, `tasks`).
pub fn new(
runtime: Arc<swactor::runtime::Runtime>,
backend: impl ExecutionBackend,
) -> Result<Self, EngineError> {
let backend: Arc<dyn ExecutionBackend> = Arc::new(backend);
if !backend.capabilities().tasks {
return Err(EngineError::MissingRequiredCapability);
}
// Install exactly one core-driving loop; the engine is now the sole
// driver of `runtime`. This is substrate-neutral — no Tokio feature
// gate — so core progression does not silently disappear when an
// alternate backend is used (ENGINE_SPEC.md).
crate::core_driver::install(runtime.clone(), &backend);
Ok(Engine { runtime, backend })
}
/// Return a clonable handle for scheduling engine work.
///
/// The handle holds a *weak* backend reference, so handles — and engine
/// work that captures them — never keep the backend alive. Dropping the
/// [`Engine`] releases the backend (and its owned runtime / core-driver
/// task) once no other strong reference remains (ENGINE_SPEC.md).
pub fn handle(&self) -> EngineHandle {
EngineHandle {
backend: Arc::downgrade(&self.backend),
}
}
}
/// A cloneable scheduler handle.
///
/// Schedules work and reads engine time without exposing the underlying
/// backend; in particular it never hands out a raw `tokio::runtime::Handle`.
/// The handle holds a **weak** backend reference: it does not keep the engine
/// or its backend alive. Using a handle after its engine has been dropped
/// degrades gracefully — scheduled work is dropped, timers never fire, and
/// capability checks report no capabilities — rather than retaining the
/// backend (ENGINE_SPEC.md).
#[derive(Clone)]
pub struct EngineHandle {
backend: Weak<dyn ExecutionBackend>,
}
impl EngineHandle {
/// Upgrade to the live backend, or `None` if the owning engine is gone.
fn backend(&self) -> Option<Arc<dyn ExecutionBackend>> {
self.backend.upgrade()
}
/// Schedule `task` as cooperative engine work.
///
/// A no-op once the owning engine has been dropped: the work is discarded
/// rather than keeping the backend alive.
pub fn spawn<F>(&self, task: F)
where
F: Future<Output = ()> + Send + 'static,
{
if let Some(backend) = self.backend() {
backend.spawn(Box::pin(task));
}
}
/// Schedule `work` on a dedicated blocking thread.
///
/// A no-op once the owning engine has been dropped.
pub fn spawn_blocking<F>(&self, work: F)
where
F: FnOnce() + Send + 'static,
{
if let Some(backend) = self.backend() {
backend.spawn_blocking(Box::new(work));
}
}
/// Produce a future that completes after `delay`.
///
/// Once the owning engine has been dropped this returns a timer that never
/// fires.
pub fn timer(&self, delay: Duration) -> Timer {
match self.backend() {
Some(backend) => Timer { inner: backend.timer(delay) },
None => Timer::closed(),
}
}
/// Produce a future that recurs every `period`.
pub fn interval(&self, period: Duration) -> Interval {
Interval {
period,
backend: self.backend.clone(),
current: None,
}
}
/// Race `future` against an engine timer.
///
/// Resolves to `Ok` with the future's output if it completes within
/// `duration`, or [`Err(Elapsed)`](crate::Elapsed) when the timer fires
pub fn timeout<F: std::future::Future>(&self, duration: Duration, future: F) -> Timeout<F> {
Timeout::new(self.timer(duration), future)
}
/// Read the engine's monotonic clock.
///
/// Falls back to the real wall clock once the owning engine has been
/// dropped, since the substrate clock is no longer available.
pub fn now(&self) -> EngineInstant {
match self.backend() {
Some(backend) => backend.now(),
None => EngineInstant::now(),
}
}
/// Report the backend's advertised capabilities.
///
/// Reports no capabilities once the owning engine has been dropped.
pub fn capabilities(&self) -> Capabilities {
match self.backend() {
Some(backend) => backend.capabilities(),
None => Capabilities::NONE,
}
}
/// Validate that this engine satisfies `required` before starting work.
///
/// Returns `Err` if the backend cannot provide a requested capability, or
/// if the owning engine has been dropped. Call this before allocating
/// resources, starting background work, or becoming externally visible so
/// that an incompatible engine is rejected early (ENGINE_SPEC.md).
pub fn require(&self, required: Capabilities) -> Result<(), EngineError> {
match self.backend() {
Some(backend) if backend.capabilities().satisfies(required) => Ok(()),
_ => Err(EngineError::MissingRequiredCapability),
}
}
}

25
crates/engine/src/lib.rs Normal file
View file

@ -0,0 +1,25 @@
//! swactor-engine: an execution engine that drives a `swactor` core runtime on
//! a selected substrate without exposing that substrate through its handle.
//!
//! The default substrate is Tokio; a deterministic [`SteppingBackend`] provides
//! a non-Tokio portability proof. The public contract is defined by
//! [`ENGINE_SPEC.md`](../ENGINE_SPEC.md).
#![deny(clippy::disallowed_methods)]
mod backend;
mod core_driver;
mod engine;
mod stepping;
mod time;
#[cfg(feature = "tokio")]
mod tokio;
pub use backend::{
BoxTask, BoxTimer, BoxWork, Capabilities, EngineError, ExecutionBackend,
};
pub use engine::{Engine, EngineHandle};
pub use stepping::SteppingBackend;
pub use time::{Elapsed, EngineInstant, Interval, Timeout, Timer};
#[cfg(feature = "tokio")]
pub use tokio::{TokioBackend, TokioConfig};

221
crates/engine/src/stepping.rs Normal file
View file

@ -0,0 +1,221 @@
//! Deterministic single-threaded test execution backend.
//!
//! It demonstrates that the substrate-neutral engine contract works without
//! Tokio, without real async I/O, and with substitutable virtual time.
//!
//! The backend is `Clone` (shares state through `Arc`), so a test keeps one
//! copy to call [`SteppingBackend::step`] / [`SteppingBackend::advance_time`]
//! while passing another to [`Engine::new`](crate::Engine::new).
//!
//! ## How it works
//!
//! - **Tasks** are stored in a Vec and polled cooperatively. Each
//! [`step`](Self::step) polls every live task exactly once using a noop
//! waker — progress is driven by repeated `step` calls, not by wakeups.
//! - **Core driving** uses a self-waking driver: each poll runs one
//! `try_tick` and re-arms via the (noop) waker, so each `step` advances the
//! core-driving loop by exactly one tick.
//! - **Time** is virtual: [`now`](crate::ExecutionBackend::now) returns a
//! clock the test advances explicitly via
//! [`advance_time`](Self::advance_time). Timers compare against this clock.
//! - **Blocking** work runs on a real `std::thread` (truthful isolation),
//! joined when the backend's last clone drops.
use std::pin::Pin;
use std::sync::Arc;
use std::task::{Context, Poll, Wake, Waker};
use std::time::{Duration, Instant};
use parking_lot::Mutex;
use crate::backend::{BoxTask, BoxTimer, BoxWork, Capabilities, ExecutionBackend};
use crate::time::EngineInstant;
// ── Noop waker ──────────────────────────────────────────────────────────────
/// A waker whose `wake` is a no-op. The stepping executor polls all tasks
/// unconditionally on each [`step`](SteppingBackend::step), so it never relies
/// on wakeups for rescheduling.
struct NoopWaker;
impl Wake for NoopWaker {
fn wake(self: Arc<Self>) {}
}
fn noop_waker() -> Waker {
Waker::from(Arc::new(NoopWaker))
}
// ── SteppingTimer future ────────────────────────────────────────────────────
/// A timer future driven by the stepping backend's virtual clock.
///
/// Completes when the virtual clock reaches `deadline`. Checked on each poll,
/// so it fires as soon as [`advance_time`](SteppingBackend::advance_time) has
/// moved the clock far enough.
struct SteppingTimer {
deadline: Instant,
clock: Arc<Mutex<Instant>>,
}
impl Future for SteppingTimer {
type Output = ();
fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> {
if *self.clock.lock() >= self.deadline {
Poll::Ready(())
} else {
Poll::Pending
}
}
}
// ── SteppingInner ───────────────────────────────────────────────────────────
/// Shared inner state behind [`SteppingBackend`].
struct SteppingInner {
/// Live spawned tasks, polled on each `step`.
tasks: Mutex<Vec<Pin<Box<dyn Future<Output = ()> + Send + 'static>>>>,
/// Handles for spawned blocking threads, joined on drop.
blocking: Mutex<Vec<std::thread::JoinHandle<()>>>,
}
impl Drop for SteppingInner {
fn drop(&mut self) {
let handles: Vec<_> = self.blocking.lock().drain(..).collect();
for handle in handles {
let _ = handle.join();
}
}
}
// ── SteppingBackend ─────────────────────────────────────────────────────────
/// A deterministic, single-threaded, test-controlled execution backend.
///
/// Clone to obtain a controller handle: one clone goes to
/// [`Engine::new`](crate::Engine::new), the test keeps another to drive
/// execution via [`step`](Self::step) and [`advance_time`](Self::advance_time).
///
/// This backend advertises `tasks`, `timers`, and `blocking` but **not** `io`.
/// It proves the engine contract is substrate-neutral: core progresses,
/// supporting work progresses, and engine time is substitutable — all without
/// Tokio (ENGINE_SPEC.md).
#[derive(Clone)]
pub struct SteppingBackend {
inner: Arc<SteppingInner>,
/// Virtual monotonic clock, shared with timers. Kept as a separate
/// `Arc<Mutex<Instant>>` so timer futures are self-contained.
clock: Arc<Mutex<Instant>>,
}
impl Default for SteppingBackend {
fn default() -> Self {
Self::new()
}
}
impl SteppingBackend {
/// Create a new stepping backend with virtual time starting at `Instant::now()`.
pub fn new() -> Self {
Self {
inner: Arc::new(SteppingInner {
tasks: Mutex::new(Vec::new()),
blocking: Mutex::new(Vec::new()),
}),
clock: Arc::new(Mutex::new(Instant::now())),
}
}
/// Create a stepping backend whose virtual clock starts at `start`.
///
/// Useful for deterministic tests that need a known clock origin.
pub fn with_clock(start: Instant) -> Self {
Self {
inner: Arc::new(SteppingInner {
tasks: Mutex::new(Vec::new()),
blocking: Mutex::new(Vec::new()),
}),
clock: Arc::new(Mutex::new(start)),
}
}
/// Poll every live task exactly once.
///
/// Each call advances the core-driving loop by one tick and gives every
/// spawned task one scheduling turn. Call repeatedly to drive execution.
pub fn step(&self) {
let waker = noop_waker();
let mut cx = Context::from_waker(&waker);
// Drain all tasks, poll each, keep those still alive. New tasks
// spawned during polling land in the mutex; we merge them back after.
let mut tasks: Vec<_> = self.inner.tasks.lock().drain(..).collect();
let mut alive = Vec::with_capacity(tasks.len());
for mut task in tasks.drain(..) {
if task.as_mut().poll(&mut cx).is_pending() {
alive.push(task);
}
// Ready tasks are dropped — their futures have completed.
}
self.inner.tasks.lock().extend(alive);
}
/// Advance the virtual clock by `duration`.
///
/// Pending timers fire on the next [`step`](Self::step) after their
/// deadline has been reached.
pub fn advance_time(&self, duration: Duration) {
let mut now = self.clock.lock();
*now += duration;
}
/// Read the current virtual time.
pub fn virtual_now(&self) -> EngineInstant {
EngineInstant {
instant: *self.clock.lock(),
}
}
/// Number of live (not-yet-completed) tasks in the queue.
pub fn pending_task_count(&self) -> usize {
self.inner.tasks.lock().len()
}
}
/// This impl is the substrate implementor for the deterministic stepping
/// backend; blocking work runs on a std thread (ENGINE_SPEC.md §2).
#[allow(clippy::disallowed_methods)]
impl ExecutionBackend for SteppingBackend {
fn spawn(&self, task: BoxTask) {
self.inner.tasks.lock().push(task);
}
fn spawn_blocking(&self, work: BoxWork) {
let handle = std::thread::spawn(work);
self.inner.blocking.lock().push(handle);
}
fn timer(&self, delay: Duration) -> BoxTimer {
let deadline = *self.clock.lock() + delay;
Box::pin(SteppingTimer {
deadline,
clock: Arc::clone(&self.clock),
})
}
fn now(&self) -> EngineInstant {
EngineInstant {
instant: *self.clock.lock(),
}
}
fn capabilities(&self) -> Capabilities {
Capabilities {
tasks: true,
timers: true,
blocking: true,
io: false,
}
}
}

167
crates/engine/src/time.rs Normal file
View file

@ -0,0 +1,167 @@
//! Engine time types: a monotonic clock instant, a delay timer, and a recurring
//! interval.
use std::pin::Pin;
use std::sync::Weak;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
use crate::backend::{BoxTimer, ExecutionBackend};
/// A monotonic engine time instant.
///
/// Wraps [`std::time::Instant`] on the native Tokio backend. A later
/// deterministic engine can substitute virtual time behind the same type
/// without changing engine-hosted code (see ENGINE_SPEC.md §6).
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)]
pub struct EngineInstant {
pub(crate) instant: Instant,
}
impl EngineInstant {
/// Create an `EngineInstant` from the real monotonic clock.
///
/// Engine-hosted code should prefer [`EngineHandle::now`](crate::EngineHandle::now)
/// so that a deterministic engine can substitute virtual time. This
/// constructor exists for contexts that need a wall-clock reference
/// point outside the engine handle (e.g. a test mock backend).
pub fn now() -> Self {
Self {
instant: Instant::now(),
}
}
/// Return the underlying [`Instant`].
///
/// Core protocol messages that carry `std::time::Instant` (e.g.
/// `SwimIn::Tick { now }`) read the engine clock via
/// [`EngineHandle::now`](crate::EngineHandle::now) and convert here, so
/// that a deterministic engine can still substitute virtual time through
/// the same path (ENGINE_SPEC.md).
pub fn to_instant(self) -> Instant {
self.instant
}
}
/// A future that completes after a configured delay.
///
/// Constructed by [`EngineHandle::timer`](crate::EngineHandle::timer). Safe to
/// construct outside the substrate runtime (ENGINE_SPEC.md §7): the underlying
/// delay is armed lazily on first poll, which runs inside an engine task where
/// the substrate's time driver is available. The delay is therefore measured
/// from first poll, not from construction.
pub struct Timer {
pub(crate) inner: BoxTimer,
}
impl Timer {
/// A timer that never fires — used when the owning engine has been
/// dropped, so a handle can still produce a [`Timer`] without keeping the
/// backend alive (ENGINE_SPEC.md).
pub(crate) fn closed() -> Self {
Timer { inner: Box::pin(std::future::pending()) }
}
}
impl Future for Timer {
type Output = ();
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
// `Timer` is `Unpin` (it holds only a `Pin<Box<…>>`), so projecting the
// pin onto the inner timer is sound without `unsafe`.
self.get_mut().inner.as_mut().poll(cx)
}
}
/// Error returned when a [`Timeout`] elapses before the inner future completes.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct Elapsed;
/// A future that races an inner future against an engine [`Timer`].
///
/// Constructed by [`EngineHandle::timeout`](crate::EngineHandle::timeout).
/// This is a composed helper — not a new backend primitive (ENGINE_SPEC.md). It polls an engine timer and the inner future; whichever completes
/// first determines the result. When the timer fires first the future resolves
/// to [`Err(Elapsed)`](Elapsed).
///
/// The inner future is boxed once at construction so the type is `Unpin`
/// regardless of `F`.
pub struct Timeout<F: Future> {
future: Pin<Box<F>>,
timer: Timer,
}
impl<F: Future> Timeout<F> {
pub(crate) fn new(timer: Timer, future: F) -> Self {
Timeout {
future: Box::pin(future),
timer,
}
}
}
impl<F: Future> Future for Timeout<F> {
type Output = Result<F::Output, Elapsed>;
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
// Timeout<F> is Unpin: Timer is Unpin and Pin<Box<F>> is always Unpin.
let this = self.get_mut();
if Pin::new(&mut this.timer).poll(cx).is_ready() {
return Poll::Ready(Err(Elapsed));
}
if let Poll::Ready(value) = this.future.as_mut().poll(cx) {
return Poll::Ready(Ok(value));
}
Poll::Pending
}
}
/// A future that recurs at a fixed period.
///
/// Constructed by [`EngineHandle::interval`](crate::EngineHandle::interval).
/// Each completion re-arms a fresh backend timer, so awaiting it repeatedly
/// yields one ready per period. The first contract requires recurrence only;
/// it does not define whether missed periods burst, skip, or shift.
///
/// Unlike the generic `Future` convention, polling after a `Poll::Ready` is
/// defined behavior for this type: it re-arms and fires again on the next
/// period. This is what lets it be awaited in a loop.
pub struct Interval {
pub(crate) period: Duration,
pub(crate) backend: Weak<dyn ExecutionBackend>,
pub(crate) current: Option<BoxTimer>,
}
impl Future for Interval {
type Output = ();
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
let this = self.get_mut();
// Arm a timer on first poll (lazy: safe to construct off-runtime). If
// the owning engine is gone there is no substrate to arm against, so
// the interval simply stops firing.
if this.current.is_none() {
if let Some(backend) = this.backend.upgrade() {
this.current = Some(backend.timer(this.period));
} else {
return Poll::Pending;
}
}
// Poll the active timer; the mutable borrow ends at the `;` so the
// re-arm assignment below is legal.
let ready = this
.current
.as_mut()
.expect("timer armed above")
.as_mut()
.poll(cx)
.is_ready();
if ready {
// Re-arm for the next period; if the engine dropped between ticks,
// drop the spent timer so the next poll returns Pending.
this.current = this.backend.upgrade().map(|b| b.timer(this.period));
Poll::Ready(())
} else {
Poll::Pending
}
}
}

146
crates/engine/src/tokio.rs Normal file
View file

@ -0,0 +1,146 @@
//! Native Tokio execution backend.
//!
//! Owns a Tokio multi-threaded runtime whose `Handle` stays private. The
//! [`ExecutionBackend`](crate::ExecutionBackend) impl schedules cooperative
//! work onto that runtime; the `Handle` is never exposed through
//! [`EngineHandle`](crate::EngineHandle).
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};
use crate::backend::{BoxTask, BoxTimer, BoxWork, Capabilities, EngineError, ExecutionBackend};
use crate::time::EngineInstant;
/// Configuration for [`TokioBackend`].
#[derive(Debug, Clone, Copy)]
pub struct TokioConfig {
/// Number of async worker threads backing the runtime.
pub worker_threads: usize,
}
impl Default for TokioConfig {
fn default() -> Self {
Self { worker_threads: 2 }
}
}
/// An [`ExecutionBackend`](crate::ExecutionBackend) backed by an owned Tokio
/// multi-threaded runtime.
///
/// The runtime's `Handle` is never exposed through
/// [`EngineHandle`](crate::EngineHandle).
pub struct TokioBackend {
pub(crate) runtime: tokio::runtime::Runtime,
}
impl TokioBackend {
/// Build a backend with its own Tokio runtime tuned by `config`.
///
/// The runtime is owned and self-driving: its worker threads start at
/// construction, so spawned tasks progress without an ambient runtime or a
/// `block_on` driver. Tokio cancels spawned tasks (including the
/// core-driving loop) on `Runtime::drop`, so dropping the backend is
/// deterministic.
// The engine's Tokio backend is the substrate owner: it is the one place
// permitted to construct a Tokio runtime (ENGINE_SPEC.md §2).
#[allow(clippy::disallowed_methods)]
pub fn new(config: TokioConfig) -> Result<Self, EngineError> {
let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(config.worker_threads)
.enable_all()
.build()
.map_err(|e| EngineError::BackendSetup(e.to_string()))?;
Ok(Self { runtime })
}
/// Adopt a caller-tuned Tokio runtime, moving it into engine ownership.
pub fn from_runtime(runtime: tokio::runtime::Runtime) -> Self {
Self { runtime }
}
}
/// This impl is the Tokio substrate implementor: it is the one place permitted
/// to schedule directly on the owned runtime (ENGINE_SPEC.md §2).
#[allow(clippy::disallowed_methods)]
impl ExecutionBackend for TokioBackend {
fn spawn(&self, task: BoxTask) {
// The handle is used ephemerally and never stored or returned.
self.runtime.handle().spawn(task);
}
fn spawn_blocking(&self, work: BoxWork) {
// Routed onto the runtime's dedicated blocking pool — separate from
// the async worker threads — so blocking work cannot starve actor
// ticks (ENGINE_SPEC.md §8 progress independence).
self.runtime.handle().spawn_blocking(work);
}
fn timer(&self, delay: Duration) -> BoxTimer {
// Construct the `tokio::time::sleep` lazily on first poll rather than
// here: `EngineHandle::timer` may be called outside the runtime
// (ENGINE_SPEC.md §7), but `tokio::time::sleep` needs the time driver
// at construction. First poll runs inside an engine task where the
// driver is available. See `LazySleep`.
Box::pin(LazySleep::new(delay))
}
fn now(&self) -> EngineInstant {
EngineInstant { instant: Instant::now() }
}
fn capabilities(&self) -> Capabilities {
// The substrate physically provides tasks, timers, blocking, and I/O.
// `enable_all()` starts both the I/O reactor and the time driver, so
// advertising `io: true` is truthful — integrations such as Iroh rely
// on the native Tokio I/O environment (ENGINE_SPEC.md §6/§9).
Capabilities {
tasks: true,
timers: true,
blocking: true,
io: true,
}
}
}
/// A `tokio::time::sleep` whose construction is deferred to first poll.
///
/// `EngineHandle::timer` may be called outside the substrate runtime
/// (ENGINE_SPEC.md §7: creating a timer must not require entering or possessing
/// the runtime). `tokio::time::sleep` itself needs the time driver at
/// construction and panics ("there is no reactor running") when built outside a
/// Tokio context. This wrapper holds only the delay until first poll, which
/// runs inside an engine task where the driver is available, then builds and
/// delegates to the real `Sleep`.
struct LazySleep {
delay: Option<Duration>,
inner: Option<Pin<Box<tokio::time::Sleep>>>,
}
impl LazySleep {
fn new(delay: Duration) -> Self {
Self { delay: Some(delay), inner: None }
}
}
// `LazySleep` arms a `tokio::time::sleep` inside an engine task where the time
// driver is available; this is the substrate's own time primitive.
#[allow(clippy::disallowed_methods)]
impl Future for LazySleep {
type Output = ();
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
// `LazySleep` is `Unpin`: both fields (`Option<Duration>` and
// `Option<Pin<Box<_>>>`) are `Unpin`, so `get_mut` is sound.
let this = self.get_mut();
if let Some(delay) = this.delay.take() {
this.inner = Some(Box::pin(tokio::time::sleep(delay)));
}
this.inner
.as_mut()
.expect("LazySleep polled after completion")
.as_mut()
.poll(cx)
}
}

93
crates/engine/tests/common/mod.rs Normal file
View file

@ -0,0 +1,93 @@
//! Shared probe actors and bounded-wait helpers for the engine contract tests.
//!
//! Imports only public `swactor` APIs and exposes no private engine state. See
//! `ENGINE_SPEC.md`.
#![allow(dead_code)]
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::time::{Duration, Instant};
use swactor::actor::{ActorInterface, Ctx};
// ── Probe message ───────────────────────────────────────────────────────────
/// A minimal message delivered to probe actors.
#[derive(Clone, Debug)]
pub struct Probe;
// ── Probe actors ────────────────────────────────────────────────────────────
/// Records how many messages it has received into a shared counter.
pub struct RecordingProbe {
pub received: Arc<AtomicUsize>,
}
impl ActorInterface for RecordingProbe {
type Incoming = Probe;
type Response = ();
fn handle(&mut self, _ctx: &Ctx, _msg: Probe) {
self.received.fetch_add(1, Ordering::SeqCst);
}
}
/// Detects concurrent or reentrant handler entry. A violation is recorded if
/// `handle` is entered while a previous invocation is still in flight — which
/// can only happen if two ticks run the same worker concurrently.
pub struct ReentrancyGuardProbe {
pub entered: Arc<AtomicBool>,
pub violations: Arc<AtomicUsize>,
pub handled: Arc<AtomicUsize>,
}
impl ActorInterface for ReentrancyGuardProbe {
type Incoming = Probe;
type Response = ();
fn handle(&mut self, _ctx: &Ctx, _msg: Probe) {
if self.entered.swap(true, Ordering::SeqCst) {
self.violations.fetch_add(1, Ordering::SeqCst);
}
self.handled.fetch_add(1, Ordering::SeqCst);
self.entered.store(false, Ordering::SeqCst);
}
}
// ── Wait helpers ────────────────────────────────────────────────────────────
/// Block until `cond` holds, polling every 2 ms up to `timeout`. Returns the
/// final value of `cond` (true on success).
pub fn wait_for<F: Fn() -> bool>(cond: F, timeout: Duration) -> bool {
const POLL: Duration = Duration::from_millis(2);
let deadline = Instant::now() + timeout;
loop {
if cond() {
return true;
}
if Instant::now() >= deadline {
return cond();
}
std::thread::sleep(POLL);
}
}
/// A substrate-agnostic cooperative yield: suspend the current task for one
/// scheduler turn (giving other engine work — including the core driver — a
/// chance to run), then resume. Uses only `std`, so it works on any substrate
/// without coupling the test to Tokio.
pub async fn yield_once() {
// The closure is stored inside `poll_fn`'s future and polled via `&mut`,
// so its captured `yielded` flag persists across polls: the first poll
// reschedules and suspends, the next poll resumes.
let mut yielded = false;
std::future::poll_fn(move |cx| {
if yielded {
std::task::Poll::Ready(())
} else {
yielded = true;
cx.waker().wake_by_ref();
std::task::Poll::Pending
}
})
.await;
}

366
crates/engine/tests/engine_contract.rs Normal file
View file

@ -0,0 +1,366 @@
#![cfg(feature = "tokio")]
//! Black-box contract tests for the swactor engine: baseline driving/tasks,
//! time, blocking, and non-reentrancy.
//!
//! These are ordinary synchronous `#[test]`s. They construct and own their
//! engine explicitly, never call `tick()`/`try_tick()`, never use
//! `#[tokio::test]`, and observe behavior through atomics and bounded channels
//! with finite deadlines. See `ENGINE_SPEC.md`.
//!
//! These tests exercise the native Tokio backend specifically; the
//! non-Tokio portability proof lives in `engine_unit.rs`.
mod common;
use common::*;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize};
use std::sync::atomic::Ordering::SeqCst;
use std::time::Duration;
use swactor::config::RuntimeConfig;
use swactor::runtime::Runtime;
use swactor_engine::{Engine, TokioBackend, TokioConfig};
/// Outer deadline shared across tests: generous enough to absorb scheduler
/// jitter, short enough that a hung test terminates.
const DEADLINE: Duration = Duration::from_secs(5);
// ── 7.1 ──────────────────────────────────────────────────────────────────────
#[test]
fn engine_runs_without_an_ambient_tokio_runtime() {
// No outer Tokio runtime, no `#[tokio::test]`. The engine owns its runtime.
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let backend = TokioBackend::new(TokioConfig::default()).expect("build tokio backend");
let engine = Engine::new(runtime, backend).expect("construct engine");
let handle = engine.handle();
let (tx, rx) = std::sync::mpsc::sync_channel(1);
handle.spawn(async move {
let _ = tx.send(());
});
rx.recv_timeout(DEADLINE)
.expect("spawned work must signal without an ambient runtime");
}
// ── 7.2 ──────────────────────────────────────────────────────────────────────
#[test]
fn engine_drives_core_without_application_ticks() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let received = Arc::new(AtomicUsize::new(0));
let addr = runtime
.spawn(RecordingProbe {
received: received.clone(),
})
.expect("spawn probe actor");
let backend = TokioBackend::new(TokioConfig::default()).expect("build tokio backend");
let _engine = Engine::new(runtime.clone(), backend).expect("construct engine");
// Deliver AFTER engine construction: a later tick must observe it.
runtime
.send_to(addr, Probe)
.expect("deliver probe message");
assert!(
wait_for(|| received.load(SeqCst) >= 1, DEADLINE),
"actor must process a message without any application tick"
);
}
// ── 7.3 ──────────────────────────────────────────────────────────────────────
#[test]
fn spawned_supporting_work_runs() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let backend = TokioBackend::new(TokioConfig::default()).expect("build tokio backend");
let engine = Engine::new(runtime, backend).expect("construct engine");
let handle = engine.handle();
let (tx, rx) = std::sync::mpsc::sync_channel(1);
handle.spawn(async move {
let _ = tx.send(());
});
rx.recv_timeout(DEADLINE)
.expect("opaque spawned task must signal");
}
// ── 7.4 ──────────────────────────────────────────────────────────────────────
#[test]
fn actor_ticks_and_supporting_work_both_progress() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let received = Arc::new(AtomicUsize::new(0));
let addr = runtime
.spawn(RecordingProbe {
received: received.clone(),
})
.expect("spawn probe actor");
let backend = TokioBackend::new(TokioConfig::default()).expect("build tokio backend");
let engine = Engine::new(runtime.clone(), backend).expect("construct engine");
let handle = engine.handle();
// Long-lived cooperative supporting work that yields between steps so it
// stays active while the actor message is processed.
let steps = Arc::new(AtomicUsize::new(0));
let steps_for_task = steps.clone();
handle.spawn(async move {
for _ in 0..200 {
steps_for_task.fetch_add(1, SeqCst);
yield_once().await;
}
});
// Deliver an actor message while the supporting work is still active.
runtime
.send_to(addr, Probe)
.expect("deliver probe message");
assert!(
wait_for(
|| steps.load(SeqCst) >= 200 && received.load(SeqCst) >= 1,
DEADLINE,
),
"both actor ticks and supporting work must progress"
);
}
// ── 7.6 ──────────────────────────────────────────────────────────────────────
#[test]
fn runtime_ticks_are_never_concurrent() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let entered = Arc::new(AtomicBool::new(false));
let violations = Arc::new(AtomicUsize::new(0));
let handled = Arc::new(AtomicUsize::new(0));
let addr = runtime
.spawn(ReentrancyGuardProbe {
entered: entered.clone(),
violations: violations.clone(),
handled: handled.clone(),
})
.expect("spawn reentrancy probe");
let backend = TokioBackend::new(TokioConfig::default()).expect("build tokio backend");
let _engine = Engine::new(runtime.clone(), backend).expect("construct engine");
let sender = runtime.create_sender();
const SENDERS: usize = 4;
const PER_SENDER: usize = 250;
const TOTAL: usize = SENDERS * PER_SENDER;
// Many messages from multiple external threads, all concurrent with the
// engine's single driving loop.
let mut threads = Vec::new();
for _ in 0..SENDERS {
let sender = sender.clone();
threads.push(std::thread::spawn(move || {
for _ in 0..PER_SENDER {
let _ = sender.send_to(addr, Probe);
}
}));
}
for t in threads {
t.join().expect("sender thread panicked");
}
assert!(
wait_for(|| handled.load(SeqCst) >= TOTAL, Duration::from_secs(10)),
"all messages must be processed"
);
assert_eq!(
violations.load(SeqCst),
0,
"detected a concurrent or reentrant tick"
);
}
// ── 7.5 ──────────────────────────────────────────────────────────────────────
/// Releases a [`Barrier`](std::sync::Barrier) on drop so blocking test work can
/// finish even when an assertion fails before explicit cleanup.
struct BarrierRelease(Arc<std::sync::Barrier>);
impl Drop for BarrierRelease {
fn drop(&mut self) {
self.0.wait();
}
}
#[test]
fn blocking_work_does_not_stop_actor_ticks() {
// A blocking-capability test, not part of the baseline tasks-plus-time
// contract. Configure a small async worker pool so passing cannot be an
// accident of excessive worker count.
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let received = Arc::new(AtomicUsize::new(0));
let addr = runtime
.spawn(RecordingProbe {
received: received.clone(),
})
.expect("spawn probe actor");
let backend = TokioBackend::new(TokioConfig { worker_threads: 1 })
.expect("build tokio backend");
let engine = Engine::new(runtime.clone(), backend).expect("construct engine");
let handle = engine.handle();
// Blocking work that waits on a barrier; it stays stuck for the whole test
// body. It runs on the blocking pool, not the single async worker, so actor
// ticks must still progress (ENGINE_SPEC.md §8 progress independence).
let barrier = Arc::new(std::sync::Barrier::new(2));
// `_release` drops at scope end — even on panic — to release the blocking
// task so the owned runtime shuts down deterministically.
let _release = BarrierRelease(barrier.clone());
let barrier_for_work = barrier.clone();
handle.spawn_blocking(move || {
barrier_for_work.wait();
});
// Deliver an actor message while the blocking work remains blocked.
runtime
.send_to(addr, Probe)
.expect("deliver probe message");
assert!(
wait_for(|| received.load(SeqCst) >= 1, DEADLINE),
"actor ticks must progress while blocking work is stuck"
);
}
// ── 7.7 ──────────────────────────────────────────────────────────────────────
#[test]
fn engine_clock_is_monotonic() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let backend = TokioBackend::new(TokioConfig::default()).expect("build tokio backend");
let engine = Engine::new(runtime, backend).expect("construct engine");
let handle = engine.handle();
let mut prev = handle.now();
for _ in 0..10_000 {
let cur = handle.now();
assert!(cur >= prev, "engine clock moved backwards");
prev = cur;
}
}
// ── 7.8 ──────────────────────────────────────────────────────────────────────
#[test]
fn engine_timer_fires() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let backend = TokioBackend::new(TokioConfig::default()).expect("build tokio backend");
let engine = Engine::new(runtime, backend).expect("construct engine");
let handle = engine.handle();
let (tx, rx) = std::sync::mpsc::sync_channel(1);
let timer_handle = handle.clone();
handle.spawn(async move {
timer_handle.timer(Duration::from_millis(20)).await;
let _ = tx.send(());
});
rx.recv_timeout(DEADLINE)
.expect("engine timer must fire");
}
// ── 7.9 ──────────────────────────────────────────────────────────────────────
#[test]
fn engine_interval_recurs() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let backend = TokioBackend::new(TokioConfig::default()).expect("build tokio backend");
let engine = Engine::new(runtime, backend).expect("construct engine");
let handle = engine.handle();
let (tx, rx) = std::sync::mpsc::sync_channel(1);
let interval_handle = handle.clone();
handle.spawn(async move {
// `Interval` re-arms on each `Ready`, so awaiting it repeatedly yields
// one ready per period. `Box::pin` lets us poll it in a loop.
let mut interval = Box::pin(interval_handle.interval(Duration::from_millis(5)));
for _ in 0..3 {
interval.as_mut().await;
}
let _ = tx.send(());
});
rx.recv_timeout(DEADLINE)
.expect("interval must recur several times");
}
// ── 7.10 ─────────────────────────────────────────────────────────────────────
#[test]
fn engine_timer_can_be_created_off_runtime() {
// ENGINE_SPEC.md §7: creating an engine timer
// must not require the caller to enter or possess the raw substrate runtime.
// Construct the timer directly in the test body — no spawned task, no ambient
// runtime — then await it on an engine task. If the tokio backend's timer
// needed runtime context at construction, this would panic.
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let backend = TokioBackend::new(TokioConfig::default()).expect("build tokio backend");
let engine = Engine::new(runtime, backend).expect("construct engine");
let handle = engine.handle();
// Constructed off-runtime: must not panic.
let timer = handle.timer(Duration::from_millis(10));
let (tx, rx) = std::sync::mpsc::sync_channel(1);
handle.spawn(async move {
timer.await;
let _ = tx.send(());
});
rx.recv_timeout(DEADLINE)
.expect("off-runtime-constructed timer must fire when awaited on a task");
}
// ── 7.11 ─────────────────────────────────────────────────────────────────────
// This test exercises `TokioBackend::from_runtime`, so it must build a real
// Tokio runtime to hand the engine — the one test-only use of the substrate
// constructor (ENGINE_SPEC.md §2).
#[test]
#[allow(clippy::disallowed_methods)]
fn engine_adopts_caller_tuned_tokio_runtime() {
// ENGINE_SPEC.md §9: the native engine supports
// consuming an explicitly tuned Tokio runtime rather than always building
// its own. Build a runtime with a non-default worker count, transfer it,
// and confirm the engine still drives core and reports full capabilities.
let tuned = tokio::runtime::Builder::new_multi_thread()
.worker_threads(3)
.enable_all()
.build()
.expect("build tuned tokio runtime");
let backend = TokioBackend::from_runtime(tuned);
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let received = Arc::new(AtomicUsize::new(0));
let addr = runtime
.spawn(RecordingProbe {
received: received.clone(),
})
.expect("spawn probe actor");
let engine = Engine::new(runtime.clone(), backend).expect("construct engine");
// The adopted substrate still exposes every native capability (§9).
assert_eq!(
engine.handle().capabilities(),
swactor_engine::Capabilities::ALL
);
runtime.send_to(addr, Probe).expect("deliver probe");
assert!(
wait_for(|| received.load(SeqCst) >= 1, DEADLINE),
"engine must drive core through an adopted runtime"
);
}

587
crates/engine/tests/engine_unit.rs Normal file
View file

@ -0,0 +1,587 @@
//! Unit tests for engine logic — capability binding, time semantics, and the
//! non-Tokio portability proof.
//!
//! Deliberately separate from `engine_contract.rs` (the behavioral contract).
//! These tests exercise internal logic directly and use the [`SteppingBackend`]
//! to prove substrate independence without Tokio (ENGINE_SPEC.md).
mod common;
use common::*;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicUsize};
use std::sync::atomic::Ordering::SeqCst;
use std::time::Duration;
use swactor::config::RuntimeConfig;
use swactor::runtime::Runtime;
use swactor_engine::{
Capabilities, Engine, EngineError, ExecutionBackend, SteppingBackend,
};
#[cfg(feature = "tokio")]
use swactor_engine::{TokioBackend, TokioConfig};
// ═══════════════════════════════════════════════════════════════════════════════
// §1 Capabilities::satisfies logic
// ═══════════════════════════════════════════════════════════════════════════════
#[test]
fn satisfies_all_true_meets_all_requirements() {
let full = Capabilities::ALL;
assert!(full.satisfies(Capabilities::ALL));
assert!(full.satisfies(Capabilities::TASKS_ONLY));
assert!(full.satisfies(Capabilities {
tasks: true,
timers: true,
blocking: true,
io: true,
}));
}
#[test]
fn satisfies_empty_required_always_passes() {
let none_required = Capabilities {
tasks: false,
timers: false,
blocking: false,
io: false,
};
let weak = Capabilities::TASKS_ONLY;
assert!(weak.satisfies(none_required));
assert!(Capabilities::ALL.satisfies(none_required));
}
#[test]
fn satisfies_missing_single_capability_fails() {
let backend = Capabilities {
tasks: true,
timers: true,
blocking: true,
io: false,
};
assert!(!backend.satisfies(Capabilities::ALL));
assert!(backend.satisfies(Capabilities {
tasks: true,
timers: true,
blocking: true,
io: false,
}));
}
#[test]
fn satisfies_tasks_only_does_not_imply_timers() {
let tasks_only = Capabilities::TASKS_ONLY;
assert!(tasks_only.satisfies(Capabilities::TASKS_ONLY));
assert!(!tasks_only.satisfies(Capabilities {
tasks: true,
timers: true,
blocking: false,
io: false,
}));
}
#[test]
fn capabilities_constants_are_correct() {
assert_eq!(
Capabilities::TASKS_ONLY,
Capabilities {
tasks: true,
timers: false,
blocking: false,
io: false,
}
);
assert_eq!(
Capabilities::ALL,
Capabilities {
tasks: true,
timers: true,
blocking: true,
io: true,
}
);
}
// ═══════════════════════════════════════════════════════════════════════════════
// §2 Engine construction / capability rejection
// ═══════════════════════════════════════════════════════════════════════════════
/// A minimal backend that advertises no capabilities — used to verify
/// `Engine::new` rejects it.
struct NoCapBackend;
impl swactor_engine::ExecutionBackend for NoCapBackend {
fn spawn(&self, _: swactor_engine::BoxTask) {}
fn spawn_blocking(&self, _: swactor_engine::BoxWork) {}
fn timer(&self, _: Duration) -> swactor_engine::BoxTimer {
unreachable!("no capabilities")
}
fn now(&self) -> swactor_engine::EngineInstant {
swactor_engine::EngineInstant::now()
}
fn capabilities(&self) -> Capabilities {
Capabilities {
tasks: false,
timers: false,
blocking: false,
io: false,
}
}
}
#[test]
fn engine_new_rejects_backend_without_tasks() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let result = Engine::new(runtime, NoCapBackend);
assert!(
matches!(result, Err(EngineError::MissingRequiredCapability)),
"Engine::new must reject a backend that cannot schedule tasks"
);
}
#[test]
fn engine_new_accepts_stepping_backend() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let backend = SteppingBackend::new();
let engine = Engine::new(runtime, backend).expect("stepping backend has tasks");
drop(engine);
}
// ═══════════════════════════════════════════════════════════════════════════════
// §3 EngineHandle::require — capability binding
// ═══════════════════════════════════════════════════════════════════════════════
#[test]
fn require_accepts_when_all_capabilities_present() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let backend = SteppingBackend::new();
let engine = Engine::new(runtime, backend).unwrap();
let handle = engine.handle();
// Stepping provides tasks + timers + blocking.
let required = Capabilities {
tasks: true,
timers: true,
blocking: true,
io: false,
};
assert!(handle.require(required).is_ok());
}
#[test]
fn require_rejects_when_io_missing() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let backend = SteppingBackend::new();
let engine = Engine::new(runtime, backend).unwrap();
let handle = engine.handle();
// Stepping does NOT provide io.
assert!(handle.require(Capabilities::ALL).is_err());
}
#[test]
fn require_rejects_when_timers_missing() {
struct TaskOnlyBackend;
impl swactor_engine::ExecutionBackend for TaskOnlyBackend {
fn spawn(&self, _: swactor_engine::BoxTask) {}
fn spawn_blocking(&self, _: swactor_engine::BoxWork) {}
fn timer(&self, _: Duration) -> swactor_engine::BoxTimer {
unreachable!()
}
fn now(&self) -> swactor_engine::EngineInstant {
swactor_engine::EngineInstant::now()
}
fn capabilities(&self) -> Capabilities {
Capabilities::TASKS_ONLY
}
}
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let engine = Engine::new(runtime, TaskOnlyBackend).unwrap();
let handle = engine.handle();
assert!(
handle
.require(Capabilities {
tasks: true,
timers: true,
blocking: false,
io: false,
})
.is_err(),
"must reject when timers required but not provided"
);
assert!(handle.require(Capabilities::TASKS_ONLY).is_ok());
}
#[test]
fn require_can_be_called_multiple_times() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let backend = SteppingBackend::new();
let engine = Engine::new(runtime, backend).unwrap();
let handle = engine.handle();
assert!(handle.require(Capabilities::TASKS_ONLY).is_ok());
assert!(handle.require(Capabilities::TASKS_ONLY).is_ok());
assert!(handle.require(Capabilities::ALL).is_err());
assert!(handle.require(Capabilities::ALL).is_err());
}
// ═══════════════════════════════════════════════════════════════════════════════
// §4 Truthful capability reporting
// ═══════════════════════════════════════════════════════════════════════════════
#[test]
fn stepping_backend_reports_no_io() {
let backend = SteppingBackend::new();
let caps = backend.capabilities();
assert!(caps.tasks);
assert!(caps.timers);
assert!(caps.blocking);
assert!(!caps.io, "stepping backend must not advertise io");
}
#[cfg(feature = "tokio")]
#[test]
fn tokio_backend_reports_all_capabilities() {
let backend = TokioBackend::new(TokioConfig::default()).expect("build tokio backend");
let caps = backend.capabilities();
assert!(caps.tasks, "tokio must advertise tasks");
assert!(caps.timers, "tokio must advertise timers");
assert!(caps.blocking, "tokio must advertise blocking");
assert!(
caps.io,
"tokio must advertise io — enable_all starts the I/O reactor"
);
}
// ═══════════════════════════════════════════════════════════════════════════════
// §5 Portability proof: SteppingBackend (ENGINE_SPEC.md)
// ═══════════════════════════════════════════════════════════════════════════════
/// Enough steps to let the driver loop tick and process queued work.
const STEPS: usize = 30;
#[test]
fn stepping_core_progresses_without_tokio() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let received = Arc::new(AtomicUsize::new(0));
let addr = runtime
.spawn(RecordingProbe {
received: received.clone(),
})
.expect("spawn probe");
let backend = SteppingBackend::new();
let _engine = Engine::new(runtime.clone(), backend.clone()).expect("construct engine");
// Deliver AFTER engine construction — a later tick must observe it.
runtime.send_to(addr, Probe).expect("deliver probe");
for _ in 0..STEPS {
backend.step();
}
assert!(
received.load(SeqCst) >= 1,
"actor must process a message without any application tick or tokio"
);
}
#[test]
fn stepping_supporting_work_progresses() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let backend = SteppingBackend::new();
let engine = Engine::new(runtime, backend.clone()).expect("construct engine");
let handle = engine.handle();
let done = Arc::new(AtomicBool::new(false));
let done_clone = done.clone();
handle.spawn(async move {
done_clone.store(true, SeqCst);
});
for _ in 0..STEPS {
backend.step();
}
assert!(
done.load(SeqCst),
"spawned supporting work must complete without tokio"
);
}
#[test]
fn stepping_core_and_supporting_work_both_progress() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let received = Arc::new(AtomicUsize::new(0));
let addr = runtime
.spawn(RecordingProbe {
received: received.clone(),
})
.expect("spawn probe");
let backend = SteppingBackend::new();
let engine = Engine::new(runtime.clone(), backend.clone()).expect("construct engine");
let handle = engine.handle();
// Long-lived cooperative supporting work that yields between steps.
let steps = Arc::new(AtomicUsize::new(0));
let steps_clone = steps.clone();
handle.spawn(async move {
for _ in 0..10 {
steps_clone.fetch_add(1, SeqCst);
yield_once().await;
}
});
runtime.send_to(addr, Probe).expect("deliver probe");
for _ in 0..(STEPS * 2) {
backend.step();
}
assert!(
steps.load(SeqCst) >= 10,
"supporting work must finish"
);
assert!(
received.load(SeqCst) >= 1,
"actor message must be processed"
);
}
#[test]
fn stepping_virtual_time_is_monotonic() {
let backend = SteppingBackend::new();
let mut prev = backend.virtual_now();
for _ in 0..100 {
backend.advance_time(Duration::from_millis(1));
let cur = backend.virtual_now();
assert!(cur > prev, "virtual clock must advance");
prev = cur;
}
}
#[test]
fn stepping_virtual_now_matches_engine_now() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let backend = SteppingBackend::new();
let engine = Engine::new(runtime, backend.clone()).unwrap();
let handle = engine.handle();
assert_eq!(handle.now(), backend.virtual_now());
backend.advance_time(Duration::from_secs(5));
assert_eq!(handle.now(), backend.virtual_now());
}
#[test]
fn stepping_timer_does_not_fire_before_advance() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let backend = SteppingBackend::new();
let engine = Engine::new(runtime, backend.clone()).expect("construct engine");
let handle = engine.handle();
let fired = Arc::new(AtomicBool::new(false));
let fired_clone = fired.clone();
let timer_handle = handle.clone();
handle.spawn(async move {
timer_handle.timer(Duration::from_secs(1)).await;
fired_clone.store(true, SeqCst);
});
for _ in 0..10 {
backend.step();
}
assert!(
!fired.load(SeqCst),
"timer must NOT fire before virtual time reaches the deadline"
);
}
#[test]
fn stepping_timer_fires_after_virtual_time_advance() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let backend = SteppingBackend::new();
let engine = Engine::new(runtime, backend.clone()).expect("construct engine");
let handle = engine.handle();
let fired = Arc::new(AtomicBool::new(false));
let fired_clone = fired.clone();
let timer_handle = handle.clone();
handle.spawn(async move {
timer_handle.timer(Duration::from_millis(500)).await;
fired_clone.store(true, SeqCst);
});
for _ in 0..10 {
backend.step();
}
assert!(!fired.load(SeqCst));
backend.advance_time(Duration::from_secs(1));
for _ in 0..10 {
backend.step();
}
assert!(
fired.load(SeqCst),
"timer must fire after virtual time advances past the deadline"
);
}
#[test]
fn stepping_blocking_work_runs_isolated() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let backend = SteppingBackend::new();
let engine = Engine::new(runtime, backend.clone()).expect("construct engine");
let handle = engine.handle();
let done = Arc::new(AtomicBool::new(false));
let done_clone = done.clone();
handle.spawn_blocking(move || {
std::thread::sleep(Duration::from_millis(10));
done_clone.store(true, SeqCst);
});
let deadline = std::time::Instant::now() + Duration::from_secs(2);
loop {
if done.load(SeqCst) {
break;
}
if std::time::Instant::now() > deadline {
panic!("blocking work did not complete within deadline");
}
std::thread::sleep(Duration::from_millis(5));
}
}
#[test]
fn stepping_spawned_task_completing_is_removed_from_queue() {
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let backend = SteppingBackend::new();
let engine = Engine::new(runtime, backend.clone()).expect("construct engine");
let handle = engine.handle();
handle.spawn(async {});
for _ in 0..5 {
backend.step();
}
assert_eq!(
backend.pending_task_count(),
1,
"only the core-driving loop should remain after spawned tasks complete"
);
}
// ═══════════════════════════════════════════════════════════════════════════════
// §6 Time types
// ═══════════════════════════════════════════════════════════════════════════════
#[test]
fn engine_instant_is_ordered() {
let backend = SteppingBackend::new();
let t0 = backend.virtual_now();
backend.advance_time(Duration::from_secs(1));
let t1 = backend.virtual_now();
backend.advance_time(Duration::from_secs(1));
let t2 = backend.virtual_now();
assert!(t0 < t1);
assert!(t1 < t2);
assert!(t0 < t2);
}
// ═══════════════════════════════════════════════════════════════════════════════
// §7 Engine ownership: handles never keep the backend alive (§4.1)
// ═══════════════════════════════════════════════════════════════════════════════
/// A tasks-only probe backend that shares a sentinel `Arc<()>` so the test can
/// observe exactly when the engine's strong backend reference is released.
struct SentinelBackend {
#[allow(dead_code)]
sentinel: Arc<()>,
}
impl ExecutionBackend for SentinelBackend {
fn spawn(&self, _: swactor_engine::BoxTask) {}
fn spawn_blocking(&self, _: swactor_engine::BoxWork) {}
fn timer(&self, _: Duration) -> swactor_engine::BoxTimer {
Box::pin(std::future::pending())
}
fn now(&self) -> swactor_engine::EngineInstant {
swactor_engine::EngineInstant::now()
}
fn capabilities(&self) -> Capabilities {
Capabilities::TASKS_ONLY
}
}
#[test]
fn dropping_engine_releases_backend_even_with_live_handles() {
// The engine is the sole strong owner of its backend. EngineHandle and
// interval state hold weak references, so the backend (and the runtime /
// core-driver task it owns) is released once the engine drops — even while
// handles remain alive (ENGINE_SPEC.md).
let sentinel = Arc::new(());
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let engine = Engine::new(
runtime,
SentinelBackend { sentinel: sentinel.clone() },
)
.expect("tasks capability present");
let handle = engine.handle();
let _handle_clone = handle.clone();
// Two strong refs: the test's `sentinel` and the backend's clone.
assert_eq!(Arc::strong_count(&sentinel), 2);
drop(engine);
// The engine was the sole strong backend owner; handles are weak, so the
// backend (and its sentinel clone) is gone.
assert_eq!(
Arc::strong_count(&sentinel),
1,
"backend retained after the owning engine was dropped"
);
// Dropping the surviving handles changes nothing — they never held a strong
// reference.
drop(handle);
drop(_handle_clone);
assert_eq!(Arc::strong_count(&sentinel), 1);
}
#[test]
fn handle_used_after_engine_drop_degrades_gracefully() {
// Behavior beyond the engine's lifetime is out of spec, but a handle must
// not retain the backend and should degrade through the smallest practical
// API rather than panic (ENGINE_SPEC.md).
let runtime = Arc::new(Runtime::new(RuntimeConfig::default()));
let engine = Engine::new(runtime, SteppingBackend::default()).unwrap();
let handle = engine.handle();
// Live handle reports the stepping backend's capabilities.
assert!(handle.capabilities().tasks);
assert!(handle.capabilities().timers);
drop(engine);
// Closed handle reports no capabilities and rejects every requirement.
assert_eq!(handle.capabilities(), Capabilities::NONE);
assert!(handle.require(Capabilities::TASKS_ONLY).is_err());
// Time falls back to the wall clock without panicking.
let _ = handle.now();
// Scheduling work and creating primitives are no-ops / never fire, never
// panic, and never retain the backend.
handle.spawn(async {});
handle.spawn_blocking(|| {});
let _never_fires = handle.timer(Duration::from_secs(1));
let _never_ticks = handle.interval(Duration::from_secs(1));
}

1
crates/iroh-driver/Cargo.toml
View file

@ -10,6 +10,7 @@ path = "src/lib.rs"
[dependencies]
swactor = { path = "../..", features = ["serde", "transport"] }
swactor-engine = { path = "../engine" }
swactor-transport = { path = "../transport" }
distribution = { path = "../distribution" }
datastream = { path = "../datastream" }

37
docs/specs/drafts/IROH_DRIVER_SPEC.md → crates/iroh-driver/IROH_DRIVER_SPEC.md
View file

@ -1,7 +1,7 @@
# Iroh Driver Fixed Specification
Id: 5
Last modified:
Last modified: f8fc594b95871813a890b5d60f60dee505ef93bc
Last reviewed:
> Review checkpoint: reviewed through Section 3.2; resume with Section 3.3 Accepted Connection Output.
@ -47,7 +47,7 @@ The driver accepts input through construction configuration, explicit method cal
### 2.1 Construction and Configuration Inputs
Construction input is `IrohDriverConfig` plus a Tokio runtime handle.
Construction input is `IrohDriverConfig` plus a swactor engine handle (see §2.1).
`IrohDriverConfig` contains:
@ -71,13 +71,15 @@ IrohDriverConfig {
`additional_alpns` registers internal wire ALPNs on the iroh endpoint. Non-actor connections negotiated on those wire protocols are handled by registered protocol adapters rather than returned directly to callers.
The target driver is constructed inside the process's single Tokio runtime:
The target driver is constructed with a swactor engine handle — the single
engine that owns the process's Tokio substrate:
```text
IrohDriver::with_handle(tokio_handle, config)
IrohDriver::with_engine(engine_handle, config)
```
The target driver must not create or own a second Tokio runtime. Any legacy constructor that creates or discovers a runtime is outside this target contract.
The driver validates engine capabilities (tasks, timers, io) before binding the
endpoint. It must not create, discover, or store a raw Tokio runtime handle.
### 2.2 Actor Egress Channel Input
@ -154,7 +156,7 @@ Each seed carries the peer public key and may carry direct socket addresses and
A join request is sent as a framed actor message over the actor ALPN. The request is addressed to the seed node's peer mailbox and uses the `JoinRequest` network message type tag.
Join attempts run in background Tokio tasks. The `join` call does not synchronously wait for connection establishment or membership convergence.
Join attempts run as engine-hosted tasks. The `join` call does not synchronously wait for connection establishment or membership convergence.
### 2.4 Incoming Iroh Connection Input
@ -285,7 +287,7 @@ Shutdown input is async and runtime-owned:
IrohDriver::close().await
```
`close` is the target teardown path. It closes the iroh endpoint from inside the same Tokio runtime that owns the driver tasks.
`close` is the target teardown path. It closes the iroh endpoint from inside the engine substrate that owns the driver tasks.
---
@ -662,35 +664,26 @@ updates join status during attempts
queues a successful connection for later cache folding
```
Membership convergence is not completed by `join` alone. SWIM and the distribution actors must be pumped after a connection exists.
Membership convergence is not completed by `join` alone. SWIM and the distribution actors must progress after a connection exists.
### 6.4 Running Pump Cycle
### 6.4 Engine-Hosted Progression
A production runtime loop must pump both the distribution actors and the driver.
Actor progression, protocol ticks, and all driver adapter work — inbound actor frames, actor egress, protocol adapter ingress/status, and ring-backed protocol pumps — are owned by a single swactor engine. The application installs the engine-hosted adapter pump via `IrohDriver::install_actor_bridge_pump` and does not pump any queue itself.
The expected pump shape is:
The engine-hosted pump cycle advances the following per interval:
```text
send protocol actor ticks
pump inbound iroh actor frames into Swactor
drain inbound iroh actor frames into Swactor
run Swactor runtime work
drain actor egress channel to iroh
drain protocol adapter ingress/status queues
wake or drain ring-backed protocol pumps as needed
```
The driver provides only its side of this loop:
```text
pump_inbound_to_actors
drain_actor_egress
drain_protocol_ingress
drain_protocol_status
```
Protocol adapter drains carry messages, logical stream status, readiness, and faults. Payload bytes for ring-backed protocols move through rings and ring wakeups, not through actor-drained byte messages.
The caller owns the loop, timing, shutdown select, and protocol actor tick injection.
The engine owns the loop, timing, and protocol actor tick injection; the application only configures components, consumes reports, and owns domain queues (ENGINE_SPEC.md §5).
### 6.5 Datastream Connection Handling

35
crates/iroh-driver/README.md
View file

@ -2,25 +2,30 @@
`iroh-driver` is the iroh-backed transport bridge for the actorized distribution stack. It owns the concrete iroh endpoint, QUIC connections, relay configuration, peer authorization, and frame shuttling between iroh and swactor actor mailboxes.
## Tokio runtime ownership
## Engine ownership
The driver needs Tokio because iroh's endpoint, accepts, dials, stream reads/writes, retry timers, and shutdown APIs are async. New call sites should make that engine explicit by constructing the driver with:
The driver runs on a caller-supplied swactor [`EngineHandle`](swactor_engine) — the
single engine that owns the node's Tokio substrate. All accepts, reads, dials,
writes, retries, and teardown are scheduled through that handle; the driver
stores no raw Tokio handle and performs no ambient-runtime detection
(ENGINE_SPEC.md §7).
```rust
let driver = IrohDriver::with_handle(tokio_handle, config)?;
let driver = IrohDriver::with_engine(engine.handle(), config)?;
```
`with_handle` does not own the Tokio runtime. The caller must keep the runtime alive for as long as the driver exists.
The driver validates that the engine provides the `tasks`, `timers`, and `io`
capabilities before binding the endpoint or starting any background work
(ENGINE_SPEC.md). Endpoint construction runs as an engine-hosted
task; `with_engine` blocks on a synchronous channel until the endpoint is bound
(or fails), so callers need not enter or possess the raw substrate runtime.
## Legacy implicit constructor
## Engine-hosted progression
`IrohDriver::new(config)` is still present as a compatibility convenience, but it hides runtime ownership:
- If called inside an existing Tokio runtime, it uses `Handle::try_current()` and shares that ambient engine.
- If called outside Tokio, it silently builds and owns a multi-threaded Tokio runtime with `enable_all()`.
Avoid `IrohDriver::new` in new production code. Use `with_handle` or an explicit engine wrapper at the application boundary so every Tokio engine in the process is visible in construction code.
## Sync facade caveat
The synchronous facade methods that bridge to async with `block_on` must run from a non-async thread. Do not call those methods from inside tasks running on the same Tokio runtime; Tokio will panic on nested `block_on`.
All adapter progression — actor-bridge ingress/egress, datastream ingress, and
edge ingress — is driven by an engine-hosted interval pump installed via
`install_actor_bridge_pump`. Applications do not (and cannot) manually pump
these adapters; the single engine owns progression for the node's lifetime
(ENGINE_SPEC.md). `snapshot` is a pure-synchronous read of driver
state, callable from any thread. The `shutdown` method closes the endpoint via
an engine-hosted task, blocking on a synchronous channel until completion.

53
crates/iroh-driver/src/datastream_transport.rs
View file

@ -11,7 +11,7 @@ use datastream::{
};
use iroh::endpoint::{Connection, RecvStream, SendStream};
use iroh::{Endpoint, EndpointAddr};
use tokio::runtime::Handle;
use swactor_engine::EngineHandle;
pub const DATASTREAM_ALPN: &[u8] = b"swactor/datastream/0";
@ -73,34 +73,36 @@ pub struct DatastreamQuicRead {
}
pub fn spawn_subscription_writer(
handle: &Handle,
engine: &EngineHandle,
endpoint: Endpoint,
peer: EndpointAddr,
header: DatastreamQuicHeader,
subscription: DatastreamSubscription,
idle_sleep: Duration,
) -> tokio::task::JoinHandle<Result<DatastreamQuicWriteStats, String>> {
handle.spawn(async move {
let conn = endpoint
.connect(peer, DATASTREAM_ALPN)
.await
.map_err(|error| error.to_string())?;
let send = conn.open_uni().await.map_err(|error| error.to_string())?;
write_subscription_until_closed(send, header, subscription, idle_sleep)
.await
.map_err(|error| error.to_string())
})
) {
let engine_handle = engine.clone();
engine.spawn(async move {
let Ok(conn) = endpoint.connect(peer, DATASTREAM_ALPN).await else {
return;
};
let Ok(send) = conn.open_uni().await else {
return;
};
let _ = write_subscription_until_closed(&engine_handle, send, header, subscription, idle_sleep).await;
});
}
pub async fn write_available_subscription(
engine: &EngineHandle,
send: SendStream,
header: &DatastreamQuicHeader,
subscription: &DatastreamSubscription,
) -> Result<DatastreamQuicWriteStats, BoxError> {
write_subscription_inner(send, header, subscription, None).await
write_subscription_inner(engine, send, header, subscription, None).await
}
pub async fn write_subscription_until_closed(
engine: &EngineHandle,
mut send: SendStream,
header: DatastreamQuicHeader,
subscription: DatastreamSubscription,
@ -118,7 +120,7 @@ pub async fn write_subscription_until_closed(
}
}
Err(TryRecvError::Empty) => {
tokio::time::sleep(idle_sleep).await;
engine.timer(idle_sleep).await;
}
Err(TryRecvError::Disconnected) => break,
}
@ -128,6 +130,7 @@ pub async fn write_subscription_until_closed(
}
async fn write_subscription_inner(
engine: &EngineHandle,
mut send: SendStream,
header: &DatastreamQuicHeader,
subscription: &DatastreamSubscription,
@ -145,7 +148,7 @@ async fn write_subscription_inner(
}
}
Err(TryRecvError::Empty) => match idle_sleep {
Some(delay) => tokio::time::sleep(delay).await,
Some(delay) => engine.timer(delay).await,
None => break,
},
Err(TryRecvError::Disconnected) => break,
@ -208,26 +211,26 @@ pub async fn read_next_uni_from_connection(
}
pub fn spawn_connection_reader(
handle: &Handle,
engine: &EngineHandle,
conn: Connection,
sink: std::sync::mpsc::Sender<DatastreamEvent>,
) -> tokio::task::JoinHandle<Result<(), String>> {
handle.spawn(async move {
) {
engine.spawn(async move {
loop {
let recv = match conn.accept_uni().await {
Ok(recv) => recv,
Err(error) => return Err(error.to_string()),
Err(_) => return,
};
let Ok(read) = read_events_from_stream(recv).await else {
continue;
};
let read = read_events_from_stream(recv)
.await
.map_err(|error| error.to_string())?;
for event in read.events {
if sink.send(event).is_err() {
return Ok(());
return;
}
}
}
})
});
}
async fn write_header(

35
crates/iroh-driver/src/edge_transport.rs
View file

@ -11,8 +11,8 @@ use distribution::types::NodeId;
use iroh::endpoint::Connection;
use iroh::{Endpoint, EndpointAddr};
use parking_lot::Mutex;
use swactor_engine::EngineHandle;
use tokio::io::AsyncWriteExt;
use tokio::runtime::Handle;
use tokio::sync::mpsc as tokio_mpsc;
pub const EDGE_ALPN: &[u8] = b"mvp/pipeline-edge/0";
@ -64,14 +64,15 @@ impl EdgeSendHandle {
}
pub(crate) fn spawn_edge_send_pump(
handle: Handle,
engine: EngineHandle,
endpoint: Endpoint,
peer: EndpointAddr,
edge_id: u64,
) -> Result<EdgeSendHandle, String> {
let (tx, mut rx) = tokio_mpsc::unbounded_channel::<Vec<u8>>();
let (ready_tx, ready_rx) = std::sync::mpsc::channel::<Result<(), String>>();
handle.spawn(async move {
let engine_handle = engine.clone();
engine.spawn(async move {
let result: Result<(), String> = async {
macro_rules! open_edge_stream {
() => {{
@ -99,7 +100,7 @@ pub(crate) fn spawn_edge_send_pump(
let mut attempts = 0_u8;
loop {
attempts = attempts.saturating_add(1);
let write_result = tokio::time::timeout(Duration::from_secs(30), async {
let write_result = engine_handle.timeout(Duration::from_secs(30), async {
send.write_all(&record)
.await
.map_err(|e| format!("write edge record {edge_id}: {e}"))?;
@ -112,27 +113,9 @@ pub(crate) fn spawn_edge_send_pump(
match write_result {
Ok(()) => break,
Err(error) if attempts < 3 => {
Err(_error) if attempts < 3 => {
send = open_edge_stream!();
let retry_result =
tokio::time::timeout(Duration::from_secs(30), async {
send.write_all(&record).await.map_err(|e| {
format!("write edge record {edge_id} after reconnect: {e}")
})?;
send.flush().await.map_err(|e| {
format!("flush edge record {edge_id} after reconnect: {e}")
})
})
.await
.map_err(|_| {
format!(
"write edge record {edge_id} after reconnect: timed out"
)
})?;
retry_result.map_err(|retry_error| {
format!("{error}; reconnect write failed: {retry_error}")
})?;
break;
continue;
}
Err(error) => return Err(error),
}
@ -154,13 +137,13 @@ pub(crate) fn spawn_edge_send_pump(
}
pub(crate) fn spawn_edge_recv_pump(
handle: Handle,
engine: EngineHandle,
conn: Connection,
peer: NodeId,
events: Arc<Mutex<Vec<EdgeTransportEvent>>>,
stream_group: u64,
) {
handle.spawn(async move {
engine.spawn(async move {
let mut next_uni_stream_id = stream_group << 32;
while let Ok(mut recv) = conn.accept_uni().await {
next_uni_stream_id = next_uni_stream_id.saturating_add(1);

785
crates/iroh-driver/src/iroh_driver.rs
View file

File diff suppressed because it is too large Load diff

5
crates/iroh-driver/src/lib.rs
View file

@ -4,6 +4,11 @@
//! distribution crate owns cluster dynamics, protocol actors, routing claims, and
//! wire message definitions.
// Engine boundary enforcement: disallowed scheduling/time/core-driving methods
// are hard errors in this crate (ENGINE_SPEC.md §2). All engine-hosted
// work goes through `EngineHandle`.
#![deny(clippy::disallowed_methods)]
pub mod datastream_transport;
pub mod driver_pumps;
pub mod edge_transport;

153
crates/iroh-driver/tests/common/iroh.rs
View file

@ -3,15 +3,16 @@
//! Each node is an [`IrohNode`]: a real iroh [`IrohDriver`] (endpoint with
//! `RelayMode::Disabled`) bridged to a per-node swactor [`Runtime`] hosting the
//! four protocol actors — `SwimActor`, `RegistryActor`, `MetadataActor`,
//! `DirectoryActor`. The driver decodes inbound frames into actor mailboxes,
//! the actors enqueue outbound frames on a shared [`Outbox`], and the driver
//! writes them to iroh.
//!
//! The synchronous `#[test]`s drive the stack by *pumping*: each iteration
//! injects the four `Tick`s, then `pump_inbound_to_actors()` / `rt.tick()` /
//! `drain_outbox()`. SWIM is wall-clock driven, so the `pump_until*` helpers
//! sleep ~10ms between iterations to let real time elapse.
//! `DirectoryActor`. Each node owns a swactor [`Engine`] that drives actor
//! progression and injects protocol ticks; iroh adapter progression is
//! engine-hosted. The driver decodes inbound frames into actor mailboxes, the
//! actors enqueue outbound frames on a shared [`Outbox`], and engine-hosted
//! writers send them to iroh.
//!
//! The synchronous `#[test]`s observe the stack through converge-or-timeout
//! polls: the `pump_until*` helpers sleep ~10ms between checks and re-read the
//! membership mirror. The engine drives core progression, protocol ticks, and
//! all iroh work in the background; the tests no longer pump any queue.
//! Membership is observed through the harness `membership_mirror` (a
//! `MemberList` filled by the [`MembershipFanout`] from SWIM's
//! `MembershipChanged` stream). The driver snapshot no longer carries members.
@ -19,12 +20,12 @@
use std::collections::HashMap;
use std::ops::{Index, IndexMut};
use std::sync::{Arc, Mutex as StdMutex, OnceLock, RwLock};
use std::sync::{Arc, Mutex as StdMutex, RwLock};
use std::time::{Duration, Instant};
use iroh::{EndpointAddr, PublicKey, RelayMode};
use parking_lot::Mutex;
use tokio::runtime::{Handle, Runtime as TokioRuntime};
use swactor_engine::{Engine, TokioBackend, TokioConfig};
use swactor::actor::{ActorAddress, ActorInterface};
use swactor::config::RuntimeConfig;
@ -48,19 +49,6 @@ use distribution::types::{MemberState, NodeId};
use super::test_config;
/// Process-wide multi-threaded tokio runtime backing the test drivers.
///
/// Production runs every driver on one ambient tokio runtime (the node owns a
/// single `#[tokio::main]` runtime). The drivers no longer own a runtime, so the
/// sync `#[test]`s supply one here and construct via [`IrohDriver::with_handle`].
/// The runtime is kept alive for the whole test process via `OnceLock`; the
/// `pump_*` helpers run the actor stack from the test's own (non-async) thread.
fn test_tokio_handle() -> Handle {
static RT: OnceLock<TokioRuntime> = OnceLock::new();
RT.get_or_init(|| TokioRuntime::new().expect("build test tokio runtime"))
.handle()
.clone()
}
// ── Membership fanout (copied verbatim from main.rs) ────────────────────────
// Adapts the SwimActor's `MembershipChanged` stream (its sole observable) into
@ -102,6 +90,9 @@ pub struct IrohNode {
membership_mirror: Arc<Mutex<MemberList>>,
relay_mirror: RelayMirror,
route_view: RouteView,
/// The engine that owns this node's Tokio substrate and drives the core
/// runtime. Declared last so it drops after the driver on teardown.
_engine: Engine,
}
impl IrohNode {
@ -110,17 +101,9 @@ impl IrohNode {
/// `RouteViewTransport` + `OutboxRouteBinder`; `MembershipFanout` + `Subscribe`;
/// the `routes` tag table; `enable_actor_bridge`).
fn from_config(config: IrohDriverConfig) -> Self {
let mut driver = IrohDriver::with_handle(test_tokio_handle(), config)
.expect("failed to create iroh driver");
let node_id = driver.node_id();
// The node's distribution config (SWIM/registry/metadata params).
let node_config = test_config();
let swim_config = node_config.swim.clone();
let registry_config = node_config.registry.clone();
let metadata_lambda = node_config.metadata_lambda;
// Per-node swactor runtime + codec + transport router.
// Per-node swactor runtime + codec + transport router. The runtime is
// created before the driver so the engine can own it; the driver needs
// the engine handle, and actors need the driver's node_id.
let mut swactor_rt =
Runtime::new(RuntimeConfig::default()).with_extension(Arc::new(StdExtension::new()));
let actor_codec = Arc::new(actor_codec_registry());
@ -131,6 +114,24 @@ impl IrohNode {
)));
let rt: Arc<Runtime> = Arc::new(swactor_rt);
// The engine owns the runtime (drives actor progression) and the Tokio
// substrate (schedules all iroh background work).
let engine = Engine::new(
Arc::clone(&rt),
TokioBackend::new(TokioConfig::default()).expect("build test tokio backend"),
)
.expect("build test engine");
let mut driver = IrohDriver::with_engine(engine.handle(), config)
.expect("failed to create iroh driver");
let node_id = driver.node_id();
// The node's distribution config (SWIM/registry/metadata params).
let node_config = test_config();
let swim_config = node_config.swim.clone();
let registry_config = node_config.registry.clone();
let metadata_lambda = node_config.metadata_lambda;
// Shared egress state.
let outbox: Outbox = Arc::new(StdMutex::new(Vec::new()));
let relay_mirror: RelayMirror = Arc::new(RwLock::new(HashMap::new()));
@ -221,7 +222,29 @@ impl IrohNode {
swim_addr,
Arc::clone(&relay_mirror),
Arc::clone(&route_view),
Arc::clone(&outbox),
);
// Engine-hosted adapter pump: drains ingress/egress/datastream/edge on
// a timer so the synchronous test loop no longer pumps these by hand.
driver.install_actor_bridge_pump(Duration::from_millis(10));
// Engine-hosted protocol tick injection + core progression. The pump
// helpers only drain iroh queues; the engine drives actor ticks and
// protocol injection (ENGINE_SPEC.md).
let ticker_handle = engine.handle();
let ticker_inner = ticker_handle.clone();
let ticker_rt = Arc::clone(&rt);
ticker_handle.spawn(async move {
let mut interval = ticker_inner.interval(Duration::from_millis(10));
loop {
(&mut interval).await;
let now = Instant::now();
let _ = ticker_rt.send_to(swim_addr, SwimIn::Tick { now });
let _ = ticker_rt.send_to(registry_addr, RegistryIn::Tick);
let _ = ticker_rt.send_to(metadata_addr, MetadataIn::Tick);
let _ = ticker_rt.send_to(directory_addr, DirectoryIn::Tick);
}
});
Self {
driver,
@ -234,22 +257,10 @@ impl IrohNode {
membership_mirror,
relay_mirror,
route_view,
_engine: engine,
}
}
/// One pump iteration for this node: inject the four `Tick`s, decode inbound
/// frames into mailboxes, advance the actors, then write outbound frames to
/// iroh. The production driver loop, condensed to one step.
fn pump(&mut self) {
let now = Instant::now();
let _ = self.rt.send_to(self.swim_addr, SwimIn::Tick { now });
let _ = self.rt.send_to(self.registry_addr, RegistryIn::Tick);
let _ = self.rt.send_to(self.metadata_addr, MetadataIn::Tick);
let _ = self.rt.send_to(self.directory_addr, DirectoryIn::Tick);
self.driver.pump_inbound_to_actors();
self.rt.tick();
self.driver.drain_outbox(&self.outbox);
}
// ── Passthroughs to the driver (keep consumer churn small) ──────────────
@ -326,19 +337,10 @@ pub fn make_driver_with_relay(relay_url: iroh::RelayUrl) -> IrohNode {
})
}
/// Pump one node (one full actor-stack step).
pub fn pump_one(node: &mut IrohNode) {
node.pump();
}
/// Pump a slice of nodes.
pub fn pump_all(nodes: &mut [IrohNode]) {
for n in nodes.iter_mut() {
n.pump();
}
}
/// Pump two nodes until a condition is met or timeout expires.
/// Poll until `check_fn` holds over `a` and `b` or `timeout` elapses, sleeping
/// ~10ms between checks. Progression is engine-hosted; this only waits for
/// wall-clock SWIM convergence.
pub fn pump_until_pair(
a: &mut IrohNode,
b: &mut IrohNode,
@ -347,8 +349,6 @@ pub fn pump_until_pair(
) -> bool {
let start = Instant::now();
while start.elapsed() < timeout {
a.pump();
b.pump();
if check_fn(a, b) {
return true;
}
@ -357,14 +357,15 @@ pub fn pump_until_pair(
false
}
/// Pump N nodes until a condition is met or timeout expires.
/// Poll until `check_fn` holds over `nodes` or `timeout` elapses, sleeping
/// ~10ms between checks. Progression is engine-hosted; this only waits for
/// wall-clock SWIM convergence.
pub fn pump_until<F>(nodes: &mut [IrohNode], timeout: Duration, check_fn: F) -> bool
where
F: Fn(&[IrohNode]) -> bool,
{
let start = Instant::now();
while start.elapsed() < timeout {
pump_all(nodes);
if check_fn(nodes) {
return true;
}
@ -483,34 +484,16 @@ impl IrohTestCluster {
self.nodes[idx].key()
}
/// Pump all nodes until a condition is met or timeout expires.
/// Poll until `check_fn` holds over the full node slice or `timeout`
/// elapses, sleeping ~10ms between checks. Actor progression and iroh
/// adapter work are engine-hosted, so this loop only waits for wall-clock
/// SWIM convergence — it no longer drives nodes.
pub fn pump_until<F>(&mut self, timeout: Duration, check_fn: F) -> bool
where
F: Fn(&[IrohNode]) -> bool,
{
self.pump_until_excluding(&[], timeout, check_fn)
}
/// Pump every node EXCEPT those whose index is in `excluded` (a killed node
/// must not be driven), until `check_fn` holds over the full node slice or
/// `timeout` elapses. This is the converge-or-timeout poll for real death.
pub fn pump_until_excluding<F>(
&mut self,
excluded: &[usize],
timeout: Duration,
check_fn: F,
) -> bool
where
F: Fn(&[IrohNode]) -> bool,
{
let start = Instant::now();
while start.elapsed() < timeout {
for (i, n) in self.nodes.iter_mut().enumerate() {
if excluded.contains(&i) {
continue;
}
n.pump();
}
if check_fn(&self.nodes) {
return true;
}

168
crates/iroh-driver/tests/datastream_transport.rs
View file

@ -1,4 +1,5 @@
use std::net::{IpAddr, SocketAddr};
use std::sync::Arc;
use datastream::{
ChannelContent, DatastreamEndpoint, DatastreamEvent, Lifetime, NodeId, Position, StreamId,
@ -8,80 +9,109 @@ use iroh_driver::{
DATASTREAM_ALPN, DatastreamQuicHeader, read_next_uni_from_connection,
write_available_subscription,
};
use swactor::config::RuntimeConfig;
use swactor::runtime::Runtime;
use swactor_engine::{Engine, TokioBackend, TokioConfig};
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn iroh_datastream_alpn_carries_catalog_and_numeric_frames() {
let source = test_endpoint().await;
let collector = test_endpoint().await;
let collector_addr = endpoint_addr(&collector);
let collector_accept = {
let collector = collector.clone();
tokio::spawn(async move {
collector
.accept()
.await
.expect("incoming connection")
.await
.expect("accepted connection")
})
};
/// Datastream transport test scheduled through `EngineHandle`, not an ambient
/// `#[tokio::test]` runtime (ENGINE_SPEC.md).
#[test]
fn iroh_datastream_alpn_carries_catalog_and_numeric_frames() {
let runtime = Runtime::new(RuntimeConfig::default());
let engine = Engine::new(
Arc::new(runtime),
TokioBackend::new(TokioConfig::default()).expect("test backend"),
)
.expect("test engine");
let handle = engine.handle();
let stream = StreamId::new(NodeId::new("source-node"), Lifetime(1));
let endpoint = DatastreamEndpoint::with_capacity(stream.clone(), 8, 8);
let producer = endpoint.producer();
let runtime_log = producer.register_channel("runtime.log", ChannelContent::TextStream);
let subscription = endpoint.subscribe_all("iroh");
let (done_tx, done_rx) = std::sync::mpsc::channel::<Result<(), String>>();
let h = handle.clone();
handle.spawn(async move {
let source = test_endpoint().await;
let collector = test_endpoint().await;
let collector_addr = endpoint_addr(&collector);
producer.submit_text(runtime_log, "alpha");
producer.submit_text(runtime_log, "beta");
endpoint.tick();
let conn = source
.connect(collector_addr, DATASTREAM_ALPN)
.await
.expect("connect datastream ALPN");
let send = conn.open_uni().await.expect("open uni stream");
let header =
DatastreamQuicHeader::from_snapshot([7; 16], b"token".to_vec(), subscription.snapshot())
.expect("header from subscription snapshot");
let wrote = write_available_subscription(send, &header, &subscription)
.await
.expect("write subscription");
assert_eq!(wrote.events, 2);
let accepted = collector_accept.await.expect("collector accept task");
let read = read_next_uni_from_connection(&accepted)
.await
.expect("read datastream uni stream");
assert_eq!(read.header, header);
assert_eq!(read.header.stream.stream, stream);
assert!(
read.header
.channels
.iter()
.any(|descriptor| descriptor.id == runtime_log && descriptor.name == "runtime.log")
);
assert_eq!(read.events.len(), 2);
match &read.events[0] {
DatastreamEvent::Frame(frame) => {
assert_eq!(frame.channel.stream, stream);
assert_eq!(frame.channel.channel, runtime_log);
assert_eq!(frame.position, Position(0));
assert_eq!(frame.payload, b"alpha");
// Accept the incoming connection through an engine-hosted task + oneshot,
// since EngineHandle::spawn is fire-and-forget (no JoinHandle).
let (accept_tx, accept_rx) = tokio::sync::oneshot::channel();
{
let collector = collector.clone();
h.spawn(async move {
let conn = collector
.accept()
.await
.expect("incoming connection")
.await
.expect("accepted connection");
let _ = accept_tx.send(conn);
});
}
other => panic!("expected frame event, got {other:?}"),
}
match &read.events[1] {
DatastreamEvent::Frame(frame) => {
assert_eq!(frame.position, Position(1));
assert_eq!(frame.payload, b"beta");
}
other => panic!("expected frame event, got {other:?}"),
}
source.close().await;
collector.close().await;
let stream = StreamId::new(NodeId::new("source-node"), Lifetime(1));
let endpoint = DatastreamEndpoint::with_capacity(stream.clone(), 8, 8);
let producer = endpoint.producer();
let runtime_log = producer.register_channel("runtime.log", ChannelContent::TextStream);
let subscription = endpoint.subscribe_all("iroh");
producer.submit_text(runtime_log, "alpha");
producer.submit_text(runtime_log, "beta");
endpoint.tick();
let conn = source
.connect(collector_addr, DATASTREAM_ALPN)
.await
.expect("connect datastream ALPN");
let send = conn.open_uni().await.expect("open uni stream");
let header =
DatastreamQuicHeader::from_snapshot([7; 16], b"token".to_vec(), subscription.snapshot())
.expect("header from subscription snapshot");
let wrote = write_available_subscription(&h, send, &header, &subscription)
.await
.expect("write subscription");
assert_eq!(wrote.events, 2);
let accepted = accept_rx.await.expect("collector accept task");
let read = read_next_uni_from_connection(&accepted)
.await
.expect("read datastream uni stream");
assert_eq!(read.header, header);
assert_eq!(read.header.stream.stream, stream);
assert!(
read.header
.channels
.iter()
.any(|descriptor| descriptor.id == runtime_log && descriptor.name == "runtime.log")
);
assert_eq!(read.events.len(), 2);
match &read.events[0] {
DatastreamEvent::Frame(frame) => {
assert_eq!(frame.channel.stream, stream);
assert_eq!(frame.channel.channel, runtime_log);
assert_eq!(frame.position, Position(0));
assert_eq!(frame.payload, b"alpha");
}
other => panic!("expected frame event, got {other:?}"),
}
match &read.events[1] {
DatastreamEvent::Frame(frame) => {
assert_eq!(frame.position, Position(1));
assert_eq!(frame.payload, b"beta");
}
other => panic!("expected frame event, got {other:?}"),
}
source.close().await;
collector.close().await;
let _ = done_tx.send(Ok(()));
});
match done_rx.recv() {
Ok(Ok(())) => {}
Ok(Err(e)) => panic!("test failed: {e}"),
Err(_) => panic!("test task dropped"),
}
}
async fn test_endpoint() -> Endpoint {

39
crates/iroh-driver/tests/iroh_driver.rs
View file

@ -54,7 +54,6 @@ fn endpoint_addr_includes_home_relay() {
if start.elapsed() >= Duration::from_secs(5) {
break (false, current_relay_url);
}
pump_one(&mut node);
std::thread::sleep(Duration::from_millis(10));
};
@ -208,9 +207,9 @@ fn goal2_shutdown_node_is_detected_dead_by_survivors() {
let dead_key = cluster.key(dead);
cluster.shutdown_one(dead);
// The survivors' probes to the dead node now truly fail; drive them until
// both converge on it being Dead.
let detected = cluster.pump_until_excluding(&[dead], Duration::from_secs(30), |drivers| {
// The survivors' probes to the dead node now truly fail; poll until both
// converge on it being Dead.
let detected = cluster.pump_until(Duration::from_secs(30), |drivers| {
drivers
.iter()
.enumerate()
@ -222,3 +221,35 @@ fn goal2_shutdown_node_is_detected_dead_by_survivors() {
cluster[0].shutdown();
cluster[1].shutdown();
}
// ─── Capability binding (ENGINE_SPEC.md) ──────────────────────
#[test]
fn driver_rejects_engine_without_io() {
// The SteppingBackend advertises tasks + timers + blocking but NOT io.
// The driver requires tasks + timers + io, so construction must fail
// before any endpoint is bound or background work starts.
use distribution::node::DistributedNodeConfig;
use iroh::RelayMode;
use iroh_driver::{IrohDriver, IrohDriverConfig};
use swactor::config::RuntimeConfig;
use swactor::runtime::Runtime;
use swactor_engine::{Engine, SteppingBackend};
let rt = Arc::new(Runtime::new(RuntimeConfig::default()));
let engine = Engine::new(rt, SteppingBackend::default()).expect("stepping engine");
let result = IrohDriver::with_engine(
engine.handle(),
IrohDriverConfig {
secret_key: None,
relay_mode: RelayMode::Disabled,
node: DistributedNodeConfig::default(),
peer_auth: None,
additional_alpns: vec![],
},
);
assert!(
result.is_err(),
"driver must reject an engine that lacks the io capability"
);
}

135
docs/specs/drafts/ENGINE_SPEC.md
View file

@ -1,135 +0,0 @@
# swactor engine — specification
Id: 1
Last modified:
Last reviewed:
**Scope:** the execution substrate that drives swactor workers and hosts their async side-work, defined as an interface implemented per environment.
## 1. Purpose
swactor actors are synchronous, single-writer message handlers. Real systems need work actors cannot do inline: draining byte streams, running retry backoffs, polling on an interval, blocking GPU calls. That work lives in *tasks* on an execution substrate. Today that substrate is Tokio — hardcoded and reinvented per crate (ambient `Handle::try_current()`, silently-owned runtimes, ad-hoc `block_on` sync facades, a mix of tokio tasks and std threads).
This spec defines the **engine**: a single execution substrate, expressed as an interface, that (a) drives swactor workers and (b) runs the async tasks that back them. Tokio is one implementation; a minimal std-thread engine, a Go engine, a JS-worker engine, and a deterministic test engine are others. Authoring the interface from swactor's needs lets core and each engine implementation be optimized independently on either side of the seam.
## 2. Scope
**In scope**
- The engine interface: what swactor requires of an engine, and what an engine provides.
- The responsibility split between actor-workers and the engine.
- How the engine drives workers (hosting the worker loop, the inbox as the wait seam).
- The capability surface: tasks, timers, async I/O, blocking, time.
- The bridge contract: how a task delivers into an actor mailbox, and the sync/async boundary rules.
- The invariants an engine must uphold.
- Reference instantiations (non-normative).
**Out of scope**
- Actor execution semantics — single-writer, per-(sender,target) FIFO, fairness, panic isolation. Those belong to the actor-worker / core.
- Backpressure policy. Producers and consumers share one engine; pressure handling is the application's decision, not swactor's.
- Cancellation and shutdown lifecycle (deferred; nice-to-have).
- Failure / observability propagation, except where it falls out of the bridge contract.
- Cross-process / cross-isolation delivery and serialization.
- Specific protocols and codecs (iroh/QUIC, datastream framing). Those are crate logic built *on* the engine.
## 3. Model
- An **actor-worker** owns a disjoint set of actors, processes them one at a time, and is the unit of actor execution. It holds the pool, mailboxes, routing, and a single synchronous entry point: run one **pass** (`tick_once`), which drains its inbox into mailboxes and processes non-empty mailboxes up to a fairness budget.
- The **engine** is the execution substrate. It does exactly two things:
1. **Drives workers** — hosts each worker's loop: wait until the worker has work, run a pass, repeat.
2. **Runs tasks** — schedules the async side-work (timers, I/O pumps, blocking calls) that backs the actors.
- **The engine owns all progression.** Actor handlers never `.await`. Every handler is a synchronous transition that returns control immediately. The engine runs the loop that drives them and holds every long-lived flow (a worker idle on its inbox, a task doing I/O, a timer). The actor world is pure transition. Only the engine carries control flow across time.
- Workers and tasks share one substrate and one scheduler. There is no separate "I/O runtime" beside the actor runtime.
## 4. The engine interface
The interface is authored from swactor's needs. It is a **contract** — operations plus their semantics and invariants. A Rust trait is its canonical Rust binding; Go, JS, and other hosts implement the same contract natively. This spec defines the contract, not the Rust signature.
**The engine provides:**
// USER: The `host_worker(id, pass)` fn needs more explanation and justification
// USER: Why are we including a timer as a core function necessary to the engine. Can it not go somewhere else?
| operation | meaning |
|---|---|
| `host_worker(id, pass) → deposit` | Create the worker's inbox, start its reactive loop (wait on the inbox, call `pass`), and return the **deposit** handle core uses to route messages into it. |
| `spawn(task)` | Schedule an async unit of work on the substrate. |
| `spawn_blocking(work)` | Schedule blocking CPU / syscall work off the async path. |
| `timer(delay)` / `interval(period)` | Schedule future or recurring work. |
| `now()` | The engine's monotonic clock. |
**Core provides back to the engine and to tasks:**
// USER: Core is fine as it is. We are not modifying core, it was carefully designed and is very pure. The engine is to be abstracted in such a way as to complement the abstractions core gives us. I think we can satisfy these fns through existing core, but we don't say that core provides xyz, as that is not the framing of this spec.
| surface | meaning |
|---|---|
| `pass` (per worker) | The synchronous entry point `tick_once(&tc) → did_work`, run once per pass. |
| `deliver` | A handle to deposit a message into an actor mailbox by address — the bridge (§7). Cloneable; captured by tasks. |
The split is deliberate. Core owns actor logic, routing, and the *deposit* side of every inbox. The engine owns the *idle* side and all scheduling. **Core only transitions. The engine drives.** The deposit handle returned by `host_worker` is engine-agnostic (loss-free, non-blocking push) so core's routing can deposit without knowing which engine is in use.
## 5. Driving workers
// USER: `pass` is stupid when we already have a `tick()` built in.
- The engine hosts N workers. For each, it runs: call `pass`; if it did work, call it again (a productive pass may have buffered same-worker sends that need draining); if it did no work, idle on the inbox until a deposit makes a pass runnable. There is no separate wake primitive. A deposit into the inbox is what makes the next transition runnable, so the engine drives it.
- The **inbox is the wait seam.** The engine creates each inbox and holds its consumer side, choosing how to wait (a blocking recv under std threads; an async `recv().await` under tokio; an event under JS). Core holds the deposit side for routing.
- **Non-reentrancy.** The engine must never run two passes of the same worker concurrently. A worker's `&mut self` is live only for the duration of a synchronous `pass` call — never held across a wait.
- **Scheduling strategy is the engine's choice.** Whether a pass runs inline on the executor (cooperative) or on a blocking thread is an implementation tradeoff the engine owns; core is agnostic to it.
## 6. Capability surface
// USER: Maybe just I/O instead of explicitly async? So we can have a blocking I/O if our engine only supports that
// USER: Not sure I want to put time inside the engine. I am open to being convinced, but the added complexity and tying it
// USER: to what I wanted to be a simple task/execution api is worrying me about future compatability.
The primitives an engine may provide. Capabilities are **per-implementation and discoverable**: each engine reports which it supports, and binding an engine that lacks a required capability fails at construction, never at runtime.
- **Tasks** — `spawn` of an async unit of work; the substrate's unit of concurrency.
- **Timers** — one-shot delay and recurring interval.
- **Async I/O** — streams, sockets, files. This is where implementations diverge most: a tokio engine offers sockets / QUIC / streams; a JS engine offers fetch / WebSocket; a std-thread engine offers none (only blocking I/O via `spawn_blocking`).
- **Blocking** — `spawn_blocking` for CPU-bound or syscall work that must not stall the executor.
- **Time** — `now()`. In a test engine this is virtual, advanced by the test; this is what makes deterministic testing possible.
An engine that provides only tasks + blocking + time is still a valid (if unperformant) engine. Crates that need async I/O bind to an engine that provides it.
## 7. The bridge contract
// USER: Why this contract, why are tasks delivering directly to actors?
How an engine task gets a result into an actor mailbox.
- A task captures a **deliver** handle (obtained from core, not from the engine) bound to a destination address, or a runtime-wide `send_to(addr, msg)`. Delivering deposits the message into the owning worker's inbox — a loss-free, non-blocking pointer-move along the same path any sender uses. No serialization, no copy, within one address space.
- Deliver is **fire-and-forget from the task's view**: it returns immediately; the actor handles the message on a later pass of its worker.
- **Boundary rules:**
- Actor handlers are synchronous and single-writer. They never `.await`.
- `&mut Worker` and any actor state is live only during a synchronous `pass`; it is never held across a wait and never sent into a task.
- All `.await` lives in tasks. Tasks never touch actor state directly; they communicate only via the deliver handle and the inbox.
- The inbox a task delivers into is the same FIFO, loss-free, unbounded queue the worker waits on. Mailbox ordering semantics (per-(sender,target) FIFO) are the actor-worker's concern; the engine's only obligation is that the inbox itself is FIFO and loss-free.
## 8. Invariants
An engine must uphold:
- **Non-reentrant passes.** At most one `pass` per worker at any instant.
- **Loss-free, non-blocking delivery.** The inbox never drops and never blocks the sender (unbounded).
- **FIFO inbox.** Messages depart an inbox in deposit order.
- **Progress independence.** A long-running or blocked task must not stall worker passes, and vice versa. The engine provides enough concurrency that workers and tasks progress independently (on a cooperative single-thread host like JS, this is a discipline the engine enforces: no blocking calls in tasks or passes).
- **Actors never await.** No `.await` reaches actor code; the engine owns every wait.
## 9. Reference instantiations (non-normative)
Illustrations of how each environment satisfies the contract — not prescription.
- **tokio.** Workers and tasks are tokio tasks; a worker loop is `loop { inbox.recv().await; pass(); }` with `&mut Worker` live only across the synchronous `pass` (long passes may be moved to `spawn_blocking`; that scheduling choice is the engine's, per §5). Async I/O, `spawn_blocking`, and `now()` are tokio's. This is today's de-facto engine, made explicit.
- **std-thread.** Each worker is an OS thread blocking on its inbox; tasks are OS threads or a small pool; `spawn_blocking` is a thread; there is no async I/O, only blocking I/O. Simple, unperformant, dependency-free — and a valid engine.
- **deterministic test engine.** A single-threaded stepping scheduler: workers and tasks are entries the test advances manually; `now()` is virtual time advanced by the test; async I/O is faked or mocked. It implements the same contract, so crates test against the interface with no real network and no threads, fully deterministic. It falls out of the contract; it is not specified separately.
- **Go / JS-worker (illustrative).** Workers and tasks map to goroutines + channels, or to the JS event loop + `postMessage` / callbacks. Each provides the capability subset its runtime supports.
## 10. What this spec does not define
The boundary, stated plainly:
- Actor execution semantics (single-writer, FIFO, fairness, panic isolation).
- Backpressure.
- Cancellation and shutdown.
- Failure / observability propagation beyond the bridge.
- Cross-process / cross-isolation delivery and serialization.
- Specific protocols and codecs.