## Coverage Report

#### Created: 2026-01-25 15:05

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

    /home/noah/src/realizar/src/gpu/resilience.rs

    Line
    Count
    Source
    1
    //! GPU Resilience Module (PMAT-802)
    2
    //!
    3
    //! Extracted from gpu/mod.rs - Retry, circuit breaker, and bulkhead patterns.
    4
    //!
    5
    //! ## Contents
    6
    //! - `RetryPolicy`, `RetryConfig` - Exponential backoff retry (IMP-076)
    7
    //! - `CircuitBreaker`, `CircuitConfig` - Circuit breaker pattern (IMP-077)
    8
    //! - `BulkheadManager`, `BulkheadConfig` - Resource isolation (IMP-078)
    9
    10
    use std::collections::HashMap;
    11
    use std::time::Duration;
    12
    13
    // ============================================================================
    14
    // M31: Retry Logic & Circuit Breakers (IMP-076, IMP-077, IMP-078)
    15
    // ============================================================================
    16
    17
    /// Error category for retry decisions (IMP-076)
    18
    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
    19
    pub enum ErrorCategory {
    20
        /// Transient error that may succeed on retry
    21
        Transient,
    22
        /// Permanent error that will not succeed on retry
    23
        Permanent,
    24
    }
    25
    26
    /// Retry decision (IMP-076)
    27
    #[derive(Debug, Clone)]
    28
    pub enum RetryDecision {
    29
        /// Retry with specified delay
    30
        Retry {
    31
            /// Delay before retry
    32
            delay: Duration,
    33
        },
    34
        /// Abort with reason
    35
        Abort {
    36
            /// Reason for abort
    37
            reason: String,
    38
        },
    39
    }
    40
    41
    /// Retry configuration (IMP-076)
    42
    #[derive(Debug, Clone)]
    43
    pub struct RetryConfig {
    44
        max_retries: u32,
    45
        base_delay: Duration,
    46
        max_delay: Duration,
    47
        jitter_factor: f64,
    48
    }
    49
    50
    impl RetryConfig {
    51
        /// Create new retry config with defaults
    52
        #[must_use]
    53
    1
        pub fn new() -> Self {
    54
    1
            Self {
    55
    1
                max_retries: 3,
    56
    1
                base_delay: Duration::from_millis(100),
    57
    1
                max_delay: Duration::from_secs(30),
    58
    1
                jitter_factor: 0.1,
    59
    1
            }
    60
    1
        }
    61
    62
        /// Set max retries
    63
        #[must_use]
    64
    1
        pub fn with_max_retries(mut self, max: u32) -> Self {
    65
    1
            self.max_retries = max;
    66
    1
            self
    67
    1
        }
    68
    69
        /// Set base delay
    70
        #[must_use]
    71
    1
        pub fn with_base_delay(mut self, delay: Duration) -> Self {
    72
    1
            self.base_delay = delay;
    73
    1
            self
    74
    1
        }
    75
    76
        /// Set max delay
    77
        #[must_use]
    78
    1
        pub fn with_max_delay(mut self, delay: Duration) -> Self {
    79
    1
            self.max_delay = delay;
    80
    1
            self
    81
    1
        }
    82
    83
        /// Set jitter factor (0.0 to 1.0)
    84
        #[must_use]
    85
    1
        pub fn with_jitter_factor(mut self, factor: f64) -> Self {
    86
    1
            self.jitter_factor = factor.clamp(0.0, 1.0);
    87
    1
            self
    88
    1
        }
    89
    }
    90
    91
    impl Default for RetryConfig {
    92
    0
        fn default() -> Self {
    93
    0
            Self::new()
    94
    0
        }
    95
    }
    96
    97
    /// Retry policy (IMP-076)
    98
    pub struct RetryPolicy {
    99
        config: RetryConfig,
    100
    }
    101
    102
    impl RetryPolicy {
    103
        /// Create new retry policy
    104
        #[must_use]
    105
    1
        pub fn new(config: RetryConfig) -> Self {
    106
    1
            Self { config }
    107
    1
        }
    108
    109
        /// Get max retries
    110
        #[must_use]
    111
    1
        pub fn max_retries(&self) -> u32 {
    112
    1
            self.config.max_retries
    113
    1
        }
    114
    115
        /// Decide whether to retry
    116
        #[must_use]
    117
    3
        pub fn should_retry(&self, attempt: u32, category: ErrorCategory) -> RetryDecision {
    118
    3
            if category == ErrorCategory::Permanent {
    119
    1
                return RetryDecision::Abort {
    120
    1
                    reason: "Permanent error".to_string(),
    121
    1
                };
    122
    2
            }
    123
    124
    2
            if attempt > self.config.max_retries {
    125
    1
                return RetryDecision::Abort {
    126
    1
                    reason: format!("Max retries ({}) exceeded", self.config.max_retries),
    127
    1
                };
    128
    1
            }
    129
    130
    1
            RetryDecision::Retry {
    131
    1
                delay: self.calculate_delay(attempt),
    132
    1
            }
    133
    3
        }
    134
    135
        /// Calculate delay with exponential backoff
    136
        #[must_use]
    137
    5
        pub fn calculate_delay(&self, attempt: u32) -> Duration {
    138
            // Exponential backoff: base * 2^attempt
    139
    5
            let exp_delay_ms = self.config.base_delay.as_millis() as u64 * (1u64 << attempt.min(20));
    140
    5
            let delay_ms = exp_delay_ms.min(self.config.max_delay.as_millis() as u64);
    141
    5
            Duration::from_millis(delay_ms)
    142
    5
        }
    143
    }
    144
    145
    /// Circuit breaker state (IMP-077)
    146
    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
    147
    pub enum CircuitState {
    148
        /// Circuit is closed, requests allowed
    149
        Closed,
    150
        /// Circuit is open, requests rejected
    151
        Open,
    152
        /// Circuit is half-open, probe requests allowed
    153
        HalfOpen,
    154
    }
    155
    156
    /// Circuit breaker configuration (IMP-077)
    157
    #[derive(Debug, Clone)]
    158
    pub struct CircuitConfig {
    159
        failure_threshold: u32,
    160
        success_threshold: u32,
    161
        timeout: Duration,
    162
    }
    163
    164
    impl CircuitConfig {
    165
        /// Create new circuit config with defaults
    166
        #[must_use]
    167
    1
        pub fn new() -> Self {
    168
    1
            Self {
    169
    1
                failure_threshold: 5,
    170
    1
                success_threshold: 2,
    171
    1
                timeout: Duration::from_secs(30),
    172
    1
            }
    173
    1
        }
    174
    175
        /// Set failure threshold to open circuit
    176
        #[must_use]
    177
    1
        pub fn with_failure_threshold(mut self, threshold: u32) -> Self {
    178
    1
            self.failure_threshold = threshold;
    179
    1
            self
    180
    1
        }
    181
    182
        /// Set success threshold to close circuit from half-open
    183
        #[must_use]
    184
    1
        pub fn with_success_threshold(mut self, threshold: u32) -> Self {
    185
    1
            self.success_threshold = threshold;
    186
    1
            self
    187
    1
        }
    188
    189
        /// Set timeout before transitioning to half-open
    190
        #[must_use]
    191
    1
        pub fn with_timeout(mut self, timeout: Duration) -> Self {
    192
    1
            self.timeout = timeout;
    193
    1
            self
    194
    1
        }
    195
    }
    196
    197
    impl Default for CircuitConfig {
    198
    0
        fn default() -> Self {
    199
    0
            Self::new()
    200
    0
        }
    201
    }
    202
    203
    /// Circuit breaker (IMP-077)
    204
    pub struct CircuitBreaker {
    205
        config: CircuitConfig,
    206
        state: std::sync::Mutex<CircuitState>,
    207
        failure_count: std::sync::atomic::AtomicU32,
    208
        success_count: std::sync::atomic::AtomicU32,
    209
        last_failure: std::sync::Mutex<Option<std::time::Instant>>,
    210
    }
    211
    212
    impl CircuitBreaker {
    213
        /// Create new circuit breaker
    214
        #[must_use]
    215
    1
        pub fn new(config: CircuitConfig) -> Self {
    216
    1
            Self {
    217
    1
                config,
    218
    1
                state: std::sync::Mutex::new(CircuitState::Closed),
    219
    1
                failure_count: std::sync::atomic::AtomicU32::new(0),
    220
    1
                success_count: std::sync::atomic::AtomicU32::new(0),
    221
    1
                last_failure: std::sync::Mutex::new(None),
    222
    1
            }
    223
    1
        }
    224
    225
        /// Get current state
    226
        #[must_use]
    227
    6
        pub fn state(&self) -> CircuitState {
    228
    6
            *self.state.lock().expect("mutex poisoned")
    229
    6
        }
    230
    231
        /// Check if request should be allowed
    232
        #[must_use]
    233
    3
        pub fn allow_request(&self) -> bool {
    234
    3
            let mut state = self.state.lock().expect("mutex poisoned");
    235
    3
            match *state {
    236
    0
                CircuitState::Closed | CircuitState::HalfOpen => true,
    237
                CircuitState::Open => {
    238
                    // Check if timeout has elapsed
    239
    3
                    let last_failure = self.last_failure.lock().expect("mutex poisoned");
    240
    3
                    if let Some(last) = *last_failure {
    241
    3
                        if last.elapsed() >= self.config.timeout {
    242
    2
                            *state = CircuitState::HalfOpen;
    243
    2
                            self.success_count
    244
    2
                                .store(0, std::sync::atomic::Ordering::SeqCst);
    245
    2
                            return true;
    246
    1
                        }
    247
    0
                    }
    248
    1
                    false
    249
                },
    250
            }
    251
    3
        }
    252
    253
        /// Record a failure
    254
    7
        pub fn record_failure(&self) {
    255
    7
            let count = self
    256
    7
                .failure_count
    257
    7
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
    258
    7
                + 1;
    259
    7
            *self.last_failure.lock().expect("mutex poisoned") = Some(std::time::Instant::now());
    260
    261
    7
            let mut state = self.state.lock().expect("mutex poisoned");
    262
    7
            match *state {
    263
                CircuitState::Closed => {
    264
    6
                    if count >= self.config.failure_threshold {
    265
    2
                        *state = CircuitState::Open;
    266
    4
                    }
    267
                },
    268
    1
                CircuitState::HalfOpen => {
    269
    1
                    *state = CircuitState::Open;
    270
    1
                },
    271
    0
                CircuitState::Open => {},
    272
            }
    273
    7
        }
    274
    275
        /// Record a success
    276
    2
        pub fn record_success(&self) {
    277
    2
            self.failure_count
    278
    2
                .store(0, std::sync::atomic::Ordering::SeqCst);
    279
    2
            let count = self
    280
    2
                .success_count
    281
    2
                .fetch_add(1, std::sync::atomic::Ordering::SeqCst)
    282
    2
                + 1;
    283
    284
    2
            let mut state = self.state.lock().expect("mutex poisoned");
    285
    2
            if *state == CircuitState::HalfOpen && count >= self.config.success_threshold {
    286
    1
                *state = CircuitState::Closed;
    287
    1
            }
    288
    2
        }
    289
    }
    290
    291
    /// Request type for bulkhead (IMP-078)
    292
    #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
    293
    pub enum RequestType {
    294
        /// Standard inference request
    295
        Inference,
    296
        /// Embedding generation request
    297
        Embedding,
    298
        /// Batch processing request
    299
        Batch,
    300
    }
    301
    302
    /// Bulkhead permit
    303
    #[derive(Debug)]
    304
    pub struct BulkheadPermit {
    305
        request_type: RequestType,
    306
    }
    307
    308
    /// Bulkhead configuration (IMP-078)
    309
    pub struct BulkheadConfig {
    310
        pools: HashMap<String, usize>,
    311
    }
    312
    313
    impl BulkheadConfig {
    314
        /// Create new bulkhead config
    315
        #[must_use]
    316
    1
        pub fn new() -> Self {
    317
    1
            Self {
    318
    1
                pools: HashMap::new(),
    319
    1
            }
    320
    1
        }
    321
    322
        /// Add a pool with specified size
    323
        #[must_use]
    324
    3
        pub fn with_pool(mut self, name: &str, size: usize) -> Self {
    325
    3
            self.pools.insert(name.to_string(), size);
    326
    3
            self
    327
    3
        }
    328
    }
    329
    330
    impl Default for BulkheadConfig {
    331
    0
        fn default() -> Self {
    332
    0
            Self::new()
    333
    0
        }
    334
    }
    335
    336
    /// Bulkhead stats
    337
    #[derive(Debug, Clone)]
    338
    pub struct BulkheadStats {
    339
        /// Number of pools
    340
        pub pool_count: usize,
    341
        /// Total capacity across all pools
    342
        pub total_capacity: usize,
    343
    }
    344
    345
    /// Bulkhead manager (IMP-078)
    346
    pub struct BulkheadManager {
    347
        inference_available: std::sync::atomic::AtomicUsize,
    348
        inference_capacity: usize,
    349
        embedding_available: std::sync::atomic::AtomicUsize,
    350
        embedding_capacity: usize,
    351
        batch_available: std::sync::atomic::AtomicUsize,
    352
        batch_capacity: usize,
    353
    }
    354
    355
    impl BulkheadManager {
    356
        /// Create new bulkhead manager
    357
        #[must_use]
    358
    1
        pub fn new(config: &BulkheadConfig) -> Self {
    359
    1
            let inference_cap = *config.pools.get("inference").unwrap_or(&10);
    360
    1
            let embedding_cap = *config.pools.get("embedding").unwrap_or(&5);
    361
    1
            let batch_cap = *config.pools.get("batch").unwrap_or(&2);
    362
    363
    1
            Self {
    364
    1
                inference_available: std::sync::atomic::AtomicUsize::new(inference_cap),
    365
    1
                inference_capacity: inference_cap,
    366
    1
                embedding_available: std::sync::atomic::AtomicUsize::new(embedding_cap),
    367
    1
                embedding_capacity: embedding_cap,
    368
    1
                batch_available: std::sync::atomic::AtomicUsize::new(batch_cap),
    369
    1
                batch_capacity: batch_cap,
    370
    1
            }
    371
    1
        }
    372
    373
        /// Get available slots for request type
    374
        #[must_use]
    375
    5
        pub fn available(&self, request_type: RequestType) -> usize {
    376
    5
            match request_type {
    377
    4
                RequestType::Inference => self
    378
    4
                    .inference_available
    379
    4
                    .load(std::sync::atomic::Ordering::SeqCst),
    380
    1
                RequestType::Embedding => self
    381
    1
                    .embedding_available
    382
    1
                    .load(std::sync::atomic::Ordering::SeqCst),
    383
    0
                RequestType::Batch => self
    384
    0
                    .batch_available
    385
    0
                    .load(std::sync::atomic::Ordering::SeqCst),
    386
            }
    387
    5
        }
    388
    389
        /// Acquire a permit
    390
        ///
    391
        /// # Errors
    392
        /// Returns error if pool is exhausted.
    393
    5
        pub fn acquire(
    394
    5
            &self,
    395
    5
            request_type: RequestType,
    396
    5
        ) -> std::result::Result<BulkheadPermit, &'static str> {
    397
    5
            let available = match request_type {
    398
    1
                RequestType::Inference => &self.inference_available,
    399
    1
                RequestType::Embedding => &self.embedding_available,
    400
    3
                RequestType::Batch => &self.batch_available,
    401
            };
    402
    403
    5
            let current = available.load(std::sync::atomic::Ordering::SeqCst);
    404
    5
            if current == 0 {
    405
    1
                return Err("Pool exhausted");
    406
    4
            }
    407
    4
            available.fetch_sub(1, std::sync::atomic::Ordering::SeqCst);
    408
    4
            Ok(BulkheadPermit { request_type })
    409
    5
        }
    410
    411
        /// Try to acquire a permit (non-blocking)
    412
        ///
    413
        /// # Errors
    414
        /// Returns error if pool is exhausted.
    415
    1
        pub fn try_acquire(
    416
    1
            &self,
    417
    1
            request_type: RequestType,
    418
    1
        ) -> std::result::Result<BulkheadPermit, &'static str> {
    419
    1
            self.acquire(request_type)
    420
    1
        }
    421
    422
        /// Release a permit
    423
    1
        pub fn release(&self, permit: &BulkheadPermit) {
    424
    1
            let available = match permit.request_type {
    425
    1
                RequestType::Inference => &self.inference_available,
    426
    0
                RequestType::Embedding => &self.embedding_available,
    427
    0
                RequestType::Batch => &self.batch_available,
    428
            };
    429
    1
            available.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
    430
    1
        }
    431
    432
        /// Get bulkhead stats
    433
        #[must_use]
    434
    1
        pub fn stats(&self) -> BulkheadStats {
    435
    1
            BulkheadStats {
    436
    1
                pool_count: 3,
    437
    1
                total_capacity: self.inference_capacity + self.embedding_capacity + self.batch_capacity,
    438
    1
            }
    439
    1
        }
    440
    }