feat(wal,ingester,server): segment rotation + rotation-triggered snapshot cadence (RFC0008.6)

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

@@ -20,6 +20,11 @@ use prost::Message;
 
 use crate::receiver::tenant::{TenantResolutionError, TenantRule, fan_out};
 
+/// The §6.9 rotation-cadence callback: receives the miner as it
+/// stands and the rotation-point high-water mark. See
+/// [`IngestPipeline::with_rotation_hook`].
+pub type RotationHook = Box<dyn FnMut(&MinerCluster, WalOffset) + Send>;
+
 /// The ingest pipeline shared across a listener's requests. The
 /// single-writer WAL forces serialization; concurrent requests queue on
 /// the mutex (the lock never spans an `.await`, so `std::sync::Mutex`
@@ -78,6 +83,7 @@ pub struct IngestPipeline {
     miner: MinerCluster,
     rule: TenantRule,
     last_durable: Option<WalOffset>,
+    rotation_hook: Option<RotationHook>,
 }
 
 impl IngestPipeline {
@@ -90,9 +96,26 @@ impl IngestPipeline {
             miner,
             rule,
             last_durable: None,
+            rotation_hook: None,
         }
     }
 
+    /// 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
+    /// durable offset in the just-closed segment. The hook runs
+    /// *before* the rotating batch's records reach the miner, so a
+    /// snapshot it takes reflects exactly the frames at or below
+    /// that mark. The caller (the server role) wires this to the
+    /// per-tenant snapshot writer; failures inside the hook are the
+    /// hook's to handle — a snapshot is a rebuildable cache, never
+    /// worth failing the ack over.
+    #[must_use]
+    pub fn with_rotation_hook(mut self, hook: RotationHook) -> Self {
+        self.rotation_hook = Some(hook);
+        self
+    }
+
     /// Seed the durable high-water mark from startup recovery
     /// (`RecoveryReport::max_delivered`). Without the seed, a process
     /// that serves zero requests writes shutdown snapshots with no
@@ -138,11 +161,42 @@ impl IngestPipeline {
 
         // Step 3: append the export as one OtlpBatch frame. Step 4: fsync
         // — the batch is durable before the ack below.
+        let before = self.last_durable;
         self.journal.append_batch(&payload)?;
         if let Some(offset) = self.journal.sync()? {
             self.last_durable = Some(offset);
         }
 
+        // §6.9 rotation cadence: a segment change between the previous
+        // durable offset and this one means the WAL rotated under this
+        // batch. Fire the hook with the rotation-point high-water mark
+        // (the old segment's last durable offset) BEFORE this batch's
+        // records reach the miner — a snapshot taken by the hook then
+        // reflects exactly the frames at or below that mark.
+        if let (Some(hook), Some(prev), Some(now)) =
+            (self.rotation_hook.as_mut(), before, self.last_durable)
+            && prev.segment != now.segment
+        {
+            // A hook panic must not unwind `ingest`: the batch is
+            // already durable and the unwind would poison the shared
+            // pipeline mutex, halting all future ingestion over a
+            // best-effort cache write. `AssertUnwindSafe` is sound
+            // for the pipeline's own state — the hook sees the miner
+            // through `&MinerCluster`, so no pipeline mutation can be
+            // torn mid-panic; the hook's own captures are its to keep
+            // consistent (it stays installed and is only ever invoked
+            // best-effort).
+            let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+                hook(&self.miner, prev);
+            }));
+            if outcome.is_err() {
+                eprintln!(
+                    "rotation hook panicked; continuing — the snapshot \
+                     is a rebuildable cache (recovery falls back to the WAL)"
+                );
+            }
+        }
+
         // Step 5: hand records to the miner (only after durability, so a
         // crash between fsync and here replays from the WAL).
         for record in &records {
@@ -299,4 +353,99 @@ mod tests {
         pipeline.ingest(request()).expect("ingest");
         assert_eq!(pipeline.last_durable(), Some(seed));
     }
+
+    /// `sync` reports offsets from a queue, so a segment change can be
+    /// staged mid-sequence.
+    struct SequenceJournal {
+        offsets: Vec<WalOffset>,
+    }
+
+    impl Journal for SequenceJournal {
+        fn append_batch(&mut self, _payload: &[u8]) -> Result<(), ReceiveError> {
+            Ok(())
+        }
+
+        fn sync(&mut self) -> Result<Option<WalOffset>, ReceiveError> {
+            Ok(Some(self.offsets.remove(0)))
+        }
+    }
+
+    #[test]
+    fn rotation_hook_fires_once_with_the_old_segments_last_durable_offset() {
+        let in_first = WalOffset {
+            segment: uuid::Uuid::from_u128(1),
+            byte: 100,
+        };
+        let in_second = WalOffset {
+            segment: uuid::Uuid::from_u128(2),
+            byte: 40,
+        };
+        let calls = Arc::new(Mutex::new(Vec::new()));
+        let seen = calls.clone();
+        let mut pipeline = IngestPipeline::new(
+            Box::new(SequenceJournal {
+                offsets: vec![in_first, in_second, in_second],
+            }),
+            MinerCluster::new(MinerConfig::default()),
+            TenantRule::service_name(),
+        )
+        .with_rotation_hook(Box::new(move |miner, mark| {
+            // Capture the miner's template count at hook time: the
+            // rotating batch must not have reached it yet.
+            let count = miner.template_count(&ourios_core::tenant::TenantId::new("checkout"));
+            seen.lock().expect("lock").push((mark, count));
+        }));
+
+        // Batch 1: no previous durable offset — never a rotation.
+        pipeline.ingest(request()).expect("batch 1");
+        assert!(calls.lock().expect("lock").is_empty());
+
+        // Batch 2: segment changed — the hook fires once with the OLD
+        // segment's last durable offset, before batch 2 hits the miner
+        // (the template count is still batch 1's).
+        pipeline.ingest(request()).expect("batch 2");
+        assert_eq!(*calls.lock().expect("lock"), vec![(in_first, 1)]);
+
+        // Batch 3: same segment — no further firing.
+        pipeline.ingest(request()).expect("batch 3");
+        assert_eq!(calls.lock().expect("lock").len(), 1);
+    }
+
+    /// A panicking hook must not unwind `ingest`: the batch is
+    /// already durable, and the unwind would poison the shared
+    /// pipeline mutex and halt all future ingestion over a
+    /// best-effort cache write.
+    #[test]
+    fn rotation_hook_panic_does_not_fail_the_ingest() {
+        let in_first = WalOffset {
+            segment: uuid::Uuid::from_u128(1),
+            byte: 100,
+        };
+        let in_second = WalOffset {
+            segment: uuid::Uuid::from_u128(2),
+            byte: 40,
+        };
+        let mut pipeline = IngestPipeline::new(
+            Box::new(SequenceJournal {
+                offsets: vec![in_first, in_second, in_second],
+            }),
+            MinerCluster::new(MinerConfig::default()),
+            TenantRule::service_name(),
+        )
+        .with_rotation_hook(Box::new(|_, _| panic!("snapshot writer blew up")));
+
+        pipeline.ingest(request()).expect("batch 1");
+        // Batch 2 rotates and the hook panics — the ingest still acks
+        // and the records still reach the miner.
+        assert_eq!(pipeline.ingest(request()).expect("batch 2 acks"), 1);
+        assert_eq!(
+            pipeline
+                .miner()
+                .template_count(&ourios_core::tenant::TenantId::new("checkout")),
+            1,
+            "the rotating batch's records reached the miner despite the panic",
+        );
+        // The pipeline stays usable afterwards.
+        assert_eq!(pipeline.ingest(request()).expect("batch 3 acks"), 1);
+    }
 }

crates/ourios-ingester/tests/rfc0008_10_rotation_cadence.rs

@@ -0,0 +1,134 @@
+//! RFC0008.10 *snapshot-cadence arm* — a WAL segment rotation
+//! triggers a per-tenant snapshot write recording the
+//! rotation-point high-water mark (RFC 0001 §6.9's primary
+//! cadence). See `docs/rfcs/0008-wal.md` §5.
+//!
+//! Drives a real `Wal` across its age cap (the §6.9 minimum,
+//! 1 s) through the live `IngestPipeline` with the same
+//! rotation hook the server role installs, then asserts the
+//! artefact on disk: stamped with the *old* segment's last
+//! durable offset, and reflecting only the records ingested
+//! before the rotating batch.
+
+mod ingest_support;
+
+use std::path::Path;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use ingest_support::{request, resource_logs, wal_config};
+use ourios_ingester::receiver::{IngestPipeline, TenantRule};
+use ourios_ingester::{recovery, snapshot_store};
+use ourios_miner::cluster::MinerCluster;
+use ourios_miner::snapshot::load_snapshot;
+use ourios_wal::{Wal, WalConfig};
+
+fn rotating_pipeline(root: &Path, snapshots_root: &Path) -> IngestPipeline {
+    let wal = Wal::open(WalConfig {
+        segment_age_secs: 1,
+        ..wal_config(root)
+    })
+    .expect("open WAL");
+    let hook_root = snapshots_root.to_path_buf();
+    IngestPipeline::new(
+        Box::new(wal),
+        MinerCluster::new(ourios_core::config::MinerConfig::default()),
+        TenantRule::service_name(),
+    )
+    .with_rotation_hook(Box::new(move |miner, mark| {
+        recovery::write_snapshots(&hook_root, miner, Some(mark)).expect("snapshot write");
+    }))
+}
+
+#[test]
+fn rotation_writes_snapshots_at_the_rotation_point_high_water() {
+    let tmp = tempfile::TempDir::new().expect("temp");
+    let snapshots_root = tmp.path().join("snapshots");
+    let mut pipeline = rotating_pipeline(tmp.path(), &snapshots_root);
+
+    // Batch A lands in the first segment; the durable mark after it
+    // is the rotation-point high-water the hook must stamp.
+    pipeline
+        .ingest(request(vec![resource_logs("svc", &["user 1 logged in"])]))
+        .expect("batch A");
+    let rotation_point = pipeline.last_durable().expect("durable after batch A");
+    assert!(
+        snapshot_store::load_all(&snapshots_root)
+            .expect("load")
+            .is_empty(),
+        "no rotation yet, no cadence write",
+    );
+
+    // Crossing the age cap makes batch B rotate; the hook fires
+    // before B's records reach the miner. (Age is a full Duration
+    // from the segment UUID's millisecond mint time, so 1.2 s of
+    // wall time is comfortably past the strict 1 s cap.)
+    std::thread::sleep(Duration::from_millis(1_200));
+    pipeline
+        .ingest(request(vec![resource_logs("svc", &["payment 9 settled"])]))
+        .expect("batch B");
+
+    let artefacts = snapshot_store::load_all(&snapshots_root).expect("load");
+    assert_eq!(artefacts.len(), 1, "one tenant, one artefact");
+    let (tenant, bytes) = &artefacts[0];
+    assert_eq!(tenant.as_str(), "svc");
+    let state = load_snapshot(bytes).expect("known version");
+    let mark = state.wal_high_water.expect("stamped with a horizon");
+    assert_eq!(
+        mark.segment,
+        rotation_point.segment.to_string(),
+        "the artefact records the rotation-point segment (the just-closed one)",
+    );
+    assert_eq!(
+        mark.byte, rotation_point.byte,
+        "…at the old segment's last durable byte",
+    );
+    assert_eq!(
+        state.leaves.len(),
+        1,
+        "the snapshot reflects only batch A — the hook ran before \
+         batch B's records reached the miner",
+    );
+
+    // The post-rotation state is intact: both batches in the miner.
+    assert_eq!(
+        pipeline
+            .miner()
+            .template_count(&ourios_core::tenant::TenantId::new("svc")),
+        2,
+        "batch B still reached the miner after the hook",
+    );
+}
+
+/// The hook only fires on a rotation — steady-state batches in one
+/// segment write nothing.
+#[test]
+fn no_rotation_means_no_cadence_write() {
+    let tmp = tempfile::TempDir::new().expect("temp");
+    let snapshots_root = tmp.path().join("snapshots");
+    let hook_root = snapshots_root.clone();
+    let fired = Arc::new(Mutex::new(0u32));
+    let count = fired.clone();
+    let wal = Wal::open(wal_config(tmp.path())).expect("open WAL");
+    let mut pipeline = IngestPipeline::new(
+        Box::new(wal),
+        MinerCluster::new(ourios_core::config::MinerConfig::default()),
+        TenantRule::service_name(),
+    )
+    .with_rotation_hook(Box::new(move |miner, mark| {
+        *count.lock().expect("lock") += 1;
+        recovery::write_snapshots(&hook_root, miner, Some(mark)).expect("snapshot write");
+    }));
+
+    for body in ["a 1", "b 2", "c 3"] {
+        pipeline
+            .ingest(request(vec![resource_logs("svc", &[body])]))
+            .expect("ingest");
+    }
+    assert_eq!(*fired.lock().expect("lock"), 0, "no rotation, no firing");
+    assert!(
+        snapshot_store::load_all(&snapshots_root)
+            .expect("load")
+            .is_empty(),
+    );
+}

crates/ourios-server/src/receiver.rs

@@ -134,8 +134,22 @@ pub async fn serve(config: ReceiverConfig) -> Result<ReceiverHandle, String> {
     // horizon — an unstamped snapshot is discarded at the next start
     // (RFC 0001 §6.9), which would overwrite the post-recovery
     // artefacts with full-replay-only ones.
+    //
+    // The rotation hook is the §6.9 *primary* cadence point: every WAL
+    // segment rotation persists per-tenant snapshots at the
+    // rotation-point high-water mark. Best-effort, like the other
+    // cadence points — a snapshot is a rebuildable cache.
+    let hook_root = snapshots_root.clone();
     let pipeline: SharedPipeline = Arc::new(Mutex::new(
-        IngestPipeline::new(Box::new(wal), miner, rule).with_last_durable(report.max_delivered),
+        IngestPipeline::new(Box::new(wal), miner, rule)
+            .with_last_durable(report.max_delivered)
+            .with_rotation_hook(Box::new(move |miner, mark| {
+                if let Err(e) = recovery::write_snapshots(&hook_root, miner, Some(mark)) {
+                    eprintln!(
+                        "rotation snapshot write failed (recovery falls back to the WAL): {e}"
+                    );
+                }
+            })),
     ));
 
     // gRPC: bind first so `:0` resolves to a real port before serving.