cellos_sink_dlq/

lib.rs

//! Dead Letter Queue (DLQ) [`EventSink`] wrapper — P3-03.
//!
//! Wraps any `Arc<dyn EventSink>` and, on a primary `emit()` failure, persists
//! the failed [`CloudEventV1`](cellos_core::CloudEventV1) as a JSON line to a
//! file under an operator-nominated directory. The wrapper is a no-op unless
//! the `CELLOS_DLQ_DIR` environment variable is set to a non-empty,
//! existing-and-writable directory at composition time.
//!
//! # Wire format
//!
//! One file is written per failed event. The file name is:
//!
//! ```text
//! <timestamp-millis>-<event-id-sanitized>.jsonl
//! ```
//!
//! where `<timestamp-millis>` is the Unix epoch milliseconds at the moment of
//! the failure (monotonically increasing within a single process barring clock
//! skew) and `<event-id-sanitized>` replaces any character outside
//! `[A-Za-z0-9._-]` with `_` so attacker-controlled or unusual CloudEvent IDs
//! cannot escape the DLQ directory or collide with shell globs.
//!
//! Each file contains exactly one JSON object on a single line:
//!
//! ```json
//! {"event": <CloudEventV1>, "error": "<primary sink error message>"}
//! ```
//!
//! Using one file per event (rather than a rolling JSONL) keeps writes append-
//! free and tolerates concurrent emitters without locking, at the cost of one
//! `inode` per failure. Operators expecting high failure rates should rotate
//! or compact `${CELLOS_DLQ_DIR}` out of band.
//!
//! # Failure semantics
//!
//! On primary `emit()` failure, the wrapper:
//!
//! 1. Best-effort writes the failed event to the DLQ.
//! 2. Logs a warning with the event id and primary error (and DLQ-write error
//!    if the persistence itself failed).
//! 3. Returns `Ok(())` to the caller — the DLQ has assumed responsibility for
//!    the event, so callers should treat the emit as "delivered to the
//!    operator's recovery channel". If the DLQ write *also* fails, the
//!    original primary error is propagated unchanged so the supervisor still
//!    sees a failure rather than silently dropping the event.
//!
//! # Activation
//!
//! ```text
//! CELLOS_DLQ_DIR=/var/lib/cellos/dlq
//! ```
//!
//! Disabled by default. When unset, empty, or pointing at a path that does
//! not exist or is not writable, [`DlqSink::from_env`] returns the inner
//! sink unwrapped (identity).

use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};

use async_trait::async_trait;
use cellos_core::ports::EventSink;
use cellos_core::{CellosError, CloudEventV1};
use tokio::io::AsyncWriteExt;

/// EventSink wrapper that persists failed CloudEvents to a DLQ directory.
pub struct DlqSink {
    inner: Arc<dyn EventSink>,
    dir: PathBuf,
}

impl DlqSink {
    /// Wrap `inner` with a DLQ persisting to `dir`. The caller is responsible
    /// for ensuring `dir` exists and is writable; [`DlqSink::from_env`]
    /// performs that pre-flight check before invoking this constructor.
    pub fn wrap(inner: Arc<dyn EventSink>, dir: impl Into<PathBuf>) -> Arc<dyn EventSink> {
        Arc::new(Self {
            inner,
            dir: dir.into(),
        })
    }

    /// Read `CELLOS_DLQ_DIR` from the environment. Returns the inner sink
    /// unwrapped (identity) when:
    ///
    /// - the variable is unset, or
    /// - the variable is empty / whitespace-only, or
    /// - the path does not exist, is not a directory, or is not writable.
    ///
    /// Otherwise returns `inner` wrapped in a [`DlqSink`].
    pub fn from_env(inner: Arc<dyn EventSink>) -> Arc<dyn EventSink> {
        let Ok(raw) = std::env::var("CELLOS_DLQ_DIR") else {
            return inner;
        };
        let trimmed = raw.trim();
        if trimmed.is_empty() {
            return inner;
        }
        let dir = PathBuf::from(trimmed);
        match dir_is_writable(&dir) {
            Ok(()) => {
                tracing::info!(
                    target: "cellos.supervisor.observability",
                    dlq_dir = %dir.display(),
                    "DLQ enabled — failed events will be persisted"
                );
                Self::wrap(inner, dir)
            }
            Err(e) => {
                tracing::warn!(
                    target: "cellos.supervisor.observability",
                    dlq_dir = %dir.display(),
                    error = %e,
                    "CELLOS_DLQ_DIR set but directory is not usable; DLQ disabled"
                );
                inner
            }
        }
    }

    async fn write_to_dlq(
        &self,
        event: &CloudEventV1,
        primary_error: &CellosError,
    ) -> Result<(), CellosError> {
        let filename = dlq_filename(&event.id);
        let path = self.dir.join(filename);

        let envelope = serde_json::json!({
            "event": event,
            "error": primary_error.to_string(),
        });
        let line = serde_json::to_string(&envelope)
            .map_err(|e| CellosError::EventSink(format!("dlq serialize: {e}")))?;

        // O_CREATE_NEW + O_WRONLY (no O_TRUNC) — one file per failure means
        // collisions on the timestamp-millis + event-id pair are vanishingly
        // unlikely, but if they do happen we want the write to fail loudly
        // rather than silently truncate a prior DLQ entry.
        let mut f = tokio::fs::OpenOptions::new()
            .create_new(true)
            .write(true)
            .open(&path)
            .await
            .map_err(|e| CellosError::EventSink(format!("dlq open {}: {e}", path.display())))?;
        f.write_all(line.as_bytes())
            .await
            .map_err(|e| CellosError::EventSink(format!("dlq write: {e}")))?;
        f.write_all(b"\n")
            .await
            .map_err(|e| CellosError::EventSink(format!("dlq newline: {e}")))?;
        f.flush()
            .await
            .map_err(|e| CellosError::EventSink(format!("dlq flush: {e}")))?;
        Ok(())
    }
}

#[async_trait]
impl EventSink for DlqSink {
    async fn emit(&self, event: &CloudEventV1) -> Result<(), CellosError> {
        match self.inner.emit(event).await {
            Ok(()) => Ok(()),
            Err(primary_err) => match self.write_to_dlq(event, &primary_err).await {
                Ok(()) => {
                    tracing::warn!(
                        target: "cellos.supervisor.observability",
                        event_id = %event.id,
                        event_type = %event.ty,
                        primary_error = %primary_err,
                        "primary event sink failed; event captured to DLQ"
                    );
                    // DLQ has taken custody of the event — caller should
                    // treat the emit as delivered.
                    Ok(())
                }
                Err(dlq_err) => {
                    tracing::error!(
                        target: "cellos.supervisor.observability",
                        event_id = %event.id,
                        event_type = %event.ty,
                        primary_error = %primary_err,
                        dlq_error = %dlq_err,
                        "primary event sink failed AND DLQ persistence failed; event lost"
                    );
                    // Both paths failed — surface the original primary error
                    // so the supervisor still sees a sink failure.
                    Err(primary_err)
                }
            },
        }
    }
}

/// Build the DLQ filename for an event id: `<unix-millis>-<sanitized-id>.jsonl`.
fn dlq_filename(event_id: &str) -> String {
    let millis = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_millis())
        .unwrap_or(0);
    format!("{millis}-{}.jsonl", sanitize_id(event_id))
}

/// Replace any character outside `[A-Za-z0-9._-]` with `_`.
///
/// Empty input collapses to `"_"` so the filename never ends with a stray
/// trailing hyphen from the timestamp separator.
fn sanitize_id(id: &str) -> String {
    if id.is_empty() {
        return "_".into();
    }
    id.chars()
        .map(|c| {
            if c.is_ascii_alphanumeric() || c == '.' || c == '_' || c == '-' {
                c
            } else {
                '_'
            }
        })
        .collect()
}

/// Pre-flight check: returns Ok if `dir` exists, is a directory, and a
/// throwaway probe file can be created and removed inside it.
fn dir_is_writable(dir: &Path) -> std::io::Result<()> {
    let meta = std::fs::metadata(dir)?;
    if !meta.is_dir() {
        return Err(std::io::Error::new(
            std::io::ErrorKind::NotADirectory,
            "CELLOS_DLQ_DIR is not a directory",
        ));
    }
    // Probe with O_CREATE_NEW + immediate unlink. We use a per-process probe
    // name to avoid colliding with a parallel supervisor doing the same
    // check.
    let probe = dir.join(format!(".cellos-dlq-probe-{}", std::process::id()));
    {
        let _f = std::fs::OpenOptions::new()
            .create_new(true)
            .write(true)
            .open(&probe)?;
    }
    let _ = std::fs::remove_file(&probe);
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Mutex;

    /// A sink that always succeeds and records the last event seen.
    struct CaptureSink(Mutex<Option<CloudEventV1>>);

    impl CaptureSink {
        fn new() -> Arc<Self> {
            Arc::new(Self(Mutex::new(None)))
        }
        fn last(&self) -> Option<CloudEventV1> {
            self.0.lock().unwrap().clone()
        }
    }

    #[async_trait]
    impl EventSink for CaptureSink {
        async fn emit(&self, event: &CloudEventV1) -> Result<(), CellosError> {
            *self.0.lock().unwrap() = Some(event.clone());
            Ok(())
        }
    }

    /// A sink that always fails with a fixed error message.
    struct FailingSink;

    #[async_trait]
    impl EventSink for FailingSink {
        async fn emit(&self, _event: &CloudEventV1) -> Result<(), CellosError> {
            Err(CellosError::EventSink("primary boom".into()))
        }
    }

    fn test_event(id: &str) -> CloudEventV1 {
        CloudEventV1 {
            specversion: "1.0".into(),
            id: id.into(),
            source: "test".into(),
            ty: "test.event".into(),
            datacontenttype: Some("application/json".into()),
            data: Some(serde_json::json!({ "k": "v" })),
            time: None,
            traceparent: None,
        }
    }

    #[tokio::test]
    async fn success_path_passes_through_to_primary() {
        // (a) When the primary emit succeeds, DlqSink must forward and write
        // nothing to the DLQ directory.
        let tmp = tempfile::tempdir().unwrap();
        let capture = CaptureSink::new();
        let sink = DlqSink::wrap(capture.clone() as Arc<dyn EventSink>, tmp.path());

        let event = test_event("evt-success");
        sink.emit(&event).await.unwrap();

        assert_eq!(
            capture.last().expect("primary saw the event").id,
            "evt-success",
        );

        let entries: Vec<_> = std::fs::read_dir(tmp.path())
            .unwrap()
            .filter_map(Result::ok)
            .collect();
        assert!(
            entries.is_empty(),
            "DLQ dir must be empty on success path, found: {:?}",
            entries.iter().map(|e| e.path()).collect::<Vec<_>>()
        );
    }

    #[tokio::test]
    async fn primary_error_lands_in_dlq_file() {
        // (b) On primary failure the failed CloudEvent must appear as a JSON
        // line under the DLQ directory and the call must return Ok (DLQ
        // assumes custody).
        let tmp = tempfile::tempdir().unwrap();
        let sink = DlqSink::wrap(Arc::new(FailingSink) as Arc<dyn EventSink>, tmp.path());

        let event = test_event("evt-failure-1");
        sink.emit(&event)
            .await
            .expect("DLQ should swallow primary error");

        let entries: Vec<_> = std::fs::read_dir(tmp.path())
            .unwrap()
            .filter_map(Result::ok)
            .collect();
        assert_eq!(entries.len(), 1, "exactly one DLQ file expected");

        let path = entries[0].path();
        let name = path.file_name().unwrap().to_string_lossy().into_owned();
        assert!(
            name.ends_with("-evt-failure-1.jsonl"),
            "filename should encode the event id: {name}"
        );

        let body = std::fs::read_to_string(&path).unwrap();
        let parsed: serde_json::Value = serde_json::from_str(body.trim_end())
            .expect("DLQ file must contain a single JSON line");
        assert_eq!(parsed["event"]["id"], "evt-failure-1");
        assert_eq!(parsed["event"]["type"], "test.event");
        assert_eq!(parsed["error"], "event sink: primary boom");
    }

    #[tokio::test]
    async fn disabled_mode_is_identity() {
        // (c) With CELLOS_DLQ_DIR unset, from_env must return the inner sink
        // unmodified — no DLQ wrapping, no behavioural change.
        // We verify by Arc-pointer-equality: the returned sink must be the
        // same Arc instance we passed in.
        // Guard the env mutation against test parallelism on the same var.
        static ENV_GUARD: Mutex<()> = Mutex::new(());
        let _g = ENV_GUARD.lock().unwrap();
        std::env::remove_var("CELLOS_DLQ_DIR");

        let inner: Arc<dyn EventSink> = CaptureSink::new();
        let returned = DlqSink::from_env(inner.clone());

        assert!(
            Arc::ptr_eq(&inner, &returned),
            "from_env with CELLOS_DLQ_DIR unset must return the inner Arc unchanged"
        );
    }

    #[tokio::test]
    async fn from_env_empty_value_is_identity() {
        static ENV_GUARD: Mutex<()> = Mutex::new(());
        let _g = ENV_GUARD.lock().unwrap();
        std::env::set_var("CELLOS_DLQ_DIR", "   ");

        let inner: Arc<dyn EventSink> = CaptureSink::new();
        let returned = DlqSink::from_env(inner.clone());
        std::env::remove_var("CELLOS_DLQ_DIR");

        assert!(
            Arc::ptr_eq(&inner, &returned),
            "whitespace-only CELLOS_DLQ_DIR must be treated as unset"
        );
    }

    #[tokio::test]
    async fn from_env_nonexistent_dir_is_identity() {
        static ENV_GUARD: Mutex<()> = Mutex::new(());
        let _g = ENV_GUARD.lock().unwrap();
        let bogus =
            std::env::temp_dir().join(format!("cellos-dlq-does-not-exist-{}", std::process::id()));
        // Make sure it really doesn't exist.
        let _ = std::fs::remove_dir_all(&bogus);
        std::env::set_var("CELLOS_DLQ_DIR", &bogus);

        let inner: Arc<dyn EventSink> = CaptureSink::new();
        let returned = DlqSink::from_env(inner.clone());
        std::env::remove_var("CELLOS_DLQ_DIR");

        assert!(
            Arc::ptr_eq(&inner, &returned),
            "non-existent CELLOS_DLQ_DIR must degrade to identity (DLQ disabled)"
        );
    }

    #[tokio::test]
    #[allow(clippy::await_holding_lock)]
    async fn from_env_writable_dir_wraps_inner() {
        static ENV_GUARD: Mutex<()> = Mutex::new(());
        let _g = ENV_GUARD.lock().unwrap();
        let tmp = tempfile::tempdir().unwrap();
        std::env::set_var("CELLOS_DLQ_DIR", tmp.path());

        let inner: Arc<dyn EventSink> = Arc::new(FailingSink);
        let wrapped = DlqSink::from_env(inner.clone());
        std::env::remove_var("CELLOS_DLQ_DIR");

        // Must NOT be the identity Arc — wrapping should have happened.
        assert!(
            !Arc::ptr_eq(&inner, &wrapped),
            "writable CELLOS_DLQ_DIR must wrap the inner sink"
        );

        // And a failing emit should now succeed at the boundary while
        // landing the event in the DLQ.
        wrapped.emit(&test_event("env-wrap")).await.unwrap();
        let count = std::fs::read_dir(tmp.path()).unwrap().count();
        assert_eq!(count, 1, "wrapped sink should have written one DLQ file");
    }

    #[test]
    fn sanitize_id_replaces_unsafe_chars() {
        assert_eq!(sanitize_id("evt.id-123_ok"), "evt.id-123_ok");
        assert_eq!(sanitize_id("../escape"), ".._escape"); // dots allowed, only / replaced
        assert_eq!(sanitize_id("evt/with/slash"), "evt_with_slash");
        assert_eq!(sanitize_id(""), "_");
    }

    #[test]
    fn dlq_filename_has_expected_shape() {
        let name = dlq_filename("abc");
        assert!(name.ends_with("-abc.jsonl"));
        let prefix = name.trim_end_matches("-abc.jsonl");
        assert!(
            prefix.chars().all(|c| c.is_ascii_digit()),
            "timestamp prefix must be all digits: {prefix}"
        );
    }
}