fix(daemon): log delivery failures dropped without a DLQ sink

CHANGELOG.md

@@ -4,6 +4,10 @@ All notable changes to RSigma are documented in this file. Each entry correspond
 
 ## [Unreleased]
 
+### Delivery failures without a DLQ sink are logged (#615)
+
+When a sink exhausted its retries and no `--dlq` was set, the daemon counted the dropped result on `rsigma_dlq_events_total` and logged it once, at `debug`, so every later dropped alert left no trace in the logs. Each drop now produces a `warn` log, rate-limited to one every 10 seconds; each warning carries the number of results dropped since the previous one and the last delivery error. A failed write to a configured DLQ sink is now counted on the new `rsigma_dlq_write_failures_total` counter, and the delivery layer and the daemon's DLQ bridge log a failure they could not queue instead of discarding it silently. The docs now say that a sink commits a failed result by handing it to the DLQ writer, so the source ack does not wait for the DLQ sink's write.
+
 ### Match detail costs one evaluation pass (#613)
 
 With `MatchDetailLevel` above `Off`, a rule that fired was evaluated again to collect its selection names and then walked a third time to collect its field matches, re-running every matcher, including regexes. Field matches are now recorded while the selections are evaluated, so a firing rule costs one detail pass, and array members are matched once per quantifier instead of again for the report. The output is unchanged. On the `eval_matched_detail` benchmark (200 rules that all fire), `Full` drops from 314 to 307 µs and `Off` from 252 to 244 µs.

crates/rsigma-cli/README.md

@@ -493,7 +493,8 @@ The first three triggers all funnel through one debounced reload task that re-re
 | `rsigma_output_queue_depth` | gauge | | Current results buffered in engine-to-sink channel |
 | `rsigma_back_pressure_events_total` | counter | | Times a source was blocked on a full event channel |
 | `rsigma_uptime_seconds` | gauge | | Daemon uptime in seconds |
-| `rsigma_dlq_events_total` | counter | | Events routed to the dead-letter queue |
+| `rsigma_dlq_events_total` | counter | | Events routed to the dead-letter queue, with or without a `--dlq` sink |
+| `rsigma_dlq_write_failures_total` | counter | | Dead-letter queue entries the `--dlq` sink failed to write |
 | `rsigma_source_resolves_total` | counter | `source_id` | Total dynamic source resolution attempts |
 | `rsigma_source_resolve_errors_total` | counter | `source_id` | Total dynamic source resolution failures |
 | `rsigma_source_resolve_latency_seconds` | histogram | | Source resolution latency |

crates/rsigma-cli/src/daemon/metrics.rs

@@ -26,6 +26,7 @@ pub struct Metrics {
     pub pipeline_latency: Histogram,
     pub batch_size_histogram: Histogram,
     pub dlq_events: IntCounter,
+    pub dlq_write_failures: IntCounter,
     pub sink_queue_depth: IntGaugeVec,
     pub sink_retries_total: IntCounterVec,
     pub sink_dropped_total: IntCounterVec,
@@ -240,6 +241,11 @@ impl Metrics {
             "Events routed to dead-letter queue",
         ))
         .unwrap();
+        let dlq_write_failures = IntCounter::with_opts(Opts::new(
+            "rsigma_dlq_write_failures_total",
+            "Dead-letter queue entries the DLQ sink failed to write",
+        ))
+        .unwrap();
         let sink_queue_depth = IntGaugeVec::new(
             Opts::new(
                 "rsigma_sink_queue_depth",
@@ -357,6 +363,9 @@ impl Metrics {
             .register(Box::new(batch_size_histogram.clone()))
             .unwrap();
         registry.register(Box::new(dlq_events.clone())).unwrap();
+        registry
+            .register(Box::new(dlq_write_failures.clone()))
+            .unwrap();
         registry
             .register(Box::new(sink_queue_depth.clone()))
             .unwrap();
@@ -1156,6 +1165,7 @@ impl Metrics {
             pipeline_latency,
             batch_size_histogram,
             dlq_events,
+            dlq_write_failures,
             sink_queue_depth,
             sink_retries_total,
             sink_dropped_total,

crates/rsigma-cli/src/daemon/server.rs

@@ -2,7 +2,7 @@ use std::collections::BTreeMap;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::sync::atomic::{AtomicI64, AtomicU64, Ordering};
-use std::time::Instant;
+use std::time::{Duration, Instant};
 
 use arc_swap::ArcSwap;
 use axum::extract::{Query, State};
@@ -39,6 +39,54 @@ struct DlqEntry {
     timestamp: String,
 }
 
+/// Minimum gap between two warnings about delivery failures dropped for lack
+/// of a DLQ sink.
+const DROPPED_FAILURE_WARN_INTERVAL: Duration = Duration::from_secs(10);
+
+/// Warns about delivery failures dropped because no `--dlq` sink is set. The
+/// first drop warns at once; later drops fold into at most one warning per
+/// [`DROPPED_FAILURE_WARN_INTERVAL`] that carries their count.
+#[derive(Default)]
+struct DroppedFailureLog {
+    last_warn: Option<Instant>,
+    pending: u64,
+}
+
+impl DroppedFailureLog {
+    /// Record one dropped failure. Returns the count it warned about, or
+    /// `None` when the warning is deferred to a later drop or [`Self::flush`].
+    fn record(&mut self, error: &str, now: Instant) -> Option<u64> {
+        self.pending += 1;
+        if self
+            .last_warn
+            .is_some_and(|last| now.duration_since(last) < DROPPED_FAILURE_WARN_INTERVAL)
+        {
+            return None;
+        }
+        let dropped = std::mem::take(&mut self.pending);
+        self.last_warn = Some(now);
+        tracing::warn!(
+            dropped,
+            error,
+            "Delivery failures dropped: no DLQ sink is configured (use --dlq to persist them)"
+        );
+        Some(dropped)
+    }
+
+    /// Warn about any drops still waiting for the next interval.
+    fn flush(&mut self) -> Option<u64> {
+        let dropped = std::mem::take(&mut self.pending);
+        if dropped == 0 {
+            return None;
+        }
+        tracing::warn!(
+            dropped,
+            "Delivery failures dropped: no DLQ sink is configured (use --dlq to persist them)"
+        );
+        Some(dropped)
+    }
+}
+
 /// One evaluated batch waiting for ordered sink/ack dispatch.
 struct EvaluatedBatch {
     outcome: BatchProcessOutcome,
@@ -1999,13 +2047,17 @@ pub async fn run_daemon(config: DaemonConfig) {
     let (df_tx, mut df_rx) = mpsc::channel::<DeliveryFailure>(buffer_size);
     let df_handle = tokio::spawn(async move {
         while let Some(failure) = df_rx.recv().await {
-            let _ = dlq_tx
-                .send(DlqEntry {
-                    original_event: failure.serialized,
-                    error: failure.error,
-                    timestamp: chrono::Utc::now().to_rfc3339(),
-                })
-                .await;
+            let entry = DlqEntry {
+                original_event: failure.serialized,
+                error: failure.error,
+                timestamp: chrono::Utc::now().to_rfc3339(),
+            };
+            if let Err(mpsc::error::SendError(entry)) = dlq_tx.send(entry).await {
+                tracing::warn!(
+                    error = %entry.error,
+                    "DLQ writer closed, delivery failure dropped"
+                );
+            }
         }
     });
 
@@ -2270,24 +2322,27 @@ pub async fn run_daemon(config: DaemonConfig) {
         dispatcher.shutdown().await;
     });
 
-    // DLQ writer task: writes DLQ entries to the configured DLQ sink.
+    // DLQ writer task: writes DLQ entries to the configured DLQ sink. Without
+    // one, only delivery failures arrive here (parse errors are not routed),
+    // and they are dropped with a rate-limited warning.
     let dlq_metrics = metrics.clone();
     let dlq_handle = tokio::spawn(async move {
         tracing::debug!("DLQ task started");
         let mut dlq_sink = dlq_sink;
-        let mut no_sink_logged = false;
+        let mut dropped = DroppedFailureLog::default();
         while let Some(entry) = dlq_rx.recv().await {
             dlq_metrics.dlq_events.inc();
             if let Some(ref mut sink) = dlq_sink {
                 let json = serde_json::to_string(&entry).unwrap_or_default();
                 if let Err(e) = sink.send_raw(&json).await {
-                    tracing::warn!(error = %e, "Failed to write to DLQ sink");
+                    dlq_metrics.dlq_write_failures.inc();
+                    tracing::warn!(error = %e, "Failed to write to DLQ sink, entry dropped");
                 }
-            } else if !no_sink_logged {
-                tracing::debug!("DLQ entry counted but no sink configured (use --dlq to persist)");
-                no_sink_logged = true;
+            } else {
+                dropped.record(&entry.error, Instant::now());
             }
         }
+        dropped.flush();
         tracing::debug!("DLQ task finished");
     });
 
@@ -4516,6 +4571,23 @@ impl rsigma_runtime::LogsService for OtlpLogsGrpcService {
 mod tests {
     use super::*;
 
+    #[test]
+    fn dropped_failure_warnings_are_rate_limited() {
+        let mut log = DroppedFailureLog::default();
+        let t0 = Instant::now();
+        assert_eq!(log.record("boom", t0), Some(1));
+        assert_eq!(log.record("boom", t0 + Duration::from_secs(1)), None);
+        assert_eq!(log.record("boom", t0 + Duration::from_secs(9)), None);
+        assert_eq!(
+            log.record("boom", t0 + DROPPED_FAILURE_WARN_INTERVAL),
+            Some(3)
+        );
+        assert_eq!(log.flush(), None);
+        assert_eq!(log.record("boom", t0 + Duration::from_secs(11)), None);
+        assert_eq!(log.flush(), Some(1));
+        assert_eq!(log.flush(), None);
+    }
+
     #[test]
     fn detection_inflight_default_scales_to_five() {
         assert_eq!(default_detect_inflight(1), 1);

crates/rsigma-cli/tests/cli_daemon_webhook.rs

@@ -12,7 +12,7 @@ mod common;
 
 use std::time::Duration;
 
-use common::{DaemonProcess, SIMPLE_RULE, http_post, poll_until, temp_file};
+use common::{DaemonProcess, SIMPLE_RULE, http_get, http_post, poll_until, temp_file};
 use wiremock::matchers::method;
 use wiremock::{Mock, MockServer, ResponseTemplate};
 
@@ -421,3 +421,57 @@ fn webhook_does_not_block_file_sink_under_fanout() {
         "an unreachable webhook must not block the file sink behind it",
     );
 }
+
+#[test]
+fn delivery_failures_without_dlq_log_one_rate_limited_warning() {
+    let cfg = WEBHOOK_RETRY_CFG
+        .replace("__URL__", "http://127.0.0.1:1")
+        .replace("attempts: 3", "attempts: 1")
+        .replace("backoff: 100ms", "backoff: 10ms");
+    let webhook = temp_file(".yml", &cfg);
+    let rule = temp_file(".yml", SIMPLE_RULE);
+    let daemon = DaemonProcess::spawn_http_with_args(
+        rule.path().to_str().unwrap(),
+        &["--webhook", webhook.path().to_str().unwrap()],
+    );
+
+    for _ in 0..3 {
+        let (status, _) = http_post(&daemon.url("/api/v1/events"), MATCHING_EVENT);
+        assert_eq!(status, 200);
+    }
+
+    let metrics = poll_until(Duration::from_secs(10), || {
+        let (_, body) = http_get(&daemon.url("/metrics"));
+        body.contains(r#"rsigma_sink_delivery_failures_total{sink="slack"} 3"#)
+            .then_some(body)
+    })
+    .expect("all three deliveries should fail");
+    assert!(
+        metrics.contains("rsigma_dlq_write_failures_total 0"),
+        "nothing reaches a DLQ sink when none is configured: {metrics}"
+    );
+
+    let warnings: Vec<String> = poll_until(Duration::from_secs(5), || {
+        let lines: Vec<String> = daemon
+            .stderr_lines()
+            .into_iter()
+            .filter(|l| l.contains("no DLQ sink is configured"))
+            .collect();
+        (!lines.is_empty()).then_some(lines)
+    })
+    .expect("a dropped delivery failure must be logged without --dlq");
+    assert_eq!(
+        warnings.len(),
+        1,
+        "later drops within the interval fold into the next warning: {warnings:?}"
+    );
+    let warning: serde_json::Value = serde_json::from_str(&warnings[0]).unwrap();
+    assert_eq!(warning["level"], "WARN");
+    assert_eq!(warning["fields"]["dropped"], 1);
+    assert!(
+        warning["fields"]["error"]
+            .as_str()
+            .is_some_and(|e| e.contains("sink delivery failure")),
+        "{warning}"
+    );
+}

crates/rsigma-cli/tests/common/mod.rs

@@ -4,6 +4,7 @@
 use std::io::{BufRead, BufReader, Write};
 use std::net::TcpStream;
 use std::process::{Command as StdCommand, Stdio};
+use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 
 use assert_cmd::Command;
@@ -101,6 +102,7 @@ impl Drop for ChildGuard {
 pub struct DaemonProcess {
     child: std::process::Child,
     api_addr: String,
+    stderr: Arc<Mutex<Vec<String>>>,
 }
 
 impl DaemonProcess {
@@ -147,7 +149,9 @@ impl DaemonProcess {
         }
 
         let stderr = guard.as_child_mut().stderr.take().unwrap();
+        let stderr_lines = Arc::new(Mutex::new(Vec::new()));
         let (tx, rx) = std::sync::mpsc::channel::<StartupEvent>();
+        let collected = stderr_lines.clone();
         std::thread::spawn(move || {
             for line in BufReader::new(stderr).lines() {
                 let Ok(line) = line else { return };
@@ -159,6 +163,7 @@ impl DaemonProcess {
                 if line.contains("Sink started") {
                     let _ = tx.send(StartupEvent::SinkStarted);
                 }
+                collected.lock().unwrap().push(line);
             }
         });
 
@@ -196,6 +201,7 @@ impl DaemonProcess {
                 return Self {
                     child: guard.disarm(),
                     api_addr,
+                    stderr: stderr_lines,
                 };
             }
             if Instant::now() >= deadline {
@@ -272,6 +278,11 @@ impl DaemonProcess {
         &self.api_addr
     }
 
+    /// The daemon's stderr log lines read so far.
+    pub fn stderr_lines(&self) -> Vec<String> {
+        self.stderr.lock().unwrap().clone()
+    }
+
     fn kill(&mut self) {
         terminate_child(&mut self.child, Duration::from_secs(5));
     }