## 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/observatory.rs

    Line
    Count
    Source
    1
    //! Actor observatory for live system introspection (RUCHY-0817)
    2
    //!
    3
    //! Provides comprehensive monitoring and debugging capabilities for the actor system,
    4
    //! including message tracing, deadlock detection, and performance analysis.
    5
    6
    #[cfg(test)]
    7
    mod tests {
    8
        use super::*;
    9
        use crate::runtime::actor::{ActorSystem, Message};
    10
        use std::sync::{Arc, Mutex};
    11
        use std::time::Duration;
    12
    13
        // Helper functions for consistent test setup
    14
        fn create_test_actor_system() -> Arc<Mutex<ActorSystem>> {
    15
            ActorSystem::new()
    16
        }
    17
    18
        fn create_test_config() -> ObservatoryConfig {
    19
            ObservatoryConfig {
    20
                max_traces: 100,
    21
                trace_retention_seconds: 3600,
    22
                enable_deadlock_detection: true,
    23
                deadlock_check_interval_ms: 1000,
    24
                enable_metrics: true,
    25
                metrics_interval_ms: 5000,
    26
                max_snapshots: 50,
    27
            }
    28
        }
    29
    30
        fn create_test_observatory() -> ActorObservatory {
    31
            let system = create_test_actor_system();
    32
            let config = create_test_config();
    33
            ActorObservatory::new(system, config)
    34
        }
    35
    36
        fn create_test_message_trace() -> MessageTrace {
    37
            MessageTrace {
    38
                trace_id: 12345,
    39
                timestamp: current_timestamp(),
    40
                source: Some(ActorId(1)),
    41
                destination: ActorId(2),
    42
                message: Message::User("test_message".to_string(), vec![]),
    43
                status: MessageStatus::Queued,
    44
                processing_duration_us: None,
    45
                error: None,
    46
                stack_depth: 1,
    47
                correlation_id: Some("corr-123".to_string()),
    48
            }
    49
        }
    50
    51
        fn create_test_actor_snapshot() -> ActorSnapshot {
    52
            ActorSnapshot {
    53
                actor_id: ActorId(1),
    54
                name: "test_actor".to_string(),
    55
                timestamp: current_timestamp(),
    56
                state: ActorState::Running,
    57
                mailbox_size: 5,
    58
                parent: Some(ActorId(0)),
    59
                children: vec![ActorId(2), ActorId(3)],
    60
                message_stats: MessageStats::default(),
    61
                memory_usage: Some(1024),
    62
            }
    63
        }
    64
    65
        fn create_test_message_filter() -> MessageFilter {
    66
            MessageFilter {
    67
                name: "test_filter".to_string(),
    68
                actor_id: Some(ActorId(1)),
    69
                actor_name_pattern: Some("test_actor".to_string()),
    70
                message_type_pattern: Some(".*message.*".to_string()),
    71
                min_processing_time_us: None,
    72
                failed_only: false,
    73
                max_stack_depth: Some(10),
    74
            }
    75
        }
    76
    77
        // ========== Observatory Configuration Tests ==========
    78
    79
        #[test]
    80
        fn test_observatory_config_default() {
    81
            let config = ObservatoryConfig::default();
    82
            assert_eq!(config.max_traces, 10000);
    83
            assert_eq!(config.trace_retention_seconds, 3600);
    84
            assert!(config.enable_deadlock_detection);
    85
            assert_eq!(config.deadlock_check_interval_ms, 1000);
    86
            assert!(config.enable_metrics);
    87
            assert_eq!(config.metrics_interval_ms, 5000);
    88
            assert_eq!(config.max_snapshots, 1000);
    89
        }
    90
    91
        #[test]
    92
        fn test_observatory_config_clone() {
    93
            let config1 = create_test_config();
    94
            let config2 = config1.clone();
    95
            assert_eq!(config1.max_traces, config2.max_traces);
    96
            assert_eq!(config1.enable_deadlock_detection, config2.enable_deadlock_detection);
    97
        }
    98
    99
        #[test]
    100
        fn test_observatory_config_debug() {
    101
            let config = create_test_config();
    102
            let debug_str = format!("{config:?}");
    103
            assert!(debug_str.contains("ObservatoryConfig"));
    104
            assert!(debug_str.contains("max_traces"));
    105
            assert!(debug_str.contains("enable_deadlock_detection"));
    106
        }
    107
    108
        #[test]
    109
        fn test_observatory_config_serialization() {
    110
            let config = create_test_config();
    111
            let json = serde_json::to_string(&config).unwrap();
    112
            let deserialized: ObservatoryConfig = serde_json::from_str(&json).unwrap();
    113
            assert_eq!(config.max_traces, deserialized.max_traces);
    114
            assert_eq!(config.enable_metrics, deserialized.enable_metrics);
    115
        }
    116
    117
        // ========== Observatory Creation and Setup Tests ==========
    118
    119
        #[test]
    120
        fn test_observatory_creation() {
    121
            let system = create_test_actor_system();
    122
            let config = create_test_config();
    123
            let observatory = ActorObservatory::new(system, config.clone());
    124
    125
            assert_eq!(observatory.config.max_traces, config.max_traces);
    126
            assert!(observatory.filters.is_empty());
    127
            assert!(observatory.start_time.elapsed() < Duration::from_secs(1));
    128
        }
    129
    130
        #[test]
    131
        fn test_observatory_with_default_config() {
    132
            let system = create_test_actor_system();
    133
            let config = ObservatoryConfig::default();
    134
            let observatory = ActorObservatory::new(system, config);
    135
    136
            assert_eq!(observatory.config.max_traces, 10000);
    137
            assert!(observatory.config.enable_deadlock_detection);
    138
        }
    139
    140
        #[test]
    141
        fn test_observatory_initialization_state() {
    142
            let observatory = create_test_observatory();
    143
    144
            // Should start with empty state
    145
            assert!(observatory.get_filters().is_empty());
    146
            let metrics = observatory.metrics.lock().unwrap();
    147
            assert_eq!(metrics.active_actors, 0);
    148
            assert_eq!(metrics.total_messages_processed, 0);
    149
        }
    150
    151
        // ========== Message Filter Management Tests ==========
    152
    153
        #[test]
    154
        fn test_add_message_filter() {
    155
            let mut observatory = create_test_observatory();
    156
            let filter = create_test_message_filter();
    157
    158
            observatory.add_filter(filter.clone());
    159
    160
            assert_eq!(observatory.get_filters().len(), 1);
    161
            assert_eq!(observatory.get_filters()[0].name, filter.name);
    162
        }
    163
    164
        #[test]
    165
        fn test_add_multiple_filters() {
    166
            let mut observatory = create_test_observatory();
    167
    168
            let filter1 = MessageFilter {
    169
                name: "filter1".to_string(),
    170
                actor_id: None,
    171
                actor_name_pattern: None,
    172
                message_type_pattern: Some("type1".to_string()),
    173
                min_processing_time_us: None,
    174
                failed_only: false,
    175
                max_stack_depth: None,
    176
            };
    177
    178
            let filter2 = MessageFilter {
    179
                name: "filter2".to_string(),
    180
                actor_id: Some(ActorId(2)),
    181
                actor_name_pattern: Some("actor2".to_string()),
    182
                message_type_pattern: None,
    183
                min_processing_time_us: Some(1000),
    184
                failed_only: true,
    185
                max_stack_depth: Some(5),
    186
            };
    187
    188
            observatory.add_filter(filter1);
    189
            observatory.add_filter(filter2);
    190
    191
            assert_eq!(observatory.get_filters().len(), 2);
    192
            assert_eq!(observatory.get_filters()[0].name, "filter1");
    193
            assert_eq!(observatory.get_filters()[1].name, "filter2");
    194
        }
    195
    196
        #[test]
    197
        fn test_remove_message_filter() {
    198
            let mut observatory = create_test_observatory();
    199
            let filter = create_test_message_filter();
    200
    201
            observatory.add_filter(filter);
    202
            assert_eq!(observatory.get_filters().len(), 1);
    203
    204
            let removed = observatory.remove_filter("test_filter");
    205
            assert!(removed);
    206
            assert!(observatory.get_filters().is_empty());
    207
        }
    208
    209
        #[test]
    210
        fn test_remove_nonexistent_filter() {
    211
            let mut observatory = create_test_observatory();
    212
            let removed = observatory.remove_filter("nonexistent");
    213
            assert!(!removed);
    214
        }
    215
    216
        #[test]
    217
        fn test_get_filters_empty() {
    218
            let observatory = create_test_observatory();
    219
            assert!(observatory.get_filters().is_empty());
    220
        }
    221
    222
        // ========== Message Tracing Tests ==========
    223
    224
        #[test]
    225
        fn test_trace_message() {
    226
            let observatory = create_test_observatory();
    227
            let trace = create_test_message_trace();
    228
    229
            let result = observatory.trace_message(trace.clone());
    230
            assert!(result.is_ok());
    231
    232
            let traces = observatory.get_traces(None, None).unwrap();
    233
            assert_eq!(traces.len(), 1);
    234
            assert_eq!(traces[0].trace_id, trace.trace_id);
    235
        }
    236
    237
        #[test]
    238
        fn test_trace_message_with_limit() {
    239
            let mut observatory = create_test_observatory();
    240
            observatory.config.max_traces = 2;
    241
    242
            // Add 3 traces, should only keep 2 most recent
    243
            for i in 0..3 {
    244
                let mut trace = create_test_message_trace();
    245
                trace.trace_id = i as u64;
    246
                observatory.trace_message(trace).unwrap();
    247
            }
    248
    249
            let traces = observatory.get_traces(None, None).unwrap();
    250
            assert_eq!(traces.len(), 2);
    251
            assert_eq!(traces[0].trace_id, 1); // First trace should be evicted
    252
            assert_eq!(traces[1].trace_id, 2);
    253
        }
    254
    255
        #[test]
    256
        fn test_trace_message_age_retention() {
    257
            let mut observatory = create_test_observatory();
    258
            observatory.config.trace_retention_seconds = 1; // 1 second retention
    259
    260
            // Add an old trace
    261
            let mut old_trace = create_test_message_trace();
    262
            old_trace.timestamp = current_timestamp() - 3600; // 1 hour ago
    263
            observatory.trace_message(old_trace).unwrap();
    264
    265
            // Add a recent trace
    266
            let recent_trace = create_test_message_trace();
    267
            observatory.trace_message(recent_trace).unwrap();
    268
    269
            let traces = observatory.get_traces(None, None).unwrap();
    270
            // Only recent trace should remain
    271
            assert_eq!(traces.len(), 1);
    272
        }
    273
    274
        #[test]
    275
        fn test_get_traces_with_limit() {
    276
            let observatory = create_test_observatory();
    277
    278
            // Add 5 traces
    279
            for i in 0..5 {
    280
                let mut trace = create_test_message_trace();
    281
                trace.trace_id = i as u64;
    282
                observatory.trace_message(trace).unwrap();
    283
            }
    284
    285
            let traces = observatory.get_traces(Some(3), None).unwrap();
    286
            assert_eq!(traces.len(), 3);
    287
        }
    288
    289
        // ========== Message Status Tests ==========
    290
    291
        #[test]
    292
        fn test_message_status_variants() {
    293
            let statuses = [MessageStatus::Queued,
    294
                MessageStatus::Processing,
    295
                MessageStatus::Completed,
    296
                MessageStatus::Failed,
    297
                MessageStatus::Dropped];
    298
    299
            assert_eq!(statuses.len(), 5);
    300
            assert_eq!(statuses[0], MessageStatus::Queued);
    301
            assert_ne!(statuses[0], MessageStatus::Processing);
    302
        }
    303
    304
        #[test]
    305
        fn test_message_status_serialization() {
    306
            let status = MessageStatus::Processing;
    307
            let json = serde_json::to_string(&status).unwrap();
    308
            let deserialized: MessageStatus = serde_json::from_str(&json).unwrap();
    309
            assert_eq!(status, deserialized);
    310
        }
    311
    312
        #[test]
    313
        fn test_message_status_debug() {
    314
            let status = MessageStatus::Failed;
    315
            let debug_str = format!("{status:?}");
    316
            assert!(debug_str.contains("Failed"));
    317
        }
    318
    319
        // ========== Actor State Tests ==========
    320
    321
        #[test]
    322
        fn test_actor_state_variants() {
    323
            let states = vec![
    324
                ActorState::Starting,
    325
                ActorState::Running,
    326
                ActorState::Processing("test_message".to_string()),
    327
                ActorState::Restarting,
    328
                ActorState::Stopping,
    329
                ActorState::Stopped,
    330
                ActorState::Failed("test_error".to_string()),
    331
            ];
    332
    333
            assert_eq!(states.len(), 7);
    334
            assert_eq!(states[0], ActorState::Starting);
    335
            assert_ne!(states[0], ActorState::Running);
    336
        }
    337
    338
        #[test]
    339
        fn test_actor_state_processing() {
    340
            let state = ActorState::Processing("handle_request".to_string());
    341
            if let ActorState::Processing(message_type) = state {
    342
                assert_eq!(message_type, "handle_request");
    343
            } else {
    344
                panic!("Expected Processing state");
    345
            }
    346
        }
    347
    348
        #[test]
    349
        fn test_actor_state_failed() {
    350
            let state = ActorState::Failed("connection_timeout".to_string());
    351
            if let ActorState::Failed(reason) = state {
    352
                assert_eq!(reason, "connection_timeout");
    353
            } else {
    354
                panic!("Expected Failed state");
    355
            }
    356
        }
    357
    358
        #[test]
    359
        fn test_actor_state_serialization() {
    360
            let state = ActorState::Processing("test".to_string());
    361
            let json = serde_json::to_string(&state).unwrap();
    362
            let deserialized: ActorState = serde_json::from_str(&json).unwrap();
    363
            assert_eq!(state, deserialized);
    364
        }
    365
    366
        // ========== Message Statistics Tests ==========
    367
    368
        #[test]
    369
        fn test_message_stats_default() {
    370
            let stats = MessageStats::default();
    371
            assert_eq!(stats.total_processed, 0);
    372
            assert_eq!(stats.messages_per_second, 0.0);
    373
            assert_eq!(stats.avg_processing_time_us, 0.0);
    374
            assert_eq!(stats.max_processing_time_us, 0);
    375
            assert_eq!(stats.failed_messages, 0);
    376
            assert!(stats.last_processed.is_none());
    377
        }
    378
    379
        #[test]
    380
        fn test_message_stats_clone() {
    381
            let mut stats1 = MessageStats::default();
    382
            stats1.total_processed = 100;
    383
            stats1.messages_per_second = 10.5;
    384
    385
            let stats2 = stats1.clone();
    386
            assert_eq!(stats1.total_processed, stats2.total_processed);
    387
            assert_eq!(stats1.messages_per_second, stats2.messages_per_second);
    388
        }
    389
    390
        #[test]
    391
        fn test_message_stats_serialization() {
    392
            let mut stats = MessageStats::default();
    393
            stats.total_processed = 500;
    394
            stats.avg_processing_time_us = 1500.0;
    395
    396
            let json = serde_json::to_string(&stats).unwrap();
    397
            let deserialized: MessageStats = serde_json::from_str(&json).unwrap();
    398
            assert_eq!(stats.total_processed, deserialized.total_processed);
    399
        }
    400
    401
        // ========== System Metrics Tests ==========
    402
    403
        #[test]
    404
        fn test_system_metrics_default() {
    405
            let metrics = SystemMetrics::default();
    406
            assert_eq!(metrics.active_actors, 0);
    407
            assert_eq!(metrics.total_messages_processed, 0);
    408
            assert_eq!(metrics.system_messages_per_second, 0.0);
    409
            assert_eq!(metrics.total_memory_usage, 0);
    410
            assert_eq!(metrics.total_queued_messages, 0);
    411
            assert_eq!(metrics.avg_mailbox_size, 0.0);
    412
            assert_eq!(metrics.recent_restarts, 0);
    413
            assert!(metrics.last_updated > 0); // Uses current timestamp
    414
        }
    415
    416
        #[test]
    417
        fn test_system_metrics_update() {
    418
            let mut metrics = SystemMetrics::default();
    419
            metrics.active_actors = 10;
    420
            metrics.total_messages_processed = 1000;
    421
            metrics.system_messages_per_second = 50.5;
    422
            metrics.last_updated = current_timestamp();
    423
    424
            assert_eq!(metrics.active_actors, 10);
    425
            assert_eq!(metrics.total_messages_processed, 1000);
    426
            assert!(metrics.last_updated > 0);
    427
        }
    428
    429
        #[test]
    430
        fn test_system_metrics_serialization() {
    431
            let mut metrics = SystemMetrics::default();
    432
            metrics.active_actors = 5;
    433
            metrics.total_memory_usage = 1_024_000;
    434
    435
            let json = serde_json::to_string(&metrics).unwrap();
    436
            let deserialized: SystemMetrics = serde_json::from_str(&json).unwrap();
    437
            assert_eq!(metrics.active_actors, deserialized.active_actors);
    438
            assert_eq!(metrics.total_memory_usage, deserialized.total_memory_usage);
    439
        }
    440
    441
        // ========== Actor Snapshot Tests ==========
    442
    443
        #[test]
    444
        fn test_actor_snapshot_creation() {
    445
            let snapshot = create_test_actor_snapshot();
    446
    447
            assert_eq!(snapshot.actor_id, ActorId(1));
    448
            assert_eq!(snapshot.name, "test_actor");
    449
            assert_eq!(snapshot.state, ActorState::Running);
    450
            assert_eq!(snapshot.mailbox_size, 5);
    451
            assert_eq!(snapshot.children.len(), 2);
    452
            assert!(snapshot.memory_usage.is_some());
    453
        }
    454
    455
        #[test]
    456
        fn test_actor_snapshot_with_no_parent() {
    457
            let mut snapshot = create_test_actor_snapshot();
    458
            snapshot.parent = None;
    459
    460
            assert!(snapshot.parent.is_none());
    461
            assert_eq!(snapshot.children.len(), 2);
    462
        }
    463
    464
        #[test]
    465
        fn test_actor_snapshot_serialization() {
    466
            let snapshot = create_test_actor_snapshot();
    467
    468
            let json = serde_json::to_string(&snapshot).unwrap();
    469
            let deserialized: ActorSnapshot = serde_json::from_str(&json).unwrap();
    470
    471
            assert_eq!(snapshot.actor_id, deserialized.actor_id);
    472
            assert_eq!(snapshot.name, deserialized.name);
    473
            assert_eq!(snapshot.state, deserialized.state);
    474
        }
    475
    476
        // ========== Message Filter Tests ==========
    477
    478
        #[test]
    479
        fn test_message_filter_creation() {
    480
            let filter = create_test_message_filter();
    481
    482
            assert_eq!(filter.name, "test_filter");
    483
            assert!(filter.actor_id.is_some());
    484
            assert!(filter.actor_name_pattern.is_some());
    485
            assert!(filter.message_type_pattern.is_some());
    486
            assert!(!filter.failed_only);
    487
        }
    488
    489
        #[test]
    490
        fn test_message_filter_with_duration_limits() {
    491
            let filter = MessageFilter {
    492
                name: "duration_filter".to_string(),
    493
                actor_id: None,
    494
                actor_name_pattern: None,
    495
                message_type_pattern: None,
    496
                min_processing_time_us: Some(1000),
    497
                failed_only: false,
    498
                max_stack_depth: None,
    499
            };
    500
    501
            assert_eq!(filter.min_processing_time_us, Some(1000));
    502
        }
    503
    504
        #[test]
    505
        fn test_message_filter_errors_only() {
    506
            let filter = MessageFilter {
    507
                name: "error_filter".to_string(),
    508
                actor_id: None,
    509
                actor_name_pattern: None,
    510
                message_type_pattern: None,
    511
                min_processing_time_us: None,
    512
                failed_only: true,
    513
                max_stack_depth: None,
    514
            };
    515
    516
            assert!(filter.failed_only);
    517
        }
    518
    519
        // ========== Message Trace Tests ==========
    520
    521
        #[test]
    522
        fn test_message_trace_creation() {
    523
            let trace = create_test_message_trace();
    524
    525
            assert_eq!(trace.trace_id, 12345);
    526
            assert!(trace.source.is_some());
    527
            assert_eq!(trace.destination, ActorId(2));
    528
            assert_eq!(trace.status, MessageStatus::Queued);
    529
            assert_eq!(trace.stack_depth, 1);
    530
            assert!(trace.correlation_id.is_some());
    531
        }
    532
    533
        #[test]
    534
        fn test_message_trace_with_processing() {
    535
            let mut trace = create_test_message_trace();
    536
            trace.status = MessageStatus::Processing;
    537
            trace.processing_duration_us = Some(1500);
    538
    539
            assert_eq!(trace.status, MessageStatus::Processing);
    540
            assert_eq!(trace.processing_duration_us, Some(1500));
    541
        }
    542
    543
        #[test]
    544
        fn test_message_trace_with_error() {
    545
            let mut trace = create_test_message_trace();
    546
            trace.status = MessageStatus::Failed;
    547
            trace.error = Some("timeout_error".to_string());
    548
    549
            assert_eq!(trace.status, MessageStatus::Failed);
    550
            assert_eq!(trace.error, Some("timeout_error".to_string()));
    551
        }
    552
    553
        #[test]
    554
        fn test_message_trace_serialization() {
    555
            let trace = create_test_message_trace();
    556
    557
            let json = serde_json::to_string(&trace).unwrap();
    558
            let deserialized: MessageTrace = serde_json::from_str(&json).unwrap();
    559
    560
            assert_eq!(trace.trace_id, deserialized.trace_id);
    561
            assert_eq!(trace.status, deserialized.status);
    562
        }
    563
    564
        // ========== Utility Function Tests ==========
    565
    566
        #[test]
    567
        fn test_current_timestamp() {
    568
            let ts1 = current_timestamp();
    569
            std::thread::sleep(Duration::from_millis(10)); // Increase sleep time for more reliable test
    570
            let ts2 = current_timestamp();
    571
    572
            assert!(ts2 >= ts1); // May be equal due to precision
    573
        }
    574
    575
        // ========== Integration Tests ==========
    576
    577
        #[test]
    578
        fn test_observatory_full_workflow() {
    579
            let mut observatory = create_test_observatory();
    580
    581
            // Add filter that matches our test message
    582
            let filter = MessageFilter {
    583
                name: "permissive_filter".to_string(),
    584
                actor_id: None, // Allow all actors
    585
                actor_name_pattern: None,
    586
                message_type_pattern: None, // Allow all message types
    587
                min_processing_time_us: None,
    588
                failed_only: false,
    589
                max_stack_depth: None,
    590
            };
    591
            observatory.add_filter(filter);
    592
            assert_eq!(observatory.get_filters().len(), 1);
    593
    594
            // Add trace
    595
            let trace = create_test_message_trace();
    596
            observatory.trace_message(trace.clone()).unwrap();
    597
    598
            // Verify trace was recorded
    599
            let traces = observatory.get_traces(None, None).unwrap();
    600
            assert_eq!(traces.len(), 1);
    601
            assert_eq!(traces[0].trace_id, trace.trace_id);
    602
    603
            // Remove filter
    604
            let removed = observatory.remove_filter("permissive_filter");
    605
            assert!(removed);
    606
            assert!(observatory.get_filters().is_empty());
    607
        }
    608
    609
        #[test]
    610
        fn test_observatory_multiple_traces() {
    611
            let observatory = create_test_observatory();
    612
    613
            // Add multiple traces with different statuses
    614
            let statuses = [MessageStatus::Queued,
    615
                MessageStatus::Processing,
    616
                MessageStatus::Completed,
    617
                MessageStatus::Failed];
    618
    619
            for (i, status) in statuses.iter().enumerate() {
    620
                let mut trace = create_test_message_trace();
    621
                trace.trace_id = i as u64;
    622
                trace.status = status.clone();
    623
                observatory.trace_message(trace).unwrap();
    624
            }
    625
    626
            let traces = observatory.get_traces(None, None).unwrap();
    627
            assert_eq!(traces.len(), 4);
    628
    629
            // Verify different statuses were recorded - count unique statuses differently
    630
            let mut status_counts = std::collections::HashMap::new();
    631
            for trace in &traces {
    632
                *status_counts.entry(&trace.status).or_insert(0) += 1;
    633
            }
    634
            assert_eq!(status_counts.len(), 4);
    635
        }
    636
    637
        #[test]
    638
        fn test_observatory_config_variations() {
    639
            let configs = vec![
    640
                ObservatoryConfig {
    641
                    max_traces: 50,
    642
                    enable_deadlock_detection: false,
    643
                    enable_metrics: false,
    644
                    ..ObservatoryConfig::default()
    645
                },
    646
                ObservatoryConfig {
    647
                    trace_retention_seconds: 7200,
    648
                    deadlock_check_interval_ms: 500,
    649
                    metrics_interval_ms: 2000,
    650
                    ..ObservatoryConfig::default()
    651
                },
    652
            ];
    653
    654
            for config in configs {
    655
                let system = create_test_actor_system();
    656
                let observatory = ActorObservatory::new(system, config.clone());
    657
                assert_eq!(observatory.config.max_traces, config.max_traces);
    658
                assert_eq!(observatory.config.enable_deadlock_detection, config.enable_deadlock_detection);
    659
            }
    660
        }
    661
    662
        #[test]
    663
        fn test_observatory_concurrent_access() {
    664
            use std::thread;
    665
    666
            let observatory = Arc::new(create_test_observatory());
    667
            let mut handles = vec![];
    668
    669
            // Spawn multiple threads to add traces concurrently
    670
            for i in 0..5 {
    671
                let obs = observatory.clone();
    672
                let handle = thread::spawn(move || {
    673
                    let mut trace = create_test_message_trace();
    674
                    trace.trace_id = i;
    675
                    obs.trace_message(trace).unwrap();
    676
                });
    677
                handles.push(handle);
    678
            }
    679
    680
            // Wait for all threads to complete
    681
            for handle in handles {
    682
                handle.join().unwrap();
    683
            }
    684
    685
            // Verify all traces were recorded
    686
            let traces = observatory.get_traces(None, None).unwrap();
    687
            assert_eq!(traces.len(), 5);
    688
        }
    689
    }
    690
    691
    use crate::runtime::actor::{ActorId, ActorSystem, Message};
    692
    use anyhow::Result;
    693
    use serde::{Deserialize, Serialize};
    694
    use std::collections::{HashMap, HashSet, VecDeque};
    695
    use std::sync::{Arc, Mutex};
    696
    use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
    697
    698
    /// Actor system observatory for live introspection and monitoring
    699
    pub struct ActorObservatory {
    700
        /// Reference to the actor system being observed
    701
        actor_system: Arc<Mutex<ActorSystem>>,
    702
    703
        /// Message trace storage
    704
        message_traces: Arc<Mutex<VecDeque<MessageTrace>>>,
    705
    706
        /// Actor state snapshots
    707
        actor_snapshots: Arc<Mutex<HashMap<ActorId, ActorSnapshot>>>,
    708
    709
        /// Deadlock detection state
    710
        deadlock_detector: Arc<Mutex<DeadlockDetector>>,
    711
    712
        /// Observatory configuration
    713
        config: ObservatoryConfig,
    714
    715
        /// Active filters for message tracing
    716
        filters: Vec<MessageFilter>,
    717
    718
        /// Performance metrics
    719
        metrics: Arc<Mutex<SystemMetrics>>,
    720
    721
        /// Observatory start time
    722
        start_time: Instant,
    723
    }
    724
    725
    /// Configuration for the actor observatory
    726
    #[derive(Debug, Clone, Serialize, Deserialize)]
    727
    pub struct ObservatoryConfig {
    728
        /// Maximum number of message traces to keep
    729
        pub max_traces: usize,
    730
    731
        /// Maximum age for message traces (in seconds)
    732
        pub trace_retention_seconds: u64,
    733
    734
        /// Enable deadlock detection
    735
        pub enable_deadlock_detection: bool,
    736
    737
        /// Deadlock detection interval (in milliseconds)
    738
        pub deadlock_check_interval_ms: u64,
    739
    740
        /// Enable performance metrics collection
    741
        pub enable_metrics: bool,
    742
    743
        /// Metrics collection interval (in milliseconds)
    744
        pub metrics_interval_ms: u64,
    745
    746
        /// Maximum number of actor snapshots to keep
    747
        pub max_snapshots: usize,
    748
    }
    749
    750
    impl Default for ObservatoryConfig {
    751
    0
        fn default() -> Self {
    752
    0
            Self {
    753
    0
                max_traces: 10000,
    754
    0
                trace_retention_seconds: 3600, // 1 hour
    755
    0
                enable_deadlock_detection: true,
    756
    0
                deadlock_check_interval_ms: 1000, // 1 second
    757
    0
                enable_metrics: true,
    758
    0
                metrics_interval_ms: 5000, // 5 seconds
    759
    0
                max_snapshots: 1000,
    760
    0
            }
    761
    0
        }
    762
    }
    763
    764
    /// Message trace entry for debugging and analysis
    765
    #[derive(Debug, Clone, Serialize, Deserialize)]
    766
    pub struct MessageTrace {
    767
        /// Unique trace ID
    768
        pub trace_id: u64,
    769
    770
        /// Timestamp when the message was traced
    771
        pub timestamp: u64,
    772
    773
        /// Source actor ID (None for external messages)
    774
        pub source: Option<ActorId>,
    775
    776
        /// Destination actor ID
    777
        pub destination: ActorId,
    778
    779
        /// The traced message
    780
        pub message: Message,
    781
    782
        /// Message processing status
    783
        pub status: MessageStatus,
    784
    785
        /// Processing duration in microseconds
    786
        pub processing_duration_us: Option<u64>,
    787
    788
        /// Error information if message processing failed
    789
        pub error: Option<String>,
    790
    791
        /// Stack depth for nested message calls
    792
        pub stack_depth: usize,
    793
    794
        /// Correlation ID for tracking message chains
    795
        pub correlation_id: Option<String>,
    796
    }
    797
    798
    /// Status of a traced message
    799
    #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
    800
    pub enum MessageStatus {
    801
        /// Message is queued for processing
    802
        Queued,
    803
        /// Message is currently being processed
    804
        Processing,
    805
        /// Message was processed successfully
    806
        Completed,
    807
        /// Message processing failed
    808
        Failed,
    809
        /// Message was dropped due to actor failure
    810
        Dropped,
    811
    }
    812
    813
    /// Snapshot of an actor's state at a point in time
    814
    #[derive(Debug, Clone, Serialize, Deserialize)]
    815
    pub struct ActorSnapshot {
    816
        /// Actor ID
    817
        pub actor_id: ActorId,
    818
    819
        /// Actor name
    820
        pub name: String,
    821
    822
        /// Snapshot timestamp
    823
        pub timestamp: u64,
    824
    825
        /// Current state of the actor
    826
        pub state: ActorState,
    827
    828
        /// Number of messages in the actor's mailbox
    829
        pub mailbox_size: usize,
    830
    831
        /// Actor's supervision parent (if any)
    832
        pub parent: Option<ActorId>,
    833
    834
        /// Actor's supervised children
    835
        pub children: Vec<ActorId>,
    836
    837
        /// Recent message processing statistics
    838
        pub message_stats: MessageStats,
    839
    840
        /// Memory usage estimate (in bytes)
    841
        pub memory_usage: Option<usize>,
    842
    }
    843
    844
    /// Current state of an actor
    845
    #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
    846
    pub enum ActorState {
    847
        /// Actor is starting up
    848
        Starting,
    849
        /// Actor is running normally
    850
        Running,
    851
        /// Actor is processing a message
    852
        Processing(String), // Message type being processed
    853
        /// Actor is restarting due to failure
    854
        Restarting,
    855
        /// Actor is stopping
    856
        Stopping,
    857
        /// Actor has stopped
    858
        Stopped,
    859
        /// Actor has failed
    860
        Failed(String), // Failure reason
    861
    }
    862
    863
    /// Message processing statistics for an actor
    864
    #[derive(Debug, Clone, Serialize, Deserialize)]
    865
    pub struct MessageStats {
    866
        /// Total messages processed
    867
        pub total_processed: u64,
    868
    869
        /// Messages processed per second (recent average)
    870
        pub messages_per_second: f64,
    871
    872
        /// Average message processing time in microseconds
    873
        pub avg_processing_time_us: f64,
    874
    875
        /// Maximum message processing time in microseconds
    876
        pub max_processing_time_us: u64,
    877
    878
        /// Number of failed message processings
    879
        pub failed_messages: u64,
    880
    881
        /// Last processing timestamp
    882
        pub last_processed: Option<u64>,
    883
    }
    884
    885
    impl Default for MessageStats {
    886
    0
        fn default() -> Self {
    887
    0
            Self {
    888
    0
                total_processed: 0,
    889
    0
                messages_per_second: 0.0,
    890
    0
                avg_processing_time_us: 0.0,
    891
    0
                max_processing_time_us: 0,
    892
    0
                failed_messages: 0,
    893
    0
                last_processed: None,
    894
    0
            }
    895
    0
        }
    896
    }
    897
    898
    /// System-wide performance metrics
    899
    #[derive(Debug, Clone, Serialize, Deserialize)]
    900
    pub struct SystemMetrics {
    901
        /// Total number of active actors
    902
        pub active_actors: usize,
    903
    904
        /// Total messages processed across all actors
    905
        pub total_messages_processed: u64,
    906
    907
        /// System-wide messages per second
    908
        pub system_messages_per_second: f64,
    909
    910
        /// Total memory usage estimate (in bytes)
    911
        pub total_memory_usage: usize,
    912
    913
        /// Number of currently queued messages across all actors
    914
        pub total_queued_messages: usize,
    915
    916
        /// Average actor mailbox size
    917
        pub avg_mailbox_size: f64,
    918
    919
        /// Number of actor restarts in the last period
    920
        pub recent_restarts: u64,
    921
    922
        /// Last metrics update timestamp
    923
        pub last_updated: u64,
    924
    }
    925
    926
    impl Default for SystemMetrics {
    927
    0
        fn default() -> Self {
    928
    0
            Self {
    929
    0
                active_actors: 0,
    930
    0
                total_messages_processed: 0,
    931
    0
                system_messages_per_second: 0.0,
    932
    0
                total_memory_usage: 0,
    933
    0
                total_queued_messages: 0,
    934
    0
                avg_mailbox_size: 0.0,
    935
    0
                recent_restarts: 0,
    936
    0
                last_updated: current_timestamp(),
    937
    0
            }
    938
    0
        }
    939
    }
    940
    941
    /// Filter for message tracing
    942
    #[derive(Debug, Clone, Serialize, Deserialize)]
    943
    pub struct MessageFilter {
    944
        /// Filter name for identification
    945
        pub name: String,
    946
    947
        /// Actor ID to filter by (None for all actors)
    948
        pub actor_id: Option<ActorId>,
    949
    950
        /// Actor name pattern to filter by
    951
        pub actor_name_pattern: Option<String>,
    952
    953
        /// Message type pattern to filter by
    954
        pub message_type_pattern: Option<String>,
    955
    956
        /// Minimum message processing time to include (microseconds)
    957
        pub min_processing_time_us: Option<u64>,
    958
    959
        /// Only include failed messages
    960
        pub failed_only: bool,
    961
    962
        /// Maximum stack depth to include
    963
        pub max_stack_depth: Option<usize>,
    964
    }
    965
    966
    /// Deadlock detection system
    967
    #[derive(Debug)]
    968
    pub struct DeadlockDetector {
    969
        /// Graph of actor message dependencies
    970
        dependency_graph: HashMap<ActorId, HashSet<ActorId>>,
    971
    972
        /// Currently blocked actors waiting for responses
    973
        blocked_actors: HashMap<ActorId, Vec<BlockedRequest>>,
    974
    975
        /// Last deadlock check timestamp
    976
        last_check: Instant,
    977
    978
        /// Detected deadlocks
    979
        detected_deadlocks: Vec<DeadlockCycle>,
    980
    }
    981
    982
    /// Information about a blocked request
    983
    #[derive(Debug, Clone)]
    984
    pub struct BlockedRequest {
    985
        /// Actor making the request
    986
        pub requester: ActorId,
    987
    988
        /// Actor being requested from
    989
        pub target: ActorId,
    990
    991
        /// When the request was made
    992
        pub timestamp: Instant,
    993
    994
        /// Timeout for the request
    995
        pub timeout: Duration,
    996
    997
        /// Message correlation ID
    998
        pub correlation_id: Option<String>,
    999
    }
    1000
    1001
    /// A detected deadlock cycle
    1002
    #[derive(Debug, Clone, Serialize, Deserialize)]
    1003
    pub struct DeadlockCycle {
    1004
        /// Actors involved in the deadlock
    1005
        pub actors: Vec<ActorId>,
    1006
    1007
        /// When the deadlock was detected
    1008
        pub detected_at: u64,
    1009
    1010
        /// Estimated duration of the deadlock
    1011
        pub duration_estimate_ms: u64,
    1012
    1013
        /// Suggested resolution strategy
    1014
        pub resolution_suggestion: String,
    1015
    }
    1016
    1017
    impl ActorObservatory {
    1018
        /// Create a new actor observatory
    1019
    0
        pub fn new(actor_system: Arc<Mutex<ActorSystem>>, config: ObservatoryConfig) -> Self {
    1020
    0
            Self {
    1021
    0
                actor_system,
    1022
    0
                message_traces: Arc::new(Mutex::new(VecDeque::new())),
    1023
    0
                actor_snapshots: Arc::new(Mutex::new(HashMap::new())),
    1024
    0
                deadlock_detector: Arc::new(Mutex::new(DeadlockDetector::new())),
    1025
    0
                config,
    1026
    0
                filters: Vec::new(),
    1027
    0
                metrics: Arc::new(Mutex::new(SystemMetrics::default())),
    1028
    0
                start_time: Instant::now(),
    1029
    0
            }
    1030
    0
        }
    1031
    1032
        /// Add a message filter for tracing
    1033
    0
        pub fn add_filter(&mut self, filter: MessageFilter) {
    1034
    0
            self.filters.push(filter);
    1035
    0
        }
    1036
    1037
        /// Remove a message filter by name
    1038
    0
        pub fn remove_filter(&mut self, name: &str) -> bool {
    1039
    0
            let initial_len = self.filters.len();
    1040
    0
            self.filters.retain(|f| f.name != name);
    1041
    0
            self.filters.len() != initial_len
    1042
    0
        }
    1043
    1044
        /// Get current list of filters
    1045
    0
        pub fn get_filters(&self) -> &[MessageFilter] {
    1046
    0
            &self.filters
    1047
    0
        }
    1048
    1049
        /// Record a message trace
    1050
    0
        pub fn trace_message(&self, trace: MessageTrace) -> Result<()> {
    1051
    0
            let mut traces = self.message_traces
    1052
    0
                .lock()
    1053
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire message traces lock"))?;
    1054
    1055
            // Apply filters
    1056
    0
            if !self.message_matches_filters(&trace) {
    1057
    0
                return Ok(());
    1058
    0
            }
    1059
    1060
    0
            traces.push_back(trace);
    1061
    1062
            // Enforce retention limits
    1063
    0
            while traces.len() > self.config.max_traces {
    1064
    0
                traces.pop_front();
    1065
    0
            }
    1066
    1067
            // Remove old traces based on age
    1068
    0
            let retention_threshold = current_timestamp() - self.config.trace_retention_seconds;
    1069
    0
            while let Some(front) = traces.front() {
    1070
    0
                if front.timestamp < retention_threshold {
    1071
    0
                    traces.pop_front();
    1072
    0
                } else {
    1073
    0
                    break;
    1074
                }
    1075
            }
    1076
    1077
    0
            Ok(())
    1078
    0
        }
    1079
    1080
        /// Get recent message traces with optional filtering
    1081
    0
        pub fn get_traces(&self, limit: Option<usize>, filter_name: Option<&str>) -> Result<Vec<MessageTrace>> {
    1082
    0
            let traces = self.message_traces
    1083
    0
                .lock()
    1084
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire message traces lock"))?;
    1085
    1086
    0
            let mut result: Vec<MessageTrace> = if let Some(filter_name) = filter_name {
    1087
    0
                traces.iter()
    1088
    0
                    .filter(|trace| self.trace_matches_filter(trace, filter_name))
    1089
    0
                    .cloned()
    1090
    0
                    .collect()
    1091
            } else {
    1092
    0
                traces.iter().cloned().collect()
    1093
            };
    1094
    1095
    0
            if let Some(limit) = limit {
    1096
    0
                result.truncate(limit);
    1097
    0
            }
    1098
    1099
    0
            Ok(result)
    1100
    0
        }
    1101
    1102
        /// Update actor snapshot
    1103
    0
        pub fn update_actor_snapshot(&self, snapshot: ActorSnapshot) -> Result<()> {
    1104
    0
            let mut snapshots = self.actor_snapshots
    1105
    0
                .lock()
    1106
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire actor snapshots lock"))?;
    1107
    1108
    0
            snapshots.insert(snapshot.actor_id, snapshot);
    1109
    1110
            // Enforce snapshot limits
    1111
    0
            if snapshots.len() > self.config.max_snapshots {
    1112
                // Remove oldest snapshots
    1113
    0
                let mut oldest_actors: Vec<_> = snapshots.iter()
    1114
    0
                    .map(|(&id, snapshot)| (id, snapshot.timestamp))
    1115
    0
                    .collect();
    1116
    0
                oldest_actors.sort_by_key(|(_, timestamp)| *timestamp);
    1117
    1118
    0
                let to_remove = snapshots.len() - self.config.max_snapshots;
    1119
    0
                for i in 0..to_remove {
    1120
    0
                    if let Some((actor_id, _)) = oldest_actors.get(i) {
    1121
    0
                        snapshots.remove(actor_id);
    1122
    0
                    }
    1123
                }
    1124
    0
            }
    1125
    1126
    0
            Ok(())
    1127
    0
        }
    1128
    1129
        /// Get current actor snapshots
    1130
    0
        pub fn get_actor_snapshots(&self) -> Result<HashMap<ActorId, ActorSnapshot>> {
    1131
    0
            Ok(self.actor_snapshots
    1132
    0
                .lock()
    1133
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire actor snapshots lock"))?
    1134
    0
                .clone())
    1135
    0
        }
    1136
    1137
        /// Get specific actor snapshot
    1138
    0
        pub fn get_actor_snapshot(&self, actor_id: ActorId) -> Result<Option<ActorSnapshot>> {
    1139
    0
            Ok(self.actor_snapshots
    1140
    0
                .lock()
    1141
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire actor snapshots lock"))?
    1142
    0
                .get(&actor_id)
    1143
    0
                .cloned())
    1144
    0
        }
    1145
    1146
        /// Perform deadlock detection
    1147
    0
        pub fn detect_deadlocks(&self) -> Result<Vec<DeadlockCycle>> {
    1148
    0
            if !self.config.enable_deadlock_detection {
    1149
    0
                return Ok(Vec::new());
    1150
    0
            }
    1151
    1152
    0
            let mut detector = self.deadlock_detector
    1153
    0
                .lock()
    1154
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire deadlock detector lock"))?;
    1155
    1156
    0
            detector.detect_cycles()
    1157
    0
        }
    1158
    1159
        /// Update system metrics
    1160
    0
        pub fn update_metrics(&self) -> Result<()> {
    1161
    0
            if !self.config.enable_metrics {
    1162
    0
                return Ok(());
    1163
    0
            }
    1164
    1165
    0
            let _system = self.actor_system
    1166
    0
                .lock()
    1167
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire actor system lock"))?;
    1168
    1169
    0
            let mut metrics = self.metrics
    1170
    0
                .lock()
    1171
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire metrics lock"))?;
    1172
    1173
    0
            let snapshots = self.actor_snapshots
    1174
    0
                .lock()
    1175
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire actor snapshots lock"))?;
    1176
    1177
            // Update metrics based on current system state
    1178
    0
            metrics.active_actors = snapshots.len();
    1179
    0
            metrics.total_messages_processed = snapshots.values()
    1180
    0
                .map(|s| s.message_stats.total_processed)
    1181
    0
                .sum();
    1182
    1183
    0
            metrics.total_queued_messages = snapshots.values()
    1184
    0
                .map(|s| s.mailbox_size)
    1185
    0
                .sum();
    1186
    1187
    0
            metrics.avg_mailbox_size = if snapshots.is_empty() {
    1188
    0
                0.0
    1189
            } else {
    1190
    0
                metrics.total_queued_messages as f64 / snapshots.len() as f64
    1191
            };
    1192
    1193
    0
            metrics.total_memory_usage = snapshots.values()
    1194
    0
                .filter_map(|s| s.memory_usage)
    1195
    0
                .sum();
    1196
    1197
    0
            metrics.last_updated = current_timestamp();
    1198
    1199
    0
            Ok(())
    1200
    0
        }
    1201
    1202
        /// Get current system metrics
    1203
    0
        pub fn get_metrics(&self) -> Result<SystemMetrics> {
    1204
    0
            Ok(self.metrics
    1205
    0
                .lock()
    1206
    0
                .map_err(|_| anyhow::anyhow!("Failed to acquire metrics lock"))?
    1207
    0
                .clone())
    1208
    0
        }
    1209
    1210
        /// Get observatory uptime
    1211
    0
        pub fn uptime(&self) -> Duration {
    1212
    0
            self.start_time.elapsed()
    1213
    0
        }
    1214
    1215
        /// Check if a message matches the configured filters
    1216
    0
        fn message_matches_filters(&self, trace: &MessageTrace) -> bool {
    1217
    0
            if self.filters.is_empty() {
    1218
    0
                return true; // No filters means include all messages
    1219
    0
            }
    1220
    1221
    0
            self.filters.iter().any(|filter| self.message_matches_filter(trace, filter))
    1222
    0
        }
    1223
    1224
        /// Check if a message matches a specific filter
    1225
    0
        fn message_matches_filter(&self, trace: &MessageTrace, filter: &MessageFilter) -> bool {
    1226
            // Filter by actor ID
    1227
    0
            if let Some(filter_actor_id) = filter.actor_id {
    1228
    0
                if trace.destination != filter_actor_id {
    1229
    0
                    return false;
    1230
    0
                }
    1231
    0
            }
    1232
    1233
            // Filter by processing time
    1234
    0
            if let Some(min_time) = filter.min_processing_time_us {
    1235
    0
                if let Some(duration) = trace.processing_duration_us {
    1236
    0
                    if duration < min_time {
    1237
    0
                        return false;
    1238
    0
                    }
    1239
                } else {
    1240
    0
                    return false;
    1241
                }
    1242
    0
            }
    1243
    1244
            // Filter by failed messages only
    1245
    0
            if filter.failed_only && trace.status != MessageStatus::Failed {
    1246
    0
                return false;
    1247
    0
            }
    1248
    1249
            // Filter by stack depth
    1250
    0
            if let Some(max_depth) = filter.max_stack_depth {
    1251
    0
                if trace.stack_depth > max_depth {
    1252
    0
                    return false;
    1253
    0
                }
    1254
    0
            }
    1255
    1256
    0
            true
    1257
    0
        }
    1258
    1259
        /// Check if a trace matches a filter by name
    1260
    0
        fn trace_matches_filter(&self, trace: &MessageTrace, filter_name: &str) -> bool {
    1261
    0
            self.filters.iter()
    1262
    0
                .find(|f| f.name == filter_name)
    1263
    0
                .is_some_and(|filter| self.message_matches_filter(trace, filter))
    1264
    0
        }
    1265
    }
    1266
    1267
    impl Default for DeadlockDetector {
    1268
    0
        fn default() -> Self {
    1269
    0
            Self::new()
    1270
    0
        }
    1271
    }
    1272
    1273
    impl DeadlockDetector {
    1274
        /// Create a new deadlock detector
    1275
    0
        pub fn new() -> Self {
    1276
    0
            Self {
    1277
    0
                dependency_graph: HashMap::new(),
    1278
    0
                blocked_actors: HashMap::new(),
    1279
    0
                last_check: Instant::now(),
    1280
    0
                detected_deadlocks: Vec::new(),
    1281
    0
            }
    1282
    0
        }
    1283
    1284
        /// Add a blocked request to track
    1285
    0
        pub fn add_blocked_request(&mut self, request: BlockedRequest) {
    1286
    0
            self.blocked_actors
    1287
    0
                .entry(request.requester)
    1288
    0
                .or_default()
    1289
    0
                .push(request.clone());
    1290
    1291
            // Update dependency graph
    1292
    0
            self.dependency_graph
    1293
    0
                .entry(request.requester)
    1294
    0
                .or_default()
    1295
    0
                .insert(request.target);
    1296
    0
        }
    1297
    1298
        /// Remove a blocked request (when resolved)
    1299
    0
        pub fn remove_blocked_request(&mut self, requester: ActorId, target: ActorId) {
    1300
    0
            if let Some(requests) = self.blocked_actors.get_mut(&requester) {
    1301
    0
                requests.retain(|r| r.target != target);
    1302
    0
                if requests.is_empty() {
    1303
    0
                    self.blocked_actors.remove(&requester);
    1304
    0
                }
    1305
    0
            }
    1306
    1307
            // Update dependency graph
    1308
    0
            if let Some(dependencies) = self.dependency_graph.get_mut(&requester) {
    1309
    0
                dependencies.remove(&target);
    1310
    0
                if dependencies.is_empty() {
    1311
    0
                    self.dependency_graph.remove(&requester);
    1312
    0
                }
    1313
    0
            }
    1314
    0
        }
    1315
    1316
        /// Detect cycles in the dependency graph (potential deadlocks)
    1317
    0
        pub fn detect_cycles(&mut self) -> Result<Vec<DeadlockCycle>> {
    1318
    0
            let mut cycles = Vec::new();
    1319
    0
            let mut visited = HashSet::new();
    1320
    0
            let mut path = Vec::new();
    1321
    1322
    0
            for &actor in self.dependency_graph.keys() {
    1323
    0
                if !visited.contains(&actor) {
    1324
    0
                    self.dfs_detect_cycle(actor, &mut visited, &mut path, &mut cycles)?;
    1325
    0
                }
    1326
            }
    1327
    1328
    0
            self.detected_deadlocks.extend(cycles.clone());
    1329
    0
            self.last_check = Instant::now();
    1330
    1331
    0
            Ok(cycles)
    1332
    0
        }
    1333
    1334
        /// Depth-first search to detect cycles
    1335
    0
        fn dfs_detect_cycle(
    1336
    0
            &self,
    1337
    0
            actor: ActorId,
    1338
    0
            visited: &mut HashSet<ActorId>,
    1339
    0
            path: &mut Vec<ActorId>,
    1340
    0
            cycles: &mut Vec<DeadlockCycle>,
    1341
    0
        ) -> Result<()> {
    1342
    0
            visited.insert(actor);
    1343
    0
            path.push(actor);
    1344
    1345
    0
            if let Some(dependencies) = self.dependency_graph.get(&actor) {
    1346
    0
                for &dependent_actor in dependencies {
    1347
    0
                    if let Some(cycle_start_index) = path.iter().position(|&a| a == dependent_actor) {
    1348
    0
                        // Found a cycle
    1349
    0
                        let cycle_actors = path[cycle_start_index..].to_vec();
    1350
    0
                        let duration_estimate = self.estimate_cycle_duration(&cycle_actors);
    1351
    0
    1352
    0
                        cycles.push(DeadlockCycle {
    1353
    0
                            actors: cycle_actors.clone(),
    1354
    0
                            detected_at: current_timestamp(),
    1355
    0
                            duration_estimate_ms: duration_estimate,
    1356
    0
                            resolution_suggestion: self.suggest_resolution(&cycle_actors),
    1357
    0
                        });
    1358
    0
                    } else if !visited.contains(&dependent_actor) {
    1359
    0
                        self.dfs_detect_cycle(dependent_actor, visited, path, cycles)?;
    1360
    0
                    }
    1361
                }
    1362
    0
            }
    1363
    1364
    0
            path.pop();
    1365
    0
            Ok(())
    1366
    0
        }
    1367
    1368
        /// Estimate how long a deadlock cycle has been active
    1369
    0
        fn estimate_cycle_duration(&self, actors: &[ActorId]) -> u64 {
    1370
    0
            let now = Instant::now();
    1371
    0
            actors.iter()
    1372
    0
                .filter_map(|&actor| self.blocked_actors.get(&actor))
    1373
    0
                .flatten()
    1374
    0
                .map(|request| now.duration_since(request.timestamp).as_millis() as u64)
    1375
    0
                .max()
    1376
    0
                .unwrap_or(0)
    1377
    0
        }
    1378
    1379
        /// Suggest a resolution strategy for a deadlock cycle
    1380
    0
        fn suggest_resolution(&self, actors: &[ActorId]) -> String {
    1381
    0
            match actors.len() {
    1382
    0
                1 => "Self-deadlock: Check for recursive message sending".to_string(),
    1383
    0
                2 => "Binary deadlock: Consider using ask with timeout or redesign interaction pattern".to_string(),
    1384
    0
                3..=5 => "Multi-actor deadlock: Implement hierarchical message ordering or use supervision".to_string(),
    1385
    0
                _ => "Complex deadlock: Consider breaking into smaller subsystems or using event sourcing".to_string(),
    1386
            }
    1387
    0
        }
    1388
    }
    1389
    1390
    /// Get current timestamp in seconds since Unix epoch
    1391
    0
    fn current_timestamp() -> u64 {
    1392
    0
        SystemTime::now()
    1393
    0
            .duration_since(UNIX_EPOCH)
    1394
    0
            .unwrap_or_default()
    1395
    0
            .as_secs()
    1396
    0
    }
    1397
    1398
    /// Create a simple message filter for testing
    1399
    impl MessageFilter {
    1400
    0
        pub fn new(name: &str) -> Self {
    1401
    0
            Self {
    1402
    0
                name: name.to_string(),
    1403
    0
                actor_id: None,
    1404
    0
                actor_name_pattern: None,
    1405
    0
                message_type_pattern: None,
    1406
    0
                min_processing_time_us: None,
    1407
    0
                failed_only: false,
    1408
    0
                max_stack_depth: None,
    1409
    0
            }
    1410
    0
        }
    1411
    1412
        /// Create a filter for a specific actor
    1413
    0
        pub fn for_actor(name: &str, actor_id: ActorId) -> Self {
    1414
    0
            Self {
    1415
    0
                name: name.to_string(),
    1416
    0
                actor_id: Some(actor_id),
    1417
    0
                actor_name_pattern: None,
    1418
    0
                message_type_pattern: None,
    1419
    0
                min_processing_time_us: None,
    1420
    0
                failed_only: false,
    1421
    0
                max_stack_depth: None,
    1422
    0
            }
    1423
    0
        }
    1424
    1425
        /// Create a filter for failed messages only
    1426
    0
        pub fn failed_messages_only(name: &str) -> Self {
    1427
    0
            Self {
    1428
    0
                name: name.to_string(),
    1429
    0
                actor_id: None,
    1430
    0
                actor_name_pattern: None,
    1431
    0
                message_type_pattern: None,
    1432
    0
                min_processing_time_us: None,
    1433
    0
                failed_only: true,
    1434
    0
                max_stack_depth: None,
    1435
    0
            }
    1436
    0
        }
    1437
    1438
        /// Create a filter for delayed messages
    1439
    0
        pub fn slow_messages(name: &str, min_time_us: u64) -> Self {
    1440
    0
            Self {
    1441
    0
                name: name.to_string(),
    1442
    0
                actor_id: None,
    1443
    0
                actor_name_pattern: None,
    1444
    0
                message_type_pattern: None,
    1445
    0
                min_processing_time_us: Some(min_time_us),
    1446
    0
                failed_only: false,
    1447
    0
                max_stack_depth: None,
    1448
    0
            }
    1449
    0
        }
    1450
    }