Merge remote-tracking branch 'origin/security' into security

# Conflicts:
#	nodedb-sql/src/parser/preprocess/search_vector.rs

Author: farhan-syah

nodedb-crdt/src/constraint_checks.rs

@@ -28,23 +28,29 @@ fn render_value(value: &LoroValue) -> String {
 }
 
 impl Validator {
+    ///
+    /// Returns `Err(EvalError)` when a CHECK predicate cannot be *evaluated*
+    /// (division/modulo by zero). This is distinct from an
+    /// ordinary constraint failure (`Ok(Some(Violation))`): the caller lifts it
+    /// to [`ValidationOutcome::EvalError`], a hard statement failure that
+    /// declarative conflict policies must never "resolve".
     pub(crate) fn check_constraint(
         &self,
         state: &impl RowLookup,
         change: &ProposedChange,
         constraint: &Constraint,
-    ) -> Option<Violation> {
+    ) -> Result<Option<Violation>, nodedb_query::EvalError> {
         match &constraint.kind {
-            ConstraintKind::Unique => self.check_unique(state, change, constraint),
+            ConstraintKind::Unique => Ok(self.check_unique(state, change, constraint)),
             ConstraintKind::ForeignKey {
                 ref_collection,
                 ref_key,
             }
             | ConstraintKind::BiTemporalFK {
                 ref_collection,
                 ref_key,
-            } => self.check_foreign_key(state, change, constraint, ref_collection, ref_key),
-            ConstraintKind::NotNull => self.check_not_null(change, constraint),
+            } => Ok(self.check_foreign_key(state, change, constraint, ref_collection, ref_key)),
+            ConstraintKind::NotNull => Ok(self.check_not_null(change, constraint)),
             ConstraintKind::Check { expr, .. } => self.check_expr(change, constraint, expr),
         }
     }
@@ -60,7 +66,7 @@ impl Validator {
         change: &ProposedChange,
         constraint: &Constraint,
         expr: &str,
-    ) -> Option<Violation> {
+    ) -> Result<Option<Violation>, nodedb_query::EvalError> {
         // Build a row Value::Object from the proposed change's fields so the
         // expression evaluator can resolve column references by name.
         let mut row = std::collections::HashMap::with_capacity(change.fields.len());
@@ -72,7 +78,10 @@ impl Validator {
         let parsed = match nodedb_query::expr_parse::parse_generated_expr(expr) {
             Ok((expr, _deps)) => expr,
             Err(e) => {
-                return Some(Violation {
+                // A malformed *stored* predicate is a catalog-integrity failure,
+                // not an evaluation error — it stays an ordinary Violation
+                // (fails closed) so a corrupt entry can't bypass its invariant.
+                return Ok(Some(Violation {
                     constraint_name: constraint.name.clone(),
                     reason: format!("invalid CHECK expression `{expr}`: {e}"),
                     hint: CompensationHint::ManualIntervention {
@@ -81,20 +90,27 @@ impl Validator {
                             constraint.name
                         ),
                     },
-                });
+                }));
             }
         };
 
+        // A division/modulo-by-zero inside the predicate
+        // is neither a PASS nor an ordinary FAIL — it's an evaluation error,
+        // propagated as `Err(EvalError)` and lifted by `validate` to
+        // `ValidationOutcome::EvalError`. That keeps it out of the conflict-
+        // policy machinery: an unevaluable predicate is a hard SQLSTATE-22012
+        // failure, never something LastWriterWins et al. may "resolve".
         match parsed.eval(&row_value) {
-            nodedb_types::Value::Bool(true) => None,
-            nodedb_types::Value::Null => None,
-            _ => Some(Violation {
+            Ok(nodedb_types::Value::Bool(true)) => Ok(None),
+            Ok(nodedb_types::Value::Null) => Ok(None),
+            Ok(_) => Ok(Some(Violation {
                 constraint_name: constraint.name.clone(),
                 reason: format!("CHECK `{}` failed: {expr}", constraint.name),
                 hint: CompensationHint::ManualIntervention {
                     reason: format!("row violates CHECK predicate `{expr}`"),
                 },
-            }),
+            })),
+            Err(e) => Err(e),
         }
     }
 

nodedb-crdt/src/pre_validate.rs

@@ -45,6 +45,16 @@ pub fn pre_validate(
                 reason: v.reason.clone(),
             }
         }
+        // An unevaluable predicate (division/modulo by zero)
+        // is rejected before it ever reaches Raft — same terminal outcome as a
+        // violation, so the delta never commits.
+        ValidationOutcome::EvalError {
+            constraint_name,
+            error,
+        } => PreValidationResult::FastReject {
+            constraint: constraint_name,
+            reason: format!("CHECK predicate failed to evaluate: {error}"),
+        },
     }
 }
 

nodedb-crdt/src/validator/policy_dispatch.rs

@@ -4,12 +4,13 @@
 
 use crate::CrdtAuthContext;
 use crate::constraint::{Constraint, ConstraintKind};
+use crate::dead_letter::CompensationHint;
 use crate::error::Result;
 use crate::policy::{ConflictPolicy, PolicyResolution, ResolvedAction};
 use crate::row_lookup::RowLookup;
 
 use super::core::Validator;
-use super::types::{ProposedChange, ValidationOutcome};
+use super::types::{ProposedChange, ValidationOutcome, Violation};
 
 impl Validator {
     /// Validate with declarative policy resolution.
@@ -195,6 +196,56 @@ impl Validator {
                     }
                 }
             }
+            ValidationOutcome::EvalError {
+                constraint_name,
+                error,
+            } => {
+                // An unevaluable predicate (division/modulo by zero)
+                // is NOT a resolvable conflict — it must never be
+                // handed to a declarative policy, which would "resolve" a delta
+                // the server cannot actually evaluate. Escalate straight to the
+                // DLQ (fails closed) regardless of the collection's configured
+                // policy, mirroring `EscalateToDlq` but for an eval error.
+                let constraint = self
+                    .constraints
+                    .all()
+                    .iter()
+                    .find(|c| c.name == constraint_name)
+                    .cloned()
+                    .unwrap_or_else(|| Constraint {
+                        name: constraint_name.clone(),
+                        collection: change.collection.clone(),
+                        field: String::new(),
+                        kind: ConstraintKind::NotNull,
+                    });
+                let reason = format!("CHECK `{constraint_name}` failed to evaluate: {error}");
+                let hint = CompensationHint::ManualIntervention {
+                    reason: reason.clone(),
+                };
+                self.dlq
+                    .enqueue(crate::dead_letter::EnqueueDeadLetterArgs {
+                        peer_id,
+                        user_id: auth.user_id,
+                        tenant_id: auth.tenant_id,
+                        delta: delta_bytes,
+                        constraint: &constraint,
+                        reason: reason.clone(),
+                        hint: hint.clone(),
+                    })?;
+                tracing::warn!(
+                    constraint = %constraint_name,
+                    collection = %change.collection,
+                    %error,
+                    "CRDT CHECK predicate raised an evaluation error; escalated to DLQ"
+                );
+                Ok(PolicyResolution::Escalate {
+                    violations: vec![Violation {
+                        constraint_name,
+                        reason,
+                        hint,
+                    }],
+                })
+            }
         }
     }
 }

nodedb-crdt/src/validator/tests.rs

@@ -415,3 +415,41 @@ fn check_constraint_rejects_malformed_expr() {
         _ => panic!("expected rejection for malformed CHECK expression"),
     }
 }
+
+/// A CHECK predicate that divides by zero is an *evaluation* error, not an
+/// ordinary violation: `validate` returns `ValidationOutcome::EvalError` so the
+/// downstream conflict-policy machinery never "resolves" an unevaluable
+/// predicate. Distinct from both `Rejected` (a genuine violation) and
+/// `Accepted`.
+#[test]
+fn check_constraint_division_by_zero_surfaces_eval_error() {
+    let state = CrdtState::new(1).unwrap();
+    let mut cs = ConstraintSet::new();
+    // `100 / age` is unevaluable when age = 0.
+    cs.add_check(
+        "people_ratio_check",
+        "people",
+        "age",
+        "100 / age > 0",
+        "100 / age > 0",
+    );
+    let validator = Validator::new(cs, 100);
+
+    let change = ProposedChange {
+        collection: "people".into(),
+        row_id: "p1".into(),
+        surrogate: nodedb_types::Surrogate::ZERO,
+        fields: vec![("age".into(), LoroValue::I64(0))],
+    };
+
+    match validator.validate(&state, &change) {
+        ValidationOutcome::EvalError {
+            constraint_name,
+            error,
+        } => {
+            assert_eq!(constraint_name, "people_ratio_check");
+            assert_eq!(error, nodedb_query::EvalError::DivisionByZero);
+        }
+        other => panic!("expected ValidationOutcome::EvalError, got {other:?}"),
+    }
+}

nodedb-crdt/src/validator/types.rs

@@ -13,6 +13,19 @@ pub enum ValidationOutcome {
     Accepted,
     /// One or more constraints violated — delta rejected.
     Rejected(Vec<Violation>),
+    /// A CHECK predicate could not be *evaluated* — division/modulo by zero.
+    /// Kept distinct from [`ValidationOutcome::Rejected`]
+    /// because an eval error is a hard statement failure (SQLSTATE `22012`),
+    /// NOT a resolvable conflict: declarative conflict policies
+    /// (`LastWriterWins`, `RenameSuffix`, …) must never "resolve" an
+    /// unevaluable predicate the way they resolve a genuine violation. It fails
+    /// closed — escalated straight to the dead-letter queue.
+    EvalError {
+        /// The CHECK constraint whose predicate failed to evaluate.
+        constraint_name: String,
+        /// The underlying evaluation error (currently only `DivisionByZero`).
+        error: nodedb_query::EvalError,
+    },
 }
 
 /// A single constraint violation.

nodedb-crdt/src/validator/validate.rs

@@ -83,8 +83,19 @@ impl Validator {
         let mut violations = Vec::new();
 
         for constraint in constraints {
-            if let Some(violation) = self.check_constraint(state, change, constraint) {
-                violations.push(violation);
+            match self.check_constraint(state, change, constraint) {
+                Ok(Some(violation)) => violations.push(violation),
+                Ok(None) => {}
+                // A predicate that could not be evaluated (division/modulo by
+                // zero) is a hard failure, not a violation to
+                // be batched with the others — surface it immediately so it
+                // bypasses conflict-policy resolution downstream.
+                Err(error) => {
+                    return ValidationOutcome::EvalError {
+                        constraint_name: constraint.name.clone(),
+                        error,
+                    };
+                }
             }
         }
 

nodedb-physical/src/physical_plan/mod.rs

@@ -324,6 +324,8 @@ impl PhysicalPlan {
             | PhysicalPlan::Graph(GraphOp::WccSuperstep(_)) => None,
             // Exchange: recurse into the child plan to extract the collection.
             PhysicalPlan::Query(QueryOp::Exchange(op)) => op.child.collection(),
+            // PostProcess: recurse into the materialized input plan.
+            PhysicalPlan::Query(QueryOp::PostProcess { input, .. }) => input.collection(),
             // ProviderScan is a catalog/constant source — no user collection.
             PhysicalPlan::Query(QueryOp::ProviderScan { .. }) => None,
             // KV ops carry their own collection (sorted-index-only ops → None).

nodedb-physical/src/physical_plan/query.rs

@@ -92,6 +92,46 @@ pub enum QueryOp {
         distinct: bool,
     },
 
+    /// Relational post-processing over a materialized child plan.
+    ///
+    /// Coordinator-resolved, exactly like [`QueryOp::Exchange`]: at resolve time
+    /// the coordinator gathers `input`'s rows, flattens them to the relational
+    /// row shape, and rewrites this node into a [`QueryOp::ProviderScan`]
+    /// carrying the same relational parameters — which applies
+    /// filter→offset→sort→distinct→project→limit on a single core. A Data-Plane
+    /// core must NEVER see this node (defensive error if it does).
+    ///
+    /// Lowered from `SqlPlan::Subquery`: an outer `ORDER BY` / `OFFSET` /
+    /// `DISTINCT` / post-reorder `LIMIT` over a subquery/derived-table body
+    /// whose leaf plan could not absorb those constraints. Constraints the leaf
+    /// DID absorb (a `WHERE` pushed into a search engine, an unordered `LIMIT`
+    /// folded into `top_k`) are applied by `input` and not repeated here.
+    PostProcess {
+        /// The subquery body to materialize. Wrapped in `Exchange{Gather}` by
+        /// the converter when the body is a sharded source, so the gather runs
+        /// exactly once over the full union before post-processing.
+        input: Box<crate::physical_plan::PhysicalPlan>,
+        /// Serialized `Vec<ScanFilter>` (MessagePack). Empty = no predicate.
+        #[serde(default)]
+        filters: Vec<u8>,
+        /// Output column names to keep. Empty = emit all columns.
+        #[serde(default)]
+        projection: Vec<String>,
+        /// Sort keys: `(column_name, ascending)`. Empty = unordered.
+        #[serde(default)]
+        sort_keys: Vec<(String, bool)>,
+        /// Maximum rows to emit after offset/sort/distinct/projection.
+        /// `None` = unlimited.
+        #[serde(default)]
+        limit: Option<usize>,
+        /// Number of rows to skip before applying limit.
+        #[serde(default)]
+        offset: usize,
+        /// SQL DISTINCT: deduplicate on the projected row.
+        #[serde(default)]
+        distinct: bool,
+    },
+
     /// Aggregate: GROUP BY + aggregate functions.
     Aggregate {
         collection: String,

nodedb-physical/src/physical_plan/routing.rs

@@ -75,6 +75,13 @@ pub fn plan_contains_cluster_partitioned_leaf(plan: &PhysicalPlan) -> bool {
             plan_contains_cluster_partitioned_leaf(child)
         }
 
+        // Recurse through PostProcess (its materialized input is the real
+        // plan). Normally resolved to a `ProviderScan` before routing, but a
+        // conservative recursion keeps routing correct if one survives.
+        PhysicalPlan::Query(QueryOp::PostProcess { input, .. }) => {
+            plan_contains_cluster_partitioned_leaf(input)
+        }
+
         // Recurse through lateral outer plans.
         PhysicalPlan::Query(QueryOp::LateralTopK { outer_plan, .. })
         | PhysicalPlan::Query(QueryOp::LateralLoop { outer_plan, .. }) => {