diff --git a/.gitignore b/.gitignore index 2ad047595..6db3449a5 100644 --- a/.gitignore +++ b/.gitignore @@ -16,5 +16,12 @@ *.swp *~ +# Claude Code runtime artefacts (per-session PID / lock files). The +# `.claude/` tree carries tracked content (`.claude/skills/` slash- +# command definitions, the personal `settings.local.json`), but the +# scheduled-task lock file is per-machine ephemera that leaked into +# PR #62 once and shouldn't again. +/.claude/scheduled_tasks.lock + # Cargo.lock is NOT ignored — Ourios is a binary (CLAUDE.md §1) and its # lockfile is part of the reproducible-build contract. diff --git a/Cargo.lock b/Cargo.lock index 70912ab06..5f93ede8d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -945,6 +945,7 @@ version = "0.0.0" dependencies = [ "blake3", "opentelemetry-proto", + "serde_json", ] [[package]] diff --git a/crates/ourios-core/Cargo.toml b/crates/ourios-core/Cargo.toml index 8963ac7e4..d5c7158a8 100644 --- a/crates/ourios-core/Cargo.toml +++ b/crates/ourios-core/Cargo.toml @@ -25,6 +25,11 @@ opentelemetry-proto = { version = "0.32", default-features = false, features = [ # §6.4. Truncated blake3 (first 16 bytes) joins audit events back to # the data record(s) that triggered them in the §6.7 drift query. blake3 = { version = "1", default-features = false } +# Drives the RFC 0005 §3.3 OTLP-canonical-JSON encoders / decoders +# (`otlp::canonical`). Pairs with `opentelemetry-proto`'s `with-serde` +# derives so the spec mapping stays single-sourced through the proto +# crate. +serde_json = { version = "1", default-features = false, features = ["std"] } [lints] workspace = true diff --git a/crates/ourios-core/src/otlp.rs b/crates/ourios-core/src/otlp.rs index 1088a27cf..3cbc52b1f 100644 --- a/crates/ourios-core/src/otlp.rs +++ b/crates/ourios-core/src/otlp.rs @@ -35,7 +35,9 @@ use crate::tenant::TenantId; // crate deep. `any_value` is also re-exported because callers // constructing a non-string `AnyValue` need its `Value` enum to // fill the `value: Option` field. -pub use opentelemetry_proto::tonic::common::v1::{AnyValue, KeyValue, any_value}; +pub use opentelemetry_proto::tonic::common::v1::{ + AnyValue, ArrayValue, KeyValue, KeyValueList, any_value, +}; /// One OTLP `LogRecord` after wire decode and tenant derivation. /// @@ -190,6 +192,258 @@ impl Body { } } +/// RFC 0005 §3.3 OTLP-canonical-JSON encoding for the columns +/// the writer stores as `BYTE_ARRAY`: `attributes`, +/// `resource_attributes`, and the `body` column for +/// `body_kind = Structured`. +/// +/// The "canonical" rule is the proto3 JSON mapping plus OTLP's +/// specific overrides (camelCase fields, `string`-encoded +/// `uint64`s, base64 for `bytes`, etc.). The +/// `opentelemetry-proto` crate's `with-serde` feature already +/// implements that spec on its proto types — these helpers are +/// thin wrappers so callers don't reach for `serde_json` +/// directly and the spec mapping stays single-sourced through +/// `opentelemetry-proto`. The same pattern rotel's OTLP HTTP +/// receiver uses on `ExportLogsServiceRequest`. +/// +/// Encoders are stable per-`AnyValue`-tree: serde derives have +/// a fixed field order and `serde_json` is deterministic, so +/// re-encoding the same in-memory tree produces byte-identical +/// output across runs — required by RFC0006.7 reproducibility. +/// +/// **Note on "canonical".** RFC 0005 §3.3 uses "canonical" to +/// mean "the single normative encoding the writer / reader +/// agree on," **not** the RFC 8785 canonical-JSON form (sorted +/// keys, normalised numbers). The two are compatible for our +/// purposes because struct field order is fixed by proto and +/// the encode → store → decode round-trip is asserted at the +/// `AnyValue` / `Vec` level (not on bytes). +pub mod canonical { + use super::{AnyValue, KeyValue}; + + /// Error returned by the canonical encoders / decoders. + /// Encoders are infallible on every `AnyValue` / + /// `Vec` the type system admits — + /// `opentelemetry-proto`'s `with-serde` ships custom + /// serializers for the proto3-JSON oddities (`f64::NAN` + /// → `"NaN"` string, `i64` → string-encoded JSON number, + /// `bytes` → base64) so the recursive primitives never + /// panic or return an error. The `Encode` arm exists only + /// for `Result`-symmetry with `Decode` and as a defence- + /// in-depth surface if a future `opentelemetry-proto` + /// release changes that contract. Decoders fan in + /// malformed-bytes errors from disk. + #[derive(Debug)] + pub enum CanonicalJsonError { + Encode(serde_json::Error), + Decode(serde_json::Error), + } + + impl core::fmt::Display for CanonicalJsonError { + fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result { + match self { + Self::Encode(e) => write!(f, "OTLP-canonical JSON encode: {e}"), + Self::Decode(e) => write!(f, "OTLP-canonical JSON decode: {e}"), + } + } + } + + impl std::error::Error for CanonicalJsonError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Encode(e) | Self::Decode(e) => Some(e), + } + } + } + + /// Encode one `AnyValue` to its OTLP-canonical JSON bytes. + /// Used by the writer / miner for `body_kind = Structured` + /// rows (the `body` column stores these bytes). + /// + /// # Errors + /// + /// In principle never; see the + /// [`CanonicalJsonError`] doc. The `Result` is kept for + /// API symmetry with [`decode_any_value`] and as a + /// forward-compat surface if a future + /// `opentelemetry-proto` release adds a fallible + /// serializer. + pub fn encode_any_value(value: &AnyValue) -> Result, CanonicalJsonError> { + serde_json::to_vec(value).map_err(CanonicalJsonError::Encode) + } + + /// Inverse of [`encode_any_value`]. Used by the reader to + /// recover the structured `AnyValue` from its stored bytes. + /// + /// # Errors + /// + /// [`CanonicalJsonError::Decode`] on malformed bytes (file + /// corruption or a foreign producer that doesn't honour the + /// §3.3 spec). + pub fn decode_any_value(bytes: &[u8]) -> Result { + serde_json::from_slice(bytes).map_err(CanonicalJsonError::Decode) + } + + /// Encode a `Vec` (the in-memory shape of + /// `attributes` / `resource_attributes`) to its + /// OTLP-canonical JSON bytes. Stored verbatim in the + /// matching `BYTE_ARRAY` column by the writer. + /// + /// # Errors + /// + /// In principle never; the recursive primitives bottom out + /// at infallible serializers (see [`encode_any_value`] / + /// the [`CanonicalJsonError`] doc). + pub fn encode_attributes(attrs: &[KeyValue]) -> Result, CanonicalJsonError> { + serde_json::to_vec(attrs).map_err(CanonicalJsonError::Encode) + } + + /// Inverse of [`encode_attributes`]. The reader uses this + /// to recover the `Vec` from a non-empty stored + /// column. + /// + /// # Errors + /// + /// [`CanonicalJsonError::Decode`] on malformed bytes. + pub fn decode_attributes(bytes: &[u8]) -> Result, CanonicalJsonError> { + serde_json::from_slice(bytes).map_err(CanonicalJsonError::Decode) + } + + #[cfg(test)] + mod tests { + use super::*; + use crate::otlp::any_value; + + fn string_av(s: &str) -> AnyValue { + AnyValue { + value: Some(any_value::Value::StringValue(s.to_string())), + } + } + + fn int_av(n: i64) -> AnyValue { + AnyValue { + value: Some(any_value::Value::IntValue(n)), + } + } + + /// Encode → decode round-trips an `AnyValue` for every + /// `Value` variant the receiver might hand the storage + /// layer. The serde derives are spec-compliant; this + /// pins the round-trip property the §3.3 reconstruction + /// guarantee depends on. + #[test] + fn any_value_round_trips_across_variants() { + for av in [ + string_av("hello world"), + int_av(-42), + AnyValue { + value: Some(any_value::Value::DoubleValue(2.71_f64)), + }, + AnyValue { + value: Some(any_value::Value::BoolValue(true)), + }, + AnyValue { + value: Some(any_value::Value::BytesValue(b"raw\x00bytes".to_vec())), + }, + AnyValue { + value: Some(any_value::Value::ArrayValue( + opentelemetry_proto::tonic::common::v1::ArrayValue { + values: vec![string_av("a"), int_av(1)], + }, + )), + }, + AnyValue { + value: Some(any_value::Value::KvlistValue( + opentelemetry_proto::tonic::common::v1::KeyValueList { + values: vec![KeyValue { + key: "k".to_string(), + value: Some(string_av("v")), + ..Default::default() + }], + }, + )), + }, + ] { + let bytes = encode_any_value(&av).expect("encode"); + let back = decode_any_value(&bytes).expect("decode"); + assert_eq!( + av, + back, + "round-trip failed for {av:?}; bytes = {}", + String::from_utf8_lossy(&bytes), + ); + } + } + + /// `Vec` round-trips at the + /// `attributes` / `resource_attributes` column boundary. + #[test] + fn attributes_round_trip() { + let attrs = vec![ + KeyValue { + key: "service.name".to_string(), + value: Some(string_av("bench-app")), + ..Default::default() + }, + KeyValue { + key: "user.id".to_string(), + value: Some(int_av(42)), + ..Default::default() + }, + ]; + let bytes = encode_attributes(&attrs).expect("encode"); + let back = decode_attributes(&bytes).expect("decode"); + assert_eq!(attrs, back); + } + + /// Re-encoding the same in-memory tree must produce + /// byte-identical bytes — RFC0006.7's reproducibility + /// requirement carries through the canonicalisation + /// boundary. + #[test] + fn encoder_is_deterministic_across_calls() { + let av = AnyValue { + value: Some(any_value::Value::KvlistValue( + opentelemetry_proto::tonic::common::v1::KeyValueList { + values: vec![ + KeyValue { + key: "alpha".to_string(), + value: Some(int_av(1)), + ..Default::default() + }, + KeyValue { + key: "beta".to_string(), + value: Some(string_av("two")), + ..Default::default() + }, + ], + }, + )), + }; + let a = encode_any_value(&av).expect("first encode"); + let b = encode_any_value(&av).expect("second encode"); + assert_eq!( + a, b, + "encoder must be byte-deterministic for the same input" + ); + } + + /// Empty attribute lists encode to a sentinel that + /// decodes back to an empty `Vec`. The writer special- + /// cases the empty case (no row-level allocation), but + /// the helper itself round-trips on the trivial input + /// for symmetry. + #[test] + fn empty_attributes_round_trip() { + let bytes = encode_attributes(&[]).expect("encode"); + assert_eq!(bytes, b"[]"); + let back = decode_attributes(&bytes).expect("decode"); + assert!(back.is_empty()); + } + } +} + impl Default for TenantId { /// Empty-tenant default exists so `OtlpLogRecord::default()` /// works in tests; production receivers always derive a diff --git a/crates/ourios-miner/src/cluster.rs b/crates/ourios-miner/src/cluster.rs index 798c2d5f6..233b611d9 100644 --- a/crates/ourios-miner/src/cluster.rs +++ b/crates/ourios-miner/src/cluster.rs @@ -1636,25 +1636,46 @@ impl MinerCluster { // per RFC §6.1 ("Always false when body_kind = // Structured"). // - // **Interim body format.** OTLP-canonical JSON encoding - // is the follow-up PR named in `ourios-core::otlp`. - // Until it ships, we still populate `body` with a - // stored representation of the `AnyValue` — its `Debug` - // form — so `reconstruct(structured) == record.body` - // holds in the §3.3 sense ("what we stored is what we - // return"). The format is **not** OTLP-canonical yet; - // the canonicalisation PR replaces this `Debug` rendering - // with the canonical JSON encoding without changing the - // schema field or `lossy_flag`. The reader-visible effect - // of the migration is the bytes in the column changing - // shape; the contract that `reconstruct` returns - // whatever the producer stored is invariant. - let stored_body = format!("{any_value:?}"); + // `body` carries the RFC 0005 §3.3 OTLP-canonical-JSON + // encoding of the `AnyValue` — the bytes the writer + // stores in the §3.2 `body` column for structured rows. + // Two interlocking invariants prevent any fallback path + // that would weaken this: + // + // - RFC 0001 §6.1 / body-representation table: + // `lossy_flag` is **always `false` when + // `body_kind = Structured`** ("the verbatim `body` + // column is the source of truth"). Setting it on the + // encoder path would mint a row shape the RFC says + // cannot exist. + // - RFC 0005 §3.3: the `body` column for structured + // rows MUST hold canonical JSON. A + // `format!("{any_value:?}")` fallback would silently + // write spec-violating bytes into a §3.3-governed + // column — the masquerading-as-JSON failure mode the + // writer's prior `StructuredBodyNotYetCanonical` + // rejection prevented. + // + // `canonical::encode_any_value` is infallible on every + // `AnyValue` value the type system admits. + // `opentelemetry-proto`'s `with-serde` ships custom + // serializers (see `proto.rs::serializer_f64`) that + // emit `"NaN"` / `"Infinity"` / `"-Infinity"` strings + // per the proto3 JSON spec rather than letting + // `serde_json`'s default `f64` path emit `null` — which + // also covers the only failure mode review raised on + // an earlier revision. The recursive variants + // (`ArrayValue`, `KvlistValue`) bottom out in the same + // primitive serializers, so encode failure is + // unreachable here. `.expect` documents the contract + // rather than swallowing a `Result` we never inspect. + let bytes = ourios_core::otlp::canonical::encode_any_value(any_value) + .expect("RFC 0005 §3.3 encoder is infallible for any spec-compliant AnyValue"); let mut rec = Self::record_envelope(record, BodyKind::Structured); rec.template_id = template_id; rec.template_version = 1; rec.confidence = 1.0; - rec.body = Some(stored_body); + rec.body = Some(String::from_utf8(bytes).expect("serde_json emits valid UTF-8")); self.emit_record(rec); template_id @@ -3035,6 +3056,59 @@ mod tests { assert_eq!(cluster.template_count(&t), 2); } + /// Pin the exact RFC 0005 §3.3 canonical-JSON bytes the + /// miner stores in `MinedRecord.body` for a structured + /// row. Catches a regression to debug formatting (the + /// prior `format!("{any_value:?}")` placeholder), AND + /// catches an `opentelemetry-proto` upgrade that breaks + /// the OTLP-JSON spec mapping (camelCase, string-encoded + /// `i64`, base64 bytes). A non-trivial `AnyValue` exercises + /// the recursive `KvlistValue` path through the encoder. + #[test] + fn structured_body_is_stored_as_otlp_canonical_json() { + use ourios_core::otlp::{KeyValue as ProtoKv, KeyValueList}; + let records = SharedRecordSink::new(); + let mut cluster = + MinerCluster::new(MinerConfig::default()).with_record_sink(Box::new(records.clone())); + let av = AnyValue { + value: Some(AvValue::KvlistValue(KeyValueList { + values: vec![ProtoKv { + key: "user.id".to_string(), + value: Some(AnyValue { + value: Some(AvValue::IntValue(42)), + }), + ..Default::default() + }], + })), + }; + let record = OtlpLogRecord { + tenant_id: TenantId::new("tenant-x"), + severity_number: 9, + scope_name: Some("bench.scope".to_string()), + body: Some(Body::Structured(av)), + ..Default::default() + }; + cluster.ingest(&record); + let emitted = records.drain(); + assert_eq!(emitted.len(), 1); + let body = emitted[0].body.as_deref().expect("structured body is Some"); + // Pinned canonical form per the proto3 JSON spec + // mapping: camelCase keys, `i64` as a quoted string, + // recursive `kvlistValue` shape. The opentelemetry-proto + // `with-serde` derives emit fields in struct-definition + // order, which is what serde_json::to_vec produces + // deterministically — RFC0006.7 reproducibility relies + // on this same byte stability. + assert_eq!( + body, r#"{"kvlistValue":{"values":[{"key":"user.id","value":{"intValue":"42"}}]}}"#, + "miner must store RFC 0005 §3.3 canonical JSON, not a debug rendering", + ); + assert!( + !emitted[0].lossy_flag, + "RFC 0001 §6.1: lossy_flag is always false on BodyKind::Structured", + ); + } + #[test] fn structured_body_with_scope_none_is_its_own_bucket() { let mut cluster = MinerCluster::new(MinerConfig::default()); diff --git a/crates/ourios-parquet/src/reader.rs b/crates/ourios-parquet/src/reader.rs index 711a324aa..00e75c383 100644 --- a/crates/ourios-parquet/src/reader.rs +++ b/crates/ourios-parquet/src/reader.rs @@ -38,7 +38,6 @@ use arrow_array::types::{ }; use arrow_array::{Array, RecordBatch, StructArray}; use ourios_core::audit::ParamType; -use ourios_core::otlp::KeyValue; use ourios_core::record::{BodyKind, MinedRecord, Param}; use ourios_core::tenant::TenantId; use parquet::arrow::arrow_reader::{ParquetRecordBatchReader, ParquetRecordBatchReaderBuilder}; @@ -137,11 +136,10 @@ impl Reader { /// data shape doesn't match what the reader expects /// (logical-type mismatch, unexpected null on a REQUIRED /// column, etc.). - /// - [`ReaderError::AttributesNotYetDecoded`] when the - /// file contains a non-empty `attributes` / - /// `resource_attributes` JSON string. The canonical-JSON - /// decoder is symmetric to the writer's encoder — both - /// are deferred to the RFC 0005 §3.3 canonicalisation PR. + /// - [`ReaderError::AttributeDecode`] when an + /// `attributes` / `resource_attributes` column's bytes + /// fail the RFC 0005 §3.3 canonical-JSON decode (corrupt + /// file or foreign-producer bytes). /// - [`ReaderError::PartitionMismatch`] when a row's /// derived partition disagrees with the writer-side /// partition supplied to [`Self::open_partition`]. @@ -159,7 +157,7 @@ impl Reader { // `From for ParquetError` lets us // route everything through the same variant. let batch = batch.map_err(|e| ReaderError::Parquet(e.into()))?; - let records = batch_to_mined_records(&batch)?; + let records = batch_to_mined_records(&batch, row_offset)?; if let Some(p) = &partition { for (idx_in_batch, r) in records.iter().enumerate() { validate_row_vs_partition(r, p, row_offset + idx_in_batch, &file_path)?; @@ -195,13 +193,16 @@ pub enum ReaderError { column: &'static str, detail: String, }, - /// Non-empty `attributes` or `resource_attributes` column. - /// The canonical-JSON decoder is deferred to the RFC 0005 - /// §3.3 canonicalisation PR (symmetric to the writer's - /// `AttributesNotYetEncoded`). - AttributesNotYetDecoded { + /// An `attributes` / `resource_attributes` column carried + /// bytes that the RFC 0005 §3.3 canonical-JSON decoder + /// couldn't parse. Treat as file corruption — the writer + /// only emits encoder-produced canonical bytes — or as a + /// foreign producer that doesn't honour the §3.3 spec. + /// Carries the row index for diagnostics. + AttributeDecode { column: &'static str, - encoded: String, + row_index: usize, + source: ourios_core::otlp::canonical::CanonicalJsonError, }, /// A row's derived partition disagrees with the partition /// supplied to [`Reader::open_partition`]. RFC 0005 §3.9 @@ -235,11 +236,15 @@ impl fmt::Display for ReaderError { Self::Conversion { column, detail } => { write!(f, "column `{column}` conversion failed: {detail}") } - Self::AttributesNotYetDecoded { column, encoded } => write!( + Self::AttributeDecode { + column, + row_index, + source, + } => write!( f, - "column `{column}` carries non-empty canonical JSON ({encoded:?}) but the \ - RFC 0005 §3.3 decoder is deferred to the canonicalisation PR (symmetric to \ - the writer's `AttributesNotYetEncoded`)", + "column `{column}` row {row_index}: RFC 0005 §3.3 canonical-JSON decode \ + failed: {source} (the writer only emits encoder-produced bytes; either the \ + file is corrupt or a foreign producer wrote it)", ), Self::PartitionMismatch { row_index, @@ -277,8 +282,8 @@ impl std::error::Error for ReaderError { Self::TimestampOverflow(e) => Some(e), Self::MissingRequiredColumn { .. } | Self::Conversion { .. } - | Self::AttributesNotYetDecoded { .. } | Self::PartitionMismatch { .. } => None, + Self::AttributeDecode { source, .. } => Some(source), } } } @@ -304,7 +309,15 @@ fn validate_row_vs_partition( /// Convert one Arrow `RecordBatch` to a `Vec` per /// RFC 0005 §3.2. Handles the §3.9 "missing OPTIONAL column → /// `None`" rule by checking column presence before unpacking. -fn batch_to_mined_records(batch: &RecordBatch) -> Result, ReaderError> { +fn batch_to_mined_records( + batch: &RecordBatch, + // File-global row offset of `batch`'s first row, threaded + // from the caller so per-row diagnostics report stable + // indices across multi-batch files. Per-batch + // `enumerate()` would reset to 0 every batch and produce + // ambiguous row numbers in `AttributeDecode` / similar. + row_offset: usize, +) -> Result, ReaderError> { let n = batch.num_rows(); let mut records: Vec = Vec::with_capacity(n); @@ -341,20 +354,33 @@ fn batch_to_mined_records(batch: &RecordBatch) -> Result, Reade let separators_lists = decode_separators_column(batch)?; for i in 0..n { + // Empty-list short-circuit mirrors the writer's + // `append_attributes` — avoids the encoder round-trip + // on every clean-attach record (the common case). let attrs_str = attributes[i].as_str(); - if attrs_str != "[]" { - return Err(ReaderError::AttributesNotYetDecoded { - column: columns::ATTRIBUTES, - encoded: attrs_str.to_string(), - }); - } + let decoded_attrs = if attrs_str == "[]" { + Vec::new() + } else { + ourios_core::otlp::canonical::decode_attributes(attrs_str.as_bytes()).map_err( + |source| ReaderError::AttributeDecode { + column: columns::ATTRIBUTES, + row_index: row_offset + i, + source, + }, + )? + }; let res_str = resource_attributes[i].as_str(); - if res_str != "[]" { - return Err(ReaderError::AttributesNotYetDecoded { - column: columns::RESOURCE_ATTRIBUTES, - encoded: res_str.to_string(), - }); - } + let decoded_resource = if res_str == "[]" { + Vec::new() + } else { + ourios_core::otlp::canonical::decode_attributes(res_str.as_bytes()).map_err( + |source| ReaderError::AttributeDecode { + column: columns::RESOURCE_ATTRIBUTES, + row_index: row_offset + i, + source, + }, + )? + }; let t_ns = u64::try_from(time_unix_nano[i]).map_err(|_| ReaderError::Conversion { column: columns::TIME_UNIX_NANO, @@ -397,9 +423,9 @@ fn batch_to_mined_records(batch: &RecordBatch) -> Result, Reade scope_version: scope_version.as_ref().and_then(|c| c[i].clone()), time_unix_nano: t_ns, observed_time_unix_nano: observed_t, - attributes: Vec::new(), + attributes: decoded_attrs, dropped_attributes_count: dropped_attributes_count[i], - resource_attributes: Vec::::new(), + resource_attributes: decoded_resource, trace_id: trace_id.as_ref().and_then(|c| c[i]), span_id: span_id.as_ref().and_then(|c| c[i]), flags: flags[i], diff --git a/crates/ourios-parquet/src/record_batch.rs b/crates/ourios-parquet/src/record_batch.rs index 0c39318bf..9ecf1c3b5 100644 --- a/crates/ourios-parquet/src/record_batch.rs +++ b/crates/ourios-parquet/src/record_batch.rs @@ -7,20 +7,20 @@ //! the declared schema. //! //! **`AnyValue` → canonical JSON.** RFC 0005 §3.3 mandates -//! OTLP-canonical JSON for the `attributes`, `resource_attributes`, -//! and (when `body_kind = Structured`) `body` columns. The -//! current builder handles **only the empty case** for the -//! `KeyValue` lists — it emits the literal `"[]"` directly into -//! the column (the RFC 0005 §3.2 `Vec::new()` ↔ `[]` rule) — and -//! returns [`BatchError::AttributesNotYetEncoded`] on any -//! non-empty input. Corpus / bench inputs today carry empty -//! attributes; the RFC 0003 receiver is what populates them, and -//! the canonicalisation PR named in the PR-E1 breadcrumb on -//! [`ourios_core::otlp::Body::Structured`] is the one that fills -//! in the proto3-JSON-with-OTLP-overrides encoder. Surfacing a -//! structured error rather than panicking (or emitting non-JSON -//! `Debug` bytes masquerading as JSON) lets the writer fail a -//! batch gracefully without crashing the ingest process. +//! OTLP-canonical JSON for the `attributes`, +//! `resource_attributes`, and (when `body_kind = Structured`) +//! `body` columns. Encoding goes through +//! [`ourios_core::otlp::canonical`], which wraps +//! `opentelemetry-proto`'s `with-serde` derives — the same spec +//! mapping rotel's OTLP HTTP receiver uses on +//! `ExportLogsServiceRequest`. The empty `Vec::new()` case +//! still short-circuits to the literal `"[]"` (the RFC 0005 §3.2 +//! `Vec::new()` ↔ `[]` rule, no per-row encoder allocation on +//! the clean-attach hot path); non-empty inputs go through +//! `canonical::encode_attributes`. For `body_kind = Structured` +//! rows, the miner's `ingest_structured` has already encoded +//! the body into the canonical bytes (`MinedRecord.body` carries +//! the bytes verbatim), so the writer just appends them. use std::fmt; use std::sync::Arc; @@ -70,12 +70,24 @@ pub enum BatchError { TimestampOverflow { field: &'static str, value: u64 }, /// A record carried a non-empty `attributes` or /// `resource_attributes` `Vec`. The canonical-JSON - /// encoder is deferred to the RFC 0005 §3.3 canonicalisation - /// PR (see the PR-E1 breadcrumb on - /// [`ourios_core::otlp::Body::Structured`]); until then the - /// writer returns this error rather than crashing the ingest - /// process. Carries the column name and entry count. - AttributesNotYetEncoded { column: &'static str, count: usize }, + /// An `attributes` / `resource_attributes` `Vec` + /// failed RFC 0005 §3.3 canonical-JSON encoding. In + /// principle unreachable on every value the type system + /// admits — `opentelemetry-proto`'s `with-serde` ships + /// proto3-JSON-compliant primitive serializers + /// (`f64::NAN` → `"NaN"` string, `i64` → string-encoded + /// number, `bytes` → base64) so the recursive variants + /// bottom out at infallible primitives. The variant + /// survives for `Result`-symmetry with the reader's + /// `AttributeDecode` and as a defence-in-depth surface if + /// a future `opentelemetry-proto` release breaks that + /// contract. Carries the column name, entry count, and + /// the underlying serde error. + AttributeEncode { + column: &'static str, + 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, @@ -86,19 +98,6 @@ pub enum BatchError { /// rejects these records rather than corrupting the /// `body_kind` semantics. UnsupportedAbsentBody, - /// A record carried `body_kind = Structured`. RFC 0005 §3.3 - /// requires the `body` column for these rows to hold - /// OTLP-canonical JSON, but the miner today populates - /// `MinedRecord.body` with an interim `Debug` rendering - /// (see the PR-E1 breadcrumb on - /// [`ourios_core::otlp::Body::Structured`]). Writing the - /// interim bytes would silently store non-canonical / - /// non-JSON content into a §3.3-governed column. Symmetric - /// to [`Self::AttributesNotYetEncoded`]: the writer fails - /// the batch until the canonicalisation PR replaces the - /// miner's `format!("{any_value:?}")` call site with a real - /// proto3-JSON-with-OTLP-overrides encoder. - StructuredBodyNotYetCanonical, /// 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 @@ -138,11 +137,16 @@ impl fmt::Display for BatchError { f, "{field} = {value} exceeds i64::MAX (RFC 0005 §3.2 u64→i64 overflow contract)", ), - Self::AttributesNotYetEncoded { column, count } => write!( + Self::AttributeEncode { + column, + count, + source, + } => write!( f, - "{column}: canonical-JSON encoding of {count} KeyValue entries is deferred to \ - the RFC 0005 §3.3 canonicalisation PR (corpus / bench inputs today carry \ - empty attributes; the RFC 0003 receiver is what populates them)", + "{column}: RFC 0005 §3.3 canonical-JSON encode of {count} KeyValue entries \ + failed: {source} (in principle unreachable — `opentelemetry-proto`'s \ + `with-serde` derives are infallible on every spec-compliant `AnyValue`; \ + this means an `opentelemetry-proto` upgrade broke that contract)", ), Self::UnsupportedAbsentBody => write!( f, @@ -151,14 +155,6 @@ impl fmt::Display for BatchError { 1=Structured); a future RFC 0005 amendment is required to represent this \ in the schema", ), - Self::StructuredBodyNotYetCanonical => write!( - f, - "record carries body_kind = Structured but the body column would receive the \ - miner's interim Debug rendering rather than RFC 0005 §3.3's OTLP-canonical \ - JSON; the canonicalisation PR (see PR-E1 breadcrumb on \ - ourios_core::otlp::Body::Structured) must land before structured rows can \ - be written", - ), Self::InvalidSeparatorsForString { expected_at_least, actual, @@ -185,11 +181,10 @@ impl std::error::Error for BatchError { fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { match self { Self::TimestampOverflow { .. } - | Self::AttributesNotYetEncoded { .. } | Self::UnsupportedAbsentBody - | Self::StructuredBodyNotYetCanonical | Self::InvalidSeparatorsForString { .. } | Self::MissingBodyForLossyString => None, + Self::AttributeEncode { source, .. } => Some(source), Self::Arrow(e) => Some(e), } } @@ -333,18 +328,16 @@ impl Builders { append_option_str(&mut self.event_name, r.event_name.as_deref()); self.body_kind.append_value(body_kind_ordinal(r.body_kind)?); - // RFC 0005 §3.3: when `body_kind = Structured`, the body - // column carries OTLP-canonical JSON. The miner today - // populates `body` with `format!("{any_value:?}")` per - // the PR-E1 breadcrumb on `ourios_core::otlp::Body:: - // Structured` — that's *not* canonical JSON, so writing - // it would store non-conforming bytes for a §3.3- - // governed column. Reject these records until the - // canonicalisation PR lands (symmetric to the - // `AttributesNotYetEncoded` deferral above). - if r.body_kind == BodyKind::Structured { - return Err(BatchError::StructuredBodyNotYetCanonical); - } + // RFC 0005 §3.3: when `body_kind = Structured`, the + // body column carries OTLP-canonical JSON — the bytes + // the miner has already encoded via + // `ourios_core::otlp::canonical::encode_any_value` (the + // miner's `ingest_structured` writes them into + // `MinedRecord.body` directly, so the writer just + // appends them verbatim). For `String` rows, the body + // 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). match r.body.as_deref() { Some(s) => self.body.append_value(s.as_bytes()), None => self.body.append_null(), @@ -504,18 +497,15 @@ fn append_separators(builder: &mut GenericListBuilder, separ /// - Empty input → appends the literal `"[]"` directly into the /// builder (RFC 0005 §3.2's `Vec::new()` ↔ `[]` round-trip /// rule). The `&'static str` argument means no per-row `String` -/// allocation — important on the hot path where corpus / bench -/// inputs today carry empty attributes for every record. -/// - Non-empty input → returns -/// [`BatchError::AttributesNotYetEncoded`] so the writer fails -/// the batch gracefully rather than crashing the ingest process. -/// The original cut emitted `format!("{attrs:?}")` (Rust `Debug` -/// rendering) which is *not* valid JSON; the structured error -/// surfaces the gap loudly without inviting downstream code to -/// silently store non-JSON masquerading as JSON. RFC 0005 §3.3 -/// names the normative encoding (proto3 JSON with OTLP -/// overrides); implementing it is the canonicalisation PR's -/// job. +/// allocation — important on the hot path where the empty +/// case is the common one for clean-attach text rows. +/// - Non-empty input → encoded via +/// [`ourios_core::otlp::canonical::encode_attributes`] (the +/// RFC 0005 §3.3 spec mapping, single-sourced through +/// `opentelemetry-proto`'s `with-serde` derives). Encode +/// failures (pathological inputs like a non-finite double +/// that the wire-decode receiver doesn't pre-filter) surface +/// as [`BatchError::AttributeEncode`]. fn append_attributes( b: &mut StringBuilder, column: &'static str, @@ -525,10 +515,20 @@ fn append_attributes( b.append_value("[]"); return Ok(()); } - Err(BatchError::AttributesNotYetEncoded { - column, - count: attrs.len(), - }) + let bytes = ourios_core::otlp::canonical::encode_attributes(attrs).map_err(|source| { + BatchError::AttributeEncode { + column, + count: attrs.len(), + source, + } + })?; + // `serde_json` emits valid UTF-8 by construction (Rust + // strings → JSON), so the `from_utf8` is infallible — + // `expect` documents the invariant rather than hiding it + // behind a `_ = ...` discard. + let as_str = std::str::from_utf8(&bytes).expect("serde_json output is valid UTF-8"); + b.append_value(as_str); + Ok(()) } #[cfg(test)] @@ -572,9 +572,10 @@ mod tests { } } - /// Sanity: the empty-attributes path serialises to literal `[]` - /// (the §3.2 `Vec::new()` ↔ `[]` round-trip rule) and does not - /// hit the `AttributesNotYetEncoded` branch. + /// Sanity: the empty-attributes path short-circuits to the + /// literal `[]` byte sequence (the §3.2 `Vec::new()` ↔ `[]` + /// rule) without invoking the encoder — keeps the + /// clean-attach hot path allocation-free. #[test] fn empty_attributes_serialise_to_open_bracket_close_bracket() { let batch = mined_records_to_batch(&[empty_record()]).expect("batch builds"); @@ -589,29 +590,33 @@ mod tests { assert_eq!(res.value(0), "[]"); } - /// Non-empty attributes return `AttributesNotYetEncoded` - /// rather than panicking via `unimplemented!()`. Pins the - /// graceful-error contract until the RFC 0005 §3.3 - /// canonicalisation PR replaces this branch with a real - /// encoder. + /// Non-empty attributes flow through the RFC 0005 §3.3 + /// canonical-JSON encoder + /// (`ourios_core::otlp::canonical::encode_attributes`) and + /// land as valid OTLP-JSON bytes in the column — + /// `decode_attributes(stored) == attrs` round-trips. Pins + /// the encoder integration that replaced the + /// `AttributesNotYetEncoded` deferral. #[test] - fn non_empty_attributes_returns_not_yet_encoded_error() { - let mut rec = empty_record(); - rec.attributes = vec![KeyValue { + fn non_empty_attributes_encode_to_canonical_json_round_trip() { + let attrs = vec![KeyValue { key: "client.address".to_string(), value: Some(AnyValue { value: Some(any_value::Value::StringValue("10.0.0.1".to_string())), }), ..KeyValue::default() }]; - let err = mined_records_to_batch(&[rec]).expect_err("non-empty attrs must error"); - match err { - BatchError::AttributesNotYetEncoded { column, count } => { - assert_eq!(column, "attributes"); - assert_eq!(count, 1); - } - other => panic!("expected AttributesNotYetEncoded, got {other:?}"), - } + let mut rec = empty_record(); + rec.attributes = attrs.clone(); + let batch = mined_records_to_batch(&[rec]).expect("batch encodes attributes"); + let attrs_idx = batch.schema().index_of(crate::columns::ATTRIBUTES).unwrap(); + let stored = batch.column(attrs_idx).as_string::().value(0); + let decoded = ourios_core::otlp::canonical::decode_attributes(stored.as_bytes()) + .expect("stored bytes are canonical JSON"); + assert_eq!( + decoded, attrs, + "encode → decode must round-trip the in-memory KeyValue list", + ); } /// RFC 0005 §3.2 invariant: clean-attach `body_kind = String` @@ -699,20 +704,25 @@ mod tests { mined_records_to_batch(&[rec]).expect("lossy_flag carve-out must not error"); } - /// `BodyKind::Structured` rows can't yet be written - /// faithfully — the miner stores an interim Debug - /// rendering of the `AnyValue` rather than canonical JSON. - /// The writer rejects until the canonicalisation PR lands. + /// `BodyKind::Structured` rows now write — the miner's + /// `ingest_structured` populates `MinedRecord.body` with + /// RFC 0005 §3.3 canonical JSON, and the writer appends + /// those bytes verbatim. Pins that the body column for a + /// structured row carries exactly the producer's bytes + /// (the §3.3 "what we stored is what we return" + /// reconstruction guarantee). #[test] - fn structured_body_kind_returns_not_yet_canonical_error() { + fn structured_body_kind_appends_producer_bytes_verbatim() { let mut rec = empty_record(); rec.body_kind = BodyKind::Structured; - rec.body = Some("{\"placeholder\":true}".to_string()); - let err = mined_records_to_batch(&[rec]).expect_err("structured body must error"); - assert!( - matches!(err, BatchError::StructuredBodyNotYetCanonical), - "expected StructuredBodyNotYetCanonical, got {err:?}", - ); + // Canonical JSON the miner would produce for an + // `AnyValue { value: Some(IntValue(42)) }`. + let canonical = "{\"intValue\":\"42\"}"; + rec.body = Some(canonical.to_string()); + let batch = mined_records_to_batch(&[rec]).expect("structured body must write"); + let body_idx = batch.schema().index_of(crate::columns::BODY).unwrap(); + let stored = batch.column(body_idx).as_binary::(); + assert_eq!(stored.value(0), canonical.as_bytes()); } /// `BodyKind::Absent` is not representable in the §3.2 @@ -730,27 +740,29 @@ mod tests { ); } - /// Same contract on the `resource_attributes` side: empty in - /// the primary `attributes` column, populated in - /// `resource_attributes`, still errors with the right column - /// name. + /// Same canonical-encode path on the `resource_attributes` + /// column: populated input round-trips through + /// `encode_attributes` / `decode_attributes` and lands in + /// the matching column. #[test] - fn non_empty_resource_attributes_errors_on_correct_column() { - let mut rec = empty_record(); - rec.resource_attributes = vec![KeyValue { + fn non_empty_resource_attributes_round_trip() { + let resource_attrs = vec![KeyValue { key: "service.name".to_string(), value: Some(AnyValue { value: Some(any_value::Value::StringValue("ourios".to_string())), }), ..KeyValue::default() }]; - let err = mined_records_to_batch(&[rec]).expect_err("non-empty resource attrs must error"); - match err { - BatchError::AttributesNotYetEncoded { column, count } => { - assert_eq!(column, "resource_attributes"); - assert_eq!(count, 1); - } - other => panic!("expected AttributesNotYetEncoded, got {other:?}"), - } + let mut rec = empty_record(); + rec.resource_attributes = resource_attrs.clone(); + let batch = mined_records_to_batch(&[rec]).expect("resource attrs must encode"); + let resource_idx = batch + .schema() + .index_of(crate::columns::RESOURCE_ATTRIBUTES) + .unwrap(); + let stored = batch.column(resource_idx).as_string::().value(0); + let decoded = ourios_core::otlp::canonical::decode_attributes(stored.as_bytes()) + .expect("stored bytes are canonical JSON"); + assert_eq!(decoded, resource_attrs); } }