feat(runtime): add resident Wasmtime actor runtime
- define versioned service and clock WIT contracts\n- enforce import allowlists, fuel, epoch, memory, I/O, and mailbox limits\n- keep one Store and Component Instance resident per serial Actor\n- add echo, counter, fault, spin, and capability probe components\n- cover lifecycle, concurrency, sandbox, and fault recovery behavior
This commit is contained in:
Maofeng
Generated
+1280
-17
File diff suppressed because it is too large
Load Diff
+11
-1
@@ -1,8 +1,15 @@
|
||||
[workspace]
|
||||
members = [
|
||||
"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"
|
||||
|
||||
[workspace.package]
|
||||
@@ -18,6 +25,9 @@ semver = "1.0.28"
|
||||
sha2 = "0.10.9"
|
||||
thiserror = "2.0.17"
|
||||
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-parser = "0.243.0"
|
||||
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