fix(procedural): harden executor against overflow, runaway loops, and cache collisions

Replace wrapping integer arithmetic with checked_add/sub/mul/div/rem
in the expression evaluator so overflows return None rather than
panicking or silently wrapping.

Replace the unlimited trigger budget (1-hour wall clock, MAX iterations)
with a trigger_default (100k iterations, 10s) to prevent runaway
procedural bodies from pinning Control Plane workers indefinitely.

Guard the plan cache against 64-bit hash collisions by verifying the
cached body SQL matches before returning a cached block, and evicting
on mismatch.

Author: farhan-syah

nodedb/src/control/planner/procedural/executor/core/mod.rs

@@ -100,7 +100,7 @@ impl<'a> StatementExecutor<'a> {
         block: &ProceduralBlock,
         bindings: &RowBindings,
     ) -> crate::Result<()> {
-        let mut budget = ExecutionBudget::unlimited();
+        let mut budget = ExecutionBudget::trigger_default();
         self.execute_block_with_exceptions(
             &block.statements,
             &block.exception_handlers,

nodedb/src/control/planner/procedural/executor/eval.rs

@@ -174,11 +174,11 @@ fn eval_binary_op(
 
     match (l, r) {
         (Value::Integer(a), Value::Integer(b)) => match op {
-            Plus => Some(Value::Integer(a + b)),
-            Minus => Some(Value::Integer(a - b)),
-            Multiply => Some(Value::Integer(a * b)),
-            Divide if *b != 0 => Some(Value::Integer(a / b)),
-            Modulo if *b != 0 => Some(Value::Integer(a % b)),
+            Plus => Some(Value::Integer(a.checked_add(*b)?)),
+            Minus => Some(Value::Integer(a.checked_sub(*b)?)),
+            Multiply => Some(Value::Integer(a.checked_mul(*b)?)),
+            Divide if *b != 0 => Some(Value::Integer(a.checked_div(*b)?)),
+            Modulo if *b != 0 => Some(Value::Integer(a.checked_rem(*b)?)),
             Gt => Some(Value::Bool(a > b)),
             GtEq => Some(Value::Bool(a >= b)),
             Lt => Some(Value::Bool(a < b)),
@@ -191,7 +191,14 @@ fn eval_binary_op(
             Plus => Some(Value::Float(a + b)),
             Minus => Some(Value::Float(a - b)),
             Multiply => Some(Value::Float(a * b)),
-            Divide if *b != 0.0 => Some(Value::Float(a / b)),
+            Divide if *b != 0.0 && b.is_finite() && *b != -0.0 => {
+                let result = a / b;
+                if result.is_finite() {
+                    Some(Value::Float(result))
+                } else {
+                    None
+                }
+            }
             Gt => Some(Value::Bool(a > b)),
             GtEq => Some(Value::Bool(a >= b)),
             Lt => Some(Value::Bool(a < b)),

nodedb/src/control/planner/procedural/executor/fuel.rs

@@ -32,14 +32,10 @@ impl ExecutionBudget {
         }
     }
 
-    /// Create an unlimited budget (for triggers with no explicit limits).
-    pub fn unlimited() -> Self {
-        Self {
-            fuel_remaining: u64::MAX,
-            deadline: Instant::now() + std::time::Duration::from_secs(3600),
-            max_iterations: u64::MAX,
-            timeout_secs: 3600,
-        }
+    /// Default budget for trigger bodies. Caps iterations and wall-clock
+    /// time to prevent runaway loops from pinning Control Plane workers.
+    pub fn trigger_default() -> Self {
+        Self::new(100_000, 10)
     }
 
     /// Consume one iteration of fuel. Returns error if exhausted.
@@ -103,8 +99,8 @@ mod tests {
     }
 
     #[test]
-    fn unlimited() {
-        let mut budget = ExecutionBudget::unlimited();
+    fn trigger_default_allows_bounded_loops() {
+        let mut budget = ExecutionBudget::trigger_default();
         for _ in 0..10_000 {
             budget.consume_iteration().unwrap();
         }

nodedb/src/control/planner/procedural/executor/plan_cache.rs

@@ -27,6 +27,7 @@ struct CacheInner {
 }
 
 struct CacheEntry {
+    body_sql: String,
     block: Arc<ProceduralBlock>,
 }
 
@@ -58,9 +59,16 @@ impl ProcedureBlockCache {
         // is uncontended in the cache-hit path (early return).
         let mut inner = self.inner.lock().unwrap_or_else(|p| p.into_inner());
 
-        // Cache hit — return immediately.
+        // Cache hit — verify the body matches (not just the hash) to guard
+        // against 64-bit hash collisions returning the wrong block.
         if let Some(entry) = inner.entries.get(&key) {
-            return Ok(Arc::clone(&entry.block));
+            if entry.body_sql == body_sql {
+                return Ok(Arc::clone(&entry.block));
+            }
+            // Hash collision with different body — evict the stale entry
+            // and fall through to re-parse.
+            inner.entries.remove(&key);
+            inner.order.retain(|k| *k != key);
         }
 
         // Cache miss — parse, cache, return.
@@ -76,6 +84,7 @@ impl ProcedureBlockCache {
         inner.entries.insert(
             key,
             CacheEntry {
+                body_sql: body_sql.to_string(),
                 block: Arc::clone(&arc),
             },
         );