feat(ingester): materialize LogRecord into OtlpLogRecord (RFC0003.7–.10)

crates/ourios-ingester/src/lib.rs

@@ -7,9 +7,11 @@
 //! - **OTLP receiver** (RFC 0003, `red`, greening) — the gRPC/HTTP
 //!   ingest front door + mining pipeline. The §5 acceptance criteria
 //!   (RFC0003.1–.15) are enumerated as `tests/rfc0003_*`; the green
-//!   slices flip them one §8 group at a time. [`receiver::decode`] is
-//!   the first to land — the §6.2 wire-decode layer (RFC0003.5) — with
-//!   tenant fan-out, transports, and the WAL-before-ack path to follow.
+//!   slices flip them one §8 group at a time. Landed: [`receiver::decode`]
+//!   (§6.2 wire decode — protobuf + OTLP/JSON, RFC0003.5/.6) and
+//!   [`receiver::materialize`] (§6.1 `LogRecord` → `OtlpLogRecord`,
+//!   RFC0003.7–.10). Tenant fan-out, the live transports, and the
+//!   WAL-before-ack path follow.
 //! - **WAL-before-ack** (RFC 0008 / `CLAUDE.md` §3.4) — durability
 //!   before acknowledgement, via the shipped `ourios-wal`. Wired into
 //!   the ingest path once the receiver lands; not exercised here.

crates/ourios-ingester/src/receiver.rs

@@ -11,7 +11,13 @@
 //!   live `tonic`/`axum` listener yet: the transports hand their decoded
 //!   payload to this same layer, so decode is specified and tested
 //!   before the framing is wired.
+//! - [`materialize`] — the §6.1 step 2–3 mapping from a decoded
+//!   `LogRecord` to the flat `OtlpLogRecord` the miner consumes (body
+//!   fork + empty-sentinel narrowing). Tenant derivation + fan-out
+//!   (RFC0003.3) layer on top of it next.
 
 pub mod decode;
+pub mod materialize;
 
 pub use decode::{DecodeError, decode_json, decode_protobuf};
+pub use materialize::materialize_record;

crates/ourios-ingester/src/receiver/materialize.rs

@@ -0,0 +1,94 @@
+//! Record materialisation (RFC 0003 §6.1 steps 2–3).
+//!
+//! Maps one decoded OTLP `LogRecord` to the flat
+//! [`OtlpLogRecord`] the miner consumes, inheriting the enclosing
+//! `Resource` attributes and `InstrumentationScope` name/version so
+//! downstream code never walks back up the OTLP hierarchy. The mapping
+//! narrows proto's "empty value = absence" sentinels into a single
+//! `Option`/`None` at this boundary (RFC0003.9), reflects
+//! `dropped_attributes_count` verbatim (RFC0003.10), and forks the body
+//! via [`Body::from_any_value`] (RFC0003.7/.8).
+//!
+//! Tenant derivation (RFC0003.3, per `ResourceLogs`) is *not* done here:
+//! `materialize_record` takes the resolved `tenant_id` as a parameter,
+//! so the fan-out slice supplies it.
+
+use opentelemetry_proto::tonic::common::v1::{InstrumentationScope, KeyValue};
+use opentelemetry_proto::tonic::logs::v1::LogRecord;
+use ourios_core::otlp::{Body, OtlpLogRecord};
+use ourios_core::tenant::TenantId;
+
+/// Materialise one decoded `LogRecord` into an [`OtlpLogRecord`] under
+/// `tenant_id`, inheriting `resource_attributes` and the enclosing
+/// `scope`.
+///
+/// Consumes `record` so its body `AnyValue` and per-record attributes
+/// *move* into the result — no deep clone of structured trees, per the
+/// §6.4 amendment. `resource_attributes` and the scope name/version are
+/// shared across the records under a `ResourceLogs`/`ScopeLogs`, so they
+/// are cloned per record.
+#[must_use]
+pub fn materialize_record(
+    record: LogRecord,
+    resource_attributes: &[KeyValue],
+    scope: Option<&InstrumentationScope>,
+    tenant_id: TenantId,
+) -> OtlpLogRecord {
+    OtlpLogRecord {
+        tenant_id,
+        // Event time: `0` = unknown per the OTLP spec, kept as `0`
+        // (a `u64`, not narrowed — absence and "epoch 0" are the same
+        // wire value and the schema models it as a plain `u64`).
+        time_unix_nano: record.time_unix_nano,
+        // Collector observation time: wire `0` = unset → `None`
+        // (RFC0003.9; the `Option<u64>` typing exists for this).
+        observed_time_unix_nano: nonzero(record.observed_time_unix_nano),
+        // `UNSPECIFIED` (`0`) is an explicit OTLP value, kept as `0`
+        // (RFC0003.9); proto's `i32` is narrowed to the schema's
+        // documented `0..=24` `u8` range — see `severity_to_u8`.
+        severity_number: severity_to_u8(record.severity_number),
+        severity_text: nonempty(record.severity_text),
+        scope_name: scope.and_then(|s| (!s.name.is_empty()).then(|| s.name.clone())),
+        scope_version: scope.and_then(|s| (!s.version.is_empty()).then(|| s.version.clone())),
+        attributes: record.attributes,
+        // Reflected verbatim from the wire, never recomputed (RFC0003.10).
+        dropped_attributes_count: record.dropped_attributes_count,
+        resource_attributes: resource_attributes.to_vec(),
+        trace_id: fixed_len(&record.trace_id),
+        span_id: fixed_len(&record.span_id),
+        flags: record.flags,
+        event_name: nonempty(record.event_name),
+        // `string_value` → mining path (`Body::String`), every other
+        // variant → `Body::Structured` verbatim (RFC0003.7/.8); `None`
+        // when the wire delivered no body.
+        body: record.body.and_then(Body::from_any_value),
+    }
+}
+
+/// Narrow proto's `i32` `severity_number` to the schema's `u8`. Valid
+/// OTLP severity is `0..=24` (`0` = UNSPECIFIED); any value outside that
+/// range — invalid-but-`u8`-representable (`25..=255`), negative, or
+/// `> 255` — maps to `0`/UNSPECIFIED, so the `OtlpLogRecord` contract the
+/// miner's template key and the Parquet schema rely on holds at this
+/// boundary.
+fn severity_to_u8(n: i32) -> u8 {
+    u8::try_from(n).ok().filter(|v| *v <= 24).unwrap_or(0)
+}
+
+/// Proto scalar `0` → `None`, else `Some` — the RFC0003.9 narrowing of a
+/// "0 = unset" wire sentinel.
+fn nonzero(v: u64) -> Option<u64> {
+    (v != 0).then_some(v)
+}
+
+/// Proto empty string → `None`, else `Some` — narrows the "empty = unset"
+/// sentinel proto uses for optional strings.
+fn nonempty(s: String) -> Option<String> {
+    (!s.is_empty()).then_some(s)
+}
+
+/// A proto `bytes` id (`trace_id` / `span_id`): exactly `N` bytes →
+/// `Some`; empty (absent) or any other length (malformed) → `None`.
+fn fixed_len<const N: usize>(bytes: &[u8]) -> Option<[u8; N]> {
+    <[u8; N]>::try_from(bytes).ok()
+}

crates/ourios-ingester/tests/rfc0003_10_dropped_attributes_count.rs

@@ -1,16 +1,23 @@
 //! RFC0003.10 — `dropped_attributes_count` preserved verbatim.
 //!
-//! Red gate (`specified → red`): `#[ignore]`'d until the receiver
-//! lands.
+//! The receiver reflects the wire-level `dropped_attributes_count` onto
+//! the materialised record exactly, and never recomputes it.
+
+use opentelemetry_proto::tonic::logs::v1::LogRecord;
+use ourios_core::tenant::TenantId;
+use ourios_ingester::receiver::materialize_record;
 
 /// Scenario RFC0003.10 — `dropped_attributes_count` preserved verbatim.
 /// See `docs/rfcs/0003-otlp-receiver.md` §5.
-#[ignore = "RFC 0003 red gate — implementation pending (RFC0003.10)"]
 #[test]
-fn rfc0003_10_dropped_attributes_count_is_reflected_not_recomputed() {
-    unimplemented!(
-        "RFC0003.10 — a wire dropped_attributes_count of 42 yields \
-         OtlpLogRecord.dropped_attributes_count == 42 exactly; the receiver \
-         reflects the wire claim and never recomputes it."
+fn rfc0003_10_dropped_attributes_count_is_reflected_verbatim() {
+    let record = LogRecord {
+        dropped_attributes_count: 42,
+        ..Default::default()
+    };
+    let materialized = materialize_record(record, &[], None, TenantId::new("tenant-a"));
+    assert_eq!(
+        materialized.dropped_attributes_count, 42,
+        "dropped_attributes_count is reflected from the wire, never recomputed",
     );
 }

crates/ourios-ingester/tests/rfc0003_7_body_structured_verbatim.rs

@@ -1,17 +1,54 @@
 //! RFC0003.7 — `Body::Structured` carries the decoded `AnyValue` verbatim.
 //!
-//! Red gate (`specified → red`): `#[ignore]`'d until the receiver
-//! lands.
+//! Materialisation routes every non-`string_value` `AnyValue` to
+//! `Body::Structured`, carrying the decoded value with no
+//! canonicalisation, reshape, or dropped fields (the §6.4 amendment).
+
+use opentelemetry_proto::tonic::common::v1::any_value::Value;
+use opentelemetry_proto::tonic::common::v1::{AnyValue, ArrayValue, KeyValue, KeyValueList};
+use opentelemetry_proto::tonic::logs::v1::LogRecord;
+use ourios_core::otlp::Body;
+use ourios_core::tenant::TenantId;
+use ourios_ingester::receiver::materialize_record;
 
 /// Scenario RFC0003.7 — `Body::Structured` carries the decoded `AnyValue` verbatim.
 /// See `docs/rfcs/0003-otlp-receiver.md` §5.
-#[ignore = "RFC 0003 red gate — implementation pending (RFC0003.7)"]
 #[test]
-fn rfc0003_7_structured_body_reaches_miner_as_verbatim_anyvalue() {
-    unimplemented!(
-        "RFC0003.7 — a structured body reaches the miner as \
-         Body::Structured(AnyValue) structurally equal to the wire AnyValue (no \
-         canonicalisation, no reshape, no dropped fields), and the same equality \
-         holds across all three transports."
-    );
+fn rfc0003_7_structured_body_is_carried_verbatim() {
+    let non_string_variants = [
+        Value::BoolValue(true),
+        Value::IntValue(-42),
+        Value::DoubleValue(1.5),
+        Value::BytesValue(vec![0xDE, 0xAD, 0xBE, 0xEF]),
+        Value::ArrayValue(ArrayValue {
+            values: vec![AnyValue {
+                value: Some(Value::IntValue(7)),
+            }],
+        }),
+        Value::KvlistValue(KeyValueList {
+            values: vec![KeyValue {
+                key: "k".to_owned(),
+                value: Some(AnyValue {
+                    value: Some(Value::StringValue("v".to_owned())),
+                }),
+                ..Default::default()
+            }],
+        }),
+    ];
+
+    for variant in non_string_variants {
+        let any_value = AnyValue {
+            value: Some(variant),
+        };
+        let record = LogRecord {
+            body: Some(any_value.clone()),
+            ..Default::default()
+        };
+        let materialized = materialize_record(record, &[], None, TenantId::new("tenant-a"));
+        assert_eq!(
+            materialized.body,
+            Some(Body::Structured(any_value)),
+            "a non-string AnyValue reaches the miner as Body::Structured, verbatim",
+        );
+    }
 }

crates/ourios-ingester/tests/rfc0003_8_body_string_lraw.rs

@@ -1,17 +1,46 @@
 //! RFC0003.8 — `Body::String` reaches the miner as the unwrapped `L_raw`.
 //!
-//! Red gate (`specified → red`): `#[ignore]`'d until the receiver
-//! lands.
+//! A `string_value` body is unwrapped to `Body::String(s)` with the
+//! original UTF-8 verbatim (no wrapping, quoting, or escaping); an
+//! absent body stays `None`.
+
+use opentelemetry_proto::tonic::common::v1::AnyValue;
+use opentelemetry_proto::tonic::common::v1::any_value::Value;
+use opentelemetry_proto::tonic::logs::v1::LogRecord;
+use ourios_core::otlp::Body;
+use ourios_core::tenant::TenantId;
+use ourios_ingester::receiver::materialize_record;
 
 /// Scenario RFC0003.8 — `Body::String` reaches the miner as the unwrapped `L_raw`.
 /// See `docs/rfcs/0003-otlp-receiver.md` §5.
-#[ignore = "RFC 0003 red gate — implementation pending (RFC0003.8)"]
 #[test]
-fn rfc0003_8_string_body_reaches_miner_unwrapped() {
-    unimplemented!(
-        "RFC0003.8 — a string body becomes OtlpLogRecord.body = \
-         Some(Body::String(s)) where s is the original UTF-8 string (no wrapping, \
-         quoting, or escaping); the value handed to MinerCluster::ingest equals s \
-         byte-for-byte (instrumented MinerCluster stub records the body argument)."
+fn rfc0003_8_string_body_is_unwrapped_verbatim() {
+    let raw = "user 42 logged in from 10.0.0.1".to_owned();
+    let record = LogRecord {
+        body: Some(AnyValue {
+            value: Some(Value::StringValue(raw.clone())),
+        }),
+        ..Default::default()
+    };
+    let materialized = materialize_record(record, &[], None, TenantId::new("tenant-a"));
+    assert_eq!(
+        materialized.body,
+        Some(Body::String(raw)),
+        "a string body is unwrapped to Body::String byte-for-byte",
+    );
+}
+
+/// Scenario RFC0003.8 — an absent body stays `None`.
+/// See `docs/rfcs/0003-otlp-receiver.md` §5.
+#[test]
+fn rfc0003_8_absent_body_is_none() {
+    let record = LogRecord {
+        body: None,
+        ..Default::default()
+    };
+    let materialized = materialize_record(record, &[], None, TenantId::new("tenant-a"));
+    assert_eq!(
+        materialized.body, None,
+        "a record with no body materialises to body = None, not an empty string",
     );
 }

crates/ourios-ingester/tests/rfc0003_9_edge_otlp_fields.rs

@@ -1,18 +1,94 @@
 //! RFC0003.9 — Edge OTLP fields pass through unchanged.
 //!
-//! Red gate (`specified → red`): `#[ignore]`'d until the receiver
-//! lands.
+//! `severity_number = 0` (UNSPECIFIED) is an explicit value, kept as
+//! `0`; empty `scope_name`/`scope_version` and wire
+//! `observed_time_unix_nano = 0` narrow to `None`; `time_unix_nano = 0`
+//! (unknown) is kept as the `u64` `0`. Nothing is coalesced, substituted,
+//! or downcast to a default, and inherited resource attributes pass
+//! through verbatim.
+
+use opentelemetry_proto::tonic::common::v1::any_value::Value;
+use opentelemetry_proto::tonic::common::v1::{AnyValue, InstrumentationScope, KeyValue};
+use opentelemetry_proto::tonic::logs::v1::LogRecord;
+use ourios_core::tenant::TenantId;
+use ourios_ingester::receiver::materialize_record;
 
 /// Scenario RFC0003.9 — Edge OTLP fields pass through unchanged.
 /// See `docs/rfcs/0003-otlp-receiver.md` §5.
-#[ignore = "RFC 0003 red gate — implementation pending (RFC0003.9)"]
 #[test]
-fn rfc0003_9_edge_otlp_fields_are_not_coalesced() {
-    unimplemented!(
-        "RFC0003.9 — severity_number = 0 (UNSPECIFIED) is kept as 0, scope_name = \
-         None, and wire observed_time_unix_nano = 0 maps to None (the Option<u64> \
-         conversion this scenario owns). The record is accepted by \
-         MinerCluster::ingest without rejection, coalescing, substitution, or \
-         downcast to a default."
+fn rfc0003_9_edge_fields_pass_through_without_coalescing() {
+    let record = LogRecord {
+        time_unix_nano: 0,          // unknown event time
+        observed_time_unix_nano: 0, // unset collector observation time
+        severity_number: 0,         // SEVERITY_NUMBER_UNSPECIFIED
+        ..Default::default()
+    };
+    let scope = InstrumentationScope {
+        name: String::new(),
+        version: String::new(),
+        ..Default::default()
+    };
+    let resource_attributes = vec![KeyValue {
+        key: "service.name".to_owned(),
+        value: Some(AnyValue {
+            value: Some(Value::StringValue("checkout".to_owned())),
+        }),
+        ..Default::default()
+    }];
+
+    let materialized = materialize_record(
+        record,
+        &resource_attributes,
+        Some(&scope),
+        TenantId::new("tenant-a"),
+    );
+
+    assert_eq!(
+        materialized.severity_number, 0,
+        "UNSPECIFIED (0) is preserved, not coalesced or substituted",
+    );
+    assert_eq!(
+        materialized.observed_time_unix_nano, None,
+        "wire observed_time_unix_nano = 0 narrows to None",
+    );
+    assert_eq!(
+        materialized.time_unix_nano, 0,
+        "unknown event time is kept as the u64 0 (not narrowed to None)",
+    );
+    assert_eq!(
+        materialized.scope_name, None,
+        "empty scope name narrows to None",
     );
+    assert_eq!(
+        materialized.scope_version, None,
+        "empty scope version narrows to None",
+    );
+    assert_eq!(
+        materialized.body, None,
+        "an absent body stays None — no substitution",
+    );
+    assert_eq!(
+        materialized.resource_attributes, resource_attributes,
+        "inherited resource attributes pass through verbatim",
+    );
+}
+
+/// Scenario RFC0003.9 — out-of-range `severity_number` narrows to UNSPECIFIED.
+/// See `docs/rfcs/0003-otlp-receiver.md` §5.
+#[test]
+fn rfc0003_9_out_of_range_severity_narrows_to_unspecified() {
+    // Valid OTLP severity is 0..=24; values outside that range (incl.
+    // u8-representable 25..=255, negative, and > 255) normalise to
+    // 0/UNSPECIFIED so the OtlpLogRecord 0..=24 contract holds.
+    for (wire, expected) in [(0i32, 0u8), (24, 24), (25, 0), (1000, 0), (-5, 0)] {
+        let record = LogRecord {
+            severity_number: wire,
+            ..Default::default()
+        };
+        let materialized = materialize_record(record, &[], None, TenantId::new("tenant-a"));
+        assert_eq!(
+            materialized.severity_number, expected,
+            "severity_number {wire} narrows to {expected}",
+        );
+    }
 }