feat(ingester): rfc 0026 green d — rejection telemetry + denial audit

crates/ourios-core/src/alias.rs

@@ -401,6 +401,7 @@ impl AliasMap {
             AuditPayload::Template { .. }
             | AuditPayload::Compaction { .. }
             | AuditPayload::RecordQuarantined { .. }
+            | AuditPayload::IngestDenied { .. }
             | AuditPayload::Unknown { .. } => {}
         }
     }

crates/ourios-core/src/audit.rs

@@ -236,6 +236,15 @@ pub enum AuditPayload {
         /// can never be written.
         error: String,
     },
+    /// An authenticated sender attempted to write outside its token's
+    /// allowed tenant set; the whole batch was rejected before the WAL
+    /// (RFC 0026 §3.2/§3.4). The event's `tenant_id` is the **offending**
+    /// derived tenant; the payload carries the rejecting token's audit
+    /// label — never the token value.
+    IngestDenied {
+        /// The rejecting token's audit/metric label (RFC 0026 §3.4).
+        token_name: String,
+    },
 }
 
 /// Stable on-disk `event_kind` ordinals (RFC 0005 §3.7 mapping).
@@ -260,6 +269,9 @@ pub const EVENT_KIND_TEMPLATE_CREATED: u8 = 6;
 /// append-only addition; old readers surface it via the
 /// [`AuditPayload::Unknown`] tolerance path.
 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;
 
 /// Canonical `event_type` strings paired with the ordinals above
 /// (RFC 0005 §3.7 / RFC 0001 §6.4 / RFC 0009 §3.6).
@@ -279,6 +291,8 @@ pub const EVENT_TYPE_ALIAS_RETRACTED: &str = "alias_retracted";
 pub const EVENT_TYPE_TEMPLATE_CREATED: &str = "template_created";
 /// See [`EVENT_TYPE_TEMPLATE_WIDENED`]. RFC 0025 §3.3 sink quarantine.
 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 `template_version` a leaf is born at (RFC 0017 §3.1). The
 /// [`TemplateChange::Created`] variant omits a version field — the invariant
@@ -305,6 +319,7 @@ impl AuditPayload {
             Self::AliasRetracted { .. } => EVENT_KIND_ALIAS_RETRACTED,
             Self::Compaction { .. } => EVENT_KIND_COMPACTION,
             Self::RecordQuarantined { .. } => EVENT_KIND_RECORD_QUARANTINED,
+            Self::IngestDenied { .. } => EVENT_KIND_INGEST_DENIED,
             Self::Unknown { event_kind, .. } => *event_kind,
         }
     }
@@ -328,6 +343,7 @@ impl AuditPayload {
             Self::AliasRetracted { .. } => EVENT_TYPE_ALIAS_RETRACTED,
             Self::Compaction { .. } => EVENT_TYPE_COMPACTION,
             Self::RecordQuarantined { .. } => EVENT_TYPE_RECORD_QUARANTINED,
+            Self::IngestDenied { .. } => EVENT_TYPE_INGEST_DENIED,
             Self::Unknown { event_type, .. } => event_type,
         }
     }
@@ -348,6 +364,7 @@ impl AuditPayload {
             | Self::AliasRetracted { .. }
             | Self::Compaction { .. }
             | Self::RecordQuarantined { .. }
+            | Self::IngestDenied { .. }
             | Self::Unknown { .. } => false,
         }
     }

crates/ourios-ingester/src/metrics.rs

@@ -318,8 +318,24 @@ impl IngestMetrics {
         }
         self.append_duration.record(elapsed.as_secs_f64(), &[]);
     }
+
+    /// Record one rejected request on the existing `ourios.ingest.batches`
+    /// counter with `error.type` (`unauthenticated` | `permission_denied`)
+    /// — the RFC 0026 §3.4 recording-errors convention: no new metric
+    /// name, the reason on a low-cardinality attribute.
+    pub fn record_rejected_batch(&self, error_type: &'static str) {
+        self.batches
+            .add(1, &[KeyValue::new(ERROR_TYPE, error_type)]);
+    }
 }
 
+/// The `error.type` value for a missing/malformed/unknown bearer
+/// (RFC 0026 §3.4).
+pub const ERROR_TYPE_UNAUTHENTICATED: &str = "unauthenticated";
+/// The `error.type` value for an authenticated cross-tenant rejection
+/// (RFC 0026 §3.4).
+pub const ERROR_TYPE_PERMISSION_DENIED: &str = "permission_denied";
+
 /// The OpenTelemetry-standard `error.type` attribute key (semconv, stable).
 /// Deliberately **not** in the Ourios weaver registry — it is an upstream
 /// OpenTelemetry attribute used here per the "recording errors on metrics"

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

@@ -40,13 +40,20 @@ use crate::receiver::pipeline::{ReceiveError, SharedPipeline};
 #[derive(Clone)]
 pub struct AuthInterceptor {
     store: Option<Arc<TokenStore>>,
+    /// Rejection telemetry (RFC 0026 §3.4). The instruments resolve by
+    /// name through the global meter, so this instance aggregates with
+    /// the pipeline's.
+    metrics: Arc<crate::metrics::IngestMetrics>,
 }
 
 impl AuthInterceptor {
     /// An interceptor over `store` (`None` = open mode pass-through).
     #[must_use]
     pub fn new(store: Option<Arc<TokenStore>>) -> Self {
-        Self { store }
+        Self {
+            store,
+            metrics: Arc::new(crate::metrics::IngestMetrics::new()),
+        }
     }
 }
 
@@ -63,8 +70,14 @@ impl tonic::service::Interceptor for AuthInterceptor {
                 Ok(request)
             }
             // One undifferentiated message: missing vs malformed vs unknown
-            // would be a probing oracle (RFC 0026 §3.2).
-            Err(_) => Err(Status::unauthenticated("a valid bearer token is required")),
+            // would be a probing oracle (RFC 0026 §3.2). §3.4: the
+            // rejection counts on `ourios.ingest.batches`
+            // (`error.type = unauthenticated`).
+            Err(_) => {
+                self.metrics
+                    .record_rejected_batch(crate::metrics::ERROR_TYPE_UNAUTHENTICATED);
+                Err(Status::unauthenticated("a valid bearer token is required"))
+            }
         }
     }
 }

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

@@ -102,6 +102,8 @@ async fn handle_logs(State(state): State<AppState>, headers: HeaderMap, body: By
         .get(header::AUTHORIZATION)
         .and_then(|value| value.to_str().ok());
     let Ok(binding) = authenticate_bearer(state.auth.as_deref(), authorization) else {
+        // RFC 0026 §3.4: the rejection counts on `ourios.ingest.batches` (`error.type = unauthenticated`).
+        state.pipeline.record_unauthenticated();
         return StatusCode::UNAUTHORIZED.into_response();
     };
 

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

@@ -126,8 +126,23 @@ pub struct IngestPipeline {
     last_durable: Mutex<Option<WalOffset>>,
     rotation_hook: Mutex<Option<RotationHook>>,
     /// Ingest throughput + WAL-before-ack latency instruments (RFC 0014
-    /// §6.3). Recorded only on a durably-acked batch.
+    /// §6.3), recorded on durably-acked batches — plus the RFC 0026 §3.4
+    /// rejection counts (`error.type` on the batches counter), recorded
+    /// on the pre-WAL denial paths.
     metrics: IngestMetrics,
+    /// RFC 0026 §3.4: the sink for `ingest_denied` audit events. Behind a
+    /// mutex — denials are the cold path.
+    ///
+    /// **Best-effort durability, deliberately.** The server wires the
+    /// buffering audit sink, so a crash before its next cadence flush can
+    /// drop denial events — and unlike template events they have no WAL
+    /// replay to recover from (the denied batch never reached the WAL,
+    /// which is the §3.2 point). The durable alerting signal is the
+    /// `error.type = permission_denied` counter; the event is forensic
+    /// detail. Making denials synchronously durable would put an fsync on
+    /// the rejection path — a write-amplification lever for any
+    /// authenticated-but-misconfigured (or hostile) sender.
+    denial_audit: Mutex<Option<Box<dyn ourios_core::audit::AuditSink + Send>>>,
 }
 
 impl IngestPipeline {
@@ -142,9 +157,33 @@ impl IngestPipeline {
             last_durable: Mutex::new(None),
             rotation_hook: Mutex::new(None),
             metrics: IngestMetrics::new(),
+            denial_audit: Mutex::new(None),
         }
     }
 
+    /// Install the RFC 0026 §3.4 denial audit sink: every tenant-binding
+    /// rejection emits an `ingest_denied` event through it (the token's
+    /// audit label + the offending tenant — never a token value).
+    #[must_use]
+    pub fn with_denial_audit_sink(
+        self,
+        sink: Box<dyn ourios_core::audit::AuditSink + Send>,
+    ) -> Self {
+        *self
+            .denial_audit
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(sink);
+        self
+    }
+
+    /// Record an authentication rejection on `ourios.ingest.batches`
+    /// (`error.type = unauthenticated`, RFC 0026 §3.4) — called by the
+    /// transports, which own the 401 surface.
+    pub fn record_unauthenticated(&self) {
+        self.metrics
+            .record_rejected_batch(crate::metrics::ERROR_TYPE_UNAUTHENTICATED);
+    }
+
     /// Install the §6.9 rotation-cadence hook: called once per
     /// detected WAL segment rotation with the miner as it stands
     /// and the **rotation-point high-water mark** — the last
@@ -213,9 +252,30 @@ impl IngestPipeline {
         binding: Option<&super::auth::AuthBinding>,
     ) -> Result<usize, ReceiveError> {
         // RFC 0026 §3.2: authz precedes every other ingest step — a denied
-        // batch does no encode, fan-out, or WAL work.
-        if let Some(binding) = binding {
-            super::auth::check_binding(&request, &self.rule, binding)?;
+        // batch does no encode, fan-out, or WAL work. §3.4: the denial counts on
+        // `ourios.ingest.batches` (`error.type = permission_denied`)
+        // and emits the audit event.
+        if let Some(binding) = binding
+            && let Err(e) = super::auth::check_binding(&request, &self.rule, binding)
+        {
+            if let ReceiveError::TenantDenied { token_name, tenant } = &e {
+                self.metrics
+                    .record_rejected_batch(crate::metrics::ERROR_TYPE_PERMISSION_DENIED);
+                let mut sink = self
+                    .denial_audit
+                    .lock()
+                    .unwrap_or_else(std::sync::PoisonError::into_inner);
+                if let Some(sink) = sink.as_mut() {
+                    sink.emit(ourios_core::audit::AuditEvent {
+                        tenant_id: tenant.clone(),
+                        timestamp: std::time::SystemTime::now(),
+                        payload: ourios_core::audit::AuditPayload::IngestDenied {
+                            token_name: token_name.clone(),
+                        },
+                    });
+                }
+            }
+            return Err(e);
         }
         // Encode before fan-out consumes the request: the WAL frame is a
         // protobuf `ExportLogsServiceRequest` (§6.5 step 3). Byte-equality

crates/ourios-ingester/tests/README.md

@@ -19,6 +19,8 @@ installers in one process race each other (see the note in
   global in-memory provider.
 - `rfc0025_quarantine.rs` — its RFC0025.5 telemetry arm installs the
   global in-memory provider.
+- `rfc0026_telemetry.rs` — the RFC0026.7 rejection-telemetry arm installs
+  the global in-memory provider.
 
 `fixtures/` holds the crash-fixture **`[[bin]]` targets** (SIGKILL'd by
 harness tests via `CARGO_BIN_EXE_*`), not test binaries.

crates/ourios-ingester/tests/it/ingest_support/mod.rs

@@ -316,6 +316,24 @@ pub fn shared_wal_pipeline(root: &Path) -> SharedPipeline {
     Arc::new(open_pipeline(root))
 }
 
+/// [`capturing_pipeline`] with an RFC 0026 denial audit sink attached.
+pub fn capturing_pipeline_with_denial_audit(
+    sink: Box<dyn ourios_core::audit::AuditSink + Send>,
+) -> (SharedPipeline, Captured) {
+    let captured = Captured::default();
+    let miner = MinerCluster::new(MinerConfig::default());
+    let pipeline = IngestPipeline::new(
+        coordinator(Box::new(CapturingJournal {
+            captured: captured.clone(),
+            byte: 0,
+        })),
+        miner,
+        TenantRule::service_name(),
+    )
+    .with_denial_audit_sink(sink);
+    (Arc::new(pipeline), captured)
+}
+
 /// A shared pipeline whose `Journal` captures appended payloads, plus the
 /// capture handle.
 pub fn capturing_pipeline() -> (SharedPipeline, Captured) {

crates/ourios-ingester/tests/it/rfc0026_auth.rs

@@ -312,15 +312,6 @@ async fn rfc0026_5_wildcard_binding_ingest() {
     assert_eq!(captured.lock().expect("captured").len(), 3);
 }
 
-/// Scenario RFC0026.7 — rejection telemetry and audit.
-/// See `docs/rfcs/0026-authentication-tenant-binding.md` §5.
-#[test]
-#[ignore = "RFC0026.7 stub — implemented in the telemetry green slice"]
-fn rfc0026_7_rejection_telemetry_and_audit() {
-    todo!(
-        "RFC0026.7 — rejections increment existing counters with \
-         error.type (unauthenticated | permission_denied); ingest authz \
-         rejection emits an audit event with the token name and offending \
-         tenant; token values never appear on any surface"
-    );
-}
+// Scenario RFC0026.7 (rejection telemetry + audit) lives in the dedicated
+// `tests/rfc0026_telemetry.rs` binary — it installs the process-global
+// OTel meter provider (the RFC0028.2 harness-exemption class).