Compare commits
3 Commits
b304b038a4
...
a5718da1da
| Author | SHA1 | Date | |
|---|---|---|---|
| a5718da1da | |||
| 572fed47b4 | |||
| 949b8d6cdb |
Generated
+7
@@ -2980,6 +2980,13 @@ dependencies = [
|
|||||||
"web-sys",
|
"web-sys",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "resident-probe-component"
|
||||||
|
version = "0.1.0"
|
||||||
|
dependencies = [
|
||||||
|
"wit-bindgen 0.41.0",
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "ring"
|
name = "ring"
|
||||||
version = "0.17.14"
|
version = "0.17.14"
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ members = [
|
|||||||
"components/spin",
|
"components/spin",
|
||||||
"components/clock-probe",
|
"components/clock-probe",
|
||||||
"components/kv-probe",
|
"components/kv-probe",
|
||||||
|
"components/resident-probe",
|
||||||
"components/wasi-clock-probe",
|
"components/wasi-clock-probe",
|
||||||
]
|
]
|
||||||
default-members = ["crates/wasmeld-runtime"]
|
default-members = ["crates/wasmeld-runtime"]
|
||||||
|
|||||||
@@ -13,6 +13,22 @@ imports,只链接实际请求且版本完全匹配的 Host 能力;未知能
|
|||||||
- `wasmeld:clock/monotonic-clock@0.1.0`:Actor 内单调时钟
|
- `wasmeld:clock/monotonic-clock@0.1.0`:Actor 内单调时钟
|
||||||
- `wasmeld:kv/store@0.1.0`:按服务隔离、跨 Revision 共享的持久化二进制 KV
|
- `wasmeld:kv/store@0.1.0`:按服务隔离、跨 Revision 共享的持久化二进制 KV
|
||||||
|
|
||||||
|
常驻型 Component 可额外导出 `wasmeld:resident/actor@0.1.0`。Runtime 将定时器、
|
||||||
|
TCP/UDP/Unix、消息订阅和扩展事件放入同一个有界 Actor mailbox;Component 每次只处理
|
||||||
|
一个事件并返回 effect。系统 socket、timer task 和 broker consumer 始终由 Host 持有,
|
||||||
|
Component 只能引用当前 Revision 内不复用的资源 ID,不能直接取得文件描述符或绕过
|
||||||
|
端点策略。
|
||||||
|
|
||||||
|
```text
|
||||||
|
Host driver -> ResidentSession -> Actor mailbox -> Wasm Component
|
||||||
|
Host driver <- validated operation <- raw effect <-
|
||||||
|
```
|
||||||
|
|
||||||
|
`ResidentSession` 负责 deny-by-default 的网络策略、资源归属、数量限制、流的暂停/半关闭/
|
||||||
|
关闭状态以及 effect 批量原子校验。TCP、UDP、Unix listener 等异步驱动运行在 Actor
|
||||||
|
线程外;协议级驱动以及文件监听、串口、系统信号等其它来源通过独立版本的 WIT 能力演进,
|
||||||
|
不需要扩大基础 service world。
|
||||||
|
|
||||||
## 结构
|
## 结构
|
||||||
|
|
||||||
```text
|
```text
|
||||||
@@ -93,6 +109,16 @@ cargo run -p wasmeld-package --bin wasmeld -- \
|
|||||||
`kv-probe` 接受 `set:<key>:<value>`、`get:<key>` 和 `delete:<key>`,用于验证 Host KV
|
`kv-probe` 接受 `set:<key>:<value>`、`get:<key>` 和 `delete:<key>`,用于验证 Host KV
|
||||||
能力;它不是公开 Gateway 的业务协议。
|
能力;它不是公开 Gateway 的业务协议。
|
||||||
|
|
||||||
|
常驻事件示例组件:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
cargo run -p wasmeld-package --bin wasmeld -- \
|
||||||
|
pack components/resident-probe/Cargo.toml --locked
|
||||||
|
```
|
||||||
|
|
||||||
|
该组件同时实现基础 service world 与 resident actor export,用于验证 stream、datagram、
|
||||||
|
timer、message 和扩展 source 的事件/effect 往返。
|
||||||
|
|
||||||
发布 WIT Package:
|
发布 WIT Package:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
|
|||||||
@@ -0,0 +1,16 @@
|
|||||||
|
[package]
|
||||||
|
name = "resident-probe-component"
|
||||||
|
version = "0.1.0"
|
||||||
|
edition.workspace = true
|
||||||
|
rust-version.workspace = true
|
||||||
|
license.workspace = true
|
||||||
|
|
||||||
|
[package.metadata.wasmeld]
|
||||||
|
id = "resident-probe"
|
||||||
|
world = "component:resident-probe/resident-probe-component@0.1.0"
|
||||||
|
|
||||||
|
[lib]
|
||||||
|
crate-type = ["cdylib"]
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
wit-bindgen.workspace = true
|
||||||
@@ -0,0 +1,72 @@
|
|||||||
|
use std::sync::atomic::{AtomicU64, Ordering};
|
||||||
|
|
||||||
|
mod bindings {
|
||||||
|
wit_bindgen::generate!({
|
||||||
|
path: "wit",
|
||||||
|
world: "resident-probe-component",
|
||||||
|
generate_all,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
use bindings::exports::wasmeld::resident::actor::{
|
||||||
|
DatagramSend, Effect, Event, Guest as ResidentGuest, MessageAck, SourceCommand, StreamWrite,
|
||||||
|
TimerArm,
|
||||||
|
};
|
||||||
|
|
||||||
|
static EVENTS_HANDLED: AtomicU64 = AtomicU64::new(0);
|
||||||
|
|
||||||
|
struct ResidentProbe;
|
||||||
|
|
||||||
|
impl bindings::Guest for ResidentProbe {
|
||||||
|
fn init(_config: Vec<u8>) -> Result<(), bindings::ServiceError> {
|
||||||
|
EVENTS_HANDLED.store(0, Ordering::Relaxed);
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn invoke(_input: Vec<u8>) -> Result<Vec<u8>, bindings::ServiceError> {
|
||||||
|
Ok(EVENTS_HANDLED
|
||||||
|
.load(Ordering::Relaxed)
|
||||||
|
.to_le_bytes()
|
||||||
|
.to_vec())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ResidentGuest for ResidentProbe {
|
||||||
|
fn handle_event(
|
||||||
|
input: Event,
|
||||||
|
) -> Result<Vec<Effect>, bindings::exports::wasmeld::resident::actor::ResidentError> {
|
||||||
|
EVENTS_HANDLED.fetch_add(1, Ordering::Relaxed);
|
||||||
|
Ok(match input {
|
||||||
|
Event::StreamData(chunk) => vec![Effect::WriteStream(StreamWrite {
|
||||||
|
stream_id: chunk.stream_id,
|
||||||
|
bytes: chunk.bytes,
|
||||||
|
})],
|
||||||
|
Event::Datagram(datagram) => vec![Effect::SendDatagram(DatagramSend {
|
||||||
|
endpoint_id: datagram.endpoint_id,
|
||||||
|
peer: datagram.peer,
|
||||||
|
bytes: datagram.bytes,
|
||||||
|
})],
|
||||||
|
Event::Timer(timer) => vec![Effect::ArmTimer(TimerArm {
|
||||||
|
timer_id: timer.timer_id,
|
||||||
|
delay_ms: 10,
|
||||||
|
interval_ms: None,
|
||||||
|
})],
|
||||||
|
Event::Message(message) => vec![Effect::AcknowledgeMessage(MessageAck {
|
||||||
|
subscription_id: message.subscription_id,
|
||||||
|
message_id: message.message_id,
|
||||||
|
})],
|
||||||
|
Event::Source(source) => vec![Effect::SourceCommand(SourceCommand {
|
||||||
|
source_id: source.source_id,
|
||||||
|
command: source.kind,
|
||||||
|
payload: source.payload,
|
||||||
|
})],
|
||||||
|
Event::StreamOpened(_)
|
||||||
|
| Event::StreamWritable(_)
|
||||||
|
| Event::StreamHalfClosed(_)
|
||||||
|
| Event::StreamClosed(_)
|
||||||
|
| Event::Shutdown => Vec::new(),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
bindings::export!(ResidentProbe with_types_in bindings);
|
||||||
@@ -0,0 +1,11 @@
|
|||||||
|
schema_version = 1
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
"wasmeld:resident" = "0.1.0"
|
||||||
|
"wasmeld:service" = "0.1.0"
|
||||||
|
|
||||||
|
[replace."wasmeld:resident"]
|
||||||
|
path = "../../wit/resident"
|
||||||
|
|
||||||
|
[replace."wasmeld:service"]
|
||||||
|
path = "../../wit/service"
|
||||||
@@ -0,0 +1,15 @@
|
|||||||
|
schema_version = 1
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "wasmeld:resident"
|
||||||
|
version = "0.1.0"
|
||||||
|
source = "path+../../wit/resident"
|
||||||
|
sha256 = "683a3c7196a6dd77b3f619227aa093430ff6995c097d39bb82d501b7c59a196c"
|
||||||
|
replaced = true
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "wasmeld:service"
|
||||||
|
version = "0.1.0"
|
||||||
|
source = "path+../../wit/service"
|
||||||
|
sha256 = "d5497307bbcd1e159f7707f385b488a6f2e26362d5256d5bc1080ac603a51305"
|
||||||
|
replaced = true
|
||||||
@@ -0,0 +1,6 @@
|
|||||||
|
package component:resident-probe@0.1.0;
|
||||||
|
|
||||||
|
world resident-probe-component {
|
||||||
|
include wasmeld:service/service-component@0.1.0;
|
||||||
|
export wasmeld:resident/actor@0.1.0;
|
||||||
|
}
|
||||||
@@ -156,6 +156,7 @@ url = "http://127.0.0.1:8080"
|
|||||||
[dependencies]
|
[dependencies]
|
||||||
"wasmeld:service" = "0.1.0"
|
"wasmeld:service" = "0.1.0"
|
||||||
"wasmeld:kv" = "0.1.0"
|
"wasmeld:kv" = "0.1.0"
|
||||||
|
"wasmeld:resident" = "0.1.0"
|
||||||
|
|
||||||
[replace."wasmeld:service"]
|
[replace."wasmeld:service"]
|
||||||
path = "../../wit/service"
|
path = "../../wit/service"
|
||||||
@@ -164,6 +165,11 @@ path = "../../wit/service"
|
|||||||
依赖解析结果写入 `wit.lock`,源码开发时应提交该文件。解析出的 `wit/deps/` 是生成目录,
|
依赖解析结果写入 `wit.lock`,源码开发时应提交该文件。解析出的 `wit/deps/` 是生成目录,
|
||||||
不应提交。
|
不应提交。
|
||||||
|
|
||||||
|
普通请求型 Component 只需要 `wasmeld:service`。需要接收 Host 定时器、网络流、UDP、
|
||||||
|
消息订阅或扩展 source 事件时,再依赖并导出 `wasmeld:resident/actor@0.1.0`;例如
|
||||||
|
`components/resident-probe`。`resident` 是独立版本化 package,不会把所有 Host 能力
|
||||||
|
合并到基础 service world。
|
||||||
|
|
||||||
同步依赖:
|
同步依赖:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
|
|||||||
@@ -5,3 +5,11 @@ wasmtime::component::bindgen!({
|
|||||||
path: "../../wit/service",
|
path: "../../wit/service",
|
||||||
world: "service-component",
|
world: "service-component",
|
||||||
});
|
});
|
||||||
|
|
||||||
|
/// Optional exports implemented only by active resident Components.
|
||||||
|
pub(crate) mod resident {
|
||||||
|
wasmtime::component::bindgen!({
|
||||||
|
path: "../../wit/resident",
|
||||||
|
world: "resident-component",
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|||||||
@@ -68,6 +68,18 @@ pub enum RuntimeError {
|
|||||||
#[error("actor for {0} stopped before returning a result")]
|
#[error("actor for {0} stopped before returning a result")]
|
||||||
ActorStopped(ServiceKey),
|
ActorStopped(ServiceKey),
|
||||||
|
|
||||||
|
#[error("service revision {0} does not export the resident actor interface")]
|
||||||
|
NotResident(ServiceKey),
|
||||||
|
|
||||||
|
#[error("resident event for {service} exceeds the {limit}-byte payload limit")]
|
||||||
|
ResidentEventTooLarge { service: ServiceKey, limit: usize },
|
||||||
|
|
||||||
|
#[error("resident event for {service} returned more than {limit} effects")]
|
||||||
|
TooManyResidentEffects { service: ServiceKey, limit: usize },
|
||||||
|
|
||||||
|
#[error("invalid resident effect: {0}")]
|
||||||
|
InvalidResidentEffect(String),
|
||||||
|
|
||||||
#[error("input for {service} exceeds the {limit}-byte limit")]
|
#[error("input for {service} exceeds the {limit}-byte limit")]
|
||||||
InputTooLarge { service: ServiceKey, limit: usize },
|
InputTooLarge { service: ServiceKey, limit: usize },
|
||||||
|
|
||||||
|
|||||||
@@ -9,6 +9,8 @@ mod bindings;
|
|||||||
mod capability;
|
mod capability;
|
||||||
mod error;
|
mod error;
|
||||||
mod manifest;
|
mod manifest;
|
||||||
|
mod resident;
|
||||||
|
mod resident_host;
|
||||||
mod runtime;
|
mod runtime;
|
||||||
|
|
||||||
pub use capability::{
|
pub use capability::{
|
||||||
@@ -16,5 +18,14 @@ pub use capability::{
|
|||||||
};
|
};
|
||||||
pub use error::RuntimeError;
|
pub use error::RuntimeError;
|
||||||
pub use manifest::{ResourceLimits, ServiceKey, ServiceManifest};
|
pub use manifest::{ResourceLimits, ServiceKey, ServiceManifest};
|
||||||
|
pub use resident::{
|
||||||
|
ComponentExecution, ResidentEffect, ResidentEvent, ResidentLimits, ResourceId,
|
||||||
|
StreamCloseReason,
|
||||||
|
};
|
||||||
|
pub use resident_host::{
|
||||||
|
NetworkScope, ResidentEndpoint, ResidentHostError, ResidentOperation, ResidentPolicy,
|
||||||
|
ResidentResourceInfo, ResidentResourceKind, ResidentResourceMetadata, ResidentResourceState,
|
||||||
|
ResidentSession,
|
||||||
|
};
|
||||||
pub use runtime::{ActorHandle, Runtime, RuntimeConfig, RuntimeStats};
|
pub use runtime::{ActorHandle, Runtime, RuntimeConfig, RuntimeStats};
|
||||||
pub use wasmeld_package::SERVICE_WORLD;
|
pub use wasmeld_package::SERVICE_WORLD;
|
||||||
|
|||||||
@@ -4,7 +4,7 @@ use std::{fmt, time::Duration};
|
|||||||
|
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
use crate::error::RuntimeError;
|
use crate::{error::RuntimeError, resident::ResidentLimits};
|
||||||
|
|
||||||
/// Immutable identity of one service revision.
|
/// Immutable identity of one service revision.
|
||||||
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
|
#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
|
||||||
@@ -103,6 +103,9 @@ pub struct ResourceLimits {
|
|||||||
pub max_input_bytes: usize,
|
pub max_input_bytes: usize,
|
||||||
/// Maximum response payload size.
|
/// Maximum response payload size.
|
||||||
pub max_output_bytes: usize,
|
pub max_output_bytes: usize,
|
||||||
|
/// Limits for active events and Host-owned resources.
|
||||||
|
#[serde(default)]
|
||||||
|
pub resident: ResidentLimits,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Default for ResourceLimits {
|
impl Default for ResourceLimits {
|
||||||
@@ -114,6 +117,7 @@ impl Default for ResourceLimits {
|
|||||||
mailbox_capacity: 16,
|
mailbox_capacity: 16,
|
||||||
max_input_bytes: 16 * 1024,
|
max_input_bytes: 16 * 1024,
|
||||||
max_output_bytes: 16 * 1024,
|
max_output_bytes: 16 * 1024,
|
||||||
|
resident: ResidentLimits::default(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -151,7 +155,7 @@ impl ResourceLimits {
|
|||||||
));
|
));
|
||||||
}
|
}
|
||||||
|
|
||||||
Ok(())
|
self.resident.validate()
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Returns the configured call deadline.
|
/// Returns the configured call deadline.
|
||||||
|
|||||||
@@ -0,0 +1,317 @@
|
|||||||
|
//! Host-owned event and effect protocol for active resident Components.
|
||||||
|
//!
|
||||||
|
//! A [`ResidentEvent`] is always produced by a Host driver and delivered
|
||||||
|
//! through the same bounded Actor mailbox as ordinary service invocations.
|
||||||
|
//! The Component never receives an operating-system descriptor. It can only
|
||||||
|
//! reference opaque [`ResourceId`] values and return [`ResidentEffect`] values
|
||||||
|
//! for the owning driver to validate and apply.
|
||||||
|
|
||||||
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
|
use crate::{RuntimeError, bindings::resident::exports::wasmeld::resident::actor as wit};
|
||||||
|
|
||||||
|
pub(crate) const RESIDENT_INTERFACE: &str = "wasmeld:resident/actor@0.1.0";
|
||||||
|
|
||||||
|
/// Component execution model discovered from its exported WIT interfaces.
|
||||||
|
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||||
|
#[serde(rename_all = "snake_case")]
|
||||||
|
pub enum ComponentExecution {
|
||||||
|
/// Retains memory but runs only for `invoke` mailbox commands.
|
||||||
|
Service,
|
||||||
|
/// Also accepts Host events and returns effects for Host-owned resources.
|
||||||
|
Resident,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Opaque Host resource identity scoped to one resident Actor revision.
|
||||||
|
///
|
||||||
|
/// IDs are never reused while the Actor is alive. `0` is reserved so an
|
||||||
|
/// uninitialized ID cannot accidentally address a real resource.
|
||||||
|
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
|
||||||
|
#[serde(transparent)]
|
||||||
|
pub struct ResourceId(u64);
|
||||||
|
|
||||||
|
impl ResourceId {
|
||||||
|
/// Creates a valid opaque resource identity.
|
||||||
|
pub fn new(value: u64) -> Option<Self> {
|
||||||
|
(value != 0).then_some(Self(value))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Returns the WIT-compatible integer representation.
|
||||||
|
pub const fn get(self) -> u64 {
|
||||||
|
self.0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Why a Host-owned stream stopped accepting events and effects.
|
||||||
|
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||||
|
#[serde(rename_all = "snake_case")]
|
||||||
|
pub enum StreamCloseReason {
|
||||||
|
PeerClosed,
|
||||||
|
HostClosed,
|
||||||
|
IdleTimeout,
|
||||||
|
ProtocolError,
|
||||||
|
TransportError,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// One bounded event delivered to a resident Component.
|
||||||
|
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||||
|
#[serde(tag = "kind", rename_all = "snake_case")]
|
||||||
|
pub enum ResidentEvent {
|
||||||
|
Timer {
|
||||||
|
timer_id: ResourceId,
|
||||||
|
scheduled_at_ns: u64,
|
||||||
|
},
|
||||||
|
StreamOpened {
|
||||||
|
endpoint_id: ResourceId,
|
||||||
|
stream_id: ResourceId,
|
||||||
|
peer: Option<String>,
|
||||||
|
},
|
||||||
|
StreamData {
|
||||||
|
stream_id: ResourceId,
|
||||||
|
bytes: Vec<u8>,
|
||||||
|
},
|
||||||
|
StreamWritable {
|
||||||
|
stream_id: ResourceId,
|
||||||
|
},
|
||||||
|
StreamHalfClosed {
|
||||||
|
stream_id: ResourceId,
|
||||||
|
},
|
||||||
|
StreamClosed {
|
||||||
|
stream_id: ResourceId,
|
||||||
|
reason: StreamCloseReason,
|
||||||
|
},
|
||||||
|
Datagram {
|
||||||
|
endpoint_id: ResourceId,
|
||||||
|
peer: String,
|
||||||
|
bytes: Vec<u8>,
|
||||||
|
},
|
||||||
|
Message {
|
||||||
|
subscription_id: ResourceId,
|
||||||
|
message_id: u64,
|
||||||
|
bytes: Vec<u8>,
|
||||||
|
},
|
||||||
|
Source {
|
||||||
|
source_id: ResourceId,
|
||||||
|
event: String,
|
||||||
|
payload: Vec<u8>,
|
||||||
|
},
|
||||||
|
Shutdown,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A requested operation returned by a resident Component.
|
||||||
|
///
|
||||||
|
/// Effects are inert until the Host supervisor checks resource ownership,
|
||||||
|
/// payload limits, current stream state, and driver policy.
|
||||||
|
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||||
|
#[serde(tag = "kind", rename_all = "snake_case")]
|
||||||
|
pub enum ResidentEffect {
|
||||||
|
WriteStream {
|
||||||
|
stream_id: ResourceId,
|
||||||
|
bytes: Vec<u8>,
|
||||||
|
},
|
||||||
|
CloseStream {
|
||||||
|
stream_id: ResourceId,
|
||||||
|
},
|
||||||
|
PauseStream {
|
||||||
|
stream_id: ResourceId,
|
||||||
|
},
|
||||||
|
ResumeStream {
|
||||||
|
stream_id: ResourceId,
|
||||||
|
},
|
||||||
|
SendDatagram {
|
||||||
|
endpoint_id: ResourceId,
|
||||||
|
peer: String,
|
||||||
|
bytes: Vec<u8>,
|
||||||
|
},
|
||||||
|
ArmTimer {
|
||||||
|
timer_id: ResourceId,
|
||||||
|
delay_ms: u64,
|
||||||
|
interval_ms: Option<u64>,
|
||||||
|
},
|
||||||
|
CancelTimer {
|
||||||
|
timer_id: ResourceId,
|
||||||
|
},
|
||||||
|
AcknowledgeMessage {
|
||||||
|
subscription_id: ResourceId,
|
||||||
|
message_id: u64,
|
||||||
|
},
|
||||||
|
SourceCommand {
|
||||||
|
source_id: ResourceId,
|
||||||
|
command: String,
|
||||||
|
payload: Vec<u8>,
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Execution limits specific to active resident Components.
|
||||||
|
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
|
||||||
|
pub struct ResidentLimits {
|
||||||
|
/// Maximum number of Host resources owned by one Actor revision.
|
||||||
|
pub max_resources: usize,
|
||||||
|
/// Maximum effects accepted from one event handler call.
|
||||||
|
pub max_effects_per_event: usize,
|
||||||
|
/// Maximum bytes in one stream read or write.
|
||||||
|
pub max_stream_chunk_bytes: usize,
|
||||||
|
/// Maximum bytes in one received or sent datagram.
|
||||||
|
pub max_datagram_bytes: usize,
|
||||||
|
/// Minimum delay accepted for Component-created timers.
|
||||||
|
pub min_timer_delay_ms: u64,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Default for ResidentLimits {
|
||||||
|
fn default() -> Self {
|
||||||
|
Self {
|
||||||
|
max_resources: 1024,
|
||||||
|
max_effects_per_event: 64,
|
||||||
|
max_stream_chunk_bytes: 64 * 1024,
|
||||||
|
max_datagram_bytes: 64 * 1024,
|
||||||
|
min_timer_delay_ms: 1,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ResidentLimits {
|
||||||
|
pub(crate) fn validate(&self) -> Result<(), RuntimeError> {
|
||||||
|
if self.max_resources == 0 {
|
||||||
|
return Err(invalid_limit("max_resources"));
|
||||||
|
}
|
||||||
|
if self.max_effects_per_event == 0 {
|
||||||
|
return Err(invalid_limit("max_effects_per_event"));
|
||||||
|
}
|
||||||
|
if self.max_stream_chunk_bytes == 0 {
|
||||||
|
return Err(invalid_limit("max_stream_chunk_bytes"));
|
||||||
|
}
|
||||||
|
if self.max_datagram_bytes == 0 {
|
||||||
|
return Err(invalid_limit("max_datagram_bytes"));
|
||||||
|
}
|
||||||
|
if self.min_timer_delay_ms == 0 {
|
||||||
|
return Err(invalid_limit("min_timer_delay_ms"));
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) fn into_wit_event(event: ResidentEvent) -> wit::Event {
|
||||||
|
match event {
|
||||||
|
ResidentEvent::Timer {
|
||||||
|
timer_id,
|
||||||
|
scheduled_at_ns,
|
||||||
|
} => wit::Event::Timer(wit::TimerFired {
|
||||||
|
timer_id: timer_id.get(),
|
||||||
|
scheduled_at_ns,
|
||||||
|
}),
|
||||||
|
ResidentEvent::StreamOpened {
|
||||||
|
endpoint_id,
|
||||||
|
stream_id,
|
||||||
|
peer,
|
||||||
|
} => wit::Event::StreamOpened(wit::StreamOpened {
|
||||||
|
endpoint_id: endpoint_id.get(),
|
||||||
|
stream_id: stream_id.get(),
|
||||||
|
peer,
|
||||||
|
}),
|
||||||
|
ResidentEvent::StreamData { stream_id, bytes } => {
|
||||||
|
wit::Event::StreamData(wit::StreamChunk {
|
||||||
|
stream_id: stream_id.get(),
|
||||||
|
bytes,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
ResidentEvent::StreamWritable { stream_id } => wit::Event::StreamWritable(stream_id.get()),
|
||||||
|
ResidentEvent::StreamHalfClosed { stream_id } => {
|
||||||
|
wit::Event::StreamHalfClosed(stream_id.get())
|
||||||
|
}
|
||||||
|
ResidentEvent::StreamClosed { stream_id, reason } => {
|
||||||
|
wit::Event::StreamClosed(wit::StreamClosed {
|
||||||
|
stream_id: stream_id.get(),
|
||||||
|
reason: match reason {
|
||||||
|
StreamCloseReason::PeerClosed => wit::CloseReason::PeerClosed,
|
||||||
|
StreamCloseReason::HostClosed => wit::CloseReason::HostClosed,
|
||||||
|
StreamCloseReason::IdleTimeout => wit::CloseReason::IdleTimeout,
|
||||||
|
StreamCloseReason::ProtocolError => wit::CloseReason::ProtocolError,
|
||||||
|
StreamCloseReason::TransportError => wit::CloseReason::TransportError,
|
||||||
|
},
|
||||||
|
})
|
||||||
|
}
|
||||||
|
ResidentEvent::Datagram {
|
||||||
|
endpoint_id,
|
||||||
|
peer,
|
||||||
|
bytes,
|
||||||
|
} => wit::Event::Datagram(wit::Datagram {
|
||||||
|
endpoint_id: endpoint_id.get(),
|
||||||
|
peer,
|
||||||
|
bytes,
|
||||||
|
}),
|
||||||
|
ResidentEvent::Message {
|
||||||
|
subscription_id,
|
||||||
|
message_id,
|
||||||
|
bytes,
|
||||||
|
} => wit::Event::Message(wit::Message {
|
||||||
|
subscription_id: subscription_id.get(),
|
||||||
|
message_id,
|
||||||
|
bytes,
|
||||||
|
}),
|
||||||
|
ResidentEvent::Source {
|
||||||
|
source_id,
|
||||||
|
event,
|
||||||
|
payload,
|
||||||
|
} => wit::Event::Source(wit::SourceEvent {
|
||||||
|
source_id: source_id.get(),
|
||||||
|
kind: event,
|
||||||
|
payload,
|
||||||
|
}),
|
||||||
|
ResidentEvent::Shutdown => wit::Event::Shutdown,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) fn from_wit_effects(
|
||||||
|
effects: Vec<wit::Effect>,
|
||||||
|
) -> Result<Vec<ResidentEffect>, RuntimeError> {
|
||||||
|
effects.into_iter().map(from_wit_effect).collect()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn from_wit_effect(effect: wit::Effect) -> Result<ResidentEffect, RuntimeError> {
|
||||||
|
Ok(match effect {
|
||||||
|
wit::Effect::WriteStream(effect) => ResidentEffect::WriteStream {
|
||||||
|
stream_id: resource_id(effect.stream_id)?,
|
||||||
|
bytes: effect.bytes,
|
||||||
|
},
|
||||||
|
wit::Effect::CloseStream(stream_id) => ResidentEffect::CloseStream {
|
||||||
|
stream_id: resource_id(stream_id)?,
|
||||||
|
},
|
||||||
|
wit::Effect::PauseStream(stream_id) => ResidentEffect::PauseStream {
|
||||||
|
stream_id: resource_id(stream_id)?,
|
||||||
|
},
|
||||||
|
wit::Effect::ResumeStream(stream_id) => ResidentEffect::ResumeStream {
|
||||||
|
stream_id: resource_id(stream_id)?,
|
||||||
|
},
|
||||||
|
wit::Effect::SendDatagram(effect) => ResidentEffect::SendDatagram {
|
||||||
|
endpoint_id: resource_id(effect.endpoint_id)?,
|
||||||
|
peer: effect.peer,
|
||||||
|
bytes: effect.bytes,
|
||||||
|
},
|
||||||
|
wit::Effect::ArmTimer(effect) => ResidentEffect::ArmTimer {
|
||||||
|
timer_id: resource_id(effect.timer_id)?,
|
||||||
|
delay_ms: effect.delay_ms,
|
||||||
|
interval_ms: effect.interval_ms,
|
||||||
|
},
|
||||||
|
wit::Effect::CancelTimer(timer_id) => ResidentEffect::CancelTimer {
|
||||||
|
timer_id: resource_id(timer_id)?,
|
||||||
|
},
|
||||||
|
wit::Effect::AcknowledgeMessage(effect) => ResidentEffect::AcknowledgeMessage {
|
||||||
|
subscription_id: resource_id(effect.subscription_id)?,
|
||||||
|
message_id: effect.message_id,
|
||||||
|
},
|
||||||
|
wit::Effect::SourceCommand(effect) => ResidentEffect::SourceCommand {
|
||||||
|
source_id: resource_id(effect.source_id)?,
|
||||||
|
command: effect.command,
|
||||||
|
payload: effect.payload,
|
||||||
|
},
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
fn resource_id(value: u64) -> Result<ResourceId, RuntimeError> {
|
||||||
|
ResourceId::new(value)
|
||||||
|
.ok_or_else(|| RuntimeError::InvalidResidentEffect("resource ID 0 is reserved".to_owned()))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn invalid_limit(field: &str) -> RuntimeError {
|
||||||
|
RuntimeError::InvalidManifest(format!("limits.resident.{field} must be greater than zero"))
|
||||||
|
}
|
||||||
File diff suppressed because it is too large
Load Diff
@@ -31,10 +31,15 @@ use wasmtime_wasi::{
|
|||||||
};
|
};
|
||||||
|
|
||||||
use crate::{
|
use crate::{
|
||||||
CapabilityDescriptor, KvBackend, RuntimeError, ServiceKey, ServiceManifest,
|
CapabilityDescriptor, ComponentExecution, KvBackend, ResidentEffect, ResidentEvent,
|
||||||
bindings::{ServiceComponent, ServiceError},
|
RuntimeError, ServiceKey, ServiceManifest,
|
||||||
|
bindings::{
|
||||||
|
ServiceComponent, ServiceError,
|
||||||
|
resident::{ResidentComponent, exports::wasmeld::resident::actor::ResidentError},
|
||||||
|
},
|
||||||
capability::{Capability, CapabilityRegistry, HostCapabilities},
|
capability::{Capability, CapabilityRegistry, HostCapabilities},
|
||||||
manifest::ResourceLimits,
|
manifest::ResourceLimits,
|
||||||
|
resident::{RESIDENT_INTERFACE, from_wit_effects, into_wit_event},
|
||||||
};
|
};
|
||||||
|
|
||||||
const DEFAULT_EPOCH_TICK: Duration = Duration::from_millis(5);
|
const DEFAULT_EPOCH_TICK: Duration = Duration::from_millis(5);
|
||||||
@@ -171,6 +176,7 @@ struct RegisteredService {
|
|||||||
manifest: ServiceManifest,
|
manifest: ServiceManifest,
|
||||||
component: Arc<Component>,
|
component: Arc<Component>,
|
||||||
capabilities: Vec<Capability>,
|
capabilities: Vec<Capability>,
|
||||||
|
execution: ComponentExecution,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Runtime {
|
impl Runtime {
|
||||||
@@ -231,6 +237,7 @@ impl Runtime {
|
|||||||
let component = Component::new(&self.inner.engine, component_bytes.as_ref())
|
let component = Component::new(&self.inner.engine, component_bytes.as_ref())
|
||||||
.map_err(RuntimeError::ComponentCompilation)?;
|
.map_err(RuntimeError::ComponentCompilation)?;
|
||||||
let capabilities = validate_component_imports(&component, &self.inner.engine)?;
|
let capabilities = validate_component_imports(&component, &self.inner.engine)?;
|
||||||
|
let execution = component_execution(&component, &self.inner.engine);
|
||||||
if capabilities.contains(&Capability::KvStore) && self.inner.kv_backend.is_none() {
|
if capabilities.contains(&Capability::KvStore) && self.inner.kv_backend.is_none() {
|
||||||
return Err(RuntimeError::CapabilityUnavailable(
|
return Err(RuntimeError::CapabilityUnavailable(
|
||||||
Capability::KvStore.descriptor().interface().to_owned(),
|
Capability::KvStore.descriptor().interface().to_owned(),
|
||||||
@@ -253,6 +260,7 @@ impl Runtime {
|
|||||||
manifest,
|
manifest,
|
||||||
component: Arc::new(component),
|
component: Arc::new(component),
|
||||||
capabilities,
|
capabilities,
|
||||||
|
execution,
|
||||||
},
|
},
|
||||||
);
|
);
|
||||||
|
|
||||||
@@ -372,6 +380,34 @@ impl Runtime {
|
|||||||
.collect())
|
.collect())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Returns whether a registered Component is passive or event-driven.
|
||||||
|
pub fn execution(&self, key: &ServiceKey) -> Result<ComponentExecution, RuntimeError> {
|
||||||
|
self.inner
|
||||||
|
.services
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| RuntimeError::LockPoisoned)?
|
||||||
|
.get(key)
|
||||||
|
.map(|service| service.execution)
|
||||||
|
.ok_or_else(|| RuntimeError::ServiceNotRegistered(key.clone()))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Delivers one Host event through the Actor's bounded serial mailbox.
|
||||||
|
pub fn dispatch_event(
|
||||||
|
&self,
|
||||||
|
key: &ServiceKey,
|
||||||
|
event: ResidentEvent,
|
||||||
|
) -> Result<Vec<ResidentEffect>, RuntimeError> {
|
||||||
|
let actor = self
|
||||||
|
.inner
|
||||||
|
.actors
|
||||||
|
.lock()
|
||||||
|
.map_err(|_| RuntimeError::LockPoisoned)?
|
||||||
|
.get(key)
|
||||||
|
.cloned()
|
||||||
|
.ok_or_else(|| RuntimeError::ActorUnavailable(key.clone()))?;
|
||||||
|
actor.dispatch_event(event)
|
||||||
|
}
|
||||||
|
|
||||||
/// Stops one Actor and releases its Store and Component Instance.
|
/// Stops one Actor and releases its Store and Component Instance.
|
||||||
pub fn stop(&self, key: &ServiceKey) -> Result<(), RuntimeError> {
|
pub fn stop(&self, key: &ServiceKey) -> Result<(), RuntimeError> {
|
||||||
let _lifecycle = self
|
let _lifecycle = self
|
||||||
@@ -528,6 +564,7 @@ impl Runtime {
|
|||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
pub struct ActorHandle {
|
pub struct ActorHandle {
|
||||||
key: ServiceKey,
|
key: ServiceKey,
|
||||||
|
execution: ComponentExecution,
|
||||||
limits: Arc<ResourceLimits>,
|
limits: Arc<ResourceLimits>,
|
||||||
sender: SyncSender<ActorCommand>,
|
sender: SyncSender<ActorCommand>,
|
||||||
status: Arc<ActorStatus>,
|
status: Arc<ActorStatus>,
|
||||||
@@ -540,6 +577,15 @@ impl ActorHandle {
|
|||||||
&self.key
|
&self.key
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Returns the execution model detected from the Component exports.
|
||||||
|
pub fn execution(&self) -> ComponentExecution {
|
||||||
|
self.execution
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) fn resident_resource_limit(&self) -> usize {
|
||||||
|
self.limits.resident.max_resources
|
||||||
|
}
|
||||||
|
|
||||||
/// Sends an invocation and waits up to the remaining service deadline.
|
/// Sends an invocation and waits up to the remaining service deadline.
|
||||||
pub fn invoke(&self, input: Vec<u8>) -> Result<Vec<u8>, RuntimeError> {
|
pub fn invoke(&self, input: Vec<u8>) -> Result<Vec<u8>, RuntimeError> {
|
||||||
if !self.is_available() {
|
if !self.is_available() {
|
||||||
@@ -583,6 +629,45 @@ impl ActorHandle {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Sends a Host-owned resource event and waits for validated effects.
|
||||||
|
pub fn dispatch_event(
|
||||||
|
&self,
|
||||||
|
event: ResidentEvent,
|
||||||
|
) -> Result<Vec<ResidentEffect>, RuntimeError> {
|
||||||
|
if !self.is_available() {
|
||||||
|
return Err(RuntimeError::ActorUnavailable(self.key.clone()));
|
||||||
|
}
|
||||||
|
validate_resident_event(&self.key, &self.limits, &event)?;
|
||||||
|
|
||||||
|
let (response_sender, response_receiver) = mpsc::sync_channel(1);
|
||||||
|
let deadline = Instant::now() + self.limits.deadline();
|
||||||
|
let command = ActorCommand::ResidentEvent {
|
||||||
|
event,
|
||||||
|
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.
|
/// Prevents new calls, asks the worker to stop, and waits for acknowledgement.
|
||||||
pub fn stop(&self) -> Result<(), RuntimeError> {
|
pub fn stop(&self) -> Result<(), RuntimeError> {
|
||||||
if !self.status.accepting.swap(false, Ordering::AcqRel) {
|
if !self.status.accepting.swap(false, Ordering::AcqRel) {
|
||||||
@@ -622,6 +707,11 @@ enum ActorCommand {
|
|||||||
deadline: Instant,
|
deadline: Instant,
|
||||||
response_sender: SyncSender<Result<Vec<u8>, RuntimeError>>,
|
response_sender: SyncSender<Result<Vec<u8>, RuntimeError>>,
|
||||||
},
|
},
|
||||||
|
ResidentEvent {
|
||||||
|
event: ResidentEvent,
|
||||||
|
deadline: Instant,
|
||||||
|
response_sender: SyncSender<Result<Vec<ResidentEffect>, RuntimeError>>,
|
||||||
|
},
|
||||||
Stop {
|
Stop {
|
||||||
response_sender: SyncSender<()>,
|
response_sender: SyncSender<()>,
|
||||||
},
|
},
|
||||||
@@ -719,6 +809,7 @@ struct ActorWorker {
|
|||||||
epoch_tick: Duration,
|
epoch_tick: Duration,
|
||||||
store: Store<HostState>,
|
store: Store<HostState>,
|
||||||
bindings: ServiceComponent,
|
bindings: ServiceComponent,
|
||||||
|
resident: Option<ResidentComponent>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl ActorWorker {
|
impl ActorWorker {
|
||||||
@@ -750,8 +841,18 @@ impl ActorWorker {
|
|||||||
|
|
||||||
let epoch_ticks = ticks_for(limits.deadline(), epoch_tick);
|
let epoch_ticks = ticks_for(limits.deadline(), epoch_tick);
|
||||||
configure_call_budget(&mut store, &limits, epoch_ticks)?;
|
configure_call_budget(&mut store, &limits, epoch_ticks)?;
|
||||||
let bindings = ServiceComponent::instantiate(&mut store, &service.component, &linker)
|
let instance = linker
|
||||||
|
.instantiate(&mut store, &service.component)
|
||||||
.map_err(|error| RuntimeError::ActorInitialization(error.to_string()))?;
|
.map_err(|error| RuntimeError::ActorInitialization(error.to_string()))?;
|
||||||
|
let bindings = ServiceComponent::new(&mut store, &instance)
|
||||||
|
.map_err(|error| RuntimeError::ActorInitialization(error.to_string()))?;
|
||||||
|
let resident = match service.execution {
|
||||||
|
ComponentExecution::Service => None,
|
||||||
|
ComponentExecution::Resident => Some(
|
||||||
|
ResidentComponent::new(&mut store, &instance)
|
||||||
|
.map_err(|error| RuntimeError::ActorInitialization(error.to_string()))?,
|
||||||
|
),
|
||||||
|
};
|
||||||
|
|
||||||
configure_call_budget(&mut store, &limits, epoch_ticks)?;
|
configure_call_budget(&mut store, &limits, epoch_ticks)?;
|
||||||
|
|
||||||
@@ -771,6 +872,7 @@ impl ActorWorker {
|
|||||||
epoch_tick,
|
epoch_tick,
|
||||||
store,
|
store,
|
||||||
bindings,
|
bindings,
|
||||||
|
resident,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -807,6 +909,54 @@ impl ActorWorker {
|
|||||||
}),
|
}),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn dispatch_event(
|
||||||
|
&mut self,
|
||||||
|
event: ResidentEvent,
|
||||||
|
deadline: Instant,
|
||||||
|
) -> Result<Vec<ResidentEffect>, 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(),
|
||||||
|
});
|
||||||
|
}
|
||||||
|
let resident = self
|
||||||
|
.resident
|
||||||
|
.as_ref()
|
||||||
|
.ok_or_else(|| RuntimeError::NotResident(self.key.clone()))?;
|
||||||
|
configure_call_budget(
|
||||||
|
&mut self.store,
|
||||||
|
&self.limits,
|
||||||
|
ticks_for(remaining, self.epoch_tick),
|
||||||
|
)?;
|
||||||
|
|
||||||
|
match resident
|
||||||
|
.wasmeld_resident_actor()
|
||||||
|
.call_handle_event(&mut self.store, &into_wit_event(event))
|
||||||
|
{
|
||||||
|
Ok(Ok(effects)) => {
|
||||||
|
if effects.len() > self.limits.resident.max_effects_per_event {
|
||||||
|
return Err(RuntimeError::TooManyResidentEffects {
|
||||||
|
service: self.key.clone(),
|
||||||
|
limit: self.limits.resident.max_effects_per_event,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
let effects = from_wit_effects(effects)?;
|
||||||
|
validate_resident_effects(&self.key, &self.limits, &effects)?;
|
||||||
|
Ok(effects)
|
||||||
|
}
|
||||||
|
Ok(Err(ResidentError::EventFailed(message))) => Err(RuntimeError::ComponentError {
|
||||||
|
kind: "resident-event",
|
||||||
|
message,
|
||||||
|
}),
|
||||||
|
Err(error) => Err(RuntimeError::ActorFault {
|
||||||
|
service: self.key.clone(),
|
||||||
|
message: error.to_string(),
|
||||||
|
}),
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
fn add_restricted_wasi(linker: &mut Linker<HostState>) -> Result<(), RuntimeError> {
|
fn add_restricted_wasi(linker: &mut Linker<HostState>) -> Result<(), RuntimeError> {
|
||||||
@@ -883,6 +1033,18 @@ fn validate_component_imports(
|
|||||||
Ok(capabilities)
|
Ok(capabilities)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn component_execution(component: &Component, engine: &Engine) -> ComponentExecution {
|
||||||
|
if component
|
||||||
|
.component_type()
|
||||||
|
.exports(engine)
|
||||||
|
.any(|(name, _)| name == RESIDENT_INTERFACE)
|
||||||
|
{
|
||||||
|
ComponentExecution::Resident
|
||||||
|
} else {
|
||||||
|
ComponentExecution::Service
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fn spawn_actor(
|
fn spawn_actor(
|
||||||
engine: Arc<Engine>,
|
engine: Arc<Engine>,
|
||||||
ticker: Arc<EpochTicker>,
|
ticker: Arc<EpochTicker>,
|
||||||
@@ -892,6 +1054,7 @@ fn spawn_actor(
|
|||||||
init_config: Vec<u8>,
|
init_config: Vec<u8>,
|
||||||
kv_backend: Option<Arc<dyn KvBackend>>,
|
kv_backend: Option<Arc<dyn KvBackend>>,
|
||||||
) -> Result<ActorHandle, RuntimeError> {
|
) -> Result<ActorHandle, RuntimeError> {
|
||||||
|
let execution = service.execution;
|
||||||
let limits = Arc::new(service.manifest.limits.clone());
|
let limits = Arc::new(service.manifest.limits.clone());
|
||||||
let (sender, receiver) = mpsc::sync_channel(limits.mailbox_capacity);
|
let (sender, receiver) = mpsc::sync_channel(limits.mailbox_capacity);
|
||||||
let (ready_sender, ready_receiver) = mpsc::sync_channel(1);
|
let (ready_sender, ready_receiver) = mpsc::sync_channel(1);
|
||||||
@@ -925,6 +1088,7 @@ fn spawn_actor(
|
|||||||
|
|
||||||
Ok(ActorHandle {
|
Ok(ActorHandle {
|
||||||
key: thread_key,
|
key: thread_key,
|
||||||
|
execution,
|
||||||
limits,
|
limits,
|
||||||
sender,
|
sender,
|
||||||
status,
|
status,
|
||||||
@@ -961,6 +1125,29 @@ fn run_actor(worker: &mut ActorWorker, receiver: Receiver<ActorCommand>, status:
|
|||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
ActorCommand::ResidentEvent {
|
||||||
|
event,
|
||||||
|
deadline,
|
||||||
|
response_sender,
|
||||||
|
} => {
|
||||||
|
let result = if Instant::now() >= deadline {
|
||||||
|
Err(RuntimeError::DeadlineExceeded {
|
||||||
|
service: worker.key.clone(),
|
||||||
|
deadline: worker.limits.deadline(),
|
||||||
|
})
|
||||||
|
} else {
|
||||||
|
worker.dispatch_event(event, deadline)
|
||||||
|
};
|
||||||
|
let fatal = is_fatal_resident(&result);
|
||||||
|
if fatal {
|
||||||
|
status.mark_stopped();
|
||||||
|
}
|
||||||
|
let _ = response_sender.send(result);
|
||||||
|
if fatal {
|
||||||
|
drain_after_failure(receiver, &worker.key);
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
}
|
||||||
ActorCommand::Stop { response_sender } => {
|
ActorCommand::Stop { response_sender } => {
|
||||||
status.mark_stopped();
|
status.mark_stopped();
|
||||||
let _ = response_sender.send(());
|
let _ = response_sender.send(());
|
||||||
@@ -980,6 +1167,11 @@ fn drain_after_failure(receiver: Receiver<ActorCommand>, key: &ServiceKey) {
|
|||||||
} => {
|
} => {
|
||||||
let _ = response_sender.send(Err(RuntimeError::ActorUnavailable(key.clone())));
|
let _ = response_sender.send(Err(RuntimeError::ActorUnavailable(key.clone())));
|
||||||
}
|
}
|
||||||
|
ActorCommand::ResidentEvent {
|
||||||
|
response_sender, ..
|
||||||
|
} => {
|
||||||
|
let _ = response_sender.send(Err(RuntimeError::ActorUnavailable(key.clone())));
|
||||||
|
}
|
||||||
ActorCommand::Stop { response_sender } => {
|
ActorCommand::Stop { response_sender } => {
|
||||||
let _ = response_sender.send(());
|
let _ = response_sender.send(());
|
||||||
}
|
}
|
||||||
@@ -987,6 +1179,113 @@ fn drain_after_failure(receiver: Receiver<ActorCommand>, key: &ServiceKey) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn validate_resident_event(
|
||||||
|
key: &ServiceKey,
|
||||||
|
limits: &ResourceLimits,
|
||||||
|
event: &ResidentEvent,
|
||||||
|
) -> Result<(), RuntimeError> {
|
||||||
|
let (payload_len, limit) = match event {
|
||||||
|
ResidentEvent::StreamData { bytes, .. } => {
|
||||||
|
(bytes.len(), limits.resident.max_stream_chunk_bytes)
|
||||||
|
}
|
||||||
|
ResidentEvent::Datagram { bytes, .. } => (bytes.len(), limits.resident.max_datagram_bytes),
|
||||||
|
ResidentEvent::Message { bytes, .. } => (bytes.len(), limits.max_input_bytes),
|
||||||
|
ResidentEvent::Source { event, payload, .. } => (
|
||||||
|
event.len().saturating_add(payload.len()),
|
||||||
|
limits.max_input_bytes,
|
||||||
|
),
|
||||||
|
ResidentEvent::Timer { .. }
|
||||||
|
| ResidentEvent::StreamOpened { .. }
|
||||||
|
| ResidentEvent::StreamWritable { .. }
|
||||||
|
| ResidentEvent::StreamHalfClosed { .. }
|
||||||
|
| ResidentEvent::StreamClosed { .. }
|
||||||
|
| ResidentEvent::Shutdown => return Ok(()),
|
||||||
|
};
|
||||||
|
if payload_len > limit {
|
||||||
|
Err(RuntimeError::ResidentEventTooLarge {
|
||||||
|
service: key.clone(),
|
||||||
|
limit,
|
||||||
|
})
|
||||||
|
} else {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn validate_resident_effects(
|
||||||
|
key: &ServiceKey,
|
||||||
|
limits: &ResourceLimits,
|
||||||
|
effects: &[ResidentEffect],
|
||||||
|
) -> Result<(), RuntimeError> {
|
||||||
|
for effect in effects {
|
||||||
|
match effect {
|
||||||
|
ResidentEffect::WriteStream { bytes, .. }
|
||||||
|
if bytes.len() > limits.resident.max_stream_chunk_bytes =>
|
||||||
|
{
|
||||||
|
return Err(invalid_resident_effect(
|
||||||
|
key,
|
||||||
|
"stream write",
|
||||||
|
limits.resident.max_stream_chunk_bytes,
|
||||||
|
));
|
||||||
|
}
|
||||||
|
ResidentEffect::SendDatagram { peer, bytes, .. } => {
|
||||||
|
if peer.is_empty() {
|
||||||
|
return Err(RuntimeError::InvalidResidentEffect(format!(
|
||||||
|
"{key} returned a datagram without a peer"
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
if bytes.len() > limits.resident.max_datagram_bytes {
|
||||||
|
return Err(invalid_resident_effect(
|
||||||
|
key,
|
||||||
|
"datagram",
|
||||||
|
limits.resident.max_datagram_bytes,
|
||||||
|
));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
ResidentEffect::ArmTimer {
|
||||||
|
delay_ms,
|
||||||
|
interval_ms,
|
||||||
|
..
|
||||||
|
} => {
|
||||||
|
let minimum = limits.resident.min_timer_delay_ms;
|
||||||
|
if *delay_ms < minimum || interval_ms.is_some_and(|value| value < minimum) {
|
||||||
|
return Err(RuntimeError::InvalidResidentEffect(format!(
|
||||||
|
"{key} returned a timer below the {minimum} ms minimum"
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
ResidentEffect::SourceCommand {
|
||||||
|
command, payload, ..
|
||||||
|
} => {
|
||||||
|
if command.is_empty() {
|
||||||
|
return Err(RuntimeError::InvalidResidentEffect(format!(
|
||||||
|
"{key} returned an empty source command"
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
if command.len().saturating_add(payload.len()) > limits.max_output_bytes {
|
||||||
|
return Err(invalid_resident_effect(
|
||||||
|
key,
|
||||||
|
"source command",
|
||||||
|
limits.max_output_bytes,
|
||||||
|
));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
ResidentEffect::WriteStream { .. }
|
||||||
|
| ResidentEffect::CloseStream { .. }
|
||||||
|
| ResidentEffect::PauseStream { .. }
|
||||||
|
| ResidentEffect::ResumeStream { .. }
|
||||||
|
| ResidentEffect::CancelTimer { .. }
|
||||||
|
| ResidentEffect::AcknowledgeMessage { .. } => {}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn invalid_resident_effect(key: &ServiceKey, kind: &str, limit: usize) -> RuntimeError {
|
||||||
|
RuntimeError::InvalidResidentEffect(format!(
|
||||||
|
"{key} returned a {kind} payload above the {limit}-byte limit"
|
||||||
|
))
|
||||||
|
}
|
||||||
|
|
||||||
fn configure_call_budget(
|
fn configure_call_budget(
|
||||||
store: &mut Store<HostState>,
|
store: &mut Store<HostState>,
|
||||||
limits: &ResourceLimits,
|
limits: &ResourceLimits,
|
||||||
@@ -1017,3 +1316,7 @@ fn component_error(kind: &'static str, error: ServiceError) -> RuntimeError {
|
|||||||
fn is_fatal(result: &Result<Vec<u8>, RuntimeError>) -> bool {
|
fn is_fatal(result: &Result<Vec<u8>, RuntimeError>) -> bool {
|
||||||
matches!(result, Err(RuntimeError::ActorFault { .. }))
|
matches!(result, Err(RuntimeError::ActorFault { .. }))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn is_fatal_resident(result: &Result<Vec<ResidentEffect>, RuntimeError>) -> bool {
|
||||||
|
matches!(result, Err(RuntimeError::ActorFault { .. }))
|
||||||
|
}
|
||||||
|
|||||||
@@ -1,6 +1,7 @@
|
|||||||
use std::{
|
use std::{
|
||||||
collections::HashMap,
|
collections::HashMap,
|
||||||
fs,
|
fs,
|
||||||
|
net::SocketAddr,
|
||||||
path::{Path, PathBuf},
|
path::{Path, PathBuf},
|
||||||
process::Command,
|
process::Command,
|
||||||
sync::{Arc, Mutex, OnceLock},
|
sync::{Arc, Mutex, OnceLock},
|
||||||
@@ -9,8 +10,11 @@ use std::{
|
|||||||
};
|
};
|
||||||
|
|
||||||
use wasmeld_runtime::{
|
use wasmeld_runtime::{
|
||||||
KV_MAX_VALUE_BYTES, KvBackend, KvBackendError, ResourceLimits, Runtime, RuntimeConfig,
|
ComponentExecution, KV_MAX_VALUE_BYTES, KvBackend, KvBackendError, NetworkScope,
|
||||||
RuntimeError, ServiceManifest,
|
ResidentEffect, ResidentEndpoint, ResidentEvent, ResidentHostError, ResidentLimits,
|
||||||
|
ResidentOperation, ResidentPolicy, ResidentResourceKind, ResidentResourceMetadata,
|
||||||
|
ResidentSession, ResourceId, ResourceLimits, Runtime, RuntimeConfig, RuntimeError,
|
||||||
|
ServiceManifest, StreamCloseReason,
|
||||||
};
|
};
|
||||||
|
|
||||||
static COMPONENTS_BUILT: OnceLock<()> = OnceLock::new();
|
static COMPONENTS_BUILT: OnceLock<()> = OnceLock::new();
|
||||||
@@ -192,6 +196,253 @@ fn explicit_clock_capability_is_linked() {
|
|||||||
runtime.stop(&key).unwrap();
|
runtime.stop(&key).unwrap();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn resident_component_handles_host_events_in_its_actor_mailbox() {
|
||||||
|
let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start");
|
||||||
|
let (key, actor) = start_component(&runtime, "resident-probe", "resident_probe_component.wasm");
|
||||||
|
let stream_id = resource_id(1);
|
||||||
|
let endpoint_id = resource_id(2);
|
||||||
|
let timer_id = resource_id(3);
|
||||||
|
let subscription_id = resource_id(4);
|
||||||
|
let source_id = resource_id(5);
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
runtime.execution(&key).unwrap(),
|
||||||
|
ComponentExecution::Resident
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
actor
|
||||||
|
.dispatch_event(ResidentEvent::StreamData {
|
||||||
|
stream_id,
|
||||||
|
bytes: b"stream".to_vec(),
|
||||||
|
})
|
||||||
|
.unwrap(),
|
||||||
|
vec![ResidentEffect::WriteStream {
|
||||||
|
stream_id,
|
||||||
|
bytes: b"stream".to_vec(),
|
||||||
|
}]
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
runtime
|
||||||
|
.dispatch_event(
|
||||||
|
&key,
|
||||||
|
ResidentEvent::Datagram {
|
||||||
|
endpoint_id,
|
||||||
|
peer: "127.0.0.1:9000".to_owned(),
|
||||||
|
bytes: b"datagram".to_vec(),
|
||||||
|
},
|
||||||
|
)
|
||||||
|
.unwrap(),
|
||||||
|
vec![ResidentEffect::SendDatagram {
|
||||||
|
endpoint_id,
|
||||||
|
peer: "127.0.0.1:9000".to_owned(),
|
||||||
|
bytes: b"datagram".to_vec(),
|
||||||
|
}]
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
actor
|
||||||
|
.dispatch_event(ResidentEvent::Timer {
|
||||||
|
timer_id,
|
||||||
|
scheduled_at_ns: 42,
|
||||||
|
})
|
||||||
|
.unwrap(),
|
||||||
|
vec![ResidentEffect::ArmTimer {
|
||||||
|
timer_id,
|
||||||
|
delay_ms: 10,
|
||||||
|
interval_ms: None,
|
||||||
|
}]
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
actor
|
||||||
|
.dispatch_event(ResidentEvent::Message {
|
||||||
|
subscription_id,
|
||||||
|
message_id: 99,
|
||||||
|
bytes: b"message".to_vec(),
|
||||||
|
})
|
||||||
|
.unwrap(),
|
||||||
|
vec![ResidentEffect::AcknowledgeMessage {
|
||||||
|
subscription_id,
|
||||||
|
message_id: 99,
|
||||||
|
}]
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
actor
|
||||||
|
.dispatch_event(ResidentEvent::Source {
|
||||||
|
source_id,
|
||||||
|
event: "flush".to_owned(),
|
||||||
|
payload: b"source".to_vec(),
|
||||||
|
})
|
||||||
|
.unwrap(),
|
||||||
|
vec![ResidentEffect::SourceCommand {
|
||||||
|
source_id,
|
||||||
|
command: "flush".to_owned(),
|
||||||
|
payload: b"source".to_vec(),
|
||||||
|
}]
|
||||||
|
);
|
||||||
|
assert_eq!(decode_counter(actor.invoke(Vec::new()).unwrap()), 5);
|
||||||
|
|
||||||
|
runtime.stop(&key).unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn passive_component_rejects_resident_events_without_stopping() {
|
||||||
|
let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start");
|
||||||
|
let (key, actor) = start_component(&runtime, "passive-echo", "echo_component.wasm");
|
||||||
|
|
||||||
|
assert_eq!(
|
||||||
|
runtime.execution(&key).unwrap(),
|
||||||
|
ComponentExecution::Service
|
||||||
|
);
|
||||||
|
assert!(matches!(
|
||||||
|
actor.dispatch_event(ResidentEvent::Shutdown),
|
||||||
|
Err(RuntimeError::NotResident(service)) if service == key
|
||||||
|
));
|
||||||
|
assert_eq!(
|
||||||
|
actor.invoke(b"still-running".to_vec()).unwrap(),
|
||||||
|
b"still-running"
|
||||||
|
);
|
||||||
|
|
||||||
|
runtime.stop(&key).unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn resident_session_owns_resources_and_validates_driver_operations() {
|
||||||
|
let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start");
|
||||||
|
let (key, actor) = start_component(
|
||||||
|
&runtime,
|
||||||
|
"resident-session",
|
||||||
|
"resident_probe_component.wasm",
|
||||||
|
);
|
||||||
|
let policy = ResidentPolicy {
|
||||||
|
tcp_listen: NetworkScope::Loopback,
|
||||||
|
udp_bind: NetworkScope::Loopback,
|
||||||
|
unix_listen_roots: vec![std::env::temp_dir()],
|
||||||
|
};
|
||||||
|
let mut session = ResidentSession::new(actor, policy).unwrap();
|
||||||
|
let tcp_id = session
|
||||||
|
.register_endpoint(ResidentEndpoint::Tcp {
|
||||||
|
name: "ingress".to_owned(),
|
||||||
|
bind: socket_addr("127.0.0.1:7000"),
|
||||||
|
})
|
||||||
|
.unwrap();
|
||||||
|
let udp_id = session
|
||||||
|
.register_endpoint(ResidentEndpoint::Udp {
|
||||||
|
name: "discovery".to_owned(),
|
||||||
|
bind: socket_addr("127.0.0.1:7001"),
|
||||||
|
})
|
||||||
|
.unwrap();
|
||||||
|
let unix_id = session
|
||||||
|
.register_endpoint(ResidentEndpoint::Unix {
|
||||||
|
name: "local-ingress".to_owned(),
|
||||||
|
path: std::env::temp_dir().join("wasmeld-resident-test.sock"),
|
||||||
|
})
|
||||||
|
.unwrap();
|
||||||
|
let timer_id = session.register_timer("heartbeat").unwrap();
|
||||||
|
let subscription_id = session.register_message_subscription("jobs").unwrap();
|
||||||
|
let source_id = session
|
||||||
|
.register_source("watcher", "fs-watch@0.1.0")
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
let (stream_id, opened) = session
|
||||||
|
.accept_stream(tcp_id, Some("127.0.0.1:51000".to_owned()))
|
||||||
|
.unwrap();
|
||||||
|
assert!(opened.is_empty());
|
||||||
|
assert_eq!(
|
||||||
|
session.stream_data(stream_id, b"stream".to_vec()).unwrap(),
|
||||||
|
vec![ResidentOperation::WriteStream {
|
||||||
|
stream_id,
|
||||||
|
bytes: b"stream".to_vec(),
|
||||||
|
}]
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
session
|
||||||
|
.datagram(udp_id, "127.0.0.1:51001".to_owned(), b"datagram".to_vec(),)
|
||||||
|
.unwrap(),
|
||||||
|
vec![ResidentOperation::SendDatagram {
|
||||||
|
endpoint_id: udp_id,
|
||||||
|
peer: "127.0.0.1:51001".to_owned(),
|
||||||
|
bytes: b"datagram".to_vec(),
|
||||||
|
}]
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
session.timer_fired(timer_id, 1).unwrap(),
|
||||||
|
vec![ResidentOperation::ArmTimer {
|
||||||
|
timer_id,
|
||||||
|
delay_ms: 10,
|
||||||
|
interval_ms: None,
|
||||||
|
}]
|
||||||
|
);
|
||||||
|
assert!(matches!(
|
||||||
|
session.resource_info(timer_id).unwrap().metadata,
|
||||||
|
ResidentResourceMetadata::Timer { armed: true }
|
||||||
|
));
|
||||||
|
assert_eq!(
|
||||||
|
session
|
||||||
|
.message(subscription_id, 7, b"work".to_vec())
|
||||||
|
.unwrap(),
|
||||||
|
vec![ResidentOperation::AcknowledgeMessage {
|
||||||
|
subscription_id,
|
||||||
|
message_id: 7,
|
||||||
|
}]
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
session
|
||||||
|
.source_event(source_id, "flush".to_owned(), b"state".to_vec())
|
||||||
|
.unwrap(),
|
||||||
|
vec![ResidentOperation::SourceCommand {
|
||||||
|
source_id,
|
||||||
|
command: "flush".to_owned(),
|
||||||
|
payload: b"state".to_vec(),
|
||||||
|
}]
|
||||||
|
);
|
||||||
|
assert!(matches!(
|
||||||
|
session.datagram(tcp_id, "127.0.0.1:1".to_owned(), Vec::new()),
|
||||||
|
Err(ResidentHostError::WrongResourceKind { .. })
|
||||||
|
));
|
||||||
|
|
||||||
|
session
|
||||||
|
.stream_closed(stream_id, StreamCloseReason::PeerClosed)
|
||||||
|
.unwrap();
|
||||||
|
assert!(matches!(
|
||||||
|
session.resource_info(stream_id),
|
||||||
|
Err(ResidentHostError::UnknownResource { .. })
|
||||||
|
));
|
||||||
|
assert_eq!(session.owner(), &key);
|
||||||
|
assert_eq!(
|
||||||
|
session.resource_info(tcp_id).unwrap().kind,
|
||||||
|
ResidentResourceKind::TcpEndpoint
|
||||||
|
);
|
||||||
|
assert_eq!(
|
||||||
|
session.resource_info(unix_id).unwrap().kind,
|
||||||
|
ResidentResourceKind::UnixEndpoint
|
||||||
|
);
|
||||||
|
|
||||||
|
session.shutdown().unwrap();
|
||||||
|
assert!(matches!(
|
||||||
|
session.register_timer("late"),
|
||||||
|
Err(ResidentHostError::ShuttingDown(_))
|
||||||
|
));
|
||||||
|
runtime.stop(&key).unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn resident_endpoint_policy_is_deny_by_default() {
|
||||||
|
let runtime = Runtime::new(RuntimeConfig::default()).expect("runtime should start");
|
||||||
|
let (key, actor) =
|
||||||
|
start_component(&runtime, "resident-policy", "resident_probe_component.wasm");
|
||||||
|
let mut session = ResidentSession::new(actor, ResidentPolicy::default()).unwrap();
|
||||||
|
|
||||||
|
assert!(matches!(
|
||||||
|
session.register_endpoint(ResidentEndpoint::Tcp {
|
||||||
|
name: "public".to_owned(),
|
||||||
|
bind: socket_addr("0.0.0.0:8080"),
|
||||||
|
}),
|
||||||
|
Err(ResidentHostError::EndpointDenied(_))
|
||||||
|
));
|
||||||
|
|
||||||
|
runtime.stop(&key).unwrap();
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn kv_capability_is_service_scoped_and_shared_across_revisions() {
|
fn kv_capability_is_service_scoped_and_shared_across_revisions() {
|
||||||
let backend = Arc::new(MemoryKvBackend::default());
|
let backend = Arc::new(MemoryKvBackend::default());
|
||||||
@@ -379,6 +630,7 @@ fn build_components_once() {
|
|||||||
"components/spin/Cargo.toml",
|
"components/spin/Cargo.toml",
|
||||||
"components/clock-probe/Cargo.toml",
|
"components/clock-probe/Cargo.toml",
|
||||||
"components/kv-probe/Cargo.toml",
|
"components/kv-probe/Cargo.toml",
|
||||||
|
"components/resident-probe/Cargo.toml",
|
||||||
"components/wasi-clock-probe/Cargo.toml",
|
"components/wasi-clock-probe/Cargo.toml",
|
||||||
] {
|
] {
|
||||||
let status = Command::new("rustup")
|
let status = Command::new("rustup")
|
||||||
@@ -421,9 +673,18 @@ fn test_limits() -> ResourceLimits {
|
|||||||
mailbox_capacity: 16,
|
mailbox_capacity: 16,
|
||||||
max_input_bytes: 16 * 1024,
|
max_input_bytes: 16 * 1024,
|
||||||
max_output_bytes: 16 * 1024,
|
max_output_bytes: 16 * 1024,
|
||||||
|
resident: ResidentLimits::default(),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn resource_id(value: u64) -> ResourceId {
|
||||||
|
ResourceId::new(value).expect("test resource id must be non-zero")
|
||||||
|
}
|
||||||
|
|
||||||
|
fn socket_addr(value: &str) -> SocketAddr {
|
||||||
|
value.parse().expect("test socket address must be valid")
|
||||||
|
}
|
||||||
|
|
||||||
fn decode_counter(bytes: Vec<u8>) -> u64 {
|
fn decode_counter(bytes: Vec<u8>) -> u64 {
|
||||||
let array: [u8; 8] = bytes
|
let array: [u8; 8] = bytes
|
||||||
.try_into()
|
.try_into()
|
||||||
@@ -447,6 +708,9 @@ fn component_world(artifact_name: &str) -> &'static str {
|
|||||||
"spin_component.wasm" => "component:spin/spin-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",
|
"clock_probe_component.wasm" => "component:clock-probe/clock-probe-component@0.1.0",
|
||||||
"kv_probe_component.wasm" => "component:kv-probe/kv-probe-component@0.1.0",
|
"kv_probe_component.wasm" => "component:kv-probe/kv-probe-component@0.1.0",
|
||||||
|
"resident_probe_component.wasm" => {
|
||||||
|
"component:resident-probe/resident-probe-component@0.1.0"
|
||||||
|
}
|
||||||
"wasi_clock_probe_component.wasm" => {
|
"wasi_clock_probe_component.wasm" => {
|
||||||
"component:wasi-clock-probe/wasi-clock-probe-component@0.1.0"
|
"component:wasi-clock-probe/wasi-clock-probe-component@0.1.0"
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,123 @@
|
|||||||
|
package wasmeld:resident@0.1.0;
|
||||||
|
|
||||||
|
/// Event/effect ABI for Components that remain active without monopolizing
|
||||||
|
/// their Actor thread. The Host owns timers, subscriptions, sockets, streams,
|
||||||
|
/// and all operating-system handles. A Component handles one bounded event and
|
||||||
|
/// returns effects before control goes back to the Runtime.
|
||||||
|
interface actor {
|
||||||
|
type resource-id = u64;
|
||||||
|
|
||||||
|
record timer-fired {
|
||||||
|
timer-id: resource-id,
|
||||||
|
scheduled-at-ns: u64,
|
||||||
|
}
|
||||||
|
|
||||||
|
record stream-opened {
|
||||||
|
endpoint-id: resource-id,
|
||||||
|
stream-id: resource-id,
|
||||||
|
peer: option<string>,
|
||||||
|
}
|
||||||
|
|
||||||
|
record stream-chunk {
|
||||||
|
stream-id: resource-id,
|
||||||
|
bytes: list<u8>,
|
||||||
|
}
|
||||||
|
|
||||||
|
enum close-reason {
|
||||||
|
peer-closed,
|
||||||
|
host-closed,
|
||||||
|
idle-timeout,
|
||||||
|
protocol-error,
|
||||||
|
transport-error,
|
||||||
|
}
|
||||||
|
|
||||||
|
record stream-closed {
|
||||||
|
stream-id: resource-id,
|
||||||
|
reason: close-reason,
|
||||||
|
}
|
||||||
|
|
||||||
|
record datagram {
|
||||||
|
endpoint-id: resource-id,
|
||||||
|
peer: string,
|
||||||
|
bytes: list<u8>,
|
||||||
|
}
|
||||||
|
|
||||||
|
record message {
|
||||||
|
subscription-id: resource-id,
|
||||||
|
message-id: u64,
|
||||||
|
bytes: list<u8>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Escape hatch for Host-owned event sources such as filesystem watchers,
|
||||||
|
/// serial devices, process signals, or platform-specific buses. `kind` and
|
||||||
|
/// `payload` are interpreted by the separately versioned Host capability
|
||||||
|
/// that created `source-id`.
|
||||||
|
record source-event {
|
||||||
|
source-id: resource-id,
|
||||||
|
kind: string,
|
||||||
|
payload: list<u8>,
|
||||||
|
}
|
||||||
|
|
||||||
|
variant event {
|
||||||
|
timer(timer-fired),
|
||||||
|
stream-opened(stream-opened),
|
||||||
|
stream-data(stream-chunk),
|
||||||
|
stream-writable(resource-id),
|
||||||
|
stream-half-closed(resource-id),
|
||||||
|
stream-closed(stream-closed),
|
||||||
|
datagram(datagram),
|
||||||
|
message(message),
|
||||||
|
source(source-event),
|
||||||
|
shutdown,
|
||||||
|
}
|
||||||
|
|
||||||
|
record stream-write {
|
||||||
|
stream-id: resource-id,
|
||||||
|
bytes: list<u8>,
|
||||||
|
}
|
||||||
|
|
||||||
|
record datagram-send {
|
||||||
|
endpoint-id: resource-id,
|
||||||
|
peer: string,
|
||||||
|
bytes: list<u8>,
|
||||||
|
}
|
||||||
|
|
||||||
|
record timer-arm {
|
||||||
|
timer-id: resource-id,
|
||||||
|
delay-ms: u64,
|
||||||
|
interval-ms: option<u64>,
|
||||||
|
}
|
||||||
|
|
||||||
|
record message-ack {
|
||||||
|
subscription-id: resource-id,
|
||||||
|
message-id: u64,
|
||||||
|
}
|
||||||
|
|
||||||
|
record source-command {
|
||||||
|
source-id: resource-id,
|
||||||
|
command: string,
|
||||||
|
payload: list<u8>,
|
||||||
|
}
|
||||||
|
|
||||||
|
variant effect {
|
||||||
|
write-stream(stream-write),
|
||||||
|
close-stream(resource-id),
|
||||||
|
pause-stream(resource-id),
|
||||||
|
resume-stream(resource-id),
|
||||||
|
send-datagram(datagram-send),
|
||||||
|
arm-timer(timer-arm),
|
||||||
|
cancel-timer(resource-id),
|
||||||
|
acknowledge-message(message-ack),
|
||||||
|
source-command(source-command),
|
||||||
|
}
|
||||||
|
|
||||||
|
variant resident-error {
|
||||||
|
event-failed(string),
|
||||||
|
}
|
||||||
|
|
||||||
|
handle-event: func(input: event) -> result<list<effect>, resident-error>;
|
||||||
|
}
|
||||||
|
|
||||||
|
world resident-component {
|
||||||
|
export actor;
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user