diff --git a/crates/ourios-core/src/alias.rs b/crates/ourios-core/src/alias.rs index 049bffdff..b6dee4c04 100644 --- a/crates/ourios-core/src/alias.rs +++ b/crates/ourios-core/src/alias.rs @@ -396,7 +396,11 @@ impl AliasMap { self.remove_id(&event.tenant_id, id); } } - AuditPayload::Template { .. } | AuditPayload::Compaction { .. } => {} + // RFC 0005 §3.7: a fold defined over named kinds ignores + // unknown kinds by construction. + AuditPayload::Template { .. } + | AuditPayload::Compaction { .. } + | AuditPayload::Unknown { .. } => {} } } diff --git a/crates/ourios-core/src/audit.rs b/crates/ourios-core/src/audit.rs index 53a7b601f..88da7a382 100644 --- a/crates/ourios-core/src/audit.rs +++ b/crates/ourios-core/src/audit.rs @@ -171,6 +171,21 @@ pub enum AuditPayload { /// Operator-supplied justification, ≤ 256 B. Empty when none. reason: String, }, + /// Reader-side catch-all per RFC 0005 §3.7's unknown-`event_kind` + /// tolerance rule (amendment 2026-06-12): an `event_kind` ordinal + /// above the reader's known range MUST NOT fail the file — the row + /// surfaces as this opaque envelope-only event. Carries the raw + /// ordinal and the stored `event_type` string verbatim so a + /// read-then-write round-trips them (the [`ParamType::Unknown`] + /// discipline applied to the kind enum). Folds defined over named + /// kinds (the alias projection, the RFC 0010 drift filter) ignore + /// it by construction; producers never construct it. + Unknown { + /// The stored ordinal, outside the §3.7 mapping table. + event_kind: u8, + /// The stored `event_type` string, preserved verbatim. + event_type: String, + }, /// A compaction consolidated a sealed partition's files /// (RFC 0009 §3.6). Carries no template identity. Compaction { @@ -239,13 +254,16 @@ impl AuditPayload { Self::AliasAsserted { .. } => EVENT_KIND_ALIAS_ASSERTED, Self::AliasRetracted { .. } => EVENT_KIND_ALIAS_RETRACTED, Self::Compaction { .. } => EVENT_KIND_COMPACTION, + Self::Unknown { event_kind, .. } => *event_kind, } } /// The canonical `event_type` string paired with - /// [`Self::event_kind`]. + /// [`Self::event_kind`] — or, for [`Self::Unknown`], the stored + /// string preserved verbatim (which is why this returns `&str` + /// rather than `&'static str`). #[must_use] - pub fn event_type(&self) -> &'static str { + pub fn event_type(&self) -> &str { match self { Self::Template { change, .. } => match change { TemplateChange::Widened { .. } => EVENT_TYPE_TEMPLATE_WIDENED, @@ -257,6 +275,7 @@ impl AuditPayload { Self::AliasAsserted { .. } => EVENT_TYPE_ALIAS_ASSERTED, Self::AliasRetracted { .. } => EVENT_TYPE_ALIAS_RETRACTED, Self::Compaction { .. } => EVENT_TYPE_COMPACTION, + Self::Unknown { event_type, .. } => event_type, } } @@ -271,10 +290,11 @@ impl AuditPayload { // `ourios.miner.alias.assertions` / // `ourios.miner.alias.retractions` (RFC 0001 §6.7); // `ourios.miner.merges` is reserved for the two structural - // widenings. - Self::AliasAsserted { .. } | Self::AliasRetracted { .. } | Self::Compaction { .. } => { - false - } + // widenings. An unknown kind cannot be assumed to be one. + Self::AliasAsserted { .. } + | Self::AliasRetracted { .. } + | Self::Compaction { .. } + | Self::Unknown { .. } => false, } } } diff --git a/crates/ourios-parquet/src/audit_reader.rs b/crates/ourios-parquet/src/audit_reader.rs index 3c2e5b442..8baa2cd29 100644 --- a/crates/ourios-parquet/src/audit_reader.rs +++ b/crates/ourios-parquet/src/audit_reader.rs @@ -16,15 +16,14 @@ //! sample`, `reason`) surface as `None`; missing baseline REQUIRED //! columns are a hard read error. //! -//! **Unknown `event_kind` ordinals** are currently surfaced as a -//! [`AuditReaderError::UnknownEventKind`] hard error. The audit -//! event enum [`AuditPayload`] has no catch-all variant; a -//! future RFC 0005 §3.8 amendment that adds a new ordinal will -//! either extend the enum (and this match) or introduce an -//! `Unknown(u8)` variant analogous to [`ParamType::Unknown`]. The -//! data side handles the analogous case (`params.type_tag = 99`) -//! via `ParamType::Unknown`; the audit side defers the choice -//! until a real new variant lands. +//! **Unknown `event_kind` ordinals** surface as +//! [`AuditPayload::Unknown`] — an opaque envelope-only event — never +//! as a file failure, per RFC 0005 §3.7's unknown-`event_kind` +//! tolerance rule (amendment 2026-06-12, the rule pinned when kinds +//! 4–5 landed). This is the [`ParamType::Unknown`] discipline applied +//! to the kind enum: every future §3.8 ordinal addition stays +//! non-breaking for readers, and folds defined over named kinds +//! ignore unknown rows by construction. use std::fmt; use std::fs::File; @@ -35,6 +34,7 @@ use std::time::{Duration, SystemTime}; use arrow_array::cast::AsArray; use arrow_array::types::{Int32Type, TimestampNanosecondType, UInt8Type, UInt32Type, UInt64Type}; use arrow_array::{Array, RecordBatch, StructArray}; +use ourios_core::alias::ActorId; use ourios_core::audit::{AuditEvent, AuditPayload, ParamType, SlotExpansion, TemplateChange}; use ourios_core::tenant::TenantId; use parquet::arrow::arrow_reader::{ParquetRecordBatchReader, ParquetRecordBatchReaderBuilder}; @@ -42,7 +42,8 @@ use parquet::errors::ParquetError; use crate::audit_columns; use crate::audit_record_batch::{ - EVENT_KIND_COMPACTION, EVENT_KIND_TEMPLATE_TYPE_EXPANDED, EVENT_KIND_TEMPLATE_WIDENED, + EVENT_KIND_ALIAS_ASSERTED, EVENT_KIND_ALIAS_RETRACTED, EVENT_KIND_COMPACTION, + EVENT_KIND_TEMPLATE_TYPE_EXPANDED, EVENT_KIND_TEMPLATE_WIDENED, EVENT_KIND_TEMPLATE_WIDENING_REJECTED_DEGENERATE, }; use crate::audit_writer::{audit_partition_matches, derive_audit_partition}; @@ -159,14 +160,6 @@ pub enum AuditReaderError { column: &'static str, detail: String, }, - /// `event_kind` ordinal isn't one of the §3.7 mapping - /// table's values. Until a future amendment adds an - /// `AuditPayload::Unknown` variant, unknown ordinals are - /// a hard error. - UnknownEventKind { - row_index: usize, - ordinal: u8, - }, /// `timestamp` nanos couldn't be converted to `SystemTime` /// (negative — pre-epoch — or out of `Duration` range). TimestampDecode { @@ -200,13 +193,6 @@ impl fmt::Display for AuditReaderError { Self::Conversion { column, detail } => { write!(f, "column `{column}` conversion failed: {detail}") } - Self::UnknownEventKind { row_index, ordinal } => write!( - f, - "row {row_index}: unknown event_kind ordinal {ordinal} — the §3.7 mapping \ - table pins 0 / 1 / 2; reading an unknown ordinal needs an \ - AuditPayload::Unknown variant which is deferred until a real new variant \ - lands via a §3.8 amendment", - ), Self::TimestampDecode { row_index, nanos } => write!( f, "row {row_index}: timestamp = {nanos} ns can't be converted to SystemTime \ @@ -244,7 +230,6 @@ impl std::error::Error for AuditReaderError { Self::Parquet(e) => Some(e), Self::MissingRequiredColumn { .. } | Self::Conversion { .. } - | Self::UnknownEventKind { .. } | Self::TimestampDecode { .. } | Self::PartitionMismatch { .. } => None, } @@ -304,10 +289,11 @@ fn batch_to_audit_events( let timestamp = required_timestamp(batch, audit_columns::TIMESTAMP, row_offset)?; let event_kind = required_u8(batch, audit_columns::EVENT_KIND, row_offset)?; // `event_type` is required-and-redundant (kept in sync with - // `event_kind` by the writer). Surface it for sanity-check - // diagnostics but use `event_kind` as the source of truth - // for variant dispatch. - let _event_type = required_string(batch, audit_columns::EVENT_TYPE, row_offset)?; + // `event_kind` by the writer); `event_kind` is the source of + // truth for variant dispatch. The string is preserved verbatim + // on the unknown-kind tolerance path (§3.7) so a read-then-write + // round-trips the envelope. + let event_type = required_string(batch, audit_columns::EVENT_TYPE, row_offset)?; // Template-group columns — OPTIONAL since the §3.7 amendment // (NULL on `compaction` rows). Required-by-convention for the // template kinds; [`require_at`] errors if a template row finds @@ -332,6 +318,10 @@ fn batch_to_audit_events( optional_string(batch, audit_columns::COMPACTION_OUTPUT_FILE)?.unwrap_or_default(); let compaction_generation = optional_u64(batch, audit_columns::COMPACTION_GENERATION)?; let compaction_rows = optional_u64(batch, audit_columns::COMPACTION_ROWS)?; + // Alias-group columns (RFC 0001 §6.7 / §3.7 amendment 2026-06-12). + let alias_representative_id = optional_u64(batch, audit_columns::ALIAS_REPRESENTATIVE_ID)?; + let alias_member_ids = optional_u64_list(batch, audit_columns::ALIAS_MEMBER_IDS)?; + let alias_actor = optional_string(batch, audit_columns::ALIAS_ACTOR)?.unwrap_or_default(); for i in 0..n { let file_row = row_offset + i; @@ -362,44 +352,32 @@ fn batch_to_audit_events( change: decode_template_change(&cols, i, file_row)?, } } - EVENT_KIND_COMPACTION => AuditPayload::Compaction { - partition: require_at( - &compaction_partition, - i, - audit_columns::COMPACTION_PARTITION, - file_row, - )?, - input_files: require_at( - &compaction_input_files, - i, - audit_columns::COMPACTION_INPUT_FILES, - file_row, - )?, - output_file: require_at( - &compaction_output_file, - i, - audit_columns::COMPACTION_OUTPUT_FILE, - file_row, - )?, - generation: require_at( - &compaction_generation, - i, - audit_columns::COMPACTION_GENERATION, - file_row, - )?, - rows: require_at( - &compaction_rows, - i, - audit_columns::COMPACTION_ROWS, - file_row, - )?, - }, - other => { - return Err(AuditReaderError::UnknownEventKind { - row_index: file_row, - ordinal: other, - }); + EVENT_KIND_COMPACTION => { + let cols = CompactionColumns { + partition: &compaction_partition, + input_files: &compaction_input_files, + output_file: &compaction_output_file, + generation: &compaction_generation, + rows: &compaction_rows, + }; + decode_compaction_payload(&cols, i, file_row)? + } + kind @ (EVENT_KIND_ALIAS_ASSERTED | EVENT_KIND_ALIAS_RETRACTED) => { + let cols = AliasColumns { + representative_id: &alias_representative_id, + member_ids: &alias_member_ids, + actor: &alias_actor, + reason: &reason, + }; + decode_alias_payload(kind, &cols, i, file_row)? } + // RFC 0005 §3.7 unknown-event_kind tolerance (amendment + // 2026-06-12): an ordinal above the known range surfaces + // as an opaque envelope-only event, never a file failure. + other => AuditPayload::Unknown { + event_kind: other, + event_type: event_type[i].clone(), + }, }; events.push(AuditEvent { @@ -424,6 +402,114 @@ struct TemplateColumns<'a> { reason: &'a [Option], } +/// Borrowed per-column slices the compaction-payload decoder reads +/// (RFC 0009 §3.6 / §3.7 amendment 2026-06-03). All five columns are +/// required-by-convention non-null for kind 3. +struct CompactionColumns<'a> { + partition: &'a [Option], + input_files: &'a [Option>], + output_file: &'a [Option], + generation: &'a [Option], + rows: &'a [Option], +} + +/// Rebuild the compaction payload for row `i` from the +/// `compaction_*` columns. +fn decode_compaction_payload( + cols: &CompactionColumns, + i: usize, + file_row: usize, +) -> Result { + Ok(AuditPayload::Compaction { + partition: require_at( + cols.partition, + i, + audit_columns::COMPACTION_PARTITION, + file_row, + )?, + input_files: require_at( + cols.input_files, + i, + audit_columns::COMPACTION_INPUT_FILES, + file_row, + )?, + output_file: require_at( + cols.output_file, + i, + audit_columns::COMPACTION_OUTPUT_FILE, + file_row, + )?, + generation: require_at( + cols.generation, + i, + audit_columns::COMPACTION_GENERATION, + file_row, + )?, + rows: require_at(cols.rows, i, audit_columns::COMPACTION_ROWS, file_row)?, + }) +} + +/// Borrowed per-column slices the alias-payload decoder reads +/// (RFC 0001 §6.7 / §3.7 amendment 2026-06-12). +struct AliasColumns<'a> { + representative_id: &'a [Option], + member_ids: &'a [Option>], + actor: &'a [Option], + reason: &'a [Option], +} + +/// Rebuild the alias payload for row `i` from the `alias_*` columns. +/// All three alias columns are required-by-convention non-null for +/// kinds 4–5; `member_ids` may be the valid empty list (distinct from +/// NULL), and an on-disk NULL `reason` decodes to the in-memory empty +/// string (the §3.7 `"" ↔ NULL` round-trip rule). +fn decode_alias_payload( + kind: u8, + cols: &AliasColumns, + i: usize, + file_row: usize, +) -> Result { + let representative_id = require_at( + cols.representative_id, + i, + audit_columns::ALIAS_REPRESENTATIVE_ID, + file_row, + )?; + let member_ids = require_at( + cols.member_ids, + i, + audit_columns::ALIAS_MEMBER_IDS, + file_row, + )?; + let actor_str = require_at(cols.actor, i, audit_columns::ALIAS_ACTOR, file_row)?; + // Aliasing is never anonymous (RFC 0001 §6.7); an empty actor is + // a writer-invariant violation. + let actor = ActorId::new(actor_str).map_err(|e| AuditReaderError::Conversion { + column: audit_columns::ALIAS_ACTOR, + detail: format!("row {file_row}: {e}"), + })?; + let reason = cols + .reason + .get(i) + .and_then(Clone::clone) + .unwrap_or_default(); + if kind == EVENT_KIND_ALIAS_ASSERTED { + Ok(AuditPayload::AliasAsserted { + representative_id, + member_ids, + actor, + reason, + }) + } else { + Ok(AuditPayload::AliasRetracted { + representative_id, + member_ids, + actor, + reason, + }) + } +} + /// Value at `col[i]`, or a `Conversion` error if it is absent / NULL — /// the writer-invariant violation a corrupt or foreign-writer file /// would produce (a template row missing a template column, or a @@ -1036,7 +1122,7 @@ fn optional_string_list( return Err(AuditReaderError::Conversion { column: name, detail: format!( - "row {row_idx} element {i}: NULL but the element field is non-nullable", + "batch row {row_idx} element {i}: NULL but the element field is non-nullable", ), }); } @@ -1047,6 +1133,54 @@ fn optional_string_list( Ok(out) } +/// Per-row `Option>` for the nullable `alias_member_ids` +/// `LIST` column. NULL list ⇒ `None` (not an alias row); +/// empty list ⇒ `Some(vec![])` — the §3.7 empty-vs-NULL distinction. +/// The element field is non-nullable, so a NULL element is a corrupt +/// row. +fn optional_u64_list( + batch: &RecordBatch, + name: &'static str, +) -> Result>>, AuditReaderError> { + let Some(col) = optional_column(batch, name) else { + return Ok(Vec::new()); + }; + let list = col + .as_list_opt::() + .ok_or_else(|| AuditReaderError::Conversion { + column: name, + detail: "column is not a LIST as declared".to_string(), + })?; + let mut out = Vec::with_capacity(list.len()); + for row_idx in 0..list.len() { + if list.is_null(row_idx) { + out.push(None); + continue; + } + let elements = list.value(row_idx); + let ids = elements.as_primitive_opt::().ok_or_else(|| { + AuditReaderError::Conversion { + column: name, + detail: "list element is not UInt64".to_string(), + } + })?; + let mut row = Vec::with_capacity(ids.len()); + for i in 0..ids.len() { + if ids.is_null(i) { + return Err(AuditReaderError::Conversion { + column: name, + detail: format!( + "batch row {row_idx} element {i}: NULL but the element field is non-nullable", + ), + }); + } + row.push(ids.value(i)); + } + out.push(Some(row)); + } + Ok(out) +} + #[cfg(test)] mod tests { //! Colocated unit tests for the audit-reader paths that the @@ -1081,45 +1215,79 @@ mod tests { } } - /// Pins file-global row indexing in `batch_to_audit_events`: - /// a forged batch where row 0 carries an unknown `event_kind` - /// must report `row_index = row_offset` (not `0`) on the - /// returned `UnknownEventKind` error. This is the multi-batch - /// invariant — without the - /// `row_offset + i` addition, a later-batch error in a real - /// file would point at row 0 of every batch instead of the - /// running file-level offset. + /// FLIPPED from expect-error (`UnknownEventKind`) to + /// expect-opaque-event per the RFC-gated contract change in + /// RFC 0005 §3.7 (amendment 2026-06-12, PR #183): a reader + /// encountering an `event_kind` ordinal above its known range + /// MUST NOT fail the file — the row surfaces as an opaque + /// envelope-only [`AuditPayload::Unknown`]. The old test pinned + /// the documented deferral ("hard error until a real new variant + /// lands"); kinds 4–5 were that variant, so the deferral is + /// resolved and the old assertion is exactly the behaviour the + /// amendment removes (`CLAUDE.md` §6.2). #[test] - fn batch_to_audit_events_reports_file_global_row_index_on_unknown_event_kind() { + fn batch_to_audit_events_surfaces_unknown_event_kind_as_opaque_event() { + // Arrange — replace the event_kind column with a single 99 + // ordinal (outside the §3.7 mapping table), keeping every + // other column intact. let valid = audit_events_to_batch(&[widened_event("acme")]).expect("batch builds"); let event_kind_idx = valid .schema() .index_of(audit_columns::EVENT_KIND) .expect("schema has event_kind"); - - // Replace the event_kind column with a single 99 ordinal — - // outside the §3.7 mapping table — keeping every other - // column intact. let mut columns: Vec = valid.columns().to_vec(); columns[event_kind_idx] = Arc::new(UInt8Array::from(vec![99u8])); let forged = RecordBatch::try_new(valid.schema(), columns).expect("forged batch type-checks"); - // Pretend this batch is the second batch of a longer file — - // the prior batches contributed 50 rows. - let err = batch_to_audit_events(&forged, 50).expect_err("unknown event_kind must error"); - match err { - AuditReaderError::UnknownEventKind { row_index, ordinal } => { - assert_eq!( - row_index, 50, - "row index must be file-global, not batch-local" - ); - assert_eq!(ordinal, 99); + // Act — decoding must NOT fail the batch. + let events = batch_to_audit_events(&forged, 50).expect("unknown kind must not error"); + + // Assert — the row decodes to the opaque envelope: the raw + // ordinal plus the stored event_type string verbatim, with + // the envelope fields (tenant, timestamp) preserved. + assert_eq!(events.len(), 1); + assert_eq!(events[0].tenant_id.as_str(), "acme"); + assert_eq!(events[0].timestamp, widened_event("acme").timestamp); + match &events[0].payload { + AuditPayload::Unknown { + event_kind, + event_type, + } => { + assert_eq!(*event_kind, 99); + // The forged batch kept the original string column — + // preserved verbatim, not re-derived from the ordinal. + assert_eq!(event_type, "template_widened"); } - other => panic!("expected UnknownEventKind, got {other:?}"), + other => panic!("expected AuditPayload::Unknown, got {other:?}"), } } + /// The opaque-envelope event round-trips: a read-then-write of an + /// [`AuditPayload::Unknown`] row preserves the envelope verbatim + /// (RFC 0005 §3.7 unknown-`event_kind` tolerance) and leaves + /// every payload column NULL, so re-decoding yields the same + /// opaque event. + #[test] + fn unknown_event_round_trips_envelope_only() { + // Arrange — an opaque event as a reader would surface it. + let unknown = AuditEvent { + tenant_id: ourios_core::tenant::TenantId::new("acme"), + timestamp: UNIX_EPOCH + Duration::from_secs(1_775_127_480), + payload: AuditPayload::Unknown { + event_kind: 42, + event_type: "some_future_kind".to_string(), + }, + }; + + // Act — write it back out and decode again. + let batch = audit_events_to_batch(std::slice::from_ref(&unknown)).expect("batch builds"); + let events = batch_to_audit_events(&batch, 0).expect("decode"); + + // Assert — byte-for-byte envelope preservation. + assert_eq!(events, vec![unknown]); + } + /// `decode_timestamp` rejects negative i64 nanos as /// `TimestampDecode` — covers the `u64::try_from` branch. /// The `checked_add` branch is defensive against narrow- diff --git a/crates/ourios-parquet/src/audit_record_batch.rs b/crates/ourios-parquet/src/audit_record_batch.rs index 1a00aa6ef..260b24668 100644 --- a/crates/ourios-parquet/src/audit_record_batch.rs +++ b/crates/ourios-parquet/src/audit_record_batch.rs @@ -16,6 +16,16 @@ //! | `TemplateTypeExpanded` | `1` | `[]` | event's slots | both = pre = post | `NULL` | //! | `TemplateWideningRejectedDegenerate` | `2` | `[]` | `[]` | both = `current_template` | JSON of `would_be_*` | //! +//! The `Compaction` (kind `3`) and `AliasAsserted` / `AliasRetracted` +//! (kinds `4` / `5`) variants populate only their own kind-prefixed +//! `compaction_*` / `alias_*` columns plus the envelope; every other +//! payload column is `NULL` (§3.8 rule 6, per kind). For the alias +//! kinds the `member_ids` list is stored **verbatim** (no sort/dedup — +//! the semantic value is the set `{representative_id} ∪ member_ids`, +//! folded by consumers), an empty list is valid and distinct from +//! `NULL`, and the in-memory empty-string `reason` maps to `NULL` on +//! disk (`"" ↔ NULL`, RFC 0005 §3.7 amendment 2026-06-12). +//! //! **Rejection-variant `reason` payload.** The in-memory //! [`TemplateChange::RejectedDegenerate`] carries //! `would_be_template: String` and `would_be_positions: Vec`, @@ -56,10 +66,11 @@ use crate::audit_schema; /// `ourios-core`; re-exported here so the reader's ordinal match and /// existing call sites resolve them at their established path. pub use ourios_core::audit::{ - EVENT_KIND_COMPACTION, EVENT_KIND_TEMPLATE_TYPE_EXPANDED, EVENT_KIND_TEMPLATE_WIDENED, - EVENT_KIND_TEMPLATE_WIDENING_REJECTED_DEGENERATE, EVENT_TYPE_COMPACTION, - EVENT_TYPE_TEMPLATE_TYPE_EXPANDED, EVENT_TYPE_TEMPLATE_WIDENED, - EVENT_TYPE_TEMPLATE_WIDENING_REJECTED_DEGENERATE, + EVENT_KIND_ALIAS_ASSERTED, EVENT_KIND_ALIAS_RETRACTED, EVENT_KIND_COMPACTION, + EVENT_KIND_TEMPLATE_TYPE_EXPANDED, EVENT_KIND_TEMPLATE_WIDENED, + EVENT_KIND_TEMPLATE_WIDENING_REJECTED_DEGENERATE, EVENT_TYPE_ALIAS_ASSERTED, + EVENT_TYPE_ALIAS_RETRACTED, EVENT_TYPE_COMPACTION, EVENT_TYPE_TEMPLATE_TYPE_EXPANDED, + EVENT_TYPE_TEMPLATE_WIDENED, EVENT_TYPE_TEMPLATE_WIDENING_REJECTED_DEGENERATE, }; /// Build an Arrow `RecordBatch` matching [`audit_schema`] from a @@ -110,17 +121,6 @@ pub enum AuditBatchError { old_template: String, new_template: String, }, - /// An `alias_asserted` / `alias_retracted` event (RFC 0001 §6.7) - /// was handed to the audit-Parquet writer, whose §3.7 schema has - /// no columns to represent the alias payload (`representative_id`, - /// `member_ids`, `actor`, `reason`). Adding those columns is the - /// RFC 0005 storage split (sibling to issue #147), deliberately - /// out of scope for the alias write-path slice — so the writer - /// rejects rather than inventing columns or dropping the event - /// silently. Carries the offending `event_type` for diagnostics. - /// Alias events are durable via the alias event log; their - /// audit-Parquet materialization waits for that split. - AliasEventNotYetPersistable { event_type: &'static str }, /// Arrow rejected the constructed `RecordBatch` (column-length /// mismatch, schema-shape mismatch). Internal bug if it ever /// fires — the array builders are constructed against @@ -152,13 +152,6 @@ impl fmt::Display for AuditBatchError { = {new_template:?}, but RFC 0005 §3.7 requires they be equal for this \ variant (template tokens don't change)", ), - Self::AliasEventNotYetPersistable { event_type } => write!( - f, - "audit event {event_type} (RFC 0001 §6.7 alias write path) is not yet \ - representable in the audit-Parquet schema; the alias columns are the RFC \ - 0005 storage split (issue #147 sibling). Alias events are durable via the \ - alias event log, not this writer.", - ), Self::Arrow(e) => write!(f, "arrow rejected RecordBatch: {e}"), } } @@ -169,8 +162,7 @@ impl std::error::Error for AuditBatchError { match self { Self::PreEpochTimestamp | Self::TimestampOverflow { .. } - | Self::TemplateMustNotChange { .. } - | Self::AliasEventNotYetPersistable { .. } => None, + | Self::TemplateMustNotChange { .. } => None, Self::Arrow(e) => Some(e), } } @@ -219,6 +211,9 @@ struct Builders { compaction_output_file: StringBuilder, compaction_generation: UInt64Builder, compaction_rows: UInt64Builder, + alias_representative_id: UInt64Builder, + alias_member_ids: GenericListBuilder, + alias_actor: StringBuilder, } impl Builders { @@ -272,6 +267,13 @@ impl Builders { compaction_output_file: StringBuilder::with_capacity(cap, 0), compaction_generation: UInt64Builder::with_capacity(cap), compaction_rows: UInt64Builder::with_capacity(cap), + alias_representative_id: UInt64Builder::with_capacity(cap), + alias_member_ids: GenericListBuilder::new(UInt64Builder::new()).with_field(Field::new( + "element", + DataType::UInt64, + false, + )), + alias_actor: StringBuilder::with_capacity(cap, 0), } } @@ -291,9 +293,11 @@ impl Builders { triggering_line_sample, change, } => { - // Template events leave the compaction columns NULL - // (§3.7 amendment 2026-06-03). + // Template events leave the compaction and alias + // columns NULL (§3.7 amendments 2026-06-03 / + // 2026-06-12). self.append_compaction_nulls(); + self.append_alias_nulls(); self.template_id.append_value(*template_id); self.triggering_line_hash .append_value(triggering_line_hash) @@ -304,15 +308,50 @@ impl Builders { } self.append_template_change(change)?; } - AuditPayload::AliasAsserted { .. } | AuditPayload::AliasRetracted { .. } => { - // RFC 0001 §6.7 alias events have no audit-Parquet - // columns yet — that schema extension is the RFC 0005 - // split (#147 sibling). Reject rather than persist a - // lossy row; alias events are durable via the alias - // event log meanwhile. - return Err(AuditBatchError::AliasEventNotYetPersistable { - event_type: e.payload.event_type(), - }); + AuditPayload::AliasAsserted { + representative_id, + member_ids, + actor, + reason, + } + | AuditPayload::AliasRetracted { + representative_id, + member_ids, + actor, + reason, + } => { + // Alias events (RFC 0001 §6.7 / §3.7 amendment + // 2026-06-12) populate only the envelope, the + // `alias_*` columns, and `reason`. `member_ids` is + // stored verbatim (no sort/dedup — round-trip is + // exact; consumers fold it as a set), an empty list + // is valid and distinct from NULL, and the in-memory + // empty-string `reason` maps to NULL (`"" ↔ NULL`). + self.append_template_nulls(); + self.append_compaction_nulls(); + if reason.is_empty() { + self.reason.append_null(); + } else { + self.reason.append_value(reason); + } + self.alias_representative_id + .append_value(*representative_id); + for id in member_ids { + self.alias_member_ids.values().append_value(*id); + } + self.alias_member_ids.append(true); + self.alias_actor.append_value(actor.as_str()); + } + AuditPayload::Unknown { .. } => { + // §3.7 unknown-event_kind tolerance: a read-then-write + // of a row a future writer produced preserves the + // envelope verbatim (`event_kind` / `event_type` are + // already appended from the payload accessors above) + // with every payload column NULL. + self.append_template_nulls(); + self.append_compaction_nulls(); + self.append_alias_nulls(); + self.reason.append_null(); } AuditPayload::Compaction { partition, @@ -322,8 +361,13 @@ impl Builders { rows, } => { // Compaction events leave every template-specific - // column NULL (§3.7 relaxed them to OPTIONAL). + // and alias column NULL (§3.7 relaxed the former to + // OPTIONAL; the latter are alias-kind-only), and the + // facts live in the `compaction_*` columns — `reason` + // stays NULL. self.append_template_nulls(); + self.append_alias_nulls(); + self.reason.append_null(); self.compaction_partition.append_value(partition); append_string_list(&mut self.compaction_input_files, input_files); self.compaction_output_file.append_value(output_file); @@ -407,7 +451,11 @@ impl Builders { Ok(()) } - /// NULL every template-specific column — for a `compaction` row. + /// NULL every template-specific column — for a non-template row. + /// `reason` is *not* part of this group: it is the shared + /// justification/diagnostic column, populated per kind by the + /// caller (the rejection JSON for kind 2, the operator's + /// justification for kinds 4–5, NULL otherwise). fn append_template_nulls(&mut self) { self.template_id.append_null(); self.old_version.append_null(); @@ -418,10 +466,9 @@ impl Builders { self.slots_expanded.append_null(); self.triggering_line_hash.append_null(); self.triggering_line_sample.append_null(); - self.reason.append_null(); } - /// NULL every compaction-specific column — for a template row. + /// NULL every compaction-specific column — for a non-compaction row. fn append_compaction_nulls(&mut self) { self.compaction_partition.append_null(); self.compaction_input_files.append_null(); @@ -430,6 +477,13 @@ impl Builders { self.compaction_rows.append_null(); } + /// NULL every alias-specific column — for a non-alias row. + fn append_alias_nulls(&mut self) { + self.alias_representative_id.append_null(); + self.alias_member_ids.append_null(); + self.alias_actor.append_null(); + } + fn finish(mut self) -> Vec { vec![ Arc::new(self.tenant_id.finish()), @@ -451,6 +505,9 @@ impl Builders { Arc::new(self.compaction_output_file.finish()), Arc::new(self.compaction_generation.finish()), Arc::new(self.compaction_rows.finish()), + Arc::new(self.alias_representative_id.finish()), + Arc::new(self.alias_member_ids.finish()), + Arc::new(self.alias_actor.finish()), ] } } @@ -616,6 +673,32 @@ mod tests { } } + /// An `alias_asserted` / `alias_retracted` audit event (RFC 0001 + /// §6.7) for the §3.7 alias kinds. + fn alias_event(asserted: bool, member_ids: Vec, reason: &str) -> AuditEvent { + let actor = ourios_core::alias::ActorId::new("op-alice").expect("non-empty actor"); + let payload = if asserted { + AuditPayload::AliasAsserted { + representative_id: 1, + member_ids, + actor, + reason: reason.to_string(), + } + } else { + AuditPayload::AliasRetracted { + representative_id: 1, + member_ids, + actor, + reason: reason.to_string(), + } + }; + AuditEvent { + tenant_id: TenantId::new("acme"), + timestamp: ts(1_775_127_700), + payload, + } + } + #[test] fn builds_batch_for_one_of_each_variant() { let batch = audit_events_to_batch(&[ @@ -623,9 +706,11 @@ mod tests { type_expanded_event(), rejection_event(), compaction_event(), + alias_event(true, vec![2, 3], "deploy re-split the login template"), + alias_event(false, vec![], ""), ]) .expect("batch builds"); - assert_eq!(batch.num_rows(), 4); + assert_eq!(batch.num_rows(), 6); assert_eq!(batch.schema(), audit_schema()); } @@ -664,29 +749,48 @@ mod tests { assert!(matches!(err, AuditBatchError::PreEpochTimestamp)); } + /// FLIPPED from `alias_events_are_rejected_pending_the_rfc_0005_split` + /// per the RFC-gated contract change in RFC 0005 §3.7 (amendment + /// 2026-06-12, PR #183): the interim `AliasEventNotYetPersistable` + /// rejection is retired now that the `alias_*` columns exist, so + /// the writer maps kinds 4–5 instead of erroring. The old test's + /// assertion ("alias events are not persistable") is exactly the + /// behaviour the amendment removes (`CLAUDE.md` §6.2). #[test] - fn alias_events_are_rejected_pending_the_rfc_0005_split() { - // RFC 0001 §6.7 alias events have no audit-Parquet columns yet - // (the RFC 0005 storage split). The writer must reject them - // rather than persist a lossy row or panic; they stay durable - // via the alias event log meanwhile. - let asserted = AuditEvent { - tenant_id: TenantId::new("acme"), - timestamp: ts(1_775_127_600), - payload: AuditPayload::AliasAsserted { - representative_id: 1, - member_ids: vec![2], - actor: ourios_core::alias::ActorId::new("op").expect("non-empty actor"), - reason: String::new(), - }, - }; - let err = audit_events_to_batch(std::slice::from_ref(&asserted)) - .expect_err("alias events are not yet persistable"); - assert!(matches!( - err, - AuditBatchError::AliasEventNotYetPersistable { - event_type: "alias_asserted" - } - )); + fn alias_events_build_a_batch_with_kinds_4_and_5() { + use arrow_array::cast::AsArray; + use arrow_array::types::UInt8Type; + + // Arrange — one assertion (kind 4) and one retraction (kind 5). + let events = [ + alias_event(true, vec![2], "merge"), + alias_event(false, vec![], ""), + ]; + + // Act. + let batch = audit_events_to_batch(&events).expect("alias events are persistable"); + + // Assert — the §3.7 dual-column mapping carries ordinals 4 / 5 + // and the paired canonical strings. + let kind_idx = batch + .schema() + .index_of(crate::audit_columns::EVENT_KIND) + .expect("event_kind column"); + let kinds = batch + .column(kind_idx) + .as_primitive::() + .values() + .to_vec(); + assert_eq!( + kinds, + vec![EVENT_KIND_ALIAS_ASSERTED, EVENT_KIND_ALIAS_RETRACTED] + ); + let type_idx = batch + .schema() + .index_of(crate::audit_columns::EVENT_TYPE) + .expect("event_type column"); + let types = batch.column(type_idx).as_string::(); + assert_eq!(types.value(0), EVENT_TYPE_ALIAS_ASSERTED); + assert_eq!(types.value(1), EVENT_TYPE_ALIAS_RETRACTED); } } diff --git a/crates/ourios-parquet/src/audit_writer.rs b/crates/ourios-parquet/src/audit_writer.rs index 9b942505a..c43fe828e 100644 --- a/crates/ourios-parquet/src/audit_writer.rs +++ b/crates/ourios-parquet/src/audit_writer.rs @@ -536,6 +536,7 @@ fn audit_writer_properties() -> Result { audit_columns::TRIGGERING_LINE_HASH, audit_columns::TRIGGERING_LINE_SAMPLE, audit_columns::REASON, + audit_columns::ALIAS_ACTOR, ] { builder = builder.set_column_statistics_enabled( ColumnPath::new(vec![no_page_idx_col.to_string()]), @@ -576,6 +577,18 @@ fn audit_writer_properties() -> Result { ]), EnabledStatistics::Chunk, ); + // `alias_member_ids` "(list values)" gets `Page index = no` on + // its list leaf per the §3.7 table (amendment 2026-06-12); + // `alias_representative_id` keeps the page-index default + // (`Page index = yes`, same shape as `template_id`). + builder = builder.set_column_statistics_enabled( + ColumnPath::new(vec![ + audit_columns::ALIAS_MEMBER_IDS.to_string(), + "list".to_string(), + "element".to_string(), + ]), + EnabledStatistics::Chunk, + ); Ok(builder.build()) } diff --git a/crates/ourios-parquet/src/lib.rs b/crates/ourios-parquet/src/lib.rs index cd1f17ff0..a8132e831 100644 --- a/crates/ourios-parquet/src/lib.rs +++ b/crates/ourios-parquet/src/lib.rs @@ -109,6 +109,11 @@ pub mod audit_columns { pub const COMPACTION_OUTPUT_FILE: &str = "compaction_output_file"; pub const COMPACTION_GENERATION: &str = "compaction_generation"; pub const COMPACTION_ROWS: &str = "compaction_rows"; + // Alias-event columns (RFC 0005 §3.7 amendment 2026-06-12 / + // RFC 0001 §6.7); NULL for all other kinds. + pub const ALIAS_REPRESENTATIVE_ID: &str = "alias_representative_id"; + pub const ALIAS_MEMBER_IDS: &str = "alias_member_ids"; + pub const ALIAS_ACTOR: &str = "alias_actor"; } /// Build the data-file Arrow schema per RFC 0005 §3.2. @@ -257,5 +262,21 @@ pub fn audit_schema() -> SchemaRef { Field::new(audit_columns::COMPACTION_OUTPUT_FILE, DataType::Utf8, true), Field::new(audit_columns::COMPACTION_GENERATION, DataType::UInt64, true), Field::new(audit_columns::COMPACTION_ROWS, DataType::UInt64, true), + // Alias-event columns (RFC 0001 §6.7 / §3.7 amendment + // 2026-06-12): OPTIONAL, NULL for all other kinds; + // required-by-convention non-null for kinds 4–5 (the + // member list possibly empty — an empty list is valid and + // distinct from NULL). + Field::new( + audit_columns::ALIAS_REPRESENTATIVE_ID, + DataType::UInt64, + true, + ), + Field::new( + audit_columns::ALIAS_MEMBER_IDS, + DataType::List(Arc::new(Field::new("element", DataType::UInt64, false))), + true, + ), + Field::new(audit_columns::ALIAS_ACTOR, DataType::Utf8, true), ])) } diff --git a/crates/ourios-parquet/tests/audit_round_trip.rs b/crates/ourios-parquet/tests/audit_round_trip.rs index f3f00c0ee..86920b1a6 100644 --- a/crates/ourios-parquet/tests/audit_round_trip.rs +++ b/crates/ourios-parquet/tests/audit_round_trip.rs @@ -20,11 +20,12 @@ use std::path::Component; use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use ourios_core::alias::ActorId; use ourios_core::audit::{ AuditEvent, AuditPayload, ParamType, SlotExpansion, TemplateChange, hash_triggering_line, }; use ourios_core::tenant::TenantId; -use ourios_parquet::{AuditReader, AuditWriter, PartitionKey}; +use ourios_parquet::{AuditReader, AuditWriter, PartitionKey, audit_columns}; use tempfile::TempDir; fn ts(offset_secs: u64) -> SystemTime { @@ -279,6 +280,128 @@ fn rfc0005_7_rejection_variant_round_trips_via_reason_column() { assert_eq!(would_be_positions, &vec![0]); } +/// An `alias_asserted` audit event (RFC 0001 §6.7 / RFC 0005 §3.7 +/// amendment 2026-06-12). `member_ids` deliberately carries a +/// duplicate and an unsorted order — the writer stores the list +/// verbatim, so the round trip must preserve it exactly. +fn alias_asserted_event(tenant: &str) -> AuditEvent { + AuditEvent { + tenant_id: TenantId::new(tenant), + timestamp: ts(1_775_127_520), + payload: AuditPayload::AliasAsserted { + representative_id: 30, + member_ids: vec![20, 10, 20], + actor: ActorId::new("op-alice").expect("non-empty actor"), + reason: "deploy 2026-06 re-split the login template".to_string(), + }, + } +} + +/// An `alias_retracted` audit event with the empty member list (the +/// common single-id retraction) and no reason — exercising the §3.7 +/// empty-list-vs-NULL distinction and the `"" ↔ NULL` reason rule. +fn alias_retracted_event(tenant: &str) -> AuditEvent { + AuditEvent { + tenant_id: TenantId::new(tenant), + timestamp: ts(1_775_127_530), + payload: AuditPayload::AliasRetracted { + representative_id: 20, + member_ids: Vec::new(), + actor: ActorId::new("op-bob").expect("non-empty actor"), + reason: String::new(), + }, + } +} + +/// Scenario RFC0005.14 — alias audit events round-trip and back the +/// v1 map derivation (amendment 2026-06-12; the derivation half lives +/// in `ourios-querier`). Per the RFC0005.12 pattern: write an +/// `alias_asserted`, an `alias_retracted`, and a `template_widened` +/// event through `AuditWriter`, read back via `AuditReader`, and +/// assert each kind's columns populated / null per §3.7 — the full +/// asserted set verbatim, the empty-list retraction (≠ NULL), the +/// actor, and the `"" ↔ NULL` reason round trip. +#[test] +fn rfc0005_14_alias_audit_events_round_trip() { + // Arrange — one of each alias kind plus a template event in the + // same partition. + let bucket = TempDir::new().unwrap(); + let template = three_variants("acme").remove(0); + let asserted = alias_asserted_event("acme"); + let retracted = alias_retracted_event("acme"); + let events = vec![template.clone(), asserted.clone(), retracted.clone()]; + let partition = audit_partition_for(&events[0]); + let mut writer = AuditWriter::open(bucket.path(), partition.clone()).expect("open"); + writer.append_events(&events).expect("append"); + let written = writer.close().expect("close"); + + // Act + let reader = AuditReader::open_partition(&written.path, partition).expect("open_partition"); + let round_tripped = reader.read_all().expect("read_all"); + + // Assert — full equality: the member set verbatim (order and the + // duplicate preserved), the actor, the non-empty reason; the + // retraction's empty member list reads back as an empty list (the + // decode requires non-NULL for alias kinds, so equality proves it + // was stored as a list, not NULL); its on-disk NULL reason decodes + // to the in-memory empty string. + assert_eq!(round_tripped, events); + + // And the §3.8-rule-6 per-kind NULL discipline at the raw column + // level: the template row's alias_* columns are NULL, the alias + // rows' template / compaction columns are NULL. + let file = std::fs::File::open(&written.path).expect("open raw"); + let batches: Vec<_> = + parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder::try_new(file) + .expect("builder") + .build() + .expect("reader") + .collect::>() + .expect("batches"); + assert_eq!(batches.len(), 1); + let batch = &batches[0]; + let null_at = |column: &str, row: usize| { + use arrow_array::Array; + let idx = batch.schema().index_of(column).expect("column exists"); + batch.column(idx).is_null(row) + }; + // Row 0 is the template event; rows 1–2 are the alias events. + for column in [ + audit_columns::ALIAS_REPRESENTATIVE_ID, + audit_columns::ALIAS_MEMBER_IDS, + audit_columns::ALIAS_ACTOR, + ] { + assert!( + null_at(column, 0), + "{column} must be NULL on a template row" + ); + assert!(!null_at(column, 1), "{column} must be set on an alias row"); + assert!(!null_at(column, 2), "{column} must be set on an alias row"); + } + for column in [ + audit_columns::TEMPLATE_ID, + audit_columns::OLD_VERSION, + audit_columns::NEW_VERSION, + audit_columns::OLD_TEMPLATE, + audit_columns::NEW_TEMPLATE, + audit_columns::POSITIONS_WIDENED, + audit_columns::SLOTS_EXPANDED, + audit_columns::TRIGGERING_LINE_HASH, + audit_columns::COMPACTION_PARTITION, + audit_columns::COMPACTION_INPUT_FILES, + audit_columns::COMPACTION_OUTPUT_FILE, + audit_columns::COMPACTION_GENERATION, + audit_columns::COMPACTION_ROWS, + ] { + assert!(null_at(column, 1), "{column} must be NULL on an alias row"); + assert!(null_at(column, 2), "{column} must be NULL on an alias row"); + } + // The `"" ↔ NULL` reason rule on disk: non-empty reason stored, + // empty reason stored as NULL. + assert!(!null_at(audit_columns::REASON, 1)); + assert!(null_at(audit_columns::REASON, 2)); +} + /// RFC0005.12 — a `compaction` audit event round-trips with its /// `compaction_*` columns populated and the template columns NULL, /// and a template event in the same file keeps the inverse (the diff --git a/crates/ourios-parquet/tests/schema_pin.rs b/crates/ourios-parquet/tests/schema_pin.rs index 30124c1e6..1bd0e8914 100644 --- a/crates/ourios-parquet/tests/schema_pin.rs +++ b/crates/ourios-parquet/tests/schema_pin.rs @@ -177,6 +177,15 @@ fn rfc0005_10_audit_schema_matches_pinned_field_list() { Field::new("compaction_output_file", DataType::Utf8, true), Field::new("compaction_generation", DataType::UInt64, true), Field::new("compaction_rows", DataType::UInt64, true), + // Alias columns (RFC 0001 §6.7 / RFC 0005 §3.7 amendment + // 2026-06-12): OPTIONAL, NULL for all other kinds. + Field::new("alias_representative_id", DataType::UInt64, true), + Field::new( + "alias_member_ids", + DataType::List(Arc::new(Field::new("element", DataType::UInt64, false))), + true, + ), + Field::new("alias_actor", DataType::Utf8, true), ]; check_schema_against(&expected, &audit_schema()); } diff --git a/crates/ourios-querier/src/alias_store.rs b/crates/ourios-querier/src/alias_store.rs new file mode 100644 index 000000000..f026ecc95 --- /dev/null +++ b/crates/ourios-querier/src/alias_store.rs @@ -0,0 +1,83 @@ +//! v1 reader-side alias-map derivation (RFC 0005 §3.7.1). +//! +//! There is **no persisted per-tenant alias-map artifact** in v1: the +//! audit stream *is* the alias store, and the querier derives the +//! requesting tenant's [`AliasMap`] at query-compile time by folding +//! the tenant's `alias_asserted` / `alias_retracted` events (RFC 0001 +//! §6.7) off the RFC 0005 §3.7 audit Parquet stream. The fold order is +//! total and deterministic — `(timestamp, file path lexicographic, +//! within-file row index)` — pinned by §3.7.1 so same-nanosecond ties +//! fold identically across re-scans. The fold semantics themselves +//! (union-on-overlap, retraction removes the asserted set's ids, +//! canonical = `min(members)`) are owned by RFC 0001 §6.7 and +//! implemented by [`ourios_core::alias::AliasMap`]; this module only +//! feeds it the ordered event stream. +//! +//! The derived map reflects exactly the alias events durably flushed +//! to the audit stream at scan time — the RFC 0001 §6.7 +//! eventual-consistency stance, with the staleness window being +//! audit-flush visibility. A future materialized per-tenant cache +//! (the RFC 0009 §3.4 manifest fork) would accelerate, not change, +//! this derivation. + +use std::path::Path; + +use ourios_core::alias::AliasMap; +use ourios_core::audit::{AuditEvent, AuditPayload}; +use ourios_core::tenant::TenantId; +use ourios_parquet::AuditReader; + +use crate::{QueryError, audit_scan}; + +/// Fold `tenant`'s alias map from its audit stream under `bucket_root` +/// per RFC 0005 §3.7.1. A tenant with no audit files (or none carrying +/// alias events) derives the empty map — every id then resolves to +/// itself. +/// +/// Alias events are rare operator actions, not ingest-volume data, so +/// the unwindowed scan is small by construction (§3.7.1); no day prune +/// applies because the fold covers the tenant's whole alias history. +pub(crate) fn derive_alias_map( + bucket_root: &Path, + tenant: &TenantId, +) -> Result { + // Lexicographic file order from the shared walk + in-file row order + // from the reader give the §3.7.1 tiebreak components… + let files = audit_scan::audit_files(bucket_root, tenant, None)?; + let mut events: Vec = Vec::new(); + for path in &files { + let read = AuditReader::open_file(path) + .and_then(AuditReader::read_all) + .map_err(|e| QueryError::Storage { + detail: format!("audit file {}: {e}", path.display()), + })?; + for event in read { + // Row-level tenant backstop (`CLAUDE.md` §3.7 / RFC 0005 + // §3.9 row-vs-path): the walk is already rooted at the + // tenant's partition, so a row claiming another tenant is + // a corrupt or foreign file — fail loudly rather than + // fold (or silently drop) it. + if event.tenant_id != *tenant { + return Err(QueryError::Storage { + detail: format!( + "audit file {} carries a row for tenant {} under tenant {}'s \ + partition root", + path.display(), + event.tenant_id.as_str(), + tenant.as_str(), + ), + }); + } + if matches!( + event.payload, + AuditPayload::AliasAsserted { .. } | AuditPayload::AliasRetracted { .. } + ) { + events.push(event); + } + } + } + // …and the stable sort by event time completes the total order: + // same-timestamp events keep their (file path, row index) order. + events.sort_by_key(|e| e.timestamp); + Ok(AliasMap::from_events(&events)) +} diff --git a/crates/ourios-querier/src/audit_scan.rs b/crates/ourios-querier/src/audit_scan.rs new file mode 100644 index 000000000..0f48cf544 --- /dev/null +++ b/crates/ourios-querier/src/audit_scan.rs @@ -0,0 +1,241 @@ +//! Shared scan over a tenant's RFC 0005 `audit/` partition subtree. +//! +//! Both audit-stream consumers — the RFC 0010 drift query +//! ([`crate::drift`]) and the RFC 0005 §3.7.1 alias-map derivation +//! ([`crate::alias_store`]) — resolve their file set through this one +//! walk so the tenancy guarantees stay in a single place: +//! +//! - **Tenant isolation is the partition root** (`CLAUDE.md` §3.7 / +//! RFC0010.4): the walk is rooted at `audit/tenant_id=/`, so no +//! other tenant's events are reachable by construction. +//! - **Canonical-path escape backstop**: every resolved `*.parquet` +//! must canonicalize *under* the tenant's canonical root — a +//! symlinked file resolving into another tenant's tree fails loudly +//! rather than being read. +//! - **Optional day-granularity window prune** (RFC 0005 §3.4 — the +//! audit layout has no `hour` segment): with a window, out-of-range +//! `day=…` leaves are skipped before they are listed. The prune is +//! conservative (an unparseable dir is never pruned) and the +//! row-level `timestamp` predicate stays the correctness authority. +//! The alias derivation passes no window — it folds the tenant's +//! whole alias history. + +use std::path::{Path, PathBuf}; + +use ourios_core::tenant::TenantId; +use ourios_parquet::percent_encode_tenant; + +use crate::QueryError; + +/// Resolve the live audit `*.parquet` files for `tenant`, optionally +/// pruned to the day partitions that could hold an event in the +/// half-open `[start, end)` window. Canonical paths are de-duplicated +/// (an in-tenant symlink can't double-count a file) and returned in +/// **lexicographic path order** — the file-path component of the +/// RFC 0005 §3.7.1 total fold order, and stable across re-scans. A +/// missing tenant directory is an empty set, not an error; any other +/// I/O failure surfaces as [`QueryError::Storage`] rather than being +/// masked as "no data". +pub(crate) fn audit_files( + bucket_root: &Path, + tenant: &TenantId, + window: Option<(u64, u64)>, +) -> Result, QueryError> { + let io_err = |op: &str, p: &Path, e: &std::io::Error| QueryError::Storage { + detail: format!("{op} {}: {e}", p.display()), + }; + let enc = percent_encode_tenant(tenant.as_str()); + let tenant_dir = bucket_root.join("audit").join(format!("tenant_id={enc}")); + + let mut files = Vec::new(); + let mut stack = vec![tenant_dir.clone()]; + while let Some(dir) = stack.pop() { + // Day-granularity partition prune (RFC 0005 §3.4 / RFC 0010 + // §6.5): an out-of-window `day=…` leaf is skipped *before* it + // is listed, so its footers are never opened. + // `day_partition_in_window` is conservative — a non-leaf or + // unparseable dir (`year=`, `month=`, `tenant_id=`) is never + // pruned, so the walk still descends to the leaves; only a + // `day=` leaf whose `[day_start, day_start + 1d)` UTC span + // misses `[start, end)` is dropped. + if let Some((start, end)) = window + && !day_partition_in_window(&dir, start, end) + { + continue; + } + let entries = match std::fs::read_dir(&dir) { + Ok(entries) => entries, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue, + Err(e) => return Err(io_err("read_dir", &dir, &e)), + }; + for entry in entries { + let entry = entry.map_err(|e| io_err("read_dir entry", &dir, &e))?; + let path = entry.path(); + match entry.file_type() { + Ok(ft) if ft.is_dir() => stack.push(path), + // `*.parquet.tmp` has extension `tmp`, so an uncommitted / + // crashed writer's temp file contributes nothing. + Ok(_) if path.extension().is_some_and(|x| x == "parquet") => files.push(path), + Ok(_) => {} + Err(e) => return Err(io_err("file_type", &path, &e)), + } + } + } + if files.is_empty() { + return Ok(files); + } + // Tenant-isolation backstop (RFC0010.4 / §3.7), mirroring the log + // path: every resolved file must canonicalize *under* the tenant's + // canonical `audit/tenant_id=…` root. The directory walk is already + // partition-local, but a symlinked `*.parquet` could resolve into + // another tenant's tree — this `starts_with` check fails such a path + // loudly rather than reading another tenant's audit events. + let tenant_root = tenant_dir + .canonicalize() + .map_err(|e| io_err("canonicalize", &tenant_dir, &e))?; + // The trust anchor is the bucket root, not the tenant dir itself: + // if `audit/tenant_id=…` (or `audit/`) were a symlink into another + // tenant's subtree, canonicalizing it as the root would make every + // foreign file pass `starts_with`. Requiring the canonical tenant + // dir to equal the path constructed under the canonical bucket + // root rejects a symlinked tenant root outright. + let bucket_canonical = bucket_root + .canonicalize() + .map_err(|e| io_err("canonicalize", bucket_root, &e))?; + let expected_root = bucket_canonical + .join("audit") + .join(format!("tenant_id={enc}")); + if tenant_root != expected_root { + return Err(QueryError::Storage { + detail: format!( + "audit tenant root {} resolves outside its expected partition path {}", + tenant_root.display(), + expected_root.display(), + ), + }); + } + // De-duplicate the canonical paths (mirroring the log path in + // `lib.rs`): two names resolving to the same file — e.g. an + // in-tenant symlink — must not be read or counted twice. + let mut seen = std::collections::HashSet::new(); + let mut validated = Vec::with_capacity(files.len()); + for file in files { + let abs = file + .canonicalize() + .map_err(|e| io_err("canonicalize", &file, &e))?; + if !abs.starts_with(&tenant_root) { + return Err(QueryError::Storage { + detail: format!( + "resolved audit file {} escapes tenant partition root {}", + abs.display(), + tenant_root.display(), + ), + }); + } + if seen.insert(abs.clone()) { + validated.push(abs); + } + } + // Lexicographic path order: deterministic regardless of the walk's + // stack order, and the §3.7.1 same-timestamp tiebreak across files. + validated.sort(); + Ok(validated) +} + +/// One day in nanoseconds — the span a `…/day=DD/` audit partition covers. +const DAY_NANOS: u64 = 86_400_000_000_000; + +/// Whether the day partition at `dir` could hold an event in the half-open +/// window `[start, end)`. Returns `true` (do not prune) whenever the trailing +/// `year/month/day` segments don't parse or aren't a real UTC instant, so a +/// query never drops in-window data on an unrecognised layout. +fn day_partition_in_window(dir: &Path, start: u64, end: u64) -> bool { + let Some((year, month, day)) = parse_day_partition(dir) else { + return true; + }; + let Some((lo, hi)) = day_span_ns(year, month, day) else { + return true; + }; + lo < end && start < hi +} + +/// Parse `(year, month, day)` from the trailing three Hive segments of an audit +/// partition directory. `None` if the deepest three components aren't `day=`, +/// `month=`, `year=` with parseable numbers (a non-leaf dir or a foreign path). +fn parse_day_partition(dir: &Path) -> Option<(i32, u32, u32)> { + let mut segments = dir.components().rev().filter_map(|c| match c { + std::path::Component::Normal(s) => s.to_str(), + _ => None, + }); + let day = segments.next()?.strip_prefix("day=")?.parse().ok()?; + let month = segments.next()?.strip_prefix("month=")?.parse().ok()?; + let year = segments.next()?.strip_prefix("year=")?.parse().ok()?; + Some((year, month, day)) +} + +/// The `[start, end)` UTC-nanosecond span of the day partition. `None` if it +/// isn't a real UTC instant or predates the 1970 epoch. +fn day_span_ns(year: i32, month: u32, day: u32) -> Option<(u64, u64)> { + let start = chrono::NaiveDate::from_ymd_opt(year, month, day)? + .and_hms_opt(0, 0, 0)? + .and_utc() + .timestamp_nanos_opt()?; + let lo = u64::try_from(start).ok()?; + Some((lo, lo.saturating_add(DAY_NANOS))) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// `day=02` on 2026-04 covers [00:00, next-00:00) UTC. + const DAY_START: u64 = 1_775_088_000_000_000_000; // 2026-04-02T00:00:00Z + + fn day_dir() -> PathBuf { + [ + "bucket", + "audit", + "tenant_id=t", + "year=2026", + "month=04", + "day=02", + ] + .iter() + .collect() + } + + #[test] + fn day_partition_prune_overlap_cases() { + let dir = day_dir(); + // A window inside the day overlaps → keep. + assert!(day_partition_in_window( + &dir, + DAY_START + 3_600_000_000_000, + DAY_START + 7_200_000_000_000, + )); + // A window touching the day's start (half-open, inclusive lo) → keep. + assert!(day_partition_in_window(&dir, DAY_START, DAY_START + 1)); + // A window entirely before the day → prune. + assert!(!day_partition_in_window( + &dir, + DAY_START - 7_200_000_000_000, + DAY_START - 3_600_000_000_000, + )); + // A window starting exactly at the day's end is excluded (half-open + // upper bound) → prune. + assert!(!day_partition_in_window( + &dir, + DAY_START + DAY_NANOS, + DAY_START + DAY_NANOS + 1, + )); + } + + #[test] + fn day_partition_prune_is_conservative_on_unparseable_paths() { + // A non-leaf / foreign path can't be proven out of range → never pruned. + let tenant_dir: PathBuf = ["bucket", "audit", "tenant_id=t"].iter().collect(); + assert!(day_partition_in_window(&tenant_dir, 0, 1)); + let foreign: PathBuf = ["some", "other", "dir"].iter().collect(); + assert!(day_partition_in_window(&foreign, 0, 1)); + } +} diff --git a/crates/ourios-querier/src/compile.rs b/crates/ourios-querier/src/compile.rs index e98dd860c..7d1acba6d 100644 --- a/crates/ourios-querier/src/compile.rs +++ b/crates/ourios-querier/src/compile.rs @@ -90,13 +90,17 @@ const NS_PER_SECOND: u64 = 1_000_000_000; /// stage, or the tenant default `[now - W, now]` when absent — RFC 0002 §4 /// P5, never unbounded) and capture the predicate + `limit` for deferred /// `Expr` building. -pub(crate) fn compile( +/// The map-independent half of [`compile`]: stage support, window +/// resolution, and the limit bound. `run_query` calls this *before* +/// deriving the alias map so an invalid query fails with its compile +/// error rather than first paying (or surfacing errors from) the +/// audit-tree scan; `compile` runs it again internally — it is pure +/// and cheap, and one source of truth beats a split. +pub(crate) fn validate( query: &Query, - tenant: &TenantId, now_unix_nano: u64, default_window_nanos: u64, - alias_map: &AliasMap, -) -> Result { +) -> Result<((u64, u64), Option), QueryError> { // This slice executes only the `range` (time window) and `limit` stages. // The aggregation / sort / projection / render stages parse into a valid // IR but are not yet wired to execution; reject them explicitly so a @@ -128,6 +132,17 @@ pub(crate) fn compile( })?), None => None, }; + Ok((window, limit)) +} + +pub(crate) fn compile( + query: &Query, + tenant: &TenantId, + now_unix_nano: u64, + default_window_nanos: u64, + alias_map: &AliasMap, +) -> Result { + let (window, limit) = validate(query, now_unix_nano, default_window_nanos)?; // Eagerly resolve every `resolves_to(n)` against the tenant's alias map // so the deferred predicate compilation in `apply` is tenant-agnostic. let mut alias_classes = BTreeMap::new(); @@ -141,6 +156,21 @@ pub(crate) fn compile( }) } +/// Whether the predicate contains any `resolves_to(n)` call. The caller uses +/// this to skip the RFC 0005 §3.7.1 alias-map derivation (an audit-tree scan) +/// for the queries that would never consult the map. +pub(crate) fn uses_resolves_to(p: &Predicate) -> bool { + match p { + Predicate::Call(Call::ResolvesTo(_)) => true, + Predicate::Not(inner) => uses_resolves_to(inner), + Predicate::And(terms) | Predicate::Or(terms) => terms.iter().any(uses_resolves_to), + Predicate::Bool(_) + | Predicate::Comparison { .. } + | Predicate::Severity { .. } + | Predicate::Call(_) => false, + } +} + /// Walk the predicate IR and, for each `resolves_to(n)`, record the tenant's /// alias expansion `n → resolves(tenant, n)` (RFC 0001 §6.7). Per-tenant /// resolution `[§3.7]` happens here once; the result rides the [`Plan`]. diff --git a/crates/ourios-querier/src/drift.rs b/crates/ourios-querier/src/drift.rs index b7c076891..6182f5d0d 100644 --- a/crates/ourios-querier/src/drift.rs +++ b/crates/ourios-querier/src/drift.rs @@ -14,9 +14,11 @@ //! other tenant's events are reachable. The window drives a day-granularity //! `year/month/day` partition prune (RFC 0005 §3.4 — the audit layout has no //! `hour` segment), then an exact `timestamp` predicate trims the boundary -//! days to the half-open `[from, to)` window (RFC 0010 §6.5). +//! days to the half-open `[from, to)` window (RFC 0010 §6.5). The walk — +//! tenant root, canonical-path escape backstop, day prune — is the shared +//! [`crate::audit_scan`], also used by the §3.7.1 alias-map derivation. -use std::path::{Path, PathBuf}; +use std::path::Path; use std::sync::Arc; use std::time::{Duration, SystemTime}; @@ -33,10 +35,10 @@ use datafusion::prelude::{SessionContext, col, lit}; use ourios_core::audit::{EVENT_TYPE_TEMPLATE_TYPE_EXPANDED, EVENT_TYPE_TEMPLATE_WIDENED}; use ourios_core::tenant::TenantId; -use ourios_parquet::{audit_columns, percent_encode_tenant}; +use ourios_parquet::audit_columns; use crate::dsl::DriftQuery; -use crate::{QueryError, QueryStats, scan_stats, storage_err, time_bound_scalar}; +use crate::{QueryError, QueryStats, audit_scan, scan_stats, storage_err, time_bound_scalar}; /// One drift row: a template that gained at least one version in the queried /// window, with the §6.3 aggregates. The columns map one-to-one onto RFC 0010 @@ -96,7 +98,7 @@ pub(crate) async fn run_drift( // (RFC0010.5). return Ok(DriftResult::default()); } - let files = audit_files_in_window(bucket_root, tenant, start, end)?; + let files = audit_scan::audit_files(bucket_root, tenant, Some((start, end)))?; if files.is_empty() { // No audit files for the window ⇒ empty drift result, not an error // (RFC0010.5). @@ -175,137 +177,6 @@ fn resolve_window(query: &DriftQuery, now: u64) -> Result<(u64, u64), QueryError Ok((from.min(to), from.max(to))) } -/// Resolve the live audit `*.parquet` files for `tenant` whose day partition -/// could hold an event in `[start, end)`. Tenancy is the partition root -/// (RFC0010.4); the day-granularity window prune (RFC 0005 §3.4) skips whole -/// `day=…` partitions that can't overlap the window — the directory is not even -/// listed, so no footer there is opened. Canonical paths are de-duplicated so a -/// symlink can't double-count a file. A missing tenant directory is an empty -/// set (RFC0010.5), not an error; any other I/O failure is surfaced as -/// [`QueryError::Storage`] rather than masked as "no drift". -fn audit_files_in_window( - bucket_root: &Path, - tenant: &TenantId, - start: u64, - end: u64, -) -> Result, QueryError> { - let io_err = |op: &str, p: &Path, e: &std::io::Error| QueryError::Storage { - detail: format!("{op} {}: {e}", p.display()), - }; - let enc = percent_encode_tenant(tenant.as_str()); - let tenant_dir = bucket_root.join("audit").join(format!("tenant_id={enc}")); - - let mut files = Vec::new(); - let mut stack = vec![tenant_dir.clone()]; - while let Some(dir) = stack.pop() { - // Day-granularity partition prune (RFC 0005 §3.4 / RFC 0010 §6.5): - // an out-of-window `day=…` leaf is skipped *before* it is listed, so - // its footers are never opened. `day_partition_in_window` is - // conservative — a non-leaf or unparseable dir (`year=`, `month=`, - // `tenant_id=`) is never pruned, so the walk still descends to the - // leaves; only a `day=` leaf whose `[day_start, day_start + 1d)` UTC - // span misses `[start, end)` is dropped. The row-level `timestamp` - // predicate stays the correctness authority. - if !day_partition_in_window(&dir, start, end) { - continue; - } - let entries = match std::fs::read_dir(&dir) { - Ok(entries) => entries, - Err(e) if e.kind() == std::io::ErrorKind::NotFound => continue, - Err(e) => return Err(io_err("read_dir", &dir, &e)), - }; - for entry in entries { - let entry = entry.map_err(|e| io_err("read_dir entry", &dir, &e))?; - let path = entry.path(); - match entry.file_type() { - Ok(ft) if ft.is_dir() => stack.push(path), - // `*.parquet.tmp` has extension `tmp`, so an uncommitted / - // crashed writer's temp file contributes nothing. - Ok(_) if path.extension().is_some_and(|x| x == "parquet") => files.push(path), - Ok(_) => {} - Err(e) => return Err(io_err("file_type", &path, &e)), - } - } - } - if files.is_empty() { - return Ok(files); - } - // Tenant-isolation backstop (RFC0010.4 / §3.7), mirroring the log - // path: every resolved file must canonicalize *under* the tenant's - // canonical `audit/tenant_id=…` root. The directory walk is already - // partition-local, but a symlinked `*.parquet` could resolve into - // another tenant's tree — this `starts_with` check fails such a path - // loudly rather than reading another tenant's audit events. - let tenant_root = tenant_dir - .canonicalize() - .map_err(|e| io_err("canonicalize", &tenant_dir, &e))?; - // De-duplicate the canonical paths (mirroring the log path in `lib.rs`): - // two names resolving to the same file — e.g. an in-tenant symlink — must - // not be read or counted twice. - let mut seen = std::collections::HashSet::new(); - let mut validated = Vec::with_capacity(files.len()); - for file in files { - let abs = file - .canonicalize() - .map_err(|e| io_err("canonicalize", &file, &e))?; - if !abs.starts_with(&tenant_root) { - return Err(QueryError::Storage { - detail: format!( - "resolved audit file {} escapes tenant partition root {}", - abs.display(), - tenant_root.display(), - ), - }); - } - if seen.insert(abs.clone()) { - validated.push(abs); - } - } - Ok(validated) -} - -/// One day in nanoseconds — the span a `…/day=DD/` audit partition covers. -const DAY_NANOS: u64 = 86_400_000_000_000; - -/// Whether the day partition at `dir` could hold an event in the half-open -/// window `[start, end)`. Returns `true` (do not prune) whenever the trailing -/// `year/month/day` segments don't parse or aren't a real UTC instant, so a -/// query never drops in-window data on an unrecognised layout. -fn day_partition_in_window(dir: &Path, start: u64, end: u64) -> bool { - let Some((year, month, day)) = parse_day_partition(dir) else { - return true; - }; - let Some((lo, hi)) = day_span_ns(year, month, day) else { - return true; - }; - lo < end && start < hi -} - -/// Parse `(year, month, day)` from the trailing three Hive segments of an audit -/// partition directory. `None` if the deepest three components aren't `day=`, -/// `month=`, `year=` with parseable numbers (a non-leaf dir or a foreign path). -fn parse_day_partition(dir: &Path) -> Option<(i32, u32, u32)> { - let mut segments = dir.components().rev().filter_map(|c| match c { - std::path::Component::Normal(s) => s.to_str(), - _ => None, - }); - let day = segments.next()?.strip_prefix("day=")?.parse().ok()?; - let month = segments.next()?.strip_prefix("month=")?.parse().ok()?; - let year = segments.next()?.strip_prefix("year=")?.parse().ok()?; - Some((year, month, day)) -} - -/// The `[start, end)` UTC-nanosecond span of the day partition. `None` if it -/// isn't a real UTC instant or predates the 1970 epoch. -fn day_span_ns(year: i32, month: u32, day: u32) -> Option<(u64, u64)> { - let start = chrono::NaiveDate::from_ymd_opt(year, month, day)? - .and_hms_opt(0, 0, 0)? - .and_utc() - .timestamp_nanos_opt()?; - let lo = u64::try_from(start).ok()?; - Some((lo, lo.saturating_add(DAY_NANOS))) -} - /// Decode the grouped+sorted aggregate batches into [`DriftRow`]s, preserving /// the engine's row order (the §6.3 `Sort`). The schema is fixed by the /// projection above (`template_id`, then the five named aggregates), so a @@ -441,57 +312,6 @@ mod tests { assert_eq!(lo, now - 3_600 * 1_000_000_000); } - /// `day=02` on 2026-04 covers [00:00, next-00:00) UTC. - const DAY_START: u64 = 1_775_088_000_000_000_000; // 2026-04-02T00:00:00Z - - fn day_dir() -> PathBuf { - [ - "bucket", - "audit", - "tenant_id=t", - "year=2026", - "month=04", - "day=02", - ] - .iter() - .collect() - } - - #[test] - fn day_partition_prune_overlap_cases() { - let dir = day_dir(); - // A window inside the day overlaps → keep. - assert!(day_partition_in_window( - &dir, - DAY_START + 3_600_000_000_000, - DAY_START + 7_200_000_000_000, - )); - // A window touching the day's start (half-open, inclusive lo) → keep. - assert!(day_partition_in_window(&dir, DAY_START, DAY_START + 1)); - // A window entirely before the day → prune. - assert!(!day_partition_in_window( - &dir, - DAY_START - 7_200_000_000_000, - DAY_START - 3_600_000_000_000, - )); - // A window starting exactly at the day's end is excluded (half-open - // upper bound) → prune. - assert!(!day_partition_in_window( - &dir, - DAY_START + DAY_NANOS, - DAY_START + DAY_NANOS + 1, - )); - } - - #[test] - fn day_partition_prune_is_conservative_on_unparseable_paths() { - // A non-leaf / foreign path can't be proven out of range → never pruned. - let tenant_dir: PathBuf = ["bucket", "audit", "tenant_id=t"].iter().collect(); - assert!(day_partition_in_window(&tenant_dir, 0, 1)); - let foreign: PathBuf = ["some", "other", "dir"].iter().collect(); - assert!(day_partition_in_window(&foreign, 0, 1)); - } - #[test] fn decode_drift_rows_rejects_a_null_group_key() { use datafusion::arrow::datatypes::{DataType, Field, Schema, TimeUnit}; diff --git a/crates/ourios-querier/src/lib.rs b/crates/ourios-querier/src/lib.rs index 630d49b9e..8c31e5833 100644 --- a/crates/ourios-querier/src/lib.rs +++ b/crates/ourios-querier/src/lib.rs @@ -37,6 +37,8 @@ #![deny(unsafe_code)] +mod alias_store; +mod audit_scan; mod compile; mod drift; pub mod dsl; @@ -469,15 +471,19 @@ impl Querier { /// `[now - default_window_nanos, now]` (RFC 0002 §4 P5 — **never** an /// unbounded scan). /// - /// `alias_map` is the in-memory RFC 0001 §6.7 alias projection the caller - /// holds for this querier process; `resolves_to(n)` expands through + /// `alias_map` selects where the RFC 0001 §6.7 alias projection comes + /// from. `None` — the production default — derives the requesting + /// tenant's map from its audit stream at compile time per RFC 0005 + /// §3.7.1 (the audit stream is the alias store in v1; the scan is + /// skipped entirely when the query has no `resolves_to`). + /// `Some(map)` injects a caller-held projection instead — the + /// test/operator override, bypassing storage. Either way, + /// `resolves_to(n)` expands through /// [`AliasMap::resolves`](ourios_core::alias::AliasMap::resolves) for /// `tenant`, so a `template_id` an operator aliased matches its whole - /// equivalence class. An id in no class resolves to `{id}` — a singleton + /// equivalence class; an id in no class resolves to `{id}` — a singleton /// `template_id IN (n)`, behaviorally identical to a bare - /// `template_id == n`. The map is a - /// projection injected by the caller; this path adds no on-disk loading - /// (physical storage is the RFC 0005 split, sibling to #147). + /// `template_id == n`. /// /// # Errors /// @@ -490,15 +496,31 @@ impl Querier { tenant: &TenantId, now_unix_nano: u64, default_window_nanos: u64, - alias_map: &ourios_core::alias::AliasMap, + alias_map: Option<&ourios_core::alias::AliasMap>, ) -> Result { - let plan = compile::compile( - query, - tenant, - now_unix_nano, - default_window_nanos, - alias_map, - )?; + // Error precedence: stage-support and window/limit validation + // runs before the alias-map derivation below, so those query + // errors surface without paying the audit-tree IO (or its + // Storage errors). Predicate compilation needs the map, so its + // errors necessarily come after. `compile` re-runs the same + // pure validation internally — one source of truth, negligible + // cost. + compile::validate(query, now_unix_nano, default_window_nanos)?; + let derived; + let map = match alias_map { + Some(map) => map, + None if compile::uses_resolves_to(&query.predicate) => { + derived = alias_store::derive_alias_map(&self.bucket_root, tenant)?; + &derived + } + // No `resolves_to` ⇒ the map is never consulted; an empty + // projection avoids the audit-tree scan. + None => { + derived = ourios_core::alias::AliasMap::new(); + &derived + } + }; + let plan = compile::compile(query, tenant, now_unix_nano, default_window_nanos, map)?; self.execute(tenant, Some(plan.window), move |df| { compile::apply(df, plan) }) diff --git a/crates/ourios-querier/tests/rfc0001_query_semantics.rs b/crates/ourios-querier/tests/rfc0001_query_semantics.rs index c2d4cd302..9c2df24d7 100644 --- a/crates/ourios-querier/tests/rfc0001_query_semantics.rs +++ b/crates/ourios-querier/tests/rfc0001_query_semantics.rs @@ -50,7 +50,7 @@ async fn rfc0001_5_bare_template_id_spans_all_versions_of_leaf() { // resolution is involved — this is by-construction). let query = ourios_querier::dsl::parse(&format!("template_id == {X}")).expect("parse"); let result = q - .run_query(&query, &tenant, NOW, DEFAULT_WINDOW_NS, &no_aliases()) + .run_query(&query, &tenant, NOW, DEFAULT_WINDOW_NS, Some(&no_aliases())) .await .expect("run_query"); @@ -107,7 +107,7 @@ async fn rfc0001_6_bare_template_id_does_not_follow_alias_chains() { let rows = async |text: &str| { let query = ourios_querier::dsl::parse(text).expect("parse"); - q.run_query(&query, &tenant, NOW, DEFAULT_WINDOW_NS, &aliases) + q.run_query(&query, &tenant, NOW, DEFAULT_WINDOW_NS, Some(&aliases)) .await .expect("run_query") .rows diff --git a/crates/ourios-querier/tests/rfc0001_time_preserved.rs b/crates/ourios-querier/tests/rfc0001_time_preserved.rs index 95e8391ca..e066a29ea 100644 --- a/crates/ourios-querier/tests/rfc0001_time_preserved.rs +++ b/crates/ourios-querier/tests/rfc0001_time_preserved.rs @@ -71,7 +71,7 @@ async fn rfc0001_10_time_unix_nano_preserved_verbatim_from_wire() { &tenant, NOW, common::DEFAULT_WINDOW_NS, - &no_aliases(), + Some(&no_aliases()), ) .await .expect("run_query"); diff --git a/crates/ourios-querier/tests/rfc0002_dsl.rs b/crates/ourios-querier/tests/rfc0002_dsl.rs index bc3f93a60..84caa12f2 100644 --- a/crates/ourios-querier/tests/rfc0002_dsl.rs +++ b/crates/ourios-querier/tests/rfc0002_dsl.rs @@ -300,7 +300,7 @@ async fn rfc0002_1_predicate_compiles_to_a_filter() { &tenant, NOW, DEFAULT_WINDOW_NS, - &common::no_aliases(), + Some(&common::no_aliases()), ) .await .expect("run_query"); @@ -410,7 +410,7 @@ async fn rfc0002_3_no_datafusion_leakage() { &tenant, NOW, DEFAULT_WINDOW_NS, - &common::no_aliases(), + Some(&common::no_aliases()), ) .await .expect_err("an out-of-scope attribute comparison must be rejected"); @@ -453,7 +453,7 @@ async fn rfc0002_4_default_time_window() { // Act — a query with NO range stage (match-all predicate). let query = ourios_querier::dsl::parse("true").expect("parse"); let r = q - .run_query(&query, &tenant, now, window, &common::no_aliases()) + .run_query(&query, &tenant, now, window, Some(&common::no_aliases())) .await .expect("run_query"); @@ -514,7 +514,7 @@ async fn rfc0002_5_severity_name_maps_to_severity_number() { &tenant, NOW, DEFAULT_WINDOW_NS, - &common::no_aliases(), + Some(&common::no_aliases()), ) .await .expect("run_query") @@ -593,7 +593,7 @@ async fn rfc0002_5_named_severity_equality_is_a_band() { &tenant, NOW, DEFAULT_WINDOW_NS, - &common::no_aliases(), + Some(&common::no_aliases()), ) .await .expect("run_query") @@ -660,7 +660,7 @@ async fn rfc0002_6_first_class_fields_resolve() { &tenant, NOW, DEFAULT_WINDOW_NS, - &common::no_aliases(), + Some(&common::no_aliases()), ) .await .expect("run_query") @@ -739,7 +739,7 @@ async fn rfc0002_6_attr_not_equal_requires_present_key() { &tenant, NOW, DEFAULT_WINDOW_NS, - &common::no_aliases(), + Some(&common::no_aliases()), ) .await .expect("run_query") @@ -798,7 +798,7 @@ async fn rfc0002_6_non_text_operators_rejected() { &tenant, NOW, DEFAULT_WINDOW_NS, - &common::no_aliases(), + Some(&common::no_aliases()), ) .await .expect_err(&format!("{text:?} must be rejected at compile")); @@ -846,7 +846,7 @@ async fn rfc0002_6_unsupported_stage_rejected() { &tenant, NOW, DEFAULT_WINDOW_NS, - &common::no_aliases(), + Some(&common::no_aliases()), ) .await .expect_err(&format!("{text:?} must be rejected, not dropped")); @@ -860,9 +860,15 @@ async fn rfc0002_6_unsupported_stage_rejected() { // A supported pipeline still runs. let ok = ourios_querier::dsl::parse("body == \"x\" | limit 5").expect("parse"); - q.run_query(&ok, &tenant, NOW, DEFAULT_WINDOW_NS, &common::no_aliases()) - .await - .expect("range + limit stay supported"); + q.run_query( + &ok, + &tenant, + NOW, + DEFAULT_WINDOW_NS, + Some(&common::no_aliases()), + ) + .await + .expect("range + limit stay supported"); } /// Scenario RFC0002.7 — Parse/serialise round-trip is idempotent. @@ -1068,7 +1074,7 @@ async fn rfc0002_9_resolves_to_expands_via_alias_map() { let rows = async |text: &str, tenant: &TenantId, map: &AliasMap| { let query = ourios_querier::dsl::parse(text).expect("parse"); - q.run_query(&query, tenant, NOW, DEFAULT_WINDOW_NS, map) + q.run_query(&query, tenant, NOW, DEFAULT_WINDOW_NS, Some(map)) .await .expect("run_query") .rows @@ -1116,6 +1122,87 @@ async fn rfc0002_9_resolves_to_expands_via_alias_map() { ); } +/// Scenario RFC0002.9, storage-backed (RFC 0005 §3.7.1 / RFC0005.14; +/// issue #148 step 3): the same `resolves_to` expansion as the test +/// above, but with NO injected map — the alias assertion is written +/// to the real RFC 0005 `audit/` stream via the production +/// `ParquetAuditSink`, and the querier DERIVES tenant `T`'s map from +/// storage at compile time. `resolves_to(A)` returns A ∪ {B} while +/// bare `template_id == A` stays exactly A, and the assertion under +/// `T` is invisible to `T2`'s derived map (`CLAUDE.md` §3.7). +#[tokio::test] +async fn rfc0002_9_storage_backed_resolves_to_expands_via_derived_map() { + use common::{DEFAULT_WINDOW_NS, NOW, TS0, at, simple, write_all, write_audit}; + use ourios_core::alias::ActorId; + use ourios_core::audit::{AuditEvent, AuditPayload}; + use ourios_core::tenant::TenantId; + use ourios_querier::Querier; + + // Arrange — the same three-template fixture under T (A, B, C) plus + // the same ids under T2, and ONE alias assertion B ≡ A for T, + // persisted through the audit sink rather than handed in. + const A: u64 = 10; + const B: u64 = 20; + const C: u64 = 30; + let bucket = tempfile::TempDir::new().expect("temp"); + write_all( + bucket.path(), + &[ + simple("T", A, TS0), + simple("T", A, TS0 + 1_000), + simple("T", B, TS0 + common::HOUR_NS), + simple("T", C, TS0 + 2 * common::HOUR_NS), + simple("T2", A, TS0), + simple("T2", B, TS0 + common::HOUR_NS), + ], + ); + write_audit( + bucket.path(), + &[AuditEvent { + tenant_id: TenantId::new("T"), + timestamp: at(TS0), + payload: AuditPayload::AliasAsserted { + representative_id: A, + member_ids: vec![B], + actor: ActorId::new("op-test").expect("actor"), + reason: "deploy re-split the login template".to_string(), + }, + }], + ); + let q = Querier::new(bucket.path()); + let t = TenantId::new("T"); + let t2 = TenantId::new("T2"); + + // No injected map: `None` selects the §3.7.1 storage derivation. + let rows = async |text: &str, tenant: &TenantId| { + let query = ourios_querier::dsl::parse(text).expect("parse"); + q.run_query(&query, tenant, NOW, DEFAULT_WINDOW_NS, None) + .await + .expect("run_query") + .rows + }; + + // Act / Assert — resolves_to(A) expands via the DERIVED {A,B} + // class: 2 A-rows + 1 B-row, C excluded … + assert_eq!( + rows("resolves_to(10)", &t).await, + 3, + "resolves_to(A) expands via the storage-derived map", + ); + // … while bare template_id == A stays exactly A. + assert_eq!( + rows("template_id == 10", &t).await, + 2, + "template_id == A is unaffected by the derived alias class", + ); + // T's stored assertion never folds into T2's derived map. + assert_eq!( + rows("resolves_to(10)", &t2).await, + 1, + "the stored T alias must not leak into T2's derived map", + ); +} + /// Scenario RFC0002.10 — A query is a YAML-safe single-line scalar. /// See `docs/rfcs/0002-query-dsl.md` §5. /// diff --git a/crates/ourios-querier/tests/rfc0005_13.rs b/crates/ourios-querier/tests/rfc0005_13.rs index 7aa4e87f5..4f5c850dd 100644 --- a/crates/ourios-querier/tests/rfc0005_13.rs +++ b/crates/ourios-querier/tests/rfc0005_13.rs @@ -60,7 +60,7 @@ async fn rows_in_window(bucket: &Path, lo: u64, hi: u64) -> u64 { &TenantId::new("a"), NOW, DEFAULT_WINDOW_NS, - &no_aliases(), + Some(&no_aliases()), ) .await .expect("run_query") @@ -219,7 +219,7 @@ async fn rfc0005_13_effective_window_prunes_row_groups() { &TenantId::new("a"), NOW, DEFAULT_WINDOW_NS, - &no_aliases(), + Some(&no_aliases()), ) .await .expect("run_query"); diff --git a/crates/ourios-querier/tests/rfc0005_14_alias_derivation.rs b/crates/ourios-querier/tests/rfc0005_14_alias_derivation.rs new file mode 100644 index 000000000..2a939a65f --- /dev/null +++ b/crates/ourios-querier/tests/rfc0005_14_alias_derivation.rs @@ -0,0 +1,290 @@ +//! Scenario RFC0005.14 — the v1 reader-side alias-map derivation +//! (RFC 0005 §3.7.1, amendment 2026-06-12; the storage round-trip half +//! lives in `ourios-parquet/tests/audit_round_trip.rs`). +//! See `docs/rfcs/0005-parquet-storage.md` §5. +//! +//! Each test writes alias events to the real RFC 0005 `audit/` stream, +//! then runs `resolves_to` through the public [`Querier::run_query`] +//! surface with **no injected map** — so the asserted row counts can +//! only come from the §3.7.1 storage-derived fold. The cross-file +//! same-timestamp tests pin the total fold order's file-path tiebreak +//! by writing one event per file and renaming the files into a crafted +//! lexicographic order (the scan orders by path, not by write time). + +mod common; + +use std::path::{Path, PathBuf}; + +use common::{DEFAULT_WINDOW_NS, HOUR_NS, NOW, TS0, at, simple, write_all}; +use ourios_core::alias::ActorId; +use ourios_core::audit::{AuditEvent, AuditPayload}; +use ourios_core::tenant::TenantId; +use ourios_parquet::{AuditWriter, PartitionKey}; +use ourios_querier::Querier; + +const A: u64 = 10; +const B: u64 = 20; + +fn alias_asserted( + tenant: &str, + representative_id: u64, + member_ids: Vec, + ts: u64, +) -> AuditEvent { + AuditEvent { + tenant_id: TenantId::new(tenant), + timestamp: at(ts), + payload: AuditPayload::AliasAsserted { + representative_id, + member_ids, + actor: ActorId::new("op-test").expect("actor"), + reason: String::new(), + }, + } +} + +fn alias_retracted( + tenant: &str, + representative_id: u64, + member_ids: Vec, + ts: u64, +) -> AuditEvent { + AuditEvent { + tenant_id: TenantId::new(tenant), + timestamp: at(ts), + payload: AuditPayload::AliasRetracted { + representative_id, + member_ids, + actor: ActorId::new("op-test").expect("actor"), + reason: String::new(), + }, + } +} + +/// Write `event` as a single-event audit file and rename it to +/// `final_name` inside its partition directory, so the test controls +/// the lexicographic file order the §3.7.1 tiebreak folds in. +fn write_audit_file_named(bucket: &Path, event: &AuditEvent, final_name: &str) -> PathBuf { + // TS0 is 2026-04-02T10:58Z; the audit partition is tenant + UTC day. + let partition = PartitionKey { + tenant_id: event.tenant_id.as_str().to_owned(), + year: 2026, + month: 4, + day: 2, + hour: 0, + }; + let mut writer = AuditWriter::open(bucket, partition).expect("open audit writer"); + writer + .append_events(std::slice::from_ref(event)) + .expect("append"); + let written = writer.close().expect("close"); + let target = written + .path + .parent() + .expect("partition dir") + .join(final_name); + std::fs::rename(&written.path, &target).expect("rename to crafted order"); + target +} + +/// Three data rows under tenant `T`: two for leaf A, one for leaf B — +/// so `resolves_to(A)` counts 2 without an active alias and 3 with one. +fn write_data_rows(bucket: &Path) { + write_all( + bucket, + &[ + simple("T", A, TS0), + simple("T", A, TS0 + 1_000), + simple("T", B, TS0 + HOUR_NS), + ], + ); +} + +async fn resolves_to_a_rows(bucket: &Path, tenant: &str) -> u64 { + let query = ourios_querier::dsl::parse(&format!("resolves_to({A})")).expect("parse"); + Querier::new(bucket) + .run_query(&query, &TenantId::new(tenant), NOW, DEFAULT_WINDOW_NS, None) + .await + .expect("run_query") + .rows +} + +/// RFC0005.14 — the derived map folds assert→retract in event-time +/// order: an assertion at `t` retracted at `t + 1` leaves no class, so +/// `resolves_to(A)` is back to exactly A's rows. +#[tokio::test] +async fn derived_map_folds_assert_then_retract_by_timestamp() { + // Arrange — data rows plus an assert/retract pair a nanosecond + // apart, in a single audit file (in-file row order is the fold + // order within one file). + let bucket = tempfile::TempDir::new().expect("temp"); + write_data_rows(bucket.path()); + let partition = PartitionKey { + tenant_id: "T".to_owned(), + year: 2026, + month: 4, + day: 2, + hour: 0, + }; + let mut writer = AuditWriter::open(bucket.path(), partition).expect("open"); + writer + .append_events(&[ + alias_asserted("T", A, vec![B], TS0), + alias_retracted("T", A, vec![B], TS0 + 1), + ]) + .expect("append"); + writer.close().expect("close"); + + // Act / Assert — assert-then-retract folds to no class: only A's + // two rows match (RFC 0001 §6.7 via RFC 0005 §3.7.1). + assert_eq!(resolves_to_a_rows(bucket.path(), "T").await, 2); +} + +/// RFC0005.14 — the §3.7.1 cross-file same-timestamp tiebreak: two +/// single-event files carry an assertion and its retraction at the +/// SAME nanosecond, so event time cannot order them — the +/// lexicographic file path must. With the assert in `a.parquet` and +/// the retract in `b.parquet`, the fold ends retracted. +#[tokio::test] +async fn cross_file_same_timestamp_tiebreak_assert_first() { + // Arrange. + let bucket = tempfile::TempDir::new().expect("temp"); + write_data_rows(bucket.path()); + write_audit_file_named( + bucket.path(), + &alias_asserted("T", A, vec![B], TS0), + "a.parquet", + ); + write_audit_file_named( + bucket.path(), + &alias_retracted("T", A, vec![B], TS0), + "b.parquet", + ); + + // Act / Assert — assert folds first (path "a" < "b"), then the + // retraction dissolves the class: only A's rows. + assert_eq!(resolves_to_a_rows(bucket.path(), "T").await, 2); +} + +/// RFC0005.14 — the mirror case: the SAME two events at the SAME +/// nanosecond, but with the retraction in the lexicographically +/// *earlier* file. The fold order flips — retract (a no-op on an +/// empty map) then assert — so the class is active and +/// `resolves_to(A)` expands to A ∪ B. Together with the test above +/// this pins that the outcome is decided by the file-path tiebreak +/// and nothing else. +#[tokio::test] +async fn cross_file_same_timestamp_tiebreak_retract_first() { + // Arrange — identical events, reversed crafted file order. + let bucket = tempfile::TempDir::new().expect("temp"); + write_data_rows(bucket.path()); + write_audit_file_named( + bucket.path(), + &alias_retracted("T", A, vec![B], TS0), + "a.parquet", + ); + write_audit_file_named( + bucket.path(), + &alias_asserted("T", A, vec![B], TS0), + "b.parquet", + ); + + // Act / Assert — the assertion folds last: the {A, B} class is + // active, so A's two rows plus B's one row match. + assert_eq!(resolves_to_a_rows(bucket.path(), "T").await, 3); +} + +/// RFC0005.14 — tenant isolation at the storage layer (`CLAUDE.md` +/// §3.7; RFC0001.14): tenant T2's alias events on disk contribute +/// nothing to T's derived map — the derivation scans only T's +/// `audit/tenant_id=T/` partition root. +#[tokio::test] +async fn second_tenants_alias_events_never_fold_into_the_derived_map() { + // Arrange — T has data rows but NO alias events; T2 asserts the + // very same {A, B} class under its own partition root, with one A + // row and one B row of its own. + let bucket = tempfile::TempDir::new().expect("temp"); + write_data_rows(bucket.path()); + write_all( + bucket.path(), + &[simple("T2", A, TS0), simple("T2", B, TS0 + 1_000)], + ); + write_audit_file_named( + bucket.path(), + &alias_asserted("T2", A, vec![B], TS0), + "a.parquet", + ); + + // Act / Assert — T's derived map is empty: resolves_to(A) is the + // singleton {A}, matching only A's two rows. T2's own map holds + // the {A, B} class, so its A row AND its B row both match — which + // distinguishes a genuinely derived map from an empty one. + assert_eq!(resolves_to_a_rows(bucket.path(), "T").await, 2); + assert_eq!(resolves_to_a_rows(bucket.path(), "T2").await, 2); +} + +/// A symlinked tenant *root* must be rejected outright: if +/// `audit/tenant_id=EVIL` is a symlink into another tenant's subtree, +/// canonicalizing it as the trust anchor would make every foreign file +/// pass the per-file `starts_with` backstop. The scan anchors trust at +/// the bucket root instead and fails loudly (`CLAUDE.md` §3.7). +#[cfg(unix)] +#[tokio::test] +async fn symlinked_tenant_root_is_rejected() { + // Arrange — T's real audit partition holds an alias event; EVIL's + // audit tenant root is a symlink straight to T's. + let bucket = tempfile::TempDir::new().expect("temp"); + write_data_rows(bucket.path()); + write_all(bucket.path(), &[simple("EVIL", A, TS0)]); + write_audit_file_named( + bucket.path(), + &alias_asserted("T", A, vec![B], TS0), + "a.parquet", + ); + let audit = bucket.path().join("audit"); + std::os::unix::fs::symlink(audit.join("tenant_id=T"), audit.join("tenant_id=EVIL")) + .expect("symlink"); + + // Act / Assert — the derivation refuses the scan instead of + // folding T's events into EVIL's map. + let query = ourios_querier::dsl::parse(&format!("resolves_to({A})")).expect("parse"); + let err = Querier::new(bucket.path()) + .run_query(&query, &TenantId::new("EVIL"), NOW, DEFAULT_WINDOW_NS, None) + .await + .expect_err("a symlinked tenant root must not be scanned"); + match err { + ourios_querier::QueryError::Storage { detail } => assert!( + detail.contains("resolves outside its expected partition path"), + "unexpected detail: {detail}", + ), + other => panic!("expected Storage, got {other:?}"), + } +} + +/// Error precedence: an invalid query fails with its compile error +/// *before* the alias-map derivation pays any audit-tree IO — even +/// when that derivation would itself error (here the tenant's audit +/// root is a plain file, so a scan would surface `Storage`). +#[tokio::test] +async fn invalid_query_fails_before_alias_derivation() { + // Arrange — poison the audit tree so any scan errors. + let bucket = tempfile::TempDir::new().expect("temp"); + write_data_rows(bucket.path()); + let audit = bucket.path().join("audit"); + std::fs::create_dir_all(&audit).expect("audit dir"); + std::fs::write(audit.join("tenant_id=T"), b"not a directory").expect("poison"); + + // Act / Assert — `count` parses but is not yet executable; the + // compile error must win over the broken audit tree's Storage + // error, proving validation runs before derivation. + let query = ourios_querier::dsl::parse(&format!("resolves_to({A}) | count by template_id")) + .expect("parse"); + let err = Querier::new(bucket.path()) + .run_query(&query, &TenantId::new("T"), NOW, DEFAULT_WINDOW_NS, None) + .await + .expect_err("unsupported stage must fail"); + assert!( + matches!(err, ourios_querier::QueryError::InvalidQuery { .. }), + "expected InvalidQuery to precede the audit scan's Storage error, got {err:?}", + ); +}