1367 lines
50 KiB
Rust
1367 lines
50 KiB
Rust
use swactor::actor::{ActorAddress, ActorInterface};
|
|
use swactor::runtime::{Ctx, Runtime, RuntimeConfig};
|
|
use swactor_wasm_actor::{ByteMessage, SharedEngine, WasmActorBuilder, WasmActorError};
|
|
|
|
use proptest::prelude::*;
|
|
|
|
fn guest_wasm(name: &str) -> Vec<u8> {
|
|
let path = format!(
|
|
"{}/tests/guests/{name}/target/wasm32-unknown-unknown/release/{name}_guest.wasm",
|
|
env!("CARGO_MANIFEST_DIR")
|
|
);
|
|
std::fs::read(&path).unwrap_or_else(|e| panic!("failed to read {path}: {e}"))
|
|
}
|
|
|
|
/// Build a message with an inbox address prepended (the guest contract).
|
|
fn framed_msg(dest: &ActorAddress, payload: &[u8]) -> ByteMessage {
|
|
let mut buf = Vec::with_capacity(32 + payload.len());
|
|
buf.extend_from_slice(&dest.0);
|
|
buf.extend_from_slice(payload);
|
|
ByteMessage(buf)
|
|
}
|
|
|
|
// ── Echo: send bytes in, same bytes come back ────────────────────────────────
|
|
|
|
#[test]
|
|
fn echo_returns_same_payload() {
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
let payload = b"hello wasm";
|
|
rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
|
|
rt.tick();
|
|
|
|
let received = inbox.try_recv().expect("inbox should have a message");
|
|
assert_eq!(received.0, payload);
|
|
}
|
|
|
|
#[test]
|
|
fn echo_preserves_binary_payload() {
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
let payload: Vec<u8> = (0..=255).collect();
|
|
rt.send_to(addr, framed_msg(inbox.addr(), &payload)).unwrap();
|
|
rt.tick();
|
|
|
|
let received = inbox.try_recv().expect("inbox should have a message");
|
|
assert_eq!(received.0, payload);
|
|
}
|
|
|
|
// ── Silent: processes messages without sending anything ───────────────────────
|
|
|
|
#[test]
|
|
fn silent_produces_no_output() {
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, guest_wasm("silent"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
rt.send_to(addr, ByteMessage(b"ignored".to_vec())).unwrap();
|
|
rt.tick();
|
|
|
|
assert!(inbox.try_recv().is_none(), "silent guest should not send anything");
|
|
}
|
|
|
|
// ── Double: one message in, two messages out ─────────────────────────────────
|
|
|
|
#[test]
|
|
fn double_sends_two_copies() {
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, guest_wasm("double"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
let payload = b"dup me";
|
|
rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
|
|
rt.tick();
|
|
|
|
let first = inbox.try_recv().expect("should receive first copy");
|
|
let second = inbox.try_recv().expect("should receive second copy");
|
|
assert_eq!(first.0, payload);
|
|
assert_eq!(second.0, payload);
|
|
assert!(inbox.try_recv().is_none(), "exactly two messages expected");
|
|
}
|
|
|
|
// ── Missing export → WasmActorError::MissingExport ───────────────────────────
|
|
|
|
#[test]
|
|
fn missing_alloc_export_returns_error() {
|
|
// Minimal valid Wasm module: (module) — no exports at all
|
|
let minimal_wasm = wat::parse_str("(module)").unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let result = WasmActorBuilder::new(engine, minimal_wasm).build();
|
|
match result {
|
|
Err(WasmActorError::MissingExport(name)) => {
|
|
assert!(
|
|
name == "memory" || name == "alloc",
|
|
"expected missing memory or alloc, got: {name}"
|
|
);
|
|
}
|
|
Err(other) => panic!("expected MissingExport, got: {other}"),
|
|
Ok(_) => panic!("expected error for module with no exports"),
|
|
}
|
|
}
|
|
|
|
// ── Engine sharing: two actors from the same engine ──────────────────────────
|
|
|
|
#[test]
|
|
fn shared_engine_serves_multiple_actors() {
|
|
let engine = SharedEngine::new().unwrap();
|
|
|
|
let echo = WasmActorBuilder::new(engine.clone(), guest_wasm("echo"))
|
|
.build()
|
|
.unwrap();
|
|
let silent = WasmActorBuilder::new(engine, guest_wasm("silent"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
|
|
let echo_addr = rt.spawn(echo).unwrap();
|
|
let _silent_addr = rt.spawn(silent).unwrap();
|
|
|
|
let payload = b"shared engine test";
|
|
rt.send_to(echo_addr, framed_msg(inbox.addr(), payload)).unwrap();
|
|
rt.tick();
|
|
|
|
let received = inbox.try_recv().expect("echo actor should still work");
|
|
assert_eq!(received.0, payload);
|
|
}
|
|
|
|
// ── Safety: edge cases that previously caused panics or corruption ────────────
|
|
|
|
#[test]
|
|
fn oob_send_traps_cleanly_and_actor_survives() {
|
|
// Guest calls swactor.send with dest_ptr pointing past the end of memory.
|
|
// The host should trap the call; the actor should survive for future messages.
|
|
let wat = r#"
|
|
(module
|
|
(import "swactor" "send" (func $send (param i32 i32 i32)))
|
|
(memory (export "memory") 1)
|
|
(func (export "alloc") (param i32) (result i32)
|
|
i32.const 0 ;; return start of memory (simplistic)
|
|
)
|
|
(func (export "handle") (param i32 i32)
|
|
;; Call send with dest_ptr = 65536 (1 page = end of memory, OOB for 32 bytes)
|
|
i32.const 65536
|
|
i32.const 0
|
|
i32.const 0
|
|
call $send
|
|
)
|
|
)
|
|
"#;
|
|
let wasm = wat::parse_str(wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
// Send a message — handle will try OOB send, which traps
|
|
rt.send_to(addr, ByteMessage(vec![42])).unwrap();
|
|
rt.tick();
|
|
|
|
// No message should arrive (the send was invalid)
|
|
assert!(inbox.try_recv().is_none(), "OOB send should not produce a message");
|
|
}
|
|
|
|
#[test]
|
|
fn alloc_oom_drops_message_actor_stays_alive() {
|
|
// Guest alloc always returns 0 (OOM). Message should be dropped,
|
|
// actor should remain alive for subsequent messages.
|
|
let wat = r#"
|
|
(module
|
|
(import "swactor" "send" (func $send (param i32 i32 i32)))
|
|
(memory (export "memory") 1)
|
|
(func (export "alloc") (param i32) (result i32)
|
|
i32.const 0 ;; always OOM
|
|
)
|
|
(func (export "handle") (param i32 i32)
|
|
;; Should never be called if alloc returned 0 for non-zero len
|
|
)
|
|
)
|
|
"#;
|
|
let wasm = wat::parse_str(wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
// Send a non-empty message — alloc returns 0, message should be dropped
|
|
rt.send_to(addr, ByteMessage(vec![1, 2, 3])).unwrap();
|
|
rt.tick();
|
|
|
|
// Actor is still alive — send another message, tick again (no panic)
|
|
rt.send_to(addr, ByteMessage(vec![4, 5, 6])).unwrap();
|
|
rt.tick();
|
|
}
|
|
|
|
#[test]
|
|
fn negative_alloc_ptr_drops_message() {
|
|
// Guest alloc returns -1. Host should detect the negative pointer and drop.
|
|
let wat = r#"
|
|
(module
|
|
(import "swactor" "send" (func $send (param i32 i32 i32)))
|
|
(memory (export "memory") 1)
|
|
(func (export "alloc") (param i32) (result i32)
|
|
i32.const -1 ;; invalid negative pointer
|
|
)
|
|
(func (export "handle") (param i32 i32))
|
|
)
|
|
"#;
|
|
let wasm = wat::parse_str(wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
rt.send_to(addr, ByteMessage(vec![1])).unwrap();
|
|
rt.tick(); // should not panic
|
|
|
|
// Actor survives
|
|
rt.send_to(addr, ByteMessage(vec![2])).unwrap();
|
|
rt.tick();
|
|
}
|
|
|
|
#[test]
|
|
fn handle_trap_drops_message_actor_survives() {
|
|
// Guest handle executes `unreachable`, causing a Wasm trap.
|
|
// Message should be dropped, actor should stay alive.
|
|
let wat = r#"
|
|
(module
|
|
(import "swactor" "send" (func $send (param i32 i32 i32)))
|
|
(memory (export "memory") 1)
|
|
(func (export "alloc") (param i32) (result i32)
|
|
i32.const 256 ;; valid allocation
|
|
)
|
|
(func (export "handle") (param i32 i32)
|
|
unreachable ;; trap!
|
|
)
|
|
)
|
|
"#;
|
|
let wasm = wat::parse_str(wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
rt.send_to(addr, ByteMessage(vec![1, 2, 3])).unwrap();
|
|
rt.tick(); // handle traps, but actor should survive
|
|
|
|
// Actor is still alive
|
|
rt.send_to(addr, ByteMessage(vec![4, 5, 6])).unwrap();
|
|
rt.tick();
|
|
}
|
|
|
|
// ── Bounds safety: alloc pointer near end of linear memory ────────────────────
|
|
|
|
#[test]
|
|
fn alloc_near_end_of_memory_drops_message_actor_survives() {
|
|
// Guest alloc returns 65500 (near end of 1-page / 65536-byte memory).
|
|
// A 100-byte message means ptr+len = 65600, which exceeds memory bounds.
|
|
// The actor should drop the message and survive — same as any other
|
|
// allocation failure — rather than being permanently killed.
|
|
let wat = r#"
|
|
(module
|
|
(import "swactor" "send" (func $send (param i32 i32 i32)))
|
|
(memory (export "memory") 1)
|
|
(func (export "alloc") (param i32) (result i32)
|
|
i32.const 65500 ;; near end of 64KiB memory
|
|
)
|
|
(func (export "handle") (param i32 i32)
|
|
;; should never be reached if bounds check works
|
|
)
|
|
)
|
|
"#;
|
|
let wasm = wat::parse_str(wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
// Send a message whose length exceeds the remaining space at ptr 65500
|
|
rt.send_to(addr, ByteMessage(vec![0u8; 100])).unwrap();
|
|
rt.tick();
|
|
|
|
// The actor should still be alive — send another message and tick without panic
|
|
rt.send_to(addr, ByteMessage(vec![1u8; 10])).unwrap();
|
|
rt.tick();
|
|
}
|
|
|
|
// ── Edge cases: empty and oversized messages ─────────────────────────────────
|
|
|
|
#[test]
|
|
fn empty_message_is_handled_without_crash() {
|
|
// A zero-length ByteMessage should pass through the alloc/handle pipeline
|
|
// without crashing. The echo guest returns nothing (len < 32 guard).
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
rt.send_to(addr, ByteMessage(vec![])).unwrap();
|
|
rt.tick();
|
|
|
|
// Echo guest does `if len < 32 { return; }` — so no reply expected
|
|
assert!(inbox.try_recv().is_none(), "empty message should produce no reply");
|
|
|
|
// Actor survives — can still process a real message
|
|
let payload = b"still alive";
|
|
rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
|
|
rt.tick();
|
|
let received = inbox.try_recv().expect("actor should still be alive");
|
|
assert_eq!(received.0, payload);
|
|
}
|
|
|
|
#[test]
|
|
fn message_larger_than_linear_memory_is_dropped() {
|
|
// A message of 65537 bytes exceeds the 1-page (64KiB) guest memory.
|
|
// Guest alloc will OOM (return 0) → message dropped, actor survives.
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
rt.send_to(addr, ByteMessage(vec![0u8; 65537])).unwrap();
|
|
rt.tick();
|
|
|
|
assert!(inbox.try_recv().is_none(), "oversized message should be dropped");
|
|
|
|
// Actor survives
|
|
let payload = b"after oversize";
|
|
rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
|
|
rt.tick();
|
|
let received = inbox.try_recv().expect("actor should survive oversized message");
|
|
assert_eq!(received.0, payload);
|
|
}
|
|
|
|
// ── Sustained load: bump allocator exhaustion ────────────────────────────────
|
|
|
|
#[test]
|
|
fn sequential_messages_degrade_gracefully_after_allocator_exhaustion() {
|
|
// The echo guest has a 64KiB bump allocator that never frees. Under
|
|
// sustained load, alloc eventually returns 0 (OOM) and messages are
|
|
// silently dropped. The actor must survive throughout — no panics,
|
|
// no poisoning.
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
let payload = b"ping";
|
|
let mut echoed = 0usize;
|
|
|
|
// Send enough messages to exhaust the 64KiB heap.
|
|
// Each framed message is 36 bytes (32 addr + 4 payload), aligned to 40.
|
|
// 65536 / 40 = ~1638, but heap offset within memory varies. Send 2000.
|
|
for _ in 0..2000 {
|
|
rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
|
|
rt.tick();
|
|
if inbox.try_recv().is_some() {
|
|
echoed += 1;
|
|
}
|
|
}
|
|
|
|
// Some messages were echoed before OOM
|
|
assert!(echoed > 0, "should echo at least some messages");
|
|
// After OOM, messages were dropped — so not all 2000 echoed
|
|
assert!(echoed < 2000, "allocator should exhaust before 2000 messages");
|
|
}
|
|
|
|
// ── Fire-and-forget: guest sends to nonexistent address ──────────────────────
|
|
|
|
#[test]
|
|
fn guest_send_to_nonexistent_address_is_silently_dropped() {
|
|
// Guest sends to an all-zero 32-byte address that isn't registered in
|
|
// the runtime. The ctx.send() error is silently dropped (fire-and-forget).
|
|
// Actor must survive.
|
|
let wat = r#"
|
|
(module
|
|
(import "swactor" "send" (func $send (param i32 i32 i32)))
|
|
(memory (export "memory") 1)
|
|
(func (export "alloc") (param i32) (result i32)
|
|
i32.const 0
|
|
)
|
|
(func (export "handle") (param i32 i32)
|
|
;; send to address at offset 0 (all zeros — no such actor)
|
|
;; with 1-byte payload at offset 32
|
|
i32.const 0 ;; dest_ptr
|
|
i32.const 32 ;; payload_ptr
|
|
i32.const 1 ;; payload_len
|
|
call $send
|
|
)
|
|
)
|
|
"#;
|
|
let wasm = wat::parse_str(wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
rt.send_to(addr, ByteMessage(vec![42])).unwrap();
|
|
rt.tick(); // guest sends to nonexistent address — should not panic
|
|
|
|
// Actor survives
|
|
rt.send_to(addr, ByteMessage(vec![99])).unwrap();
|
|
rt.tick();
|
|
}
|
|
|
|
// ── Builder validation: wrong export signatures ──────────────────────────────
|
|
|
|
#[test]
|
|
fn wrong_handle_signature_is_rejected() {
|
|
// Module exports `handle` with wrong signature: (i32) -> i32 instead of (i32, i32) -> ()
|
|
// Builder maps get_typed_func errors to MissingExport (signature mismatch = not found).
|
|
let wat = r#"
|
|
(module
|
|
(memory (export "memory") 1)
|
|
(func (export "alloc") (param i32) (result i32) i32.const 0)
|
|
(func (export "handle") (param i32) (result i32) i32.const 0)
|
|
)
|
|
"#;
|
|
let wasm = wat::parse_str(wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let result = WasmActorBuilder::new(engine, wasm).build();
|
|
assert!(result.is_err(), "should reject wrong handle signature");
|
|
}
|
|
|
|
#[test]
|
|
fn wrong_memory_export_name_returns_missing_export() {
|
|
// Module has a memory, but exported as "mem" instead of "memory"
|
|
let wat = r#"
|
|
(module
|
|
(memory (export "mem") 1)
|
|
(func (export "alloc") (param i32) (result i32) i32.const 0)
|
|
(func (export "handle") (param i32 i32))
|
|
)
|
|
"#;
|
|
let wasm = wat::parse_str(wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let result = WasmActorBuilder::new(engine, wasm).build();
|
|
match result {
|
|
Err(WasmActorError::MissingExport("memory")) => {} // expected
|
|
Err(other) => panic!("expected MissingExport(\"memory\"), got: {other}"),
|
|
Ok(_) => panic!("should reject module without 'memory' export"),
|
|
}
|
|
}
|
|
|
|
// ── Lifecycle: graceful stop of WASM actor ───────────────────────────────────
|
|
|
|
#[test]
|
|
fn graceful_stop_cleans_up_wasm_actor() {
|
|
// After stopping a WASM actor, it should be removed from the runtime.
|
|
// The wasmtime Store is dropped cleanly (no leak, no crash).
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
// Verify the actor works
|
|
let payload = b"before stop";
|
|
rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
|
|
rt.tick();
|
|
let received = inbox.try_recv().expect("actor should echo before stop");
|
|
assert_eq!(received.0, payload);
|
|
|
|
// Stop the actor
|
|
rt.stop_actor(addr).unwrap();
|
|
rt.tick(); // process stop signal
|
|
rt.tick(); // cleanup_dead phase
|
|
|
|
// Actor is gone — send should fail
|
|
let result = rt.send_to(addr, ByteMessage(vec![1]));
|
|
assert!(result.is_err(), "send to stopped actor should fail");
|
|
}
|
|
|
|
// ── Host import validation: negative payload_len ─────────────────────────────
|
|
|
|
#[test]
|
|
fn negative_payload_len_in_send_traps_actor_survives() {
|
|
// Guest calls swactor.send with payload_len = -1. The host import
|
|
// should trap (negative argument check), and the actor should survive.
|
|
let wat = r#"
|
|
(module
|
|
(import "swactor" "send" (func $send (param i32 i32 i32)))
|
|
(memory (export "memory") 1)
|
|
(func (export "alloc") (param i32) (result i32)
|
|
i32.const 256
|
|
)
|
|
(func (export "handle") (param i32 i32)
|
|
i32.const 0 ;; dest_ptr
|
|
i32.const 0 ;; payload_ptr
|
|
i32.const -1 ;; payload_len (negative!)
|
|
call $send
|
|
)
|
|
)
|
|
"#;
|
|
let wasm = wat::parse_str(wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
rt.send_to(addr, ByteMessage(vec![1, 2, 3])).unwrap();
|
|
rt.tick(); // guest calls send with negative len — should trap
|
|
|
|
// Actor survives
|
|
rt.send_to(addr, ByteMessage(vec![4, 5, 6])).unwrap();
|
|
rt.tick();
|
|
}
|
|
|
|
// ── Independent stores: two echo actors from same engine + bytes ─────────────
|
|
|
|
#[test]
|
|
fn two_echo_actors_from_same_engine_are_independent() {
|
|
let engine = SharedEngine::new().unwrap();
|
|
let wasm_bytes = guest_wasm("echo");
|
|
|
|
let actor_a = WasmActorBuilder::new(engine.clone(), wasm_bytes.clone())
|
|
.build()
|
|
.unwrap();
|
|
let actor_b = WasmActorBuilder::new(engine, wasm_bytes)
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox_a = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let inbox_b = rt.new_inbox::<ByteMessage>().unwrap();
|
|
|
|
let addr_a = rt.spawn(actor_a).unwrap();
|
|
let addr_b = rt.spawn(actor_b).unwrap();
|
|
|
|
// Send different payloads to each
|
|
rt.send_to(addr_a, framed_msg(inbox_a.addr(), b"for-a")).unwrap();
|
|
rt.send_to(addr_b, framed_msg(inbox_b.addr(), b"for-b")).unwrap();
|
|
rt.tick();
|
|
|
|
let recv_a = inbox_a.try_recv().expect("actor A should echo");
|
|
let recv_b = inbox_b.try_recv().expect("actor B should echo");
|
|
assert_eq!(recv_a.0, b"for-a");
|
|
assert_eq!(recv_b.0, b"for-b");
|
|
|
|
// Cross-check: no leakage between actors
|
|
assert!(inbox_a.try_recv().is_none());
|
|
assert!(inbox_b.try_recv().is_none());
|
|
}
|
|
|
|
// ── Watch integration: native watcher observes WASM actor death ──────────────
|
|
|
|
struct ExitWatcher {
|
|
exit_count: std::sync::Arc<std::sync::atomic::AtomicUsize>,
|
|
last_reason: std::sync::Arc<std::sync::Mutex<Option<swactor::actor::ExitReason>>>,
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
struct WatchThis(ActorAddress);
|
|
|
|
impl ActorInterface for ExitWatcher {
|
|
type Incoming = WatchThis;
|
|
type Response = ();
|
|
fn handle(&mut self, ctx: &Ctx, msg: WatchThis) {
|
|
ctx.watch(msg.0);
|
|
}
|
|
fn on_actor_exit(&mut self, _ctx: &Ctx, exited: swactor::actor::ActorExited) {
|
|
self.exit_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
|
|
*self.last_reason.lock().unwrap() = Some(exited.reason);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn native_watcher_notified_when_wasm_actor_stops() {
|
|
// A native actor watches a WASM actor. When the WASM actor is stopped,
|
|
// the watcher should receive ActorExited with ExitReason::Stopped.
|
|
let engine = SharedEngine::new().unwrap();
|
|
let wasm = WasmActorBuilder::new(engine, guest_wasm("echo"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
|
|
let exit_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
|
|
let last_reason = std::sync::Arc::new(std::sync::Mutex::new(None));
|
|
let watcher = ExitWatcher {
|
|
exit_count: exit_count.clone(),
|
|
last_reason: last_reason.clone(),
|
|
};
|
|
|
|
let wasm_addr = rt.spawn(wasm).unwrap();
|
|
let watcher_addr = rt.spawn(watcher).unwrap();
|
|
|
|
// Tell watcher to watch the WASM actor
|
|
rt.send_to(watcher_addr, WatchThis(wasm_addr)).unwrap();
|
|
for _ in 0..3 { rt.tick(); }
|
|
|
|
// Stop the WASM actor
|
|
rt.stop_actor(wasm_addr).unwrap();
|
|
for _ in 0..5 { rt.tick(); }
|
|
|
|
assert_eq!(exit_count.load(std::sync::atomic::Ordering::SeqCst), 1);
|
|
assert_eq!(
|
|
*last_reason.lock().unwrap(),
|
|
Some(swactor::actor::ExitReason::Stopped)
|
|
);
|
|
}
|
|
|
|
// ── WASM-to-WASM: two WASM actors communicating ─────────────────────────────
|
|
|
|
#[test]
|
|
fn wasm_to_wasm_message_relay() {
|
|
// Echo A echoes to Echo B's address, Echo B echoes to an external inbox.
|
|
// This verifies the full WASM→runtime→WASM→runtime→inbox path.
|
|
let engine = SharedEngine::new().unwrap();
|
|
let wasm_bytes = guest_wasm("echo");
|
|
|
|
let actor_a = WasmActorBuilder::new(engine.clone(), wasm_bytes.clone())
|
|
.build()
|
|
.unwrap();
|
|
let actor_b = WasmActorBuilder::new(engine, wasm_bytes)
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
|
|
let addr_a = rt.spawn(actor_a).unwrap();
|
|
let addr_b = rt.spawn(actor_b).unwrap();
|
|
|
|
// Send to actor A: "echo your payload to actor B"
|
|
// Actor A receives [addr_b | payload_for_b]
|
|
// Actor A echoes payload_for_b to addr_b
|
|
// payload_for_b itself is [inbox_addr | final_payload]
|
|
// Actor B receives [inbox_addr | final_payload]
|
|
// Actor B echoes final_payload to inbox
|
|
let final_payload = b"relayed";
|
|
let payload_for_b = framed_msg(inbox.addr(), final_payload);
|
|
let msg_for_a = framed_msg(&addr_b, &payload_for_b.0);
|
|
|
|
rt.send_to(addr_a, msg_for_a).unwrap();
|
|
rt.tick(); // A receives, echoes to B
|
|
rt.tick(); // B receives, echoes to inbox
|
|
|
|
let received = inbox.try_recv().expect("should receive relayed message");
|
|
assert_eq!(received.0, final_payload);
|
|
}
|
|
|
|
// ── Integration: WasmActor alongside a native Rust actor ─────────────────────
|
|
|
|
#[derive(Clone)]
|
|
struct ForwardToWasm {
|
|
wasm_addr: ActorAddress,
|
|
inbox_addr: ActorAddress,
|
|
}
|
|
|
|
struct Forwarder;
|
|
|
|
impl ActorInterface for Forwarder {
|
|
type Incoming = ForwardToWasm;
|
|
type Response = ();
|
|
|
|
fn handle(&mut self, ctx: &Ctx, msg: ForwardToWasm) {
|
|
// Build the framed message and forward to the wasm actor
|
|
let payload = b"from native";
|
|
let framed = framed_msg(&msg.inbox_addr, payload);
|
|
let _ = ctx.send(msg.wasm_addr, framed);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn native_actor_communicates_with_wasm_actor() {
|
|
let engine = SharedEngine::new().unwrap();
|
|
let wasm = WasmActorBuilder::new(engine, guest_wasm("echo"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
|
|
let wasm_addr = rt.spawn(wasm).unwrap();
|
|
let forwarder_addr = rt.spawn(Forwarder).unwrap();
|
|
|
|
rt.send_to(
|
|
forwarder_addr,
|
|
ForwardToWasm {
|
|
wasm_addr,
|
|
inbox_addr: *inbox.addr(),
|
|
},
|
|
)
|
|
.unwrap();
|
|
|
|
// Tick 1: Forwarder receives message and sends to WasmActor
|
|
rt.tick();
|
|
// Tick 2: WasmActor receives the forwarded message and echoes to inbox
|
|
rt.tick();
|
|
|
|
let received = inbox.try_recv().expect("wasm actor should have echoed");
|
|
assert_eq!(received.0, b"from native");
|
|
}
|
|
|
|
// ── Stale outbox: sends before trap leak into next handle ─────────────────────
|
|
|
|
#[test]
|
|
fn outbox_entries_from_trapped_handle_do_not_leak_into_next_call() {
|
|
// A guest that calls swactor.send() successfully, then traps.
|
|
// The outbox contains the send from before the trap.
|
|
// On the next handle call (which succeeds without sending), the stale
|
|
// outbox entry should NOT be delivered.
|
|
//
|
|
// Counter incremented BEFORE the if-branch so it persists past the trap.
|
|
let wat = r#"
|
|
(module
|
|
(import "swactor" "send" (func $send (param i32 i32 i32)))
|
|
(memory (export "memory") 1)
|
|
(global $counter (mut i32) (i32.const 0))
|
|
|
|
(func (export "alloc") (param i32) (result i32)
|
|
i32.const 256
|
|
)
|
|
(func (export "handle") (param $ptr i32) (param $len i32)
|
|
;; Increment counter first (survives trap)
|
|
global.get $counter
|
|
i32.const 1
|
|
i32.add
|
|
global.set $counter
|
|
|
|
;; If counter was 0 (now 1): send then trap
|
|
global.get $counter
|
|
i32.const 1
|
|
i32.eq
|
|
if
|
|
local.get $ptr ;; dest_ptr (first 32 bytes = inbox address)
|
|
i32.const 32 ;; payload_ptr
|
|
i32.const 1 ;; payload_len
|
|
call $send
|
|
unreachable ;; trap after send
|
|
end
|
|
;; counter > 1: do nothing (no send)
|
|
)
|
|
)
|
|
"#;
|
|
let wasm = wat::parse_str(wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
// First message: guest sends to outbox then traps — stale entry in outbox
|
|
// Use framed_msg so the first 32 bytes are the inbox address
|
|
rt.send_to(addr, framed_msg(inbox.addr(), b"x")).unwrap();
|
|
rt.tick();
|
|
|
|
// No message should have been delivered (handle trapped before outbox drain)
|
|
assert!(inbox.try_recv().is_none(), "trapped handle should not deliver messages");
|
|
|
|
// Second message: guest does nothing (counter=2, no send, no trap).
|
|
// If the outbox wasn't cleared, the stale entry would be drained here.
|
|
rt.send_to(addr, framed_msg(inbox.addr(), b"y")).unwrap();
|
|
rt.tick();
|
|
|
|
// Should still be empty — the stale outbox entry must not leak
|
|
assert!(
|
|
inbox.try_recv().is_none(),
|
|
"stale outbox entry from trapped call should not leak into next handle"
|
|
);
|
|
}
|
|
|
|
// ── Builder validation: invalid WASM bytes ───────────────────────────────────
|
|
|
|
#[test]
|
|
fn invalid_wasm_bytes_returns_wasmtime_error() {
|
|
let garbage = vec![0u8, 1, 2, 3]; // not valid wasm
|
|
let engine = SharedEngine::new().unwrap();
|
|
let result = WasmActorBuilder::new(engine, garbage).build();
|
|
match result {
|
|
Err(WasmActorError::Wasmtime(_)) => {} // expected — compilation failure
|
|
Err(other) => panic!("expected Wasmtime error for invalid bytes, got: {other}"),
|
|
Ok(_) => panic!("should reject invalid wasm bytes"),
|
|
}
|
|
}
|
|
|
|
// ── Guest sends zero-length payload ──────────────────────────────────────────
|
|
|
|
#[test]
|
|
fn guest_send_with_zero_length_payload_delivers_empty_message() {
|
|
// Guest calls swactor.send with payload_len=0. This should produce
|
|
// a ByteMessage(vec![]) at the destination.
|
|
let wat = r#"
|
|
(module
|
|
(import "swactor" "send" (func $send (param i32 i32 i32)))
|
|
(memory (export "memory") 1)
|
|
(func (export "alloc") (param i32) (result i32)
|
|
i32.const 256 ;; valid allocation
|
|
)
|
|
(func (export "handle") (param $ptr i32) (param $len i32)
|
|
;; Send with the first 32 bytes as dest, zero-length payload
|
|
local.get $ptr
|
|
i32.const 32 ;; payload_ptr (doesn't matter, len is 0)
|
|
i32.const 0 ;; payload_len
|
|
call $send
|
|
)
|
|
)
|
|
"#;
|
|
let wasm = wat::parse_str(wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
// Build a framed message with the inbox address as the first 32 bytes
|
|
let msg = framed_msg(inbox.addr(), b"ignored-payload");
|
|
rt.send_to(addr, msg).unwrap();
|
|
rt.tick();
|
|
|
|
let received = inbox.try_recv().expect("should receive zero-length message");
|
|
assert!(received.0.is_empty(), "payload should be empty");
|
|
}
|
|
|
|
// ── Multiple sequential traps: actor survives repeated failures ──────────────
|
|
|
|
#[test]
|
|
fn actor_survives_multiple_sequential_traps() {
|
|
// After 3 consecutive traps, the actor should still be alive and
|
|
// able to process a non-trapping message.
|
|
let wat = r#"
|
|
(module
|
|
(import "swactor" "send" (func $send (param i32 i32 i32)))
|
|
(memory (export "memory") 1)
|
|
(global $counter (mut i32) (i32.const 0))
|
|
(func (export "alloc") (param i32) (result i32)
|
|
i32.const 256
|
|
)
|
|
(func (export "handle") (param $ptr i32) (param $len i32)
|
|
global.get $counter
|
|
i32.const 3
|
|
i32.lt_u
|
|
if
|
|
;; First 3 calls: trap
|
|
global.get $counter
|
|
i32.const 1
|
|
i32.add
|
|
global.set $counter
|
|
unreachable
|
|
end
|
|
;; 4th+ call: echo the message back using first 32 bytes as dest
|
|
local.get $ptr
|
|
local.get $ptr
|
|
i32.const 32
|
|
i32.add
|
|
local.get $len
|
|
i32.const 32
|
|
i32.sub
|
|
call $send
|
|
)
|
|
)
|
|
"#;
|
|
let wasm = wat::parse_str(wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
// 3 trapping messages
|
|
for _ in 0..3 {
|
|
rt.send_to(addr, framed_msg(inbox.addr(), b"will trap")).unwrap();
|
|
rt.tick();
|
|
assert!(inbox.try_recv().is_none(), "trapped call should produce nothing");
|
|
}
|
|
|
|
// 4th message: should succeed
|
|
let payload = b"survived";
|
|
rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
|
|
rt.tick();
|
|
|
|
let received = inbox.try_recv().expect("actor should work after multiple traps");
|
|
assert_eq!(received.0, payload);
|
|
}
|
|
|
|
// ── Builder: module with start function that traps ───────────────────────────
|
|
|
|
#[test]
|
|
fn module_with_trapping_start_function_returns_error() {
|
|
// WASM modules can have a (start) function that runs during instantiation.
|
|
// If it traps, build() should return an error.
|
|
let wat = r#"
|
|
(module
|
|
(memory (export "memory") 1)
|
|
(func (export "alloc") (param i32) (result i32) i32.const 256)
|
|
(func (export "handle") (param i32 i32))
|
|
(func $init unreachable)
|
|
(start $init)
|
|
)
|
|
"#;
|
|
let wasm = wat::parse_str(wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let result = WasmActorBuilder::new(engine, wasm).build();
|
|
assert!(result.is_err(), "module with trapping start function should fail to build");
|
|
}
|
|
|
|
// ── Self-send: guest sends message back to own address ───────────────────────
|
|
|
|
#[test]
|
|
fn guest_self_send_creates_feedback_loop() {
|
|
// Echo guest sends its payload to a destination. If we set the dest
|
|
// to the actor's OWN address, it creates a feedback loop. The actor
|
|
// should process the self-sent message on the next tick.
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
// Frame: dest=self, payload=[inbox_addr | "hello"]
|
|
// Echo will send [inbox_addr | "hello"] back to itself.
|
|
// On next tick, it receives [inbox_addr | "hello"], echoes "hello" to inbox.
|
|
let inner_msg = framed_msg(inbox.addr(), b"hello");
|
|
let self_msg = framed_msg(&addr, &inner_msg.0);
|
|
|
|
rt.send_to(addr, self_msg).unwrap();
|
|
rt.tick(); // actor echoes inner_msg to self
|
|
rt.tick(); // actor receives inner_msg, echoes "hello" to inbox
|
|
|
|
let received = inbox.try_recv().expect("should receive after self-send loop");
|
|
assert_eq!(received.0, b"hello");
|
|
}
|
|
|
|
// ── Amplification: guest sends many messages in one handle ───────────────────
|
|
|
|
#[test]
|
|
fn guest_sending_many_messages_in_one_handle_all_delivered() {
|
|
// A guest that calls swactor.send N times in a single handle call.
|
|
// All N messages should be delivered via the outbox drain.
|
|
let wat = r#"
|
|
(module
|
|
(import "swactor" "send" (func $send (param i32 i32 i32)))
|
|
(memory (export "memory") 1)
|
|
(func (export "alloc") (param i32) (result i32)
|
|
i32.const 256
|
|
)
|
|
(func (export "handle") (param $ptr i32) (param $len i32)
|
|
;; Send 10 messages, each with 0-byte payload
|
|
;; dest_ptr = $ptr (first 32 bytes of the incoming message)
|
|
(local $i i32)
|
|
(local.set $i (i32.const 0))
|
|
(block $break
|
|
(loop $loop
|
|
(br_if $break (i32.ge_u (local.get $i) (i32.const 10)))
|
|
(call $send (local.get $ptr) (i32.const 32) (i32.const 0))
|
|
(local.set $i (i32.add (local.get $i) (i32.const 1)))
|
|
(br $loop)
|
|
)
|
|
)
|
|
)
|
|
)
|
|
"#;
|
|
let wasm = wat::parse_str(wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
rt.send_to(addr, framed_msg(inbox.addr(), b"x")).unwrap();
|
|
rt.tick();
|
|
|
|
let mut count = 0;
|
|
while inbox.try_recv().is_some() {
|
|
count += 1;
|
|
}
|
|
assert_eq!(count, 10, "guest should have sent exactly 10 messages");
|
|
}
|
|
|
|
// ── Overlapping send regions: dest_ptr and payload_ptr overlap ───────────────
|
|
|
|
#[test]
|
|
fn overlapping_dest_and_payload_in_send_works() {
|
|
// Guest calls send with dest_ptr=0, payload_ptr=16, payload_len=32.
|
|
// The dest region [0..32] and payload region [16..48] overlap.
|
|
// Both are read-only in the host, so this should work without corruption.
|
|
let wat = r#"
|
|
(module
|
|
(import "swactor" "send" (func $send (param i32 i32 i32)))
|
|
(memory (export "memory") 1)
|
|
(func (export "alloc") (param i32) (result i32)
|
|
i32.const 256
|
|
)
|
|
(func (export "handle") (param $ptr i32) (param $len i32)
|
|
;; Overlapping regions
|
|
local.get $ptr ;; dest_ptr (first 32 bytes of message)
|
|
local.get $ptr
|
|
i32.const 16
|
|
i32.add ;; payload_ptr = ptr + 16 (overlaps with dest)
|
|
i32.const 32 ;; payload_len = 32
|
|
call $send
|
|
)
|
|
)
|
|
"#;
|
|
let wasm = wat::parse_str(wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
// Build a message where bytes [0..32] = inbox addr, [32..] = payload
|
|
// Guest reads dest from [0..32] (inbox addr) and payload from [16..48]
|
|
rt.send_to(addr, framed_msg(inbox.addr(), b"overlap-test-padding!")).unwrap();
|
|
rt.tick();
|
|
|
|
let received = inbox.try_recv().expect("overlapping send should deliver");
|
|
assert_eq!(received.0.len(), 32, "payload should be 32 bytes from overlapping region");
|
|
}
|
|
|
|
// ── Type mismatch: non-ByteMessage sent to WASM actor ────────────────────────
|
|
|
|
#[test]
|
|
fn non_byte_message_to_wasm_actor_is_silently_ignored() {
|
|
// Sending a message of the wrong type (not ByteMessage) to a WASM actor.
|
|
// The runtime's handle_any downcast fails, counting a type mismatch.
|
|
// The actor should survive and still process valid ByteMessages.
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
// Send wrong type — u32 instead of ByteMessage
|
|
// This goes through send_any with Box::new(42u32), downcast to ByteMessage fails.
|
|
rt.send_to(addr, 42u32).unwrap();
|
|
rt.tick(); // type mismatch — silently ignored
|
|
|
|
// Actor still alive — send a valid message
|
|
let payload = b"after mismatch";
|
|
rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
|
|
rt.tick();
|
|
|
|
let received = inbox.try_recv().expect("actor should work after type mismatch");
|
|
assert_eq!(received.0, payload);
|
|
}
|
|
|
|
// ── Guest state persistence: mutable global survives across messages ─────────
|
|
|
|
#[test]
|
|
fn guest_mutable_state_persists_across_messages() {
|
|
// A guest module with a mutable global counter. Each handle call increments
|
|
// the counter and includes it in the reply payload. Verifies that the
|
|
// wasmtime Store and linear memory persist between handle() calls.
|
|
let wat = r#"
|
|
(module
|
|
(import "swactor" "send" (func $send (param i32 i32 i32)))
|
|
(memory (export "memory") 1)
|
|
(global $counter (mut i32) (i32.const 0))
|
|
|
|
(func (export "alloc") (param i32) (result i32)
|
|
i32.const 256
|
|
)
|
|
(func (export "handle") (param $ptr i32) (param $len i32)
|
|
;; Increment counter
|
|
global.get $counter
|
|
i32.const 1
|
|
i32.add
|
|
global.set $counter
|
|
|
|
;; Write counter value to memory at offset 200
|
|
(i32.store8 (i32.const 200) (global.get $counter))
|
|
|
|
;; Send counter byte as payload to dest at $ptr
|
|
local.get $ptr ;; dest_ptr (first 32 bytes of message)
|
|
i32.const 200 ;; payload_ptr (counter byte)
|
|
i32.const 1 ;; payload_len
|
|
call $send
|
|
)
|
|
)
|
|
"#;
|
|
let wasm = wat::parse_str(wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
// Send 3 messages, each should get an incrementing counter
|
|
for expected in 1..=3u8 {
|
|
rt.send_to(addr, framed_msg(inbox.addr(), b"x")).unwrap();
|
|
rt.tick();
|
|
let received = inbox.try_recv().expect("should receive counter reply");
|
|
assert_eq!(received.0, vec![expected], "counter should increment per message");
|
|
}
|
|
}
|
|
|
|
// ── Spawn WASM from handler: native actor spawns WASM actor during handle ────
|
|
|
|
struct WasmSpawner {
|
|
engine: SharedEngine,
|
|
wasm_bytes: Vec<u8>,
|
|
}
|
|
|
|
#[derive(Clone)]
|
|
struct SpawnAndForward {
|
|
inbox_addr: ActorAddress,
|
|
payload: Vec<u8>,
|
|
}
|
|
|
|
impl ActorInterface for WasmSpawner {
|
|
type Incoming = SpawnAndForward;
|
|
type Response = ();
|
|
fn handle(&mut self, ctx: &Ctx, msg: SpawnAndForward) {
|
|
let actor = WasmActorBuilder::new(self.engine.clone(), self.wasm_bytes.clone())
|
|
.build()
|
|
.unwrap();
|
|
let wasm_addr = ctx.spawn(actor);
|
|
let _ = ctx.send(wasm_addr.unwrap(), framed_msg(&msg.inbox_addr, &msg.payload));
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn native_handler_spawns_wasm_actor_and_forwards_message() {
|
|
let engine = SharedEngine::new().unwrap();
|
|
let wasm_bytes = guest_wasm("echo");
|
|
|
|
let spawner = WasmSpawner {
|
|
engine: engine.clone(),
|
|
wasm_bytes: wasm_bytes.clone(),
|
|
};
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let spawner_addr = rt.spawn(spawner).unwrap();
|
|
|
|
rt.send_to(
|
|
spawner_addr,
|
|
SpawnAndForward {
|
|
inbox_addr: *inbox.addr(),
|
|
payload: b"spawned-echo".to_vec(),
|
|
},
|
|
)
|
|
.unwrap();
|
|
|
|
// Tick 1: Spawner receives message, spawns WASM actor, sends to it
|
|
rt.tick();
|
|
// Tick 2: WASM actor processes message and echoes to inbox
|
|
rt.tick();
|
|
|
|
let received = inbox.try_recv().expect("dynamically spawned WASM actor should echo");
|
|
assert_eq!(received.0, b"spawned-echo");
|
|
}
|
|
|
|
// ── Bounded mailbox: WASM actor with backpressure ────────────────────────────
|
|
|
|
#[test]
|
|
fn bounded_mailbox_applies_to_wasm_actor() {
|
|
// With a bounded mailbox of capacity 3, sending 10 messages should
|
|
// result in only 3 being processed (DropNewest policy).
|
|
use swactor::runtime::MailboxOverflow;
|
|
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let config = RuntimeConfig {
|
|
default_mailbox_capacity: 3,
|
|
mailbox_overflow: MailboxOverflow::DropNewest,
|
|
..RuntimeConfig::default()
|
|
};
|
|
let rt = Runtime::new(config);
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
// Send 10 messages before any tick — only first 3 should be kept
|
|
for i in 0u8..10 {
|
|
rt.send_to(addr, framed_msg(inbox.addr(), &[i])).unwrap();
|
|
}
|
|
rt.tick();
|
|
|
|
let mut received = Vec::new();
|
|
while let Some(msg) = inbox.try_recv() {
|
|
received.push(msg.0[0]);
|
|
}
|
|
|
|
assert_eq!(received.len(), 3, "bounded mailbox should limit to 3 messages");
|
|
// DropNewest keeps the first 3 sent
|
|
assert_eq!(received, vec![0, 1, 2]);
|
|
}
|
|
|
|
// ── Property: alloc failures never kill the actor ────────────────────────────
|
|
|
|
proptest! {
|
|
#[test]
|
|
fn prop_any_alloc_return_value_never_kills_actor(alloc_val in -100i32..70000) {
|
|
// Regardless of what alloc returns (negative, zero, OOB, valid),
|
|
// sending a message should never kill the actor.
|
|
let alloc_const = format!("i32.const {alloc_val}");
|
|
let wat = format!(r#"
|
|
(module
|
|
(import "swactor" "send" (func $send (param i32 i32 i32)))
|
|
(memory (export "memory") 1)
|
|
(func (export "alloc") (param i32) (result i32)
|
|
{alloc_const}
|
|
)
|
|
(func (export "handle") (param i32 i32))
|
|
)
|
|
"#);
|
|
let wasm = wat::parse_str(&wat).unwrap();
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
// Send a message — should never panic or poison
|
|
rt.send_to(addr, ByteMessage(vec![0u8; 100])).unwrap();
|
|
rt.tick();
|
|
|
|
// Actor should still accept messages (not poisoned)
|
|
let result = rt.send_to(addr, ByteMessage(vec![1]));
|
|
prop_assert!(result.is_ok(), "actor should survive any alloc return value: {alloc_val}");
|
|
}
|
|
}
|
|
|
|
// ── Multi-worker: WASM actors across threads ─────────────────────────────────
|
|
|
|
#[test]
|
|
fn wasm_actor_works_on_multi_worker_runtime() {
|
|
// Spawn a WASM echo actor on a 2-worker runtime and verify message
|
|
// round-trip works across threads. This is a smoke test for Send safety
|
|
// of wasmtime Store<HostState>.
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let config = RuntimeConfig {
|
|
num_threads: 2,
|
|
..RuntimeConfig::default()
|
|
};
|
|
let rt = Runtime::new(config);
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
let payload = b"multi-worker";
|
|
rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
|
|
|
|
// Use run() to drive the runtime on background threads
|
|
let handle = rt.run().unwrap();
|
|
std::thread::sleep(std::time::Duration::from_millis(50));
|
|
|
|
let received = inbox.try_recv().expect("wasm actor should echo on MT runtime");
|
|
assert_eq!(received.0, payload);
|
|
|
|
handle.shutdown();
|
|
}
|
|
|
|
// ── Property-based: arbitrary bytes round-trip through echo ──────────────────
|
|
|
|
proptest! {
|
|
#[test]
|
|
fn prop_echo_roundtrips_arbitrary_bytes(payload in proptest::collection::vec(any::<u8>(), 0..500)) {
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
let msg = framed_msg(inbox.addr(), &payload);
|
|
rt.send_to(addr, msg).unwrap();
|
|
rt.tick();
|
|
|
|
if payload.is_empty() {
|
|
// Echo guest: if total len < 32, no reply (32B addr + 0B payload = 32, but
|
|
// the framed message is 32 + 0 = 32 bytes, and echo checks `len < 32`)
|
|
// Actually: framed_msg produces 32 + payload.len() bytes. When payload
|
|
// is empty, total is 32, and echo checks `if len < 32 { return; }`.
|
|
// len == 32 passes the check! So dest_ptr = ptr, payload_ptr = ptr+32,
|
|
// payload_len = 0 → sends a 0-byte message.
|
|
// Let's just check: if we got something, it matches.
|
|
if let Some(received) = inbox.try_recv() {
|
|
prop_assert_eq!(received.0, payload);
|
|
}
|
|
} else {
|
|
let received = inbox.try_recv().expect("echo should return non-empty payload");
|
|
prop_assert_eq!(received.0, payload);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn prop_double_always_sends_exactly_two_copies(payload in proptest::collection::vec(any::<u8>(), 1..500)) {
|
|
let engine = SharedEngine::new().unwrap();
|
|
let actor = WasmActorBuilder::new(engine, guest_wasm("double"))
|
|
.build()
|
|
.unwrap();
|
|
|
|
let rt = Runtime::new(RuntimeConfig::default());
|
|
let inbox = rt.new_inbox::<ByteMessage>().unwrap();
|
|
let addr = rt.spawn(actor).unwrap();
|
|
|
|
let msg = framed_msg(inbox.addr(), &payload);
|
|
rt.send_to(addr, msg).unwrap();
|
|
rt.tick();
|
|
|
|
let first = inbox.try_recv().expect("double should send first copy");
|
|
let second = inbox.try_recv().expect("double should send second copy");
|
|
prop_assert_eq!(&first.0, &payload);
|
|
prop_assert_eq!(&second.0, &payload);
|
|
prop_assert!(inbox.try_recv().is_none(), "exactly two messages expected");
|
|
}
|
|
}
|