Drain all available events from the ring buffer (up to DRAIN_BATCH_LIMIT). Returns the drained events for async processing (trigger dispatch).
(
rx: &mut EventConsumerRx,
metrics: &CoreMetrics,
core_id: usize,
last_sequence: &mut u64,
last_lsn: &mut Lsn,
)
| 88 | /// Drain all available events from the ring buffer (up to DRAIN_BATCH_LIMIT). |
| 89 | /// Returns the drained events for async processing (trigger dispatch). |
| 90 | pub fn drain_ring_buffer( |
| 91 | rx: &mut EventConsumerRx, |
| 92 | metrics: &CoreMetrics, |
| 93 | core_id: usize, |
| 94 | last_sequence: &mut u64, |
| 95 | last_lsn: &mut Lsn, |
| 96 | ) -> Vec<WriteEvent> { |
| 97 | let mut events = Vec::new(); |
| 98 | while let Some(event) = rx.try_recv() { |
| 99 | detect_sequence_gap(core_id, &event, *last_sequence, metrics); |
| 100 | record_event(core_id, &event, metrics); |
| 101 | |
| 102 | *last_sequence = event.sequence; |
| 103 | if event.lsn.is_ahead_of(*last_lsn) { |
| 104 | *last_lsn = event.lsn; |
| 105 | } |
| 106 | |
| 107 | events.push(event); |
| 108 | if (events.len() as u32).is_multiple_of(DRAIN_BATCH_LIMIT) { |
| 109 | break; |
| 110 | } |
| 111 | } |
| 112 | events |
| 113 | } |
| 114 | |
| 115 | /// Drain the ring buffer, skipping events with sequence <= last_sequence. |
| 116 | /// Used after WAL catchup to discard stale events that overlap with replay. |
no test coverage detected