mod common; use common::*; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; // ── Identity Hasher Correctness ──────────────────────────────────────────── /// Given: 200 actors each expecting a unique numbered message /// When: Each actor receives its number and replies with (self_addr, number) /// Then: All 200 replies match -- no message was misrouted by the identity hasher #[test] fn many_actors_all_receive_correct_messages() { #[derive(Clone)] struct NumberedMsg { n: usize, reply_to: ActorAddress, } #[derive(Clone, Debug, PartialEq)] struct NumberedReply { from: ActorAddress, n: usize, } struct NumberedActor; impl ActorInterface for NumberedActor { type Incoming = NumberedMsg; type Response = (); fn handle(&mut self, ctx: &Ctx, msg: NumberedMsg) { let _ = ctx.send( msg.reply_to, NumberedReply { from: ctx.self_addr(), n: msg.n, }, ); } } let rt = std_runtime(RuntimeConfig { max_actors: 300, channel_buffer_size: 1024, num_threads: 1, ..Default::default() }); let inbox = rt.new_inbox::().unwrap(); let inbox_addr = *inbox.addr(); // Spawn 200 actors let mut addrs = Vec::new(); for _ in 0..200 { addrs.push(rt.spawn(NumberedActor).unwrap()); } rt.tick(); // on_start // Send unique numbered message to each for (i, addr) in addrs.iter().enumerate() { rt.send_to( *addr, NumberedMsg { n: i, reply_to: inbox_addr, }, ) .unwrap(); } rt.tick(); // process + reply rt.tick(); // deliver replies // Verify all 200 replies let mut replies: Vec = Vec::new(); while let Some(reply) = inbox.try_recv() { replies.push(reply); } assert_eq!(replies.len(), 200, "should receive exactly 200 replies"); // Verify each reply came from the correct actor with the correct number for (i, addr) in addrs.iter().enumerate() { let reply = replies.iter().find(|r| r.n == i); assert!( reply.is_some(), "missing reply for actor #{i}" ); assert_eq!( reply.unwrap().from, *addr, "reply #{i} came from wrong actor" ); } } /// Given: A 100-actor ring where each actor forwards to the next /// When: A message enters the ring and traverses all 100 hops /// Then: The message completes the full circuit (address_map lookups all correct) #[test] fn ring_routing_unchanged_after_hasher_optimization() { #[derive(Clone)] struct RingHop { hops_remaining: usize, final_dest: ActorAddress, } #[derive(Clone, Debug, PartialEq)] struct RingDone(usize); // total hops completed struct RingNode { next: ActorAddress, } impl ActorInterface for RingNode { type Incoming = RingHop; type Response = (); fn handle(&mut self, ctx: &Ctx, msg: RingHop) { if msg.hops_remaining == 0 { let _ = ctx.send(msg.final_dest, RingDone(100)); } else { let _ = ctx.send( self.next, RingHop { hops_remaining: msg.hops_remaining - 1, final_dest: msg.final_dest, }, ); } } } let rt = std_runtime(RuntimeConfig { max_actors: 200, channel_buffer_size: 1024, num_threads: 1, ..Default::default() }); let inbox = rt.new_inbox::().unwrap(); let inbox_addr = *inbox.addr(); // Build chain backwards: last node sends to inbox, first node receives let mut addrs = Vec::new(); let mut next = inbox_addr; for _ in (0..100).rev() { let node = RingNode { next }; let addr = rt.spawn(node).unwrap(); addrs.push(addr); next = addr; } addrs.reverse(); // addrs[0] is start of chain rt.tick(); // on_start // Inject message at the start rt.send_to( addrs[0], RingHop { hops_remaining: 99, final_dest: inbox_addr, }, ) .unwrap(); // Tick enough times for the message to traverse all 100 actors for _ in 0..110 { rt.tick(); } let result = inbox.try_recv(); assert!(result.is_some(), "ring message should complete all 100 hops"); assert_eq!(result.unwrap(), RingDone(100)); } /// Given: An actor that calls ctx.stop_self() upon receiving a trigger message /// When: The trigger is sent, then 5 more messages are sent, then ticked /// Then: The actor is removed, only messages before stop are processed #[test] fn stop_self_with_pending_messages_still_works() { let processed = Arc::new(AtomicUsize::new(0)); #[derive(Clone)] struct Msg(bool); // true = trigger stop struct StopOnTrigger(Arc); impl ActorInterface for StopOnTrigger { type Incoming = Msg; type Response = (); fn handle(&mut self, ctx: &Ctx, msg: Msg) { self.0.fetch_add(1, Ordering::Relaxed); if msg.0 { ctx.stop_self(); } } } let rt = std_runtime(RuntimeConfig::default()); let p = processed.clone(); let addr = rt.spawn(StopOnTrigger(p)).unwrap(); rt.tick(); // on_start // Send: 2 normal, 1 trigger, 5 more normal rt.send_to(addr, Msg(false)).unwrap(); rt.send_to(addr, Msg(false)).unwrap(); rt.send_to(addr, Msg(true)).unwrap(); // stop trigger rt.send_to(addr, Msg(false)).unwrap(); rt.send_to(addr, Msg(false)).unwrap(); rt.send_to(addr, Msg(false)).unwrap(); rt.send_to(addr, Msg(false)).unwrap(); rt.send_to(addr, Msg(false)).unwrap(); rt.tick(); // process messages -- stops after trigger rt.tick(); // cleanup // Only 3 messages should be processed (2 normal + 1 trigger) assert_eq!( processed.load(Ordering::Relaxed), 3, "should process exactly the messages up to and including the stop trigger" ); // Subsequent sends should fail assert!(rt.send_to(addr, Msg(false)).is_err()); }