feat(engine): implement bitemporal temporal-purge across storage engines

Add per-engine purge logic that drops superseded bitemporal row versions
(those with finite system_from_ms) once they fall outside the configured
audit-retain window, while never removing the single surviving current
version of any key.

- Graph edge store (`edge_store/temporal/purge.rs`): scans versioned
  edge rows, purges stale non-current versions in keyed batches.
- Sparse btree-versioned (`btree_versioned/purge.rs`): purges stale
  document-strict and metadata versions from redb versioned tables.
- Columnar (`mutation/engine.rs`): expose `delete_bitmap_mut`,
  `memtable_segment_id`, and `pk_col_indices` accessors used by the
  columnar purge path to tombstone superseded segment rows without
  touching the main write pipeline.
- New `engine/bitemporal/` module: `BitemporalRetentionRegistry`
  (per-collection policy store) and `enforcement` loop that computes
  cutoffs and dispatches `MetaOp::TemporalPurge*` plans.

Author: farhan-syah

nodedb-columnar/src/mutation/engine.rs

@@ -94,6 +94,24 @@ impl MutationEngine {
         self.delete_bitmaps.get(&segment_id)
     }
 
+    /// Mutable access to a segment's delete bitmap. Creates an empty one
+    /// on first access so callers can `mark_deleted_batch` unconditionally.
+    /// Used by temporal-purge paths that tombstone superseded row positions
+    /// without going through the single-row `insert` / `delete` paths.
+    pub fn delete_bitmap_mut(&mut self, segment_id: u32) -> &mut DeleteBitmap {
+        self.delete_bitmaps.entry(segment_id).or_default()
+    }
+
+    /// The virtual segment id used for rows still in the memtable.
+    pub fn memtable_segment_id(&self) -> u32 {
+        self.memtable_segment_id
+    }
+
+    /// The schema's primary-key column indices, in schema order.
+    pub fn pk_col_indices(&self) -> &[usize] {
+        &self.pk_col_indices
+    }
+
     /// Access all delete bitmaps.
     pub fn delete_bitmaps(&self) -> &HashMap<u32, DeleteBitmap> {
         &self.delete_bitmaps

nodedb/src/engine/bitemporal/enforcement.rs

@@ -0,0 +1,238 @@
+//! Bitemporal audit-retention enforcement loop.
+//!
+//! Runs on the Event Plane (Tokio, `Send + Sync`). Every tick:
+//!
+//! 1. Snapshot the [`BitemporalRetentionRegistry`].
+//! 2. For each entry with `audit_retain_ms > 0`, compute
+//!    `cutoff_system_ms = now - audit_retain_ms`.
+//! 3. Dispatch the appropriate `MetaOp::TemporalPurge{EdgeStore,
+//!    DocumentStrict, Columnar}` to the owning Data Plane core via the
+//!    shared `sync_dispatch` async path.
+//! 4. On success, parse the returned purge count from the response
+//!    payload and append a `RecordType::TemporalPurge` record to the
+//!    WAL for durable audit.
+//!
+//! NEVER does storage I/O directly — all physical work happens in the
+//! Data Plane. The only persistent side-effect emitted from this loop
+//! is the WAL audit record.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use tokio::sync::watch;
+use tracing::{info, warn};
+
+use super::registry::{BitemporalEngineKind, BitemporalRetentionRegistry, Entry};
+use crate::bridge::envelope::PhysicalPlan;
+use crate::bridge::physical_plan::MetaOp;
+use crate::control::state::SharedState;
+
+/// Default tick interval when no shorter deadline is needed. One hour
+/// matches the timeseries retention loop's default; operators can lower
+/// by providing a shorter `tick_interval`.
+const DEFAULT_TICK_MS: u64 = 3_600_000;
+
+/// Dispatch deadline per purge op. Purges can scan many segments /
+/// redb ranges; a 30-second bound matches the existing retention path.
+const DISPATCH_DEADLINE_SECS: u64 = 30;
+
+/// Startup delay so the loop doesn't race the Data Plane warm-up.
+const STARTUP_DELAY_SECS: u64 = 10;
+
+/// Spawn the bitemporal-retention enforcement loop as a background
+/// Tokio task. Returns a `JoinHandle` for shutdown coordination.
+pub fn spawn_bitemporal_retention_loop(
+    shared_state: Arc<SharedState>,
+    registry: Arc<BitemporalRetentionRegistry>,
+    shutdown: watch::Receiver<bool>,
+    tick_interval: Option<Duration>,
+) -> tokio::task::JoinHandle<()> {
+    let tick = tick_interval.unwrap_or(Duration::from_millis(DEFAULT_TICK_MS));
+    tokio::spawn(async move {
+        enforcement_loop(shared_state, registry, shutdown, tick).await;
+    })
+}
+
+async fn enforcement_loop(
+    state: Arc<SharedState>,
+    registry: Arc<BitemporalRetentionRegistry>,
+    mut shutdown: watch::Receiver<bool>,
+    tick: Duration,
+) {
+    tokio::time::sleep(Duration::from_secs(STARTUP_DELAY_SECS)).await;
+
+    loop {
+        tokio::select! {
+            _ = tokio::time::sleep(tick) => {}
+            _ = shutdown.changed() => {
+                if *shutdown.borrow() {
+                    info!("bitemporal retention loop shutting down");
+                    return;
+                }
+            }
+        }
+
+        let entries = registry.snapshot();
+        if entries.is_empty() {
+            continue;
+        }
+        for entry in entries {
+            run_one(&state, &entry).await;
+        }
+    }
+}
+
+async fn run_one(state: &Arc<SharedState>, entry: &Entry) {
+    let audit_ms = entry.retention.audit_retain_ms;
+    if audit_ms == 0 {
+        return; // "retain forever" — no purge.
+    }
+    let now_ms = std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .unwrap_or_default()
+        .as_millis() as i64;
+    let cutoff_system_ms = now_ms.saturating_sub(audit_ms as i64);
+
+    let tenant_id = entry.tenant_id;
+    let plan = match entry.engine {
+        BitemporalEngineKind::EdgeStore => PhysicalPlan::Meta(MetaOp::TemporalPurgeEdgeStore {
+            tenant_id: tenant_id.as_u32(),
+            collection: entry.collection.clone(),
+            cutoff_system_ms,
+        }),
+        BitemporalEngineKind::DocumentStrict => {
+            PhysicalPlan::Meta(MetaOp::TemporalPurgeDocumentStrict {
+                tenant_id: tenant_id.as_u32(),
+                collection: entry.collection.clone(),
+                cutoff_system_ms,
+            })
+        }
+        BitemporalEngineKind::Columnar => PhysicalPlan::Meta(MetaOp::TemporalPurgeColumnar {
+            tenant_id: tenant_id.as_u32(),
+            collection: entry.collection.clone(),
+            cutoff_system_ms,
+        }),
+    };
+
+    match crate::control::server::pgwire::ddl::sync_dispatch::dispatch_async(
+        state,
+        tenant_id,
+        &entry.collection,
+        plan,
+        Duration::from_secs(DISPATCH_DEADLINE_SECS),
+    )
+    .await
+    {
+        Ok(payload) => {
+            let purged = parse_count_from_payload(entry.engine, &payload);
+            if purged > 0
+                && let Err(e) = state.wal.append_temporal_purge(
+                    tenant_id,
+                    entry.engine.wire_tag(),
+                    &entry.collection,
+                    cutoff_system_ms,
+                    purged,
+                )
+            {
+                warn!(
+                    tenant = tenant_id.as_u32(),
+                    collection = %entry.collection,
+                    error = %e,
+                    "temporal-purge wal append failed"
+                );
+            }
+            if purged > 0 {
+                info!(
+                    tenant = tenant_id.as_u32(),
+                    collection = %entry.collection,
+                    engine = ?entry.engine,
+                    purged,
+                    cutoff_ms = cutoff_system_ms,
+                    "bitemporal audit-retention purge"
+                );
+            }
+        }
+        Err(e) => {
+            warn!(
+                tenant = tenant_id.as_u32(),
+                collection = %entry.collection,
+                engine = ?entry.engine,
+                error = %e,
+                "bitemporal temporal-purge dispatch failed"
+            );
+        }
+    }
+}
+
+/// Decode the purge count from a response payload.
+///
+/// EdgeStore / Columnar → 8 bytes LE u64 (count).
+/// DocumentStrict → 16 bytes: docs LE u64 ++ index-entries LE u64; we
+///   sum them for the audit-record count so operators see total rows
+///   reclaimed across the two versioned tables.
+fn parse_count_from_payload(engine: BitemporalEngineKind, payload: &[u8]) -> u64 {
+    match engine {
+        BitemporalEngineKind::EdgeStore | BitemporalEngineKind::Columnar => {
+            if payload.len() >= 8 {
+                u64::from_le_bytes(payload[..8].try_into().unwrap_or([0; 8]))
+            } else {
+                0
+            }
+        }
+        BitemporalEngineKind::DocumentStrict => {
+            if payload.len() >= 16 {
+                let docs = u64::from_le_bytes(payload[..8].try_into().unwrap_or([0; 8]));
+                let idx = u64::from_le_bytes(payload[8..16].try_into().unwrap_or([0; 8]));
+                docs.saturating_add(idx)
+            } else {
+                0
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn parse_count_edgestore_8_bytes() {
+        let payload = 42u64.to_le_bytes().to_vec();
+        assert_eq!(
+            parse_count_from_payload(BitemporalEngineKind::EdgeStore, &payload),
+            42
+        );
+    }
+
+    #[test]
+    fn parse_count_columnar_8_bytes() {
+        let payload = 7u64.to_le_bytes().to_vec();
+        assert_eq!(
+            parse_count_from_payload(BitemporalEngineKind::Columnar, &payload),
+            7
+        );
+    }
+
+    #[test]
+    fn parse_count_document_strict_sums_docs_and_idx() {
+        let mut payload = Vec::with_capacity(16);
+        payload.extend_from_slice(&5u64.to_le_bytes());
+        payload.extend_from_slice(&11u64.to_le_bytes());
+        assert_eq!(
+            parse_count_from_payload(BitemporalEngineKind::DocumentStrict, &payload),
+            16
+        );
+    }
+
+    #[test]
+    fn parse_count_truncated_returns_zero() {
+        assert_eq!(
+            parse_count_from_payload(BitemporalEngineKind::EdgeStore, &[1, 2, 3]),
+            0
+        );
+        assert_eq!(
+            parse_count_from_payload(BitemporalEngineKind::DocumentStrict, &[1; 8]),
+            0
+        );
+    }
+}

nodedb/src/engine/bitemporal/mod.rs

@@ -0,0 +1,17 @@
+//! Bitemporal audit-retention: per-collection configuration registry and
+//! background enforcement loop that purges superseded versions older than
+//! each collection's `audit_retain_ms` window, while preserving the single
+//! latest version of every logical row.
+//!
+//! Population of the registry is the DDL layer's responsibility (e.g.
+//! `CREATE COLLECTION ... WITH BITEMPORAL RETENTION (AUDIT_RETAIN = '7d')`).
+//! Enforcement runs on the Event Plane as a Tokio background loop and
+//! dispatches `MetaOp::TemporalPurge{EdgeStore,DocumentStrict,Columnar}`
+//! to the owning Data Plane core. Successful dispatches emit a durable
+//! `RecordType::TemporalPurge` audit record via `WalManager`.
+
+pub mod enforcement;
+pub mod registry;
+
+pub use enforcement::spawn_bitemporal_retention_loop;
+pub use registry::{BitemporalEngineKind, BitemporalRetentionRegistry, RegisterError};