Function process_retry_queue

nodedb/src/event/consumer.rs:397–424

Process the retry queue: DLQ exhausted entries and retry ready ones.

(
    retry_queue: &mut TriggerRetryQueue,
    trigger_dlq: &Arc<std::sync::Mutex<TriggerDlq>>,
    shared_state: &Arc<SharedState>,
)

/// Process the retry queue: DLQ exhausted entries and retry ready ones.
async fn process_retry_queue(
    retry_queue: &mut TriggerRetryQueue,
    trigger_dlq: &Arc<std::sync::Mutex<TriggerDlq>>,
    shared_state: &Arc<SharedState>,
) {
    let (ready, exhausted) = retry_queue.drain_due();
    if !exhausted.is_empty() {
        let mut dlq = trigger_dlq.lock().unwrap_or_else(|p| p.into_inner());
        for entry in &exhausted {
            let _ = dlq.enqueue(super::trigger::dlq::DlqEnqueueParams {
                tenant_id: entry.tenant_id,
                source_collection: entry.collection.clone(),
                row_id: entry.row_id.clone(),
                operation: entry.operation.clone(),
                trigger_name: entry.trigger_name.clone(),
                error: entry.last_error.clone(),
                retry_count: entry.attempts,
                source_lsn: entry.source_lsn,
                source_sequence: entry.source_sequence,
            });
        }
        // dlq MutexGuard dropped before any await.
    }

    for entry in ready {
        super::trigger::dispatcher::retry_single(&entry, shared_state, retry_queue).await;
    }
}

#[cfg(test)]
mod tests {