diff --git a/examples/README.md b/examples/README.md index 655cc3f..23b7fb7 100644 --- a/examples/README.md +++ b/examples/README.md @@ -55,13 +55,13 @@ relative to process startup. Restarting the process starts a new poll schedule; polls are not aligned to UTC boundaries. `flush.every` does not start a background timer. It divides time into -Unix-epoch-aligned windows. With the examples' `10m` policy, the windows are: +Unix-epoch-aligned windows. The two example configurations intentionally use +different cadences: -```text -12:00:00–12:10:00 UTC -12:10:00–12:20:00 UTC -12:20:00–12:30:00 UTC -``` +| Example | Flush window | Air temperature | Rainfall | PM2.5 | +|---|---:|---:|---:|---:| +| Dataset repository | 10 minutes | 1 minute | 5 minutes | 1 hour | +| Storage bucket | 15 minutes | 5 minutes | 5 minutes | 15 minutes | At startup and before every poll, the pipeline calls `Sink::advance()`. That check uploads any closed windows before collection begins, including when the @@ -69,32 +69,32 @@ following collection fails or returns no records. A UTC boundary closes a window, but the boundary itself does not run `advance()`; delivery waits for the next pipeline check. -This distinction matters when the poll interval is longer than the flush -window. Both example configurations poll PM2.5 hourly while using 10-minute -windows. If the process starts at `12:03`: +The dataset example demonstrates a poll interval longer than its flush window. +If it starts at `12:03`: ```text -12:03 Advance, poll PM2.5, and append fresh records to the 12:00 window. -12:10 The 12:00 window closes. No pipeline task runs at this boundary. -13:03 Advance uploads the 12:00 window, then the next poll fills 13:00. -14:03 Advance uploads the 13:00 window, then polling continues. +12:03 Poll PM2.5 and append fresh records to the 12:00–12:10 window. +12:10 The window closes, but no pipeline task runs at this boundary. +13:03 Advance uploads 12:00, then the next poll fills the 13:00 window. ``` -PM2.5 therefore normally produces one occupied window per successful fresh -poll and uploads it at the next hourly check. It does not create empty files -for the intervening 10-minute windows, and the schedule does not become -`1h10m`. +It normally produces one occupied PM2.5 window per successful fresh poll and +uploads it at the next hourly check. Empty intervening windows do not produce +files, and the schedule does not become `1h10m`. -The faster collectors check for closed windows more often: +The bucket example polls PM2.5 every 15 minutes into 15-minute windows. If it +also starts at `12:03`: -| Collector | Poll interval | Behavior with `flush.every = "10m"` | -|---|---:|---| -| Air temperature | 1 minute | Polls share each window; delivery usually follows within about one minute. | -| Rainfall | 5 minutes | Polls share each window; delivery usually follows within about five minutes. | -| PM2.5 | 1 hour | One fresh poll usually occupies a window; delivery waits for the next hourly check. | +```text +12:03 Poll PM2.5 and append fresh records to the 12:00–12:15 window. +12:15 The window closes, but no pipeline task runs at this boundary. +12:18 Advance uploads 12:00, then the next poll fills the 12:15 window. +``` -`flush.max_records` is a separate safety valve. Reaching it sends the current -records before the wall-clock window closes. +The air-temperature and rainfall collectors check for closed windows more +often because their poll intervals are shorter. `flush.max_records` is a +separate safety valve; reaching it sends the current records before the +wall-clock window closes. ## Restarts and shutdown @@ -157,8 +157,8 @@ records. ## Configuration reference -The flush policy is shared by all collectors, while each collector has its own -poll interval: +The dataset repository configuration uses a shared 10-minute flush policy and +per-collector poll intervals: ```toml spool_dir = "/var/lib/meathook/spool" diff --git a/examples/meathook_bucket.toml b/examples/meathook_bucket.toml index b1a0534..d047a6b 100644 --- a/examples/meathook_bucket.toml +++ b/examples/meathook_bucket.toml @@ -12,17 +12,17 @@ spool_dir = "./spool-test-bucket" # PVC mount on k8s; separate from the # run side by side [flush] # FlushPolicy for each pipeline's durable JSONL tier -every = "10m" +every = "15m" max_records = 50_000 [sink.bucket] id = "zeon256/meathook-test" [collectors.air_temperature] -interval = "1m" +interval = "5m" [collectors.rainfall] interval = "5m" [collectors.pm25] -interval = "1h" +interval = "15m" diff --git a/src/pipeline.rs b/src/pipeline.rs index c93651e..1e86a20 100644 --- a/src/pipeline.rs +++ b/src/pipeline.rs @@ -35,6 +35,19 @@ pub enum ShutdownPolicy { PreserveActiveWindow, } +trait WallClock { + fn now_utc(&self) -> OffsetDateTime; +} + +#[derive(Debug, Clone, Copy)] +struct SystemClock; + +impl WallClock for SystemClock { + fn now_utc(&self) -> OffsetDateTime { + OffsetDateTime::now_utc() + } +} + /// A collector polled on `poll_interval`, feeding a sink stack. /// /// Consecutive polls of "latest reading" APIs return repeats, so an optional @@ -114,13 +127,20 @@ where /// /// Collector and sink errors are logged, never fatal: the loop keeps /// ticking and durable layers retry on their own cadence. - pub async fn run(mut self, cancel: CancellationToken) { + pub async fn run(self, cancel: CancellationToken) { + self.run_with_clock(cancel, SystemClock).await; + } + + async fn run_with_clock(mut self, cancel: CancellationToken, clock: W) + where + W: WallClock, + { let name = self.collector.name().to_owned(); info!(pipeline = %name, interval = ?self.poll_interval, "pipeline starting"); // Recover closed durable windows without splitting the wall-clock // window which is still active at startup. - if let Err(error) = self.sink.advance(OffsetDateTime::now_utc()).await { + if let Err(error) = self.sink.advance(clock.now_utc()).await { warn!(pipeline = %name, %error, "startup window advancement failed"); } @@ -130,7 +150,7 @@ where loop { tokio::select! { () = cancel.cancelled() => break, - _instant = interval.tick() => self.tick(&name).await, + _instant = interval.tick() => self.tick(&name, &clock).await, } } @@ -141,7 +161,7 @@ where } ShutdownPolicy::PreserveActiveWindow => { info!(pipeline = %name, "pipeline shutting down; preserving active wall-clock window"); - self.sink.advance(OffsetDateTime::now_utc()).await + self.sink.advance(clock.now_utc()).await } }; if let Err(error) = result { @@ -149,8 +169,8 @@ where } } - async fn tick(&mut self, name: &str) { - let start = OffsetDateTime::now_utc(); + async fn tick(&mut self, name: &str, clock: &impl WallClock) { + let start = clock.now_utc(); // This heartbeat is independent of collection success or batch // size, so an idle collector still closes elapsed windows. if let Err(error) = self.sink.advance(start).await { @@ -174,7 +194,7 @@ where let meta = WindowMeta { pipeline: name.to_owned(), start, - end: OffsetDateTime::now_utc(), + end: clock.now_utc(), }; if let Err(error) = self.sink.ingest(&meta, records).await { warn!(pipeline = %name, %error, "sink ingest failed"); @@ -204,13 +224,128 @@ where #[cfg(test)] mod tests { + use std::collections::VecDeque; use std::convert::Infallible; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; + use ::time::macros::datetime; + use parking_lot::Mutex; + use serde::{Deserialize, Serialize}; + use tokio::sync::Notify; + use super::*; - use crate::test_util::SharedSink; + use crate::test_util::{FakeObjectSink, SharedSink}; + use crate::{FlushPolicy, JsonlStore, Tier}; + + #[derive(Clone)] + struct ManualClock(Arc>); + + impl ManualClock { + fn at(now: OffsetDateTime) -> Self { + Self(Arc::new(Mutex::new(now))) + } + + fn set(&self, now: OffsetDateTime) { + *self.0.lock() = now; + } + } + + impl WallClock for ManualClock { + fn now_utc(&self) -> OffsetDateTime { + *self.0.lock() + } + } + + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] + struct Pm25Reading { + region: String, + timestamp: String, + value: f64, + } + + enum Pm25Poll { + Records(Vec), + Empty, + Error, + } + + #[derive(Clone, Default)] + struct PollProgress { + completed: Arc, + changed: Arc, + } + + impl PollProgress { + async fn wait_for(&self, target: usize) { + loop { + if self.completed.load(Ordering::SeqCst) >= target { + return; + } + let changed = self.changed.notified(); + if self.completed.load(Ordering::SeqCst) >= target { + return; + } + changed.await; + } + } + } + + struct ScriptedPm25 { + polls: VecDeque, + progress: PollProgress, + } + + impl ScriptedPm25 { + fn new(polls: impl IntoIterator) -> Self { + Self { + polls: polls.into_iter().collect(), + progress: PollProgress::default(), + } + } + + fn progress(&self) -> PollProgress { + self.progress.clone() + } + } + + #[derive(Debug, thiserror::Error)] + #[error("scripted PM2.5 provider failure")] + struct ScriptedPm25Error; + + impl Collector for ScriptedPm25 { + type Record = Pm25Reading; + type Error = ScriptedPm25Error; + fn name(&self) -> &'static str { + "pm25" + } + + fn collect( + &mut self, + ) -> impl Future, ScriptedPm25Error>> + Send { + let result = match self.polls.pop_front().unwrap_or(Pm25Poll::Empty) { + Pm25Poll::Records(records) => Ok(records), + Pm25Poll::Empty => Ok(vec![]), + Pm25Poll::Error => Err(ScriptedPm25Error), + }; + self.progress.completed.fetch_add(1, Ordering::SeqCst); + self.progress.changed.notify_one(); + std::future::ready(result) + } + } + + fn pm25_rows(timestamp: &str, base: f64) -> Vec { + ["east", "west", "north", "south", "central"] + .into_iter() + .enumerate() + .map(|(offset, region)| Pm25Reading { + region: region.to_owned(), + timestamp: timestamp.to_owned(), + value: base + offset as f64, + }) + .collect() + } /// Emits `(tick, i)` pairs, overlapping the previous tick's batch to /// exercise dedup: tick n emits keys n and n+1. struct FakeCollector { @@ -348,4 +483,168 @@ mod tests { assert_eq!(sink.advances(), calls.load(Ordering::SeqCst) + 2); assert!(!sink.flushed()); } + #[tokio::test(start_paused = true)] + async fn configured_pm25_poll_uploads_previous_fifteen_minute_window() { + let spool = tempfile::tempdir().unwrap(); + let spool_dir = spool.path().join("pm25"); + let remote = FakeObjectSink::default(); + let first = pm25_rows("2026-08-25T12:00:00Z", 10.0); + let second = pm25_rows("2026-08-25T12:15:00Z", 20.0); + let collector = ScriptedPm25::new([ + Pm25Poll::Records(first.clone()), + Pm25Poll::Records(second.clone()), + ]); + let progress = collector.progress(); + let stack = Tier::new( + JsonlStore::new(&spool_dir), + FlushPolicy::every(Duration::from_secs(900)), + remote.clone(), + ); + let pipeline = Pipeline::new(collector, stack, Duration::from_secs(900)) + .with_shutdown_policy(ShutdownPolicy::PreserveActiveWindow) + .with_key_fn(|row: &Pm25Reading| (row.region.clone(), row.timestamp.clone())); + let clock = ManualClock::at(datetime!(2026-08-25 12:03 UTC)); + let cancel = CancellationToken::new(); + let handle = tokio::spawn(pipeline.run_with_clock(cancel.clone(), clock.clone())); + + progress.wait_for(1).await; + assert!(remote.objects().is_empty()); + + clock.set(datetime!(2026-08-25 12:18 UTC)); + time::advance(Duration::from_secs(900)).await; + progress.wait_for(2).await; + + cancel.cancel(); + handle.await.unwrap(); + + let objects = remote.objects(); + assert_eq!(objects.len(), 1); + let (path, content) = objects.iter().next().unwrap(); + assert!(path.starts_with("data/pm25/2026-08-25/12-00-00-")); + assert_eq!( + serde_json::from_slice::>(content).unwrap(), + first + ); + + let segments = std::fs::read_dir(spool_dir) + .unwrap() + .collect::, _>>() + .unwrap(); + assert_eq!(segments.len(), 1); + let active = std::fs::read_to_string(segments[0].path()).unwrap(); + let active = active + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .collect::>(); + assert_eq!(active, second); + } + + #[tokio::test(start_paused = true)] + async fn provider_error_still_advances_closed_pm25_window() { + let spool = tempfile::tempdir().unwrap(); + let spool_dir = spool.path().join("pm25"); + let remote = FakeObjectSink::default(); + let collector = ScriptedPm25::new([ + Pm25Poll::Records(pm25_rows("2026-08-25T12:00:00Z", 10.0)), + Pm25Poll::Error, + ]); + let progress = collector.progress(); + let stack = Tier::new( + JsonlStore::new(&spool_dir), + FlushPolicy::every(Duration::from_secs(900)), + remote.clone(), + ); + let pipeline = Pipeline::new(collector, stack, Duration::from_secs(900)) + .with_shutdown_policy(ShutdownPolicy::PreserveActiveWindow) + .with_key_fn(|row: &Pm25Reading| (row.region.clone(), row.timestamp.clone())); + let clock = ManualClock::at(datetime!(2026-08-25 12:03 UTC)); + let cancel = CancellationToken::new(); + let handle = tokio::spawn(pipeline.run_with_clock(cancel.clone(), clock.clone())); + + progress.wait_for(1).await; + clock.set(datetime!(2026-08-25 12:18 UTC)); + time::advance(Duration::from_secs(900)).await; + progress.wait_for(2).await; + + cancel.cancel(); + handle.await.unwrap(); + + let objects = remote.objects(); + assert_eq!(objects.len(), 1); + assert!( + objects + .keys() + .next() + .unwrap() + .starts_with("data/pm25/2026-08-25/12-00-00-") + ); + assert_eq!(std::fs::read_dir(spool_dir).unwrap().count(), 0); + } + + #[tokio::test] + async fn repeated_pm25_snapshot_is_deduplicated_until_timestamp_changes() { + let first = pm25_rows("2026-08-25T12:00:00Z", 10.0); + let second = pm25_rows("2026-08-25T12:15:00Z", 20.0); + let collector = ScriptedPm25::new([ + Pm25Poll::Records(first.clone()), + Pm25Poll::Records(first.clone()), + Pm25Poll::Records(second.clone()), + ]); + let sink = SharedSink::new(); + let mut pipeline = Pipeline::new(collector, sink.clone(), Duration::from_secs(900)) + .with_key_fn(|row: &Pm25Reading| (row.region.clone(), row.timestamp.clone())); + let clock = ManualClock::at(datetime!(2026-08-25 12:03 UTC)); + + pipeline.tick("pm25", &clock).await; + clock.set(datetime!(2026-08-25 12:04 UTC)); + pipeline.tick("pm25", &clock).await; + clock.set(datetime!(2026-08-25 12:18 UTC)); + pipeline.tick("pm25", &clock).await; + + let mut expected = first; + expected.extend(second); + assert_eq!(sink.records(), expected); + } + + #[tokio::test] + async fn restart_resets_pm25_dedupe_and_can_repeat_source_rows() { + let spool = tempfile::tempdir().unwrap(); + let spool_dir = spool.path().join("pm25"); + let remote = FakeObjectSink::default(); + let repeated = pm25_rows("2026-08-25T12:00:00Z", 10.0); + let clock = ManualClock::at(datetime!(2026-08-25 12:03 UTC)); + + { + let stack = Tier::new( + JsonlStore::new(&spool_dir), + FlushPolicy::every(Duration::from_secs(900)), + remote.clone(), + ); + let collector = ScriptedPm25::new([Pm25Poll::Records(repeated.clone())]); + let mut pipeline = Pipeline::new(collector, stack, Duration::from_secs(900)) + .with_key_fn(|row: &Pm25Reading| (row.region.clone(), row.timestamp.clone())); + pipeline.tick("pm25", &clock).await; + } + + let stack = Tier::new( + JsonlStore::new(&spool_dir), + FlushPolicy::every(Duration::from_secs(900)), + remote.clone(), + ); + let collector = ScriptedPm25::new([Pm25Poll::Records(repeated), Pm25Poll::Empty]); + let mut restarted = Pipeline::new(collector, stack, Duration::from_secs(900)) + .with_key_fn(|row: &Pm25Reading| (row.region.clone(), row.timestamp.clone())); + + clock.set(datetime!(2026-08-25 12:04 UTC)); + restarted.tick("pm25", &clock).await; + clock.set(datetime!(2026-08-25 12:18 UTC)); + restarted.tick("pm25", &clock).await; + + let objects = remote.objects(); + assert_eq!(objects.len(), 1); + let rows = + serde_json::from_slice::>(objects.values().next().unwrap()).unwrap(); + assert_eq!(rows.len(), 10); + assert_eq!(&rows[..5], &rows[5..]); + } } diff --git a/src/store/jsonl.rs b/src/store/jsonl.rs index da803da..f9d8691 100644 --- a/src/store/jsonl.rs +++ b/src/store/jsonl.rs @@ -295,7 +295,7 @@ mod tests { use super::*; use crate::layer::{FlushPolicy, Tier}; use crate::sink::Sink; - use crate::test_util::{SharedSink, meta_at}; + use crate::test_util::{FakeObjectSink, SharedSink, meta_at}; use time::OffsetDateTime; fn policy() -> FlushPolicy { @@ -555,6 +555,51 @@ mod tests { assert_eq!(fs::read_dir(&store_dir).unwrap().count(), 0); } + #[tokio::test] + async fn acknowledgement_loss_replays_the_same_remote_object_after_restart() { + let dir = tempfile::tempdir().unwrap(); + let store_dir = dir.path().join("pm25"); + let remote = FakeObjectSink::default(); + let fifteen_minutes = FlushPolicy::every(Duration::from_secs(900)); + + remote.fail_after_put_once(); + { + let mut first = Tier::new(JsonlStore::new(&store_dir), fifteen_minutes, remote.clone()); + first + .ingest(&meta_at("pm25", 43_380), vec![1, 2]) + .await + .unwrap(); + assert!(remote.objects().is_empty()); + + assert!( + first + .advance(OffsetDateTime::from_unix_timestamp(46_980).unwrap()) + .await + .is_err() + ); + assert_eq!(remote.objects().len(), 1); + assert!(store_dir.join("43200.jsonl").exists()); + } + + { + let mut restarted = Tier::new( + JsonlStore::::new(&store_dir), + fifteen_minutes, + remote.clone(), + ); + restarted + .advance(OffsetDateTime::from_unix_timestamp(50_580).unwrap()) + .await + .unwrap(); + } + + let attempts = remote.attempts(); + assert_eq!(attempts.len(), 2); + assert_eq!(attempts[0], attempts[1]); + assert_eq!(remote.objects().len(), 1); + assert!(!store_dir.join("43200.jsonl").exists()); + } + #[tokio::test] async fn append_is_write_ahead_on_disk() { let dir = tempfile::tempdir().unwrap(); diff --git a/src/test_util.rs b/src/test_util.rs index a39ff5e..0593c72 100644 --- a/src/test_util.rs +++ b/src/test_util.rs @@ -1,18 +1,92 @@ //! Shared fakes for unit tests: a `Vec`-backed sink with a failure toggle. use parking_lot::Mutex; +use serde::Serialize; +use serde::de::DeserializeOwned; +use std::collections::BTreeMap; use std::sync::Arc; use std::sync::atomic::AtomicUsize; use std::sync::atomic::{AtomicBool, Ordering}; use time::OffsetDateTime; -use crate::sink::{Sink, WindowMeta}; +use crate::encode::{Encoder, JsonEncoder}; +use crate::sink::{Sink, WindowMeta, object_path}; #[derive(Debug, thiserror::Error)] #[error("test sink failure")] pub struct TestSinkFailure; +#[derive(Debug, thiserror::Error)] +pub enum FakeObjectError { + #[error("failed to encode test object: {0}")] + Encode(#[from] serde_json::Error), + #[error("test object was written but its acknowledgement was lost")] + AcknowledgementLost, +} + +/// Deterministic object-storage sink for tests. +/// +/// Objects are keyed by the same window-and-content path as remote sinks. +/// `fail_after_put_once` models a remote write whose acknowledgement is lost: +/// the object remains visible while the upstream tier retains its segment. +#[derive(Clone, Default)] +pub struct FakeObjectSink { + objects: Arc>>>, + attempts: Arc>>, + fail_after_put: Arc, +} + +impl FakeObjectSink { + pub fn fail_after_put_once(&self) { + self.fail_after_put.store(true, Ordering::SeqCst); + } + + #[must_use] + pub fn objects(&self) -> BTreeMap> { + self.objects.lock().clone() + } + + #[must_use] + pub fn attempts(&self) -> Vec { + self.attempts.lock().clone() + } +} + +impl Sink for FakeObjectSink +where + R: Serialize + DeserializeOwned + Send + 'static, +{ + type Error = FakeObjectError; + + fn ingest( + &mut self, + meta: &WindowMeta, + records: Vec, + ) -> impl Future> + Send { + let result = (|| { + if records.is_empty() { + return Ok(()); + } + + let content = JsonEncoder.encode(&records)?; + let path = object_path(meta, &content, JsonEncoder::EXT); + self.attempts.lock().push(path.clone()); + self.objects.lock().insert(path, content); + + if self.fail_after_put.swap(false, Ordering::SeqCst) { + return Err(FakeObjectError::AcknowledgementLost); + } + Ok(()) + })(); + std::future::ready(result) + } + + fn flush(&mut self) -> impl Future> + Send { + std::future::ready(Ok(())) + } +} + type Batches = Arc)>>>; /// Clonable terminal sink recording every batch it accepts.