feat(receiver): composite tenant derivation — RFC 0045 implementation

crates/ourios-ingester/src/lib.rs

@@ -40,6 +40,7 @@ pub mod publish;
 pub mod receiver;
 pub mod record_sink;
 pub mod recovery;
+pub mod rule_epochs;
 pub mod snapshot_store;
 
 pub use compactor::{Compactor, IngestError, SweepReport, run_sweep, run_sweep_with_promoted};

crates/ourios-ingester/src/receiver.rs

@@ -42,6 +42,7 @@ pub mod propagation;
 pub mod tenant;
 pub mod tls;
 pub mod tls_serve;
+pub mod watch;
 
 pub use auth::{AuthBinding, AuthResolver, Unauthenticated, authenticate_bearer};
 pub use commit::CommitCoordinator;
@@ -51,4 +52,7 @@ pub use pipeline::{IngestPipeline, Journal, ReceiveError, SharedPipeline};
 pub use propagation::{
     HeaderExtractor, MetadataExtractor, extract_context, extract_context_from_metadata,
 };
-pub use tenant::{TenantResolutionError, TenantRule, fan_out};
+pub use tenant::{
+    TenantDerivation, TenantResolutionError, TenantRule, TenantRuleError, fan_out, fan_out_observed,
+};
+pub use watch::DivergenceWatch;

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

@@ -29,7 +29,8 @@ use tracing::Instrument as _;
 
 use crate::metrics::IngestMetrics;
 use crate::receiver::commit::CommitCoordinator;
-use crate::receiver::tenant::{TenantResolutionError, TenantRule, fan_out};
+use crate::receiver::tenant::{TenantResolutionError, TenantRule, fan_out_observed};
+use crate::receiver::watch::DivergenceWatch;
 
 /// The §6.9 rotation-cadence callback: receives the miner as it
 /// stands and the rotation-point high-water mark. See
@@ -121,6 +122,9 @@ pub struct IngestPipeline {
     coordinator: Arc<CommitCoordinator>,
     miner: Mutex<MinerCluster>,
     rule: TenantRule,
+    /// RFC 0045 §3.4 — the divergence detector, when any watch key is
+    /// configured beyond the rule's own keys.
+    watch: Option<DivergenceWatch>,
     /// The durable high-water mark after the most recent acked batch (or
     /// the startup seed). Behind a mutex: concurrent acks update it, and
     /// the rotation-detection read-then-write must see a consistent value.
@@ -164,6 +168,7 @@ impl IngestPipeline {
             coordinator,
             miner: Mutex::new(miner),
             rule,
+            watch: None,
             last_durable: Mutex::new(None),
             rotation_hook: Mutex::new(None),
             encode_pool: None,
@@ -172,6 +177,14 @@ impl IngestPipeline {
         }
     }
 
+    /// Attach the RFC 0045 §3.4 divergence detector; every derived group
+    /// is observed after fan-out.
+    #[must_use]
+    pub fn with_tenant_watch(mut self, watch: Option<DivergenceWatch>) -> Self {
+        self.watch = watch;
+        self
+    }
+
     /// Enable the RFC 0035 §3.1 ordered/concurrent ingest split: the
     /// gated section runs only Drain match + template-id assignment
     /// (+ audit), and the Parquet-sink emit runs on `pool`. The pool's
@@ -337,7 +350,11 @@ impl IngestPipeline {
 
         // Steps 1–2: fan out per tenant. An unresolvable Resource rejects
         // the entire batch here, before any WAL write (RFC0003.4).
-        let records = fan_out(request, &self.rule)?;
+        let records = fan_out_observed(request, &self.rule, |tenant, attributes| {
+            if let Some(watch) = &self.watch {
+                watch.observe(tenant, attributes);
+            }
+        })?;
 
         // Empty fast path (RFC0003.12): no records → success, no WAL
         // frame, miner untouched.