diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index fe202061..e25bdec4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -384,8 +384,8 @@ jobs: set -euo pipefail fga model transform --file model.fga | diff - model.json - # RFC 0047 §5 (RFC0047.1–.9): the OpenFGA resolver, the layer-2 - # visibility two-step and the MCP tool gate on the served binary against a real + # RFC 0047 §5 (RFC0047.1–.11): the OpenFGA resolver, the layer-2 + # visibility two-step, the MCP tool gate, the graph emitter and erasure against a real # `openfga/openfga` container (testcontainers, image pinned by digest in # the tests) loaded with the in-tree model — the same posture as # `dex-oidc`. A required check (in `ci-success`'s `needs`). @@ -407,6 +407,7 @@ jobs: --ignored --exact rfc0047_openfga::rfc0047_1_to_3_resolver_end_to_end rfc0047_visibility::rfc0047_4_to_9_visibility_end_to_end + rfc0047_emitter::rfc0047_10_11_emitter_and_erasure_end_to_end # Emission-time semconv conformance: boot the real `ourios-server`, # point its OTLP export (metrics + the dogfooded logs signal) at diff --git a/crates/ourios-core/src/alias.rs b/crates/ourios-core/src/alias.rs index 198d8f53..9164bfed 100644 --- a/crates/ourios-core/src/alias.rs +++ b/crates/ourios-core/src/alias.rs @@ -402,6 +402,7 @@ impl AliasMap { | AuditPayload::Compaction { .. } | AuditPayload::RecordQuarantined { .. } | AuditPayload::IngestDenied { .. } + | AuditPayload::ConversationErased { .. } | AuditPayload::Unknown { .. } => {} } } diff --git a/crates/ourios-core/src/audit.rs b/crates/ourios-core/src/audit.rs index 8c2ea01d..90360961 100644 --- a/crates/ourios-core/src/audit.rs +++ b/crates/ourios-core/src/audit.rs @@ -245,6 +245,21 @@ pub enum AuditPayload { /// The rejecting token's audit/metric label (RFC 0026 §3.4). token_name: String, }, + /// A conversation was erased from the tenant (RFC 0047 §3.6): its rows + /// dropped by the compaction rewrite of every partition, then its + /// graph tuples deleted — in that order; the event is written after + /// both. System-scoped like [`Self::Compaction`]; the event's + /// `tenant_id` is the tenant the conversation lived in. + ConversationErased { + /// The raw conversation id (the promoted-column value). + conversation_id: String, + /// Partitions rewritten by the erasure pass. + partitions_rewritten: u64, + /// Rows dropped across those rewrites. + rows_dropped: u64, + /// Graph tuples deleted for the conversation object. + tuples_deleted: u64, + }, } /// Stable on-disk `event_kind` ordinals (RFC 0005 §3.7 mapping). @@ -272,6 +287,9 @@ pub const EVENT_KIND_RECORD_QUARANTINED: u8 = 7; /// `ingest_denied` — an authenticated cross-tenant write attempt was /// rejected pre-WAL (RFC 0026 §3.2). pub const EVENT_KIND_INGEST_DENIED: u8 = 8; +/// `conversation_erased` — a conversation's rows were dropped by the +/// compaction rewrite and its graph tuples deleted (RFC 0047 §3.6). +pub const EVENT_KIND_CONVERSATION_ERASED: u8 = 9; /// Canonical `event_type` strings paired with the ordinals above /// (RFC 0005 §3.7 / RFC 0001 §6.4 / RFC 0009 §3.6). @@ -293,6 +311,8 @@ pub const EVENT_TYPE_TEMPLATE_CREATED: &str = "template_created"; pub const EVENT_TYPE_RECORD_QUARANTINED: &str = "record_quarantined"; /// The string form of [`EVENT_KIND_INGEST_DENIED`]. pub const EVENT_TYPE_INGEST_DENIED: &str = "ingest_denied"; +/// The string form of [`EVENT_KIND_CONVERSATION_ERASED`]. +pub const EVENT_TYPE_CONVERSATION_ERASED: &str = "conversation_erased"; /// The `template_version` a leaf is born at (RFC 0017 §3.1). The /// [`TemplateChange::Created`] variant omits a version field — the invariant @@ -320,6 +340,7 @@ impl AuditPayload { Self::Compaction { .. } => EVENT_KIND_COMPACTION, Self::RecordQuarantined { .. } => EVENT_KIND_RECORD_QUARANTINED, Self::IngestDenied { .. } => EVENT_KIND_INGEST_DENIED, + Self::ConversationErased { .. } => EVENT_KIND_CONVERSATION_ERASED, Self::Unknown { event_kind, .. } => *event_kind, } } @@ -344,6 +365,7 @@ impl AuditPayload { Self::Compaction { .. } => EVENT_TYPE_COMPACTION, Self::RecordQuarantined { .. } => EVENT_TYPE_RECORD_QUARANTINED, Self::IngestDenied { .. } => EVENT_TYPE_INGEST_DENIED, + Self::ConversationErased { .. } => EVENT_TYPE_CONVERSATION_ERASED, Self::Unknown { event_type, .. } => event_type, } } @@ -365,6 +387,7 @@ impl AuditPayload { | Self::Compaction { .. } | Self::RecordQuarantined { .. } | Self::IngestDenied { .. } + | Self::ConversationErased { .. } | Self::Unknown { .. } => false, } } diff --git a/crates/ourios-core/src/auth/openfga/client.rs b/crates/ourios-core/src/auth/openfga/client.rs index c82b5b60..947e8fca 100644 --- a/crates/ourios-core/src/auth/openfga/client.rs +++ b/crates/ourios-core/src/auth/openfga/client.rs @@ -39,7 +39,7 @@ const MAX_CACHE_ENTRIES: usize = 4096; const MAX_ERROR_BODY_BYTES: usize = 512; /// One relationship tuple / tuple key on the wire. -#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] pub struct TupleKey { /// `:` or a userset `:#`. pub user: String, @@ -225,6 +225,42 @@ struct WriteBody<'a> { authorization_model_id: Option<&'a str>, } +/// `Read` filter: every tuple on `object` (optionally one `relation`). +#[derive(Serialize)] +struct ReadTupleKey<'a> { + object: &'a str, + #[serde(skip_serializing_if = "Option::is_none")] + relation: Option<&'a str>, +} + +#[derive(Serialize)] +struct ReadBody<'a> { + tuple_key: ReadTupleKey<'a>, + page_size: u32, + #[serde(skip_serializing_if = "Option::is_none")] + continuation_token: Option<&'a str>, +} + +#[derive(Deserialize)] +struct ReadResponse { + #[serde(default)] + tuples: Vec, + #[serde(default)] + continuation_token: Option, +} + +#[derive(Deserialize)] +struct ReadTuple { + key: TupleKey, +} + +/// `Read` page size — `OpenFGA`'s maximum. +const READ_PAGE_SIZE: u32 = 100; +/// The bound on tuples one `read_by_object` returns: an object with more +/// is not a conversation the emitter wrote (a few relations per +/// participant), and an unbounded read is not fail-closed. +const MAX_READ_TUPLES: usize = 10_000; + /// The `OpenFGA` HTTP API over one store. #[derive(Clone)] pub struct OpenFgaClient { @@ -393,6 +429,53 @@ impl OpenFgaClient { Ok(kept) } + /// `Read`: every tuple on `object` (RFC 0047 §3.6 — "no wildcard + /// delete exists": erasure reads the object's tuples, then deletes + /// them). Paginated to completion; bounded. + /// + /// # Errors + /// + /// [`OpenFgaError::Unavailable`] on transport/timeout/non-2xx; + /// [`OpenFgaError::BoundExceeded`] past the read bound. + pub async fn read_by_object(&self, object: &str) -> Result, OpenFgaError> { + let mut tuples = Vec::new(); + let mut continuation: Option = None; + loop { + let body = ReadBody { + tuple_key: ReadTupleKey { + object, + relation: None, + }, + page_size: READ_PAGE_SIZE, + continuation_token: continuation.as_deref(), + }; + let response = self + .post("read", &body)? + .send() + .await + .map_err(|e| transport(&e))?; + let response = ok_status(response).await?; + let bytes = response + .bytes() + .await + .map_err(|e| OpenFgaError::Unavailable(format!("read read: {e}")))?; + let page: ReadResponse = serde_json::from_slice(&bytes) + .map_err(|e| OpenFgaError::Unavailable(format!("decode read: {e}")))?; + for tuple in page.tuples { + if tuples.len() >= MAX_READ_TUPLES { + return Err(OpenFgaError::BoundExceeded { + bound: MAX_READ_TUPLES, + }); + } + tuples.push(tuple.key); + } + match page.continuation_token { + Some(token) if !token.is_empty() => continuation = Some(token), + _ => return Ok(tuples), + } + } + } + /// `Write`: add `writes` and remove `deletes` in one transactional, /// **idempotent** call (`on_duplicate` / `on_missing` = `ignore`, so a /// tuple already present or already gone is not an error). `OpenFGA` diff --git a/crates/ourios-core/src/auth/openfga/mod.rs b/crates/ourios-core/src/auth/openfga/mod.rs index e04f9f17..7e61b5f3 100644 --- a/crates/ourios-core/src/auth/openfga/mod.rs +++ b/crates/ourios-core/src/auth/openfga/mod.rs @@ -215,6 +215,9 @@ pub const DEFAULT_LIST_TIMEOUT_MS: u64 = 2_000; pub const DEFAULT_SERVER_LIST_OBJECTS_DEADLINE_MS: u64 = 3_000; /// The `OpenFGA` object type of a conversation — the one bindable type in v1. pub const CONVERSATION_TYPE: &str = "conversation"; +/// The RFC 0027 MCP tools as graph objects (RFC 0047 §3.5): +/// `tool:/`; the emitter writes their `parent` tuples per tenant. +pub const MCP_TOOL_NAMES: [&str; 3] = ["query_logs", "list_templates", "template_drift"]; impl fmt::Debug for OpenFgaConfig { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { @@ -602,13 +605,19 @@ pub struct TenantObjects { impl TenantObjects { /// The graph objects of `tenant`, or `None` when the tenant id cannot - /// form an object id. + /// form an object id — raw, or once its segment is encoded (the + /// encoding can only grow it; a tenant whose encoded segment plus the + /// `/` separator leaves no room for a conversation id has no + /// tenant-scoped objects). #[must_use] pub fn new(tenant: &str) -> Option { if !is_object_id(tenant) { return None; } let encoded = encode_tenant_segment(tenant); + if encoded.len() + 1 >= MAX_OBJECT_ID_BYTES { + return None; + } Some(Self { tenant_object: format!("{TENANT_TYPE}:{tenant}"), conversation_prefix: format!("{CONVERSATION_TYPE}:{encoded}/"), @@ -635,6 +644,16 @@ impl TenantObjects { format!("{}{id}", self.conversation_prefix) } + /// Whether `conversation:/` is a valid object: `id` must be + /// an object id itself and the combined id half must fit `OpenFGA`'s + /// 256-byte limit. The emitter skips ids that do not. + #[must_use] + pub fn conversation_fits(&self, id: &str) -> bool { + is_object_id(id) + && self.conversation_prefix.len() - CONVERSATION_TYPE.len() - 1 + id.len() + <= MAX_OBJECT_ID_BYTES + } + /// `tool:/`. #[must_use] pub fn tool(&self, name: &str) -> String { @@ -908,6 +927,15 @@ mod tests { for bad in ["", "a b", "a:b", "a#b"] { assert!(TenantObjects::new(bad).is_none(), "{bad:?}"); } + // The encoding may not push the segment past the object-id limit, + // and a conversation id must fit next to it. + let slashes = "/".repeat(90); // 90 raw bytes → 270 encoded + assert!(TenantObjects::new(&slashes).is_none()); + let long = "x".repeat(200); + let t = TenantObjects::new(&long).expect("fits alone"); + assert!(t.conversation_fits("c-1")); + assert!(!t.conversation_fits(&"y".repeat(60)), "201 + 60 > 256"); + assert!(!t.conversation_fits("a b")); } /// The principal vocabulary renders exactly the model's type names. diff --git a/crates/ourios-ingester/src/compactor.rs b/crates/ourios-ingester/src/compactor.rs index 5a40001b..b5aa94b9 100644 --- a/crates/ourios-ingester/src/compactor.rs +++ b/crates/ourios-ingester/src/compactor.rs @@ -9,18 +9,34 @@ //! ([`crate::metrics::CompactionMetrics`]), and hands each result to a //! caller-supplied observer for logging. +#[cfg(feature = "openfga")] +use std::collections::BTreeSet; use std::path::PathBuf; +#[cfg(feature = "openfga")] +use std::sync::Arc; use std::time::{Duration, Instant, SystemTime}; use ourios_core::audit::{AuditEvent, AuditPayload, AuditSink, NoOpAuditSink}; +use ourios_core::record::MinedRecord; use ourios_core::tenant::TenantId; use ourios_parquet::{ - Committed, CompactionError, CompactionPolicy, PartitionKey, PromotedAttributes, Store, - compact_partition_with_promoted, gc_orphans, percent_decode_tenant, plan_candidates, + Committed, CompactionError, CompactionPolicy, PartitionKey, PromotedAttributes, RowHooks, + Store, compact_partition_hooked, gc_orphans, hour_partitions, percent_decode_tenant, + percent_encode_tenant, plan_candidates, }; +#[cfg(feature = "openfga")] +use crate::graph_emitter::GraphEmitter; use crate::metrics::CompactionMetrics; +/// The tuples one sweep derives for the graph (RFC 0047 §3.3). +#[cfg(feature = "openfga")] +type GraphTuples = BTreeSet; +/// Nothing to derive without the graph. +#[cfg(not(feature = "openfga"))] +#[derive(Default)] +struct GraphTuples; + /// Failure during a compaction sweep. #[derive(Debug)] #[non_exhaustive] @@ -111,6 +127,159 @@ pub struct SweepReport { /// Tenants whose planning *errored* are omitted (their candidate /// count is unknown; they're recorded in [`Self::errors`]). pub per_tenant: Vec, + /// RFC 0047 §3.6 erasure requests this sweep acted on, in the order + /// they were processed. + pub erasures: Vec, + /// RFC 0047 §3.3 graph tuples the emitter wrote after this sweep + /// (`0` without an emitter). + pub graph_tuples_emitted: usize, +} + +/// One pending RFC 0047 §3.6 erasure: a durable request marker in the +/// store (`erasure/tenant_id=/conversation=`, written by an +/// operator through [`request_erasure`]) naming the conversation to +/// remove from a tenant. The marker's body carries the phase, so a +/// sweep interrupted after the rewrite retries only the tuple deletion. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ErasureRequest { + /// The tenant the conversation lives in. + pub tenant: String, + /// The raw conversation id (the promoted-column value). + pub conversation_id: String, + /// The marker object's key. + pub marker: String, + /// Where the request stands. + pub phase: ErasurePhase, +} + +/// The two phases of an erasure — rows first, tuples after (RFC 0047 +/// §3.6: a dangling tuple is harmless, a dangling row is a leak). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ErasurePhase { + /// The Parquet rewrite has not completed for every partition. + Rows, + /// Rows are gone; the graph tuples remain to be deleted. + Tuples, +} + +/// What one sweep did for one erasure request. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ErasureOutcome { + /// The request as it was when the sweep picked it up. + pub request: ErasureRequest, + /// Partitions rewritten this sweep (`0` when the rows phase was + /// already done). + pub partitions_rewritten: u64, + /// Rows dropped this sweep. + pub rows_dropped: u64, + /// The phase after this sweep's blocking pass: `Tuples` once every + /// partition rewrote cleanly, else still `Rows` (retried next sweep). + pub phase: ErasurePhase, + /// Graph tuples deleted (set by the async phase; `None` until then). + pub tuples_deleted: Option, + /// Whether the marker was removed — the erasure is complete. + pub finished: bool, +} + +/// The object-store prefix of erasure request markers. +pub const ERASURE_PREFIX: &str = "erasure/"; +const ERASURE_PHASE_ROWS: &[u8] = br#"{"phase":"rows"}"#; +const ERASURE_PHASE_TUPLES: &[u8] = br#"{"phase":"tuples"}"#; + +/// The marker key for erasing `conversation_id` from `tenant`. +#[must_use] +pub fn erasure_marker_key(tenant: &str, conversation_id: &str) -> String { + format!( + "{ERASURE_PREFIX}tenant_id={}/conversation={}", + percent_encode_tenant(tenant), + percent_encode_tenant(conversation_id) + ) +} + +/// Request the erasure of `conversation_id` from `tenant` (RFC 0047 §3.6): +/// writes the durable marker the next sweep acts on. Idempotent — +/// create-if-absent, so a repeated request never resets an erasure already +/// in flight (one whose rows are gone and whose marker is in the `tuples` +/// phase). +/// +/// # Errors +/// +/// [`IngestError::Io`] when the marker cannot be written. +pub fn request_erasure( + store: &Store, + tenant: &str, + conversation_id: &str, +) -> Result<(), IngestError> { + let key = erasure_marker_key(tenant, conversation_id); + match store.put_if_absent_blocking(&key, ERASURE_PHASE_ROWS.to_vec()) { + Ok(()) => Ok(()), + Err(e) if e.is_already_exists() => Ok(()), + Err(e) => Err(store_error("put erasure marker", &key, &e)), + } +} + +/// The pending erasure requests, in deterministic (lexicographic key) +/// order. +/// +/// # Errors +/// +/// [`IngestError::Io`] when the marker prefix cannot be listed or a marker +/// cannot be read — a transient store error never regresses a marker's +/// phase. +pub fn pending_erasures(store: &Store) -> Result, IngestError> { + let mut keys = store + .list_blocking(Some(ERASURE_PREFIX)) + .map_err(|e| store_error("list erasure markers", ERASURE_PREFIX, &e))?; + keys.sort(); + let mut requests = Vec::new(); + for key in keys { + let Some(rest) = key.strip_prefix(ERASURE_PREFIX) else { + continue; + }; + let Some((tenant_segment, conversation_segment)) = rest.split_once('/') else { + continue; + }; + let (Some(tenant), Some(conversation)) = ( + tenant_segment + .strip_prefix("tenant_id=") + .and_then(percent_decode_tenant), + conversation_segment + .strip_prefix("conversation=") + .and_then(percent_decode_tenant), + ) else { + continue; + }; + let body = store + .get_blocking_opt(&key) + .map_err(|e| store_error("read erasure marker", &key, &e))?; + // A marker removed between the listing and this read is finished. + let Some(body) = body else { + continue; + }; + // The marker body is `{"phase": "rows" | "tuples"}`, parsed + // leniently (whitespace, key order) — operator tooling writes these. + let phase = match serde_json::from_slice::(&body) { + Ok(marker) if marker["phase"].as_str().map(str::trim) == Some("tuples") => { + ErasurePhase::Tuples + } + _ => ErasurePhase::Rows, + }; + requests.push(ErasureRequest { + tenant, + conversation_id: conversation, + marker: key, + phase, + }); + } + Ok(requests) +} + +fn store_error(op: &'static str, key: &str, e: &ourios_parquet::StoreError) -> IngestError { + IngestError::Io { + op, + path: PathBuf::from(key), + source: std::io::Error::other(e.to_string()), + } } /// Per-tenant candidate vs. compacted counts for one sweep — the basis @@ -140,7 +309,7 @@ pub struct CompactedFile { /// Run one compaction sweep over `store`, as of wall-clock /// `now_unix_nanos`: for each tenant, select its sealed candidate /// partitions ([`plan_candidates`]) and consolidate each -/// ([`compact_partition_with_promoted`]), accumulating a [`SweepReport`]. +/// ([`compact_partition_hooked`]), accumulating a [`SweepReport`]. /// /// Resilient: a tenant whose planning fails, or a partition whose /// consolidation fails, is recorded in [`SweepReport::errors`] and @@ -172,6 +341,48 @@ pub fn run_sweep( /// # Errors /// /// See [`run_sweep`]. +pub fn run_sweep_with_promoted( + store: &Store, + now_unix_nanos: u64, + policy: &CompactionPolicy, + promoted: &PromotedAttributes, +) -> Result { + run_sweep_hooked( + store, + now_unix_nanos, + policy, + promoted, + &mut SweepHooks::default(), + ) +} + +/// The RFC 0047 hooks a sweep runs with: `observe` sees every row the +/// sweep rewrites, per tenant (the graph feed, §3.3); `erasure_match` +/// decides which rows an erasure request drops (§3.6) — without it, +/// pending erasures are recorded as errors, never silently skipped. +#[derive(Default)] +pub struct SweepHooks<'a> { + /// `(tenant, rows)` for every input file the sweep decodes. + pub observe: Option<&'a mut SweepObserver<'a>>, + /// `(row, conversation_id)` → whether the row belongs to the conversation. + pub erasure_match: Option<&'a ErasureMatch<'a>>, +} + +/// A [`SweepHooks::observe`] callback. +pub type SweepObserver<'a> = dyn FnMut(&str, &[MinedRecord]) + 'a; +/// A [`SweepHooks::erasure_match`] predicate. +pub type ErasureMatch<'a> = dyn Fn(&MinedRecord, &str) -> bool + 'a; + +/// [`run_sweep_with_promoted`] with [`SweepHooks`]: the consolidation +/// pass, then the RFC 0047 §3.6 erasure pass — every pending request in +/// the `Rows` phase rewrites each of its tenant's partitions with the +/// conversation's rows dropped; once every partition rewrote cleanly the +/// marker advances to `Tuples` (the tuple deletion is the async caller's, +/// after this pass — never before the rewrite). +/// +/// # Errors +/// +/// As [`run_sweep`]. // RFC 0038: one span per compaction sweep — coarse and periodic. Opened inside // the callee (the tick `spawn_blocking`s this), and the per-tenant / per-file // loops below stay span-free. @@ -180,11 +391,12 @@ pub fn run_sweep( name = "sweep partitions", fields(otel.kind = "internal") )] -pub fn run_sweep_with_promoted( +pub fn run_sweep_hooked( store: &Store, now_unix_nanos: u64, policy: &CompactionPolicy, promoted: &PromotedAttributes, + hooks: &mut SweepHooks<'_>, ) -> Result { let mut report = SweepReport::default(); for tenant in tenants(store)? { @@ -209,7 +421,18 @@ pub fn run_sweep_with_promoted( partition.year, partition.month, partition.day, partition.hour, )), } - match compact_partition_with_promoted(store, &partition, promoted) { + let tenant_name = tenant.as_str(); + let mut observe = hooks + .observe + .as_deref_mut() + .map(|observe| move |rows: &[MinedRecord]| observe(tenant_name, rows)); + let mut row_hooks = RowHooks { + observe: observe + .as_mut() + .map(|observe| observe as &mut dyn FnMut(&[MinedRecord])), + drop: None, + }; + match compact_partition_hooked(store, &partition, promoted, &mut row_hooks) { Ok(outcome) => { if let Some(committed) = &outcome.committed { report.partitions_compacted += 1; @@ -243,9 +466,90 @@ pub fn run_sweep_with_promoted( partitions_compacted: compacted_here, }); } + erase_pending(store, promoted, hooks.erasure_match, &mut report)?; Ok(report) } +/// The RFC 0047 §3.6 erasure pass (rows phase). +fn erase_pending( + store: &Store, + promoted: &PromotedAttributes, + erasure_match: Option<&ErasureMatch<'_>>, + report: &mut SweepReport, +) -> Result<(), IngestError> { + for request in pending_erasures(store)? { + let mut outcome = ErasureOutcome { + request: request.clone(), + partitions_rewritten: 0, + rows_dropped: 0, + phase: request.phase, + tuples_deleted: None, + finished: false, + }; + if request.phase == ErasurePhase::Rows { + let Some(matches) = erasure_match else { + report.errors.push(format!( + "erase {:?} {:?}: no conversation column configured \ + (auth.openfga.visibility.objects) — request left pending", + request.tenant, request.conversation_id + )); + report.erasures.push(outcome); + continue; + }; + let id = request.conversation_id.as_str(); + let drop = |record: &MinedRecord| matches(record, id); + let mut clean = true; + let partitions = match hour_partitions(store, &request.tenant) { + Ok(partitions) => partitions, + Err(e) => { + report + .errors + .push(format!("erase {:?}: list partitions: {e}", request.tenant)); + report.erasures.push(outcome); + continue; + } + }; + for partition in partitions { + let mut row_hooks = RowHooks { + observe: None, + drop: Some(&drop), + }; + match compact_partition_hooked(store, &partition, promoted, &mut row_hooks) { + Ok(o) => { + if o.committed.is_some() { + outcome.partitions_rewritten += 1; + outcome.rows_dropped += o.rows_dropped; + } + } + Err(e) => { + clean = false; + report.errors.push(format!( + "erase {:?} {:?} {:04}-{:02}-{:02}T{:02}: {e}", + request.tenant, + request.conversation_id, + partition.year, + partition.month, + partition.day, + partition.hour, + )); + } + } + } + if clean { + match store.put_blocking(&request.marker, ERASURE_PHASE_TUPLES.to_vec()) { + Ok(()) => outcome.phase = ErasurePhase::Tuples, + Err(e) => report.errors.push(format!( + "erase {:?} {:?}: advance marker: {e}", + request.tenant, request.conversation_id + )), + } + } + } + report.erasures.push(outcome); + } + Ok(()) +} + /// Raw tenant ids present in the store, decoded from the immediate /// `data/tenant_id=` child common-prefixes /// ([`Store::list_common_prefixes_blocking`], RFC 0019 §3.3), sorted + @@ -289,18 +593,23 @@ pub struct Compactor { /// Defaults to [`NoOpAuditSink`]; set via [`Self::with_audit_sink`] /// (the WAL-backed sink replaces it once `ourios-wal` lands). audit_sink: Box, + /// The RFC 0047 §3.3 graph emitter, when the graph is configured. + #[cfg(feature = "openfga")] + emitter: Option>, } impl std::fmt::Debug for Compactor { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { // `AuditSink` is not `Debug`; name it without its contents. - f.debug_struct("Compactor") - .field("store", &self.store) + let mut d = f.debug_struct("Compactor"); + d.field("store", &self.store) .field("policy", &self.policy) .field("interval", &self.interval) .field("promoted", &self.promoted) - .field("audit_sink", &"Box") - .finish() + .field("audit_sink", &"Box"); + #[cfg(feature = "openfga")] + d.field("emitter", &self.emitter); + d.finish() } } @@ -318,9 +627,21 @@ impl Compactor { interval, promoted: PromotedAttributes::default(), audit_sink: Box::new(NoOpAuditSink::new()), + #[cfg(feature = "openfga")] + emitter: None, } } + /// Feed the RFC 0047 §3.3 graph from every row the sweep rewrites, and + /// complete §3.6 erasures by deleting the conversation's tuples after + /// the rewrite. + #[cfg(feature = "openfga")] + #[must_use] + pub fn with_graph_emitter(mut self, emitter: Arc) -> Self { + self.emitter = Some(emitter); + self + } + /// Set the RFC 0022 promoted attribute set consolidated files re-project /// under (`storage.promoted_attributes`, §3.2/§3.4). #[must_use] @@ -354,20 +675,14 @@ impl Compactor { where F: FnMut(Result), { - // Destructure up front into owned locals: `audit_sink` moves into each - // sweep's blocking task and back out, and `store`/`policy` are used - // inside the loop without holding `self`. let Self { store, policy, interval, promoted, - // Own the audit sink locally so it can move into each sweep's - // blocking task and back out. Its `emit` performs Parquet `put`s - // through the store — now S3 network I/O (RFC 0019 slice 2d) — so it - // must run on the blocking pool alongside the sweep, never on the - // async task where slow S3 would stall the runtime. mut audit_sink, + #[cfg(feature = "openfga")] + emitter, } = self; // Built (and zero-seeded) once, before the loop, so the metric // set is visible to the exporter even before the first sweep. @@ -380,25 +695,15 @@ impl Compactor { ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); loop { ticker.tick().await; - // `Store` is a cheap `Arc` handle; clone it into the blocking task - // (compaction is blocking I/O). `policy` is `Copy`, so the `move` - // closure copies it and the outer binding stays valid each loop. - let store = store.clone(); - let promoted = promoted.clone(); - let (result, elapsed, sink) = tokio::task::spawn_blocking(move || { - let start = Instant::now(); - let result = run_sweep_with_promoted(&store, now_unix_nanos(), &policy, &promoted); - // Emit the committed-compaction audit events here, on the - // blocking pool, since the sink's `put`s are blocking store I/O. - if let Ok(report) = &result { - for event in &report.compaction_events { - audit_sink.emit(event.clone()); - } - } - (result, start.elapsed(), audit_sink) - }) - .await - .expect("compaction sweep task should not panic"); + let (result, elapsed, sink) = sweep_once( + store.clone(), + policy, + promoted.clone(), + audit_sink, + #[cfg(feature = "openfga")] + emitter.clone(), + ) + .await; audit_sink = sink; metrics.record_sweep(&result, elapsed); on_sweep(result); @@ -406,6 +711,185 @@ impl Compactor { } } +/// One full sweep as the daemon runs it: the blocking pass (consolidation, +/// erasure rewrites, compaction audit events) on the blocking pool, then +/// — with an emitter — the async graph phase (write the tuples the pass +/// derived; delete the tuples of every erasure whose rows are gone; then, +/// back on the blocking pool, the `conversation_erased` audit event and +/// the marker removal). Returns the report, the wall-clock spent, and the +/// audit sink handed back. Runs the same way whether called by +/// [`Compactor::run`] or a test. +/// +/// # Panics +/// +/// If a blocking task panics — `run_sweep` returns errors rather than +/// panicking, so this signals a bug, surfaced loudly rather than silently +/// stalling the daemon. +pub async fn sweep_once( + store: Store, + policy: CompactionPolicy, + promoted: PromotedAttributes, + audit_sink: Box, + #[cfg(feature = "openfga")] emitter: Option>, +) -> ( + Result, + Duration, + Box, +) { + let start = Instant::now(); + #[cfg(feature = "openfga")] + let blocking_emitter = emitter.clone(); + let blocking_store = store.clone(); + // `Store` is a cheap `Arc` handle; clone it into the blocking task + // (compaction is blocking I/O). `policy` is `Copy`. The audit sink moves + // into the task and back out: its `emit` performs Parquet `put`s through + // the store — S3 network I/O (RFC 0019) — so it must run on the blocking + // pool, never on the async task where slow S3 would stall the runtime. + #[cfg_attr(not(feature = "openfga"), allow(unused_mut))] + let (mut result, mut audit_sink, tuples) = tokio::task::spawn_blocking(move || { + let mut audit_sink = audit_sink; + let tuples: std::cell::RefCell = std::cell::RefCell::default(); + let result = { + #[cfg(feature = "openfga")] + let tuples_ref = &tuples; + #[cfg(feature = "openfga")] + let mut observe = blocking_emitter.as_ref().map(|emitter| { + let emitter = Arc::clone(emitter); + move |tenant: &str, rows: &[MinedRecord]| { + let mut tuples = tuples_ref.borrow_mut(); + tuples.extend(emitter.derive(tenant, rows)); + tuples.extend(GraphEmitter::tool_tuples(tenant)); + } + }); + #[cfg(feature = "openfga")] + let erasure_match = blocking_emitter.as_ref().map(|emitter| { + let emitter = Arc::clone(emitter); + move |record: &MinedRecord, id: &str| emitter.conversation_matches(record, id) + }); + let mut hooks = SweepHooks { + #[cfg(feature = "openfga")] + observe: observe.as_mut().map(|f| f as &mut SweepObserver<'_>), + #[cfg(feature = "openfga")] + erasure_match: erasure_match.as_ref().map(|f| f as &ErasureMatch<'_>), + #[cfg(not(feature = "openfga"))] + observe: None, + #[cfg(not(feature = "openfga"))] + erasure_match: None, + }; + run_sweep_hooked( + &blocking_store, + now_unix_nanos(), + &policy, + &promoted, + &mut hooks, + ) + }; + if let Ok(report) = &result { + for event in &report.compaction_events { + audit_sink.emit(event.clone()); + } + } + (result, audit_sink, tuples.into_inner()) + }) + .await + .expect("compaction sweep task should not panic"); + + #[cfg(feature = "openfga")] + if let (Ok(report), Some(emitter)) = (&mut result, emitter.as_ref()) { + graph_phase(&store, emitter, report, &mut audit_sink, tuples).await; + } + #[cfg(not(feature = "openfga"))] + let GraphTuples = tuples; + (result, start.elapsed(), audit_sink) +} + +/// The async graph phase of a sweep (RFC 0047 §3.3 / §3.6). +#[cfg(feature = "openfga")] +async fn graph_phase( + store: &Store, + emitter: &Arc, + report: &mut SweepReport, + audit_sink: &mut Box, + tuples: GraphTuples, +) { + if !tuples.is_empty() { + match emitter.emit(&tuples).await { + Ok(written) => report.graph_tuples_emitted = written.tuples, + Err(e) => report.errors.push(format!("graph emit: {e}")), + } + } + let mut completed: Vec<(usize, AuditEvent)> = Vec::new(); + for (index, outcome) in report.erasures.iter_mut().enumerate() { + if outcome.phase != ErasurePhase::Tuples { + continue; + } + let request = &outcome.request; + match emitter + .erase_conversation(&request.tenant, &request.conversation_id) + .await + { + Ok(deleted) => { + outcome.tuples_deleted = Some(deleted); + completed.push(( + index, + AuditEvent { + tenant_id: TenantId::new(&request.tenant), + timestamp: SystemTime::now(), + payload: AuditPayload::ConversationErased { + conversation_id: request.conversation_id.clone(), + partitions_rewritten: outcome.partitions_rewritten, + rows_dropped: outcome.rows_dropped, + tuples_deleted: to_u64(deleted), + }, + }, + )); + } + Err(e) => report.errors.push(format!( + "erase {:?} {:?}: graph tuples: {e} — retried next sweep", + request.tenant, request.conversation_id + )), + } + } + if completed.is_empty() { + return; + } + // Back on the blocking pool for the audit `put`s and the marker + // deletes — after the tuples are gone. + let store = store.clone(); + let markers: Vec<(usize, String, AuditEvent)> = completed + .into_iter() + .map(|(index, event)| (index, report.erasures[index].request.marker.clone(), event)) + .collect(); + let mut sink = std::mem::replace(audit_sink, Box::new(NoOpAuditSink::new())); + let (sink, finished, errors) = tokio::task::spawn_blocking(move || { + let mut finished = Vec::new(); + let mut errors = Vec::new(); + for (index, marker, event) in markers { + // The marker removal is the at-most-once transition: only the + // process that removes it writes the audit event. A marker + // already gone was finished (and audited) elsewhere; a failed + // delete leaves the marker in the `tuples` phase — the next + // sweep repeats the (idempotent) tuple deletion and retries. + match store.delete_blocking(&marker) { + Ok(()) => { + sink.emit(event); + finished.push(index); + } + Err(e) if e.is_not_found() => finished.push(index), + Err(e) => errors.push(format!("erase: remove marker {marker}: {e}")), + } + } + (sink, finished, errors) + }) + .await + .expect("erasure completion task should not panic"); + *audit_sink = sink; + for index in finished { + report.erasures[index].finished = true; + } + report.errors.extend(errors); +} + /// Saturating `usize` → `u64` (lossless on 64-bit; saturates rather /// than truncating on a theoretically wider target). pub(crate) fn to_u64(value: usize) -> u64 { @@ -466,17 +950,17 @@ mod tests { /// A local [`Store`] rooted at `bucket` — the seam every sweep runs /// through (RFC 0019 §3.3). - fn store_at(bucket: &Path) -> Store { + pub(super) fn store_at(bucket: &Path) -> Store { Store::local(bucket).expect("local store") } /// 2026-04-02T10:58:00 UTC (hour 10). - const TS0: u64 = 1_775_127_480_000_000_000; + pub(super) const TS0: u64 = 1_775_127_480_000_000_000; const HOUR: u64 = 3_600_000_000_000; /// Well past hour 10's end + grace. const NOW_SEALED: u64 = TS0 + 2 * HOUR; - fn rec(tenant: &str, template_id: u64, ts_ns: u64) -> MinedRecord { + pub(super) fn rec(tenant: &str, template_id: u64, ts_ns: u64) -> MinedRecord { MinedRecord { tenant_id: TenantId::new(tenant), template_id, @@ -765,3 +1249,431 @@ mod tests { assert_eq!(compacted.expect("sweep ok"), 1); } } + +#[cfg(all(test, feature = "openfga"))] +mod graph_tests { + use std::path::Path; + use std::sync::{Arc, Mutex}; + + use axum::Router; + use axum::extract::State; + use axum::routing::post; + use ourios_core::audit::{AuditPayload, SharedAuditSink}; + use ourios_core::auth::openfga::{OpenFgaSpec, TupleKey, build_openfga_config}; + use ourios_core::otlp::any_value::Value; + use ourios_core::otlp::{AnyValue, KeyValue}; + use ourios_core::record::MinedRecord; + use ourios_parquet::{ + CompactionPolicy, PartitionKey, PromotedAttributes, PromotedKey, Reader, Store, Writer, + }; + use serde_json::json; + + use super::{ErasurePhase, pending_erasures, request_erasure, sweep_once}; + use crate::graph_emitter::GraphEmitter; + + /// A fake `OpenFGA` store: `/write` applies writes/deletes (asserting the + /// ≤ 100 chunk), `/read` answers by object. + #[derive(Clone, Default)] + struct Fake { + tuples: Arc>>, + writes: Arc>>, + } + + fn json(value: &serde_json::Value) -> ([(&'static str, &'static str); 1], String) { + ([("content-type", "application/json")], value.to_string()) + } + + async fn write( + State(fake): State, + body: axum::body::Bytes, + ) -> ([(&'static str, &'static str); 1], String) { + let request: serde_json::Value = serde_json::from_slice(&body).expect("json"); + let mut tuples = fake.tuples.lock().expect("lock"); + if let Some(keys) = request["writes"]["tuple_keys"].as_array() { + assert!(keys.len() <= 100, "RFC 0047 §3.3: ≤ 100 tuples per Write"); + assert_eq!(request["writes"]["on_duplicate"], "ignore"); + fake.writes.lock().expect("lock").push(keys.len()); + for key in keys { + let key: TupleKey = serde_json::from_value(key.clone()).expect("tuple"); + if !tuples.contains(&key) { + tuples.push(key); + } + } + } + if let Some(keys) = request["deletes"]["tuple_keys"].as_array() { + assert!(keys.len() <= 100); + assert_eq!(request["deletes"]["on_missing"], "ignore"); + for key in keys { + let key: TupleKey = serde_json::from_value(key.clone()).expect("tuple"); + tuples.retain(|t| *t != key); + } + } + json(&json!({})) + } + + async fn read( + State(fake): State, + body: axum::body::Bytes, + ) -> ([(&'static str, &'static str); 1], String) { + let request: serde_json::Value = serde_json::from_slice(&body).expect("json"); + let object = request["tuple_key"]["object"].as_str().expect("object"); + let tuples = fake.tuples.lock().expect("lock"); + let matching: Vec = tuples + .iter() + .filter(|t| t.object == object) + .map(|t| json!({ "key": t })) + .collect(); + json(&json!({ "tuples": matching, "continuation_token": "" })) + } + + async fn serve(fake: Fake) -> String { + let app = Router::new() + .route("/stores/{store}/write", post(write)) + .route("/stores/{store}/read", post(read)) + .with_state(fake); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind"); + let url = format!("http://{}", listener.local_addr().expect("addr")); + tokio::spawn(async move { + axum::serve(listener, app).await.expect("serve"); + }); + url + } + + fn emitter(url: &str) -> Arc { + use ourios_core::auth::openfga::{VisibilityObjectSpec, VisibilitySpec}; + let config = build_openfga_config(&OpenFgaSpec { + api_url: Some(url.to_string()), + store_id: Some("s".to_string()), + request_timeout_secs: Some("2".to_string()), + visibility: VisibilitySpec { + objects: vec![VisibilityObjectSpec { + object_type: Some("conversation".to_string()), + column: Some("attr.gen_ai.conversation.id".to_string()), + }], + ..VisibilitySpec::default() + }, + ..OpenFgaSpec::default() + }) + .expect("config"); + Arc::new( + GraphEmitter::from_config(&config) + .expect("client") + .expect("bound"), + ) + } + + fn kv(key: &str, value: &str) -> KeyValue { + KeyValue { + key: key.to_string(), + value: Some(AnyValue { + value: Some(Value::StringValue(value.to_string())), + }), + ..Default::default() + } + } + + fn promoted() -> PromotedAttributes { + PromotedAttributes::new_typed( + [], + [ + PromotedKey::string("gen_ai.conversation.id".to_string()), + PromotedKey::string("user.hash".to_string()), + ], + ) + } + + /// One file per call, in the sealed hour partition, `rows` records with + /// `conversation`/`user` attributes. + fn write_rows(store: &Store, conversation: &str, user: &str, agent: Option<&str>, n: u64) { + let rows: Vec = (0..n) + .map(|i| { + let mut r = super::tests::rec("acme", 1, super::tests::TS0 + i * 1_000); + r.attributes = vec![ + kv("gen_ai.conversation.id", conversation), + kv("user.hash", user), + ]; + if let Some(agent) = agent { + r.attributes.push(kv("gen_ai.agent.id", agent)); + } + r + }) + .collect(); + let partition = PartitionKey::derive(&rows[0]).expect("derive"); + let mut w = Writer::open_in_with_promoted( + store, + partition, + ourios_parquet::DEFAULT_ZSTD_LEVEL, + promoted(), + ) + .expect("open writer"); + w.append_records(&rows).expect("append"); + w.close().expect("close"); + } + + fn live_rows(store: &Store, bucket: &Path) -> Vec { + let mut rows = Vec::new(); + for key in store.list_blocking(Some("data/")).expect("list") { + if !key.ends_with(".parquet") { + continue; + } + let bytes = store.get_blocking(&key).expect("get"); + let reader = Reader::open_bytes(bytes.into()).expect("open"); + rows.extend(reader.read_all().expect("read")); + } + let _ = bucket; + rows + } + + /// Scenario RFC0047.10 — the sweep feeds the graph: after a sweep the + /// `parent`, `participant`, `actor` (and binding, and tool) tuples exist + /// with tenant-prefixed ids; a second sweep writes nothing new (the + /// partition is consolidated, nothing is rewritten); every `Write` is + /// ≤ 100 tuples. See `docs/rfcs/0047-rebac-resolver-and-graph-visibility.md` §5. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn rfc0047_10_sweep_emits_tuples_idempotently() { + let fake = Fake::default(); + let url = serve(fake.clone()).await; + let bucket = tempfile::TempDir::new().expect("temp"); + let store = super::tests::store_at(bucket.path()); + // Two files → a sealed candidate; 130 distinct conversations so the + // tuple set spans more than one chunk. + write_rows(&store, "c-1", "alice", Some("bot"), 3); + for i in 0..130 { + write_rows(&store, &format!("c-{}", i + 10), "bob", None, 1); + } + let emitter = emitter(&url); + let (result, _, sink) = sweep_once( + store.clone(), + CompactionPolicy::default(), + promoted(), + Box::new(SharedAuditSink::new()), + Some(Arc::clone(&emitter)), + ) + .await; + let report = result.expect("sweep"); + assert_eq!(report.partitions_compacted, 1, "{report:?}"); + assert!(report.errors.is_empty(), "{:?}", report.errors); + let tuples = fake.tuples.lock().expect("lock").clone(); + let t = |u: &str, r: &str, o: &str| TupleKey::new(u, r, o); + for tuple in [ + t("tenant:acme", "parent", "conversation:acme/c-1"), + t("user:alice", "participant", "conversation:acme/c-1"), + t("user:alice", "scoped_reader", "tenant:acme"), + t("agent:bot", "actor", "conversation:acme/c-1"), + t("agent:bot", "scoped_reader", "tenant:acme"), + t("tenant:acme", "parent", "conversation:acme/c-42"), + t("user:bob", "participant", "conversation:acme/c-42"), + t("user:bob", "scoped_reader", "tenant:acme"), + t("tenant:acme", "parent", "tool:acme/query_logs"), + ] { + assert!(tuples.contains(&tuple), "missing {tuple:?}"); + } + assert_eq!( + report.graph_tuples_emitted, + tuples.len(), + "every tuple sent once" + ); + let writes = fake.writes.lock().expect("lock").clone(); + assert!( + writes.len() >= 2 && writes.iter().all(|n| *n <= 100), + "{writes:?}" + ); + + // Second sweep: nothing to consolidate, nothing rewritten, nothing sent. + let before = fake.writes.lock().expect("lock").len(); + let (result, _, _) = sweep_once( + store.clone(), + CompactionPolicy::default(), + promoted(), + sink, + Some(emitter), + ) + .await; + let report = result.expect("sweep"); + assert_eq!(report.partitions_compacted, 0); + assert_eq!(report.graph_tuples_emitted, 0); + assert_eq!( + fake.writes.lock().expect("lock").len(), + before, + "nothing new" + ); + assert_eq!(fake.tuples.lock().expect("lock").len(), tuples.len()); + } + + /// Scenario RFC0047.11 — erasure removes tuples after rows: a requested + /// erasure rewrites the tenant's partitions with the conversation's rows + /// dropped, then deletes its tuples, then writes the `conversation_erased` + /// audit event after every compaction event, then removes the marker; the + /// object is unlisted (no tuple on it remains) and other conversations' + /// tuples are untouched. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + #[allow(clippy::too_many_lines)] // one store, one graph: rows → tuples → audit → marker in sequence + async fn rfc0047_11_erasure_removes_tuples_after_rows() { + let fake = Fake::default(); + let url = serve(fake.clone()).await; + let bucket = tempfile::TempDir::new().expect("temp"); + let store = super::tests::store_at(bucket.path()); + write_rows(&store, "c-1", "alice", Some("bot"), 3); + write_rows(&store, "c-2", "bob", None, 2); + let emitter = emitter(&url); + let audit = SharedAuditSink::new(); + // Sweep 1: consolidate + feed the graph. + let (result, _, sink) = sweep_once( + store.clone(), + CompactionPolicy::default(), + promoted(), + Box::new(audit.clone()), + Some(Arc::clone(&emitter)), + ) + .await; + result.expect("sweep"); + assert!( + fake.tuples + .lock() + .expect("lock") + .iter() + .any(|t| t.object == "conversation:acme/c-1") + ); + let _ = audit.drain(); + + // Request the erasure of c-1; sweep 2 performs it. A repeated request + // is a no-op (create-if-absent) — it never resets a marker's phase. + request_erasure(&store, "acme", "c-1").expect("request"); + request_erasure(&store, "acme", "c-1").expect("repeat"); + assert_eq!(pending_erasures(&store).expect("pending").len(), 1); + store + .put_blocking( + &super::erasure_marker_key("acme", "c-9"), + b" { \"phase\" : \"tuples\" }\n".to_vec(), + ) + .expect("hand-written marker"); + request_erasure(&store, "acme", "c-9").expect("repeat"); + let phases: Vec<_> = pending_erasures(&store) + .expect("pending") + .into_iter() + .map(|r| (r.conversation_id, r.phase)) + .collect(); + assert!( + phases.contains(&("c-9".to_string(), ErasurePhase::Tuples)), + "{phases:?}" + ); + store + .delete_blocking(&super::erasure_marker_key("acme", "c-9")) + .expect("cleanup"); + let (result, _, _) = sweep_once( + store.clone(), + CompactionPolicy::default(), + promoted(), + sink, + Some(Arc::clone(&emitter)), + ) + .await; + let report = result.expect("sweep"); + assert!(report.errors.is_empty(), "{:?}", report.errors); + assert_eq!(report.erasures.len(), 1, "{report:?}"); + let outcome = &report.erasures[0]; + assert_eq!(outcome.request.conversation_id, "c-1"); + assert_eq!(outcome.rows_dropped, 3); + assert_eq!(outcome.partitions_rewritten, 1); + assert_eq!(outcome.phase, ErasurePhase::Tuples); + assert_eq!( + outcome.tuples_deleted, + Some(3), + "parent + participant + actor" + ); + assert!(outcome.finished); + + // Rows: only c-2's remain. + let rows = live_rows(&store, bucket.path()); + assert_eq!(rows.len(), 2, "c-1's three rows are gone"); + assert!(rows.iter().all(|r| { + r.attributes.iter().any(|kv| { + kv.key == "gen_ai.conversation.id" + && kv.value.as_ref().and_then(|v| v.value.as_ref()) + == Some(&Value::StringValue("c-2".to_string())) + }) + })); + // Tuples: no tuple on the object remains; c-2's untouched; the + // binding tuples (tenant-scoped, not object-scoped) stay. + let tuples = fake.tuples.lock().expect("lock").clone(); + assert!(!tuples.iter().any(|t| t.object == "conversation:acme/c-1")); + assert!(tuples.iter().any(|t| t.object == "conversation:acme/c-2")); + assert!(tuples.contains(&TupleKey::new("user:alice", "scoped_reader", "tenant:acme"))); + // Marker gone; nothing pending. + assert!(pending_erasures(&store).expect("pending").is_empty()); + // Audit order: the erasure event comes after every compaction + // event of the sweep (the rewrite), carrying the counts. + let events = audit.drain(); + let erased = events + .iter() + .position(|e| matches!(e.payload, AuditPayload::ConversationErased { .. })) + .expect("conversation_erased event"); + assert_eq!(erased, events.len() - 1, "last event of the sweep"); + match &events[erased].payload { + AuditPayload::ConversationErased { + conversation_id, + partitions_rewritten, + rows_dropped, + tuples_deleted, + } => { + assert_eq!(conversation_id, "c-1"); + assert_eq!(*partitions_rewritten, 1); + assert_eq!(*rows_dropped, 3); + assert_eq!(*tuples_deleted, 3); + } + other => panic!("{other:?}"), + } + assert_eq!(events[erased].tenant_id.as_str(), "acme"); + } + + /// RFC0047.11 (raw ids): a conversation whose id can never be a graph + /// object (here: whitespace) still has its rows erased — matched on the + /// stored value — with zero tuples to delete and an honest audit event. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn rfc0047_11_erasure_matches_raw_ids() { + let fake = Fake::default(); + let url = serve(fake.clone()).await; + let bucket = tempfile::TempDir::new().expect("temp"); + let store = super::tests::store_at(bucket.path()); + write_rows(&store, "odd id", "alice", None, 2); + write_rows(&store, "c-2", "bob", None, 1); + let emitter = emitter(&url); + let audit = SharedAuditSink::new(); + let (result, _, sink) = sweep_once( + store.clone(), + CompactionPolicy::default(), + promoted(), + Box::new(audit.clone()), + Some(Arc::clone(&emitter)), + ) + .await; + result.expect("sweep"); + assert!( + !fake + .tuples + .lock() + .expect("lock") + .iter() + .any(|t| t.object.contains("odd")), + "no tuple was ever minted for a non-object-id conversation" + ); + request_erasure(&store, "acme", "odd id").expect("request"); + let (result, _, _) = sweep_once( + store.clone(), + CompactionPolicy::default(), + promoted(), + sink, + Some(emitter), + ) + .await; + let report = result.expect("sweep"); + assert!(report.errors.is_empty(), "{:?}", report.errors); + let outcome = &report.erasures[0]; + assert_eq!(outcome.rows_dropped, 2, "rows matched on the raw value"); + assert_eq!(outcome.tuples_deleted, Some(0)); + assert!(outcome.finished); + assert_eq!(live_rows(&store, bucket.path()).len(), 1); + } +} diff --git a/crates/ourios-ingester/src/graph_emitter.rs b/crates/ourios-ingester/src/graph_emitter.rs new file mode 100644 index 00000000..34b54898 --- /dev/null +++ b/crates/ourios-ingester/src/graph_emitter.rs @@ -0,0 +1,412 @@ +//! RFC 0047 §3.3 / §3.6 — feeding the authorization graph from the data, +//! and taking a conversation back out of it. +//! +//! The emitter derives relationship tuples from stored rows — the promoted +//! `gen_ai.conversation.id`, `user.hash` / `enduser.pseudo.id` and +//! `gen_ai.agent.id` values — and writes them to `OpenFGA` in ≤ 100-tuple +//! idempotent batches. It is fed by the compaction sweep (every row it +//! rewrites) and by the receiver's flush cadence (every batch it publishes), +//! so a conversation is visible to fine-grained principals seconds after it +//! is stored. Erasure reads a conversation object's tuples and deletes them +//! — after the Parquet rewrite that dropped the rows, never before. +//! +//! Object naming is [`TenantObjects`] — the one place the rule lives — so +//! the emitter and the planner can never disagree. + +use std::collections::BTreeSet; +use std::sync::Arc; + +use opentelemetry::KeyValue; +use opentelemetry::global; +use opentelemetry::metrics::Counter; +use ourios_core::auth::openfga::{ + MCP_TOOL_NAMES, OpenFgaClient, OpenFgaConfig, OpenFgaError, PrincipalKind, TenantObjects, + TupleKey, is_object_id, +}; +use ourios_core::record::MinedRecord; +use ourios_parquet::promoted::{self, project_string_value}; +use ourios_semconv as semconv; + +/// `OpenFGA`'s cap on tuples per transactional `Write` (RFC 0047 §3.3). +pub const WRITE_CHUNK: usize = 100; +/// The user-identity attribute keys the emitter reads (RFC 0047 §3.3). +pub const USER_KEYS: [&str; 2] = ["user.hash", "enduser.pseudo.id"]; +/// The agent-identity attribute key the emitter reads (RFC 0047 §3.3). +pub const AGENT_KEY: &str = "gen_ai.agent.id"; + +const OPERATION_WRITE: &str = "write"; +const OPERATION_DELETE: &str = "delete"; +const ERROR_TYPE: &str = "error.type"; +const ERROR_TYPE_UPSTREAM_UNAVAILABLE: &str = "upstream_unavailable"; + +/// The graph emitter (RFC 0047 §3.3): derives tuples from rows and writes +/// them; erases a conversation's tuples (§3.6). +pub struct GraphEmitter { + client: OpenFgaClient, + /// Which attribute family and key carries the conversation id — the + /// same column the planner filters on. + conversation: ConversationKey, + tuples: Counter, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +enum ConversationKey { + Log(String), + Resource(String), +} + +impl std::fmt::Debug for GraphEmitter { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("GraphEmitter") + .field("conversation", &self.conversation) + .finish_non_exhaustive() + } +} + +/// What one emitter call did. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct Emitted { + /// Tuples sent (idempotent — an existing tuple is not an error). + pub tuples: usize, + /// `Write` calls issued (≤ 100 tuples each). + pub batches: usize, +} + +impl GraphEmitter { + /// An emitter for the configured graph, or `None` when no + /// `conversation` object type is bound (`auth.openfga.visibility.objects`) + /// — nothing to derive tuples for. + /// + /// # Errors + /// + /// When the HTTP client cannot be built (a startup error). + pub fn from_config(config: &OpenFgaConfig) -> Result, String> { + let Some(object) = + config.visibility().objects().iter().find(|object| { + object.object_type() == ourios_core::auth::openfga::CONVERSATION_TYPE + }) + else { + return Ok(None); + }; + let conversation = if let Some(key) = object.column().strip_prefix(promoted::ATTR_PREFIX) { + ConversationKey::Log(key.to_string()) + } else if let Some(key) = object.column().strip_prefix(promoted::RESOURCE_PREFIX) { + ConversationKey::Resource(key.to_string()) + } else { + return Err(format!( + "auth.openfga.visibility.objects: conversation column `{}` is not a promoted \ + column name", + object.column() + )); + }; + Ok(Some(Self { + client: OpenFgaClient::new(config)?, + conversation, + tuples: global::meter("ourios.graph") + .u64_counter(semconv::OURIOS_GRAPH_TUPLES) + .with_unit("{tuple}") + .build(), + })) + } + + /// Whether `record` belongs to conversation `id` — the erasure filter + /// (RFC 0047 §3.6), reading the same column the tuples were derived + /// from, on the **raw** value: a conversation whose id can never be a + /// graph object is still erasable from the rows. + #[must_use] + pub fn conversation_matches(&self, record: &MinedRecord, id: &str) -> bool { + self.raw_conversation_id(record) == Some(id) + } + + /// The conversation id of `record` as stored, if any. + fn raw_conversation_id<'a>(&self, record: &'a MinedRecord) -> Option<&'a str> { + match &self.conversation { + ConversationKey::Log(key) => project_string_value(&record.attributes, key), + ConversationKey::Resource(key) => { + project_string_value(&record.resource_attributes, key) + } + } + } + + /// The conversation id of `record`, when it carries one that can be a + /// graph object id. + fn conversation_id<'a>(&self, record: &'a MinedRecord) -> Option<&'a str> { + let id = self.raw_conversation_id(record)?; + is_object_id(id).then_some(id) + } + + /// The RFC 0047 §3.3 tuples of `records` in `tenant`: for every distinct + /// conversation `conversation:T/#parent@tenant:T`; per + /// (conversation, user) `#participant@user:` plus the + /// `tenant:T#scoped_reader@user:` binding tuple; per (conversation, + /// agent) `#actor@agent:` plus its binding tuple. Deduplicated; + /// values that cannot be object ids are skipped. Empty when the tenant + /// itself cannot be a graph object. + #[must_use] + pub fn derive(&self, tenant: &str, records: &[MinedRecord]) -> BTreeSet { + let mut tuples = BTreeSet::new(); + let Some(objects) = TenantObjects::new(tenant) else { + return tuples; + }; + for record in records { + let Some(id) = self.conversation_id(record) else { + continue; + }; + if !objects.conversation_fits(id) { + continue; + } + let conversation = objects.conversation(id); + tuples.insert(TupleKey::new(objects.tenant(), "parent", &conversation)); + for key in USER_KEYS { + if let Some(user) = project_string_value(&record.attributes, key) + && is_object_id(user) + { + let user = format!("{}:{user}", PrincipalKind::User.type_name()); + tuples.insert(TupleKey::new(&user, "participant", &conversation)); + tuples.insert(TupleKey::new(&user, "scoped_reader", objects.tenant())); + } + } + if let Some(agent) = project_string_value(&record.attributes, AGENT_KEY) + && is_object_id(agent) + { + let agent = format!("{}:{agent}", PrincipalKind::Agent.type_name()); + tuples.insert(TupleKey::new(&agent, "actor", &conversation)); + tuples.insert(TupleKey::new(&agent, "scoped_reader", objects.tenant())); + } + } + tuples + } + + /// The per-tenant tool objects (RFC 0047 §3.5): `tool:T/#parent@ + /// tenant:T` for every MCP tool, so operators grant `caller` only. + #[must_use] + pub fn tool_tuples(tenant: &str) -> BTreeSet { + let Some(objects) = TenantObjects::new(tenant) else { + return BTreeSet::new(); + }; + MCP_TOOL_NAMES + .iter() + .map(|tool| TupleKey::new(objects.tenant(), "parent", objects.tool(tool))) + .collect() + } + + /// Write `tuples` in ≤ 100-tuple idempotent batches. Stops at the first + /// failed batch (later batches are retried by the next sweep — every + /// write is idempotent). + /// + /// # Errors + /// + /// [`OpenFgaError`] from the failed batch. + pub async fn emit(&self, tuples: &BTreeSet) -> Result { + let all: Vec = tuples.iter().cloned().collect(); + let mut emitted = Emitted::default(); + for chunk in all.chunks(WRITE_CHUNK) { + match self.client.write(chunk, &[]).await { + Ok(()) => { + self.record(OPERATION_WRITE, chunk.len(), None); + emitted.tuples += chunk.len(); + emitted.batches += 1; + } + Err(e) => { + self.record( + OPERATION_WRITE, + chunk.len(), + Some(ERROR_TYPE_UPSTREAM_UNAVAILABLE), + ); + return Err(e); + } + } + } + Ok(emitted) + } + + /// Erase a conversation from the graph (RFC 0047 §3.6): read the + /// object's tuples, delete them in ≤ 100-tuple batches. Returns the + /// number deleted. Call **after** the Parquet rewrite that dropped the + /// rows — a dangling tuple is harmless, a dangling row is a leak. + /// + /// # Errors + /// + /// [`OpenFgaError`] from the read or a failed batch; the erasure is + /// retried by the next sweep (deletes are idempotent). + pub async fn erase_conversation(&self, tenant: &str, id: &str) -> Result { + let objects = TenantObjects::new(tenant).ok_or(OpenFgaError::InvalidTenant)?; + // A conversation the emitter could never have named has no tuples + // — nothing to read or delete; the rows were still dropped. + if !objects.conversation_fits(id) { + return Ok(0); + } + let tuples = self + .client + .read_by_object(&objects.conversation(id)) + .await?; + let mut deleted = 0; + for chunk in tuples.chunks(WRITE_CHUNK) { + match self.client.write(&[], chunk).await { + Ok(()) => { + self.record(OPERATION_DELETE, chunk.len(), None); + deleted += chunk.len(); + } + Err(e) => { + self.record( + OPERATION_DELETE, + chunk.len(), + Some(ERROR_TYPE_UPSTREAM_UNAVAILABLE), + ); + return Err(e); + } + } + } + Ok(deleted) + } + + fn record(&self, operation: &'static str, count: usize, error_type: Option<&'static str>) { + let count = u64::try_from(count).unwrap_or(u64::MAX); + match error_type { + None => self.tuples.add( + count, + &[KeyValue::new( + semconv::OURIOS_GRAPH_TUPLE_OPERATION, + operation, + )], + ), + Some(error_type) => self.tuples.add( + count, + &[ + KeyValue::new(semconv::OURIOS_GRAPH_TUPLE_OPERATION, operation), + KeyValue::new(ERROR_TYPE, error_type), + ], + ), + } + } +} + +/// A shared emitter handle for the roles that feed it. +pub type SharedGraphEmitter = Arc; + +#[cfg(test)] +mod tests { + use ourios_core::auth::openfga::{OpenFgaSpec, TupleKey, build_openfga_config}; + use ourios_core::otlp::any_value::Value; + use ourios_core::otlp::{AnyValue, KeyValue}; + use ourios_core::record::{BodyKind, MinedRecord}; + use ourios_core::tenant::TenantId; + + use super::GraphEmitter; + + fn kv(key: &str, value: &str) -> KeyValue { + KeyValue { + key: key.to_string(), + value: Some(AnyValue { + value: Some(Value::StringValue(value.to_string())), + }), + ..Default::default() + } + } + + fn record(attrs: Vec) -> MinedRecord { + MinedRecord { + tenant_id: TenantId::new("acme"), + template_id: 1, + template_version: 1, + severity_number: 9, + severity_text: None, + scope_name: None, + scope_version: None, + scope_attributes: Vec::new(), + resource_schema_url: None, + scope_schema_url: None, + time_unix_nano: 1, + observed_time_unix_nano: None, + attributes: attrs, + dropped_attributes_count: 0, + resource_attributes: Vec::new(), + trace_id: None, + span_id: None, + flags: 0, + event_name: None, + body_kind: BodyKind::Absent, + params: Vec::new(), + separators: Vec::new(), + body: None, + confidence: 1.0, + lossy_flag: false, + } + } + + fn emitter() -> GraphEmitter { + use ourios_core::auth::openfga::{VisibilityObjectSpec, VisibilitySpec}; + let config = build_openfga_config(&OpenFgaSpec { + api_url: Some("http://openfga.invalid:8080".to_string()), + store_id: Some("s".to_string()), + visibility: VisibilitySpec { + objects: vec![VisibilityObjectSpec { + object_type: Some("conversation".to_string()), + column: Some("attr.gen_ai.conversation.id".to_string()), + }], + ..VisibilitySpec::default() + }, + ..OpenFgaSpec::default() + }) + .expect("config"); + GraphEmitter::from_config(&config) + .expect("client") + .expect("conversation bound") + } + + fn t(user: &str, relation: &str, object: &str) -> TupleKey { + TupleKey::new(user, relation, object) + } + + /// RFC0047.10 (derivation): the §3.3 table — parent per conversation, + /// participant + binding per user, actor + binding per agent — + /// deduplicated across rows, tenant-prefixed, with values that cannot + /// be object ids skipped and rows without a conversation ignored. + #[test] + fn derives_the_section_3_3_tuples() { + let emitter = emitter(); + let rows = vec![ + record(vec![ + kv("gen_ai.conversation.id", "c-1"), + kv("user.hash", "alice"), + kv("gen_ai.agent.id", "bot"), + ]), + record(vec![ + kv("gen_ai.conversation.id", "c-1"), + kv("user.hash", "alice"), + ]), + record(vec![ + kv("gen_ai.conversation.id", "c-2"), + kv("enduser.pseudo.id", "bob"), + ]), + // no conversation → nothing + record(vec![kv("user.hash", "carol")]), + // a user value that cannot be an object id → skipped, the + // conversation itself still gets its parent tuple + record(vec![ + kv("gen_ai.conversation.id", "c-3"), + kv("user.hash", "has space"), + ]), + ]; + let tuples = emitter.derive("acme", &rows); + let expected = [ + t("tenant:acme", "parent", "conversation:acme/c-1"), + t("user:alice", "participant", "conversation:acme/c-1"), + t("user:alice", "scoped_reader", "tenant:acme"), + t("agent:bot", "actor", "conversation:acme/c-1"), + t("agent:bot", "scoped_reader", "tenant:acme"), + t("tenant:acme", "parent", "conversation:acme/c-2"), + t("user:bob", "participant", "conversation:acme/c-2"), + t("user:bob", "scoped_reader", "tenant:acme"), + t("tenant:acme", "parent", "conversation:acme/c-3"), + ]; + assert_eq!(tuples, expected.into_iter().collect()); + assert!(emitter.derive("bad tenant", &rows).is_empty()); + assert_eq!(GraphEmitter::tool_tuples("acme").len(), 3); + assert!(GraphEmitter::tool_tuples("acme").contains(&t( + "tenant:acme", + "parent", + "tool:acme/query_logs" + ))); + } +} diff --git a/crates/ourios-ingester/src/lib.rs b/crates/ourios-ingester/src/lib.rs index 3eea7fe0..d0fdf948 100644 --- a/crates/ourios-ingester/src/lib.rs +++ b/crates/ourios-ingester/src/lib.rs @@ -35,6 +35,8 @@ pub mod audit_sink; pub mod compactor; pub mod encode_pool; +#[cfg(feature = "openfga")] +pub mod graph_emitter; pub mod metrics; pub mod publish; pub mod receiver; diff --git a/crates/ourios-ingester/src/metrics.rs b/crates/ourios-ingester/src/metrics.rs index 68de6ce5..32f46139 100644 --- a/crates/ourios-ingester/src/metrics.rs +++ b/crates/ourios-ingester/src/metrics.rs @@ -808,6 +808,8 @@ mod tests { let (guard, exporter) = ourios_telemetry::init_in_memory("ourios-test"); let metrics = CompactionMetrics::new(); let report = SweepReport { + erasures: Vec::new(), + graph_tuples_emitted: 0, tenants_scanned: 2, partitions_compacted: 2, files_compacted: 7, diff --git a/crates/ourios-ingester/src/publish.rs b/crates/ourios-ingester/src/publish.rs index cfa6182b..553f5cac 100644 --- a/crates/ourios-ingester/src/publish.rs +++ b/crates/ourios-ingester/src/publish.rs @@ -71,13 +71,37 @@ impl Drained { pub struct PublishCoordinator { record: SharedParquetSink, audit: SharedParquetAuditSink, + /// The RFC 0047 §3.3 graph emitter — fed with every published batch + /// (the flush-cadence bridge), when the graph is configured. + #[cfg(feature = "openfga")] + graph: Option>, } impl PublishCoordinator { /// Build a coordinator over the two shared sinks. #[must_use] pub fn new(record: SharedParquetSink, audit: SharedParquetAuditSink) -> Self { - Self { record, audit } + Self { + record, + audit, + #[cfg(feature = "openfga")] + graph: None, + } + } + + /// Feed the RFC 0047 §3.3 graph from every batch this coordinator + /// publishes: tuples are derived from the records about to be written + /// and sent — asynchronously, best-effort, idempotent — once the batch + /// is durable. The compaction sweep re-derives the same tuples later, + /// so a failed send here only delays visibility. + #[cfg(feature = "openfga")] + #[must_use] + pub fn with_graph_emitter( + mut self, + emitter: std::sync::Arc, + ) -> Self { + self.graph = Some(emitter); + self } /// Atomically take the audit buffer + the aged record partitions (the @@ -144,7 +168,43 @@ impl PublishCoordinator { self.record.requeue(drained.records); return false; } - self.record.publish_owned(drained.records, trigger) + #[cfg(feature = "openfga")] + let tuples = self.graph.as_ref().map(|emitter| { + let mut tuples = std::collections::BTreeSet::new(); + for (partition, records) in &drained.records { + tuples.extend(emitter.derive(&partition.tenant_id, records)); + tuples.extend(crate::graph_emitter::GraphEmitter::tool_tuples( + &partition.tenant_id, + )); + } + tuples + }); + let published = self.record.publish_owned(drained.records, trigger); + #[cfg(feature = "openfga")] + if published + && let (Some(emitter), Some(tuples)) = (self.graph.clone(), tuples) + && !tuples.is_empty() + { + // Off the publish path: the graph is fed after the batch is + // durable, and never delays the next flush. + if let Ok(handle) = tokio::runtime::Handle::try_current() { + handle.spawn(async move { + if let Err(e) = emitter.emit(&tuples).await { + tracing::warn!( + error = %e, + "graph emit after flush failed; the sweep re-derives these tuples \ + (RFC 0047 §3.3)" + ); + } + }); + } else { + tracing::warn!( + "graph emit after flush skipped: no runtime handle; the sweep re-derives \ + these tuples (RFC 0047 §3.3)" + ); + } + } + published } /// The record sink handle (for the receiver's existing flush/snapshot paths). diff --git a/crates/ourios-parquet/src/audit_reader.rs b/crates/ourios-parquet/src/audit_reader.rs index 87c5e090..67747178 100644 --- a/crates/ourios-parquet/src/audit_reader.rs +++ b/crates/ourios-parquet/src/audit_reader.rs @@ -43,8 +43,8 @@ use parquet::errors::ParquetError; use crate::audit_columns; use crate::audit_record_batch::{ EVENT_KIND_ALIAS_ASSERTED, EVENT_KIND_ALIAS_RETRACTED, EVENT_KIND_COMPACTION, - EVENT_KIND_INGEST_DENIED, EVENT_KIND_RECORD_QUARANTINED, EVENT_KIND_TEMPLATE_CREATED, - EVENT_KIND_TEMPLATE_TYPE_EXPANDED, EVENT_KIND_TEMPLATE_WIDENED, + EVENT_KIND_CONVERSATION_ERASED, EVENT_KIND_INGEST_DENIED, EVENT_KIND_RECORD_QUARANTINED, + EVENT_KIND_TEMPLATE_CREATED, 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}; @@ -349,6 +349,7 @@ fn batch_to_audit_events( // Quarantine-group columns (RFC 0025 §3.3) — absent in // pre-amendment files (§3.7 absent-column tolerance). let (quarantine_partition, quarantine_error, denied_token_name) = rejection_columns(batch)?; + let erasure = erasure_columns(batch)?; for i in 0..n { let file_row = row_offset + i; @@ -398,6 +399,7 @@ fn batch_to_audit_events( decode_quarantine_payload(&cols, i, file_row)? } EVENT_KIND_INGEST_DENIED => decode_denied_payload(&denied_token_name, i, file_row)?, + EVENT_KIND_CONVERSATION_ERASED => decode_erasure_payload(&erasure, i, file_row)?, kind @ (EVENT_KIND_ALIAS_ASSERTED | EVENT_KIND_ALIAS_RETRACTED) => { let cols = AliasColumns { representative_id: &alias_representative_id, @@ -491,6 +493,50 @@ fn rejection_columns(batch: &RecordBatch) -> Result>, + partitions: Vec>, + rows: Vec>, + tuples: Vec>, +} + +/// The erasure-group columns — absent in earlier files (§3.7 +/// absent-column tolerance). +fn erasure_columns(batch: &RecordBatch) -> Result { + Ok(ErasureColumns { + conversation_id: optional_string(batch, audit_columns::ERASURE_CONVERSATION_ID)? + .unwrap_or_default(), + partitions: optional_u64(batch, audit_columns::ERASURE_PARTITIONS)?, + rows: optional_u64(batch, audit_columns::ERASURE_ROWS)?, + tuples: optional_u64(batch, audit_columns::ERASURE_TUPLES)?, + }) +} + +/// Rebuild the `conversation_erased` payload for row `i` (RFC 0047 §3.6). +fn decode_erasure_payload( + cols: &ErasureColumns, + i: usize, + file_row: usize, +) -> Result { + Ok(AuditPayload::ConversationErased { + conversation_id: require_at( + &cols.conversation_id, + i, + audit_columns::ERASURE_CONVERSATION_ID, + file_row, + )?, + partitions_rewritten: require_at( + &cols.partitions, + i, + audit_columns::ERASURE_PARTITIONS, + file_row, + )?, + rows_dropped: require_at(&cols.rows, i, audit_columns::ERASURE_ROWS, file_row)?, + tuples_deleted: require_at(&cols.tuples, i, audit_columns::ERASURE_TUPLES, file_row)?, + }) +} + /// Rebuild the `ingest_denied` payload for row `i` from its column /// (RFC 0026 §3.4). fn decode_denied_payload( diff --git a/crates/ourios-parquet/src/audit_record_batch.rs b/crates/ourios-parquet/src/audit_record_batch.rs index ce5af582..5ebd745f 100644 --- a/crates/ourios-parquet/src/audit_record_batch.rs +++ b/crates/ourios-parquet/src/audit_record_batch.rs @@ -69,8 +69,8 @@ use crate::audit_schema; /// existing call sites resolve them at their established path. pub use ourios_core::audit::{ EVENT_KIND_ALIAS_ASSERTED, EVENT_KIND_ALIAS_RETRACTED, EVENT_KIND_COMPACTION, - EVENT_KIND_INGEST_DENIED, EVENT_KIND_RECORD_QUARANTINED, EVENT_KIND_TEMPLATE_CREATED, - EVENT_KIND_TEMPLATE_TYPE_EXPANDED, EVENT_KIND_TEMPLATE_WIDENED, + EVENT_KIND_CONVERSATION_ERASED, EVENT_KIND_INGEST_DENIED, EVENT_KIND_RECORD_QUARANTINED, + EVENT_KIND_TEMPLATE_CREATED, 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_CREATED, EVENT_TYPE_TEMPLATE_TYPE_EXPANDED, EVENT_TYPE_TEMPLATE_WIDENED, @@ -214,6 +214,10 @@ struct Builders { quarantine_partition: StringBuilder, quarantine_error: StringBuilder, denied_token_name: StringBuilder, + erasure_conversation_id: StringBuilder, + erasure_partitions: UInt64Builder, + erasure_rows: UInt64Builder, + erasure_tuples: UInt64Builder, compaction_input_files: GenericListBuilder, compaction_output_file: StringBuilder, compaction_generation: UInt64Builder, @@ -272,6 +276,10 @@ impl Builders { quarantine_partition: StringBuilder::with_capacity(cap, 0), quarantine_error: StringBuilder::with_capacity(cap, 0), denied_token_name: StringBuilder::with_capacity(cap, 0), + erasure_conversation_id: StringBuilder::with_capacity(cap, 0), + erasure_partitions: UInt64Builder::with_capacity(cap), + erasure_rows: UInt64Builder::with_capacity(cap), + erasure_tuples: UInt64Builder::with_capacity(cap), compaction_input_files: GenericListBuilder::new(StringBuilder::new()) .with_field(Field::new("element", DataType::Utf8, false)), compaction_output_file: StringBuilder::with_capacity(cap, 0), @@ -308,7 +316,7 @@ impl Builders { // 2026-06-12). self.append_compaction_nulls(); self.append_quarantine_nulls(); - self.append_denied_nulls(); + self.append_trailing_nulls(); self.append_alias_nulls(); self.template_id.append_value(*template_id); self.triggering_line_hash @@ -342,7 +350,7 @@ impl Builders { self.append_template_nulls(); self.append_compaction_nulls(); self.append_quarantine_nulls(); - self.append_denied_nulls(); + self.append_trailing_nulls(); if reason.is_empty() { self.reason.append_null(); } else { @@ -365,7 +373,7 @@ impl Builders { self.append_template_nulls(); self.append_compaction_nulls(); self.append_quarantine_nulls(); - self.append_denied_nulls(); + self.append_trailing_nulls(); self.append_alias_nulls(); self.reason.append_null(); } @@ -375,45 +383,30 @@ impl Builders { output_file, generation, rows, - } => { - // Compaction events leave every template-specific - // 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); - self.compaction_generation.append_value(*generation); - self.compaction_rows.append_value(*rows); - self.append_quarantine_nulls(); - self.append_denied_nulls(); - } + } => self.append_compaction(partition, input_files, output_file, *generation, *rows), AuditPayload::RecordQuarantined { partition, error } => { // Quarantine events (RFC 0025 §3.3) populate only the // envelope and the `quarantine_*` columns. self.append_template_nulls(); self.append_compaction_nulls(); self.append_alias_nulls(); - self.append_denied_nulls(); + self.append_trailing_nulls(); self.reason.append_null(); self.quarantine_partition.append_value(partition); self.quarantine_error.append_value(error); } - AuditPayload::IngestDenied { token_name } => { - // Denial events (RFC 0026 §3.4) populate only the - // envelope (whose `tenant_id` is the offending tenant) - // and the token's audit label. - self.append_template_nulls(); - self.append_compaction_nulls(); - self.append_alias_nulls(); - self.append_quarantine_nulls(); - self.reason.append_null(); - self.denied_token_name.append_value(token_name); - } + AuditPayload::IngestDenied { token_name } => self.append_denied(token_name), + AuditPayload::ConversationErased { + conversation_id, + partitions_rewritten, + rows_dropped, + tuples_deleted, + } => self.append_erasure( + conversation_id, + *partitions_rewritten, + *rows_dropped, + *tuples_deleted, + ), } Ok(()) @@ -549,6 +542,78 @@ impl Builders { self.denied_token_name.append_null(); } + /// Compaction events leave every template-specific 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. + fn append_compaction( + &mut self, + partition: &str, + input_files: &[String], + output_file: &str, + generation: u64, + rows: u64, + ) { + 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); + self.compaction_generation.append_value(generation); + self.compaction_rows.append_value(rows); + self.append_quarantine_nulls(); + self.append_trailing_nulls(); + } + + /// Denial events (RFC 0026 §3.4) populate only the envelope (whose + /// `tenant_id` is the offending tenant) and the token's audit label. + fn append_denied(&mut self, token_name: &str) { + self.append_template_nulls(); + self.append_compaction_nulls(); + self.append_alias_nulls(); + self.append_quarantine_nulls(); + self.reason.append_null(); + self.denied_token_name.append_value(token_name); + self.append_erasure_nulls(); + } + + /// Erasure events (RFC 0047 §3.6) populate only the envelope and the + /// `erasure_*` columns. + fn append_erasure( + &mut self, + conversation_id: &str, + partitions_rewritten: u64, + rows_dropped: u64, + tuples_deleted: u64, + ) { + self.append_template_nulls(); + self.append_compaction_nulls(); + self.append_alias_nulls(); + self.append_quarantine_nulls(); + self.reason.append_null(); + self.append_denied_nulls(); + self.erasure_conversation_id.append_value(conversation_id); + self.erasure_partitions.append_value(partitions_rewritten); + self.erasure_rows.append_value(rows_dropped); + self.erasure_tuples.append_value(tuples_deleted); + } + + /// NULL the trailing rejection + erasure columns (every kind that is + /// neither `ingest_denied` nor `conversation_erased`). + fn append_trailing_nulls(&mut self) { + self.append_denied_nulls(); + self.append_erasure_nulls(); + } + + /// NULL the `conversation_erased`-only columns (every other kind). + fn append_erasure_nulls(&mut self) { + self.erasure_conversation_id.append_null(); + self.erasure_partitions.append_null(); + self.erasure_rows.append_null(); + self.erasure_tuples.append_null(); + } + fn finish(mut self) -> Vec { vec![ Arc::new(self.tenant_id.finish()), @@ -576,6 +641,10 @@ impl Builders { Arc::new(self.quarantine_partition.finish()), Arc::new(self.quarantine_error.finish()), Arc::new(self.denied_token_name.finish()), + Arc::new(self.erasure_conversation_id.finish()), + Arc::new(self.erasure_partitions.finish()), + Arc::new(self.erasure_rows.finish()), + Arc::new(self.erasure_tuples.finish()), ] } } diff --git a/crates/ourios-parquet/src/compaction.rs b/crates/ourios-parquet/src/compaction.rs index ee17a13e..ae806a31 100644 --- a/crates/ourios-parquet/src/compaction.rs +++ b/crates/ourios-parquet/src/compaction.rs @@ -56,9 +56,11 @@ const HOUR_NANOS: u64 = 3_600_000_000_000; pub struct CompactionOutcome { /// Number of live files before compaction. pub files_before: usize, - /// Rows in the consolidated file (equal to the total input rows). - /// `0` on a no-op. + /// Rows in the consolidated file (the total input rows minus any an + /// RFC 0047 §3.6 erasure dropped). `0` on a no-op. pub rows: u64, + /// Rows an erasure filter removed (RFC 0047 §3.6); `0` without one. + pub rows_dropped: u64, /// The commit, or `None` when compaction was a no-op (fewer than /// two live files — nothing to consolidate — or a lost CAS race that /// left the work for a later sweep). @@ -209,6 +211,45 @@ pub fn compact_partition_with_promoted( ) } +/// Row-level hooks a caller threads through the rewrite (RFC 0047 §3.3 / +/// §3.6): `observe` sees every input row once, as decoded (the graph +/// emitter's feed); `drop` removes rows from the consolidated output (a +/// conversation-scoped erasure) — with a `drop` filter the partition is +/// rewritten even when it holds a single file, so the erasure lands. +#[derive(Default)] +pub struct RowHooks<'a> { + /// Called once per input file with its decoded rows, before any drop. + pub observe: Option<&'a mut RowObserver<'a>>, + /// Rows for which this returns `true` are not written back. + pub drop: Option<&'a RowFilter<'a>>, +} + +/// A [`RowHooks::observe`] callback. +pub type RowObserver<'a> = dyn FnMut(&[MinedRecord]) + 'a; +/// A [`RowHooks::drop`] predicate. +pub type RowFilter<'a> = dyn Fn(&MinedRecord) -> bool + 'a; + +/// [`compact_partition_with_promoted`] with [`RowHooks`]. +/// +/// # Errors +/// +/// See [`compact_partition`]. +pub fn compact_partition_hooked( + store: &Store, + partition: &PartitionKey, + promoted: &PromotedAttributes, + hooks: &mut RowHooks<'_>, +) -> Result { + compact_sorted_hooked( + store, + partition, + promoted, + ClusterKeys::for_promoted(promoted), + SortTuning::default(), + hooks, + ) +} + /// Like [`compact_partition`] but rotating compacted row groups at an /// explicit `flush_bytes` threshold instead of the RFC 0036 §3.3 /// **adaptive** default (`OURIOS_COMPACTED_RG_BYTES` env, else the value @@ -292,6 +333,24 @@ fn compact_sorted( promoted: &PromotedAttributes, keys: ClusterKeys, tuning: SortTuning, +) -> Result { + compact_sorted_hooked( + store, + partition, + promoted, + keys, + tuning, + &mut RowHooks::default(), + ) +} + +fn compact_sorted_hooked( + store: &Store, + partition: &PartitionKey, + promoted: &PromotedAttributes, + keys: ClusterKeys, + tuning: SortTuning, + hooks: &mut RowHooks<'_>, ) -> Result { let key = manifest_key(partition); let (existing, etag) = @@ -300,7 +359,10 @@ fn compact_sorted( None => (None, None), }; let mut inputs = live_file_keys(store, partition, existing.as_ref())?; - if inputs.len() < 2 { + // Consolidation needs two files; an erasure (`drop`) rewrites any + // non-empty partition. + let minimum_inputs = if hooks.drop.is_some() { 1 } else { 2 }; + if inputs.len() < minimum_inputs { return Ok(no_op_outcome(inputs.len())); } // §3.1 tie-break: the input-file ordinal is sorted-basename order. @@ -364,7 +426,7 @@ fn compact_sorted( estimated_output_bytes, ) .map_err(CompactionError::Write)?; - let (row_count, bytes_read) = sort_inputs_into( + let (row_count, bytes_read, rows_dropped) = sort_inputs_into( &mut writer, store, partition, @@ -372,6 +434,7 @@ fn compact_sorted( keys, tuning, &inputs, + hooks, )?; let written = writer.close().map_err(CompactionError::Write)?; let bytes_written = written.bytes_written; @@ -413,6 +476,7 @@ fn compact_sorted( Ok(CompactionOutcome { files_before: inputs.len(), rows: row_count, + rows_dropped, committed: Some(Committed { file: consolidated, generation, @@ -518,6 +582,7 @@ enum SortState { /// rows are held at once to sort in place and skip spilling — bounded /// by one seal-target's worth of input, so no larger than decoding a /// single worst-case input file (the [`SortTuning`] tradeoff). +#[allow(clippy::too_many_arguments)] // one call site; the tuple of sort inputs is the seam fn sort_inputs_into( writer: &mut Writer, store: &Store, @@ -526,9 +591,11 @@ fn sort_inputs_into( keys: ClusterKeys, tuning: SortTuning, inputs: &[String], -) -> Result<(u64, u64), CompactionError> { + hooks: &mut RowHooks<'_>, +) -> Result<(u64, u64, u64), CompactionError> { let mut row_count: u64 = 0; let mut bytes_read: u64 = 0; + let mut rows_dropped: u64 = 0; let mut state = SortState::Buffering(Vec::new()); for input in inputs { let bytes = store @@ -538,6 +605,17 @@ fn sort_inputs_into( let reader = Reader::open_partition_bytes(Bytes::from(bytes), partition.clone(), input) .map_err(CompactionError::Read)?; let mut records = reader.read_all().map_err(CompactionError::Read)?; + // RFC 0047 §3.3/§3.6: the graph feed sees every row once (before + // any drop); an erasure removes its rows before the sort. + if let Some(observe) = hooks.observe.as_deref_mut() { + observe(&records); + } + if let Some(drop) = hooks.drop { + let before = records.len(); + records.retain(|record| !drop(record)); + rows_dropped = rows_dropped + .saturating_add(u64::try_from(before - records.len()).unwrap_or(u64::MAX)); + } #[cfg(test)] residency::add(records.len()); // `usize <= u64` on every supported target; saturate rather than panic @@ -602,7 +680,7 @@ fn sort_inputs_into( drop(scratch); } } - Ok((row_count, bytes_read)) + Ok((row_count, bytes_read, rows_dropped)) } /// Stable §3.1 sort of one input's decoded rows: promoted @@ -1083,10 +1161,14 @@ fn is_candidate( /// the immediate common-prefixes (cheap), never the full object set. This is the /// object-store equivalent of the pre-RFC-0019 level-by-level `read_dir` walk, /// not a recursive `O(N_objects)` scan. Each level's segment is parsed in the -/// canonical zero-padded form ([`parse_partition_segment`]); a non-canonical +/// canonical zero-padded form (`parse_partition_segment`); a non-canonical /// child prefix is dropped exactly as the old walk dropped non-canonical dirs. /// Returned sorted chronologically (oldest first) and deduplicated. -fn hour_partitions(store: &Store, tenant: &str) -> Result, CompactionError> { +/// +/// # Errors +/// +/// [`CompactionError`] when the store's prefixes cannot be listed. +pub fn hour_partitions(store: &Store, tenant: &str) -> Result, CompactionError> { let root = format!("data/tenant_id={}", percent_encode_tenant(tenant)); let mut partitions = Vec::new(); for (year_prefix, year) in numbered_child_prefixes(store, &root, "year", 4)? { @@ -1236,6 +1318,7 @@ fn no_op_outcome(files_before: usize) -> CompactionOutcome { CompactionOutcome { files_before, rows: 0, + rows_dropped: 0, committed: None, gc_failures: 0, bytes_read: 0, diff --git a/crates/ourios-parquet/src/lib.rs b/crates/ourios-parquet/src/lib.rs index 0a5e76a4..193fcc9c 100644 --- a/crates/ourios-parquet/src/lib.rs +++ b/crates/ourios-parquet/src/lib.rs @@ -43,9 +43,9 @@ pub use audit_record_batch::{AuditBatchError, audit_events_to_batch}; pub use audit_sink::ParquetAuditSink; pub use audit_writer::{AuditWriter, AuditWriterError, AuditWrittenFile, derive_audit_partition}; pub use compaction::{ - Committed, CompactionError, CompactionOutcome, CompactionPolicy, OrphanGc, compact_partition, - compact_partition_with_flush_threshold, compact_partition_with_promoted, gc_orphans, - plan_candidates, + Committed, CompactionError, CompactionOutcome, CompactionPolicy, OrphanGc, RowHooks, + compact_partition, compact_partition_hooked, compact_partition_with_flush_threshold, + compact_partition_with_promoted, gc_orphans, hour_partitions, plan_candidates, }; pub use manifest::{MANIFEST_FILENAME, Manifest, ManifestError, Published}; pub use partition::{ @@ -139,6 +139,13 @@ pub mod audit_columns { /// The rejecting token's audit label on an `ingest_denied` event /// (RFC 0026 §3.4) — never the token value. pub const DENIED_TOKEN_NAME: &str = "denied_token_name"; + /// `conversation_erased` (RFC 0047 §3.6, kind 9): the erased + /// conversation id and the pass's counts. OPTIONAL, NULL for every + /// other kind; appended after the denial column (§3.7 additive). + pub const ERASURE_CONVERSATION_ID: &str = "erasure_conversation_id"; + pub const ERASURE_PARTITIONS: &str = "erasure_partitions"; + pub const ERASURE_ROWS: &str = "erasure_rows"; + pub const ERASURE_TUPLES: &str = "erasure_tuples"; } /// Build the data-file Arrow schema per RFC 0005 §3.2. @@ -331,5 +338,11 @@ pub fn audit_schema() -> SchemaRef { Field::new(audit_columns::QUARANTINE_PARTITION, DataType::Utf8, true), Field::new(audit_columns::QUARANTINE_ERROR, DataType::Utf8, true), Field::new(audit_columns::DENIED_TOKEN_NAME, DataType::Utf8, true), + // Erasure columns (RFC 0047 §3.6, kind 9) — OPTIONAL, NULL for + // every other kind (§3.7 additive). + Field::new(audit_columns::ERASURE_CONVERSATION_ID, DataType::Utf8, true), + Field::new(audit_columns::ERASURE_PARTITIONS, DataType::UInt64, true), + Field::new(audit_columns::ERASURE_ROWS, DataType::UInt64, true), + Field::new(audit_columns::ERASURE_TUPLES, DataType::UInt64, true), ])) } diff --git a/crates/ourios-parquet/tests/it/audit_round_trip.rs b/crates/ourios-parquet/tests/it/audit_round_trip.rs index 1b69de52..ae1d16c1 100644 --- a/crates/ourios-parquet/tests/it/audit_round_trip.rs +++ b/crates/ourios-parquet/tests/it/audit_round_trip.rs @@ -150,6 +150,22 @@ fn denied_event(tenant: &str) -> AuditEvent { } } +/// A `conversation_erased` event (RFC 0047 §3.6): the tenant on the +/// envelope, the erased conversation and the pass's counts in the +/// `erasure_*` columns. +fn erased_event(tenant: &str) -> AuditEvent { + AuditEvent { + tenant_id: TenantId::new(tenant), + timestamp: ts(1_775_127_540), + payload: AuditPayload::ConversationErased { + conversation_id: "c-7".to_string(), + partitions_rewritten: 3, + rows_dropped: 12, + tuples_deleted: 4, + }, + } +} + fn audit_partition_for(event: &AuditEvent) -> PartitionKey { // Audit partitioning shares the data-side `PartitionKey` // shape; the writer/reader compare only tenant + year/month/ @@ -242,6 +258,31 @@ fn rfc0026_ingest_denied_audit_event_round_trips() { ); } +/// RFC 0047 §3.6 — the `conversation_erased` audit event round-trips: kind +/// 9, the tenant on the envelope, the id and counts in the `erasure_*` +/// columns, every other payload column NULL — next to a denial and a +/// compaction row so the null discipline holds across kinds. +#[test] +fn rfc0047_conversation_erased_audit_event_round_trips() { + let bucket = TempDir::new().unwrap(); + let events = vec![ + erased_event("acme"), + denied_event("acme"), + compaction_event("acme"), + ]; + 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"); + + let reader = AuditReader::open_partition(&written.path, partition).expect("open_partition"); + let round_tripped = reader.read_all().expect("read_all"); + assert_eq!(round_tripped, events, "erasure rows round-trip exactly"); + assert_eq!(round_tripped[0].payload.event_kind(), 9); + assert_eq!(round_tripped[0].payload.event_type(), "conversation_erased"); +} + /// `AuditReader::open_bytes` reads an audit file from in-memory bytes — the /// RFC 0019 `Store` read path the querier's audit scan moves onto (so the scan /// reads through the object-storage seam, local or S3, rather than `std::fs`). diff --git a/crates/ourios-parquet/tests/it/schema_pin.rs b/crates/ourios-parquet/tests/it/schema_pin.rs index 5f0da6e3..00148f69 100644 --- a/crates/ourios-parquet/tests/it/schema_pin.rs +++ b/crates/ourios-parquet/tests/it/schema_pin.rs @@ -197,6 +197,13 @@ fn rfc0005_10_audit_schema_matches_pinned_field_list() { // The ingest_denied column (RFC 0026 §3.4 amendment, // 2026-07-06): OPTIONAL, appended after the quarantine group. Field::new("denied_token_name", DataType::Utf8, true), + // The conversation_erased columns (RFC 0047 §3.6 amendment, + // 2026-08-18, RFC 0005 §3.7 kind 9): OPTIONAL, appended after + // the denial column. + Field::new("erasure_conversation_id", DataType::Utf8, true), + Field::new("erasure_partitions", DataType::UInt64, true), + Field::new("erasure_rows", DataType::UInt64, true), + Field::new("erasure_tuples", DataType::UInt64, true), ]; check_schema_against(&expected, &audit_schema()); } diff --git a/crates/ourios-semconv/src/lib.rs b/crates/ourios-semconv/src/lib.rs index 9c4acf3e..b410ed5b 100644 --- a/crates/ourios-semconv/src/lib.rs +++ b/crates/ourios-semconv/src/lib.rs @@ -69,6 +69,9 @@ pub const OURIOS_COMPACTION_ROWS: &str = "ourios.compaction.rows"; /// `ourios.compaction.sweeps` (counter, unit `{sweep}`). pub const OURIOS_COMPACTION_SWEEPS: &str = "ourios.compaction.sweeps"; +/// `ourios.graph.tuples` (counter, unit `{tuple}`). +pub const OURIOS_GRAPH_TUPLES: &str = "ourios.graph.tuples"; + /// `ourios.ingest.batches` (counter, unit `{batch}`). pub const OURIOS_INGEST_BATCHES: &str = "ourios.ingest.batches"; @@ -192,6 +195,9 @@ pub const OURIOS_AUDIT_SINK_FLUSH_OUTCOME: &str = "ourios.audit_sink.flush.outco /// `ourios.compaction.result` attribute key. pub const OURIOS_COMPACTION_RESULT: &str = "ourios.compaction.result"; +/// `ourios.graph.tuple.operation` attribute key. +pub const OURIOS_GRAPH_TUPLE_OPERATION: &str = "ourios.graph.tuple.operation"; + /// `ourios.ingest.json.lenient` attribute key. pub const OURIOS_INGEST_JSON_LENIENT: &str = "ourios.ingest.json.lenient"; diff --git a/crates/ourios-server/src/main.rs b/crates/ourios-server/src/main.rs index 1799c7fa..13bd8659 100644 --- a/crates/ourios-server/src/main.rs +++ b/crates/ourios-server/src/main.rs @@ -867,6 +867,14 @@ async fn main() -> Result<(), Box> { // learns the actual ports. // One resolver, built once, shared by every enabled network role. let resolver = auth_resolver(&config).await?; + // RFC 0047 §3.3: the graph emitter — built for every role that stores or + // rewrites rows (receiver flush cadence, compaction sweep) when the graph + // binds a conversation object; no startup round-trip. + let graph_emitter = match config.auth.as_ref().and_then(|auth| auth.openfga.as_ref()) { + Some(openfga) => ourios_ingester::graph_emitter::GraphEmitter::from_config(openfga)? + .map(std::sync::Arc::new), + None => None, + }; let receiver = match &config.receiver { // The receiver's RFC 0014 data write path runs on the resolved store @@ -886,6 +894,7 @@ async fn main() -> Result<(), Box> { promoted: config.promoted.clone(), auth: resolver.clone().expect("resolver built for enabled roles"), encode_workers: params.encode_workers, + graph_emitter: graph_emitter.clone(), }) .await?; println!("receiver gRPC listening on {}", handle.grpc_addr); @@ -908,15 +917,17 @@ async fn main() -> Result<(), Box> { // the disabled state is logged so it's visible in a multi-pod rollout. let compactor = if config.compaction_enabled { let audit_store = store.clone(); - Some( - Compactor::new( - store, - CompactionPolicy::default(), - config.compaction_interval, - ) - .with_promoted_attributes(config.promoted.clone()) - .with_audit_sink(Box::new(ParquetAuditSink::new(audit_store))), + let mut compactor = Compactor::new( + store, + CompactionPolicy::default(), + config.compaction_interval, ) + .with_promoted_attributes(config.promoted.clone()) + .with_audit_sink(Box::new(ParquetAuditSink::new(audit_store))); + if let Some(emitter) = graph_emitter.clone() { + compactor = compactor.with_graph_emitter(emitter); + } + Some(compactor) } else { tracing::info!(name: ourios_semconv::EVENT_OURIOS_SERVER_COMPACTION_DISABLED, "compaction disabled for this process (OURIOS_COMPACTION_ENABLED)"); None diff --git a/crates/ourios-server/src/receiver.rs b/crates/ourios-server/src/receiver.rs index 9a02cb97..556a93d9 100644 --- a/crates/ourios-server/src/receiver.rs +++ b/crates/ourios-server/src/receiver.rs @@ -274,6 +274,9 @@ pub struct ReceiverConfig { /// (`receiver.encode_workers`; the config layer validates ≥ 1 and /// defaults to the host's available cores). pub encode_workers: usize, + /// The RFC 0047 §3.3 graph emitter, fed on the flush cadence, when the + /// graph is configured with a bound conversation object. + pub graph_emitter: Option>, } /// A running receiver role: the **resolved** bound addresses (so a `:0` @@ -544,7 +547,10 @@ pub async fn serve(config: ReceiverConfig) -> Result { // The cadence age-sweep publishes through the coordinator: atomic drain // under the miner lock + audit-ordered off-lock write (issue #302 #1/#2). - let coordinator = PublishCoordinator::new(sink.clone(), audit_sink.clone()); + let mut coordinator = PublishCoordinator::new(sink.clone(), audit_sink.clone()); + if let Some(emitter) = config.graph_emitter.clone() { + coordinator = coordinator.with_graph_emitter(emitter); + } let flush_tick = spawn_age_sweep( pipeline.clone(), coordinator, @@ -886,6 +892,7 @@ mod tests { store, promoted: PromotedAttributes::default(), auth: AuthResolver::static_only(None), + graph_emitter: None, encode_workers: 2, }) .await @@ -917,6 +924,7 @@ mod tests { store, promoted: PromotedAttributes::default(), auth: AuthResolver::static_only(None), + graph_emitter: None, encode_workers: 2, }) .await @@ -1051,6 +1059,7 @@ mod tests { store, promoted: PromotedAttributes::default(), auth: AuthResolver::static_only(None), + graph_emitter: None, encode_workers: 2, }) .await diff --git a/crates/ourios-server/tests/it/main.rs b/crates/ourios-server/tests/it/main.rs index 9cbef07b..445e4a08 100644 --- a/crates/ourios-server/tests/it/main.rs +++ b/crates/ourios-server/tests/it/main.rs @@ -34,5 +34,6 @@ mod rfc0039_4_sampling; mod rfc0043_5_event_name_query; mod rfc0046_out_of_band_tenancy; mod rfc0047_9_tool_gate; +mod rfc0047_emitter; mod rfc0047_openfga; mod rfc0047_visibility; diff --git a/crates/ourios-server/tests/it/rfc0047_9_tool_gate.rs b/crates/ourios-server/tests/it/rfc0047_9_tool_gate.rs index 7cda75dc..2f18305d 100644 --- a/crates/ourios-server/tests/it/rfc0047_9_tool_gate.rs +++ b/crates/ourios-server/tests/it/rfc0047_9_tool_gate.rs @@ -1,6 +1,7 @@ //! Scenario RFC0047.9 — the MCP tool gate, in-process against a fake -//! `OpenFGA` (no container): the tenant-wide bypass issues no `Check`, a -//! scoped principal needs an explicit `caller` grant per tool, and an +//! `OpenFGA` (no container): a tenant-wide reader skips the per-tool +//! `can_call` check (the two-step's own `Check(can_read_content, tenant)` +//! still runs), a scoped principal needs an explicit `caller` grant per tool, and an //! unanswerable graph fails the call closed. Complements the served-binary //! arm in `rfc0047_visibility` (real container). //! See `docs/rfcs/0047-rebac-resolver-and-graph-visibility.md` §5. diff --git a/crates/ourios-server/tests/it/rfc0047_emitter.rs b/crates/ourios-server/tests/it/rfc0047_emitter.rs new file mode 100644 index 00000000..81d836ac --- /dev/null +++ b/crates/ourios-server/tests/it/rfc0047_emitter.rs @@ -0,0 +1,259 @@ +//! RFC 0047 §3.3 / §3.6 against a **real `OpenFGA` container** (testcontainers; +//! CI-gated like the other RFC 0047 container tests — `#[ignore]`d in the +//! default run): the compaction sweep feeds the graph from stored rows, +//! and a requested erasure removes the rows, then the tuples. +//! +//! Scenarios RFC0047.10 (emitter) and RFC0047.11 (erasure), plus the +//! end-to-end proof that emitted tuples bind a participant on the served +//! binary (no operator-written conversation tuples anywhere). +//! See `docs/rfcs/0047-rebac-resolver-and-graph-visibility.md` §5. + +use std::sync::Arc; +use std::time::Duration; + +use ourios_core::audit::{AuditPayload, SharedAuditSink}; +use ourios_core::auth::openfga::{ + OpenFgaClient, OpenFgaSpec, TupleKey, VisibilityObjectSpec, VisibilitySpec, + build_openfga_config, +}; +use ourios_ingester::compactor::{pending_erasures, request_erasure, sweep_once}; +use ourios_ingester::graph_emitter::GraphEmitter; +use ourios_parquet::{CompactionPolicy, Store}; +use testcontainers_modules::testcontainers::core::ContainerPort; +use testcontainers_modules::testcontainers::runners::AsyncRunner; +use testcontainers_modules::testcontainers::{GenericImage, ImageExt}; +use tokio::time::timeout; + +use crate::rfc0029_oidc::claim_binding::spawn_with_auth_and_storage; +use crate::rfc0029_oidc::ingest_binding::{make_key, serve_issuer}; +use crate::rfc0047_openfga::{OPENFGA_IMAGE, OPENFGA_TAG, mint, provision, tuple}; +use crate::rfc0047_visibility::{conversations, promoted, query, row_at, write_records}; + +/// 2026-04-02T10:58:00 UTC — a long-sealed hour, so the partition is a +/// compaction candidate under the default policy. +const TS0: u64 = 1_775_127_480_000_000_000; +/// A window that reaches back to `TS0` from any 2026 wall clock. +const ALL: &str = "true | range(-365d, now) | limit 100"; + +fn row(i: u64, conversation: &str, user: &str) -> ourios_core::record::MinedRecord { + row_at(TS0 + i * 1_000, conversation, user) +} + +/// Scenarios RFC0047.10 / RFC0047.11 on a real graph. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +#[allow(clippy::too_many_lines)] // one container, one store, both scenarios in sequence +#[ignore = "RFC0047.10–.11 — needs Docker (real OpenFGA container); run by the openfga-resolver CI job via --ignored"] +async fn rfc0047_10_11_emitter_and_erasure_end_to_end() { + // --- OpenFGA ----------------------------------------------------------- + let container = GenericImage::new(OPENFGA_IMAGE, OPENFGA_TAG) + .with_exposed_port(ContainerPort::Tcp(8080)) + .with_cmd(["run"]) + .start() + .await + .expect("openfga started"); + let port = container + .get_host_port_ipv4(8080) + .await + .expect("mapped port"); + let api_url = format!("http://127.0.0.1:{port}"); + let http = reqwest::Client::new(); + timeout(Duration::from_secs(60), async { + loop { + if let Ok(response) = http.get(format!("{api_url}/healthz")).send().await + && response.status().is_success() + { + return; + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + }) + .await + .expect("openfga healthy before timeout"); + let (store_id, model_id) = provision(&api_url).await; + let config = build_openfga_config(&OpenFgaSpec { + api_url: Some(api_url.clone()), + store_id: Some(store_id.clone()), + authorization_model_id: Some(model_id.clone()), + visibility: VisibilitySpec { + objects: vec![VisibilityObjectSpec { + object_type: Some("conversation".to_string()), + column: Some("attr.gen_ai.conversation.id".to_string()), + }], + ..VisibilitySpec::default() + }, + ..OpenFgaSpec::default() + }) + .expect("config"); + let fga = OpenFgaClient::new(&config).expect("client"); + let emitter = Arc::new( + GraphEmitter::from_config(&config) + .expect("emitter") + .expect("conversation bound"), + ); + + // --- Parquet: two files in one sealed hour → a compaction candidate ---- + let tmp = tempfile::TempDir::new().expect("temp"); + let bucket = tmp.path().to_path_buf(); + write_records( + &bucket, + &[ + row(1, "c-1", "alice"), + row(2, "c-1", "alice"), + row(3, "c-2", "bob"), + ], + ); + write_records(&bucket, &[row(4, "c-3", "carol"), row(5, "c-2", "bob")]); + let store = Store::local(&bucket).expect("store"); + let audit = SharedAuditSink::new(); + + // --- RFC0047.10: the sweep feeds the graph ----------------------------- + let (result, _, sink) = sweep_once( + store.clone(), + CompactionPolicy::default(), + promoted(), + Box::new(audit.clone()), + Some(Arc::clone(&emitter)), + ) + .await; + let report = result.expect("sweep"); + assert_eq!(report.partitions_compacted, 1, "{report:?}"); + assert!(report.errors.is_empty(), "{:?}", report.errors); + assert!(report.graph_tuples_emitted > 0); + for (user, relation, object) in [ + ("tenant:acme", "parent", "conversation:acme/c-1"), + ("user:alice", "participant", "conversation:acme/c-1"), + ("user:alice", "scoped_reader", "tenant:acme"), + ("user:bob", "participant", "conversation:acme/c-2"), + ("tenant:acme", "parent", "conversation:acme/c-3"), + ("tenant:acme", "parent", "tool:acme/query_logs"), + ] { + let tuples = fga.read_by_object(object).await.expect("read"); + assert!( + tuples.contains(&TupleKey::new(user, relation, object)), + "missing {user} {relation} {object}: {tuples:?}" + ); + } + // Idempotent: a second sweep has nothing to rewrite and writes nothing new. + let count = |object: &str| { + let fga = fga.clone(); + let object = object.to_string(); + async move { fga.read_by_object(&object).await.expect("read").len() } + }; + let c1_before = count("conversation:acme/c-1").await; + let (result, _, sink) = sweep_once( + store.clone(), + CompactionPolicy::default(), + promoted(), + sink, + Some(Arc::clone(&emitter)), + ) + .await; + let report = result.expect("sweep"); + assert_eq!(report.partitions_compacted, 0); + assert_eq!(report.graph_tuples_emitted, 0); + assert_eq!(count("conversation:acme/c-1").await, c1_before); + + // --- The emitted tuples bind a participant on the served binary ------ + // No operator wrote a single conversation tuple: alice's `participant` + // and binding tuples came from her rows. + let (encoding, jwk) = make_key("key-1"); + let issuer = serve_issuer(jwk).await; + let storage_yaml = " promoted_attributes:\n log: [gen_ai.conversation.id, user.hash, model, {key: cost_usd, type: f64}]\n"; + let auth_yaml = format!( + "auth:\n\ + \x20\x20oidc:\n\ + \x20\x20\x20\x20issuer: {issuer}\n\ + \x20\x20\x20\x20audience: ourios\n\ + \x20\x20openfga:\n\ + \x20\x20\x20\x20api_url: {api_url}\n\ + \x20\x20\x20\x20store_id: {store_id}\n\ + \x20\x20\x20\x20authorization_model_id: {model_id}\n\ + \x20\x20\x20\x20session_ttl_secs: 1\n\ + \x20\x20\x20\x20visibility:\n\ + \x20\x20\x20\x20\x20\x20objects:\n\ + \x20\x20\x20\x20\x20\x20\x20\x20- type: conversation\n\ + \x20\x20\x20\x20\x20\x20\x20\x20\x20\x20column: attr.gen_ai.conversation.id\n" + ); + let (mut child, _grpc, _http, querier) = + spawn_with_auth_and_storage(&tmp, storage_yaml, &auth_yaml, &[]).await; + let alice = mint(&encoding, &issuer, "alice", &[], false); + let (status, body) = query(&http, querier, &alice, "acme", ALL).await; + assert_eq!(status, 200, "{body}"); + assert_eq!( + conversations(&body), + ["c-1", "c-1"], + "alice sees exactly her conversation, through tuples the data produced" + ); + let bob = mint(&encoding, &issuer, "bob", &[], false); + let (status, body) = query(&http, querier, &bob, "acme", ALL).await; + assert_eq!(status, 200, "{body}"); + assert_eq!(conversations(&body), ["c-2", "c-2"]); + child.kill().await.expect("kill the server"); + + // --- RFC0047.11: erasure removes tuples after rows --------------------- + let _ = audit.drain(); + request_erasure(&store, "acme", "c-1").expect("request"); + let (result, _, _) = sweep_once( + store.clone(), + CompactionPolicy::default(), + promoted(), + sink, + Some(Arc::clone(&emitter)), + ) + .await; + let report = result.expect("sweep"); + assert!(report.errors.is_empty(), "{:?}", report.errors); + let outcome = &report.erasures[0]; + assert_eq!(outcome.rows_dropped, 2); + assert_eq!(outcome.tuples_deleted, Some(2), "parent + participant"); + assert!(outcome.finished); + assert!(pending_erasures(&store).expect("pending").is_empty()); + assert!( + fga.read_by_object("conversation:acme/c-1") + .await + .expect("read") + .is_empty(), + "no tuple for the object remains" + ); + assert!( + !fga.check( + &tuple("user:alice", "can_read_content", "conversation:acme/c-1"), + &[] + ) + .await + .expect("check"), + "unreachable" + ); + assert!( + !fga.read_by_object("conversation:acme/c-2") + .await + .expect("read") + .is_empty(), + "other conversations untouched" + ); + // Audit ordering: the erasure event is the last event of the sweep, + // after the rewrite's compaction events. + let events = audit.drain(); + let last = events.last().expect("events"); + assert!( + matches!(&last.payload, AuditPayload::ConversationErased { conversation_id, rows_dropped: 2, tuples_deleted: 2, .. } if conversation_id == "c-1"), + "{last:?}" + ); + // The rows are gone from the store: the served binary no longer + // returns c-1 to a tenant-wide reader either. + fga.write(&[tuple("user:zed", "reader", "tenant:acme")], &[]) + .await + .expect("grant"); + let (mut child, _grpc, _http, querier) = + spawn_with_auth_and_storage(&tmp, storage_yaml, &auth_yaml, &[]).await; + let zed = mint(&encoding, &issuer, "zed", &[], false); + let (status, body) = query(&http, querier, &zed, "acme", ALL).await; + assert_eq!(status, 200, "{body}"); + assert_eq!( + conversations(&body), + ["c-2", "c-2", "c-3"], + "c-1's rows are gone" + ); + child.kill().await.expect("kill the server"); + drop(container); +} diff --git a/crates/ourios-server/tests/it/rfc0047_visibility.rs b/crates/ourios-server/tests/it/rfc0047_visibility.rs index 6c75c76f..38a7b804 100644 --- a/crates/ourios-server/tests/it/rfc0047_visibility.rs +++ b/crates/ourios-server/tests/it/rfc0047_visibility.rs @@ -60,8 +60,13 @@ fn recent_ns(offset: u64) -> u64 { } /// One `acme` row in `conversation` with the given `user.hash`, a content -/// attribute, a model and a $1.50 cost. -fn row(i: u64, conversation: &str, user: &str) -> MinedRecord { +/// attribute, a model and a $1.50 cost, timestamped a minute ago. +pub(crate) fn row(i: u64, conversation: &str, user: &str) -> MinedRecord { + row_at(recent_ns(i), conversation, user) +} + +/// [`row`] at an explicit timestamp. +pub(crate) fn row_at(time_unix_nano: u64, conversation: &str, user: &str) -> MinedRecord { MinedRecord { tenant_id: TenantId::new("acme"), template_id: 1, @@ -73,7 +78,7 @@ fn row(i: u64, conversation: &str, user: &str) -> MinedRecord { scope_attributes: Vec::new(), resource_schema_url: None, scope_schema_url: None, - time_unix_nano: recent_ns(i), + time_unix_nano, observed_time_unix_nano: None, attributes: vec![ kv("gen_ai.conversation.id", conversation), @@ -100,7 +105,7 @@ fn row(i: u64, conversation: &str, user: &str) -> MinedRecord { } } -fn promoted() -> PromotedAttributes { +pub(crate) fn promoted() -> PromotedAttributes { PromotedAttributes::new_typed( [], [ @@ -115,7 +120,7 @@ fn promoted() -> PromotedAttributes { ) } -fn write_records(bucket: &Path, recs: &[MinedRecord]) { +pub(crate) fn write_records(bucket: &Path, recs: &[MinedRecord]) { let store = Store::local(bucket).expect("local store"); let mut by_part: HashMap> = HashMap::new(); for r in recs { @@ -133,7 +138,7 @@ fn write_records(bucket: &Path, recs: &[MinedRecord]) { } /// `POST /v1/query` with a bearer and tenant; returns (status, JSON body). -async fn query( +pub(crate) async fn query( http: &reqwest::Client, addr: std::net::SocketAddr, bearer: &str, @@ -227,7 +232,7 @@ async fn mcp_call( } /// The sorted conversation ids of a row response. -fn conversations(body: &serde_json::Value) -> Vec { +pub(crate) fn conversations(body: &serde_json::Value) -> Vec { let mut ids: Vec = body["records"] .as_array() .expect("records") diff --git a/docs/guides/authentication.md b/docs/guides/authentication.md index 57e23a95..a4584481 100644 --- a/docs/guides/authentication.md +++ b/docs/guides/authentication.md @@ -185,6 +185,30 @@ enumeration runs per query. The branch a query took is recorded on `ourios.query.visibility` (`ourios.query.visibility.branch`) and the request span. +### Feeding the graph from the data + +With a `conversation` object bound, Ourios writes the data-derived +tuples itself (RFC 0047 §3.3) — nobody hand-writes conversation grants: +for every stored row the compaction sweep rewrites, and for every batch +the receiver flushes, the emitter derives +`conversation:T/#parent@tenant:T`, `#participant@user:` (plus `tenant:T#scoped_reader@user:<…>`), +`#actor@agent:` (plus its binding tuple) and the +per-tenant `tool:T/#parent@tenant:T` objects, and writes them in +idempotent ≤ 100-tuple batches (`ourios.graph.tuples`). Operators write +only the administrative tuples (`tenant#reader/writer/owner/ +metadata_reader`, `team#member`, `tool#caller`, `conversation#delegate`). + +**Erasing a conversation** (RFC 0047 §3.6): write a request marker into +the object store — `erasure/tenant_id=/conversation=` (the +same percent-encoding as `data/tenant_id=…`, body `{"phase":"rows"}`; +`ourios-server`'s compactor exposes it as +`ourios_ingester::compactor::request_erasure`) — and the next sweep +rewrites every partition of the tenant with the conversation's rows +dropped, then deletes its graph tuples, then writes a +`conversation_erased` audit event and removes the marker. Rows first, +tuples after — a dangling tuple is harmless, a dangling row is a leak. + The binding is cached per credential for `session_ttl_secs` and is **fail-closed**: an unreachable or slow OpenFGA answers `503` on the query and MCP surfaces and `UNAVAILABLE`/`503` on ingest, and diff --git a/docs/rfcs/0005-parquet-storage.md b/docs/rfcs/0005-parquet-storage.md index 2a69d33e..090a4750 100644 --- a/docs/rfcs/0005-parquet-storage.md +++ b/docs/rfcs/0005-parquet-storage.md @@ -621,6 +621,10 @@ mapping: | `3` | `compaction` | `Compaction` | RFC 0009 §3.6 (amendment 2026-06-03) | | `4` | `alias_asserted` | `AliasAsserted` | RFC 0001 §6.7 (amendment 2026-06-12) | | `5` | `alias_retracted` | `AliasRetracted` | RFC 0001 §6.7 (amendment 2026-06-12) | +| `6` | `template_created` | `Template { change: Created }` | RFC 0017 §3.1 | +| `7` | `record_quarantined` | `RecordQuarantined` | RFC 0025 §3.3 | +| `8` | `ingest_denied` | `IngestDenied` | RFC 0026 §3.4 | +| `9` | `conversation_erased` | `ConversationErased` | RFC 0047 §3.6 (amendment 2026-08-18: OPTIONAL `erasure_conversation_id` / `erasure_partitions` / `erasure_rows` / `erasure_tuples` columns, NULL for every other kind) | Adding a new ordinal is a §3.8 additive amendment; the mapping table is the source of truth and a new ordinal lands as a new diff --git a/docs/rfcs/0047-rebac-resolver-and-graph-visibility.md b/docs/rfcs/0047-rebac-resolver-and-graph-visibility.md index a29924c8..ff11f5bd 100644 --- a/docs/rfcs/0047-rebac-resolver-and-graph-visibility.md +++ b/docs/rfcs/0047-rebac-resolver-and-graph-visibility.md @@ -1,7 +1,7 @@ --- rfc: 0047 title: ReBAC resolver (OpenFGA) and graph-fed visibility inside a tenant -status: red +status: green author: Jens Holdgaard Pedersen drafting-assistance: Claude created: 2026-08-17 @@ -11,13 +11,13 @@ superseded-by: — # RFC 0047 — ReBAC resolver and graph-fed visibility -> **Status: `red` (2026-08-18).** Slices 1–3 are green: RFC0047.1–.3 -> (the layer-1 resolver), RFC0047.4–.8 (the planner two-step, masking, -> bounded enumeration) and RFC0047.9 (the MCP tool gate) pass on the served -> binary against a real OpenFGA container (`openfga-resolver` CI job) with -> the in-tree model; RFC0047.12 gates CI. RFC0047.10–.11 (emitter, erasure) -> are the remaining slice (slice 4); the RFC0047.5 request-carried contextual-tuple arm is deferred -> (§3.3, §7). Prerequisite: RFC 0046 (out-of-band tenancy, `green`) — the tenant is an +> **Status: `green` (2026-08-18).** All twelve §5 criteria pass: +> RFC0047.1–.3 (the layer-1 resolver), .4–.8 (the planner two-step, +> masking, bounded enumeration), .9 (the MCP tool gate), .10–.11 (the graph +> emitter and erasure) on the served binary against a real OpenFGA container +> (`openfga-resolver` CI job) with the in-tree model, and .12 gating CI. The +> RFC0047.5 request-carried contextual-tuple arm is deferred (§3.3, §7); +> the erasure request channel is the §3.6 slice-4 decision. Prerequisite: RFC 0046 (out-of-band tenancy, `green`) — the tenant is an > opaque, coarse, credential-selected object, which is exactly the object > type this RFC binds the authorization graph to. Grounded in the #688 > OpenFGA spike (resolver seam holds, p50 1.4 ms), two OpenFGA-assistant @@ -260,6 +260,23 @@ a `delegate` grant on `conversation:T/` MUST be paired by the operator with `tenant:T#scoped_reader@agent:` for the same reason (documented next to the model; the `.fga.yaml` fixture shows the pair). +**Slice-4 decisions (implemented).** The emitter lives in the ingester +(`graph_emitter`), fed from **both** hooks: the compaction sweep observes +every input row it decodes (once, before any drop) and the receiver's +`PublishCoordinator` derives tuples from every batch it publishes and sends +them off the flush path once the batch is durable. The conversation key is +the column bound in `visibility.objects` (`attr.` or `resource.` stripped); +the user keys are `user.hash` and `enduser.pseudo.id`, the agent key +`gen_ai.agent.id`; a value that cannot be an object id is skipped. It also +writes the per-tenant `tool:T/#parent@tenant:T` objects (§3.5), so +operators grant `caller` only. Writes are `on_duplicate = ignore` +(OpenFGA ≥ 1.10, per the OpenFGA assistant), so a resend is a no-op and +no read-then-diff pass exists; the sweep sends what it derived after the +blocking pass, in ≤ 100-tuple batches, counted on +`ourios.graph.tuples{ourios.graph.tuple.operation}`. Data stored before the +graph was configured is fed the next time its partition is rewritten +(compaction) — a backfill sweep is a follow-on (§9). + **Freshness.** A conversation is invisible to fine-grained principals until its tuples land (seconds after the next sweep; the emitter is also invoked on the receiver's flush cadence for the tenants it flushed). Two bridges, @@ -419,6 +436,26 @@ exists — so a deleted conversation is unreachable *and* unlisted. Tuple deletion follows the Parquet rewrite, never precedes it (a dangling tuple is harmless; a dangling row is a leak). +**Slice-4 decisions (implemented).** The RFC left the *request* channel +open; the choice is a durable **marker object in the store** — +`erasure/tenant_id=/conversation=` (the partition +percent-encoding; body `{"phase":"rows"}`) — because the object store is +the source of truth (`CLAUDE.md` §3.6), it needs no new network surface +or credential, and an operator can write it with the tooling they already +have (`ourios_ingester::compactor::request_erasure` in-process). The sweep's +blocking pass rewrites every hour partition of the tenant through the same +compaction rewrite with the conversation's rows dropped (a single-file +partition is rewritten too), advances the marker to `{"phase":"tuples"}` +once every partition rewrote cleanly, and only then — in the async phase, +after the blocking pass — reads the object's tuples and deletes them in +≤ 100-tuple batches (`on_missing = ignore`), writes the new +`conversation_erased` audit event (RFC 0005 §3.7 kind 9, carrying +`partitions_rewritten` / `rows_dropped` / `tuples_deleted`) after the +sweep's compaction events, and removes the marker. A sweep interrupted +between the phases retries only the tuple deletion; an unreachable graph +leaves the marker and retries next sweep. Without a bound conversation +object a marker is recorded as a sweep error, never silently dropped. + ### 3.7 Operational posture Fail-closed everywhere OpenFGA is consulted (resolution, planner, tool @@ -619,8 +656,11 @@ planner's returned row set equals the naive "rows whose conversation ∈ ## 9. Follow-ons (recorded, not built here) -Cluster/service ownership as graph objects (from the reviewed model) once a -scenario needs them; time-boxed grants (`temporal_grant` CEL condition on +A backfill sweep that feeds the graph from data stored before the graph was +configured (today only a rewrite re-derives); an operator-facing erasure +surface over the store marker (a CLI verb / MCP tool); cluster/service +ownership as graph objects (from the reviewed model) once a scenario needs +them; time-boxed grants (`temporal_grant` CEL condition on `tool#caller` / `conversation#delegate`); an upstream write-up for the OTel community once this is validated in a real deployment (#688 Q7). diff --git a/semconv/registry/attributes.yaml b/semconv/registry/attributes.yaml index 5b00556c..cc555bd7 100644 --- a/semconv/registry/attributes.yaml +++ b/semconv/registry/attributes.yaml @@ -181,6 +181,21 @@ groups: the total), so the B1 pruned fraction is derived in the backend as `pruned / (scanned + pruned)` (RFC 0016; OTel usage/state convention — record raw counts, derive the ratio). + - id: ourios.graph.tuple.operation + type: + members: + - id: write + value: "write" + stability: development + brief: Tuples written by the emitter (idempotent, `on_duplicate = ignore`). + - id: delete + value: "delete" + stability: development + brief: Tuples deleted by a conversation erasure (`on_missing = ignore`). + stability: development + brief: >- + Whether a graph tuple batch was a write or a delete (RFC 0047 §3.3 / + §3.6). - id: ourios.query.visibility.branch type: members: diff --git a/semconv/registry/metrics.yaml b/semconv/registry/metrics.yaml index e85d1ac3..ed65443a 100644 --- a/semconv/registry/metrics.yaml +++ b/semconv/registry/metrics.yaml @@ -427,6 +427,21 @@ groups: - ref: ourios.query.visibility.branch requirement_level: required + - id: metric.ourios.graph.tuples + type: metric + metric_name: ourios.graph.tuples + stability: development + brief: >- + Relationship tuples the RFC 0047 §3.3 emitter wrote to (or, for an + erasure, deleted from) the authorization graph, by operation. A + failed batch carries `error.type` (`upstream_unavailable`) with the + batch's tuple count. + instrument: counter + unit: "{tuple}" + attributes: + - ref: ourios.graph.tuple.operation + requirement_level: required + # Audit sink (issue #302): the miner's template-audit write path. A # buffering sink mirroring the RFC 0014 record sink — events buffer on the # request path and flush off the runtime. `flush.outcome` splits a failed