diff --git a/crates/std/src/ctx_ext.rs b/crates/std/src/ctx_ext.rs index cf856bd..fe83613 100644 --- a/crates/std/src/ctx_ext.rs +++ b/crates/std/src/ctx_ext.rs @@ -2,6 +2,7 @@ use swactor::actor::{ActorAddress, ActorInterface, Ctx, Message, MonitorRef}; use swactor::Error; use crate::StdExtension; +use crate::timer_wheel::{CloneMsg, TimerRequest}; fn get_ext<'a>(ctx: &'a Ctx) -> &'a StdExtension { ctx.extension() @@ -96,6 +97,43 @@ impl CtxWatching for Ctx<'_> { } } +/// Timer extension for [`Ctx`]. +/// +/// Provides `send_after_ticks` / `send_interval_ticks` via the per-worker +/// [`TimerWheel`](crate::timer_wheel::TimerWheel). +pub trait CtxTimers { + /// Schedule a one-shot timer: deliver `msg` to `addr` after `ticks` worker ticks. + /// + /// The message is delivered as a normal mailbox message during the fire tick, + /// before `tick_all` processes messages. The timer is tick-counted (deterministic), + /// not wall-clock based. + fn send_after_ticks(&self, addr: ActorAddress, msg: M, ticks: u64); + + /// Schedule a repeating timer: deliver a clone of `msg` to `addr` every `period` ticks. + /// + /// The first delivery happens after `period` ticks. The message is cloned for each + /// delivery. The timer continues until the target actor is stopped/poisoned. + fn send_interval_ticks(&self, addr: ActorAddress, msg: M, period: u64); +} + +impl CtxTimers for Ctx<'_> { + fn send_after_ticks(&self, addr: ActorAddress, msg: M, ticks: u64) { + self.raw_inner().post_worker_request(Box::new(TimerRequest::Once { + dest: addr, + msg: Box::new(msg), + ticks, + })); + } + + fn send_interval_ticks(&self, addr: ActorAddress, msg: M, period: u64) { + self.raw_inner().post_worker_request(Box::new(TimerRequest::Interval { + dest: addr, + msg: Box::new(msg) as Box, + period, + })); + } +} + /// Group extension for [`Ctx`]. /// /// Provides `join_group`, `leave_group`, `publish`, and `group_members` via diff --git a/crates/std/src/extension.rs b/crates/std/src/extension.rs index a7a86f2..713495e 100644 --- a/crates/std/src/extension.rs +++ b/crates/std/src/extension.rs @@ -1,11 +1,12 @@ use std::any::Any; use swactor::actor::{ActorAddress, Down, ExitReason, StopReason}; -use swactor::extension::RuntimeExtension; +use swactor::extension::{RuntimeExtension, WorkerExtension}; use crate::group_registry::GroupRegistry; use crate::monitor_registry::MonitorRegistry; use crate::name_registry::NameRegistry; +use crate::timer_wheel::TimerWheel; use crate::watch_registry::WatchRegistry; /// Standard library extension — provides naming, monitoring, watching, and group registries. @@ -80,4 +81,8 @@ impl RuntimeExtension for StdExtension { fn as_any(&self) -> &dyn Any { self } + + fn create_worker_extension(&self) -> Option> { + Some(Box::new(TimerWheel::new())) + } } diff --git a/crates/std/src/lib.rs b/crates/std/src/lib.rs index 4153dfb..98df904 100644 --- a/crates/std/src/lib.rs +++ b/crates/std/src/lib.rs @@ -4,6 +4,7 @@ pub mod name_registry; pub mod monitor_registry; pub mod watch_registry; pub mod group_registry; +pub(crate) mod timer_wheel; mod extension; mod ctx_ext; mod runtime_ext; @@ -11,5 +12,5 @@ mod runtime_ext; pub use supervisor::{ChildSpec, RestartPolicy, Supervisor, SupervisorStrategy}; pub use router::{Router, RoutingStrategy}; pub use extension::StdExtension; -pub use ctx_ext::{CtxMonitoring, CtxNaming, CtxGroups, CtxWatching}; +pub use ctx_ext::{CtxMonitoring, CtxNaming, CtxGroups, CtxWatching, CtxTimers}; pub use runtime_ext::{RuntimeNaming, RuntimeGroups, RuntimeWatching}; diff --git a/crates/std/src/timer_wheel.rs b/crates/std/src/timer_wheel.rs new file mode 100644 index 0000000..6ca4e94 --- /dev/null +++ b/crates/std/src/timer_wheel.rs @@ -0,0 +1,148 @@ +use std::any::Any; + +use swactor::actor::{ActorAddress, Message}; +use swactor::extension::WorkerExtension; + +// ─── Cloneable Message Trait ──────────────────────────────────────────────── + +/// Type-erased cloneable message for interval timers. +/// Since `Message: Clone`, all actor messages implement this. +pub(crate) trait CloneMsg: Send { + fn clone_boxed(&self) -> Box; +} + +impl CloneMsg for M { + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } +} + +// ─── Timer Request ────────────────────────────────────────────────────────── + +/// Timer request from a handler, queued for processing after tick_all. +pub(crate) enum TimerRequest { + /// One-shot: deliver `msg` to `dest` after `ticks` worker ticks. + Once { + dest: ActorAddress, + msg: Box, + ticks: u64, + }, + /// Repeating: deliver a clone of `msg` to `dest` every `period` ticks. + Interval { + dest: ActorAddress, + msg: Box, + period: u64, + }, +} + +// ─── Timer Wheel ──────────────────────────────────────────────────────────── + +struct OnceTimer { + fire_at: u64, + dest: ActorAddress, + msg: Box, +} + +struct IntervalTimer { + next_fire: u64, + period: u64, + dest: ActorAddress, + msg: Box, +} + +/// Per-worker tick-counting timer wheel. +/// +/// Timers are deterministic (tick-counted, not wall-clock). One-shot timers +/// fire once and are consumed; interval timers fire repeatedly every N ticks. +pub struct TimerWheel { + current_tick: u64, + once_timers: Vec, + interval_timers: Vec, +} + +impl TimerWheel { + pub fn new() -> Self { + Self { + current_tick: 0, + once_timers: Vec::new(), + interval_timers: Vec::new(), + } + } + + /// Advance the tick counter and collect all due timer messages. + fn fire(&mut self) -> Vec<(ActorAddress, Box)> { + self.current_tick += 1; + let tick = self.current_tick; + let mut result = Vec::new(); + + // Fire one-shot timers (swap-remove for O(1) removal) + let mut i = 0; + while i < self.once_timers.len() { + if self.once_timers[i].fire_at <= tick { + let timer = self.once_timers.swap_remove(i); + result.push((timer.dest, timer.msg)); + } else { + i += 1; + } + } + + // Fire interval timers + for timer in &mut self.interval_timers { + if timer.next_fire <= tick { + let msg = timer.msg.clone_boxed(); + result.push((timer.dest, msg)); + timer.next_fire = tick + timer.period; + } + } + + result + } + + /// Remove interval timers whose target was just removed from the worker. + fn gc_dead_intervals(&mut self, dead: &[ActorAddress]) { + if dead.is_empty() { + return; + } + self.interval_timers + .retain(|t| !dead.iter().any(|d| *d == t.dest)); + } + + fn add_once(&mut self, dest: ActorAddress, msg: Box, ticks: u64) { + self.once_timers.push(OnceTimer { + fire_at: self.current_tick + ticks, + dest, + msg, + }); + } + + fn add_interval(&mut self, dest: ActorAddress, msg: Box, period: u64) { + let period = period.max(1); // prevent zero-period infinite loop + self.interval_timers.push(IntervalTimer { + next_fire: self.current_tick + period, + period, + dest, + msg, + }); + } +} + +impl WorkerExtension for TimerWheel { + fn on_tick(&mut self) -> Vec<(ActorAddress, Box)> { + self.fire() + } + + fn handle_request(&mut self, request: Box) { + if let Ok(req) = request.downcast::() { + match *req { + TimerRequest::Once { dest, msg, ticks } => self.add_once(dest, msg, ticks), + TimerRequest::Interval { dest, msg, period } => { + self.add_interval(dest, msg, period) + } + } + } + } + + fn gc_dead(&mut self, dead: &[ActorAddress]) { + self.gc_dead_intervals(dead); + } +} diff --git a/fuzz/Cargo.toml b/fuzz/Cargo.toml index a43cf35..c9a04d2 100644 --- a/fuzz/Cargo.toml +++ b/fuzz/Cargo.toml @@ -11,6 +11,7 @@ cargo-fuzz = true libfuzzer-sys = { version = "0.4", features = ["arbitrary-derive"] } arbitrary = { version = "1", features = ["derive"] } swactor = { path = "..", default-features = true } +swactor-std = { path = "../crates/std" } # Prevent this from interfering with workspaces [workspace] diff --git a/fuzz/fuzz_targets/fuzz_runtime.rs b/fuzz/fuzz_targets/fuzz_runtime.rs index 54349ee..6b3e303 100644 --- a/fuzz/fuzz_targets/fuzz_runtime.rs +++ b/fuzz/fuzz_targets/fuzz_runtime.rs @@ -7,9 +7,12 @@ use std::sync::OnceLock; use arbitrary::Arbitrary; use libfuzzer_sys::fuzz_target; +use std::sync::Arc; + use swactor::actor::{ActorAddress, ActorInterface}; use swactor::config::RuntimeConfig; use swactor::runtime::{Ctx, Inbox, Runtime}; +use swactor_std::{CtxTimers, StdExtension}; // ─── Run Logging ──────────────────────────────────────────────────────────── // FUZZ_LOG=1 → trace every run @@ -881,7 +884,8 @@ fuzz_target!(|input: FuzzInput| { num_threads: 1, ..Default::default() }; - let rt = Runtime::new(config); + let rt = Runtime::new(config) + .with_extension(Arc::new(StdExtension::new())); let mut state = FuzzState::new(rt, tracing); let scenario_limit = input.scenarios.len().min(64); diff --git a/src/actor.rs b/src/actor.rs index 55d6a71..29c4a9f 100644 --- a/src/actor.rs +++ b/src/actor.rs @@ -199,47 +199,19 @@ pub struct Down { /// Not a `Message` — intercepted in `tick_all` before reaching `handle_any`. pub(crate) struct StopSignal; -/// Type-erased cloneable message for interval timers. -/// Since `Message: Clone`, all actor messages can implement this. -pub(crate) trait CloneMsg: Send { - fn clone_boxed(&self) -> Box; -} - -impl CloneMsg for M { - fn clone_boxed(&self) -> Box { - Box::new(self.clone()) - } -} - -/// Timer request from a handler, queued for processing after tick_all. -pub(crate) enum TimerRequest { - /// One-shot: deliver `msg` to `dest` after `ticks` worker ticks. - Once { - dest: ActorAddress, - msg: Box, - ticks: u64, - }, - /// Repeating: deliver a clone of `msg` to `dest` every `period` ticks. - Interval { - dest: ActorAddress, - msg: Box, - period: u64, - }, -} - /// Object-safe inner trait for sending type-erased messages. /// -/// Minimal core interface: send, spawn, stop, timers, and extension access. -/// Registry methods (naming, monitoring, groups) are provided by extension -/// traits in `swactor-std`. +/// Minimal core interface: send, spawn, stop, and extension access. +/// Registry methods (naming, monitoring, groups) and timer scheduling +/// are provided by extension traits in `swactor-std`. #[allow(private_interfaces)] pub trait ContextInner { fn send_any(&self, addr: ActorAddress, msg: Box) -> Result<(), Error>; fn spawn_any(&self, addr: ActorAddress, actor: Box); /// Request graceful stop for an actor. Takes effect after the current message. fn request_stop(&self, addr: ActorAddress); - /// Schedule a timer (one-shot or interval). - fn schedule_timer(&self, request: TimerRequest); + /// Post a request to the per-worker extension (e.g., timer scheduling). + fn post_worker_request(&self, request: Box); /// Access the runtime extension (if installed). fn extension(&self) -> Option<&dyn crate::extension::RuntimeExtension>; } @@ -303,29 +275,4 @@ impl<'a> Ctx<'a> { self.inner.send_any(addr, Box::new(StopSignal)) } - /// Schedule a one-shot timer: deliver `msg` to `addr` after `ticks` worker ticks. - /// - /// The message is delivered as a normal mailbox message during the fire tick, - /// before `tick_all` processes messages. The timer is tick-counted (deterministic), - /// not wall-clock based. - pub fn send_after_ticks(&self, addr: ActorAddress, msg: M, ticks: u64) { - self.inner.schedule_timer(TimerRequest::Once { - dest: addr, - msg: Box::new(msg), - ticks, - }); - } - - /// Schedule a repeating timer: deliver a clone of `msg` to `addr` every `period` ticks. - /// - /// The first delivery happens after `period` ticks. The message is cloned for each - /// delivery. The timer continues until the target actor is stopped/poisoned. - pub fn send_interval_ticks(&self, addr: ActorAddress, msg: M, period: u64) { - self.inner.schedule_timer(TimerRequest::Interval { - dest: addr, - msg: Box::new(msg), - period, - }); - } - } diff --git a/src/extension.rs b/src/extension.rs index 7028bd2..51681ba 100644 --- a/src/extension.rs +++ b/src/extension.rs @@ -25,4 +25,29 @@ pub trait RuntimeExtension: Send + Sync { /// Downcast support for Ctx extension traits. fn as_any(&self) -> &dyn Any; + + /// Create a per-worker extension instance. Called once per worker during init. + /// + /// Unlike `RuntimeExtension` (shared across all workers), each worker owns + /// its own `WorkerExtension` instance for per-worker state like timer wheels. + fn create_worker_extension(&self) -> Option> { + None + } +} + +/// Per-worker extension state, created by [`RuntimeExtension::create_worker_extension`]. +/// +/// Each worker owns its own instance. Core calls these methods during tick phases: +/// - `on_tick`: phase 2.5 — before tick_all, returns messages to deliver +/// - `handle_request`: phase 5.5 — processes deferred requests from handlers +/// - `gc_dead`: after cleanup_dead — removes state for dead actors +pub trait WorkerExtension: Send { + /// Called each tick before tick_all. Returns messages to deliver. + fn on_tick(&mut self) -> Vec<(ActorAddress, Box)>; + + /// Process a deferred request posted during handle() via `post_worker_request`. + fn handle_request(&mut self, request: Box); + + /// Clean up state for dead actors. + fn gc_dead(&mut self, dead: &[ActorAddress]); } diff --git a/src/runtime.rs b/src/runtime.rs index 7af01b0..ce04231 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -7,7 +7,7 @@ use std::thread::{self, JoinHandle}; use std::thread::Thread; use crate::Instant; -use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message, StopSignal, TimerRequest}; +use crate::actor::{Actor, ActorAddress, ActorInterface, AnyActor, Message, StopSignal}; use crate::channel::{Receiver, Sender}; // Re-export config types so existing code using `runtime::RuntimeConfig` still works pub use crate::config::{BackoffPolicy, MailboxOverflow, RuntimeConfig}; @@ -245,6 +245,12 @@ impl Runtime { /// /// Must be called before `run()` or `tick()`. pub fn with_extension(mut self, ext: Arc) -> Self { + // Create per-worker extensions (e.g., timer wheels) + for worker in self.tick_workers.get_mut().iter_mut() { + if let Some(wext) = ext.create_worker_extension() { + worker.worker_ext = Some(wext); + } + } self.extension = Some(ext); self } @@ -503,11 +509,10 @@ impl ContextInner for Runtime { } } - fn schedule_timer(&self, _request: TimerRequest) { - // Timers are per-worker and tick-counted; scheduling from outside + fn post_worker_request(&self, _request: Box) { + // Worker requests (e.g., timers) are per-worker; posting from outside // a worker context (e.g., rt.spawn() callback) is not supported. - // Use rt.send_to() with a delay loop instead. - eprintln!("swactor: schedule_timer called outside worker context — ignored"); + eprintln!("swactor: post_worker_request called outside worker context — ignored"); } fn extension(&self) -> Option<&dyn RuntimeExtension> { diff --git a/src/worker.rs b/src/worker.rs index 9f6fc09..e1c0190 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -6,107 +6,14 @@ use std::sync::Arc; use std::thread; use crate::Instant; -use crate::actor::{ActorAddress, AnyActor, CloneMsg, ContextInner, Ctx, StopReason, StopSignal, TimerRequest}; +use crate::actor::{ActorAddress, AnyActor, ContextInner, Ctx, StopReason, StopSignal}; use crate::channel::Receiver; use crate::config::MailboxOverflow; use crate::delivery::{AddrBuildHasher, AddrMap, Envelope, TickContext, WorkerId}; use crate::stats::{ActorSnapshot, TickTiming, WorkerStats}; use crate::Error; -// ─── Per-Worker Timer Wheel ───────────────────────────────────────────────── - -struct OnceTimer { - fire_at: u64, - dest: ActorAddress, - msg: Box, -} - -struct IntervalTimer { - next_fire: u64, - period: u64, - dest: ActorAddress, - msg: Box, -} - -/// Per-worker tick-counting timer wheel. -/// -/// Timers are deterministic (tick-counted, not wall-clock). One-shot timers -/// fire once and are consumed; interval timers fire repeatedly every N ticks. -struct TimerWheel { - current_tick: u64, - once_timers: Vec, - interval_timers: Vec, -} - -impl TimerWheel { - fn new() -> Self { - Self { - current_tick: 0, - once_timers: Vec::new(), - interval_timers: Vec::new(), - } - } - - /// Advance the tick counter and collect all due timer messages. - /// Returns the messages to be routed by the caller (may target local or remote actors/inboxes). - fn fire(&mut self) -> Vec<(ActorAddress, Box)> { - self.current_tick += 1; - let tick = self.current_tick; - let mut result = Vec::new(); - - // Fire one-shot timers (swap-remove for O(1) removal) - let mut i = 0; - while i < self.once_timers.len() { - if self.once_timers[i].fire_at <= tick { - let timer = self.once_timers.swap_remove(i); - result.push((timer.dest, timer.msg)); - } else { - i += 1; - } - } - - // Fire interval timers - for timer in &mut self.interval_timers { - if timer.next_fire <= tick { - let msg = timer.msg.clone_boxed(); - result.push((timer.dest, msg)); - timer.next_fire = tick + timer.period; - } - } - - result - } - - /// Remove interval timers whose target was just removed from the worker. - /// Only GCs timers for addresses in `dead` — inboxes and cross-worker actors - /// are not in the local pool but are still valid targets. - fn gc_dead_intervals(&mut self, dead: &[ActorAddress]) { - if dead.is_empty() { - return; - } - self.interval_timers.retain(|t| !dead.iter().any(|d| *d == t.dest)); - } - - /// Add a one-shot timer. - fn add_once(&mut self, dest: ActorAddress, msg: Box, ticks: u64) { - self.once_timers.push(OnceTimer { - fire_at: self.current_tick + ticks, - dest, - msg, - }); - } - - /// Add an interval timer. First fire is after `period` ticks. - fn add_interval(&mut self, dest: ActorAddress, msg: Box, period: u64) { - let period = period.max(1); // prevent zero-period infinite loop - self.interval_timers.push(IntervalTimer { - next_fire: self.current_tick + period, - period, - dest, - msg, - }); - } -} +use crate::extension::WorkerExtension; // ─── Worker ───────────────────────────────────────────────────────────────── @@ -119,8 +26,8 @@ pub(crate) struct Worker { stats: Arc, /// Reusable scratch buffer for building per-actor snapshots. snapshot_buf: Vec, - /// Per-worker tick-counting timer wheel. - timers: TimerWheel, + /// Per-worker extension (e.g., timer wheel). Created by RuntimeExtension factory. + pub(crate) worker_ext: Option>, } impl Worker { @@ -139,7 +46,7 @@ impl Worker { spawn_rx, stats, snapshot_buf: Vec::new(), - timers: TimerWheel::new(), + worker_ext: None, } } @@ -175,32 +82,31 @@ impl Worker { } let t2 = Instant::now(); - // 2.5. Fire due timers → deliver to mailboxes before tick_all - let timer_msgs = self.timers.fire(); - for (dest, msg) in timer_msgs { - if self.pool.contains(&dest) { - // Same-worker: deliver directly to actor's mailbox - self.pool.deliver(&dest, msg); - } else { - // Inbox or cross-worker: route through address map / inbox registry - match tc.address_map.lookup(&dest) { - Some(wid) => { - tc.transfer_txs[wid.as_usize()].send(Envelope::new(dest, msg)); - crate::runtime::notify_worker(tc.worker_threads, wid.as_usize()); - } - None => { - let _ = tc.inbox_registry.try_deliver(dest, msg); + // 2.5. Fire per-worker extension (e.g., timers) → deliver before tick_all + if let Some(ext) = &mut self.worker_ext { + for (dest, msg) in ext.on_tick() { + if self.pool.contains(&dest) { + self.pool.deliver(&dest, msg); + } else { + match tc.address_map.lookup(&dest) { + Some(wid) => { + tc.transfer_txs[wid.as_usize()].send(Envelope::new(dest, msg)); + crate::runtime::notify_worker(tc.worker_threads, wid.as_usize()); + } + None => { + let _ = tc.inbox_registry.try_deliver(dest, msg); + } } } + did_work = true; } - did_work = true; } // 3. Tick all actors with WorkerContext let pending_local: RefCell)>> = RefCell::new(Vec::new()); let stop_requests: RefCell> = RefCell::new(Vec::new()); - let timer_requests: RefCell> = RefCell::new(Vec::new()); + let worker_requests: RefCell>> = RefCell::new(Vec::new()); let processed; { @@ -209,7 +115,7 @@ impl Worker { tc, pending_local: &pending_local, stop_requests: &stop_requests, - timer_requests: &timer_requests, + worker_requests: &worker_requests, stats: &self.stats, }; processed = self.pool.tick_all(&worker_ctx, &self.stats, tc.config.actor_message_budget, &stop_requests); @@ -245,15 +151,10 @@ impl Worker { self.pool.deliver(&addr, msg); } - // 5.5. Process timer requests from handlers - for request in timer_requests.into_inner() { - match request { - TimerRequest::Once { dest, msg, ticks } => { - self.timers.add_once(dest, msg, ticks); - } - TimerRequest::Interval { dest, msg, period } => { - self.timers.add_interval(dest, msg, period); - } + // 5.5. Process worker extension requests from handlers (e.g., timer scheduling) + if let Some(ext) = &mut self.worker_ext { + for request in worker_requests.into_inner() { + ext.handle_request(request); } } @@ -308,14 +209,14 @@ impl Worker { let cleanup_pending: RefCell)>> = RefCell::new(Vec::new()); let cleanup_stops: RefCell> = RefCell::new(Vec::new()); - let cleanup_timers: RefCell> = RefCell::new(Vec::new()); + let cleanup_requests: RefCell>> = RefCell::new(Vec::new()); let dead = { let cleanup_ctx = WorkerContext { worker_id: self.id, tc, pending_local: &cleanup_pending, stop_requests: &cleanup_stops, - timer_requests: &cleanup_timers, + worker_requests: &cleanup_requests, stats: &self.stats, }; let dead = self.pool.cleanup_dead(&cleanup_ctx); @@ -362,9 +263,11 @@ impl Worker { self.pool.deliver(&addr, msg); } - // GC orphaned interval timers for actors that were just removed - let dead_addrs: Vec = dead.iter().map(|(a, _)| *a).collect(); - self.timers.gc_dead_intervals(&dead_addrs); + // GC per-worker extension state for dead actors (e.g., orphaned interval timers) + if let Some(ext) = &mut self.worker_ext { + let dead_addrs: Vec = dead.iter().map(|(a, _)| *a).collect(); + ext.gc_dead(&dead_addrs); + } did_work } @@ -408,7 +311,7 @@ struct WorkerContext<'a> { tc: &'a TickContext<'a>, pending_local: &'a RefCell)>>, stop_requests: &'a RefCell>, - timer_requests: &'a RefCell>, + worker_requests: &'a RefCell>>, stats: &'a WorkerStats, } @@ -445,8 +348,8 @@ impl ContextInner for WorkerContext<'_> { self.stop_requests.borrow_mut().push(addr); } - fn schedule_timer(&self, request: TimerRequest) { - self.timer_requests.borrow_mut().push(request); + fn post_worker_request(&self, request: Box) { + self.worker_requests.borrow_mut().push(request); } fn extension(&self) -> Option<&dyn crate::extension::RuntimeExtension> { diff --git a/tests/proptest_runtime.rs b/tests/proptest_runtime.rs index 832f9b0..96cacbe 100644 --- a/tests/proptest_runtime.rs +++ b/tests/proptest_runtime.rs @@ -8,9 +8,12 @@ use std::collections::HashMap; use proptest::prelude::*; use proptest_state_machine::{prop_state_machine, ReferenceStateMachine, StateMachineTest}; +use std::sync::Arc; + use swactor::actor::{ActorAddress, ActorInterface}; use swactor::config::{MailboxOverflow, RuntimeConfig}; use swactor::runtime::{Ctx, Inbox, Runtime}; +use swactor_std::{CtxTimers, StdExtension}; // ─── Shared Actor Types ──────────────────────────────────────────────────── @@ -141,7 +144,8 @@ proptest! { /// One-shot timer fires at exactly the right tick for any delay. #[test] fn one_shot_timer_fires_at_correct_tick(delay in 1u64..20) { - let rt = Runtime::new(RuntimeConfig::default()); + let rt = Runtime::new(RuntimeConfig::default()) + .with_extension(Arc::new(StdExtension::new())); let inbox = rt.new_inbox::().unwrap(); struct TimerActor { target: ActorAddress, delay: u64 } @@ -175,7 +179,8 @@ proptest! { /// Interval timer fires at correct periodic ticks for any period. #[test] fn interval_timer_fires_at_correct_period(period in 1u64..10) { - let rt = Runtime::new(RuntimeConfig::default()); + let rt = Runtime::new(RuntimeConfig::default()) + .with_extension(Arc::new(StdExtension::new())); let inbox = rt.new_inbox::().unwrap(); struct IntervalActor { target: ActorAddress, period: u64 } diff --git a/tests/runtime_api.rs b/tests/runtime_api.rs index 0a85827..a7e3fb1 100644 --- a/tests/runtime_api.rs +++ b/tests/runtime_api.rs @@ -3,8 +3,8 @@ use std::sync::Arc; use swactor::actor::{ActorAddress, ActorInterface, Down, MonitorRef, StopReason}; use swactor_std::{ - ChildSpec, CtxGroups, CtxMonitoring, CtxNaming, RestartPolicy, Router, RoutingStrategy, - RuntimeGroups, RuntimeNaming, StdExtension, Supervisor, SupervisorStrategy, + ChildSpec, CtxGroups, CtxMonitoring, CtxNaming, CtxTimers, RestartPolicy, Router, + RoutingStrategy, RuntimeGroups, RuntimeNaming, StdExtension, Supervisor, SupervisorStrategy, }; use swactor::runtime::{Ctx, Inbox, MailboxOverflow, Runtime, RuntimeConfig};