From ef87f7e1b92e442851d1dae424453ad1c49b7ae5 Mon Sep 17 00:00:00 2001 From: Developer Date: Thu, 12 Feb 2026 11:11:30 +0000 Subject: [PATCH] feat: per-actor message budget for tick fairness Research across ractor, tokio, Erlang/OTP BEAM, Linux CFS, and libuv revealed that tick_all drained the entire mailbox per actor per tick, allowing one hot actor to starve all others on the same worker. - Add `actor_message_budget` to RuntimeConfig (default: 64 msgs/actor/tick) - Modify tick_all to break after budget messages, yielding to next actor - budget=0 restores unlimited (backward compatible) behavior - 3 new fairness tests validating hot-cold actor scenarios - New fairness benchmark group (cold_latency_under_pressure, throughput_by_budget) - Fix RuntimeConfig struct literals across workspace crates Inspired by BEAM's 4000-reduction budget and tokio's 128-op cooperative budget. All 45 tests pass (42 original + 3 new). Co-Authored-By: Claude Opus 4.6 --- CLAUDE/TASK.md | 46 +++++++++ CLAUDE/notes/baseline_benchmarks.md | 32 ++++++ CLAUDE/notes/constraints.md | 30 ++++++ CLAUDE/notes/progress.md | 58 +++++++++++ CLAUDE/notes/research_synthesis.md | 90 +++++++++++++++++ benches/mt_benchmarks.rs | 1 + benches/runtime_benchmarks.rs | 97 ++++++++++++++++++- crates/python/src/lib.rs | 1 + .../examples/bench_dashboard.rs | 1 + src/config.rs | 10 ++ src/worker.rs | 12 ++- tests/runtime_api.rs | 84 ++++++++++++++++ 12 files changed, 459 insertions(+), 3 deletions(-) create mode 100644 CLAUDE/TASK.md create mode 100644 CLAUDE/notes/baseline_benchmarks.md create mode 100644 CLAUDE/notes/constraints.md create mode 100644 CLAUDE/notes/progress.md create mode 100644 CLAUDE/notes/research_synthesis.md diff --git a/CLAUDE/TASK.md b/CLAUDE/TASK.md new file mode 100644 index 0000000..0a64f0c --- /dev/null +++ b/CLAUDE/TASK.md @@ -0,0 +1,46 @@ +Plan: + You are to improve this codebase via: + - investigating similar codebases + - identifying and summarizing their design decisions when compared to swactor: + - runtime engine + - benchmarking + - testing + - overall performance + - etc. + - implementing improvements based on your anaylsis + +Workflow: + - Read `CLAUDE/TASK.md` and `CLAUDE/notes/progress.md` + - Identify what stage you are on. + - Read and update yourself as necessary. + - Proceed to accomplishing the next task as written in `progress.md` + - For each attempt at any step, keep a record. If you reach attempt 3, step back, document, and try something else. + - When done, because attempt limit or task success: + - update `progress.md` with: + - Completed this session + - Next steps (specific, actionable) + - Open Questions + - Blockers + - make a commit + - compress your context and start the loop again + +Style: + - Do not add to existing modules in the root swactor `src/` they should stay as they are. You may modify but not change module structure. + - Integration tests in `tests/`, benchmark code in `benches/` (cap benchmark execution time at 2 minutes max). You may modify these as you wish. + - Report all your changes to architecture with changes to the `docs/` items + - all notes you wish to keep across iterations shall go in the `CLAUDE/notes/` folder + +Example loop (not restrictive, feel free to ignore if prudent): + - Pick a related codebase and a concept to execute (benchmarks, test coverage, engine performance under various scenarios) + - compare to swactor + - make analysis + - implement plan + - execute + - evaluate + - repeat + +Before git commit: + - all `cargo test` passes, including feature gated material + - if a test fails, investigate do not ignore or delete + - You can combine tests but not skip code paths or delete them for active code + - if a fix takes > 3 attempts, log and move on \ No newline at end of file diff --git a/CLAUDE/notes/baseline_benchmarks.md b/CLAUDE/notes/baseline_benchmarks.md new file mode 100644 index 0000000..3eb2174 --- /dev/null +++ b/CLAUDE/notes/baseline_benchmarks.md @@ -0,0 +1,32 @@ +# Baseline Benchmarks (pre-improvements) + +## Latency (single-threaded) +| Benchmark | Time | +|-----------|------| +| spawn | 1.28 µs | +| message_roundtrip | 2.24 µs | +| send_fire_and_forget | 1.50 µs | +| inbox_creation | 1.63 µs | + +## Throughput (single-threaded) +| Benchmark | Time | Throughput | +|-----------|------|-----------| +| single_actor/100 | 15.0 µs | 6.65 Melem/s | +| single_actor/1000 | 57.5 µs | 17.4 Melem/s | +| single_actor/10000 | 474.6 µs | 21.1 Melem/s | +| multi_actor/10x100 | 80.6 µs | 12.4 Melem/s | +| multi_actor/100x100 | 610.6 µs | 16.4 Melem/s | +| multi_actor/100x1000 | 6.10 ms | 16.4 Melem/s | +| ring/10 | 6.3 µs | 1.75 Melem/s | +| ring/100 | 99.9 µs | 1.01 Melem/s | +| ring/500 | 919.0 µs | 545 Kelem/s | +| spawn/100 | 33.5 µs | 2.99 Melem/s | +| spawn/1000 | 326.2 µs | 3.07 Melem/s | +| spawn/5000 | 1.69 ms | 2.96 Melem/s | + +## Key Observations +- Single-actor throughput scales well: 6.65M → 21.1M msgs/s as batch size grows (amortized overhead) +- Multi-actor throughput lower due to iteration overhead across actors +- Ring throughput degrades with ring size (expected: each message traverses more actors) +- Spawn throughput steady at ~3M/s regardless of batch size +- Message roundtrip latency: 2.24µs (spawn + deliver + process + reply + deliver) diff --git a/CLAUDE/notes/constraints.md b/CLAUDE/notes/constraints.md new file mode 100644 index 0000000..402e536 --- /dev/null +++ b/CLAUDE/notes/constraints.md @@ -0,0 +1,30 @@ +# Task Constraints (from user) + +## Scope of Study +- **Broad survey**: Not just Rust actor frameworks — include: + - Rust: ractor, actix, kameo, coerce, stakker, xactor, bastion + - Non-actor runtimes: tokio, C++ node/libuv event loop + - OS-level: `process` scheduling/logic in operating systems + - Classic actor systems: Erlang/OTP, Akka/Pekko (JVM) + - Any widely-used, well-reputed system + +## Priority & Approach +- **Interleaved**: Pick a topic → analyze competitors → benchmark swactor → improve → repeat +- **Also improve testing methodology and coverage** based on analysis +- **Look at bug report histories** of competitor projects for insights +- Behavioral tests only (Given/When/Then), no white-box/structural tests + +## Code Structure Rules +- **src/ is frozen**: No new files, no new modules, no structural changes. Only modify existing files in-place. +- **No new dependencies** on the root crate (swactor's Cargo.toml) +- May add new crates to `crates/` but they must NOT be pulled into `src/` +- Cap at ~5 new crates — if approaching that, prune back +- Integration tests in `tests/`, benchmarks in `benches/` +- Benchmark execution capped at 2 minutes max +- All notes go in `CLAUDE/notes/` +- Report architecture changes in `docs/` + +## Commit Rules +- All `cargo test` must pass (including feature-gated) +- Never skip/delete tests for active code +- If a fix takes >3 attempts, log and move on diff --git a/CLAUDE/notes/progress.md b/CLAUDE/notes/progress.md new file mode 100644 index 0000000..36c1511 --- /dev/null +++ b/CLAUDE/notes/progress.md @@ -0,0 +1,58 @@ +# Progress Log + +## Current Stage: Phase 1 — Research + First Improvement Cycle + +### Status: Cycle 1 COMPLETE + +## Plan Overview +1. **Phase 0**: Codebase audit — understand current swactor architecture, existing tests, benchmarks ✅ +2. **Phase 1**: Broad survey + interleaved improvements +3. **Phase 2**: Deeper improvements based on findings +4. **Phase 3**: Testing methodology improvements +5. **Phase 4**: Final evaluation & documentation + +## Completed This Session + +### Cycle 1: Fairness (Message Budget) +- **Research**: Studied ractor, tokio, Erlang/OTP BEAM, Linux CFS/EEVDF, libuv +- **Finding**: `tick_all` drained ENTIRE mailbox per actor per tick — critical fairness bug + - BEAM uses 4000 reduction budget, tokio uses 128-op cooperative budget + - Swactor had zero budget — one hot actor could starve all others on same worker +- **Implementation**: Added `actor_message_budget` to `RuntimeConfig` (default: 64) + - Modified `tick_all` to break after `budget` messages per actor + - `budget=0` means unlimited (backward compatible) +- **Tests**: 3 new fairness tests (hot_actor_does_not_starve_cold_actor, unlimited_budget_drains_all, budget_messages_drain_across_multiple_ticks) +- **Benchmarks**: Added fairness benchmark group (cold_latency_under_pressure, throughput_by_budget) +- **Fixes**: Updated RuntimeConfig struct literals across crates (python, runtime-dashboard, mt_benchmarks) +- **Result**: 45 tests pass (42 original + 3 new), all workspace crates compile + +### Research Notes +- Full analysis in `CLAUDE/notes/research_synthesis.md` +- Baseline benchmarks in `CLAUDE/notes/baseline_benchmarks.md` +- Constraints in `CLAUDE/notes/constraints.md` + +## Next Steps +- [ ] **Cycle 2: Stress testing + property-based tests** + - Concurrent spawn+send stress tests + - Multi-threaded fairness validation + - Property: message ordering preserved under budget + - Property: all messages eventually delivered with budget > 0 +- [ ] **Cycle 3: Adaptive backoff with thread parking** + - Replace spinning with condvar-based parking (from tokio parker design) + - Benchmark latency improvement under varying load +- [ ] **Cycle 4: Enhanced benchmarks** + - Message size sensitivity (8B, 64B, 256B, 1KB) + - Latency percentiles (p50, p99, p999) + - Many-to-one fanin contention + - Cross-worker vs same-worker delivery comparison +- [ ] **Cycle 5: Work stealing exploration** + - Evaluate feasibility of actor migration between workers + - BEAM two-tier approach: reactive steal + periodic migration + +## Open Questions +- Should budget be configurable per-actor (not just per-runtime)? +- Is 64 the right default budget? Benchmarks show budget=32 slightly faster for throughput +- Thread parking: how to handle the notification mechanism without adding deps? + +## Blockers +- (none) diff --git a/CLAUDE/notes/research_synthesis.md b/CLAUDE/notes/research_synthesis.md new file mode 100644 index 0000000..058ad6d --- /dev/null +++ b/CLAUDE/notes/research_synthesis.md @@ -0,0 +1,90 @@ +# Research Synthesis: Competitor Analysis + +## Frameworks Studied +1. **Ractor** (Rust) — async task-per-actor on tokio +2. **Tokio** (Rust) — work-stealing async runtime +3. **Erlang/OTP BEAM** — reduction-counted preemptive scheduler +4. **Linux CFS/EEVDF** — vruntime fairness, work stealing, adaptive ticks +5. **libuv/Node.js** — single-threaded event loop with phase-based execution + +## Critical Finding: Swactor Fairness Bug + +`tick_all` in `worker.rs` drains the ENTIRE mailbox for each actor before moving to the next: +```rust +while let Some(msg) = slot.mailbox.pop_front() { + // processes ALL messages for actor A before moving to actor B +} +``` + +If actor A has 10,000 queued messages, all other actors on the same worker are completely starved +until A finishes. Every other runtime studied prevents this: +- **BEAM**: 4000 reductions per process, then preempt +- **Tokio**: 128-256 operation cooperative budget per task +- **libuv**: Round-robin across handlers; no single handler drains completely +- **Linux CFS**: vruntime-based fairness; time slices enforced + +## Ranked Improvement Opportunities + +### P0: Per-Actor Message Budget (Fairness) +- **Impact**: Prevents starvation; critical for production workloads +- **Effort**: Small — modify `tick_all` loop in `worker.rs` +- **Source**: BEAM reductions, tokio coop budget +- **Design**: Process up to N messages per actor per tick, configurable via RuntimeConfig + +### P1: Improved Benchmarks +- **Impact**: Can't improve what you can't measure +- **Effort**: Medium — new benchmark scenarios +- **Source**: Ractor benchmarks, tokio benchmarks +- **New scenarios needed**: + - Fairness: imbalanced load (1 hot actor + 99 cold actors) + - Message size sensitivity (8B, 64B, 256B, 1KB) + - Contention: many-to-one fanin + - Latency percentiles (p50, p99, p999) + - Cross-worker vs same-worker message delivery + +### P2: Better Testing Coverage +- **Impact**: Catches regressions, validates fairness guarantees +- **Effort**: Medium +- **Source**: BEAM testing patterns, tokio Loom +- **New tests needed**: + - Fairness: hot actor doesn't starve cold actors + - Backpressure: tiny buffer under load + - Concurrent spawn+send races + - Multi-threaded delivery guarantees + +### P3: Adaptive Backoff with Thread Parking +- **Impact**: Better latency under varying load; power savings +- **Effort**: Medium — modify `run` loop in `worker.rs` +- **Source**: Tokio parker, Linux NO_HZ +- **Design**: Replace spinning with condvar-based parking; use notification to wake + +### P4: Work Stealing (Future) +- **Impact**: Dynamic load balancing +- **Effort**: Large — significant architectural change +- **Source**: Tokio steal-half, BEAM migration plans +- **Note**: Would require stealing actors between workers, which changes ownership + +## Key Design Comparisons + +| Dimension | Swactor | Ractor | Tokio | BEAM | +|-----------|---------|--------|-------|------| +| Scheduling | Sync tick | Async task-per-actor | Work-stealing | Reduction preemption | +| Fairness | None (drain all) | N/A (1 task = 1 actor) | Coop budget (128) | 4000 reductions | +| Backpressure | Bounded crossbeam ring | None (unbounded) | Bounded MPSC | Off-heap mailbox | +| Work stealing | None | Tokio handles it | Steal-half, N/2 searchers | Steal + migrate | +| Panic handling | catch_unwind + poison | AssertUnwindSafe + supervisor | N/A | Process isolation | +| Message passing | Box downcast | Box downcast | Typed channels | Term copying | + +## Ractor Bug History Lessons +- Destructive `get_children()` — snapshot methods must not mutate state +- OutputPort silent drops — bounded channels need explicit backpressure, not silent overflow +- Remote actor latency regression — cross-runtime messaging needs careful ordering +- Memory bloat per actor — each channel/structure per actor adds up at scale + +## Tokio Patterns to Adopt +1. LIFO slot for same-worker sends (cache locality) +2. Searcher count limiting (N/2 max) for cross-worker stealing +3. Steal-half strategy (amortize overhead) +4. Global queue interval checking (reduce contention) +5. Loom-style testing for lock-free code +6. Single allocation per actor context (hot/cold layout) diff --git a/benches/mt_benchmarks.rs b/benches/mt_benchmarks.rs index 2d7979a..12c0a1c 100644 --- a/benches/mt_benchmarks.rs +++ b/benches/mt_benchmarks.rs @@ -22,6 +22,7 @@ fn mt_config(threads: usize, max_actors: usize, max_messages: usize) -> RuntimeC sleep_increment_us: 10, sleep_max_us: 100, }, + ..Default::default() } } diff --git a/benches/runtime_benchmarks.rs b/benches/runtime_benchmarks.rs index 06ad48b..b4ba60b 100644 --- a/benches/runtime_benchmarks.rs +++ b/benches/runtime_benchmarks.rs @@ -272,5 +272,100 @@ fn throughput_benchmarks(c: &mut Criterion) { group.finish(); } -criterion_group!(benches, latency_benchmarks, throughput_benchmarks); +// --------------------------------------------------------------------------- +// Fairness benchmarks +// --------------------------------------------------------------------------- + +fn fairness_benchmarks(c: &mut Criterion) { + let mut group = c.benchmark_group("fairness"); + + // F1 — Cold-actor latency under hot-actor pressure + // Measures how quickly a cold actor responds when a hot actor has a full mailbox + for hot_msgs in [100, 1_000, 10_000] { + group.bench_with_input( + BenchmarkId::new("cold_latency_under_pressure", hot_msgs), + &hot_msgs, + |b, &hot_msgs| { + b.iter_batched( + || { + let rt = Runtime::new(RuntimeConfig { + max_actors: 100, + channel_buffer_size: hot_msgs + 100, + num_threads: 1, + ..Default::default() + }); + let hot_addr = rt.spawn(SinkActor).unwrap(); + let cold_addr = rt.spawn(EchoActor).unwrap(); + rt.tick(); // register actors + + // Load hot actor + for i in 0..hot_msgs { + rt.send_to(hot_addr, CountMessage(i as u64)).unwrap(); + } + + let inbox = rt.new_inbox::().unwrap(); + let inbox_addr = *inbox.addr(); + (rt, cold_addr, inbox, inbox_addr) + }, + |(rt, cold_addr, inbox, inbox_addr)| { + // Send to cold actor and measure ticks until reply + rt.send_to(cold_addr, PingMessage { reply_to: inbox_addr }).unwrap(); + for _ in 0..200 { + rt.tick(); + if inbox.try_recv().is_some() { + return; + } + } + panic!("Cold actor did not respond"); + }, + BatchSize::SmallInput, + ); + }, + ); + } + + // F2 — Total throughput with budget vs without (ensures budget doesn't kill throughput) + for budget in [0usize, 32, 64, 128] { + let label = if budget == 0 { "unlimited".to_string() } else { format!("{budget}") }; + group.throughput(Throughput::Elements(10_000)); + group.bench_with_input( + BenchmarkId::new("throughput_by_budget", &label), + &budget, + |b, &budget| { + b.iter_batched( + || { + let rt = Runtime::new(RuntimeConfig { + max_actors: 200, + channel_buffer_size: 11_000, + num_threads: 1, + actor_message_budget: budget, + ..Default::default() + }); + let mut addrs = Vec::new(); + for _ in 0..10 { + addrs.push(rt.spawn(SinkActor).unwrap()); + } + rt.tick(); + for &addr in &addrs { + for i in 0..1_000 { + rt.send_to(addr, CountMessage(i as u64)).unwrap(); + } + } + rt + }, + |rt| { + for _ in 0..500 { + rt.tick(); + } + }, + BatchSize::LargeInput, + ); + }, + ); + } + + group.finish(); +} + +criterion_group!(benches, latency_benchmarks, throughput_benchmarks, fairness_benchmarks); criterion_main!(benches); diff --git a/crates/python/src/lib.rs b/crates/python/src/lib.rs index ce41d43..25a95d5 100644 --- a/crates/python/src/lib.rs +++ b/crates/python/src/lib.rs @@ -282,6 +282,7 @@ impl From for RuntimeConfig { sleep_increment_us: py.sleep_increment_us, sleep_max_us: py.sleep_max_us, }, + ..Default::default() } } } diff --git a/crates/runtime-dashboard/examples/bench_dashboard.rs b/crates/runtime-dashboard/examples/bench_dashboard.rs index a227b9b..aef4283 100644 --- a/crates/runtime-dashboard/examples/bench_dashboard.rs +++ b/crates/runtime-dashboard/examples/bench_dashboard.rs @@ -87,6 +87,7 @@ fn bench_config(threads: usize, max_actors: usize, max_messages: usize) -> Runti sleep_increment_us: 10, sleep_max_us: 100, }, + ..Default::default() } } diff --git a/src/config.rs b/src/config.rs index ef8c9b7..7ac08a0 100644 --- a/src/config.rs +++ b/src/config.rs @@ -29,6 +29,10 @@ pub struct RuntimeConfig { pub channel_buffer_size: usize, pub num_threads: usize, pub backoff_policy: BackoffPolicy, + /// Maximum messages processed per actor per tick. + /// Prevents a single actor with a large mailbox from starving others. + /// `0` means unlimited (drain entire mailbox). + pub actor_message_budget: usize, } /// 8kB for the `Box<..>` before counting the rest of the memory @@ -38,6 +42,11 @@ const DEFAULT_MAX_ACTORS: usize = 1_000; /// When the ring is full, messages overflow into an unbounded backup queue. const DEFAULT_CHANNEL_BUFFER_SIZE: usize = 1_000; +/// Default per-actor message budget per tick. +/// Inspired by BEAM's reduction budget (4000) and tokio's cooperative budget (128). +/// 64 is a good default: high enough for throughput, low enough for fairness. +const DEFAULT_ACTOR_MESSAGE_BUDGET: usize = 64; + impl Default for RuntimeConfig { fn default() -> Self { Self { @@ -45,6 +54,7 @@ impl Default for RuntimeConfig { channel_buffer_size: DEFAULT_CHANNEL_BUFFER_SIZE, num_threads: 1, backoff_policy: BackoffPolicy::default(), + actor_message_budget: DEFAULT_ACTOR_MESSAGE_BUDGET, } } } diff --git a/src/worker.rs b/src/worker.rs index a257a30..2a38ce7 100644 --- a/src/worker.rs +++ b/src/worker.rs @@ -84,7 +84,7 @@ impl Worker { pending_local: &pending_local, stats: &self.stats, }; - processed = self.pool.tick_all(&worker_ctx, &self.stats); + processed = self.pool.tick_all(&worker_ctx, &self.stats, tc.config.actor_message_budget); if processed > 0 { did_work = true; } @@ -270,7 +270,10 @@ impl ActorPool { } /// Tick all actors in the pool. Returns the number of messages processed. - pub fn tick_all(&mut self, inner: &dyn ContextInner, stats: &WorkerStats) -> usize { + /// + /// Each actor processes up to `budget` messages per tick (0 = unlimited). + /// This prevents a single hot actor from starving others on the same worker. + pub fn tick_all(&mut self, inner: &dyn ContextInner, stats: &WorkerStats, budget: usize) -> usize { let mut count = 0; for (&addr, slot) in self.actors.iter_mut() { if slot.poisoned { @@ -279,6 +282,7 @@ impl ActorPool { continue; } let ctx = Ctx::new(inner, addr); + let mut actor_count = 0usize; while let Some(msg) = slot.mailbox.pop_front() { let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| { slot.actor.handle_any(&ctx, msg) @@ -302,6 +306,10 @@ impl ActorPool { } } count += 1; + actor_count += 1; + if budget > 0 && actor_count >= budget { + break; + } } } count diff --git a/tests/runtime_api.rs b/tests/runtime_api.rs index 33273f7..6c4355c 100644 --- a/tests/runtime_api.rs +++ b/tests/runtime_api.rs @@ -1194,3 +1194,87 @@ fn custom_thread_count_respected() { // Then the runtime created the requested number of workers assert_eq!(s.num_workers, 4, "runtime should respect the requested thread count"); } + +// ═══════════════════════════════════════════════════════════════════════════ +// Fairness (message budget) +// ═══════════════════════════════════════════════════════════════════════════ + +#[test] +fn hot_actor_does_not_starve_cold_actor() { + // Given: one "hot" actor with 1000 queued messages and one "cold" actor with 1 message + let rt = Runtime::new(RuntimeConfig::default()); + let hot_counter = Arc::new(AtomicUsize::new(0)); + let cold_inbox = rt.new_inbox::().unwrap(); + + let hot_addr = rt.spawn(CountingPingActor { counter: hot_counter.clone() }).unwrap(); + let cold_addr = rt.spawn(PingPongActor).unwrap(); + + // Load the hot actor with 1000 messages (needs a dummy inbox for replies) + let dummy = rt.new_inbox::().unwrap(); + for _ in 0..1000 { + rt.send_to(hot_addr, Ping { reply_to: *dummy.addr() }).unwrap(); + } + // Send one message to the cold actor + rt.send_to(cold_addr, Ping { reply_to: *cold_inbox.addr() }).unwrap(); + + // When: we tick a limited number of times (default budget = 64 msgs/actor/tick) + // After 1 tick: hot actor processes 64, cold actor processes 1 + rt.tick(); + + // Then: the cold actor replied even though the hot actor had 1000 queued messages + let cold_reply = cold_inbox.try_recv(); + assert!( + cold_reply.is_some(), + "cold actor must not be starved by hot actor; message budget should enforce fairness" + ); + // And the hot actor only processed its budget, not all 1000 + let hot_processed = hot_counter.load(Ordering::SeqCst); + assert!( + hot_processed <= 64, + "hot actor should process at most the budget (64) per tick, got {hot_processed}" + ); +} + +#[test] +fn unlimited_budget_drains_all_messages() { + // Given: a runtime with unlimited budget (0) + let rt = Runtime::new(RuntimeConfig { + actor_message_budget: 0, + ..Default::default() + }); + let counter = Arc::new(AtomicUsize::new(0)); + let dummy = rt.new_inbox::().unwrap(); + let addr = rt.spawn(CountingPingActor { counter: counter.clone() }).unwrap(); + + // When: 500 messages are queued and we tick once + for _ in 0..500 { + rt.send_to(addr, Ping { reply_to: *dummy.addr() }).unwrap(); + } + rt.tick(); + rt.tick(); + + // Then: all 500 are processed in a single pass (no budget limit) + let processed = counter.load(Ordering::SeqCst); + assert_eq!(processed, 500, "unlimited budget should drain all messages"); +} + +#[test] +fn budget_messages_drain_across_multiple_ticks() { + // Given: an actor with more messages than the budget + let rt = Runtime::new(RuntimeConfig::default()); // budget=64 + let counter = Arc::new(AtomicUsize::new(0)); + let dummy = rt.new_inbox::().unwrap(); + let addr = rt.spawn(CountingPingActor { counter: counter.clone() }).unwrap(); + + // When: 200 messages are queued + for _ in 0..200 { + rt.send_to(addr, Ping { reply_to: *dummy.addr() }).unwrap(); + } + + // Then: it takes multiple ticks to drain them all + for _ in 0..10 { + rt.tick(); + } + let processed = counter.load(Ordering::SeqCst); + assert_eq!(processed, 200, "all messages should eventually be processed across ticks"); +}