feat(runtime): add resident Wasmtime actor runtime
- define versioned service and clock WIT contracts - enforce import allowlists, fuel, epoch, memory, I/O, and mailbox limits - keep one Store and Component Instance resident per serial Actor - add echo, counter, fault, spin, and capability probe components - cover lifecycle, concurrency, sandbox, and fault recovery behavior
This commit is contained in:
Generated
+1280
-17
File diff suppressed because it is too large
Load Diff
+11
-1
@@ -1,8 +1,15 @@
|
|||||||
[workspace]
|
[workspace]
|
||||||
members = [
|
members = [
|
||||||
"crates/wasmeld-package",
|
"crates/wasmeld-package",
|
||||||
|
"crates/wasmeld-runtime",
|
||||||
|
"components/echo",
|
||||||
|
"components/counter",
|
||||||
|
"components/fault",
|
||||||
|
"components/spin",
|
||||||
|
"components/clock-probe",
|
||||||
|
"components/wasi-clock-probe",
|
||||||
]
|
]
|
||||||
default-members = ["crates/wasmeld-package"]
|
default-members = ["crates/wasmeld-runtime"]
|
||||||
resolver = "2"
|
resolver = "2"
|
||||||
|
|
||||||
[workspace.package]
|
[workspace.package]
|
||||||
@@ -18,6 +25,9 @@ semver = "1.0.28"
|
|||||||
sha2 = "0.10.9"
|
sha2 = "0.10.9"
|
||||||
thiserror = "2.0.17"
|
thiserror = "2.0.17"
|
||||||
toml = "0.9.8"
|
toml = "0.9.8"
|
||||||
|
wasmtime = { version = "=41.0.0", default-features = false, features = ["component-model", "cranelift", "runtime", "std"] }
|
||||||
|
wasmtime-wasi = { version = "=41.0.0", default-features = false, features = ["p2"] }
|
||||||
|
wit-bindgen = "=0.41.0"
|
||||||
wit-component = "=0.243.0"
|
wit-component = "=0.243.0"
|
||||||
wit-parser = "0.243.0"
|
wit-parser = "0.243.0"
|
||||||
zip = { version = "2.4.2", default-features = false, features = ["deflate"] }
|
zip = { version = "2.4.2", default-features = false, features = ["deflate"] }
|
||||||
|
|||||||
@@ -0,0 +1,16 @@
|
|||||||
|
[package]
|
||||||
|
name = "clock-probe-component"
|
||||||
|
version = "0.1.0"
|
||||||
|
edition.workspace = true
|
||||||
|
rust-version.workspace = true
|
||||||
|
license.workspace = true
|
||||||
|
|
||||||
|
[package.metadata.wasmeld]
|
||||||
|
id = "clock-probe"
|
||||||
|
world = "component:clock-probe/clock-probe-component@0.1.0"
|
||||||
|
|
||||||
|
[lib]
|
||||||
|
crate-type = ["cdylib"]
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
wit-bindgen.workspace = true
|
||||||
@@ -0,0 +1,24 @@
|
|||||||
|
mod bindings {
|
||||||
|
wit_bindgen::generate!({
|
||||||
|
path: "wit",
|
||||||
|
world: "clock-probe-component",
|
||||||
|
generate_all,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
struct ClockProbe;
|
||||||
|
|
||||||
|
impl bindings::Guest for ClockProbe {
|
||||||
|
fn init(_config: Vec<u8>) -> Result<(), bindings::ServiceError> {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn invoke(input: Vec<u8>) -> Result<Vec<u8>, bindings::ServiceError> {
|
||||||
|
let now = bindings::wasmeld::clock::monotonic_clock::now();
|
||||||
|
let mut output = now.to_le_bytes().to_vec();
|
||||||
|
output.extend_from_slice(&input);
|
||||||
|
Ok(output)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
bindings::export!(ClockProbe with_types_in bindings);
|
||||||
@@ -0,0 +1,11 @@
|
|||||||
|
schema_version = 1
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
"wasmeld:clock" = "0.1.0"
|
||||||
|
"wasmeld:service" = "0.1.0"
|
||||||
|
|
||||||
|
[replace."wasmeld:clock"]
|
||||||
|
path = "../../wit/clock"
|
||||||
|
|
||||||
|
[replace."wasmeld:service"]
|
||||||
|
path = "../../wit/service"
|
||||||
@@ -0,0 +1,15 @@
|
|||||||
|
schema_version = 1
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "wasmeld:clock"
|
||||||
|
version = "0.1.0"
|
||||||
|
source = "path+../../wit/clock"
|
||||||
|
sha256 = "93079c69d1b9cb3afc28da6fe4a7e37f724d0a09e780c847f8524d3920a90960"
|
||||||
|
replaced = true
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "wasmeld:service"
|
||||||
|
version = "0.1.0"
|
||||||
|
source = "path+../../wit/service"
|
||||||
|
sha256 = "d5497307bbcd1e159f7707f385b488a6f2e26362d5256d5bc1080ac603a51305"
|
||||||
|
replaced = true
|
||||||
@@ -0,0 +1,6 @@
|
|||||||
|
package component:clock-probe@0.1.0;
|
||||||
|
|
||||||
|
world clock-probe-component {
|
||||||
|
include wasmeld:service/service-component@0.1.0;
|
||||||
|
import wasmeld:clock/monotonic-clock@0.1.0;
|
||||||
|
}
|
||||||
@@ -0,0 +1,16 @@
|
|||||||
|
[package]
|
||||||
|
name = "counter-component"
|
||||||
|
version = "0.1.0"
|
||||||
|
edition.workspace = true
|
||||||
|
rust-version.workspace = true
|
||||||
|
license.workspace = true
|
||||||
|
|
||||||
|
[package.metadata.wasmeld]
|
||||||
|
id = "counter"
|
||||||
|
world = "component:counter/counter-component@0.1.0"
|
||||||
|
|
||||||
|
[lib]
|
||||||
|
crate-type = ["cdylib"]
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
wit-bindgen.workspace = true
|
||||||
@@ -0,0 +1,28 @@
|
|||||||
|
use core::sync::atomic::{AtomicU64, Ordering};
|
||||||
|
|
||||||
|
mod bindings {
|
||||||
|
wit_bindgen::generate!({
|
||||||
|
path: "wit",
|
||||||
|
world: "counter-component",
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
static COUNTER: AtomicU64 = AtomicU64::new(0);
|
||||||
|
|
||||||
|
struct Counter;
|
||||||
|
|
||||||
|
impl bindings::Guest for Counter {
|
||||||
|
fn init(_config: Vec<u8>) -> Result<(), bindings::ServiceError> {
|
||||||
|
// Calls are serialized by the host Actor, so this component intentionally
|
||||||
|
// uses one mutable global to make resident memory observable in tests.
|
||||||
|
COUNTER.store(0, Ordering::Relaxed);
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn invoke(_input: Vec<u8>) -> Result<Vec<u8>, bindings::ServiceError> {
|
||||||
|
let next = COUNTER.fetch_add(1, Ordering::Relaxed).wrapping_add(1);
|
||||||
|
Ok(next.to_le_bytes().to_vec())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
bindings::export!(Counter with_types_in bindings);
|
||||||
@@ -0,0 +1,7 @@
|
|||||||
|
schema_version = 1
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
"wasmeld:service" = "0.1.0"
|
||||||
|
|
||||||
|
[replace."wasmeld:service"]
|
||||||
|
path = "../../wit/service"
|
||||||
@@ -0,0 +1,8 @@
|
|||||||
|
schema_version = 1
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "wasmeld:service"
|
||||||
|
version = "0.1.0"
|
||||||
|
source = "path+../../wit/service"
|
||||||
|
sha256 = "d5497307bbcd1e159f7707f385b488a6f2e26362d5256d5bc1080ac603a51305"
|
||||||
|
replaced = true
|
||||||
@@ -0,0 +1,5 @@
|
|||||||
|
package component:counter@0.1.0;
|
||||||
|
|
||||||
|
world counter-component {
|
||||||
|
include wasmeld:service/service-component@0.1.0;
|
||||||
|
}
|
||||||
@@ -0,0 +1,16 @@
|
|||||||
|
[package]
|
||||||
|
name = "echo-component"
|
||||||
|
version = "0.1.0"
|
||||||
|
edition.workspace = true
|
||||||
|
rust-version.workspace = true
|
||||||
|
license.workspace = true
|
||||||
|
|
||||||
|
[package.metadata.wasmeld]
|
||||||
|
id = "echo"
|
||||||
|
world = "component:echo/echo-component@0.1.0"
|
||||||
|
|
||||||
|
[lib]
|
||||||
|
crate-type = ["cdylib"]
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
wit-bindgen.workspace = true
|
||||||
@@ -0,0 +1,20 @@
|
|||||||
|
mod bindings {
|
||||||
|
wit_bindgen::generate!({
|
||||||
|
path: "wit",
|
||||||
|
world: "echo-component",
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
struct Echo;
|
||||||
|
|
||||||
|
impl bindings::Guest for Echo {
|
||||||
|
fn init(_config: Vec<u8>) -> Result<(), bindings::ServiceError> {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn invoke(input: Vec<u8>) -> Result<Vec<u8>, bindings::ServiceError> {
|
||||||
|
Ok(input)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
bindings::export!(Echo with_types_in bindings);
|
||||||
@@ -0,0 +1,7 @@
|
|||||||
|
schema_version = 1
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
"wasmeld:service" = "0.1.0"
|
||||||
|
|
||||||
|
[replace."wasmeld:service"]
|
||||||
|
path = "../../wit/service"
|
||||||
@@ -0,0 +1,8 @@
|
|||||||
|
schema_version = 1
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "wasmeld:service"
|
||||||
|
version = "0.1.0"
|
||||||
|
source = "path+../../wit/service"
|
||||||
|
sha256 = "d5497307bbcd1e159f7707f385b488a6f2e26362d5256d5bc1080ac603a51305"
|
||||||
|
replaced = true
|
||||||
@@ -0,0 +1,5 @@
|
|||||||
|
package component:echo@0.1.0;
|
||||||
|
|
||||||
|
world echo-component {
|
||||||
|
include wasmeld:service/service-component@0.1.0;
|
||||||
|
}
|
||||||
@@ -0,0 +1,16 @@
|
|||||||
|
[package]
|
||||||
|
name = "fault-component"
|
||||||
|
version = "0.1.0"
|
||||||
|
edition.workspace = true
|
||||||
|
rust-version.workspace = true
|
||||||
|
license.workspace = true
|
||||||
|
|
||||||
|
[package.metadata.wasmeld]
|
||||||
|
id = "fault"
|
||||||
|
world = "component:fault/fault-component@0.1.0"
|
||||||
|
|
||||||
|
[lib]
|
||||||
|
crate-type = ["cdylib"]
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
wit-bindgen.workspace = true
|
||||||
@@ -0,0 +1,21 @@
|
|||||||
|
mod bindings {
|
||||||
|
wit_bindgen::generate!({
|
||||||
|
path: "wit",
|
||||||
|
world: "fault-component",
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
struct Fault;
|
||||||
|
|
||||||
|
impl bindings::Guest for Fault {
|
||||||
|
fn init(_config: Vec<u8>) -> Result<(), bindings::ServiceError> {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn invoke(input: Vec<u8>) -> Result<Vec<u8>, bindings::ServiceError> {
|
||||||
|
assert_ne!(input, b"fault", "intentional test fault");
|
||||||
|
Ok(input)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
bindings::export!(Fault with_types_in bindings);
|
||||||
@@ -0,0 +1,7 @@
|
|||||||
|
schema_version = 1
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
"wasmeld:service" = "0.1.0"
|
||||||
|
|
||||||
|
[replace."wasmeld:service"]
|
||||||
|
path = "../../wit/service"
|
||||||
@@ -0,0 +1,8 @@
|
|||||||
|
schema_version = 1
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "wasmeld:service"
|
||||||
|
version = "0.1.0"
|
||||||
|
source = "path+../../wit/service"
|
||||||
|
sha256 = "d5497307bbcd1e159f7707f385b488a6f2e26362d5256d5bc1080ac603a51305"
|
||||||
|
replaced = true
|
||||||
@@ -0,0 +1,5 @@
|
|||||||
|
package component:fault@0.1.0;
|
||||||
|
|
||||||
|
world fault-component {
|
||||||
|
include wasmeld:service/service-component@0.1.0;
|
||||||
|
}
|
||||||
@@ -0,0 +1,16 @@
|
|||||||
|
[package]
|
||||||
|
name = "spin-component"
|
||||||
|
version = "0.1.0"
|
||||||
|
edition.workspace = true
|
||||||
|
rust-version.workspace = true
|
||||||
|
license.workspace = true
|
||||||
|
|
||||||
|
[package.metadata.wasmeld]
|
||||||
|
id = "spin"
|
||||||
|
world = "component:spin/spin-component@0.1.0"
|
||||||
|
|
||||||
|
[lib]
|
||||||
|
crate-type = ["cdylib"]
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
wit-bindgen.workspace = true
|
||||||
@@ -0,0 +1,43 @@
|
|||||||
|
mod bindings {
|
||||||
|
wit_bindgen::generate!({
|
||||||
|
path: "wit",
|
||||||
|
world: "spin-component",
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
static mut SPIN_STATE: u64 = 0;
|
||||||
|
|
||||||
|
struct Spin;
|
||||||
|
|
||||||
|
impl bindings::Guest for Spin {
|
||||||
|
fn init(config: Vec<u8>) -> Result<(), bindings::ServiceError> {
|
||||||
|
if config == b"spin" {
|
||||||
|
burn(u64::MAX);
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn invoke(input: Vec<u8>) -> Result<Vec<u8>, bindings::ServiceError> {
|
||||||
|
let steps = input
|
||||||
|
.get(..size_of::<u64>())
|
||||||
|
.and_then(|bytes| bytes.try_into().ok())
|
||||||
|
.map(u64::from_le_bytes)
|
||||||
|
.unwrap_or_default();
|
||||||
|
|
||||||
|
burn(steps);
|
||||||
|
Ok(Vec::new())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The volatile write makes every loop iteration observable to the optimizer.
|
||||||
|
fn burn(steps: u64) {
|
||||||
|
let mut state = unsafe { core::ptr::read_volatile(&raw const SPIN_STATE) };
|
||||||
|
|
||||||
|
for _ in 0..steps {
|
||||||
|
state = state.rotate_left(7).wrapping_add(0x9e37_79b9_7f4a_7c15);
|
||||||
|
unsafe { core::ptr::write_volatile(&raw mut SPIN_STATE, state) };
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
bindings::export!(Spin with_types_in bindings);
|
||||||
@@ -0,0 +1,7 @@
|
|||||||
|
schema_version = 1
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
"wasmeld:service" = "0.1.0"
|
||||||
|
|
||||||
|
[replace."wasmeld:service"]
|
||||||
|
path = "../../wit/service"
|
||||||
@@ -0,0 +1,8 @@
|
|||||||
|
schema_version = 1
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "wasmeld:service"
|
||||||
|
version = "0.1.0"
|
||||||
|
source = "path+../../wit/service"
|
||||||
|
sha256 = "d5497307bbcd1e159f7707f385b488a6f2e26362d5256d5bc1080ac603a51305"
|
||||||
|
replaced = true
|
||||||
@@ -0,0 +1,5 @@
|
|||||||
|
package component:spin@0.1.0;
|
||||||
|
|
||||||
|
world spin-component {
|
||||||
|
include wasmeld:service/service-component@0.1.0;
|
||||||
|
}
|
||||||
@@ -0,0 +1,16 @@
|
|||||||
|
[package]
|
||||||
|
name = "wasi-clock-probe-component"
|
||||||
|
version = "0.1.0"
|
||||||
|
edition.workspace = true
|
||||||
|
rust-version.workspace = true
|
||||||
|
license.workspace = true
|
||||||
|
|
||||||
|
[package.metadata.wasmeld]
|
||||||
|
id = "wasi-clock-probe"
|
||||||
|
world = "component:wasi-clock-probe/wasi-clock-probe-component@0.1.0"
|
||||||
|
|
||||||
|
[lib]
|
||||||
|
crate-type = ["cdylib"]
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
wit-bindgen.workspace = true
|
||||||
@@ -0,0 +1,21 @@
|
|||||||
|
mod bindings {
|
||||||
|
wit_bindgen::generate!({
|
||||||
|
path: "wit",
|
||||||
|
world: "wasi-clock-probe-component",
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
struct WasiClockProbe;
|
||||||
|
|
||||||
|
impl bindings::Guest for WasiClockProbe {
|
||||||
|
fn init(_config: Vec<u8>) -> Result<(), bindings::ServiceError> {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn invoke(input: Vec<u8>) -> Result<Vec<u8>, bindings::ServiceError> {
|
||||||
|
let _ = std::time::Instant::now();
|
||||||
|
Ok(input)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
bindings::export!(WasiClockProbe with_types_in bindings);
|
||||||
@@ -0,0 +1,7 @@
|
|||||||
|
schema_version = 1
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
"wasmeld:service" = "0.1.0"
|
||||||
|
|
||||||
|
[replace."wasmeld:service"]
|
||||||
|
path = "../../wit/service"
|
||||||
@@ -0,0 +1,8 @@
|
|||||||
|
schema_version = 1
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "wasmeld:service"
|
||||||
|
version = "0.1.0"
|
||||||
|
source = "path+../../wit/service"
|
||||||
|
sha256 = "d5497307bbcd1e159f7707f385b488a6f2e26362d5256d5bc1080ac603a51305"
|
||||||
|
replaced = true
|
||||||
@@ -0,0 +1,5 @@
|
|||||||
|
package component:wasi-clock-probe@0.1.0;
|
||||||
|
|
||||||
|
world wasi-clock-probe-component {
|
||||||
|
include wasmeld:service/service-component@0.1.0;
|
||||||
|
}
|
||||||
@@ -0,0 +1,16 @@
|
|||||||
|
[package]
|
||||||
|
name = "wasmeld-runtime"
|
||||||
|
version = "0.1.0"
|
||||||
|
edition.workspace = true
|
||||||
|
rust-version.workspace = true
|
||||||
|
license.workspace = true
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
serde.workspace = true
|
||||||
|
thiserror.workspace = true
|
||||||
|
toml.workspace = true
|
||||||
|
wasmeld-package = { path = "../wasmeld-package" }
|
||||||
|
wasmtime.workspace = true
|
||||||
|
wasmtime-wasi.workspace = true
|
||||||
|
|
||||||
|
[dev-dependencies]
|
||||||
@@ -0,0 +1,16 @@
|
|||||||
|
//! Typed host and guest bindings generated from independently versioned WIT packages.
|
||||||
|
|
||||||
|
// The service world defines the exports every runnable component implements.
|
||||||
|
wasmtime::component::bindgen!({
|
||||||
|
path: "../../wit/service",
|
||||||
|
world: "service-component",
|
||||||
|
});
|
||||||
|
|
||||||
|
pub(crate) mod clock {
|
||||||
|
// Host capabilities are generated separately so adding a capability does
|
||||||
|
// not expand the baseline service contract for every component.
|
||||||
|
wasmtime::component::bindgen!({
|
||||||
|
path: "../../wit/clock",
|
||||||
|
world: "clock-host",
|
||||||
|
});
|
||||||
|
}
|
||||||
@@ -0,0 +1,81 @@
|
|||||||
|
//! Runtime error taxonomy exposed to management layers.
|
||||||
|
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
|
use thiserror::Error;
|
||||||
|
|
||||||
|
use crate::manifest::ServiceKey;
|
||||||
|
|
||||||
|
/// Failures from Runtime configuration, sandbox validation, or Actor execution.
|
||||||
|
#[derive(Debug, Error)]
|
||||||
|
pub enum RuntimeError {
|
||||||
|
#[error("service id and revision must not be empty")]
|
||||||
|
InvalidServiceKey,
|
||||||
|
|
||||||
|
#[error("invalid manifest: {0}")]
|
||||||
|
InvalidManifest(String),
|
||||||
|
|
||||||
|
#[error("manifest parse failed: {0}")]
|
||||||
|
ManifestParse(#[from] toml::de::Error),
|
||||||
|
|
||||||
|
#[error("service revision {0} is not registered")]
|
||||||
|
ServiceNotRegistered(ServiceKey),
|
||||||
|
|
||||||
|
#[error("service revision {0} is already registered")]
|
||||||
|
ServiceAlreadyRegistered(ServiceKey),
|
||||||
|
|
||||||
|
#[error("component compilation failed: {0}")]
|
||||||
|
ComponentCompilation(#[source] wasmtime::Error),
|
||||||
|
|
||||||
|
#[error("component imports unsupported capability {0}")]
|
||||||
|
UnsupportedImport(String),
|
||||||
|
|
||||||
|
#[error("runtime creation failed: {0}")]
|
||||||
|
RuntimeCreation(#[source] wasmtime::Error),
|
||||||
|
|
||||||
|
#[error("failed to read component artifact {path}: {source}")]
|
||||||
|
ArtifactRead {
|
||||||
|
path: String,
|
||||||
|
#[source]
|
||||||
|
source: std::io::Error,
|
||||||
|
},
|
||||||
|
|
||||||
|
#[error("failed to spawn actor thread: {0}")]
|
||||||
|
ActorThread(#[source] std::io::Error),
|
||||||
|
|
||||||
|
#[error("actor initialization failed: {0}")]
|
||||||
|
ActorInitialization(String),
|
||||||
|
|
||||||
|
#[error("actor for {0} is unavailable")]
|
||||||
|
ActorUnavailable(ServiceKey),
|
||||||
|
|
||||||
|
#[error("actor mailbox for {0} is full")]
|
||||||
|
ActorOverloaded(ServiceKey),
|
||||||
|
|
||||||
|
#[error("actor for {0} stopped before returning a result")]
|
||||||
|
ActorStopped(ServiceKey),
|
||||||
|
|
||||||
|
#[error("input for {service} exceeds the {limit}-byte limit")]
|
||||||
|
InputTooLarge { service: ServiceKey, limit: usize },
|
||||||
|
|
||||||
|
#[error("output for {service} exceeds the {limit}-byte limit")]
|
||||||
|
OutputTooLarge { service: ServiceKey, limit: usize },
|
||||||
|
|
||||||
|
#[error("actor for {service} exceeded its {deadline:?} execution deadline")]
|
||||||
|
DeadlineExceeded {
|
||||||
|
service: ServiceKey,
|
||||||
|
deadline: Duration,
|
||||||
|
},
|
||||||
|
|
||||||
|
#[error("actor for {service} faulted: {message}")]
|
||||||
|
ActorFault {
|
||||||
|
service: ServiceKey,
|
||||||
|
message: String,
|
||||||
|
},
|
||||||
|
|
||||||
|
#[error("component returned {kind}: {message}")]
|
||||||
|
ComponentError { kind: &'static str, message: String },
|
||||||
|
|
||||||
|
#[error("runtime lock was poisoned")]
|
||||||
|
LockPoisoned,
|
||||||
|
}
|
||||||
@@ -0,0 +1,16 @@
|
|||||||
|
//! A minimal resident WebAssembly Component runtime.
|
||||||
|
//!
|
||||||
|
//! The runtime intentionally has no network protocol or business capabilities.
|
||||||
|
//! It provides a restricted WASI compatibility context, links explicitly
|
||||||
|
//! imported host capabilities, loads a Component that implements the stable
|
||||||
|
//! service exports, and invokes it through one serial Actor.
|
||||||
|
|
||||||
|
mod bindings;
|
||||||
|
mod error;
|
||||||
|
mod manifest;
|
||||||
|
mod runtime;
|
||||||
|
|
||||||
|
pub use error::RuntimeError;
|
||||||
|
pub use manifest::{ResourceLimits, ServiceKey, ServiceManifest};
|
||||||
|
pub use runtime::{ActorHandle, Runtime, RuntimeConfig, RuntimeStats};
|
||||||
|
pub use wasmeld_package::SERVICE_WORLD;
|
||||||
@@ -0,0 +1,161 @@
|
|||||||
|
//! Runtime-owned service identity and resource policy.
|
||||||
|
|
||||||
|
use std::{fmt, time::Duration};
|
||||||
|
|
||||||
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
|
use crate::error::RuntimeError;
|
||||||
|
|
||||||
|
/// Immutable identity of one service revision.
|
||||||
|
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
|
||||||
|
pub struct ServiceKey {
|
||||||
|
id: String,
|
||||||
|
revision: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ServiceKey {
|
||||||
|
/// Creates a non-empty service identity.
|
||||||
|
pub fn new(id: impl Into<String>, revision: impl Into<String>) -> Result<Self, RuntimeError> {
|
||||||
|
let id = id.into();
|
||||||
|
let revision = revision.into();
|
||||||
|
|
||||||
|
if id.trim().is_empty() || revision.trim().is_empty() {
|
||||||
|
return Err(RuntimeError::InvalidServiceKey);
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(Self { id, revision })
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Returns the stable service identifier.
|
||||||
|
pub fn id(&self) -> &str {
|
||||||
|
&self.id
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Returns the immutable service revision.
|
||||||
|
pub fn revision(&self) -> &str {
|
||||||
|
&self.revision
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl fmt::Display for ServiceKey {
|
||||||
|
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||||
|
write!(formatter, "{}@{}", self.id, self.revision)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Runtime manifest generated by the Console for one registered component.
|
||||||
|
#[derive(Clone, Debug, Deserialize, Serialize)]
|
||||||
|
pub struct ServiceManifest {
|
||||||
|
/// Stable service identifier.
|
||||||
|
pub id: String,
|
||||||
|
/// Immutable component revision.
|
||||||
|
pub revision: String,
|
||||||
|
/// Local path to the compiled Component.
|
||||||
|
pub component: String,
|
||||||
|
/// Fully versioned WIT world implemented by the Component.
|
||||||
|
pub world: String,
|
||||||
|
/// Platform-owned execution and I/O limits.
|
||||||
|
pub limits: ResourceLimits,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ServiceManifest {
|
||||||
|
/// Parses and validates a TOML runtime manifest.
|
||||||
|
pub fn from_toml(source: &str) -> Result<Self, RuntimeError> {
|
||||||
|
let manifest = toml::from_str::<Self>(source)?;
|
||||||
|
manifest.validate()?;
|
||||||
|
Ok(manifest)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Derives the map key used by Runtime and Console state.
|
||||||
|
pub fn key(&self) -> Result<ServiceKey, RuntimeError> {
|
||||||
|
ServiceKey::new(self.id.clone(), self.revision.clone())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Validates component path, WIT world, identity, and resource limits.
|
||||||
|
pub fn validate(&self) -> Result<(), RuntimeError> {
|
||||||
|
self.key()?;
|
||||||
|
|
||||||
|
if self.component.trim().is_empty() {
|
||||||
|
return Err(RuntimeError::InvalidManifest(
|
||||||
|
"component must not be empty".to_owned(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
wasmeld_package::validate_world(&self.world)
|
||||||
|
.map_err(|error| RuntimeError::InvalidManifest(error.to_string()))?;
|
||||||
|
|
||||||
|
self.limits.validate()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Per-Actor sandbox, scheduling, mailbox, and payload limits.
|
||||||
|
#[derive(Clone, Debug, Deserialize, Serialize)]
|
||||||
|
pub struct ResourceLimits {
|
||||||
|
/// Maximum linear memory available to the Store.
|
||||||
|
pub memory_bytes: usize,
|
||||||
|
/// Wasmtime fuel replenished before each component call.
|
||||||
|
pub fuel_per_call: u64,
|
||||||
|
/// End-to-end deadline for one initialization or invocation.
|
||||||
|
pub deadline_ms: u64,
|
||||||
|
/// Number of queued commands accepted by the bounded Actor mailbox.
|
||||||
|
pub mailbox_capacity: usize,
|
||||||
|
/// Maximum request payload size.
|
||||||
|
pub max_input_bytes: usize,
|
||||||
|
/// Maximum response payload size.
|
||||||
|
pub max_output_bytes: usize,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for ResourceLimits {
|
||||||
|
fn default() -> Self {
|
||||||
|
Self {
|
||||||
|
memory_bytes: 64 * 1024 * 1024,
|
||||||
|
fuel_per_call: 1_000_000,
|
||||||
|
deadline_ms: 500,
|
||||||
|
mailbox_capacity: 16,
|
||||||
|
max_input_bytes: 16 * 1024,
|
||||||
|
max_output_bytes: 16 * 1024,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ResourceLimits {
|
||||||
|
/// Rejects limits that cannot enforce a meaningful execution boundary.
|
||||||
|
pub fn validate(&self) -> Result<(), RuntimeError> {
|
||||||
|
if self.memory_bytes == 0 {
|
||||||
|
return Err(RuntimeError::InvalidManifest(
|
||||||
|
"limits.memory_bytes must be greater than zero".to_owned(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
if self.fuel_per_call == 0 {
|
||||||
|
return Err(RuntimeError::InvalidManifest(
|
||||||
|
"limits.fuel_per_call must be greater than zero".to_owned(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
if self.deadline_ms == 0 {
|
||||||
|
return Err(RuntimeError::InvalidManifest(
|
||||||
|
"limits.deadline_ms must be greater than zero".to_owned(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
if self.mailbox_capacity == 0 {
|
||||||
|
return Err(RuntimeError::InvalidManifest(
|
||||||
|
"limits.mailbox_capacity must be greater than zero".to_owned(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
if self.max_input_bytes == 0 || self.max_output_bytes == 0 {
|
||||||
|
return Err(RuntimeError::InvalidManifest(
|
||||||
|
"input and output limits must be greater than zero".to_owned(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Returns the configured call deadline.
|
||||||
|
pub fn deadline(&self) -> Duration {
|
||||||
|
Duration::from_millis(self.deadline_ms)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,931 @@
|
|||||||
|
//! Wasmtime sandbox and resident Actor lifecycle.
|
||||||
|
//!
|
||||||
|
//! Each started service revision owns one thread, Store, Component Instance,
|
||||||
|
//! and bounded synchronous mailbox. The Actor processes commands serially, so
|
||||||
|
//! component memory remains resident without allowing concurrent entry into the
|
||||||
|
//! same Store.
|
||||||
|
|
||||||
|
use std::{
|
||||||
|
collections::HashMap,
|
||||||
|
fs,
|
||||||
|
sync::{
|
||||||
|
Arc, Mutex,
|
||||||
|
atomic::{AtomicBool, Ordering},
|
||||||
|
mpsc::{self, Receiver, SyncSender, TrySendError},
|
||||||
|
},
|
||||||
|
thread,
|
||||||
|
time::{Duration, Instant},
|
||||||
|
};
|
||||||
|
|
||||||
|
use wasmtime::{
|
||||||
|
Config, Engine, Store, StoreLimits, StoreLimitsBuilder,
|
||||||
|
component::{Component, HasData, Linker, ResourceTable},
|
||||||
|
};
|
||||||
|
use wasmtime_wasi::{
|
||||||
|
WasiCtx, WasiCtxBuilder, WasiCtxView, WasiView,
|
||||||
|
cli::{WasiCli, WasiCliView as _},
|
||||||
|
clocks::{WasiClocks, WasiClocksView as _},
|
||||||
|
filesystem::{WasiFilesystem, WasiFilesystemView as _},
|
||||||
|
p2::bindings::sync::{cli, clocks, filesystem, io},
|
||||||
|
};
|
||||||
|
|
||||||
|
use crate::{
|
||||||
|
RuntimeError, ServiceKey, ServiceManifest,
|
||||||
|
bindings::{ServiceComponent, ServiceError, clock as clock_bindings},
|
||||||
|
manifest::ResourceLimits,
|
||||||
|
};
|
||||||
|
|
||||||
|
const DEFAULT_EPOCH_TICK: Duration = Duration::from_millis(5);
|
||||||
|
const MAX_EPOCH_TICK: Duration = Duration::from_millis(10);
|
||||||
|
const DEFAULT_MAX_WASM_STACK: usize = 512 * 1024;
|
||||||
|
const ACTOR_RESOURCE_COUNT_LIMIT: usize = 16;
|
||||||
|
const ACTOR_TABLE_ELEMENT_LIMIT: usize = 16 * 1024;
|
||||||
|
const MONOTONIC_CLOCK_IMPORT: &str = "wasmeld:clock/monotonic-clock@0.1.0";
|
||||||
|
const ALLOWED_WASI_IMPORTS: &[&str] = &[
|
||||||
|
"wasi:cli/environment@",
|
||||||
|
"wasi:cli/exit@",
|
||||||
|
"wasi:cli/stderr@",
|
||||||
|
"wasi:cli/stdin@",
|
||||||
|
"wasi:cli/stdout@",
|
||||||
|
"wasi:clocks/wall-clock@",
|
||||||
|
"wasi:filesystem/preopens@",
|
||||||
|
"wasi:filesystem/types@",
|
||||||
|
"wasi:io/error@",
|
||||||
|
"wasi:io/streams@",
|
||||||
|
];
|
||||||
|
|
||||||
|
/// Process-wide Wasmtime Engine settings.
|
||||||
|
#[derive(Clone, Debug)]
|
||||||
|
pub struct RuntimeConfig {
|
||||||
|
/// Frequency used to advance Wasmtime epoch interruption.
|
||||||
|
pub epoch_tick: Duration,
|
||||||
|
/// Maximum native stack reservation for WebAssembly execution.
|
||||||
|
pub max_wasm_stack: usize,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for RuntimeConfig {
|
||||||
|
fn default() -> Self {
|
||||||
|
Self {
|
||||||
|
epoch_tick: DEFAULT_EPOCH_TICK,
|
||||||
|
max_wasm_stack: DEFAULT_MAX_WASM_STACK,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl RuntimeConfig {
|
||||||
|
fn validate(&self) -> Result<(), RuntimeError> {
|
||||||
|
if self.epoch_tick.is_zero() {
|
||||||
|
return Err(RuntimeError::InvalidManifest(
|
||||||
|
"runtime epoch_tick must be greater than zero".to_owned(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
if self.epoch_tick > MAX_EPOCH_TICK {
|
||||||
|
return Err(RuntimeError::InvalidManifest(format!(
|
||||||
|
"runtime epoch_tick must not exceed {MAX_EPOCH_TICK:?}"
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
|
||||||
|
if self.max_wasm_stack == 0 {
|
||||||
|
return Err(RuntimeError::InvalidManifest(
|
||||||
|
"runtime max_wasm_stack must be greater than zero".to_owned(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Registry and lifecycle manager for resident Component Actors.
|
||||||
|
pub struct Runtime {
|
||||||
|
inner: Arc<RuntimeInner>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Point-in-time Runtime service and Actor counts.
|
||||||
|
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
||||||
|
pub struct RuntimeStats {
|
||||||
|
/// Number of compiled, registered service revisions.
|
||||||
|
pub registered_services: usize,
|
||||||
|
/// Number of Actors currently accepting calls.
|
||||||
|
pub running_actors: usize,
|
||||||
|
}
|
||||||
|
|
||||||
|
struct RuntimeInner {
|
||||||
|
engine: Arc<Engine>,
|
||||||
|
epoch_tick: Duration,
|
||||||
|
services: Mutex<HashMap<ServiceKey, RegisteredService>>,
|
||||||
|
actors: Mutex<HashMap<ServiceKey, ActorHandle>>,
|
||||||
|
lifecycle: Mutex<()>,
|
||||||
|
ticker: Arc<EpochTicker>,
|
||||||
|
}
|
||||||
|
|
||||||
|
struct EpochTicker {
|
||||||
|
shutdown: mpsc::Sender<()>,
|
||||||
|
worker: Mutex<Option<thread::JoinHandle<()>>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl EpochTicker {
|
||||||
|
fn start(engine: Arc<Engine>, tick: Duration) -> Result<Arc<Self>, RuntimeError> {
|
||||||
|
let (shutdown, receiver) = mpsc::channel();
|
||||||
|
let worker = thread::Builder::new()
|
||||||
|
.name("wasmeld-runtime-epoch".to_owned())
|
||||||
|
.spawn(move || {
|
||||||
|
loop {
|
||||||
|
match receiver.recv_timeout(tick) {
|
||||||
|
Ok(()) | Err(mpsc::RecvTimeoutError::Disconnected) => return,
|
||||||
|
Err(mpsc::RecvTimeoutError::Timeout) => engine.increment_epoch(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
.map_err(RuntimeError::ActorThread)?;
|
||||||
|
|
||||||
|
Ok(Arc::new(Self {
|
||||||
|
shutdown,
|
||||||
|
worker: Mutex::new(Some(worker)),
|
||||||
|
}))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Drop for EpochTicker {
|
||||||
|
fn drop(&mut self) {
|
||||||
|
let _ = self.shutdown.send(());
|
||||||
|
|
||||||
|
if let Ok(mut worker) = self.worker.lock()
|
||||||
|
&& let Some(handle) = worker.take()
|
||||||
|
{
|
||||||
|
let _ = handle.join();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Clone)]
|
||||||
|
struct RegisteredService {
|
||||||
|
manifest: ServiceManifest,
|
||||||
|
component: Arc<Component>,
|
||||||
|
capabilities: Vec<HostCapability>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
||||||
|
enum HostCapability {
|
||||||
|
MonotonicClock,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Runtime {
|
||||||
|
/// Creates the shared Wasmtime Engine and epoch interruption worker.
|
||||||
|
pub fn new(config: RuntimeConfig) -> Result<Self, RuntimeError> {
|
||||||
|
config.validate()?;
|
||||||
|
|
||||||
|
let mut wasmtime_config = Config::new();
|
||||||
|
wasmtime_config.wasm_component_model(true);
|
||||||
|
wasmtime_config.async_support(false);
|
||||||
|
wasmtime_config.consume_fuel(true);
|
||||||
|
wasmtime_config.epoch_interruption(true);
|
||||||
|
wasmtime_config.max_wasm_stack(config.max_wasm_stack);
|
||||||
|
|
||||||
|
let engine =
|
||||||
|
Arc::new(Engine::new(&wasmtime_config).map_err(RuntimeError::RuntimeCreation)?);
|
||||||
|
let ticker = EpochTicker::start(Arc::clone(&engine), config.epoch_tick)?;
|
||||||
|
|
||||||
|
Ok(Self {
|
||||||
|
inner: Arc::new(RuntimeInner {
|
||||||
|
engine,
|
||||||
|
epoch_tick: config.epoch_tick,
|
||||||
|
services: Mutex::new(HashMap::new()),
|
||||||
|
actors: Mutex::new(HashMap::new()),
|
||||||
|
lifecycle: Mutex::new(()),
|
||||||
|
ticker,
|
||||||
|
}),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Compiles and registers a Component without starting an Actor.
|
||||||
|
///
|
||||||
|
/// Imports are checked against the restricted WASI and explicit Wasmeld
|
||||||
|
/// capability allowlist before the service becomes visible.
|
||||||
|
pub fn register(
|
||||||
|
&self,
|
||||||
|
manifest: ServiceManifest,
|
||||||
|
component_bytes: impl AsRef<[u8]>,
|
||||||
|
) -> Result<ServiceKey, RuntimeError> {
|
||||||
|
manifest.validate()?;
|
||||||
|
let key = manifest.key()?;
|
||||||
|
let component = Component::new(&self.inner.engine, component_bytes.as_ref())
|
||||||
|
.map_err(RuntimeError::ComponentCompilation)?;
|
||||||
|
let capabilities = validate_component_imports(&component, &self.inner.engine)?;
|
||||||
|
|
||||||
|
let mut services = self
|
||||||
|
.inner
|
||||||
|
.services
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| RuntimeError::LockPoisoned)?;
|
||||||
|
|
||||||
|
if services.contains_key(&key) {
|
||||||
|
return Err(RuntimeError::ServiceAlreadyRegistered(key));
|
||||||
|
}
|
||||||
|
|
||||||
|
services.insert(
|
||||||
|
key.clone(),
|
||||||
|
RegisteredService {
|
||||||
|
manifest,
|
||||||
|
component: Arc::new(component),
|
||||||
|
capabilities,
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
Ok(key)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Reads a Component from the manifest path and delegates to [`Self::register`].
|
||||||
|
pub fn register_from_file(
|
||||||
|
&self,
|
||||||
|
manifest: ServiceManifest,
|
||||||
|
) -> Result<ServiceKey, RuntimeError> {
|
||||||
|
manifest.validate()?;
|
||||||
|
let component_path = manifest.component.clone();
|
||||||
|
let bytes = fs::read(&component_path).map_err(|source| RuntimeError::ArtifactRead {
|
||||||
|
path: component_path,
|
||||||
|
source,
|
||||||
|
})?;
|
||||||
|
self.register(manifest, bytes)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Starts or returns the resident Actor for one registered service revision.
|
||||||
|
pub fn start(
|
||||||
|
&self,
|
||||||
|
key: &ServiceKey,
|
||||||
|
init_config: Vec<u8>,
|
||||||
|
) -> Result<ActorHandle, RuntimeError> {
|
||||||
|
let _lifecycle = self
|
||||||
|
.inner
|
||||||
|
.lifecycle
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| RuntimeError::LockPoisoned)?;
|
||||||
|
|
||||||
|
{
|
||||||
|
let mut actors = self
|
||||||
|
.inner
|
||||||
|
.actors
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| RuntimeError::LockPoisoned)?;
|
||||||
|
|
||||||
|
if let Some(actor) = actors.get(key) {
|
||||||
|
if actor.is_available() {
|
||||||
|
return Ok(actor.clone());
|
||||||
|
}
|
||||||
|
|
||||||
|
if actor.is_alive() {
|
||||||
|
return Err(RuntimeError::ActorUnavailable(key.clone()));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
actors.remove(key);
|
||||||
|
}
|
||||||
|
|
||||||
|
let service = self
|
||||||
|
.inner
|
||||||
|
.services
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| RuntimeError::LockPoisoned)?
|
||||||
|
.get(key)
|
||||||
|
.cloned()
|
||||||
|
.ok_or_else(|| RuntimeError::ServiceNotRegistered(key.clone()))?;
|
||||||
|
|
||||||
|
if service.manifest.limits.deadline() < self.inner.epoch_tick {
|
||||||
|
return Err(RuntimeError::InvalidManifest(format!(
|
||||||
|
"service deadline for {key} must be at least the runtime epoch_tick"
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
|
||||||
|
let actor = spawn_actor(
|
||||||
|
Arc::clone(&self.inner.engine),
|
||||||
|
Arc::clone(&self.inner.ticker),
|
||||||
|
self.inner.epoch_tick,
|
||||||
|
key.clone(),
|
||||||
|
service,
|
||||||
|
init_config,
|
||||||
|
)?;
|
||||||
|
|
||||||
|
self.inner
|
||||||
|
.actors
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| RuntimeError::LockPoisoned)?
|
||||||
|
.insert(key.clone(), actor.clone());
|
||||||
|
|
||||||
|
Ok(actor)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Enqueues one call on the service Actor's bounded serial mailbox.
|
||||||
|
pub fn invoke(&self, key: &ServiceKey, input: Vec<u8>) -> Result<Vec<u8>, RuntimeError> {
|
||||||
|
let actor = self
|
||||||
|
.inner
|
||||||
|
.actors
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| RuntimeError::LockPoisoned)?
|
||||||
|
.get(key)
|
||||||
|
.cloned()
|
||||||
|
.ok_or_else(|| RuntimeError::ActorUnavailable(key.clone()))?;
|
||||||
|
actor.invoke(input)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Stops one Actor and releases its Store and Component Instance.
|
||||||
|
pub fn stop(&self, key: &ServiceKey) -> Result<(), RuntimeError> {
|
||||||
|
let _lifecycle = self
|
||||||
|
.inner
|
||||||
|
.lifecycle
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| RuntimeError::LockPoisoned)?;
|
||||||
|
|
||||||
|
let actor = self
|
||||||
|
.inner
|
||||||
|
.actors
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| RuntimeError::LockPoisoned)?
|
||||||
|
.get(key)
|
||||||
|
.cloned()
|
||||||
|
.ok_or_else(|| RuntimeError::ActorUnavailable(key.clone()))?;
|
||||||
|
|
||||||
|
actor.stop()?;
|
||||||
|
self.inner
|
||||||
|
.actors
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| RuntimeError::LockPoisoned)?
|
||||||
|
.remove(key);
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Registers a new revision and immediately starts its Actor.
|
||||||
|
pub fn reload(
|
||||||
|
&self,
|
||||||
|
manifest: ServiceManifest,
|
||||||
|
component_bytes: impl AsRef<[u8]>,
|
||||||
|
init_config: Vec<u8>,
|
||||||
|
) -> Result<ActorHandle, RuntimeError> {
|
||||||
|
let key = self.register(manifest, component_bytes)?;
|
||||||
|
self.start(&key, init_config)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Returns counts without exposing internal Engine or Store state.
|
||||||
|
pub fn stats(&self) -> Result<RuntimeStats, RuntimeError> {
|
||||||
|
let registered_services = self
|
||||||
|
.inner
|
||||||
|
.services
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| RuntimeError::LockPoisoned)?
|
||||||
|
.len();
|
||||||
|
let running_actors = self
|
||||||
|
.inner
|
||||||
|
.actors
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| RuntimeError::LockPoisoned)?
|
||||||
|
.values()
|
||||||
|
.filter(|actor| actor.is_available())
|
||||||
|
.count();
|
||||||
|
Ok(RuntimeStats {
|
||||||
|
registered_services,
|
||||||
|
running_actors,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Stops every Actor and returns the first stop failure, if any.
|
||||||
|
pub fn stop_all(&self) -> Result<(), RuntimeError> {
|
||||||
|
let _lifecycle = self
|
||||||
|
.inner
|
||||||
|
.lifecycle
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| RuntimeError::LockPoisoned)?;
|
||||||
|
let actors = self
|
||||||
|
.inner
|
||||||
|
.actors
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| RuntimeError::LockPoisoned)?
|
||||||
|
.values()
|
||||||
|
.cloned()
|
||||||
|
.collect::<Vec<_>>();
|
||||||
|
let mut first_error = None;
|
||||||
|
|
||||||
|
for actor in actors {
|
||||||
|
if !actor.is_available() {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
if let Err(error) = actor.stop()
|
||||||
|
&& first_error.is_none()
|
||||||
|
{
|
||||||
|
first_error = Some(error);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
self.inner
|
||||||
|
.actors
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| RuntimeError::LockPoisoned)?
|
||||||
|
.clear();
|
||||||
|
|
||||||
|
match first_error {
|
||||||
|
Some(error) => Err(error),
|
||||||
|
None => Ok(()),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Cloneable command handle for one resident Actor.
|
||||||
|
///
|
||||||
|
/// Clones share the same bounded mailbox and status flags; they do not clone
|
||||||
|
/// the Wasmtime Store or Component Instance.
|
||||||
|
#[derive(Clone)]
|
||||||
|
pub struct ActorHandle {
|
||||||
|
key: ServiceKey,
|
||||||
|
limits: Arc<ResourceLimits>,
|
||||||
|
sender: SyncSender<ActorCommand>,
|
||||||
|
status: Arc<ActorStatus>,
|
||||||
|
_ticker: Arc<EpochTicker>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ActorHandle {
|
||||||
|
/// Returns the service revision served by this Actor.
|
||||||
|
pub fn key(&self) -> &ServiceKey {
|
||||||
|
&self.key
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Sends an invocation and waits up to the remaining service deadline.
|
||||||
|
pub fn invoke(&self, input: Vec<u8>) -> Result<Vec<u8>, RuntimeError> {
|
||||||
|
if !self.is_available() {
|
||||||
|
return Err(RuntimeError::ActorUnavailable(self.key.clone()));
|
||||||
|
}
|
||||||
|
|
||||||
|
if input.len() > self.limits.max_input_bytes {
|
||||||
|
return Err(RuntimeError::InputTooLarge {
|
||||||
|
service: self.key.clone(),
|
||||||
|
limit: self.limits.max_input_bytes,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
let (response_sender, response_receiver) = mpsc::sync_channel(1);
|
||||||
|
let deadline = Instant::now() + self.limits.deadline();
|
||||||
|
let command = ActorCommand::Invoke {
|
||||||
|
input,
|
||||||
|
deadline,
|
||||||
|
response_sender,
|
||||||
|
};
|
||||||
|
|
||||||
|
match self.sender.try_send(command) {
|
||||||
|
Ok(()) => {}
|
||||||
|
Err(TrySendError::Full(_)) => {
|
||||||
|
return Err(RuntimeError::ActorOverloaded(self.key.clone()));
|
||||||
|
}
|
||||||
|
Err(TrySendError::Disconnected(_)) => {
|
||||||
|
return Err(RuntimeError::ActorUnavailable(self.key.clone()));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
match response_receiver.recv_timeout(deadline.saturating_duration_since(Instant::now())) {
|
||||||
|
Ok(result) => result,
|
||||||
|
Err(mpsc::RecvTimeoutError::Timeout) => Err(RuntimeError::DeadlineExceeded {
|
||||||
|
service: self.key.clone(),
|
||||||
|
deadline: self.limits.deadline(),
|
||||||
|
}),
|
||||||
|
Err(mpsc::RecvTimeoutError::Disconnected) => {
|
||||||
|
Err(RuntimeError::ActorStopped(self.key.clone()))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Prevents new calls, asks the worker to stop, and waits for acknowledgement.
|
||||||
|
pub fn stop(&self) -> Result<(), RuntimeError> {
|
||||||
|
if !self.status.accepting.swap(false, Ordering::AcqRel) {
|
||||||
|
return Err(RuntimeError::ActorUnavailable(self.key.clone()));
|
||||||
|
}
|
||||||
|
|
||||||
|
let (response_sender, response_receiver) = mpsc::sync_channel(1);
|
||||||
|
match self.sender.try_send(ActorCommand::Stop { response_sender }) {
|
||||||
|
Ok(()) => {}
|
||||||
|
Err(TrySendError::Full(_)) => {
|
||||||
|
self.status.accepting.store(true, Ordering::Release);
|
||||||
|
return Err(RuntimeError::ActorOverloaded(self.key.clone()));
|
||||||
|
}
|
||||||
|
Err(TrySendError::Disconnected(_)) => {
|
||||||
|
self.status.alive.store(false, Ordering::Release);
|
||||||
|
return Err(RuntimeError::ActorUnavailable(self.key.clone()));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
response_receiver
|
||||||
|
.recv_timeout(self.limits.deadline())
|
||||||
|
.map_err(|_| RuntimeError::ActorStopped(self.key.clone()))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn is_available(&self) -> bool {
|
||||||
|
self.status.alive.load(Ordering::Acquire) && self.status.accepting.load(Ordering::Acquire)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn is_alive(&self) -> bool {
|
||||||
|
self.status.alive.load(Ordering::Acquire)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
enum ActorCommand {
|
||||||
|
Invoke {
|
||||||
|
input: Vec<u8>,
|
||||||
|
deadline: Instant,
|
||||||
|
response_sender: SyncSender<Result<Vec<u8>, RuntimeError>>,
|
||||||
|
},
|
||||||
|
Stop {
|
||||||
|
response_sender: SyncSender<()>,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
struct ActorStatus {
|
||||||
|
accepting: AtomicBool,
|
||||||
|
alive: AtomicBool,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ActorStatus {
|
||||||
|
fn new() -> Self {
|
||||||
|
Self {
|
||||||
|
accepting: AtomicBool::new(false),
|
||||||
|
alive: AtomicBool::new(false),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn mark_ready(&self) {
|
||||||
|
self.alive.store(true, Ordering::Release);
|
||||||
|
self.accepting.store(true, Ordering::Release);
|
||||||
|
}
|
||||||
|
|
||||||
|
fn mark_stopped(&self) {
|
||||||
|
self.accepting.store(false, Ordering::Release);
|
||||||
|
self.alive.store(false, Ordering::Release);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct HostState {
|
||||||
|
store_limits: StoreLimits,
|
||||||
|
wasi: WasiCtx,
|
||||||
|
wasi_table: ResourceTable,
|
||||||
|
clock_origin: Instant,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl HostState {
|
||||||
|
fn new(limits: &ResourceLimits) -> Self {
|
||||||
|
let store_limits = StoreLimitsBuilder::new()
|
||||||
|
.memory_size(limits.memory_bytes)
|
||||||
|
.table_elements(ACTOR_TABLE_ELEMENT_LIMIT)
|
||||||
|
.instances(ACTOR_RESOURCE_COUNT_LIMIT)
|
||||||
|
.tables(ACTOR_RESOURCE_COUNT_LIMIT)
|
||||||
|
.memories(ACTOR_RESOURCE_COUNT_LIMIT)
|
||||||
|
.trap_on_grow_failure(true)
|
||||||
|
.build();
|
||||||
|
|
||||||
|
let mut wasi = WasiCtxBuilder::new();
|
||||||
|
wasi.allow_tcp(false)
|
||||||
|
.allow_udp(false)
|
||||||
|
.allow_ip_name_lookup(false)
|
||||||
|
.socket_addr_check(|_, _| Box::pin(async { false }));
|
||||||
|
|
||||||
|
Self {
|
||||||
|
store_limits,
|
||||||
|
wasi: wasi.build(),
|
||||||
|
wasi_table: ResourceTable::new(),
|
||||||
|
clock_origin: Instant::now(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl HasData for HostState {
|
||||||
|
type Data<'a> = &'a mut HostState;
|
||||||
|
}
|
||||||
|
|
||||||
|
impl clock_bindings::wasmeld::clock::monotonic_clock::Host for HostState {
|
||||||
|
fn now(&mut self) -> u64 {
|
||||||
|
u64::try_from(self.clock_origin.elapsed().as_nanos()).unwrap_or(u64::MAX)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl WasiView for HostState {
|
||||||
|
fn ctx(&mut self) -> WasiCtxView<'_> {
|
||||||
|
WasiCtxView {
|
||||||
|
ctx: &mut self.wasi,
|
||||||
|
table: &mut self.wasi_table,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct WasiIoTable;
|
||||||
|
|
||||||
|
impl HasData for WasiIoTable {
|
||||||
|
type Data<'a> = &'a mut ResourceTable;
|
||||||
|
}
|
||||||
|
|
||||||
|
struct ActorWorker {
|
||||||
|
key: ServiceKey,
|
||||||
|
limits: Arc<ResourceLimits>,
|
||||||
|
epoch_tick: Duration,
|
||||||
|
store: Store<HostState>,
|
||||||
|
bindings: ServiceComponent,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ActorWorker {
|
||||||
|
fn create(
|
||||||
|
engine: Arc<Engine>,
|
||||||
|
epoch_tick: Duration,
|
||||||
|
key: ServiceKey,
|
||||||
|
service: RegisteredService,
|
||||||
|
init_config: Vec<u8>,
|
||||||
|
) -> Result<Self, RuntimeError> {
|
||||||
|
let limits = Arc::new(service.manifest.limits.clone());
|
||||||
|
if init_config.len() > limits.max_input_bytes {
|
||||||
|
return Err(RuntimeError::InputTooLarge {
|
||||||
|
service: key,
|
||||||
|
limit: limits.max_input_bytes,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut store = Store::new(&engine, HostState::new(&limits));
|
||||||
|
store.limiter(|state| &mut state.store_limits);
|
||||||
|
|
||||||
|
let mut linker = Linker::new(&engine);
|
||||||
|
add_restricted_wasi(&mut linker)?;
|
||||||
|
add_host_capabilities(&mut linker, &service.capabilities)?;
|
||||||
|
|
||||||
|
let epoch_ticks = ticks_for(limits.deadline(), epoch_tick);
|
||||||
|
configure_call_budget(&mut store, &limits, epoch_ticks)?;
|
||||||
|
let bindings = ServiceComponent::instantiate(&mut store, &service.component, &linker)
|
||||||
|
.map_err(|error| RuntimeError::ActorInitialization(error.to_string()))?;
|
||||||
|
|
||||||
|
configure_call_budget(&mut store, &limits, epoch_ticks)?;
|
||||||
|
|
||||||
|
match bindings.call_init(&mut store, &init_config) {
|
||||||
|
Ok(Ok(())) => {}
|
||||||
|
Ok(Err(error)) => {
|
||||||
|
return Err(component_error("init", error));
|
||||||
|
}
|
||||||
|
Err(error) => {
|
||||||
|
return Err(RuntimeError::ActorInitialization(error.to_string()));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(Self {
|
||||||
|
key,
|
||||||
|
limits,
|
||||||
|
epoch_tick,
|
||||||
|
store,
|
||||||
|
bindings,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
fn invoke(&mut self, input: Vec<u8>, deadline: Instant) -> Result<Vec<u8>, RuntimeError> {
|
||||||
|
let remaining = deadline.saturating_duration_since(Instant::now());
|
||||||
|
if remaining.is_zero() {
|
||||||
|
return Err(RuntimeError::DeadlineExceeded {
|
||||||
|
service: self.key.clone(),
|
||||||
|
deadline: self.limits.deadline(),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
configure_call_budget(
|
||||||
|
&mut self.store,
|
||||||
|
&self.limits,
|
||||||
|
ticks_for(remaining, self.epoch_tick),
|
||||||
|
)?;
|
||||||
|
|
||||||
|
match self.bindings.call_invoke(&mut self.store, &input) {
|
||||||
|
Ok(Ok(output)) => {
|
||||||
|
if output.len() > self.limits.max_output_bytes {
|
||||||
|
return Err(RuntimeError::OutputTooLarge {
|
||||||
|
service: self.key.clone(),
|
||||||
|
limit: self.limits.max_output_bytes,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(output)
|
||||||
|
}
|
||||||
|
Ok(Err(error)) => Err(component_error("invoke", error)),
|
||||||
|
Err(error) => Err(RuntimeError::ActorFault {
|
||||||
|
service: self.key.clone(),
|
||||||
|
message: error.to_string(),
|
||||||
|
}),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn add_restricted_wasi(linker: &mut Linker<HostState>) -> Result<(), RuntimeError> {
|
||||||
|
let exit_options = cli::exit::LinkOptions::default();
|
||||||
|
link_wasi(clocks::wall_clock::add_to_linker::<HostState, WasiClocks>(
|
||||||
|
linker,
|
||||||
|
|state| state.clocks(),
|
||||||
|
))?;
|
||||||
|
link_wasi(filesystem::preopens::add_to_linker::<
|
||||||
|
HostState,
|
||||||
|
WasiFilesystem,
|
||||||
|
>(linker, |state| state.filesystem()))?;
|
||||||
|
link_wasi(
|
||||||
|
filesystem::types::add_to_linker::<HostState, WasiFilesystem>(linker, |state| {
|
||||||
|
state.filesystem()
|
||||||
|
}),
|
||||||
|
)?;
|
||||||
|
link_wasi(cli::exit::add_to_linker::<HostState, WasiCli>(
|
||||||
|
linker,
|
||||||
|
&exit_options,
|
||||||
|
|state| state.cli(),
|
||||||
|
))?;
|
||||||
|
link_wasi(cli::environment::add_to_linker::<HostState, WasiCli>(
|
||||||
|
linker,
|
||||||
|
|state| state.cli(),
|
||||||
|
))?;
|
||||||
|
link_wasi(cli::stdin::add_to_linker::<HostState, WasiCli>(
|
||||||
|
linker,
|
||||||
|
|state| state.cli(),
|
||||||
|
))?;
|
||||||
|
link_wasi(cli::stdout::add_to_linker::<HostState, WasiCli>(
|
||||||
|
linker,
|
||||||
|
|state| state.cli(),
|
||||||
|
))?;
|
||||||
|
link_wasi(cli::stderr::add_to_linker::<HostState, WasiCli>(
|
||||||
|
linker,
|
||||||
|
|state| state.cli(),
|
||||||
|
))?;
|
||||||
|
link_wasi(io::error::add_to_linker::<HostState, WasiIoTable>(
|
||||||
|
linker,
|
||||||
|
|state| state.ctx().table,
|
||||||
|
))?;
|
||||||
|
link_wasi(io::streams::add_to_linker::<HostState, WasiIoTable>(
|
||||||
|
linker,
|
||||||
|
|state| state.ctx().table,
|
||||||
|
))?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn link_wasi(result: wasmtime::Result<()>) -> Result<(), RuntimeError> {
|
||||||
|
result.map_err(|error| RuntimeError::ActorInitialization(error.to_string()))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn add_host_capabilities(
|
||||||
|
linker: &mut Linker<HostState>,
|
||||||
|
capabilities: &[HostCapability],
|
||||||
|
) -> Result<(), RuntimeError> {
|
||||||
|
for capability in capabilities {
|
||||||
|
match capability {
|
||||||
|
HostCapability::MonotonicClock => link_wasi(
|
||||||
|
clock_bindings::wasmeld::clock::monotonic_clock::add_to_linker::<
|
||||||
|
HostState,
|
||||||
|
HostState,
|
||||||
|
>(linker, |state| state),
|
||||||
|
)?,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn validate_component_imports(
|
||||||
|
component: &Component,
|
||||||
|
engine: &Engine,
|
||||||
|
) -> Result<Vec<HostCapability>, RuntimeError> {
|
||||||
|
// Import validation is deny-by-default. A host function is linked only
|
||||||
|
// after its exact WIT identity or WASI family has passed this allowlist.
|
||||||
|
let mut capabilities = Vec::new();
|
||||||
|
for (name, _) in component.component_type().imports(engine) {
|
||||||
|
if name == MONOTONIC_CLOCK_IMPORT {
|
||||||
|
capabilities.push(HostCapability::MonotonicClock);
|
||||||
|
} else if !ALLOWED_WASI_IMPORTS
|
||||||
|
.iter()
|
||||||
|
.any(|allowed| name.starts_with(allowed))
|
||||||
|
{
|
||||||
|
return Err(RuntimeError::UnsupportedImport(name.to_owned()));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
Ok(capabilities)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn spawn_actor(
|
||||||
|
engine: Arc<Engine>,
|
||||||
|
ticker: Arc<EpochTicker>,
|
||||||
|
epoch_tick: Duration,
|
||||||
|
key: ServiceKey,
|
||||||
|
service: RegisteredService,
|
||||||
|
init_config: Vec<u8>,
|
||||||
|
) -> Result<ActorHandle, RuntimeError> {
|
||||||
|
let limits = Arc::new(service.manifest.limits.clone());
|
||||||
|
let (sender, receiver) = mpsc::sync_channel(limits.mailbox_capacity);
|
||||||
|
let (ready_sender, ready_receiver) = mpsc::sync_channel(1);
|
||||||
|
let thread_key = key.clone();
|
||||||
|
let status = Arc::new(ActorStatus::new());
|
||||||
|
let worker_status = Arc::clone(&status);
|
||||||
|
let worker_ticker = Arc::clone(&ticker);
|
||||||
|
|
||||||
|
thread::Builder::new()
|
||||||
|
.name(format!("wasm-actor-{thread_key}"))
|
||||||
|
.spawn(move || {
|
||||||
|
let _ticker = worker_ticker;
|
||||||
|
|
||||||
|
match ActorWorker::create(engine, epoch_tick, key, service, init_config) {
|
||||||
|
Ok(mut worker) => {
|
||||||
|
worker_status.mark_ready();
|
||||||
|
let _ = ready_sender.send(Ok(()));
|
||||||
|
run_actor(&mut worker, receiver, &worker_status);
|
||||||
|
}
|
||||||
|
Err(error) => {
|
||||||
|
worker_status.mark_stopped();
|
||||||
|
let _ = ready_sender.send(Err(error));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
.map_err(RuntimeError::ActorThread)?;
|
||||||
|
|
||||||
|
ready_receiver.recv().map_err(|_| {
|
||||||
|
RuntimeError::ActorInitialization("actor thread stopped during initialization".to_owned())
|
||||||
|
})??;
|
||||||
|
|
||||||
|
Ok(ActorHandle {
|
||||||
|
key: thread_key,
|
||||||
|
limits,
|
||||||
|
sender,
|
||||||
|
status,
|
||||||
|
_ticker: ticker,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
fn run_actor(worker: &mut ActorWorker, receiver: Receiver<ActorCommand>, status: &ActorStatus) {
|
||||||
|
// This is the only loop that enters the Store. Multiple ActorHandle clones
|
||||||
|
// therefore cannot invoke the same Component Instance concurrently.
|
||||||
|
while let Ok(command) = receiver.recv() {
|
||||||
|
match command {
|
||||||
|
ActorCommand::Invoke {
|
||||||
|
input,
|
||||||
|
deadline,
|
||||||
|
response_sender,
|
||||||
|
} => {
|
||||||
|
let result = if Instant::now() >= deadline {
|
||||||
|
Err(RuntimeError::DeadlineExceeded {
|
||||||
|
service: worker.key.clone(),
|
||||||
|
deadline: worker.limits.deadline(),
|
||||||
|
})
|
||||||
|
} else {
|
||||||
|
worker.invoke(input, deadline)
|
||||||
|
};
|
||||||
|
let fatal = is_fatal(&result);
|
||||||
|
if fatal {
|
||||||
|
status.mark_stopped();
|
||||||
|
}
|
||||||
|
let _ = response_sender.send(result);
|
||||||
|
|
||||||
|
if fatal {
|
||||||
|
drain_after_failure(receiver, &worker.key);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
ActorCommand::Stop { response_sender } => {
|
||||||
|
status.mark_stopped();
|
||||||
|
let _ = response_sender.send(());
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
status.mark_stopped();
|
||||||
|
}
|
||||||
|
|
||||||
|
fn drain_after_failure(receiver: Receiver<ActorCommand>, key: &ServiceKey) {
|
||||||
|
while let Ok(command) = receiver.try_recv() {
|
||||||
|
match command {
|
||||||
|
ActorCommand::Invoke {
|
||||||
|
response_sender, ..
|
||||||
|
} => {
|
||||||
|
let _ = response_sender.send(Err(RuntimeError::ActorUnavailable(key.clone())));
|
||||||
|
}
|
||||||
|
ActorCommand::Stop { response_sender } => {
|
||||||
|
let _ = response_sender.send(());
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn configure_call_budget(
|
||||||
|
store: &mut Store<HostState>,
|
||||||
|
limits: &ResourceLimits,
|
||||||
|
epoch_ticks: u64,
|
||||||
|
) -> Result<(), RuntimeError> {
|
||||||
|
store
|
||||||
|
.set_fuel(limits.fuel_per_call)
|
||||||
|
.map_err(|error| RuntimeError::ActorInitialization(error.to_string()))?;
|
||||||
|
store.set_epoch_deadline(epoch_ticks);
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn ticks_for(deadline: Duration, epoch_tick: Duration) -> u64 {
|
||||||
|
let deadline_nanos = deadline.as_nanos();
|
||||||
|
let tick_nanos = epoch_tick.as_nanos().max(1);
|
||||||
|
let ticks = deadline_nanos.div_ceil(tick_nanos).max(1);
|
||||||
|
u64::try_from(ticks).unwrap_or(u64::MAX)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn component_error(kind: &'static str, error: ServiceError) -> RuntimeError {
|
||||||
|
let message = match error {
|
||||||
|
ServiceError::InitFailed(message) | ServiceError::CallFailed(message) => message,
|
||||||
|
};
|
||||||
|
|
||||||
|
RuntimeError::ComponentError { kind, message }
|
||||||
|
}
|
||||||
|
|
||||||
|
fn is_fatal(result: &Result<Vec<u8>, RuntimeError>) -> bool {
|
||||||
|
matches!(result, Err(RuntimeError::ActorFault { .. }))
|
||||||
|
}
|
||||||
@@ -0,0 +1,323 @@
|
|||||||
|
use std::{
|
||||||
|
fs,
|
||||||
|
path::{Path, PathBuf},
|
||||||
|
process::Command,
|
||||||
|
sync::OnceLock,
|
||||||
|
thread,
|
||||||
|
time::{Duration, Instant},
|
||||||
|
};
|
||||||
|
|
||||||
|
use wasmeld_runtime::{ResourceLimits, Runtime, RuntimeConfig, RuntimeError, ServiceManifest};
|
||||||
|
|
||||||
|
static COMPONENTS_BUILT: OnceLock<()> = OnceLock::new();
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn echo_component_round_trips_bytes() {
|
||||||
|
let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start");
|
||||||
|
let (key, actor) = start_component(&runtime, "echo", "echo_component.wasm");
|
||||||
|
|
||||||
|
assert_eq!(actor.invoke(b"hello wasm".to_vec()).unwrap(), b"hello wasm");
|
||||||
|
|
||||||
|
runtime.stop(&key).unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn counter_memory_is_resident_and_resets_after_restart() {
|
||||||
|
let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start");
|
||||||
|
let (key, actor) = start_component(&runtime, "counter", "counter_component.wasm");
|
||||||
|
|
||||||
|
assert_eq!(decode_counter(actor.invoke(Vec::new()).unwrap()), 1);
|
||||||
|
assert_eq!(decode_counter(actor.invoke(Vec::new()).unwrap()), 2);
|
||||||
|
|
||||||
|
runtime.stop(&key).unwrap();
|
||||||
|
let restarted = runtime.start(&key, Vec::new()).unwrap();
|
||||||
|
assert_eq!(decode_counter(restarted.invoke(Vec::new()).unwrap()), 1);
|
||||||
|
|
||||||
|
runtime.stop(&key).unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn actor_serializes_concurrent_counter_calls() {
|
||||||
|
let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start");
|
||||||
|
let (key, actor) = start_component(&runtime, "counter-concurrent", "counter_component.wasm");
|
||||||
|
|
||||||
|
let mut workers = Vec::new();
|
||||||
|
for _ in 0..8 {
|
||||||
|
let actor = actor.clone();
|
||||||
|
workers.push(thread::spawn(move || {
|
||||||
|
decode_counter(actor.invoke(Vec::new()).unwrap())
|
||||||
|
}));
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut values = workers
|
||||||
|
.into_iter()
|
||||||
|
.map(|worker| worker.join().expect("worker should not panic"))
|
||||||
|
.collect::<Vec<_>>();
|
||||||
|
values.sort_unstable();
|
||||||
|
assert_eq!(values, (1..=8).collect::<Vec<_>>());
|
||||||
|
|
||||||
|
runtime.stop(&key).unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn runtime_reports_stats_and_stops_all_actors() {
|
||||||
|
let runtime = Runtime::new(RuntimeConfig::default()).unwrap();
|
||||||
|
let (key, actor) = start_component(&runtime, "runtime-control", "echo_component.wasm");
|
||||||
|
|
||||||
|
assert_eq!(runtime.stats().unwrap().registered_services, 1);
|
||||||
|
assert_eq!(runtime.stats().unwrap().running_actors, 1);
|
||||||
|
|
||||||
|
runtime.stop_all().unwrap();
|
||||||
|
assert_eq!(runtime.stats().unwrap().registered_services, 1);
|
||||||
|
assert_eq!(runtime.stats().unwrap().running_actors, 0);
|
||||||
|
assert!(matches!(
|
||||||
|
actor.invoke(b"stopped".to_vec()),
|
||||||
|
Err(RuntimeError::ActorUnavailable(unavailable)) if unavailable == key
|
||||||
|
));
|
||||||
|
|
||||||
|
runtime.start(&key, Vec::new()).unwrap();
|
||||||
|
assert_eq!(
|
||||||
|
runtime.invoke(&key, b"started-again".to_vec()).unwrap(),
|
||||||
|
b"started-again"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn faulted_actor_can_be_started_again() {
|
||||||
|
let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start");
|
||||||
|
let (key, actor) = start_component(&runtime, "fault", "fault_component.wasm");
|
||||||
|
|
||||||
|
assert!(matches!(
|
||||||
|
actor.invoke(b"fault".to_vec()),
|
||||||
|
Err(RuntimeError::ActorFault { .. })
|
||||||
|
));
|
||||||
|
|
||||||
|
let restarted = runtime.start(&key, Vec::new()).unwrap();
|
||||||
|
assert_eq!(restarted.invoke(b"ready".to_vec()).unwrap(), b"ready");
|
||||||
|
|
||||||
|
runtime.stop(&key).unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn fuel_limit_stops_cpu_bound_component() {
|
||||||
|
let runtime = Runtime::new(RuntimeConfig {
|
||||||
|
epoch_tick: Duration::from_millis(10),
|
||||||
|
..RuntimeConfig::default()
|
||||||
|
})
|
||||||
|
.expect("runtime should start");
|
||||||
|
let key = register_component(
|
||||||
|
&runtime,
|
||||||
|
"spin-fuel",
|
||||||
|
"spin_component.wasm",
|
||||||
|
ResourceLimits {
|
||||||
|
fuel_per_call: 10_000,
|
||||||
|
deadline_ms: 50,
|
||||||
|
..test_limits()
|
||||||
|
},
|
||||||
|
);
|
||||||
|
let actor = runtime.start(&key, Vec::new()).unwrap();
|
||||||
|
|
||||||
|
let error = actor.invoke(u64::MAX.to_le_bytes().to_vec()).unwrap_err();
|
||||||
|
assert!(
|
||||||
|
matches!(&error, RuntimeError::ActorFault { .. }),
|
||||||
|
"expected a Wasm trap after fuel exhaustion, got {error:?}"
|
||||||
|
);
|
||||||
|
assert!(matches!(
|
||||||
|
actor.invoke(Vec::new()),
|
||||||
|
Err(RuntimeError::ActorUnavailable(_))
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn epoch_interrupts_cpu_bound_initialization() {
|
||||||
|
let runtime = Runtime::new(RuntimeConfig {
|
||||||
|
epoch_tick: Duration::from_millis(2),
|
||||||
|
..RuntimeConfig::default()
|
||||||
|
})
|
||||||
|
.expect("runtime should start");
|
||||||
|
let key = register_component(
|
||||||
|
&runtime,
|
||||||
|
"spin-epoch",
|
||||||
|
"spin_component.wasm",
|
||||||
|
ResourceLimits {
|
||||||
|
fuel_per_call: 10_000_000_000,
|
||||||
|
deadline_ms: 100,
|
||||||
|
..test_limits()
|
||||||
|
},
|
||||||
|
);
|
||||||
|
|
||||||
|
let started_at = Instant::now();
|
||||||
|
let error = match runtime.start(&key, b"spin".to_vec()) {
|
||||||
|
Ok(_) => panic!("epoch budget should interrupt the spin component"),
|
||||||
|
Err(error) => error,
|
||||||
|
};
|
||||||
|
assert!(
|
||||||
|
matches!(&error, RuntimeError::ActorInitialization(_)),
|
||||||
|
"expected initialization to be interrupted, got {error:?}"
|
||||||
|
);
|
||||||
|
let elapsed = started_at.elapsed();
|
||||||
|
assert!(
|
||||||
|
elapsed >= Duration::from_millis(20),
|
||||||
|
"initialization failed before the epoch budget could expire"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
elapsed < Duration::from_secs(2),
|
||||||
|
"epoch interruption took too long"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn explicit_clock_capability_is_linked() {
|
||||||
|
let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start");
|
||||||
|
let (key, actor) = start_component(&runtime, "clock-probe", "clock_probe_component.wasm");
|
||||||
|
|
||||||
|
let first = actor.invoke(b"first".to_vec()).unwrap();
|
||||||
|
let second = actor.invoke(b"second".to_vec()).unwrap();
|
||||||
|
assert_eq!(&first[8..], b"first");
|
||||||
|
assert_eq!(&second[8..], b"second");
|
||||||
|
assert!(decode_clock(&second) >= decode_clock(&first));
|
||||||
|
|
||||||
|
runtime.stop(&key).unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn registration_rejects_non_whitelisted_wasi_imports() {
|
||||||
|
let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start");
|
||||||
|
let artifact_name = "wasi_clock_probe_component.wasm";
|
||||||
|
let artifact = component_artifact(artifact_name);
|
||||||
|
let manifest = ServiceManifest {
|
||||||
|
id: "wasi-clock-probe".to_owned(),
|
||||||
|
revision: "0.1.0".to_owned(),
|
||||||
|
component: artifact.display().to_string(),
|
||||||
|
world: component_world(artifact_name).to_owned(),
|
||||||
|
limits: test_limits(),
|
||||||
|
};
|
||||||
|
|
||||||
|
let error = runtime
|
||||||
|
.register(manifest, fs::read(artifact).unwrap())
|
||||||
|
.expect_err("monotonic clock must not be a V1 capability");
|
||||||
|
assert!(matches!(
|
||||||
|
error,
|
||||||
|
RuntimeError::UnsupportedImport(name) if name.starts_with("wasi:clocks/monotonic-clock@")
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
fn start_component(
|
||||||
|
runtime: &Runtime,
|
||||||
|
service_id: &str,
|
||||||
|
artifact_name: &str,
|
||||||
|
) -> (wasmeld_runtime::ServiceKey, wasmeld_runtime::ActorHandle) {
|
||||||
|
let key = register_component(runtime, service_id, artifact_name, test_limits());
|
||||||
|
let actor = runtime.start(&key, Vec::new()).unwrap();
|
||||||
|
(key, actor)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn register_component(
|
||||||
|
runtime: &Runtime,
|
||||||
|
service_id: &str,
|
||||||
|
artifact_name: &str,
|
||||||
|
limits: ResourceLimits,
|
||||||
|
) -> wasmeld_runtime::ServiceKey {
|
||||||
|
let artifact = component_artifact(artifact_name);
|
||||||
|
let manifest = ServiceManifest {
|
||||||
|
id: service_id.to_owned(),
|
||||||
|
revision: "0.1.0".to_owned(),
|
||||||
|
component: artifact.display().to_string(),
|
||||||
|
world: component_world(artifact_name).to_owned(),
|
||||||
|
limits,
|
||||||
|
};
|
||||||
|
|
||||||
|
let bytes = fs::read(&artifact).expect("component artifact should be readable");
|
||||||
|
runtime.register(manifest, bytes).unwrap()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn component_artifact(name: &str) -> PathBuf {
|
||||||
|
build_components_once();
|
||||||
|
workspace_root()
|
||||||
|
.join("target/wasm32-wasip2/release")
|
||||||
|
.join(name)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn build_components_once() {
|
||||||
|
COMPONENTS_BUILT.get_or_init(|| {
|
||||||
|
let toolchain =
|
||||||
|
std::env::var("WASMELD_COMPONENT_TOOLCHAIN").unwrap_or_else(|_| "1.90.0".to_owned());
|
||||||
|
for manifest in [
|
||||||
|
"components/echo/Cargo.toml",
|
||||||
|
"components/counter/Cargo.toml",
|
||||||
|
"components/fault/Cargo.toml",
|
||||||
|
"components/spin/Cargo.toml",
|
||||||
|
"components/clock-probe/Cargo.toml",
|
||||||
|
"components/wasi-clock-probe/Cargo.toml",
|
||||||
|
] {
|
||||||
|
let status = Command::new("rustup")
|
||||||
|
.args([
|
||||||
|
"run",
|
||||||
|
&toolchain,
|
||||||
|
"cargo",
|
||||||
|
"build",
|
||||||
|
"--manifest-path",
|
||||||
|
manifest,
|
||||||
|
"--target",
|
||||||
|
"wasm32-wasip2",
|
||||||
|
"--release",
|
||||||
|
])
|
||||||
|
.current_dir(workspace_root())
|
||||||
|
.status()
|
||||||
|
.expect("component build command should start");
|
||||||
|
assert!(status.success(), "component build should succeed");
|
||||||
|
}
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
fn workspace_root() -> &'static Path {
|
||||||
|
static ROOT: OnceLock<PathBuf> = OnceLock::new();
|
||||||
|
ROOT.get_or_init(|| {
|
||||||
|
PathBuf::from(env!("CARGO_MANIFEST_DIR"))
|
||||||
|
.parent()
|
||||||
|
.and_then(Path::parent)
|
||||||
|
.expect("runtime crate must live in the workspace")
|
||||||
|
.to_path_buf()
|
||||||
|
})
|
||||||
|
.as_path()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn test_limits() -> ResourceLimits {
|
||||||
|
ResourceLimits {
|
||||||
|
memory_bytes: 64 * 1024 * 1024,
|
||||||
|
fuel_per_call: 1_000_000,
|
||||||
|
deadline_ms: 500,
|
||||||
|
mailbox_capacity: 16,
|
||||||
|
max_input_bytes: 16 * 1024,
|
||||||
|
max_output_bytes: 16 * 1024,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn decode_counter(bytes: Vec<u8>) -> u64 {
|
||||||
|
let array: [u8; 8] = bytes
|
||||||
|
.try_into()
|
||||||
|
.expect("counter response must be a u64 payload");
|
||||||
|
u64::from_le_bytes(array)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn decode_clock(bytes: &[u8]) -> u64 {
|
||||||
|
u64::from_le_bytes(
|
||||||
|
bytes[..8]
|
||||||
|
.try_into()
|
||||||
|
.expect("clock response must start with a u64 payload"),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn component_world(artifact_name: &str) -> &'static str {
|
||||||
|
match artifact_name {
|
||||||
|
"echo_component.wasm" => "component:echo/echo-component@0.1.0",
|
||||||
|
"counter_component.wasm" => "component:counter/counter-component@0.1.0",
|
||||||
|
"fault_component.wasm" => "component:fault/fault-component@0.1.0",
|
||||||
|
"spin_component.wasm" => "component:spin/spin-component@0.1.0",
|
||||||
|
"clock_probe_component.wasm" => "component:clock-probe/clock-probe-component@0.1.0",
|
||||||
|
"wasi_clock_probe_component.wasm" => {
|
||||||
|
"component:wasi-clock-probe/wasi-clock-probe-component@0.1.0"
|
||||||
|
}
|
||||||
|
other => panic!("no WIT world registered for {other}"),
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -0,0 +1,9 @@
|
|||||||
|
package wasmeld:clock@0.1.0;
|
||||||
|
|
||||||
|
interface monotonic-clock {
|
||||||
|
now: func() -> u64;
|
||||||
|
}
|
||||||
|
|
||||||
|
world clock-host {
|
||||||
|
import monotonic-clock;
|
||||||
|
}
|
||||||
@@ -0,0 +1,11 @@
|
|||||||
|
package wasmeld:service@0.1.0;
|
||||||
|
|
||||||
|
world service-component {
|
||||||
|
variant service-error {
|
||||||
|
init-failed(string),
|
||||||
|
call-failed(string),
|
||||||
|
}
|
||||||
|
|
||||||
|
export init: func(config: list<u8>) -> result<_, service-error>;
|
||||||
|
export invoke: func(input: list<u8>) -> result<list<u8>, service-error>;
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user