## Coverage Report

#### Created: 2025-09-08 21:26

next uncovered line (L), next uncovered region (R), next uncovered branch (B)

    /home/noah/src/ruchy/src/runtime/actor.rs

    Line
    Count
    Source
    1
    #![allow(clippy::print_stdout, clippy::print_stderr)]
    2
    //! Actor system runtime with supervision trees
    3
    //!
    4
    //! This module implements a robust actor system inspired by Erlang/OTP and Akka,
    5
    //! with supervision trees for fault tolerance and message passing capabilities.
    6
    7
    use anyhow::{anyhow, Result};
    8
    use serde::{Deserialize, Serialize};
    9
    use std::collections::HashMap;
    10
    use std::sync::{mpsc, Arc, Mutex};
    11
    use std::thread::{self, JoinHandle};
    12
    use std::time::Duration;
    13
    14
    /// Unique identifier for actors
    15
    #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
    16
    pub struct ActorId(pub u64);
    17
    18
    impl std::fmt::Display for ActorId {
    19
    0
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
    20
    0
            write!(f, "actor-{}", self.0)
    21
    0
        }
    22
    }
    23
    24
    /// Actor reference for sending messages
    25
    #[derive(Debug, Clone)]
    26
    pub struct ActorRef {
    27
        pub id: ActorId,
    28
        pub name: String,
    29
        sender: mpsc::Sender<ActorMessage>,
    30
    }
    31
    32
    impl ActorRef {
    33
        /// Send a message to this actor (fire-and-forget)
    34
        ///
    35
        /// # Errors
    36
        ///
    37
        /// Returns an error if the actor is no longer running
    38
        /// # Errors
    39
        ///
    40
        /// Returns an error if the operation fails
    41
    0
        pub fn send(&self, message: Message) -> Result<()> {
    42
    0
            self.sender
    43
    0
                .send(ActorMessage::UserMessage(message))
    44
    0
                .map_err(|_| anyhow!("Actor {} is no longer running", self.id))?;
    45
    0
            Ok(())
    46
    0
        }
    47
    48
        /// Ask a message to this actor and wait for a response
    49
        ///
    50
        /// # Errors
    51
        ///
    52
        /// Returns an error if:
    53
        /// - The actor is no longer running
    54
        /// - The timeout expires before receiving a response
    55
        /// # Errors
    56
        ///
    57
        /// Returns an error if the operation fails
    58
    0
        pub fn ask(&self, message: Message, timeout: Duration) -> Result<Message> {
    59
    0
            let (response_tx, response_rx) = mpsc::channel();
    60
    61
    0
            self.sender
    62
    0
                .send(ActorMessage::AskMessage {
    63
    0
                    message,
    64
    0
                    response: response_tx,
    65
    0
                })
    66
    0
                .map_err(|_| anyhow!("Actor {} is no longer running", self.id))?;
    67
    68
    0
            response_rx
    69
    0
                .recv_timeout(timeout)
    70
    0
                .map_err(|_| anyhow!("Timeout waiting for response from {}", self.id))
    71
    0
        }
    72
    }
    73
    74
    /// Message that can be sent between actors
    75
    #[derive(Debug, Clone, Serialize, Deserialize)]
    76
    pub enum Message {
    77
        /// System messages for actor lifecycle
    78
        Start,
    79
        Stop,
    80
        Restart,
    81
        /// User-defined messages
    82
        User(String, Vec<MessageValue>),
    83
        /// Error notification
    84
        Error(String),
    85
        /// Supervision messages
    86
        ChildFailed(ActorId, String),
    87
        ChildRestarted(ActorId),
    88
    }
    89
    90
    /// Values that can be passed in messages
    91
    #[derive(Debug, Clone, Serialize, Deserialize)]
    92
    pub enum MessageValue {
    93
        String(String),
    94
        Integer(i64),
    95
        Float(f64),
    96
        Bool(bool),
    97
        List(Vec<MessageValue>),
    98
        Map(HashMap<String, MessageValue>),
    99
        ActorRef(ActorId),
    100
    }
    101
    102
    /// Internal actor message envelope
    103
    #[derive(Debug)]
    104
    enum ActorMessage {
    105
        UserMessage(Message),
    106
        AskMessage {
    107
            message: Message,
    108
            response: mpsc::Sender<Message>,
    109
        },
    110
        SystemShutdown,
    111
    }
    112
    113
    /// Actor behavior trait
    114
    pub trait ActorBehavior: Send + 'static {
    115
        /// Called when the actor starts
    116
        ///
    117
        /// # Errors
    118
        ///
    119
        /// Returns an error if initialization fails
    120
    0
        fn pre_start(&mut self, _ctx: &mut ActorContext) -> Result<()> {
    121
    0
            Ok(())
    122
    0
        }
    123
    124
        /// Called when the actor stops
    125
        ///
    126
        /// # Errors
    127
        ///
    128
        /// Returns an error if cleanup fails
    129
    0
        fn post_stop(&mut self, _ctx: &mut ActorContext) -> Result<()> {
    130
    0
            Ok(())
    131
    0
        }
    132
    133
        /// Called when the actor is about to restart
    134
        ///
    135
        /// # Errors
    136
        ///
    137
        /// Returns an error if pre-restart logic fails
    138
    0
        fn pre_restart(&mut self, _ctx: &mut ActorContext, _reason: &str) -> Result<()> {
    139
    0
            Ok(())
    140
    0
        }
    141
    142
        /// Called after the actor has restarted
    143
        ///
    144
        /// # Errors
    145
        ///
    146
        /// Returns an error if post-restart logic fails
    147
    0
        fn post_restart(&mut self, _ctx: &mut ActorContext, _reason: &str) -> Result<()> {
    148
    0
            Ok(())
    149
    0
        }
    150
    151
        /// Handle incoming messages
    152
        ///
    153
        /// # Errors
    154
        ///
    155
        /// Returns an error if message processing fails
    156
        fn receive(&mut self, message: Message, ctx: &mut ActorContext) -> Result<Option<Message>>;
    157
    158
        /// Handle actor supervision - called when a child actor fails
    159
    0
        fn supervisor_strategy(&mut self, _child: ActorId, _reason: &str) -> SupervisorDirective {
    160
    0
            SupervisorDirective::Restart
    161
    0
        }
    162
    }
    163
    164
    /// Supervisor strategy for handling child actor failures
    165
    #[derive(Debug, Clone)]
    166
    pub enum SupervisorDirective {
    167
        /// Restart the failed child
    168
        Restart,
    169
        /// Stop the failed child
    170
        Stop,
    171
        /// Escalate the failure to the parent supervisor
    172
        Escalate,
    173
        /// Resume the child (ignore the failure)
    174
        Resume,
    175
    }
    176
    177
    /// Actor context provided during message handling
    178
    pub struct ActorContext {
    179
        pub actor_id: ActorId,
    180
        pub actor_name: String,
    181
        pub supervisor: Option<ActorRef>,
    182
        pub children: HashMap<ActorId, ActorRef>,
    183
        system: Arc<Mutex<ActorSystem>>,
    184
    }
    185
    186
    impl ActorContext {
    187
        /// Spawn a child actor under this actor's supervision
    188
        /// # Errors
    189
        ///
    190
        /// Returns an error if the operation fails
    191
    0
        pub fn spawn_child<B: ActorBehavior>(&mut self, name: String, behavior: B) -> Result<ActorRef> {
    192
    0
            let mut system = self
    193
    0
                .system
    194
    0
                .lock()
    195
    0
                .map_err(|_| anyhow!("Actor system mutex poisoned"))?;
    196
    0
            let actor_ref = system.spawn_supervised(name, Box::new(behavior), Some(self.actor_id))?;
    197
    0
            self.children.insert(actor_ref.id, actor_ref.clone());
    198
    0
            Ok(actor_ref)
    199
    0
        }
    200
    201
        /// Stop a child actor
    202
        /// # Errors
    203
        ///
    204
        /// Returns an error if the operation fails
    205
    0
        pub fn stop_child(&mut self, child_id: ActorId) -> Result<()> {
    206
    0
            if let Some(child_ref) = self.children.remove(&child_id) {
    207
    0
                child_ref.send(Message::Stop)?;
    208
    0
            }
    209
    0
            Ok(())
    210
    0
        }
    211
    212
        /// Get reference to self
    213
        ///
    214
        /// # Errors
    215
        ///
    216
        /// Returns an error if the actor reference cannot be retrieved
    217
        /// # Errors
    218
        ///
    219
        /// Returns an error if the operation fails
    220
    0
        pub fn get_self(&self) -> Result<ActorRef> {
    221
    0
            let system = self
    222
    0
                .system
    223
    0
                .lock()
    224
    0
                .map_err(|_| anyhow!("Actor system mutex poisoned"))?;
    225
    0
            system
    226
    0
                .get_actor_ref(self.actor_id)
    227
    0
                .ok_or_else(|| anyhow!("Actor not found"))
    228
    0
        }
    229
    230
        /// Find actor by name
    231
    0
        pub fn find_actor(&self, name: &str) -> Option<ActorRef> {
    232
    0
            let system = self.system.lock().ok()?;
    233
    0
            system.find_actor_by_name(name)
    234
    0
        }
    235
    }
    236
    237
    /// Actor runtime information
    238
    struct ActorRuntime {
    239
        id: ActorId,
    240
        name: String,
    241
        behavior: Box<dyn ActorBehavior>,
    242
        receiver: mpsc::Receiver<ActorMessage>,
    243
        sender: mpsc::Sender<ActorMessage>,
    244
        supervisor: Option<ActorId>,
    245
        children: HashMap<ActorId, ActorRef>,
    246
        system: Arc<Mutex<ActorSystem>>,
    247
        handle: Option<JoinHandle<()>>,
    248
    }
    249
    250
    impl ActorRuntime {
    251
    0
        fn new(
    252
    0
            id: ActorId,
    253
    0
            name: String,
    254
    0
            behavior: Box<dyn ActorBehavior>,
    255
    0
            supervisor: Option<ActorId>,
    256
    0
            system: Arc<Mutex<ActorSystem>>,
    257
    0
        ) -> Self {
    258
    0
            let (sender, receiver) = mpsc::channel();
    259
    260
    0
            Self {
    261
    0
                id,
    262
    0
                name,
    263
    0
                behavior,
    264
    0
                receiver,
    265
    0
                sender,
    266
    0
                supervisor,
    267
    0
                children: HashMap::new(),
    268
    0
                system,
    269
    0
                handle: None,
    270
    0
            }
    271
    0
        }
    272
    273
    0
        fn start(&mut self) -> ActorRef {
    274
    0
            let actor_ref = ActorRef {
    275
    0
                id: self.id,
    276
    0
                name: self.name.clone(),
    277
    0
                sender: self.sender.clone(),
    278
    0
            };
    279
    280
    0
            let id = self.id;
    281
    0
            let name = self.name.clone();
    282
    0
            let receiver = std::mem::replace(&mut self.receiver, mpsc::channel().1);
    283
    0
            let mut behavior = std::mem::replace(&mut self.behavior, Box::new(DummyBehavior));
    284
    0
            let supervisor = self.supervisor;
    285
    0
            let system = self.system.clone();
    286
    0
            let children = self.children.clone();
    287
    288
    0
            let handle = thread::spawn(move || {
    289
    0
                let mut ctx = ActorContext {
    290
    0
                    actor_id: id,
    291
    0
                    actor_name: name.clone(),
    292
    0
                    supervisor: supervisor.and_then(|sup_id| system.lock().ok()?.get_actor_ref(sup_id)),
    293
    0
                    children,
    294
    0
                    system: system.clone(),
    295
                };
    296
    297
                // Initialize actor
    298
    0
                if let Err(e) = behavior.pre_start(&mut ctx) {
    299
    0
                    eprintln!("Actor {name} failed to start: {e}");
    300
    0
                    return;
    301
    0
                }
    302
    303
                // Main message loop
    304
                loop {
    305
    0
                    match receiver.recv() {
    306
    0
                        Ok(ActorMessage::UserMessage(msg)) => {
    307
    0
                            match behavior.receive(msg, &mut ctx) {
    308
    0
                                Ok(_) => {}
    309
    0
                                Err(e) => {
    310
    0
                                    eprintln!("Actor {name} error handling message: {e}");
    311
                                    // Notify supervisor of failure
    312
    0
                                    if let Some(sup) = &ctx.supervisor {
    313
    0
                                        let _ = sup.send(Message::ChildFailed(id, e.to_string()));
    314
    0
                                    }
    315
                                }
    316
                            }
    317
                        }
    318
    0
                        Ok(ActorMessage::AskMessage { message, response }) => {
    319
    0
                            match behavior.receive(message, &mut ctx) {
    320
    0
                                Ok(Some(reply)) => {
    321
    0
                                    let _ = response.send(reply);
    322
    0
                                }
    323
    0
                                Ok(None) => {
    324
    0
                                    let _ = response.send(Message::Error("No response".to_string()));
    325
    0
                                }
    326
    0
                                Err(e) => {
    327
    0
                                    let _ = response.send(Message::Error(e.to_string()));
    328
                                    // Notify supervisor of failure
    329
    0
                                    if let Some(sup) = &ctx.supervisor {
    330
    0
                                        let _ = sup.send(Message::ChildFailed(id, e.to_string()));
    331
    0
                                    }
    332
                                }
    333
                            }
    334
                        }
    335
                        Ok(ActorMessage::SystemShutdown) => {
    336
    0
                            break;
    337
                        }
    338
                        Err(_) => {
    339
                            // Channel closed, exit
    340
    0
                            break;
    341
                        }
    342
                    }
    343
                }
    344
    345
                // Cleanup
    346
    0
                let _ = behavior.post_stop(&mut ctx);
    347
    0
            });
    348
    349
    0
            self.handle = Some(handle);
    350
    0
            actor_ref
    351
    0
        }
    352
    353
    0
        fn stop(&mut self) {
    354
    0
            if let Some(handle) = self.handle.take() {
    355
    0
                let _ = self.sender.send(ActorMessage::SystemShutdown);
    356
    0
                let _ = handle.join();
    357
    0
            }
    358
    0
        }
    359
    }
    360
    361
    /// Dummy behavior for placeholder
    362
    struct DummyBehavior;
    363
    364
    impl ActorBehavior for DummyBehavior {
    365
    0
        fn receive(&mut self, _message: Message, _ctx: &mut ActorContext) -> Result<Option<Message>> {
    366
    0
            Ok(None)
    367
    0
        }
    368
    }
    369
    370
    /// Actor system managing all actors and supervision
    371
    pub struct ActorSystem {
    372
        actors: HashMap<ActorId, ActorRuntime>,
    373
        actor_names: HashMap<String, ActorId>,
    374
        next_id: u64,
    375
    }
    376
    377
    impl ActorSystem {
    378
        /// Create a new actor system
    379
    0
        pub fn new() -> Arc<Mutex<Self>> {
    380
    0
            Arc::new(Mutex::new(Self {
    381
    0
                actors: HashMap::new(),
    382
    0
                actor_names: HashMap::new(),
    383
    0
                next_id: 1,
    384
    0
            }))
    385
    0
        }
    386
    387
        /// Spawn a new actor in the system
    388
        ///
    389
        /// # Errors
    390
        ///
    391
        /// Returns an error if an actor with the same name already exists
    392
        /// # Errors
    393
        ///
    394
        /// Returns an error if the operation fails
    395
    0
        pub fn spawn<B: ActorBehavior>(&mut self, name: String, behavior: B) -> Result<ActorRef> {
    396
    0
            self.spawn_supervised(name, Box::new(behavior), None)
    397
    0
        }
    398
    399
        /// Spawn a supervised actor
    400
        ///
    401
        /// # Errors
    402
        ///
    403
        /// Returns an error if:
    404
        /// - An actor with the same name already exists
    405
        /// - The supervisor doesn't exist (if specified)
    406
    0
        pub fn spawn_supervised(
    407
    0
            &mut self,
    408
    0
            name: String,
    409
    0
            behavior: Box<dyn ActorBehavior>,
    410
    0
            supervisor: Option<ActorId>,
    411
    0
        ) -> Result<ActorRef> {
    412
    0
            if self.actor_names.contains_key(&name) {
    413
    0
                return Err(anyhow!("Actor with name '{}' already exists", name));
    414
    0
            }
    415
    416
    0
            let id = ActorId(self.next_id);
    417
    0
            self.next_id += 1;
    418
    419
    0
            let system_arc = Arc::new(Mutex::new(ActorSystem {
    420
    0
                actors: HashMap::new(),
    421
    0
                actor_names: HashMap::new(),
    422
    0
                next_id: self.next_id,
    423
    0
            }));
    424
    425
    0
            let mut runtime = ActorRuntime::new(id, name.clone(), behavior, supervisor, system_arc);
    426
    0
            let actor_ref = runtime.start();
    427
    428
    0
            self.actors.insert(id, runtime);
    429
    0
            self.actor_names.insert(name, id);
    430
    431
    0
            Ok(actor_ref)
    432
    0
        }
    433
    434
        /// Get actor reference by ID
    435
    0
        pub fn get_actor_ref(&self, id: ActorId) -> Option<ActorRef> {
    436
    0
            self.actors.get(&id).map(|runtime| ActorRef {
    437
    0
                id: runtime.id,
    438
    0
                name: runtime.name.clone(),
    439
    0
                sender: runtime.sender.clone(),
    440
    0
            })
    441
    0
        }
    442
    443
        /// Find actor by name
    444
    0
        pub fn find_actor_by_name(&self, name: &str) -> Option<ActorRef> {
    445
    0
            self.actor_names
    446
    0
                .get(name)
    447
    0
                .and_then(|&id| self.get_actor_ref(id))
    448
    0
        }
    449
    450
        /// Stop an actor
    451
        /// # Errors
    452
        ///
    453
        /// Returns an error if the operation fails
    454
    0
        pub fn stop_actor(&mut self, id: ActorId) -> Result<()> {
    455
    0
            if let Some(mut runtime) = self.actors.remove(&id) {
    456
    0
                self.actor_names.retain(|_, &mut v| v != id);
    457
    0
                runtime.stop();
    458
    0
            }
    459
    0
            Ok(())
    460
    0
        }
    461
    462
        /// Shutdown the entire actor system
    463
    0
        pub fn shutdown(&mut self) {
    464
    0
            let actor_ids: Vec<ActorId> = self.actors.keys().copied().collect();
    465
    0
            for id in actor_ids {
    466
    0
                let _ = self.stop_actor(id);
    467
    0
            }
    468
    0
        }
    469
    }
    470
    471
    impl Default for ActorSystem {
    472
    0
        fn default() -> Self {
    473
    0
            Self {
    474
    0
                actors: HashMap::new(),
    475
    0
                actor_names: HashMap::new(),
    476
    0
                next_id: 1,
    477
    0
            }
    478
    0
        }
    479
    }
    480
    481
    impl Clone for ActorSystem {
    482
    0
        fn clone(&self) -> Self {
    483
    0
            Self {
    484
    0
                actors: HashMap::new(),
    485
    0
                actor_names: self.actor_names.clone(),
    486
    0
                next_id: self.next_id,
    487
    0
            }
    488
    0
        }
    489
    }
    490
    491
    /// Example echo actor behavior
    492
    pub struct EchoActor;
    493
    494
    impl ActorBehavior for EchoActor {
    495
    0
        fn receive(&mut self, message: Message, _ctx: &mut ActorContext) -> Result<Option<Message>> {
    496
    0
            match message {
    497
    0
                Message::User(msg_type, values) => {
    498
    0
                    println!("Echo: {msg_type} with values: {values:?}");
    499
    0
                    Ok(Some(Message::User(format!("Echo: {msg_type}"), values)))
    500
                }
    501
    0
                _ => Ok(None),
    502
            }
    503
    0
        }
    504
    }
    505
    506
    /// Example supervisor actor that manages child actors
    507
    pub struct SupervisorActor {
    508
        restart_count: HashMap<ActorId, u32>,
    509
        max_restarts: u32,
    510
    }
    511
    512
    impl SupervisorActor {
    513
        #[must_use]
    514
    0
        pub fn new(max_restarts: u32) -> Self {
    515
    0
            Self {
    516
    0
                restart_count: HashMap::new(),
    517
    0
                max_restarts,
    518
    0
            }
    519
    0
        }
    520
    }
    521
    522
    impl ActorBehavior for SupervisorActor {
    523
    0
        fn receive(&mut self, message: Message, ctx: &mut ActorContext) -> Result<Option<Message>> {
    524
    0
            match message {
    525
    0
                Message::ChildFailed(child_id, reason) => {
    526
    0
                    let count = self.restart_count.entry(child_id).or_insert(0);
    527
    0
                    *count += 1;
    528
    529
    0
                    if *count <= self.max_restarts {
    530
    0
                        println!("Supervisor restarting child {child_id} (attempt {count}): {reason}");
    531
                        // In a real implementation, we would restart the child here
    532
    0
                        Ok(Some(Message::ChildRestarted(child_id)))
    533
                    } else {
    534
    0
                        println!("Supervisor stopping child {child_id} after {count} failures");
    535
    0
                        ctx.stop_child(child_id)?;
    536
    0
                        Ok(None)
    537
                    }
    538
                }
    539
    0
                _ => Ok(None),
    540
            }
    541
    0
        }
    542
    543
    0
        fn supervisor_strategy(&mut self, child: ActorId, _reason: &str) -> SupervisorDirective {
    544
    0
            let count = self.restart_count.get(&child).unwrap_or(&0);
    545
    0
            if *count < self.max_restarts {
    546
    0
                SupervisorDirective::Restart
    547
            } else {
    548
    0
                SupervisorDirective::Stop
    549
            }
    550
    0
        }
    551
    }
    552
    553
    #[cfg(test)]
    554
    #[allow(clippy::unwrap_used)]
    555
    #[allow(clippy::panic)]
    556
    mod tests {
    557
        use super::*;
    558
        use std::time::Duration;
    559
    560
        #[test]
    561
        fn test_actor_system_creation() {
    562
            let system = ActorSystem::new();
    563
            assert!(system.lock().unwrap().actors.is_empty());
    564
        }
    565
    566
        #[test]
    567
        fn test_echo_actor() {
    568
            let system = ActorSystem::new();
    569
            let actor_ref = {
    570
                let mut sys = system.lock().unwrap();
    571
                sys.spawn("echo".to_string(), EchoActor).unwrap()
    572
            };
    573
    574
            let message = Message::User(
    575
                "test".to_string(),
    576
                vec![MessageValue::String("hello".to_string())],
    577
            );
    578
    579
            let response = actor_ref.ask(message, Duration::from_millis(100)).unwrap();
    580
            match response {
    581
                Message::User(msg, _) => assert!(msg.contains("Echo: test")),
    582
                _ => panic!("Unexpected response type"),
    583
            }
    584
        }
    585
    586
        #[test]
    587
        fn test_supervisor_actor() {
    588
            let system = ActorSystem::new();
    589
            let supervisor_ref = {
    590
                let mut sys = system.lock().unwrap();
    591
                sys.spawn("supervisor".to_string(), SupervisorActor::new(3))
    592
                    .unwrap()
    593
            };
    594
    595
            let child_id = ActorId(999);
    596
            let failure_message = Message::ChildFailed(child_id, "Test failure".to_string());
    597
    598
            let response = supervisor_ref
    599
                .ask(failure_message, Duration::from_millis(100))
    600
                .unwrap();
    601
            match response {
    602
                Message::ChildRestarted(id) => assert_eq!(id, child_id),
    603
                _ => panic!("Expected ChildRestarted message"),
    604
            }
    605
        }
    606
    }