Merge pull request #165 from Agencybuilds/feat/dead-letter-queue

feat: Implement Dead Letter Queue for Failed Message Processing

Author: JamesEjembi

docs/architecture/dead-letter-queue.md

@@ -0,0 +1,23 @@
+# Dead Letter Queue Architecture
+
+## Overview
+The Dead Letter Queue (DLQ) is designed to handle failed message processing across all services. It ensures that critical messages are not lost when downstream services are unavailable, or processing fails repeatedly. 
+
+## Technical Requirements
+- **Performance**: Operations target < 100ms P99 for critical paths.
+- **Availability**: 99.99% uptime via resilient in-memory queuing combined with persistent flushing (future scope).
+- **Scope**: System-wide implementation affecting core pools and net pipelines.
+
+## Component Design
+`DeadLetterQueue` holds a bounded deque of `FailedMessage` records. Each record contains:
+- `message_id`
+- `payload`
+- `error_reason`
+- `timestamp`
+- `retry_count`
+
+## Blue-Green Strategy & Canary Analysis
+The DLQ component will be initially deployed to canary nodes to monitor processing and dropped message rates. Upon stable metric emission, it will be rolled out via a blue-green deployment strategy.
+
+## Monitoring & Alerting
+Built-in metrics track `total_enqueued`, `total_processed`, and `total_dropped`. Dashboards will monitor these metrics. An alert is triggered if `total_dropped` increases over time.

docs/runbooks/dlq-monitoring.md

@@ -0,0 +1,26 @@
+# Runbook: Dead Letter Queue (DLQ) Monitoring and Alerting
+
+## Overview
+This runbook covers procedures to follow when DLQ alerts are triggered.
+
+## Alerts
+
+### 1. High Dropped Message Rate
+**Condition**: `dlq.metrics.total_dropped` increases by more than 100 within a 5-minute window.
+**Impact**: Messages are being lost because the queue is full, possibly due to a stuck downstream processor or unusually high failure rates.
+**Actions**:
+1. Check the error reasons for the failed messages.
+2. Verify if downstream processing nodes are healthy.
+3. Temporarily increase the queue capacity if necessary and safe.
+
+### 2. High Enqueue Rate
+**Condition**: Spike in `total_enqueued`.
+**Impact**: Elevated message processing failures.
+**Actions**:
+1. Identify the service producing the failed messages.
+2. Check network stability and dependency health.
+
+## Dashboards
+- DLQ processing rate
+- DLQ depth
+- Drop rate over time

src/core/dlq.rs

@@ -0,0 +1,97 @@
+use alloc::vec::Vec;
+use alloc::string::String;
+use alloc::collections::VecDeque;
+
+#[derive(Clone, Debug, PartialEq)]
+pub struct FailedMessage {
+    pub message_id: String,
+    pub payload: Vec<u8>,
+    pub error_reason: String,
+    pub timestamp: u64,
+    pub retry_count: u8,
+}
+
+pub struct DeadLetterQueue {
+    pub queue: VecDeque<FailedMessage>,
+    pub max_size: usize,
+    pub metrics: DlqMetrics,
+}
+
+#[derive(Default, Clone, Debug, PartialEq)]
+pub struct DlqMetrics {
+    pub total_enqueued: u64,
+    pub total_processed: u64,
+    pub total_dropped: u64,
+}
+
+impl DeadLetterQueue {
+    pub fn new(max_size: usize) -> Self {
+        Self {
+            queue: VecDeque::new(),
+            max_size,
+            metrics: DlqMetrics::default(),
+        }
+    }
+
+    pub fn enqueue(&mut self, message: FailedMessage) -> Result<(), &'static str> {
+        if self.queue.len() >= self.max_size {
+            self.metrics.total_dropped += 1;
+            return Err("Queue is full");
+        }
+        self.queue.push_back(message);
+        self.metrics.total_enqueued += 1;
+        Ok(())
+    }
+
+    pub fn dequeue(&mut self) -> Option<FailedMessage> {
+        if let Some(message) = self.queue.pop_front() {
+            self.metrics.total_processed += 1;
+            Some(message)
+        } else {
+            None
+        }
+    }
+
+    pub fn get_metrics(&self) -> DlqMetrics {
+        self.metrics.clone()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alloc::string::ToString;
+    use alloc::vec;
+
+    #[test]
+    fn test_enqueue_dequeue() {
+        let mut dlq = DeadLetterQueue::new(10);
+        let msg = FailedMessage {
+            message_id: "1".to_string(),
+            payload: vec![1, 2, 3],
+            error_reason: "timeout".to_string(),
+            timestamp: 123456789,
+            retry_count: 0,
+        };
+        assert!(dlq.enqueue(msg.clone()).is_ok());
+        let dequeued = dlq.dequeue();
+        assert_eq!(dequeued, Some(msg));
+        assert_eq!(dlq.metrics.total_enqueued, 1);
+        assert_eq!(dlq.metrics.total_processed, 1);
+    }
+
+    #[test]
+    fn test_queue_full() {
+        let mut dlq = DeadLetterQueue::new(1);
+        let msg = FailedMessage {
+            message_id: "1".to_string(),
+            payload: vec![1, 2, 3],
+            error_reason: "timeout".to_string(),
+            timestamp: 123456789,
+            retry_count: 0,
+        };
+        assert!(dlq.enqueue(msg.clone()).is_ok());
+        assert!(dlq.enqueue(msg.clone()).is_err());
+        assert_eq!(dlq.metrics.total_dropped, 1);
+    }
+}

src/core/mod.rs

@@ -1,3 +1,4 @@
 pub mod liveness;
 pub mod poc;
 pub mod pool;
+pub mod dlq;