meathook-rs 0.2.0

//! [`JsonlStore`]: durable write-ahead segment files (JSON lines).
//!
//! The **disk is the buffer**: `append` writes records as JSON lines to the
//! window's segment file and fsyncs before returning, so once a tick's
//! ingest returns those records survive `SIGKILL`. `commit` deletes the
//! segment only after the tier's downstream sink accepted it — a failed
//! downstream leaves the segment in place to be retried at the next firing.
//!
//! On-disk layout (one directory per pipeline):
//!
//! ```text
//! {dir}/{window_start_unix}.jsonl
//! ```
//!
//! Segment files are named by the start of their flush window (unix
//! seconds, aligned by the tier), so windows are reconstructed from the
//! filename alone — leftover segments from a crashed run replay with their
//! original window and land at the same storage path (idempotent). Torn
//! final lines (crash mid-append) and corrupt lines are skipped with a
//! warning on read.

use std::fs;
use std::io::{self, Write};
use std::marker::PhantomData;
use std::path::{Path, PathBuf};

use serde::Serialize;
use serde::de::DeserializeOwned;
use tracing::warn;

use super::{Segment, Store};

/// Error from a [`JsonlStore`].
#[derive(Debug, thiserror::Error)]
pub enum JsonlStoreError {
    /// Reading or writing a segment file (or the store directory) failed.
    #[error("store I/O error at {path}: {source}")]
    Io {
        /// The file or directory the operation failed on.
        path: PathBuf,
        /// The underlying I/O error.
        #[source]
        source: io::Error,
    },
    /// A record could not be serialized to a JSON line.
    #[error("failed to serialize record for spooling: {0}")]
    Serialize(#[source] serde_json::Error),
}

fn io_err(path: &Path, source: io::Error) -> JsonlStoreError {
    JsonlStoreError::Io {
        path: path.to_owned(),
        source,
    }
}

/// Durable write-ahead store rooted at one directory (one per pipeline).
/// See the [module docs](self) for the on-disk protocol.
///
/// Construction is infallible and does no I/O; the directory is created on
/// first use. The [`Store::pipeline_hint`] is the last component of `dir`
/// (override with [`with_pipeline_name`](Self::with_pipeline_name)), so
/// point each pipeline at `spool_root.join(pipeline_name)`.
///
/// File I/O uses synchronous `std::fs` calls: appends are a few kilobytes
/// plus an fsync, which is acceptable to block the runtime for at the
/// collection rates this crate targets.
pub struct JsonlStore<R> {
    /// Directory holding this store's segment files, one per window.
    dir: PathBuf,
    /// Hint reported via [`Store::pipeline_hint`]: the last component of
    /// `dir` unless overridden with
    /// [`with_pipeline_name`](Self::with_pipeline_name).
    pipeline: String,
    /// Whether `dir` has been created; construction does no I/O, so this
    /// happens lazily on first use.
    initialized: bool,
    /// Ties the store to one record type without owning any (`fn() -> R`
    /// keeps `Send`/`Sync` independent of `R`).
    _record: PhantomData<fn() -> R>,
}

impl<R> JsonlStore<R> {
    /// Create a store rooted at `dir`. The pipeline hint is derived from
    /// the last component of `dir`.
    #[must_use]
    pub fn new(dir: impl Into<PathBuf>) -> Self {
        let dir = dir.into();
        let pipeline = dir.file_name().map_or_else(
            || "unknown".to_owned(),
            |n| n.to_string_lossy().into_owned(),
        );
        Self {
            dir,
            pipeline,
            initialized: false,
            _record: PhantomData,
        }
    }

    /// Override the pipeline hint derived from the directory name.
    #[must_use]
    pub fn with_pipeline_name(mut self, name: impl Into<String>) -> Self {
        self.pipeline = name.into();
        self
    }

    fn ensure_dir(&mut self) -> Result<(), JsonlStoreError> {
        if !self.initialized {
            fs::create_dir_all(&self.dir).map_err(|e| io_err(&self.dir, e))?;
            self.initialized = true;
        }
        Ok(())
    }

    fn segment_path(&self, window: i64) -> PathBuf {
        self.dir.join(format!("{window}.jsonl"))
    }

    /// All segment files in the store directory, oldest first.
    fn list_segments(&self) -> Result<Vec<(i64, PathBuf)>, JsonlStoreError> {
        let mut segments = vec![];
        let entries = fs::read_dir(&self.dir).map_err(|e| io_err(&self.dir, e))?;
        for entry in entries {
            let entry = entry.map_err(|e| io_err(&self.dir, e))?;
            let path = entry.path();
            if path.extension().is_none_or(|ext| ext != "jsonl") {
                continue;
            }
            let Some(start) = path
                .file_stem()
                .and_then(|s| s.to_str())
                .and_then(|s| s.parse::<i64>().ok())
            else {
                warn!(path = %path.display(), "ignoring unrecognized file in store dir");
                continue;
            };
            segments.push((start, path));
        }
        segments.sort_unstable_by_key(|(start, _)| *start);
        Ok(segments)
    }
}

impl<R> Store<R> for JsonlStore<R>
where
    R: Serialize + DeserializeOwned + Send + 'static,
{
    type Error = JsonlStoreError;
    type Segment<'a>
        = JsonlSegment<R>
    where
        Self: 'a;

    /// Append records to the window's segment file, fsyncing the file (and
    /// the directory when the segment is new) before returning.
    async fn append(&mut self, window: i64, records: Vec<R>) -> Result<(), JsonlStoreError> {
        self.ensure_dir()?;
        let path = self.segment_path(window);

        let mut lines = vec![];
        for record in &records {
            serde_json::to_writer(&mut lines, record).map_err(JsonlStoreError::Serialize)?;
            lines.push(b'\n');
        }

        let is_new = !path.exists();
        let mut file = fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&path)
            .map_err(|e| io_err(&path, e))?;
        file.write_all(&lines).map_err(|e| io_err(&path, e))?;
        file.sync_all().map_err(|e| io_err(&path, e))?;
        if is_new {
            fs::File::open(&self.dir)
                .and_then(|d| d.sync_all())
                .map_err(|e| io_err(&self.dir, e))?;
        }
        Ok(())
    }

    async fn oldest(
        &mut self,
        after: Option<i64>,
    ) -> Result<Option<JsonlSegment<R>>, JsonlStoreError> {
        self.ensure_dir()?;
        Ok(self
            .list_segments()?
            .into_iter()
            .find(|(window, _)| after.is_none_or(|a| *window > a))
            .map(|(window, path)| JsonlSegment {
                window,
                path,
                _record: PhantomData,
            }))
    }

    fn pipeline_hint(&self) -> Option<&str> {
        Some(&self.pipeline)
    }
}

/// Oldest segment file checked out of a [`JsonlStore`]. Holds only the
/// path; the file itself is the retained copy until [`Segment::commit`]
/// deletes it.
pub struct JsonlSegment<R> {
    /// Window start (unix seconds), parsed from the segment filename.
    window: i64,
    /// The segment file: reads re-open it, [`Segment::commit`] deletes it.
    path: PathBuf,
    /// Ties the segment to its store's record type.
    _record: PhantomData<fn() -> R>,
}

impl<R> Segment<R> for JsonlSegment<R>
where
    R: DeserializeOwned + Send + 'static,
{
    type Error = JsonlStoreError;

    fn window(&self) -> i64 {
        self.window
    }

    async fn records(&mut self) -> Result<Vec<R>, JsonlStoreError> {
        let contents = fs::read_to_string(&self.path).map_err(|e| io_err(&self.path, e))?;
        let lines = contents
            .lines()
            .filter(|l| !l.is_empty())
            .collect::<Vec<&str>>();
        let mut records = Vec::with_capacity(lines.len());
        let last = lines.len().saturating_sub(1);
        for (i, line) in lines.iter().enumerate() {
            match serde_json::from_str::<R>(line) {
                Ok(record) => records.push(record),
                Err(error) if i == last => {
                    warn!(
                        path = %self.path.display(),
                        %error,
                        "skipping torn final line in store segment (crash mid-append)"
                    );
                }
                Err(error) => {
                    warn!(
                        path = %self.path.display(),
                        line = i,
                        %error,
                        "skipping corrupt line in store segment"
                    );
                }
            }
        }
        Ok(records)
    }

    async fn commit(self) -> Result<(), JsonlStoreError> {
        fs::remove_file(&self.path).map_err(|e| io_err(&self.path, e))
    }
}

#[cfg(test)]
mod tests {
    use std::time::Duration;

    use super::*;
    use crate::layer::{FlushPolicy, SinkExt};
    use crate::sink::Sink;
    use crate::test_util::{SharedSink, meta_at};

    fn policy() -> FlushPolicy {
        FlushPolicy::new(Duration::from_secs(3600), usize::MAX)
    }

    #[tokio::test]
    async fn tier_ingest_is_write_ahead() {
        let dir = tempfile::tempdir().unwrap();
        let store_dir = dir.path().join("p");
        let inner = SharedSink::new();
        let mut tier = inner.clone().tier(JsonlStore::new(&store_dir), policy());

        tier.ingest(&meta_at("p", 10), vec![1, 2]).await.unwrap();
        tier.ingest(&meta_at("p", 20), vec![3]).await.unwrap();

        // Records are on disk before any flush fires.
        let files = fs::read_dir(&store_dir).unwrap().collect::<Vec<_>>();
        assert_eq!(files.len(), 1);
        let contents = fs::read_to_string(files[0].as_ref().unwrap().path()).unwrap();
        assert_eq!(contents.lines().count(), 3);
        assert!(inner.batches().is_empty());
    }

    #[tokio::test]
    async fn replays_leftover_segments_on_first_use() {
        let dir = tempfile::tempdir().unwrap();
        let store_dir = dir.path().join("weather");
        fs::create_dir_all(&store_dir).unwrap();
        // Two leftover segments from a "previous run", an hour apart.
        fs::write(store_dir.join("3600.jsonl"), "1\n2\n").unwrap();
        fs::write(store_dir.join("7200.jsonl"), "3\n").unwrap();

        let inner = SharedSink::new();
        let mut tier = inner
            .clone()
            .tier(JsonlStore::<i32>::new(&store_dir), policy());
        tier.flush().await.unwrap();

        let batches = inner.batches();
        assert_eq!(batches.len(), 2);
        // Oldest first, meta reconstructed from the filename; the pipeline
        // name comes from the store's hint (no live meta seen yet).
        assert_eq!(batches[0].0.start.unix_timestamp(), 3600);
        assert_eq!(batches[0].0.pipeline, "weather");
        assert_eq!(batches[0].1, vec![1, 2]);
        assert_eq!(batches[1].0.start.unix_timestamp(), 7200);
        assert_eq!(batches[1].1, vec![3]);
        // Segments are gone after a successful replay.
        assert_eq!(fs::read_dir(&store_dir).unwrap().count(), 0);
    }

    #[tokio::test]
    async fn tier_flush_tolerates_torn_final_line() {
        let dir = tempfile::tempdir().unwrap();
        let store_dir = dir.path().join("p");
        fs::create_dir_all(&store_dir).unwrap();
        fs::write(store_dir.join("3600.jsonl"), "1\n2\n{\"trunc").unwrap();

        let inner = SharedSink::new();
        let mut tier = inner
            .clone()
            .tier(JsonlStore::<i32>::new(&store_dir), policy());
        tier.flush().await.unwrap();

        assert_eq!(inner.batches()[0].1, vec![1, 2]);
        assert_eq!(fs::read_dir(&store_dir).unwrap().count(), 0);
    }

    #[tokio::test]
    async fn tier_retains_segments_across_failing_downstream() {
        let dir = tempfile::tempdir().unwrap();
        let store_dir = dir.path().join("p");
        let inner = SharedSink::new();
        let mut tier = inner.clone().tier(JsonlStore::new(&store_dir), policy());

        tier.ingest(&meta_at("p", 10), vec![1, 2]).await.unwrap();

        inner.set_fail(true);
        assert!(tier.flush().await.is_err());
        assert_eq!(fs::read_dir(&store_dir).unwrap().count(), 1);
        assert!(inner.batches().is_empty());

        inner.set_fail(false);
        tier.flush().await.unwrap();
        assert_eq!(inner.batches()[0].1, vec![1, 2]);
        assert_eq!(fs::read_dir(&store_dir).unwrap().count(), 0);
    }

    #[tokio::test]
    async fn tier_max_records_drains_active_segment() {
        let dir = tempfile::tempdir().unwrap();
        let store_dir = dir.path().join("p");
        let inner = SharedSink::new();
        let mut tier = inner.clone().tier(
            JsonlStore::new(&store_dir),
            FlushPolicy::new(Duration::from_secs(3600), 3),
        );

        tier.ingest(&meta_at("p", 10), vec![1, 2]).await.unwrap();
        assert!(inner.batches().is_empty());
        tier.ingest(&meta_at("p", 20), vec![3]).await.unwrap();

        assert_eq!(inner.batches()[0].1, vec![1, 2, 3]);
        assert_eq!(fs::read_dir(&store_dir).unwrap().count(), 0);
    }

    #[tokio::test]
    async fn append_is_write_ahead_on_disk() {
        let dir = tempfile::tempdir().unwrap();
        let store_dir = dir.path().join("p");
        let mut store = JsonlStore::new(&store_dir);

        store.append(0, vec![1, 2]).await.unwrap();
        store.append(0, vec![3]).await.unwrap();

        let contents = fs::read_to_string(store_dir.join("0.jsonl")).unwrap();
        assert_eq!(contents, "1\n2\n3\n");
    }

    #[tokio::test]
    async fn oldest_is_oldest_window_and_commit_deletes() {
        let dir = tempfile::tempdir().unwrap();
        let store_dir = dir.path().join("p");
        let mut store = JsonlStore::new(&store_dir);
        store.append(7200, vec![3]).await.unwrap();
        store.append(3600, vec![1, 2]).await.unwrap();

        let mut seg = store.oldest(None).await.unwrap().unwrap();
        assert_eq!(seg.window(), 3600);
        assert_eq!(seg.records().await.unwrap(), vec![1, 2]);
        seg.commit().await.unwrap();
        assert!(!store_dir.join("3600.jsonl").exists());

        let seg = store.oldest(None).await.unwrap().unwrap();
        assert_eq!(seg.window(), 7200);
    }

    #[tokio::test]
    async fn oldest_after_skips_windows_at_or_below_cursor() {
        let dir = tempfile::tempdir().unwrap();
        let store_dir = dir.path().join("p");
        let mut store = JsonlStore::new(&store_dir);
        store.append(3600, vec![1]).await.unwrap();
        store.append(7200, vec![2]).await.unwrap();

        let seg = store.oldest(Some(3600)).await.unwrap().unwrap();
        assert_eq!(seg.window(), 7200);
        assert!(store.oldest(Some(7200)).await.unwrap().is_none());

        // Skipped segment files are untouched.
        assert!(store_dir.join("3600.jsonl").exists());
    }

    #[tokio::test]
    async fn ignores_non_jsonl_files() {
        let dir = tempfile::tempdir().unwrap();
        let store_dir = dir.path().join("p");
        fs::create_dir_all(&store_dir).unwrap();
        fs::write(store_dir.join("notes.txt"), "hi").unwrap();
        fs::write(store_dir.join("weird.jsonl"), "1\n").unwrap();
        fs::write(store_dir.join("100.jsonl"), "1\n").unwrap();

        let mut store: JsonlStore<i32> = JsonlStore::new(&store_dir);
        let seg = store.oldest(None).await.unwrap().unwrap();
        assert_eq!(seg.window(), 100);
    }

    #[tokio::test]
    async fn segment_tolerates_torn_final_line() {
        let dir = tempfile::tempdir().unwrap();
        let store_dir = dir.path().join("p");
        fs::create_dir_all(&store_dir).unwrap();
        fs::write(store_dir.join("3600.jsonl"), "1\n2\n{\"trunc").unwrap();

        let mut store: JsonlStore<i32> = JsonlStore::new(&store_dir);
        let mut seg = store.oldest(None).await.unwrap().unwrap();
        assert_eq!(seg.records().await.unwrap(), vec![1, 2]);
    }

    #[test]
    fn pipeline_hint_is_dir_derived_and_overridable() {
        let store: JsonlStore<i32> = JsonlStore::new("/var/spool/weather");
        assert_eq!(Store::<i32>::pipeline_hint(&store), Some("weather"));

        let store = store.with_pipeline_name("rain");
        assert_eq!(Store::<i32>::pipeline_hint(&store), Some("rain"));
    }
}