## Coverage Report

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

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

    /home/noah/src/trueno/src/brick/patterns.rs

    Line
    Count
    Source
    1
    //! Async and Buffer Pattern Catalog
    2
    //!
    3
    //! Utility patterns from Phase 12 (E.10) for async operations,
    4
    //! buffer management, and flow control.
    5
    //!
    6
    //! # Patterns Included
    7
    //!
    8
    //! - **LCP-12**: AsyncResult - async compute with sync fallback
    9
    //! - **AWP-11**: BoundedQueue - bounded message queue with back-pressure
    10
    //! - **AWP-13**: ReserveStrategy - buffer reservation strategies
    11
    //! - **LCP-08**: GraphReuseCounter - graph reuse tracking
    12
    //! - **AWP-03**: DualWakerState - dual-waker backpressure
    13
    //! - **AWP-04**: StreamCapacity - HTTP/2 flow control
    14
    //! - **AWP-09**: WakeSkipState - smart wake skip optimization
    15
    16
    use std::collections::VecDeque;
    17
    18
    // ============================================================================
    19
    // LCP-12: Async Compute with Sync Fallback
    20
    // ============================================================================
    21
    22
    /// Result of an async operation with fallback capability.
    23
    #[derive(Debug, Clone)]
    24
    pub enum AsyncResult<T, E> {
    25
        /// Operation completed asynchronously
    26
        Async(T),
    27
        /// Operation completed synchronously (fallback)
    28
        Sync(T),
    29
        /// Operation failed
    30
        Error(E),
    31
    }
    32
    33
    impl<T, E> AsyncResult<T, E> {
    34
        /// Check if result was obtained asynchronously.
    35
        #[must_use]
    36
    0
        pub fn is_async(&self) -> bool {
    37
    0
            matches!(self, AsyncResult::Async(_))
    38
    0
        }
    39
    40
        /// Check if result was obtained synchronously (fallback).
    41
        #[must_use]
    42
    0
        pub fn is_sync(&self) -> bool {
    43
    0
            matches!(self, AsyncResult::Sync(_))
    44
    0
        }
    45
    46
        /// Check if operation failed.
    47
        #[must_use]
    48
    0
        pub fn is_error(&self) -> bool {
    49
    0
            matches!(self, AsyncResult::Error(_))
    50
    0
        }
    51
    52
        /// Get the result value, regardless of async/sync.
    53
    0
        pub fn into_result(self) -> Result<T, E> {
    54
    0
            match self {
    55
    0
                AsyncResult::Async(v) | AsyncResult::Sync(v) => Ok(v),
    56
    0
                AsyncResult::Error(e) => Err(e),
    57
            }
    58
    0
        }
    59
    60
        /// Map the success value.
    61
    0
        pub fn map<U>(self, f: impl FnOnce(T) -> U) -> AsyncResult<U, E> {
    62
    0
            match self {
    63
    0
                AsyncResult::Async(v) => AsyncResult::Async(f(v)),
    64
    0
                AsyncResult::Sync(v) => AsyncResult::Sync(f(v)),
    65
    0
                AsyncResult::Error(e) => AsyncResult::Error(e),
    66
            }
    67
    0
        }
    68
    }
    69
    70
    // ============================================================================
    71
    // AWP-11: Bounded Message Queue
    72
    // ============================================================================
    73
    74
    /// Bounded message queue with back-pressure.
    75
    ///
    76
    /// # Example
    77
    /// ```rust
    78
    /// use trueno::brick::BoundedQueue;
    79
    ///
    80
    /// let mut queue: BoundedQueue<i32> = BoundedQueue::new(3);
    81
    ///
    82
    /// assert!(queue.try_push(1).is_ok());
    83
    /// assert!(queue.try_push(2).is_ok());
    84
    /// assert!(queue.try_push(3).is_ok());
    85
    /// assert!(queue.try_push(4).is_err()); // Queue full
    86
    ///
    87
    /// assert_eq!(queue.pop(), Some(1));
    88
    /// assert!(queue.try_push(4).is_ok()); // Space available
    89
    /// ```
    90
    #[derive(Debug)]
    91
    pub struct BoundedQueue<T> {
    92
        items: VecDeque<T>,
    93
        capacity: usize,
    94
    }
    95
    96
    impl<T> BoundedQueue<T> {
    97
        /// Create a new bounded queue.
    98
    0
        pub fn new(capacity: usize) -> Self {
    99
    0
            Self {
    100
    0
                items: VecDeque::with_capacity(capacity),
    101
    0
                capacity,
    102
    0
            }
    103
    0
        }
    104
    105
        /// Try to push an item. Returns error if queue is full.
    106
    0
        pub fn try_push(&mut self, item: T) -> Result<(), T> {
    107
    0
            if self.items.len() >= self.capacity {
    108
    0
                Err(item)
    109
            } else {
    110
    0
                self.items.push_back(item);
    111
    0
                Ok(())
    112
            }
    113
    0
        }
    114
    115
        /// Pop an item from the front.
    116
    0
        pub fn pop(&mut self) -> Option<T> {
    117
    0
            self.items.pop_front()
    118
    0
        }
    119
    120
        /// Peek at the front item.
    121
        #[must_use]
    122
    0
        pub fn peek(&self) -> Option<&T> {
    123
    0
            self.items.front()
    124
    0
        }
    125
    126
        /// Get the number of items in the queue.
    127
        #[must_use]
    128
    0
        pub fn len(&self) -> usize {
    129
    0
            self.items.len()
    130
    0
        }
    131
    132
        /// Check if the queue is empty.
    133
        #[must_use]
    134
    0
        pub fn is_empty(&self) -> bool {
    135
    0
            self.items.is_empty()
    136
    0
        }
    137
    138
        /// Check if the queue is full.
    139
        #[must_use]
    140
    0
        pub fn is_full(&self) -> bool {
    141
    0
            self.items.len() >= self.capacity
    142
    0
        }
    143
    144
        /// Get the capacity.
    145
        #[must_use]
    146
    0
        pub fn capacity(&self) -> usize {
    147
    0
            self.capacity
    148
    0
        }
    149
    150
        /// Get remaining capacity.
    151
        #[must_use]
    152
    0
        pub fn remaining(&self) -> usize {
    153
    0
            self.capacity.saturating_sub(self.items.len())
    154
    0
        }
    155
    156
        /// Clear all items.
    157
    0
        pub fn clear(&mut self) {
    158
    0
            self.items.clear();
    159
    0
        }
    160
    }
    161
    162
    impl<T> Default for BoundedQueue<T> {
    163
    0
        fn default() -> Self {
    164
    0
            Self::new(16)
    165
    0
        }
    166
    }
    167
    168
    // ============================================================================
    169
    // AWP-13: Buffer Reserve Strategy
    170
    // ============================================================================
    171
    172
    /// Strategy for buffer reservation.
    173
    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
    174
    pub enum ReserveStrategy {
    175
        /// Reserve exact amount needed
    176
        Exact,
    177
        /// Reserve with 50% growth headroom
    178
        Grow50,
    179
        /// Reserve with 100% growth headroom (double)
    180
        Double,
    181
        /// Reserve to next power of two
    182
        PowerOfTwo,
    183
    }
    184
    185
    /// Reserve buffer capacity according to strategy.
    186
    ///
    187
    /// # Example
    188
    /// ```rust
    189
    /// use trueno::brick::{reserve_capacity, ReserveStrategy};
    190
    ///
    191
    /// assert_eq!(reserve_capacity(100, ReserveStrategy::Exact), 100);
    192
    /// assert_eq!(reserve_capacity(100, ReserveStrategy::Grow50), 150);
    193
    /// assert_eq!(reserve_capacity(100, ReserveStrategy::Double), 200);
    194
    /// assert_eq!(reserve_capacity(100, ReserveStrategy::PowerOfTwo), 128);
    195
    /// ```
    196
    #[must_use]
    197
    0
    pub fn reserve_capacity(needed: usize, strategy: ReserveStrategy) -> usize {
    198
    0
        match strategy {
    199
    0
            ReserveStrategy::Exact => needed,
    200
    0
            ReserveStrategy::Grow50 => needed + needed / 2,
    201
    0
            ReserveStrategy::Double => needed * 2,
    202
    0
            ReserveStrategy::PowerOfTwo => needed.next_power_of_two(),
    203
        }
    204
    0
    }
    205
    206
    /// Buffer with configurable reserve strategy.
    207
    #[derive(Debug)]
    208
    pub struct StrategicBuffer {
    209
        data: Vec<u8>,
    210
        strategy: ReserveStrategy,
    211
    }
    212
    213
    impl StrategicBuffer {
    214
        /// Create a new buffer with the given strategy.
    215
    0
        pub fn new(strategy: ReserveStrategy) -> Self {
    216
    0
            Self {
    217
    0
                data: Vec::new(),
    218
    0
                strategy,
    219
    0
            }
    220
    0
        }
    221
    222
        /// Create with initial capacity.
    223
    0
        pub fn with_capacity(capacity: usize, strategy: ReserveStrategy) -> Self {
    224
    0
            Self {
    225
    0
                data: Vec::with_capacity(reserve_capacity(capacity, strategy)),
    226
    0
                strategy,
    227
    0
            }
    228
    0
        }
    229
    230
        /// Ensure capacity for additional bytes.
    231
    0
        pub fn reserve(&mut self, additional: usize) {
    232
    0
            let needed = self.data.len() + additional;
    233
    0
            if needed > self.data.capacity() {
    234
    0
                let new_cap = reserve_capacity(needed, self.strategy);
    235
    0
                self.data.reserve(new_cap - self.data.capacity());
    236
    0
            }
    237
    0
        }
    238
    239
        /// Write bytes to the buffer.
    240
    0
        pub fn write(&mut self, bytes: &[u8]) {
    241
    0
            self.reserve(bytes.len());
    242
    0
            self.data.extend_from_slice(bytes);
    243
    0
        }
    244
    245
        /// Get the data.
    246
        #[must_use]
    247
    0
        pub fn as_slice(&self) -> &[u8] {
    248
    0
            &self.data
    249
    0
        }
    250
    251
        /// Get current length.
    252
        #[must_use]
    253
    0
        pub fn len(&self) -> usize {
    254
    0
            self.data.len()
    255
    0
        }
    256
    257
        /// Check if empty.
    258
        #[must_use]
    259
    0
        pub fn is_empty(&self) -> bool {
    260
    0
            self.data.is_empty()
    261
    0
        }
    262
    263
        /// Get capacity.
    264
        #[must_use]
    265
    0
        pub fn capacity(&self) -> usize {
    266
    0
            self.data.capacity()
    267
    0
        }
    268
    269
        /// Clear the buffer.
    270
    0
        pub fn clear(&mut self) {
    271
    0
            self.data.clear();
    272
    0
        }
    273
    }
    274
    275
    impl Default for StrategicBuffer {
    276
    0
        fn default() -> Self {
    277
    0
            Self::new(ReserveStrategy::Double)
    278
    0
        }
    279
    }
    280
    281
    // ============================================================================
    282
    // LCP-08: Graph Reuse Counter
    283
    // ============================================================================
    284
    285
    /// Counter for tracking graph reuse in inference optimization.
    286
    ///
    287
    /// Tracks how many times a computation graph has been reused,
    288
    /// enabling optimization decisions like caching or recompilation.
    289
    #[derive(Debug, Clone, Default)]
    290
    pub struct GraphReuseCounter {
    291
        /// Number of times this graph has been executed
    292
        reuse_count: u64,
    293
        /// Threshold for considering graph "hot"
    294
        hot_threshold: u64,
    295
        /// Whether to enable caching
    296
        cache_enabled: bool,
    297
    }
    298
    299
    impl GraphReuseCounter {
    300
        /// Create a new counter with hot threshold.
    301
    0
        pub fn new(hot_threshold: u64) -> Self {
    302
    0
            Self {
    303
    0
                reuse_count: 0,
    304
    0
                hot_threshold,
    305
    0
                cache_enabled: false,
    306
    0
            }
    307
    0
        }
    308
    309
        /// Record a graph execution.
    310
    0
        pub fn record_use(&mut self) {
    311
    0
            self.reuse_count += 1;
    312
    0
            if self.reuse_count >= self.hot_threshold {
    313
    0
                self.cache_enabled = true;
    314
    0
            }
    315
    0
        }
    316
    317
        /// Check if graph is considered "hot" (heavily reused).
    318
        #[must_use]
    319
    0
        pub fn is_hot(&self) -> bool {
    320
    0
            self.reuse_count >= self.hot_threshold
    321
    0
        }
    322
    323
        /// Check if caching should be enabled.
    324
        #[must_use]
    325
    0
        pub fn should_cache(&self) -> bool {
    326
    0
            self.cache_enabled
    327
    0
        }
    328
    329
        /// Get the current reuse count.
    330
        #[must_use]
    331
    0
        pub fn count(&self) -> u64 {
    332
    0
            self.reuse_count
    333
    0
        }
    334
    335
        /// Reset the counter.
    336
    0
        pub fn reset(&mut self) {
    337
    0
            self.reuse_count = 0;
    338
    0
            self.cache_enabled = false;
    339
    0
        }
    340
    }
    341
    342
    // ============================================================================
    343
    // AWP-03: Dual-Waker Payload Backpressure
    344
    // ============================================================================
    345
    346
    /// Dual-waker state for async backpressure.
    347
    ///
    348
    /// Tracks two wakers: one for the producer, one for the consumer.
    349
    /// Enables efficient producer/consumer coordination.
    350
    #[derive(Debug, Default)]
    351
    pub struct DualWakerState {
    352
        /// Producer is waiting
    353
        producer_waiting: bool,
    354
        /// Consumer is waiting
    355
        consumer_waiting: bool,
    356
        /// Buffer fill level (0-100%)
    357
        fill_percent: u8,
    358
        /// High watermark for backpressure (%)
    359
        high_watermark: u8,
    360
        /// Low watermark for resume (%)
    361
        low_watermark: u8,
    362
    }
    363
    364
    impl DualWakerState {
    365
        /// Create new state with watermarks.
    366
    0
        pub fn new(low_watermark: u8, high_watermark: u8) -> Self {
    367
    0
            Self {
    368
    0
                producer_waiting: false,
    369
    0
                consumer_waiting: false,
    370
    0
                fill_percent: 0,
    371
    0
                high_watermark: high_watermark.min(100),
    372
    0
                low_watermark: low_watermark.min(high_watermark),
    373
    0
            }
    374
    0
        }
    375
    376
        /// Update fill level and determine who should wake.
    377
    0
        pub fn update_fill(&mut self, fill_percent: u8) -> WakeDecision {
    378
    0
            let old_fill = self.fill_percent;
    379
    0
            self.fill_percent = fill_percent.min(100);
    380
    381
            // Crossed high watermark going up - pause producer
    382
    0
            if old_fill < self.high_watermark && self.fill_percent >= self.high_watermark {
    383
    0
                return WakeDecision::PauseProducer;
    384
    0
            }
    385
    386
            // Crossed low watermark going down - resume producer
    387
    0
            if old_fill > self.low_watermark && self.fill_percent <= self.low_watermark {
    388
    0
                return WakeDecision::WakeProducer;
    389
    0
            }
    390
    391
            // Data available - wake consumer if waiting
    392
    0
            if self.fill_percent > 0 && self.consumer_waiting {
    393
    0
                return WakeDecision::WakeConsumer;
    394
    0
            }
    395
    396
    0
            WakeDecision::None
    397
    0
        }
    398
    399
        /// Producer is now waiting.
    400
    0
        pub fn producer_wait(&mut self) {
    401
    0
            self.producer_waiting = true;
    402
    0
        }
    403
    404
        /// Consumer is now waiting.
    405
    0
        pub fn consumer_wait(&mut self) {
    406
    0
            self.consumer_waiting = true;
    407
    0
        }
    408
    409
        /// Producer was woken.
    410
    0
        pub fn producer_woke(&mut self) {
    411
    0
            self.producer_waiting = false;
    412
    0
        }
    413
    414
        /// Consumer was woken.
    415
    0
        pub fn consumer_woke(&mut self) {
    416
    0
            self.consumer_waiting = false;
    417
    0
        }
    418
    419
        /// Check if producer should be allowed to produce.
    420
        #[must_use]
    421
    0
        pub fn can_produce(&self) -> bool {
    422
    0
            self.fill_percent < self.high_watermark
    423
    0
        }
    424
    425
        /// Check if consumer has data to consume.
    426
        #[must_use]
    427
    0
        pub fn can_consume(&self) -> bool {
    428
    0
            self.fill_percent > 0
    429
    0
        }
    430
    }
    431
    432
    /// Decision on which waker to invoke.
    433
    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
    434
    pub enum WakeDecision {
    435
        /// No action needed
    436
        None,
    437
        /// Wake the producer (buffer drained below low watermark)
    438
        WakeProducer,
    439
        /// Wake the consumer (data available)
    440
        WakeConsumer,
    441
        /// Pause the producer (buffer above high watermark)
    442
        PauseProducer,
    443
    }
    444
    445
    // ============================================================================
    446
    // AWP-04: HTTP/2 Stream Capacity
    447
    // ============================================================================
    448
    449
    /// HTTP/2 flow control window state.
    450
    ///
    451
    /// Tracks send and receive window sizes for stream-level flow control.
    452
    #[derive(Debug, Clone)]
    453
    pub struct StreamCapacity {
    454
        /// Connection-level send window
    455
        connection_send: i32,
    456
        /// Stream-level send window
    457
        stream_send: i32,
    458
        /// Receive window (how much we can receive)
    459
        receive_window: i32,
    460
        /// Initial window size
    461
        initial_window: i32,
    462
        /// Whether stream is blocked on flow control
    463
        is_blocked: bool,
    464
    }
    465
    466
    impl StreamCapacity {
    467
        /// Default window size (HTTP/2 spec: 65535).
    468
        pub const DEFAULT_WINDOW: i32 = 65535;
    469
    470
        /// Create with default windows.
    471
    0
        pub fn new() -> Self {
    472
    0
            Self {
    473
    0
                connection_send: Self::DEFAULT_WINDOW,
    474
    0
                stream_send: Self::DEFAULT_WINDOW,
    475
    0
                receive_window: Self::DEFAULT_WINDOW,
    476
    0
                initial_window: Self::DEFAULT_WINDOW,
    477
    0
                is_blocked: false,
    478
    0
            }
    479
    0
        }
    480
    481
        /// Create with custom initial window.
    482
    0
        pub fn with_initial_window(initial: i32) -> Self {
    483
    0
            Self {
    484
    0
                connection_send: initial,
    485
    0
                stream_send: initial,
    486
    0
                receive_window: initial,
    487
    0
                initial_window: initial,
    488
    0
                is_blocked: false,
    489
    0
            }
    490
    0
        }
    491
    492
        /// Reserve capacity for sending.
    493
    0
        pub fn reserve_send(&mut self, bytes: i32) -> Result<(), FlowControlError> {
    494
    0
            if bytes < 0 {
    495
    0
                return Err(FlowControlError::NegativeReservation);
    496
    0
            }
    497
    498
    0
            let available = self.available_send();
    499
    0
            if bytes > available {
    500
    0
                self.is_blocked = true;
    501
    0
                return Err(FlowControlError::InsufficientCapacity {
    502
    0
                    requested: bytes,
    503
    0
                    available,
    504
    0
                });
    505
    0
            }
    506
    507
    0
            self.stream_send -= bytes;
    508
    0
            self.connection_send -= bytes;
    509
    0
            self.is_blocked = false;
    510
    0
            Ok(())
    511
    0
        }
    512
    513
        /// Release send capacity (after WINDOW_UPDATE).
    514
    0
        pub fn release_send(&mut self, bytes: i32) {
    515
    0
            self.stream_send += bytes;
    516
    0
            self.connection_send += bytes;
    517
    0
            if self.available_send() > 0 {
    518
    0
                self.is_blocked = false;
    519
    0
            }
    520
    0
        }
    521
    522
        /// Consume receive window (data received).
    523
    0
        pub fn consume_receive(&mut self, bytes: i32) {
    524
    0
            self.receive_window -= bytes;
    525
    0
        }
    526
    527
        /// Replenish receive window (sending WINDOW_UPDATE).
    528
    0
        pub fn replenish_receive(&mut self, bytes: i32) {
    529
    0
            self.receive_window += bytes;
    530
    0
        }
    531
    532
        /// Get available send capacity.
    533
        #[must_use]
    534
    0
        pub fn available_send(&self) -> i32 {
    535
    0
            self.stream_send.min(self.connection_send).max(0)
    536
    0
        }
    537
    538
        /// Get available receive capacity.
    539
        #[must_use]
    540
    0
        pub fn available_receive(&self) -> i32 {
    541
    0
            self.receive_window.max(0)
    542
    0
        }
    543
    544
        /// Check if stream is blocked on flow control.
    545
        #[must_use]
    546
    0
        pub fn is_blocked(&self) -> bool {
    547
    0
            self.is_blocked
    548
    0
        }
    549
    550
        /// Check if receive window needs replenishment.
    551
        #[must_use]
    552
    0
        pub fn needs_window_update(&self) -> bool {
    553
    0
            self.receive_window < self.initial_window / 2
    554
    0
        }
    555
    }
    556
    557
    impl Default for StreamCapacity {
    558
    0
        fn default() -> Self {
    559
    0
            Self::new()
    560
    0
        }
    561
    }
    562
    563
    /// Flow control errors.
    564
    #[derive(Debug, Clone, PartialEq, Eq)]
    565
    pub enum FlowControlError {
    566
        /// Tried to reserve negative bytes
    567
        NegativeReservation,
    568
        /// Not enough capacity
    569
        InsufficientCapacity {
    570
            requested: i32,
    571
            available: i32,
    572
        },
    573
    }
    574
    575
    // ============================================================================
    576
    // AWP-09: Smart Payload Wake Skip
    577
    // ============================================================================
    578
    579
    /// Wake skip optimization state.
    580
    ///
    581
    /// Tracks whether a wakeup is actually needed or can be skipped
    582
    /// to avoid unnecessary context switches.
    583
    #[derive(Debug, Default)]
    584
    pub struct WakeSkipState {
    585
        /// Number of items pending
    586
        pending_items: usize,
    587
        /// Whether there's a registered waker
    588
        has_waker: bool,
    589
        /// Last poll had work to do
    590
        last_poll_had_work: bool,
    591
        /// Consecutive empty polls
    592
        empty_poll_count: u32,
    593
        /// Threshold for skipping wakes
    594
        skip_threshold: u32,
    595
    }
    596
    597
    impl WakeSkipState {
    598
        /// Create with skip threshold.
    599
    0
        pub fn new(skip_threshold: u32) -> Self {
    600
    0
            Self {
    601
    0
                pending_items: 0,
    602
    0
                has_waker: false,
    603
    0
                last_poll_had_work: false,
    604
    0
                empty_poll_count: 0,
    605
    0
                skip_threshold,
    606
    0
            }
    607
    0
        }
    608
    609
        /// Register that a waker exists.
    610
    0
        pub fn register_waker(&mut self) {
    611
    0
            self.has_waker = true;
    612
    0
        }
    613
    614
        /// Clear waker registration.
    615
    0
        pub fn clear_waker(&mut self) {
    616
    0
            self.has_waker = false;
    617
    0
        }
    618
    619
        /// Add pending items.
    620
    0
        pub fn add_pending(&mut self, count: usize) {
    621
    0
            self.pending_items += count;
    622
    0
        }
    623
    624
        /// Remove pending items.
    625
    0
        pub fn remove_pending(&mut self, count: usize) {
    626
    0
            self.pending_items = self.pending_items.saturating_sub(count);
    627
    0
        }
    628
    629
        /// Record poll result.
    630
    0
        pub fn record_poll(&mut self, had_work: bool) {
    631
    0
            self.last_poll_had_work = had_work;
    632
    0
            if had_work {
    633
    0
                self.empty_poll_count = 0;
    634
    0
            } else {
    635
    0
                self.empty_poll_count += 1;
    636
    0
            }
    637
    0
        }
    638
    639
        /// Determine if wake should be skipped.
    640
        #[must_use]
    641
    0
        pub fn should_skip_wake(&self) -> bool {
    642
            // Skip if:
    643
            // 1. No waker registered
    644
            // 2. Already has pending items (will be polled anyway)
    645
            // 3. Had recent empty polls (probably will be empty again)
    646
    0
            if !self.has_waker {
    647
    0
                return true;
    648
    0
            }
    649
    0
            if self.pending_items > 0 && self.last_poll_had_work {
    650
    0
                return true; // Already has work queued
    651
    0
            }
    652
    0
            if self.empty_poll_count >= self.skip_threshold {
    653
    0
                return true; // Likely to be empty again
    654
    0
            }
    655
    0
            false
    656
    0
        }
    657
    658
        /// Check if wake is needed.
    659
        #[must_use]
    660
    0
        pub fn needs_wake(&self) -> bool {
    661
    0
            !self.should_skip_wake() && self.pending_items > 0
    662
    0
        }
    663
    664
        /// Get pending count.
    665
        #[must_use]
    666
    0
        pub fn pending(&self) -> usize {
    667
    0
            self.pending_items
    668
    0
        }
    669
    670
        /// Reset empty poll tracking (after successful wake).
    671
    0
        pub fn reset_tracking(&mut self) {
    672
    0
            self.empty_poll_count = 0;
    673
    0
        }
    674
    }
    675
    676
    // ============================================================================
    677
    // Tests
    678
    // ============================================================================
    679
    680
    #[cfg(test)]
    681
    mod tests {
    682
        use super::*;
    683
    684
        // ------------------------------------------------------------------------
    685
        // AsyncResult Tests (LCP-12)
    686
        // ------------------------------------------------------------------------
    687
    688
        #[test]
    689
        fn test_async_result_states() {
    690
            let async_val: AsyncResult<i32, &str> = AsyncResult::Async(42);
    691
            let sync_val: AsyncResult<i32, &str> = AsyncResult::Sync(42);
    692
            let err: AsyncResult<i32, &str> = AsyncResult::Error("fail");
    693
    694
            assert!(async_val.is_async());
    695
            assert!(!async_val.is_sync());
    696
            assert!(!async_val.is_error());
    697
    698
            assert!(!sync_val.is_async());
    699
            assert!(sync_val.is_sync());
    700
            assert!(!sync_val.is_error());
    701
    702
            assert!(err.is_error());
    703
            assert!(!err.is_async());
    704
            assert!(!err.is_sync());
    705
    706
            assert_eq!(async_val.into_result(), Ok(42));
    707
            assert_eq!(sync_val.into_result(), Ok(42));
    708
            assert_eq!(err.into_result(), Err("fail"));
    709
        }
    710
    711
        #[test]
    712
        fn test_async_result_map() {
    713
            let async_val: AsyncResult<i32, &str> = AsyncResult::Async(10);
    714
            let sync_val: AsyncResult<i32, &str> = AsyncResult::Sync(10);
    715
            let err: AsyncResult<i32, &str> = AsyncResult::Error("fail");
    716
    717
            let mapped_async = async_val.map(|x| x * 2);
    718
            let mapped_sync = sync_val.map(|x| x * 2);
    719
            let mapped_err = err.map(|x| x * 2);
    720
    721
            assert!(matches!(mapped_async, AsyncResult::Async(20)));
    722
            assert!(matches!(mapped_sync, AsyncResult::Sync(20)));
    723
            assert!(matches!(mapped_err, AsyncResult::Error("fail")));
    724
        }
    725
    726
        // ------------------------------------------------------------------------
    727
        // BoundedQueue Tests (AWP-11)
    728
        // ------------------------------------------------------------------------
    729
    730
        #[test]
    731
        fn test_bounded_queue_basic() {
    732
            let mut queue: BoundedQueue<i32> = BoundedQueue::new(5);
    733
    734
            assert!(queue.is_empty());
    735
            assert!(!queue.is_full());
    736
    737
            queue.try_push(1).unwrap();
    738
            queue.try_push(2).unwrap();
    739
            queue.try_push(3).unwrap();
    740
    741
            assert_eq!(queue.len(), 3);
    742
            assert_eq!(queue.pop(), Some(1));
    743
            assert_eq!(queue.pop(), Some(2));
    744
            assert_eq!(queue.len(), 1);
    745
        }
    746
    747
        #[test]
    748
        fn test_bounded_queue_backpressure() {
    749
            let mut queue: BoundedQueue<i32> = BoundedQueue::new(3);
    750
    751
            assert!(queue.try_push(1).is_ok());
    752
            assert!(queue.try_push(2).is_ok());
    753
            assert!(queue.try_push(3).is_ok());
    754
            assert!(queue.is_full());
    755
    756
            assert!(queue.try_push(4).is_err());
    757
    758
            queue.pop();
    759
            assert!(queue.try_push(4).is_ok());
    760
        }
    761
    762
        #[test]
    763
        fn test_bounded_queue_comprehensive() {
    764
            let mut queue: BoundedQueue<String> = BoundedQueue::new(3);
    765
    766
            // Test peek
    767
            assert!(queue.peek().is_none());
    768
            queue.try_push("first".to_string()).unwrap();
    769
            assert_eq!(queue.peek(), Some(&"first".to_string()));
    770
    771
            // Test remaining
    772
            assert_eq!(queue.remaining(), 2);
    773
            queue.try_push("second".to_string()).unwrap();
    774
            assert_eq!(queue.remaining(), 1);
    775
    776
            // Test clear
    777
            queue.clear();
    778
            assert!(queue.is_empty());
    779
            assert_eq!(queue.remaining(), 3);
    780
        }
    781
    782
        // ------------------------------------------------------------------------
    783
        // ReserveStrategy Tests (AWP-13)
    784
        // ------------------------------------------------------------------------
    785
    786
        #[test]
    787
        fn test_reserve_strategy_variants() {
    788
            assert_eq!(reserve_capacity(100, ReserveStrategy::Exact), 100);
    789
            assert_eq!(reserve_capacity(100, ReserveStrategy::Grow50), 150);
    790
            assert_eq!(reserve_capacity(100, ReserveStrategy::Double), 200);
    791
            assert_eq!(reserve_capacity(100, ReserveStrategy::PowerOfTwo), 128);
    792
        }
    793
    794
        #[test]
    795
        fn test_reserve_capacity_edge_cases() {
    796
            assert_eq!(reserve_capacity(0, ReserveStrategy::Exact), 0);
    797
            assert_eq!(reserve_capacity(0, ReserveStrategy::Double), 0);
    798
            assert_eq!(reserve_capacity(1, ReserveStrategy::PowerOfTwo), 1);
    799
            assert_eq!(reserve_capacity(3, ReserveStrategy::PowerOfTwo), 4);
    800
        }
    801
    802
        #[test]
    803
        fn test_strategic_buffer() {
    804
            let mut buf = StrategicBuffer::with_capacity(10, ReserveStrategy::Double);
    805
            buf.write(b"hello");
    806
            assert_eq!(buf.len(), 5);
    807
            assert_eq!(buf.as_slice(), b"hello");
    808
    809
            buf.write(b" world");
    810
            assert_eq!(buf.len(), 11);
    811
            assert_eq!(buf.as_slice(), b"hello world");
    812
    813
            buf.clear();
    814
            assert!(buf.is_empty());
    815
        }
    816
    817
        // ------------------------------------------------------------------------
    818
        // GraphReuseCounter Tests (LCP-08)
    819
        // ------------------------------------------------------------------------
    820
    821
        #[test]
    822
        fn test_graph_reuse_counter() {
    823
            let mut counter = GraphReuseCounter::new(3);
    824
    825
            assert!(!counter.is_hot());
    826
            assert!(!counter.should_cache());
    827
    828
            counter.record_use();
    829
            counter.record_use();
    830
            assert!(!counter.is_hot());
    831
    832
            counter.record_use();
    833
            assert!(counter.is_hot());
    834
            assert!(counter.should_cache());
    835
            assert_eq!(counter.count(), 3);
    836
    837
            counter.reset();
    838
            assert!(!counter.is_hot());
    839
            assert_eq!(counter.count(), 0);
    840
        }
    841
    842
        // ------------------------------------------------------------------------
    843
        // DualWakerState Tests (AWP-03)
    844
        // ------------------------------------------------------------------------
    845
    846
        #[test]
    847
        fn test_dual_waker_state() {
    848
            let mut state = DualWakerState::new(20, 80);
    849
    850
            assert!(state.can_produce());
    851
            assert!(!state.can_consume());
    852
    853
            // Fill above high watermark
    854
            let decision = state.update_fill(85);
    855
            assert_eq!(decision, WakeDecision::PauseProducer);
    856
            assert!(!state.can_produce());
    857
            assert!(state.can_consume());
    858
    859
            // Drain below low watermark
    860
            let decision = state.update_fill(15);
    861
            assert_eq!(decision, WakeDecision::WakeProducer);
    862
            assert!(state.can_produce());
    863
        }
    864
    865
        #[test]
    866
        fn test_dual_waker_consumer_wake() {
    867
            let mut state = DualWakerState::new(20, 80);
    868
            state.consumer_wait();
    869
    870
            // Add data while consumer waiting
    871
            let decision = state.update_fill(50);
    872
            assert_eq!(decision, WakeDecision::WakeConsumer);
    873
        }
    874
    875
        #[test]
    876
        fn test_dual_waker_edge_cases() {
    877
            let mut state = DualWakerState::new(20, 80);
    878
    879
            state.producer_wait();
    880
            state.consumer_wait();
    881
    882
            state.producer_woke();
    883
            state.consumer_woke();
    884
    885
            // Test clamping
    886
            state.update_fill(200); // Should clamp to 100
    887
            assert!(!state.can_produce());
    888
        }
    889
    890
        // ------------------------------------------------------------------------
    891
        // StreamCapacity Tests (AWP-04)
    892
        // ------------------------------------------------------------------------
    893
    894
        #[test]
    895
        fn test_stream_capacity_basic() {
    896
            let mut cap = StreamCapacity::new();
    897
    898
            assert_eq!(cap.available_send(), StreamCapacity::DEFAULT_WINDOW);
    899
            assert!(!cap.is_blocked());
    900
    901
            cap.reserve_send(1000).unwrap();
    902
            assert_eq!(
    903
                cap.available_send(),
    904
                StreamCapacity::DEFAULT_WINDOW - 1000
    905
            );
    906
        }
    907
    908
        #[test]
    909
        fn test_stream_capacity_blocking() {
    910
            let mut cap = StreamCapacity::with_initial_window(100);
    911
    912
            let result = cap.reserve_send(150);
    913
            assert!(result.is_err());
    914
            assert!(cap.is_blocked());
    915
    916
            cap.release_send(100);
    917
            assert!(!cap.is_blocked());
    918
        }
    919
    920
        #[test]
    921
        fn test_stream_capacity_window_ops() {
    922
            let mut cap = StreamCapacity::new();
    923
            // Default window is 65535, needs_window_update when < 32767
    924
    925
            // Consume more than half to trigger update
    926
            cap.consume_receive(40000);
    927
            assert!(cap.needs_window_update()); // 25535 < 32767
    928
    929
            cap.replenish_receive(20000);
    930
            assert!(!cap.needs_window_update()); // 45535 > 32767
    931
        }
    932
    933
        #[test]
    934
        fn test_flow_control_error() {
    935
            let err1 = FlowControlError::NegativeReservation;
    936
            let err2 = FlowControlError::InsufficientCapacity {
    937
                requested: 100,
    938
                available: 50,
    939
            };
    940
    941
            assert_eq!(err1, FlowControlError::NegativeReservation);
    942
            assert!(matches!(
    943
                err2,
    944
                FlowControlError::InsufficientCapacity { .. }
    945
            ));
    946
        }
    947
    948
        // ------------------------------------------------------------------------
    949
        // WakeSkipState Tests (AWP-09)
    950
        // ------------------------------------------------------------------------
    951
    952
        #[test]
    953
        fn test_wake_skip_state() {
    954
            let mut state = WakeSkipState::new(3);
    955
    956
            // No waker = always skip
    957
            assert!(state.should_skip_wake());
    958
    959
            state.register_waker();
    960
            assert!(!state.should_skip_wake());
    961
    962
            // Accumulate empty polls
    963
            state.record_poll(false);
    964
            state.record_poll(false);
    965
            state.record_poll(false);
    966
            assert!(state.should_skip_wake());
    967
    968
            // Reset tracking
    969
            state.reset_tracking();
    970
            assert!(!state.should_skip_wake());
    971
        }
    972
    973
        #[test]
    974
        fn test_wake_skip_needs_wake() {
    975
            let mut state = WakeSkipState::new(3);
    976
            state.register_waker();
    977
    978
            // No pending items
    979
            assert!(!state.needs_wake());
    980
    981
            // Add pending
    982
            state.add_pending(5);
    983
            assert!(state.needs_wake());
    984
            assert_eq!(state.pending(), 5);
    985
    986
            // Remove some
    987
            state.remove_pending(3);
    988
            assert_eq!(state.pending(), 2);
    989
        }
    990
    991
        #[test]
    992
        fn test_wake_skip_tracking() {
    993
            let mut state = WakeSkipState::new(5);
    994
            state.register_waker();
    995
    996
            // Work resets empty poll count
    997
            state.record_poll(false);
    998
            state.record_poll(false);
    999
            state.record_poll(true); // Had work
    1000
            state.record_poll(false);
    1001
    1002
            // Should not skip yet (only 1 empty after work)
    1003
            assert!(!state.should_skip_wake());
    1004
        }
    1005
    1006
        // ------------------------------------------------------------------------
    1007
        // Falsification Tests
    1008
        // ------------------------------------------------------------------------
    1009
    1010
        /// FALSIFICATION: BoundedQueue capacity invariant.
    1011
        /// Queue must never exceed its capacity.
    1012
        #[test]
    1013
        fn test_falsify_bounded_queue_capacity_invariant() {
    1014
            for cap in [1, 5, 10, 100] {
    1015
                let mut queue: BoundedQueue<usize> = BoundedQueue::new(cap);
    1016
    1017
                // Try to push more than capacity
    1018
                for i in 0..cap * 2 {
    1019
                    let _ = queue.try_push(i);
    1020
                }
    1021
    1022
                assert!(
    1023
                    queue.len() <= cap,
    1024
                    "FALSIFICATION FAILED: Queue exceeded capacity {} with len {}",
    1025
                    cap,
    1026
                    queue.len()
    1027
                );
    1028
            }
    1029
        }
    1030
    1031
        /// FALSIFICATION: ReserveStrategy must always return >= needed.
    1032
        #[test]
    1033
        fn test_falsify_reserve_strategy_minimum() {
    1034
            for needed in [0, 1, 10, 100, 1000, 10000] {
    1035
                for strategy in [
    1036
                    ReserveStrategy::Exact,
    1037
                    ReserveStrategy::Grow50,
    1038
                    ReserveStrategy::Double,
    1039
                    ReserveStrategy::PowerOfTwo,
    1040
                ] {
    1041
                    let reserved = reserve_capacity(needed, strategy);
    1042
                    assert!(
    1043
                        reserved >= needed,
    1044
                        "FALSIFICATION FAILED: reserve_capacity({}, {:?}) = {} < {}",
    1045
                        needed,
    1046
                        strategy,
    1047
                        reserved,
    1048
                        needed
    1049
                    );
    1050
                }
    1051
            }
    1052
        }
    1053
    1054
        /// FALSIFICATION: GraphReuseCounter hot transition.
    1055
        /// Must become hot exactly at threshold.
    1056
        #[test]
    1057
        fn test_falsify_graph_reuse_threshold() {
    1058
            for threshold in [1, 5, 10, 100] {
    1059
                let mut counter = GraphReuseCounter::new(threshold);
    1060
    1061
                // Before threshold
    1062
                for _ in 0..threshold - 1 {
    1063
                    counter.record_use();
    1064
                    assert!(
    1065
                        !counter.is_hot(),
    1066
                        "FALSIFICATION FAILED: Became hot before threshold {}",
    1067
                        threshold
    1068
                    );
    1069
                }
    1070
    1071
                // At threshold
    1072
                counter.record_use();
    1073
                assert!(
    1074
                    counter.is_hot(),
    1075
                    "FALSIFICATION FAILED: Not hot at threshold {}",
    1076
                    threshold
    1077
                );
    1078
            }
    1079
        }
    1080
    1081
        /// FALSIFICATION: StreamCapacity window consistency.
    1082
        #[test]
    1083
        fn test_falsify_stream_capacity_consistency() {
    1084
            let mut cap = StreamCapacity::new();
    1085
            let initial = cap.available_send();
    1086
    1087
            // Reserve and release should return to initial
    1088
            cap.reserve_send(10000).unwrap();
    1089
            cap.release_send(10000);
    1090
            assert_eq!(
    1091
                cap.available_send(),
    1092
                initial,
    1093
                "FALSIFICATION FAILED: Window not restored after reserve+release"
    1094
            );
    1095
        }
    1096
    }