diff --git a/crates/ourios-miner/src/cluster.rs b/crates/ourios-miner/src/cluster.rs index 6e784963e..92bf70743 100644 --- a/crates/ourios-miner/src/cluster.rs +++ b/crates/ourios-miner/src/cluster.rs @@ -1153,10 +1153,10 @@ impl MinerCluster { // record with `BodyKind::Absent` and the // template-id sentinel; tokenize/mask didn't // run, so there's no separator / param info to - // carry. `lossy_flag = true` because there is no - // template, so reconstruction is not possible. - let mut rec = Self::record_envelope(record, BodyKind::Absent); - rec.lossy_flag = true; + // carry. `lossy_flag = false` per RFC 0025 §3.1: + // absence is not loss — reconstruction is defined + // and total (it renders nothing). + let rec = Self::record_envelope(record, BodyKind::Absent); self.emit_record(rec, service); NO_TEMPLATE } @@ -4331,7 +4331,10 @@ mod tests { assert_eq!(rec.tenant_id, t); assert_eq!(rec.template_id, NO_TEMPLATE); assert_eq!(rec.body_kind, BodyKind::Absent); - assert!(rec.lossy_flag, "Body::None records are lossy (no template)"); + assert!( + !rec.lossy_flag, + "absence is not loss (RFC 0025 §3.1) — reconstruction renders nothing, exactly" + ); assert!(rec.separators.is_empty()); assert!(rec.params.is_empty()); assert!(rec.body.is_none()); diff --git a/crates/ourios-parquet/src/reader.rs b/crates/ourios-parquet/src/reader.rs index 7bd043e23..c29235cb1 100644 --- a/crates/ourios-parquet/src/reader.rs +++ b/crates/ourios-parquet/src/reader.rs @@ -682,9 +682,13 @@ fn decode_body_kind(ord: u8) -> Result { match ord { 0 => Ok(BodyKind::String), 1 => Ok(BodyKind::Structured), + 2 => Ok(BodyKind::Absent), other => Err(ReaderError::Conversion { column: columns::BODY_KIND, - detail: format!("unknown ordinal {other} (RFC 0005 §3.2 pins 0=String, 1=Structured)"), + detail: format!( + "unknown ordinal {other} (RFC 0005 §3.2 pins 0=String, 1=Structured, \ + 2=Absent per RFC 0025)" + ), }), } } diff --git a/crates/ourios-parquet/src/record_batch.rs b/crates/ourios-parquet/src/record_batch.rs index 85fcde09d..925eb0731 100644 --- a/crates/ourios-parquet/src/record_batch.rs +++ b/crates/ourios-parquet/src/record_batch.rs @@ -166,16 +166,12 @@ pub enum BatchError { count: usize, source: ourios_core::otlp::canonical::CanonicalJsonError, }, - /// A record carried [`BodyKind::Absent`] (the in-memory - /// "wire delivered no body" variant). RFC 0005 §3.2's - /// `body_kind` column pins exactly two ordinals (`0 = String, - /// 1 = Structured`); silently mapping `Absent` to one of - /// them would misclassify wire-absent rows. Until a future - /// RFC 0005 amendment either adds a third ordinal or adds a - /// separate `body_present` boolean column, the writer - /// rejects these records rather than corrupting the - /// `body_kind` semantics. - UnsupportedAbsentBody, + /// A `body_kind = Absent` record carried body bytes. RFC 0025 + /// §3.1 pins the on-disk contract as ordinal 2 **with a `NULL` + /// `body` cell**; silently writing the bytes would smuggle an + /// undefined state into the schema, so the writer rejects the + /// producer bug loudly. + NonNullBodyForAbsent, /// A clean-attach `body_kind = String` record had too few /// `separators` entries to satisfy the RFC 0005 §3.2 /// invariant ("`tokens.len() + 1` elements when @@ -226,13 +222,6 @@ impl fmt::Display for BatchError { `with-serde` derives are infallible on every spec-compliant `AnyValue`; \ this means an `opentelemetry-proto` upgrade broke that contract)", ), - Self::UnsupportedAbsentBody => write!( - f, - "record carries BodyKind::Absent (wire-absent body), which RFC 0005 §3.2's \ - body_kind column does not yet encode (the column pins ordinals 0=String, \ - 1=Structured); a future RFC 0005 amendment is required to represent this \ - in the schema", - ), Self::InvalidSeparatorsForString { expected_at_least, actual, @@ -250,6 +239,12 @@ impl fmt::Display for BatchError { reconstruction path returns the retained body verbatim — without one, the \ record is unreconstructable on read", ), + Self::NonNullBodyForAbsent => write!( + f, + "body_kind = Absent record carries body bytes, but RFC 0025 §3.1 pins the \ + Absent on-disk contract as a NULL body cell — a producer bug, rejected \ + rather than written", + ), Self::Arrow(e) => write!(f, "arrow rejected RecordBatch: {e}"), } } @@ -259,9 +254,9 @@ impl std::error::Error for BatchError { fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { match self { Self::TimestampOverflow { .. } - | Self::UnsupportedAbsentBody | Self::InvalidSeparatorsForString { .. } - | Self::MissingBodyForLossyString => None, + | Self::MissingBodyForLossyString + | Self::NonNullBodyForAbsent => None, Self::AttributeEncode { source, .. } => Some(source), Self::Arrow(e) => Some(e), } @@ -421,7 +416,7 @@ impl Builders { self.flags.append_value(r.flags); append_option_str(&mut self.event_name, r.event_name.as_deref()); - self.body_kind.append_value(body_kind_ordinal(r.body_kind)?); + self.body_kind.append_value(body_kind_ordinal(r.body_kind)); // RFC 0005 §3.3: when `body_kind = Structured`, the // body column carries Ourios-canonical JSON — the bytes // the miner has already encoded via @@ -432,6 +427,9 @@ impl Builders { // is the retained line bytes on the §6.6 lossy path // (or `None` on the clean-attach path, reconstructed // from `template + params + separators` by the reader). + if r.body_kind == BodyKind::Absent && r.body.is_some() { + return Err(BatchError::NonNullBodyForAbsent); + } match r.body.as_deref() { Some(s) => self.body.append_value(s.as_bytes()), None => self.body.append_null(), @@ -534,16 +532,14 @@ fn append_option_str(b: &mut StringBuilder, v: Option<&str>) { } /// Map an in-memory [`BodyKind`] to the §3.2 on-disk `body_kind` -/// ordinal. The schema pins exactly two ordinals (`0 = String, -/// 1 = Structured`); `BodyKind::Absent` has no on-disk -/// representation today and the writer rejects records carrying -/// it via [`BatchError::UnsupportedAbsentBody`] rather than -/// silently misclassifying them. -fn body_kind_ordinal(k: BodyKind) -> Result { +/// ordinal (`0 = String, 1 = Structured, 2 = Absent` — the third +/// ordinal is the RFC 0025 §3.1 amendment; wire-absent rows carry a +/// `NULL` body cell). +fn body_kind_ordinal(k: BodyKind) -> u8 { match k { - BodyKind::String => Ok(0), - BodyKind::Structured => Ok(1), - BodyKind::Absent => Err(BatchError::UnsupportedAbsentBody), + BodyKind::String => 0, + BodyKind::Structured => 1, + BodyKind::Absent => 2, } } @@ -840,18 +836,42 @@ mod tests { assert_eq!(stored.value(0), canonical.as_bytes()); } - /// `BodyKind::Absent` is not representable in the §3.2 - /// `body_kind` column today (the ordinals pin to - /// `0 = String, 1 = Structured`). The writer rejects such - /// records rather than silently lumping them with String. + /// RFC 0025 §3.1: `BodyKind::Absent` writes ordinal 2 with a + /// `NULL` body cell — the contract change that retired the old + /// `UnsupportedAbsentBody` rejection (approved via RFC 0025; + /// wire-absent bodies are spec-legal and must persist). + #[test] + fn absent_body_kind_writes_ordinal_two_with_null_body() { + let mut rec = empty_record(); + rec.body_kind = BodyKind::Absent; + rec.body = None; + let batch = mined_records_to_batch(&[rec]).expect("Absent body must write"); + let kind_idx = batch.schema().index_of(crate::columns::BODY_KIND).unwrap(); + let kinds = batch + .column(kind_idx) + .as_any() + .downcast_ref::() + .expect("body_kind is UInt8"); + assert_eq!(kinds.value(0), 2, "Absent is ordinal 2 (RFC 0025 §3.1)"); + let body_idx = batch.schema().index_of(crate::columns::BODY).unwrap(); + assert!( + batch.column(body_idx).is_null(0), + "the body cell is NULL for Absent rows", + ); + } + + /// The §3.1 contract's other direction: an Absent record + /// carrying body bytes is a producer bug and must be rejected + /// loudly, never written. #[test] - fn absent_body_kind_returns_unsupported_error() { + fn absent_body_kind_with_bytes_is_rejected() { let mut rec = empty_record(); rec.body_kind = BodyKind::Absent; - let err = mined_records_to_batch(&[rec]).expect_err("Absent body must error"); + rec.body = Some("stray bytes".to_string()); + let err = mined_records_to_batch(&[rec]).expect_err("must reject"); assert!( - matches!(err, BatchError::UnsupportedAbsentBody), - "expected UnsupportedAbsentBody, got {err:?}", + matches!(err, BatchError::NonNullBodyForAbsent), + "expected NonNullBodyForAbsent, got {err:?}", ); } diff --git a/crates/ourios-parquet/tests/rfc0024_properties.rs b/crates/ourios-parquet/tests/rfc0024_properties.rs index 158d90414..53b21bfa9 100644 --- a/crates/ourios-parquet/tests/rfc0024_properties.rs +++ b/crates/ourios-parquet/tests/rfc0024_properties.rs @@ -78,7 +78,8 @@ fn fail(what: &str, e: impl std::fmt::Display) -> TestCaseError { /// Mine `batch`, store every writable row, read it back, and assert /// fidelity. Adversarial timestamps can exceed `i64::MAX`, which the /// writer *rejects by contract* (RFC 0005 §3.2 timestamp overflow) — -/// those rows are asserted rejected, everything else must round-trip. +/// those rows are asserted rejected, everything else (absent bodies +/// included, per RFC 0025) must round-trip. fn assert_round_trip(batch: &[OtlpLogRecord]) -> Result<(), TestCaseError> { let sink = SharedRecordSink::new(); let mut cluster = @@ -97,24 +98,17 @@ fn assert_round_trip(batch: &[OtlpLogRecord]) -> Result<(), TestCaseError> { let mut groups: Vec<(PartitionKey, Vec<(usize, MinedRecord)>)> = Vec::new(); let mut writable = vec![false; batch.len()]; for (i, mined) in emitted.into_iter().enumerate() { - let absent_body = mined.body_kind == ourios_core::record::BodyKind::Absent; let ts_overflow = !fits_i64(mined.time_unix_nano) || mined.observed_time_unix_nano.is_some_and(|t| !fits_i64(t)); - if absent_body || ts_overflow { - // - Absent body: KNOWN GAP (#362, found by this suite) — - // no §3.2 on-disk representation until the RFC 0005 - // amendment lands; this arm then turns into a - // round-trip. - // - Timestamp overflow: the §3.2 u64→i64 contract, on - // *either* timestamp column. - // Both must be loud rejections, never silent drops. + if ts_overflow { + // Timestamp overflow: the §3.2 u64→i64 contract, on + // *either* timestamp column — the writer's one remaining + // documented loud rejection. (Absent bodies round-trip + // since RFC 0025 §3.1 gave them ordinal 2.) let err = mined_records_to_batch(std::slice::from_ref(&mined)) .expect_err("the writer must reject this record, not silently map it"); prop_assert!( - matches!( - err, - BatchError::UnsupportedAbsentBody | BatchError::TimestampOverflow { .. } - ), + matches!(err, BatchError::TimestampOverflow { .. }), "record {}: rejected for an undocumented reason: {}", i, err diff --git a/crates/ourios-parquet/tests/rfc0025_absent_body.rs b/crates/ourios-parquet/tests/rfc0025_absent_body.rs index 3dc5dc0f9..b0bc3feab 100644 --- a/crates/ourios-parquet/tests/rfc0025_absent_body.rs +++ b/crates/ourios-parquet/tests/rfc0025_absent_body.rs @@ -4,30 +4,190 @@ //! quarantine (`.4`/`.5`) in //! `crates/ourios-ingester/tests/rfc0025_quarantine.rs`. //! -//! Stubs are `#[ignore]`d so the default run stays green while the -//! RFC is red; each names the green slice that discharges it. +//! The pre-amendment file is the committed +//! `testdata/rfc0025/pre-amendment.parquet` fixture (the RFC 0021 §6 +//! committed-fixture discipline), generated by the pre-RFC 0025 +//! writer — two ordinals only, no `Absent` rows. + +use ourios_core::audit::ParamType; +use ourios_core::record::{BodyKind, MinedRecord, Param}; +use ourios_core::tenant::TenantId; +use ourios_parquet::{PartitionKey, Reader, Writer}; +use tempfile::TempDir; + +/// 2026-04-02T10:58:00Z — the fixture baseline instant. +const TS0: u64 = 1_775_127_480_000_000_000; + +fn repo_root() -> std::path::PathBuf { + std::path::PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .parent() + .and_then(std::path::Path::parent) + .expect("workspace root") + .to_path_buf() +} + +fn fixture_path() -> std::path::PathBuf { + repo_root().join("testdata/rfc0025/pre-amendment.parquet") +} + +/// A minimal record of each pre-amendment kind, deterministic so the +/// fixture and the parity assertion agree forever. +fn pre_records() -> Vec { + vec![ + // Clean-attach String row (reconstructable shape). + MinedRecord { + tenant_id: TenantId::new("rfc0025"), + template_id: 11, + template_version: 1, + severity_number: 9, + severity_text: Some("INFO".to_string()), + scope_name: Some("lib.fixture".to_string()), + scope_version: None, + scope_attributes: Vec::new(), + resource_schema_url: None, + scope_schema_url: None, + time_unix_nano: TS0, + observed_time_unix_nano: None, + attributes: Vec::new(), + dropped_attributes_count: 0, + resource_attributes: Vec::new(), + trace_id: None, + span_id: None, + flags: 0, + event_name: None, + body_kind: BodyKind::String, + params: vec![Param { + type_tag: ParamType::Num, + value: "7".to_string(), + }], + separators: vec![String::new(), " ".to_string()], + body: None, + confidence: 1.0, + lossy_flag: false, + }, + // Structured row (canonical-JSON body bytes). + MinedRecord { + tenant_id: TenantId::new("rfc0025"), + template_id: 12, + template_version: 1, + severity_number: 9, + severity_text: None, + scope_name: None, + scope_version: None, + scope_attributes: Vec::new(), + resource_schema_url: None, + scope_schema_url: None, + time_unix_nano: TS0 + 1_000, + observed_time_unix_nano: None, + attributes: Vec::new(), + dropped_attributes_count: 0, + resource_attributes: Vec::new(), + trace_id: None, + span_id: None, + flags: 0, + event_name: None, + body_kind: BodyKind::Structured, + params: Vec::new(), + separators: Vec::new(), + body: Some(r#"{"intValue":"42"}"#.to_string()), + confidence: 1.0, + // Always false for structured rows (RFC 0001 §6.1) — + // the body carries canonical JSON, reconstruction is + // defined. + lossy_flag: false, + }, + ] +} + +/// An absent-body record as the miner emits it post-RFC 0025: +/// `NO_TEMPLATE`, empty params/separators, `body = None`, and — +/// per §3.1 — `lossy_flag = false` (absence reconstructs exactly, +/// to nothing). +fn absent_record(ts_offset: u64) -> MinedRecord { + MinedRecord { + body_kind: BodyKind::Absent, + template_id: 0, + template_version: 0, + time_unix_nano: TS0 + ts_offset, + body: None, + params: Vec::new(), + separators: Vec::new(), + confidence: 0.0, + lossy_flag: false, + ..pre_records().remove(0) + } +} /// Scenario RFC0025.1 — absent bodies round-trip. /// See `docs/rfcs/0025-absent-body-representation.md` §5. #[test] -#[ignore = "RFC0025.1 stub — implemented in the schema green slice"] fn rfc0025_1_absent_bodies_round_trip() { - todo!( - "RFC0025.1 — a BodyKind::Absent record writes under body_kind \ - ordinal 2 with a NULL body cell and reads back with every \ - RFC 0005 §3.2 column intact; the RFC 0024 P1 pinned-rejection \ - arm for absent bodies flips to a round-trip assertion" - ); + let bucket = TempDir::new().expect("temp dir"); + let originals = vec![ + pre_records().remove(0), + absent_record(2_000), + absent_record(3_000), + ]; + + let partition = PartitionKey::derive(&originals[0]).expect("derive"); + let mut writer = Writer::open(bucket.path(), partition.clone()).expect("open writer"); + writer.append_records(&originals).expect("append"); + let written = writer.close().expect("close"); + + let reader = Reader::open_partition(&written.path, partition).expect("open_partition"); + let round_tripped = reader.read_all().expect("read_all"); + + assert_eq!(round_tripped, originals, "every RFC 0005 §3.2 column"); + for r in round_tripped + .iter() + .filter(|r| r.body_kind == BodyKind::Absent) + { + assert_eq!(r.body, None, "the body cell is NULL for Absent rows"); + assert!(!r.lossy_flag, "absence is not loss (RFC 0025 §3.1)"); + } } /// Scenario RFC0025.2 — old files unaffected. /// See `docs/rfcs/0025-absent-body-representation.md` §5. #[test] -#[ignore = "RFC0025.2 stub — implemented in the schema green slice"] fn rfc0025_2_old_files_unaffected() { - todo!( - "RFC0025.2 — a pre-amendment committed fixture reads identically \ - under the amended reader (the RFC 0021 §6 committed-fixture \ - parity discipline)" + let bucket = TempDir::new().expect("temp dir"); + let partition = PartitionKey::derive(&pre_records()[0]).expect("derive"); + let dir = partition.data_path(bucket.path()); + std::fs::create_dir_all(&dir).expect("mkdir partition"); + std::fs::copy(fixture_path(), dir.join("0-pre-amendment.parquet")).expect( + "committed fixture missing — regenerate via the ignored rfc0025_fixture test \ + (pre-amendment writer only)", + ); + + let reader = Reader::open_partition(&dir.join("0-pre-amendment.parquet"), partition) + .expect("open_partition"); + let rows = reader.read_all().expect("read_all"); + assert_eq!( + rows, + pre_records(), + "pre-amendment files read identically under the amended reader", ); } + +/// Regenerates `testdata/rfc0025/pre-amendment.parquet`. Generated +/// once with the **pre-RFC 0025** writer (two `body_kind` ordinals) +/// and committed; kept for provenance — the rows deliberately avoid +/// `Absent`, so the output is stable across the amendment. Run +/// manually (`cargo test -p ourios-parquet --test rfc0025_absent_body \ +/// -- --ignored rfc0025_fixture`); never in CI. +#[test] +#[ignore = "fixture generator — run manually, commit the output"] +fn rfc0025_fixture() { + let out_dir = fixture_path(); + let out_dir = out_dir.parent().expect("fixture parent"); + std::fs::create_dir_all(out_dir).expect("mkdir testdata/rfc0025"); + + let scratch = TempDir::new().expect("temp dir"); + let partition = PartitionKey::derive(&pre_records()[0]).expect("derive"); + let mut writer = Writer::open(scratch.path(), partition).expect("open writer"); + writer.append_records(&pre_records()).expect("append"); + let written = writer.close().expect("close"); + std::fs::copy(&written.path, fixture_path()).expect("install fixture"); + eprintln!("fixture written to {}", fixture_path().display()); +} diff --git a/testdata/rfc0025/pre-amendment.parquet b/testdata/rfc0025/pre-amendment.parquet new file mode 100644 index 000000000..39697a793 Binary files /dev/null and b/testdata/rfc0025/pre-amendment.parquet differ