From 2cf3002e352cc3bd363c65904ef28307c70b5cb0 Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Wed, 17 Jun 2026 19:30:41 +0200 Subject: [PATCH 1/4] =?UTF-8?q?feat(ingester):=20rfc0014=20green=20pt1=20?= =?UTF-8?q?=E2=80=94=20ParquetRecordSink=20+=20flush=20policy?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Build the buffering production write path (RFC 0014): `ParquetRecordSink` implements `RecordSink`, accumulating mined records per partition and flushing each to a Parquet object on the RFC 0013 `Store` seam (`encode_records_to_parquet` + `put_blocking`, UUIDv7-named). Hybrid flush policy per the specified decisions: - Size — the `emit` that crosses `target_bytes` flushes the partition (RFC0014.1). - Age — `flush_aged` (batch-window tick) flushes partitions whose oldest record reached `max_buffer_age`, inclusive (RFC0014.2). - Rotation — `flush_all` force-flushes EVERY partition, incl. sub-threshold (RFC0014.3). - Hard ceiling — `emit` flushes the largest partition inline before it would exceed `ceiling_bytes`, so buffered bytes never exceed it (RFC0014.4). Buffers are keyed by `PartitionKey` (carries tenant_id) → tenant-scoped by construction (RFC0014.6). A flush failure retains the buffer (the WAL is the durability of record), counted for observability. Un-ignores RFC0014.1/.2/.3/.4/.6 (driven against a LocalFileSystem `Store`, read back to prove no loss + tenant isolation). RFC0014.5 (crash recovery) stays `#[ignore]`d — green part 2 wires the sink into the ingest pipeline + extends the RFC 0008 crash harness; RFC stays `red` until then. The sink takes a `Store` (local or S3) so it's S3-ready; the server-wiring that constructs/injects it + greens RFC0013.6 is the follow-on. Co-Authored-By: Claude Opus 4.8 --- crates/ourios-ingester/Cargo.toml | 2 +- crates/ourios-ingester/src/lib.rs | 1 + crates/ourios-ingester/src/record_sink.rs | 278 ++++++++++++++++++ .../tests/rfc0014_ingest_write_path.rs | 247 ++++++++++++++-- 4 files changed, 503 insertions(+), 25 deletions(-) create mode 100644 crates/ourios-ingester/src/record_sink.rs diff --git a/crates/ourios-ingester/Cargo.toml b/crates/ourios-ingester/Cargo.toml index 3a3704d6f..37927167a 100644 --- a/crates/ourios-ingester/Cargo.toml +++ b/crates/ourios-ingester/Cargo.toml @@ -81,7 +81,7 @@ tonic = { version = "0.14", default-features = false, features = ["codegen"] } # Parse the snapshot-recorded WAL high-water segment id back into the # `uuid::Uuid` of `ourios_wal::WalOffset` (RFC 0001 §6.9 / RFC 0008 # §6.6). Same major as ourios-wal's, so the types unify. -uuid = { version = "1", default-features = false, features = ["std"] } +uuid = { version = "1", default-features = false, features = ["std", "v7"] } [dev-dependencies] tempfile = "3" diff --git a/crates/ourios-ingester/src/lib.rs b/crates/ourios-ingester/src/lib.rs index daf3a1f11..7d48b952b 100644 --- a/crates/ourios-ingester/src/lib.rs +++ b/crates/ourios-ingester/src/lib.rs @@ -35,6 +35,7 @@ pub mod compactor; pub mod metrics; pub mod receiver; +pub mod record_sink; pub mod recovery; pub mod snapshot_store; diff --git a/crates/ourios-ingester/src/record_sink.rs b/crates/ourios-ingester/src/record_sink.rs new file mode 100644 index 000000000..f666716c8 --- /dev/null +++ b/crates/ourios-ingester/src/record_sink.rs @@ -0,0 +1,278 @@ +//! Production data write path (RFC 0014): a buffering [`RecordSink`] that +//! accumulates mined records per partition and flushes each to a Parquet +//! object on the RFC 0013 [`Store`] seam. +//! +//! Flush policy (RFC 0014 §3.2, hybrid): a partition flushes when its buffered +//! bytes reach [`FlushConfig::target_bytes`] (size, evaluated on `emit`), when +//! its oldest record reaches [`FlushConfig::max_buffer_age`] (age, evaluated by +//! [`ParquetRecordSink::flush_aged`] on the batch-window tick), or when the WAL +//! segment rotates ([`ParquetRecordSink::flush_all`], force-flushing *every* +//! partition). Total buffered bytes are kept under +//! [`FlushConfig::ceiling_bytes`] by flushing the largest partition inline +//! before `emit` would exceed it (RFC 0014 §3.4 — a hard ceiling). +//! +//! Records reach the sink only after the WAL is durable (`CLAUDE.md` §3.4), so +//! a buffer is a bounded accelerator, never the durability of record: a crash +//! re-mines the un-flushed tail from the WAL. A flush failure therefore retains +//! the buffer (counted, retried on the next trigger) rather than dropping data. +//! Buffers are keyed by [`PartitionKey`], which carries `tenant_id`, so they +//! are tenant-scoped by construction (`CLAUDE.md` §3.7). + +use std::collections::HashMap; +use std::path::Path; +use std::time::{Duration, Instant}; + +use ourios_core::record::{MinedRecord, RecordSink}; +use ourios_parquet::{ + DEFAULT_ZSTD_LEVEL, PartitionKey, Store, StoreError, WriterError, encode_records_to_parquet, +}; +use uuid::Uuid; + +/// Flush-policy knobs (RFC 0014 §3; RFC 0004 config at the call site). +#[derive(Debug, Clone)] +pub struct FlushConfig { + /// Size trigger: a partition flushes once its estimated buffered bytes + /// reach this. Production targets RFC 0005 §3.5's 256 MiB–2 GiB file band; + /// tests use small values. (Tuning is RFC 0014 §7.) + pub target_bytes: usize, + /// Age trigger: a partition flushes once its oldest buffered record's age + /// reaches this (inclusive), bounding low-volume staleness. + pub max_buffer_age: Duration, + /// Hard ceiling on total buffered bytes across all partitions; `emit` + /// flushes inline to stay at or under it. + pub ceiling_bytes: usize, +} + +/// A failed flush of one partition. Non-fatal — the buffer is retained and the +/// WAL remains the durability of record. +#[derive(Debug)] +pub enum FlushError { + /// Encoding the buffered records to Parquet failed. + Encode(WriterError), + /// Writing the encoded object to the store failed. + Store(StoreError), +} + +impl std::fmt::Display for FlushError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Encode(e) => write!(f, "encode buffered records: {e}"), + Self::Store(e) => write!(f, "put Parquet object: {e}"), + } + } +} + +impl std::error::Error for FlushError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Encode(e) => Some(e), + Self::Store(e) => Some(e), + } + } +} + +struct PartitionBuffer { + records: Vec, + est_bytes: usize, + oldest: Instant, +} + +/// The buffering Parquet record sink — the production replacement for +/// `NoOpRecordSink` (RFC 0014). See the module docs for the flush policy. +pub struct ParquetRecordSink { + store: Store, + config: FlushConfig, + buffers: HashMap, + total_bytes: usize, + flushes: u64, + records_flushed: u64, + flush_errors: u64, + derive_errors: u64, +} + +/// Cheap per-record footprint estimate driving the size trigger + ceiling. Not +/// the exact encoded (compressed) size — a conservative over-estimate is fine +/// for triggering; precise estimation is RFC 0014 §7. +fn estimate_bytes(r: &MinedRecord) -> usize { + let opt = |o: &Option| o.as_ref().map_or(0, String::len); + // Fixed per-record overhead plus the variable-length payloads. + 96 + opt(&r.body) + + opt(&r.severity_text) + + opt(&r.scope_name) + + opt(&r.scope_version) + + r.params.iter().map(|p| p.value.len() + 8).sum::() + + r.separators.iter().map(String::len).sum::() + // Attributes are encoded as JSON; a flat per-entry estimate suffices. + + (r.attributes.len() + r.resource_attributes.len()) * 48 +} + +/// `/`-delimited object key for a partition's flushed file: the RFC 0005 §3.4 +/// Hive path (relative to the store root) plus a `UUIDv7` name. Mirrors +/// `ourios_parquet::Writer`'s key; object keys are `/`-delimited on every host. +fn object_key(partition: &PartitionKey) -> String { + let rel = partition.data_path(Path::new("")); + format!( + "{}/{}.parquet", + rel.to_string_lossy() + .replace(std::path::MAIN_SEPARATOR, "/"), + Uuid::now_v7() + ) +} + +impl ParquetRecordSink { + /// Create a sink flushing to `store` under `config`. + #[must_use] + pub fn new(store: Store, config: FlushConfig) -> Self { + Self { + store, + config, + buffers: HashMap::new(), + total_bytes: 0, + flushes: 0, + records_flushed: 0, + flush_errors: 0, + derive_errors: 0, + } + } + + /// Total estimated bytes currently buffered across all partitions. + #[must_use] + pub fn buffered_bytes(&self) -> usize { + self.total_bytes + } + + /// Number of partitions with a non-empty buffer. + #[must_use] + pub fn buffered_partitions(&self) -> usize { + self.buffers.len() + } + + /// Count of successful partition flushes. + #[must_use] + pub fn flushes(&self) -> u64 { + self.flushes + } + + /// Total records written out across all successful flushes. + #[must_use] + pub fn records_flushed(&self) -> u64 { + self.records_flushed + } + + /// Records currently buffered (not yet flushed) across all partitions. + #[must_use] + pub fn buffered_records(&self) -> usize { + self.buffers.values().map(|b| b.records.len()).sum() + } + + /// Force-flush every buffered partition — the WAL-segment-rotation trigger + /// (RFC0014.3), including sub-threshold low-volume partitions. + pub fn flush_all(&mut self) { + let keys: Vec = self.buffers.keys().cloned().collect(); + for key in keys { + self.flush_partition_swallow(&key); + } + } + + /// Flush partitions whose oldest record has reached `max_buffer_age` — the + /// age trigger (RFC0014.2), driven by the batch-window tick. + pub fn flush_aged(&mut self) { + let max = self.config.max_buffer_age; + let keys: Vec = self + .buffers + .iter() + .filter(|(_, b)| b.oldest.elapsed() >= max) + .map(|(k, _)| k.clone()) + .collect(); + for key in keys { + self.flush_partition_swallow(&key); + } + } + + /// Encode + put one partition's buffer. On success the buffer is removed + /// and the counters advance; the caller (via [`Self::flush_partition_swallow`]) + /// retains it on error. + fn flush_partition(&mut self, key: &PartitionKey) -> Result<(), FlushError> { + let bytes = match self.buffers.get(key) { + Some(buf) if !buf.records.is_empty() => { + encode_records_to_parquet(&buf.records, DEFAULT_ZSTD_LEVEL) + .map_err(FlushError::Encode)? + } + _ => return Ok(()), + }; + self.store + .put_blocking(&object_key(key), bytes) + .map_err(FlushError::Store)?; + if let Some(buf) = self.buffers.remove(key) { + self.total_bytes = self.total_bytes.saturating_sub(buf.est_bytes); + self.flushes += 1; + self.records_flushed += buf.records.len() as u64; + } + Ok(()) + } + + /// [`Self::flush_partition`] for the infallible `emit` / tick / rotation + /// paths: a failed flush retains the buffer (the WAL is the durability of + /// record) and is counted for observability. + fn flush_partition_swallow(&mut self, key: &PartitionKey) { + if self.flush_partition(key).is_err() { + self.flush_errors += 1; + } + } + + /// Flush the largest buffered partition to reclaim memory. Returns whether + /// a flush actually succeeded (so the ceiling loop stops if the store is + /// unavailable rather than spinning). + fn flush_largest(&mut self) -> bool { + let Some(key) = self + .buffers + .iter() + .filter(|(_, b)| !b.records.is_empty()) + .max_by_key(|(_, b)| b.est_bytes) + .map(|(k, _)| k.clone()) + else { + return false; + }; + if self.flush_partition(&key).is_ok() { + true + } else { + self.flush_errors += 1; + false + } + } +} + +impl RecordSink for ParquetRecordSink { + fn emit(&mut self, record: MinedRecord) { + let Ok(key) = PartitionKey::derive(&record) else { + // Un-partitionable (timestamp overflow, §3.4 fallback exhausted): + // can't route it. The WAL still holds it; count and drop here. + self.derive_errors += 1; + return; + }; + let est = estimate_bytes(&record); + + // Ceiling (RFC0014.4): make room before appending so the total never + // exceeds the ceiling. Stop if nothing more can be flushed (store down + // / single oversized buffer) rather than spinning. + while self.total_bytes + est > self.config.ceiling_bytes && self.flush_largest() {} + + let buf = self + .buffers + .entry(key.clone()) + .or_insert_with(|| PartitionBuffer { + records: Vec::new(), + est_bytes: 0, + oldest: Instant::now(), + }); + buf.records.push(record); + buf.est_bytes += est; + let over_target = buf.est_bytes >= self.config.target_bytes; + self.total_bytes += est; + + // Size trigger (RFC0014.1): the emit that crosses the target flushes. + if over_target { + self.flush_partition_swallow(&key); + } + } +} diff --git a/crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs b/crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs index d238f0a7f..318bd4381 100644 --- a/crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs +++ b/crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs @@ -1,36 +1,149 @@ //! RFC 0014 — ingest write path (record sink + flush policy) acceptance //! scenarios (§5). //! -//! **Status: `red`.** These are the failing stubs that drive the `green` -//! implementation: each encodes one RFC 0014 §5 scenario and currently -//! `todo!()`s. They are `#[ignore]`d so the default `cargo test` (and CI) -//! stays green while the buffering `ParquetRecordSink` is built — `green` -//! replaces each body with a real assertion against the sink (per-partition -//! buffers; hybrid size + age + WAL-rotation flush; hard byte-ceiling with -//! blocking backpressure) and removes the `#[ignore]`. -//! -//! Placement may shift at `green`: RFC0014.5 (crash recovery) extends the RFC -//! 0008 WAL harness here in `ourios-ingester`; the buffer-trigger scenarios -//! (.1–.4, .6) may move next to the sink wherever it lands. +//! `.1`–`.4`/`.6` drive the [`ParquetRecordSink`] directly against a +//! `LocalFileSystem`-backed [`Store`] (synthetic `MinedRecord` streams); +//! reading the flushed objects back proves no loss + tenant isolation. `.5` +//! (crash recovery) stays `#[ignore]`d until the sink is wired into the ingest +//! pipeline and the RFC 0008 WAL crash harness is extended (`green`, part 2). //! //! See `docs/rfcs/0014-ingest-write-path.md` §5/§6. +use std::path::Path; +use std::time::Duration; + +use ourios_core::audit::ParamType; +use ourios_core::record::{BodyKind, MinedRecord, Param, RecordSink}; +use ourios_core::tenant::TenantId; +use ourios_ingester::record_sink::{FlushConfig, ParquetRecordSink}; +use ourios_parquet::{Reader, Store}; + +/// A clean-round-trip record for `tenant` at in-hour offset `i`. +fn rec_for(tenant: &str, i: u64) -> MinedRecord { + MinedRecord { + tenant_id: TenantId::new(tenant), + template_id: 1, + template_version: 1, + severity_number: 9, + severity_text: Some("INFO".to_string()), + scope_name: Some("lib.cart".to_string()), + scope_version: Some("1.0.0".to_string()), + time_unix_nano: 1_775_127_480_000_000_000 + i * 1_000, + observed_time_unix_nano: Some(1_775_127_480_000_000_000 + i * 1_000 + 1), + attributes: Vec::new(), + dropped_attributes_count: 0, + resource_attributes: Vec::new(), + trace_id: None, + span_id: None, + flags: 0x01, + event_name: None, + body_kind: BodyKind::String, + params: vec![Param { + type_tag: ParamType::Num, + value: format!("{i}"), + }], + separators: vec![String::new(), " ".to_string()], + body: None, + confidence: 1.0, + lossy_flag: false, + } +} + +/// Every flushed `*.parquet` object under `root`, one inner `Vec` per file +/// (so per-file tenant isolation can be asserted). +fn parquet_files(root: &Path) -> Vec> { + let mut files = Vec::new(); + let mut stack = vec![root.to_path_buf()]; + while let Some(dir) = stack.pop() { + let Ok(entries) = std::fs::read_dir(&dir) else { + continue; + }; + for entry in entries.flatten() { + let path = entry.path(); + if path.is_dir() { + stack.push(path); + } else if path.extension().is_some_and(|x| x == "parquet") { + let rows = Reader::open_file(&path) + .expect("open_file") + .read_all() + .expect("read_all"); + files.push(rows); + } + } + } + files +} + +fn all_rows(root: &Path) -> Vec { + parquet_files(root).into_iter().flatten().collect() +} + +fn sink(dir: &Path, config: FlushConfig) -> ParquetRecordSink { + ParquetRecordSink::new(Store::local(dir).expect("local store"), config) +} + +const HUGE: usize = 1 << 40; +const FOREVER: Duration = Duration::from_secs(86_400); + /// Scenario RFC0014.1 — Size trigger: the emit that crosses the size target /// flushes the partition to one right-sized Parquet object. /// See `docs/rfcs/0014-ingest-write-path.md` §5. #[test] -#[ignore = "RFC0014.1 — red until the ParquetRecordSink + flush policy land (green)"] fn rfc0014_1_size_trigger() { - todo!("RFC0014.1: a partition flushes on the emit that crosses the size target") + let dir = tempfile::TempDir::new().expect("temp dir"); + let mut s = sink( + dir.path(), + FlushConfig { + target_bytes: 500, + max_buffer_age: FOREVER, + ceiling_bytes: HUGE, + }, + ); + let records: Vec = (0..50).map(|i| rec_for("tenant-a", i)).collect(); + for r in &records { + s.emit(r.clone()); + } + // The size trigger alone (no age/rotation flush called) has fired. + assert!(s.flushes() >= 1, "size trigger flushed mid-stream"); + // Drain the buffered tail, then confirm no loss across the flushed objects. + s.flush_all(); + let mut got = all_rows(dir.path()); + assert_eq!( + got.len(), + records.len(), + "every record published, none lost" + ); + got.sort_by_key(|r| r.params[0].value.parse::().unwrap_or_default()); + assert_eq!(got, records, "rows recovered byte-for-byte"); } /// Scenario RFC0014.2 — Age trigger: a sub-target low-volume partition flushes /// on the next batch-window tick once its oldest record reaches `max_buffer_age`. /// See `docs/rfcs/0014-ingest-write-path.md` §5. #[test] -#[ignore = "RFC0014.2 — red until the ParquetRecordSink + flush policy land (green)"] fn rfc0014_2_age_trigger() { - todo!("RFC0014.2: low-volume partition flushes on age") + let dir = tempfile::TempDir::new().expect("temp dir"); + // Size never triggers; age is inclusive at zero, so any buffered record is + // immediately "aged" — `flush_aged` (the tick) is the only thing that flushes. + let mut s = sink( + dir.path(), + FlushConfig { + target_bytes: HUGE, + max_buffer_age: Duration::ZERO, + ceiling_bytes: HUGE, + }, + ); + for i in 0..5 { + s.emit(rec_for("tenant-a", i)); + } + assert_eq!( + s.flushes(), + 0, + "size/ceiling did not flush a low-volume partition" + ); + s.flush_aged(); + assert_eq!(s.flushes(), 1, "the age sweep flushed the partition"); + assert_eq!(all_rows(dir.path()).len(), 5); } /// Scenario RFC0014.3 — Rotation force-flush: a WAL segment rotation flushes @@ -38,18 +151,69 @@ fn rfc0014_2_age_trigger() { /// the sealed segment. /// See `docs/rfcs/0014-ingest-write-path.md` §5. #[test] -#[ignore = "RFC0014.3 — red until the ParquetRecordSink + flush policy land (green)"] fn rfc0014_3_rotation_force_flush() { - todo!("RFC0014.3: rotation flushes every partition") + let dir = tempfile::TempDir::new().expect("temp dir"); + let mut s = sink( + dir.path(), + FlushConfig { + target_bytes: HUGE, + max_buffer_age: FOREVER, + ceiling_bytes: HUGE, + }, + ); + // Three tenants → three partitions, all below every other trigger. + for t in ["tenant-x", "tenant-y", "tenant-z"] { + for i in 0..3 { + s.emit(rec_for(t, i)); + } + } + assert_eq!(s.flushes(), 0, "nothing flushed before rotation"); + assert_eq!(s.buffered_partitions(), 3); + + s.flush_all(); // the WAL-segment-rotation trigger + + assert_eq!(s.flushes(), 3, "every partition flushed on rotation"); + assert_eq!( + s.buffered_partitions(), + 0, + "no buffered record predates the seal" + ); + assert_eq!(all_rows(dir.path()).len(), 9); } -/// Scenario RFC0014.4 — Bounded memory: the sink early-flushes under pressure -/// and, at the hard ceiling, `emit` blocks so buffered bytes never exceed it. +/// Scenario RFC0014.4 — Bounded memory: the sink flushes inline so buffered +/// bytes never exceed the hard ceiling; nothing is lost. /// See `docs/rfcs/0014-ingest-write-path.md` §5. #[test] -#[ignore = "RFC0014.4 — red until the ParquetRecordSink + flush policy land (green)"] fn rfc0014_4_bounded_memory() { - todo!("RFC0014.4: hard ceiling, never exceeded") + let dir = tempfile::TempDir::new().expect("temp dir"); + // Size/age never trigger; only the ceiling does. 100 records far exceed a + // 1 KiB ceiling, so the sink must flush inline to stay bounded. + let ceiling = 1024; + let mut s = sink( + dir.path(), + FlushConfig { + target_bytes: HUGE, + max_buffer_age: FOREVER, + ceiling_bytes: ceiling, + }, + ); + let n = 100; + for i in 0..n { + s.emit(rec_for("tenant-a", i)); + assert!( + s.buffered_bytes() <= ceiling, + "ceiling held after each emit: {} <= {ceiling}", + s.buffered_bytes(), + ); + } + assert!(s.flushes() >= 1, "the ceiling forced inline flushes"); + s.flush_all(); + assert_eq!( + all_rows(dir.path()).len() as u64, + n, + "no loss under backpressure" + ); } /// Scenario RFC0014.5 — No acknowledged-data loss: a crash with a non-empty @@ -57,7 +221,7 @@ fn rfc0014_4_bounded_memory() { /// record (`CLAUDE.md` §3.4). /// See `docs/rfcs/0014-ingest-write-path.md` §5. #[test] -#[ignore = "RFC0014.5 — red until the ParquetRecordSink + flush policy land (green)"] +#[ignore = "RFC0014.5 — green part 2: wire the sink into the ingest pipeline + extend the RFC 0008 crash harness"] fn rfc0014_5_no_acknowledged_data_loss() { todo!("RFC0014.5: crash mid-buffer loses no acknowledged data (WAL replay)") } @@ -66,7 +230,42 @@ fn rfc0014_5_no_acknowledged_data_loss() { /// only one tenant's rows; no buffer or flush crosses tenants (`CLAUDE.md` §3.7). /// See `docs/rfcs/0014-ingest-write-path.md` §5. #[test] -#[ignore = "RFC0014.6 — red until the ParquetRecordSink + flush policy land (green)"] fn rfc0014_6_tenant_isolation() { - todo!("RFC0014.6: no cross-tenant buffer or flush") + let dir = tempfile::TempDir::new().expect("temp dir"); + let mut s = sink( + dir.path(), + FlushConfig { + target_bytes: HUGE, + max_buffer_age: FOREVER, + ceiling_bytes: HUGE, + }, + ); + for i in 0..10 { + s.emit(rec_for("tenant-x", i)); + s.emit(rec_for("tenant-y", i)); + } + s.flush_all(); + + let files = parquet_files(dir.path()); + assert!(!files.is_empty(), "objects were published"); + for file in &files { + let tenants: std::collections::BTreeSet<&str> = + file.iter().map(|r| r.tenant_id.as_str()).collect(); + assert_eq!( + tenants.len(), + 1, + "each object holds exactly one tenant: {tenants:?}" + ); + } + let x = files + .iter() + .flatten() + .filter(|r| r.tenant_id.as_str() == "tenant-x") + .count(); + let y = files + .iter() + .flatten() + .filter(|r| r.tenant_id.as_str() == "tenant-y") + .count(); + assert_eq!((x, y), (10, 10), "both tenants' rows present, unmixed"); } From b5f35898cba9d8207f0f36d949ed661947a05c71 Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Wed, 17 Jun 2026 19:42:26 +0200 Subject: [PATCH 2/4] fix(ingester): precise ceiling-degradation docs; private FlushError; tighten RFC0014 tests (review) --- crates/ourios-ingester/src/record_sink.rs | 23 +++++++++++++------ .../tests/rfc0014_ingest_write_path.rs | 22 ++++++++++++++---- 2 files changed, 33 insertions(+), 12 deletions(-) diff --git a/crates/ourios-ingester/src/record_sink.rs b/crates/ourios-ingester/src/record_sink.rs index f666716c8..1ac41b441 100644 --- a/crates/ourios-ingester/src/record_sink.rs +++ b/crates/ourios-ingester/src/record_sink.rs @@ -38,15 +38,19 @@ pub struct FlushConfig { /// Age trigger: a partition flushes once its oldest buffered record's age /// reaches this (inclusive), bounding low-volume staleness. pub max_buffer_age: Duration, - /// Hard ceiling on total buffered bytes across all partitions; `emit` - /// flushes inline to stay at or under it. + /// Ceiling on total buffered bytes across all partitions; `emit` flushes + /// inline to stay at or under it whenever the store accepts writes. On a + /// flush failure the buffer is retained (the WAL is the durability of + /// record) and the ceiling may be transiently exceeded — surfaced as a + /// flush error — rather than stalling ingest. pub ceiling_bytes: usize, } /// A failed flush of one partition. Non-fatal — the buffer is retained and the -/// WAL remains the durability of record. +/// WAL remains the durability of record. Internal: the public `emit` / tick / +/// rotation surface is infallible (errors are swallowed + counted). #[derive(Debug)] -pub enum FlushError { +enum FlushError { /// Encoding the buffered records to Parquet failed. Encode(WriterError), /// Writing the encoded object to the store failed. @@ -252,9 +256,14 @@ impl RecordSink for ParquetRecordSink { }; let est = estimate_bytes(&record); - // Ceiling (RFC0014.4): make room before appending so the total never - // exceeds the ceiling. Stop if nothing more can be flushed (store down - // / single oversized buffer) rather than spinning. + // Ceiling (RFC0014.4): flush the largest partition inline to make room + // before appending, so buffered bytes stay at or under the ceiling + // whenever the store accepts writes. If a flush fails (store + // unavailable) or nothing more can be flushed (a single oversized + // buffer), the loop stops rather than spinning — the record is still + // retained below (the WAL is the durability of record), and the + // ceiling may be transiently exceeded (counted via `flush_errors`) + // instead of deadlocking the ingest path. while self.total_bytes + est > self.config.ceiling_bytes && self.flush_largest() {} let buf = self diff --git a/crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs b/crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs index 318bd4381..44e2f4584 100644 --- a/crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs +++ b/crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs @@ -113,7 +113,12 @@ fn rfc0014_1_size_trigger() { records.len(), "every record published, none lost" ); - got.sort_by_key(|r| r.params[0].value.parse::().unwrap_or_default()); + got.sort_by_key(|r| { + r.params[0] + .value + .parse::() + .expect("param round-trips as a parseable number") + }); assert_eq!(got, records, "rows recovered byte-for-byte"); } @@ -257,14 +262,21 @@ fn rfc0014_6_tenant_isolation() { "each object holds exactly one tenant: {tenants:?}" ); } - let x = files + let rows: Vec<&MinedRecord> = files.iter().flatten().collect(); + let tenants: std::collections::BTreeSet<&str> = + rows.iter().map(|r| r.tenant_id.as_str()).collect(); + assert_eq!( + tenants, + ["tenant-x", "tenant-y"].into_iter().collect(), + "exactly the two emitted tenants, no extras", + ); + assert_eq!(rows.len(), 20, "no extra rows beyond the 20 emitted"); + let x = rows .iter() - .flatten() .filter(|r| r.tenant_id.as_str() == "tenant-x") .count(); - let y = files + let y = rows .iter() - .flatten() .filter(|r| r.tenant_id.as_str() == "tenant-y") .count(); assert_eq!((x, y), (10, 10), "both tenants' rows present, unmixed"); From 1fd8df3dfd8499b83810e91c7fd96a20ef7f6a74 Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Wed, 17 Jun 2026 19:48:33 +0200 Subject: [PATCH 3/4] fix(ingester): pointer-width-safe HUGE sentinel; accurate ceiling/estimate docs (copilot) --- crates/ourios-ingester/src/record_sink.rs | 24 +++++++++++-------- .../tests/rfc0014_ingest_write_path.rs | 3 ++- 2 files changed, 16 insertions(+), 11 deletions(-) diff --git a/crates/ourios-ingester/src/record_sink.rs b/crates/ourios-ingester/src/record_sink.rs index 1ac41b441..5a92387b7 100644 --- a/crates/ourios-ingester/src/record_sink.rs +++ b/crates/ourios-ingester/src/record_sink.rs @@ -7,9 +7,11 @@ //! its oldest record reaches [`FlushConfig::max_buffer_age`] (age, evaluated by //! [`ParquetRecordSink::flush_aged`] on the batch-window tick), or when the WAL //! segment rotates ([`ParquetRecordSink::flush_all`], force-flushing *every* -//! partition). Total buffered bytes are kept under -//! [`FlushConfig::ceiling_bytes`] by flushing the largest partition inline -//! before `emit` would exceed it (RFC 0014 §3.4 — a hard ceiling). +//! partition). Buffered bytes are kept under [`FlushConfig::ceiling_bytes`] +//! by flushing the largest partition inline before `emit` would exceed it +//! (RFC 0014 §3.4) — a hard ceiling whenever the store accepts writes; a flush +//! failure retains the buffer and may transiently exceed it rather than +//! stalling ingest (see [`FlushConfig::ceiling_bytes`]). //! //! Records reach the sink only after the WAL is durable (`CLAUDE.md` §3.4), so //! a buffer is a bounded accelerator, never the durability of record: a crash @@ -39,10 +41,11 @@ pub struct FlushConfig { /// reaches this (inclusive), bounding low-volume staleness. pub max_buffer_age: Duration, /// Ceiling on total buffered bytes across all partitions; `emit` flushes - /// inline to stay at or under it whenever the store accepts writes. On a - /// flush failure the buffer is retained (the WAL is the durability of - /// record) and the ceiling may be transiently exceeded — surfaced as a - /// flush error — rather than stalling ingest. + /// inline to stay at or under it whenever the store accepts writes. If a + /// flush fails, or a single record alone exceeds the ceiling (nothing left + /// to flush), the buffer is retained (the WAL is the durability of record) + /// and the ceiling may be transiently exceeded — rather than stalling + /// ingest. (A failed flush attempt is also counted as a flush error.) pub ceiling_bytes: usize, } @@ -94,9 +97,10 @@ pub struct ParquetRecordSink { derive_errors: u64, } -/// Cheap per-record footprint estimate driving the size trigger + ceiling. Not -/// the exact encoded (compressed) size — a conservative over-estimate is fine -/// for triggering; precise estimation is RFC 0014 §7. +/// Cheap per-record footprint estimate driving the size trigger + ceiling — a +/// rough heuristic over the larger variable-length fields, not the exact +/// encoded (compressed) size and not every field. Good enough to bound memory +/// and roughly right-size files; precise estimation is RFC 0014 §7. fn estimate_bytes(r: &MinedRecord) -> usize { let opt = |o: &Option| o.as_ref().map_or(0, String::len); // Fixed per-record overhead plus the variable-length payloads. diff --git a/crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs b/crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs index 44e2f4584..fd7620969 100644 --- a/crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs +++ b/crates/ourios-ingester/tests/rfc0014_ingest_write_path.rs @@ -82,7 +82,8 @@ fn sink(dir: &Path, config: FlushConfig) -> ParquetRecordSink { ParquetRecordSink::new(Store::local(dir).expect("local store"), config) } -const HUGE: usize = 1 << 40; +/// A "never trigger" sentinel for the size/ceiling knobs (pointer-width safe). +const HUGE: usize = usize::MAX; const FOREVER: Duration = Duration::from_secs(86_400); /// Scenario RFC0014.1 — Size trigger: the emit that crosses the size target From 52291f4e9631ec02ab8709dff89b9c8161c3a628 Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Wed, 17 Jun 2026 19:55:38 +0200 Subject: [PATCH 4/4] fix(ingester): saturating byte-counter adds in the sink (copilot) --- crates/ourios-ingester/src/record_sink.rs | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) diff --git a/crates/ourios-ingester/src/record_sink.rs b/crates/ourios-ingester/src/record_sink.rs index 5a92387b7..7e3e805fc 100644 --- a/crates/ourios-ingester/src/record_sink.rs +++ b/crates/ourios-ingester/src/record_sink.rs @@ -268,7 +268,9 @@ impl RecordSink for ParquetRecordSink { // retained below (the WAL is the durability of record), and the // ceiling may be transiently exceeded (counted via `flush_errors`) // instead of deadlocking the ingest path. - while self.total_bytes + est > self.config.ceiling_bytes && self.flush_largest() {} + while self.total_bytes.saturating_add(est) > self.config.ceiling_bytes + && self.flush_largest() + {} let buf = self .buffers @@ -279,9 +281,12 @@ impl RecordSink for ParquetRecordSink { oldest: Instant::now(), }); buf.records.push(record); - buf.est_bytes += est; + // Saturating (matching `saturating_sub` on flush) so the byte counters + // stay monotonic and the triggers can't be corrupted by wraparound + // under prolonged retention (e.g. a store outage). + buf.est_bytes = buf.est_bytes.saturating_add(est); let over_target = buf.est_bytes >= self.config.target_bytes; - self.total_bytes += est; + self.total_bytes = self.total_bytes.saturating_add(est); // Size trigger (RFC0014.1): the emit that crosses the target flushes. if over_target {