refactor(nodedb-crdt): split validator monolith into focused submodules

The 792-line validator.rs was a growing catch-all. Split into:
- core.rs        — Validator struct, constructors, accessors
- types.rs       — ValidationOutcome, Violation, ProposedChange
- validate.rs    — core validation logic
- policy_dispatch.rs — conflict-policy resolution
- bitemporal.rs  — bitemporal time-window helpers (valid-until sentinels,
                   liveness predicate, window-overlap check)
- tests.rs       — unit tests (moved from inline)

mod.rs re-exports the full public surface unchanged; no behavioral
differences in this commit.

Author: farhan-syah

nodedb-crdt/src/validator.rs

@@ -0,0 +1,75 @@
+//! Bitemporal-aware validation helpers.
+//!
+//! CRDT constraint validation for bitemporal collections must scope to the
+//! "current" version of each row — rows whose `_ts_valid_until` is open
+//! (MAX or unset). A UNIQUE collision between an old (superseded) version
+//! and a new live row is not a violation; two live rows with the same
+//! value IS a violation.
+
+/// Reserved column name for the valid-time upper bound.
+pub const VALID_UNTIL: &str = "_ts_valid_until";
+
+/// Reserved column name for the valid-time lower bound.
+pub const VALID_FROM: &str = "_ts_valid_from";
+
+/// Reserved column name for the system-time stamp.
+pub const SYSTEM_TS: &str = "_ts_system";
+
+/// Sentinel value representing "no upper bound" on valid-time.
+pub const VALID_UNTIL_OPEN: i64 = i64::MAX;
+
+/// Is a row "currently live" at the given system time?
+///
+/// A row is live when `valid_until` is open (MAX) or strictly greater
+/// than the query time. Rows with `valid_until == now` are considered
+/// just-superseded and NOT live.
+pub fn is_live(valid_until: i64, now_ms: i64) -> bool {
+    valid_until == VALID_UNTIL_OPEN || valid_until > now_ms
+}
+
+/// Do two valid-time windows overlap?
+///
+/// Windows are half-open `[from, until)`. Two windows overlap iff each
+/// starts strictly before the other ends.
+pub fn windows_overlap(a_from: i64, a_until: i64, b_from: i64, b_until: i64) -> bool {
+    a_from < b_until && b_from < a_until
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn live_with_open_until() {
+        assert!(is_live(VALID_UNTIL_OPEN, 1_000));
+    }
+
+    #[test]
+    fn superseded_row_not_live() {
+        assert!(!is_live(500, 1_000));
+        assert!(!is_live(1_000, 1_000));
+    }
+
+    #[test]
+    fn future_valid_until_is_live() {
+        assert!(is_live(2_000, 1_000));
+    }
+
+    #[test]
+    fn overlap_disjoint() {
+        assert!(!windows_overlap(0, 100, 100, 200));
+        assert!(!windows_overlap(200, 300, 0, 100));
+    }
+
+    #[test]
+    fn overlap_shared_interval() {
+        assert!(windows_overlap(0, 150, 100, 200));
+        assert!(windows_overlap(100, 200, 0, 150));
+    }
+
+    #[test]
+    fn overlap_contained() {
+        assert!(windows_overlap(0, 1000, 100, 200));
+        assert!(windows_overlap(100, 200, 0, 1000));
+    }
+}

nodedb-crdt/src/validator/bitemporal.rs

@@ -0,0 +1,115 @@
+//! Validator struct, constructors, and accessors.
+
+use std::collections::{HashMap, HashSet};
+
+use crate::constraint::ConstraintSet;
+use crate::dead_letter::DeadLetterQueue;
+use crate::deferred::DeferredQueue;
+use crate::policy::PolicyRegistry;
+use crate::signing::DeltaSigner;
+
+/// The constraint validator.
+///
+/// Validates proposed changes against a set of constraints and the current
+/// committed state. Violations are resolved via declarative policies:
+/// - AUTO_RESOLVED — policy handles it (e.g., LAST_WRITER_WINS)
+/// - DEFERRED — queued for exponential backoff retry (CASCADE_DEFER)
+/// - WEBHOOK_REQUIRED — caller must POST to webhook for decision
+/// - ESCALATE — route to dead-letter queue (fallback)
+pub struct Validator {
+    pub(super) constraints: ConstraintSet,
+    pub(super) dlq: DeadLetterQueue,
+    pub(super) policies: PolicyRegistry,
+    pub(super) deferred: DeferredQueue,
+    /// Monotonic suffix counter: (collection, field) -> next suffix number
+    pub(super) suffix_counter: HashMap<(String, String), u64>,
+    /// Optional delta signature verifier. When set, signed deltas are
+    /// verified before constraint validation.
+    pub(super) delta_verifier: Option<DeltaSigner>,
+    /// Collections known to be bitemporal. UNIQUE checks for rows in these
+    /// collections scope to currently-live rows (open `_ts_valid_until`)
+    /// so superseded versions don't falsely collide with live writes.
+    pub(super) bitemporal_collections: HashSet<String>,
+}
+
+impl Validator {
+    /// Create a new validator with default (ephemeral) policies.
+    pub fn new(constraints: ConstraintSet, dlq_capacity: usize) -> Self {
+        Self::new_with_policies(constraints, dlq_capacity, PolicyRegistry::new(), 1000)
+    }
+
+    /// Create a new validator with custom policies and deferred queue.
+    pub fn new_with_policies(
+        constraints: ConstraintSet,
+        dlq_capacity: usize,
+        policies: PolicyRegistry,
+        deferred_capacity: usize,
+    ) -> Self {
+        Self {
+            constraints,
+            dlq: DeadLetterQueue::new(dlq_capacity),
+            policies,
+            deferred: DeferredQueue::new(deferred_capacity),
+            suffix_counter: HashMap::new(),
+            delta_verifier: None,
+            bitemporal_collections: HashSet::new(),
+        }
+    }
+
+    /// Register a collection as bitemporal. UNIQUE constraints for rows in
+    /// this collection will scope to currently-live rows only.
+    pub fn mark_bitemporal(&mut self, collection: impl Into<String>) {
+        self.bitemporal_collections.insert(collection.into());
+    }
+
+    /// Is the given collection registered as bitemporal?
+    pub fn is_bitemporal(&self, collection: &str) -> bool {
+        self.bitemporal_collections.contains(collection)
+    }
+
+    /// Access the dead-letter queue.
+    pub fn dlq(&self) -> &DeadLetterQueue {
+        &self.dlq
+    }
+
+    /// Mutable access to the DLQ (for dequeue/retry).
+    pub fn dlq_mut(&mut self) -> &mut DeadLetterQueue {
+        &mut self.dlq
+    }
+
+    /// Access the policy registry.
+    pub fn policies(&self) -> &PolicyRegistry {
+        &self.policies
+    }
+
+    /// Mutable access to the policy registry.
+    pub fn policies_mut(&mut self) -> &mut PolicyRegistry {
+        &mut self.policies
+    }
+
+    /// Access the deferred queue.
+    pub fn deferred(&self) -> &DeferredQueue {
+        &self.deferred
+    }
+
+    /// Mutable access to the deferred queue.
+    pub fn deferred_mut(&mut self) -> &mut DeferredQueue {
+        &mut self.deferred
+    }
+
+    /// Set the delta signature verifier. When set, deltas with non-zero
+    /// signatures in their CrdtAuthContext will be verified before validation.
+    pub fn set_delta_verifier(&mut self, verifier: DeltaSigner) {
+        self.delta_verifier = Some(verifier);
+    }
+
+    /// Access the delta verifier.
+    pub fn delta_verifier(&self) -> Option<&DeltaSigner> {
+        self.delta_verifier.as_ref()
+    }
+
+    /// Mutable access to the delta verifier.
+    pub fn delta_verifier_mut(&mut self) -> Option<&mut DeltaSigner> {
+        self.delta_verifier.as_mut()
+    }
+}

nodedb-crdt/src/validator/core.rs

@@ -0,0 +1,18 @@
+//! Constraint validator — checks deltas against committed state.
+//!
+//! This is the core of the CRDT/SQL bridge. When a delta arrives from a peer,
+//! the validator checks all applicable constraints against the leader's
+//! committed state. If any constraint is violated, the delta is rejected
+//! with a compensation hint.
+
+pub mod bitemporal;
+mod core;
+mod policy_dispatch;
+mod types;
+mod validate;
+
+#[cfg(test)]
+mod tests;
+
+pub use core::Validator;
+pub use types::{ProposedChange, ValidationOutcome, Violation};