fix(mem): saturate memory release on underflow instead of panicking

BudgetGovernor::release() previously used fetch_sub which panics in
debug builds (and wraps in release) when releasing more bytes than are
currently allocated. This can occur legitimately during WAL replay,
where samples are ingested without a prior reserve() call. Replace
with a CAS loop that saturates to zero and emits a tracing::warn so
the drift is observable without crashing the server.

Also modernize if-let chains, Option combinators, and minor idioms
across catalog, interpolate, window, eval, sql_plan_convert,
collection_insert, prepared parser, kv hash_table, kv sorted_index,
timeseries grouped scan, and trigger retry to use Rust 2024 let-chain
syntax and is_none_or/is_some_and where appropriate.

Author: farhan-syah

nodedb-lite/src/query/catalog.rs

@@ -35,62 +35,62 @@ impl<S: StorageEngine> LiteCatalog<S> {
 impl<S: StorageEngine> SqlCatalog for LiteCatalog<S> {
     fn get_collection(&self, name: &str) -> Option<CollectionInfo> {
         // Check strict collections first.
-        if let Ok(strict) = self.strict.lock() {
-            if let Some(schema) = strict.schema(name) {
-                let columns = schema
-                    .columns
-                    .iter()
-                    .map(|c| ColumnInfo {
-                        name: c.name.clone(),
-                        data_type: convert_column_type(&c.column_type),
-                        nullable: c.nullable,
-                        is_primary_key: c.primary_key,
-                    })
-                    .collect();
-                let pk = schema
-                    .columns
-                    .iter()
-                    .find(|c| c.primary_key)
-                    .map(|c| c.name.clone());
-                return Some(CollectionInfo {
-                    name: name.into(),
-                    engine: EngineType::DocumentStrict,
-                    columns,
-                    primary_key: pk,
-                    has_auto_tier: false,
-                });
-            }
+        if let Ok(strict) = self.strict.lock()
+            && let Some(schema) = strict.schema(name)
+        {
+            let columns = schema
+                .columns
+                .iter()
+                .map(|c| ColumnInfo {
+                    name: c.name.clone(),
+                    data_type: convert_column_type(&c.column_type),
+                    nullable: c.nullable,
+                    is_primary_key: c.primary_key,
+                })
+                .collect();
+            let pk = schema
+                .columns
+                .iter()
+                .find(|c| c.primary_key)
+                .map(|c| c.name.clone());
+            return Some(CollectionInfo {
+                name: name.into(),
+                engine: EngineType::DocumentStrict,
+                columns,
+                primary_key: pk,
+                has_auto_tier: false,
+            });
         }
 
         // Check columnar collections.
-        if let Ok(columnar) = self.columnar.lock() {
-            if columnar.schema(name).is_some() {
-                return Some(CollectionInfo {
-                    name: name.into(),
-                    engine: EngineType::Columnar,
-                    columns: Vec::new(),
-                    primary_key: None,
-                    has_auto_tier: false,
-                });
-            }
+        if let Ok(columnar) = self.columnar.lock()
+            && columnar.schema(name).is_some()
+        {
+            return Some(CollectionInfo {
+                name: name.into(),
+                engine: EngineType::Columnar,
+                columns: Vec::new(),
+                primary_key: None,
+                has_auto_tier: false,
+            });
         }
 
         // Check CRDT (schemaless) collections.
-        if let Ok(crdt) = self.crdt.lock() {
-            if crdt.collection_names().iter().any(|n| n == name) {
-                return Some(CollectionInfo {
-                    name: name.into(),
-                    engine: EngineType::DocumentSchemaless,
-                    columns: vec![ColumnInfo {
-                        name: "id".into(),
-                        data_type: SqlDataType::String,
-                        nullable: false,
-                        is_primary_key: true,
-                    }],
-                    primary_key: Some("id".into()),
-                    has_auto_tier: false,
-                });
-            }
+        if let Ok(crdt) = self.crdt.lock()
+            && crdt.collection_names().iter().any(|n| n == name)
+        {
+            return Some(CollectionInfo {
+                name: name.into(),
+                engine: EngineType::DocumentSchemaless,
+                columns: vec![ColumnInfo {
+                    name: "id".into(),
+                    data_type: SqlDataType::String,
+                    nullable: false,
+                    is_primary_key: true,
+                }],
+                primary_key: Some("id".into()),
+                has_auto_tier: false,
+            });
         }
 
         None

nodedb-mem/src/budget.rs

@@ -76,12 +76,32 @@ impl Budget {
     }
 
     /// Release `size` bytes back to the budget.
+    ///
+    /// Saturates to zero if `size` exceeds the current allocation (which can
+    /// happen when data is replayed from WAL without a matching reservation).
     pub fn release(&self, size: usize) {
-        let prev = self.allocated.fetch_sub(size, Ordering::Release);
-        debug_assert!(
-            prev >= size,
-            "BUG: released more memory than allocated ({size} > {prev})"
-        );
+        loop {
+            let current = self.allocated.load(Ordering::Acquire);
+            let new_val = current.saturating_sub(size);
+            match self.allocated.compare_exchange_weak(
+                current,
+                new_val,
+                Ordering::Release,
+                Ordering::Relaxed,
+            ) {
+                Ok(_) => {
+                    if size > current {
+                        tracing::warn!(
+                            released = size,
+                            allocated = current,
+                            "memory release exceeds allocation (WAL replay or accounting drift)"
+                        );
+                    }
+                    return;
+                }
+                Err(_) => continue,
+            }
+        }
     }
 
     /// Current allocated bytes.

nodedb-query/src/ts_functions/interpolate.rs

@@ -110,19 +110,19 @@ fn fill_linear(values: &[Option<f64>], timestamps_ns: &[i64]) -> Vec<f64> {
     }
 
     // Edge: fill leading NULLs with first known value.
-    if let Some(first) = values.iter().position(|v| v.is_some()) {
-        if let Some(fv) = values[first] {
-            for r in result.iter_mut().take(first) {
-                *r = fv;
-            }
+    if let Some(first) = values.iter().position(|v| v.is_some())
+        && let Some(fv) = values[first]
+    {
+        for r in result.iter_mut().take(first) {
+            *r = fv;
         }
     }
     // Edge: fill trailing NULLs with last known value.
-    if let Some(last) = values.iter().rposition(|v| v.is_some()) {
-        if let Some(lv) = values[last] {
-            for r in result.iter_mut().skip(last + 1) {
-                *r = lv;
-            }
+    if let Some(last) = values.iter().rposition(|v| v.is_some())
+        && let Some(lv) = values[last]
+    {
+        for r in result.iter_mut().skip(last + 1) {
+            *r = lv;
         }
     }
 

nodedb-sql/src/planner/window.rs

@@ -20,10 +20,10 @@ pub fn extract_window_functions(
             ast::SelectItem::ExprWithAlias { expr, alias } => (expr, normalize_ident(alias)),
             _ => continue,
         };
-        if let ast::Expr::Function(func) = expr {
-            if func.over.is_some() {
-                specs.push(convert_window_spec(func, &alias)?);
-            }
+        if let ast::Expr::Function(func) = expr
+            && func.over.is_some()
+        {
+            specs.push(convert_window_spec(func, &alias)?);
         }
     }
     Ok(specs)

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

@@ -108,12 +108,11 @@ fn try_eval_constant(sql: &str) -> Option<nodedb_types::Value> {
     let stmts = sqlparser::parser::Parser::parse_sql(&dialect, &select_sql).ok()?;
     let stmt = stmts.first()?;
 
-    if let sqlparser::ast::Statement::Query(query) = stmt {
-        if let sqlparser::ast::SetExpr::Select(select) = &*query.body {
-            if let Some(sqlparser::ast::SelectItem::UnnamedExpr(expr)) = select.projection.first() {
-                return eval_expr(expr);
-            }
-        }
+    if let sqlparser::ast::Statement::Query(query) = stmt
+        && let sqlparser::ast::SetExpr::Select(select) = &*query.body
+        && let Some(sqlparser::ast::SelectItem::UnnamedExpr(expr)) = select.projection.first()
+    {
+        return eval_expr(expr);
     }
     None
 }

nodedb/src/control/planner/sql_plan_convert.rs

@@ -227,22 +227,20 @@ fn convert_one(
                 .collect();
 
             // AUTO_TIER: split query across retention tiers if enabled.
-            if *tiered {
-                if let Some(registry) = &ctx.retention_registry {
-                    if let Some(policy) = registry.get(tenant_id.as_u32(), collection) {
-                        if policy.auto_tier {
-                            return Ok(super::auto_tier::plan_tiered_scan(
-                                &policy,
-                                tenant_id,
-                                *time_range,
-                                filter_bytes,
-                                group_by.clone(),
-                                agg_pairs,
-                                gap_fill.clone(),
-                            ));
-                        }
-                    }
-                }
+            if *tiered
+                && let Some(registry) = &ctx.retention_registry
+                && let Some(policy) = registry.get(tenant_id.as_u32(), collection)
+                && policy.auto_tier
+            {
+                return Ok(super::auto_tier::plan_tiered_scan(
+                    &policy,
+                    tenant_id,
+                    *time_range,
+                    filter_bytes,
+                    group_by.clone(),
+                    agg_pairs,
+                    gap_fill.clone(),
+                ));
             }
 
             let proj_names = extract_projection_names(projection);

nodedb/src/control/server/pgwire/ddl/collection_insert.rs

@@ -306,28 +306,28 @@ pub async fn insert_document(
 
     // Track field names in catalog so DataFusion can resolve them in queries.
     // This makes schemaless fields visible for WHERE, GROUP BY, ORDER BY, etc.
-    if let Some(catalog) = state.credentials.catalog() {
-        if let Ok(Some(mut coll)) = catalog.get_collection(tenant_id.as_u32(), &parsed.coll_name) {
-            let mut changed = false;
-            for (name, val) in &fields {
-                if name == "id" {
-                    continue;
-                }
-                if !coll.fields.iter().any(|(n, _)| n == name) {
-                    let type_str = match val {
-                        serde_json::Value::Number(n) if n.is_f64() => "FLOAT",
-                        serde_json::Value::Number(_) => "INT",
-                        serde_json::Value::Bool(_) => "BOOL",
-                        _ => "TEXT",
-                    };
-                    coll.fields.push((name.clone(), type_str.to_string()));
-                    changed = true;
-                }
+    if let Some(catalog) = state.credentials.catalog()
+        && let Ok(Some(mut coll)) = catalog.get_collection(tenant_id.as_u32(), &parsed.coll_name)
+    {
+        let mut changed = false;
+        for (name, val) in &fields {
+            if name == "id" {
+                continue;
             }
-            if changed {
-                let _ = catalog.put_collection(&coll);
+            if !coll.fields.iter().any(|(n, _)| n == name) {
+                let type_str = match val {
+                    serde_json::Value::Number(n) if n.is_f64() => "FLOAT",
+                    serde_json::Value::Number(_) => "INT",
+                    serde_json::Value::Bool(_) => "BOOL",
+                    _ => "TEXT",
+                };
+                coll.fields.push((name.clone(), type_str.to_string()));
+                changed = true;
             }
         }
+        if changed {
+            let _ = catalog.put_collection(&coll);
+        }
     }
 
     // Fire SYNC AFTER INSERT triggers (execute in write path, same transaction).

nodedb/src/control/server/pgwire/handler/prepared/parser.rs

@@ -110,10 +110,10 @@ impl QueryParser for NodeDbQueryParser {
 fn count_placeholders(sql: &str) -> usize {
     let mut max_idx = 0usize;
     for part in sql.split('$') {
-        if let Some(num_str) = part.split(|c: char| !c.is_ascii_digit()).next() {
-            if let Ok(idx) = num_str.parse::<usize>() {
-                max_idx = max_idx.max(idx);
-            }
+        if let Some(num_str) = part.split(|c: char| !c.is_ascii_digit()).next()
+            && let Ok(idx) = num_str.parse::<usize>()
+        {
+            max_idx = max_idx.max(idx);
         }
     }
     max_idx

nodedb/src/engine/kv/hash_table.rs

@@ -209,9 +209,7 @@ impl KvHashTable {
         if let Some(idx) = Self::probe_find_index_static(&self.slots, h, key) {
             // probe_find_index_static guarantees slots[idx] is Some.
             let old_value = {
-                let Some(slot) = self.slots[idx].as_ref() else {
-                    return None;
-                };
+                let slot = self.slots[idx].as_ref()?;
                 let v = extract_value_from(&slot.value, &self.overflow);
                 free_value_from(&slot.value, &mut self.overflow);
                 v
@@ -228,9 +226,7 @@ impl KvHashTable {
         if let Some(old_slots) = self.rehash_source.as_mut()
             && let Some(idx) = Self::probe_find_index_static(old_slots, h, key)
         {
-            let Some(old_entry) = old_slots[idx].take() else {
-                return None;
-            };
+            let old_entry = old_slots[idx].take()?;
             let old_value = extract_value_from(&old_entry.value, &self.overflow);
             free_value_from(&old_entry.value, &mut self.overflow);
             let new_kv_value = store_value_in(&mut self.overflow, value, self.inline_threshold);

nodedb/src/engine/kv/sorted_index/tree.rs

@@ -433,16 +433,16 @@ impl OrderStatTree {
 
         // Pruning: left subtree is all < current node.
         // Visit left if current > min (left might contain values in [min, current)).
-        let go_left = min.map_or(true, |m| key > m);
+        let go_left = min.is_none_or(|m| key > m);
         // Pruning: right subtree is all > current node.
         // Visit right if current < max (right might contain values in (current, max]).
-        let go_right = max.map_or(true, |m| key < m);
+        let go_right = max.is_none_or(|m| key < m);
 
         if go_left {
             self.range_collect(n.left, min, max, result);
         }
 
-        let in_range = min.map_or(true, |m| key >= m) && max.map_or(true, |m| key <= m);
+        let in_range = min.is_none_or(|m| key >= m) && max.is_none_or(|m| key <= m);
         if in_range {
             result.push((&n.sort_key, &n.primary_key));
         }