feat(engine): add memory-pressure backpressure guards and wire durable outbound queues

Write-path changes across document, vector, batch, and query layers:

- Reject writes with `LiteError::Backpressure` when the memory governor
  reports `PressureLevel::Critical` in `document_put_impl`,
  `vector_insert_impl`, `document_put_with_vector_batch_impl`, and the
  batch vector ingest path.
- Wire all outbound enqueue calls through `reconcile_outbound_enqueue`
  so backpressure errors are uniformly logged and propagated.
- FTS and spatial enqueue calls switched from direct durable enqueue to
  the new staging path (`stage_index`, `stage_delete`, `stage_insert`);
  the flush task spills staged entries to durable storage periodically.
- `NodeDbLite::open_inner` wires `StreamSeqTracker`, loads the Lite
  identity (`lite_id` + `epoch`) for the handshake, opens all outbound
  queues with the configured cap, and wires FTS/spatial outbound queues
  into the query engine for SQL-path writes.
- `flush()` now drains the KV write buffer first and spills FTS/spatial
  staging buffers to durable queues on each flush tick.
- Add `LiteError::Backpressure` variant; fix typo in SPDX identifiers.

Author: farhan-syah

nodedb-lite/src/error.rs

@@ -44,6 +44,10 @@ pub enum LiteError {
     /// failure that cannot be recovered automatically (OPFS has no rename).
     #[error("OPFS worker bridge failed: {detail}")]
     WorkerFailed { detail: String },
+
+    /// An error during key derivation, salt I/O, or encryption setup.
+    #[error("encryption error: {detail}")]
+    Encryption { detail: String },
 }
 
 impl From<nodedb_types::columnar::SchemaError> for LiteError {
@@ -80,4 +84,26 @@ mod tests {
         let ndb: nodedb_types::error::NodeDbError = e.into();
         assert!(ndb.to_string().contains("test"));
     }
+
+    #[test]
+    fn lite_error_encryption_display_and_convert() {
+        let e = LiteError::Encryption {
+            detail: "argon2 key derivation failed".into(),
+        };
+        let rendered = e.to_string();
+        assert!(rendered.contains("encryption error"));
+        assert!(rendered.contains("argon2 key derivation failed"));
+
+        let ndb: nodedb_types::error::NodeDbError = e.into();
+        assert!(ndb.to_string().contains("argon2 key derivation failed"));
+    }
+
+    #[test]
+    fn lite_error_backpressure_display() {
+        let e = LiteError::Backpressure {
+            detail: "outbound queue full".into(),
+        };
+        assert!(e.to_string().contains("backpressure"));
+        assert!(e.to_string().contains("outbound queue full"));
+    }
 }

nodedb-lite/src/nodedb/batch.rs

@@ -24,6 +24,15 @@ impl<S: StorageEngine> NodeDbLite<S> {
             return Ok(());
         }
 
+        if self.governor.pressure() == crate::memory::PressureLevel::Critical {
+            return Err(NodeDbError::storage(
+                crate::error::LiteError::Backpressure {
+                    detail: "batch vector insert rejected: memory governor is at Critical pressure"
+                        .into(),
+                },
+            ));
+        }
+
         let dim = vectors[0].1.len();
 
         {

nodedb-lite/src/nodedb/collection/kv.rs

@@ -114,6 +114,14 @@ impl<S: StorageEngine> NodeDbLite<S> {
         value: &[u8],
         deadline_ms: u64,
     ) -> NodeDbResult<()> {
+        if self.governor.pressure() == crate::memory::PressureLevel::Critical {
+            return Err(nodedb_types::error::NodeDbError::storage(
+                crate::error::LiteError::Backpressure {
+                    detail: "KV write rejected: memory governor is at Critical pressure".into(),
+                },
+            ));
+        }
+
         let rkey = kv_key(collection, key.as_bytes());
         let encoded = encode_value(deadline_ms, value);
 
@@ -126,10 +134,7 @@ impl<S: StorageEngine> NodeDbLite<S> {
                 key: rkey.clone(),
                 value: encoded,
             });
-            let n = buf.ops.len();
-            self.kv_overlay_len
-                .store(buf.overlay.len(), std::sync::atomic::Ordering::Release);
-            n >= KV_FLUSH_THRESHOLD
+            buf.ops.len() >= KV_FLUSH_THRESHOLD
         };
 
         // Invalidate any cached value for this key so subsequent reads go to storage.
@@ -166,34 +171,25 @@ impl<S: StorageEngine> NodeDbLite<S> {
     pub async fn kv_get(&self, collection: &str, key: &str) -> NodeDbResult<Option<Vec<u8>>> {
         let rkey = kv_key(collection, key.as_bytes());
 
-        // Fast path: when the overlay is empty (common case in read-heavy
-        // workloads between flushes), skip the mutex acquire entirely and
-        // go straight to storage. The single-writer design + Release stores
-        // on overlay mutation make this safe: observing `len == 0` means
-        // the writer either hasn't started or has completed; either way the
-        // overlay holds no relevant entry.
-        if self
-            .kv_overlay_len
-            .load(std::sync::atomic::Ordering::Acquire)
-            > 0
-        {
-            // Scope the guard so it is not live at any await point.
-            let overlay_result: Option<Option<Vec<u8>>> = {
-                let buf = self.kv_write_buf.lock_or_recover();
-                buf.overlay.get(&rkey).map(|entry| match entry {
-                    Some(stored) => decode_value(stored).and_then(|(deadline, user_bytes)| {
-                        if is_expired(deadline) {
-                            None
-                        } else {
-                            Some(user_bytes.to_vec())
-                        }
-                    }),
-                    None => None,
-                })
-            };
-            if let Some(result) = overlay_result {
-                return Ok(result);
-            }
+        // Always acquire the write-buffer lock to check the overlay first.
+        // This prevents torn reads that could occur if an unconditional
+        // Acquire load of a length counter raced with a concurrent writer.
+        // Scope the guard so it is not live at any await point.
+        let overlay_result: Option<Option<Vec<u8>>> = {
+            let buf = self.kv_write_buf.lock_or_recover();
+            buf.overlay.get(&rkey).map(|entry| match entry {
+                Some(stored) => decode_value(stored).and_then(|(deadline, user_bytes)| {
+                    if is_expired(deadline) {
+                        None
+                    } else {
+                        Some(user_bytes.to_vec())
+                    }
+                }),
+                None => None,
+            })
+        };
+        if let Some(result) = overlay_result {
+            return Ok(result);
         }
 
         // Cache check: look up the composite key before hitting storage.
@@ -260,10 +256,7 @@ impl<S: StorageEngine> NodeDbLite<S> {
                 ns: Namespace::Kv,
                 key: rkey.clone(),
             });
-            let n = buf.ops.len();
-            self.kv_overlay_len
-                .store(buf.overlay.len(), std::sync::atomic::Ordering::Release);
-            n >= KV_FLUSH_THRESHOLD
+            buf.ops.len() >= KV_FLUSH_THRESHOLD
         };
         // Evict the expired entry so future reads don't serve stale data.
         {
@@ -286,10 +279,7 @@ impl<S: StorageEngine> NodeDbLite<S> {
                 ns: Namespace::Kv,
                 key: rkey.clone(),
             });
-            let n = buf.ops.len();
-            self.kv_overlay_len
-                .store(buf.overlay.len(), std::sync::atomic::Ordering::Release);
-            n >= KV_FLUSH_THRESHOLD
+            buf.ops.len() >= KV_FLUSH_THRESHOLD
         };
 
         // Invalidate the cache so subsequent reads don't return stale data.
@@ -405,10 +395,7 @@ impl<S: StorageEngine> NodeDbLite<S> {
                         key: rkey.clone(),
                     });
                 }
-                let n = buf.ops.len();
-                self.kv_overlay_len
-                    .store(buf.overlay.len(), std::sync::atomic::Ordering::Release);
-                n >= KV_FLUSH_THRESHOLD
+                buf.ops.len() >= KV_FLUSH_THRESHOLD
             };
             // Evict expired keys from the cache.
             {
@@ -543,16 +530,15 @@ impl<S: StorageEngine> NodeDbLite<S> {
     }
 
     /// Internal: flush write buffer to storage without touching CRDT.
-    async fn kv_flush_inner(&self) -> NodeDbResult<usize> {
+    /// `pub(in crate::nodedb)` so the global `flush()` can drain the KV buffer.
+    pub(in crate::nodedb) async fn kv_flush_inner(&self) -> NodeDbResult<usize> {
         let ops: Vec<WriteOp> = {
             let mut buf = self.kv_write_buf.lock_or_recover();
             if buf.ops.is_empty() {
                 return Ok(0);
             }
             let ops = std::mem::take(&mut buf.ops);
             buf.overlay.clear();
-            self.kv_overlay_len
-                .store(0, std::sync::atomic::Ordering::Release);
             ops
         };
 

nodedb-lite/src/nodedb/core/flush.rs

@@ -17,6 +17,11 @@ use super::types::{
 impl<S: StorageEngine> NodeDbLite<S> {
     /// Persist all in-memory state to storage (call before shutdown).
     pub async fn flush(&self) -> NodeDbResult<()> {
+        // Drain the buffered KV writes first — they have their own batch-commit
+        // path. Without this, `flush()` (and the auto-flush timer) would not
+        // persist KV `put`s, contradicting "persist all in-memory state".
+        self.kv_flush_inner().await?;
+
         let mut ops = Vec::new();
 
         // ── Persist CRDT snapshot (CRC32C wrapped) ──
@@ -83,6 +88,8 @@ impl<S: StorageEngine> NodeDbLite<S> {
                 value: names_bytes,
             });
 
+            // Mutated only via the native segment-ext path, compiled out on wasm32.
+            #[cfg_attr(target_arch = "wasm32", allow(unused_mut))]
             let mut segment_data = Vec::new();
             for (name, index) in csr_map.iter() {
                 match index.checkpoint_to_bytes() {
@@ -180,6 +187,8 @@ impl<S: StorageEngine> NodeDbLite<S> {
                 value: names_bytes,
             });
 
+            // Mutated only via the native segment-ext path, compiled out on wasm32.
+            #[cfg_attr(target_arch = "wasm32", allow(unused_mut))]
             let mut segment_data = Vec::new();
             for (name, index) in indices.iter() {
                 let key = format!("hnsw:{name}");
@@ -289,6 +298,27 @@ impl<S: StorageEngine> NodeDbLite<S> {
         .await
         .map_err(|e| NodeDbError::storage(format!("fts flush: {e}")))?;
 
+        // ── Spill FTS + spatial staging buffers to durable queues ────────────
+        // These queues accumulate sync entries written synchronously by
+        // `index_document_text`, `remove_document_text`, `spatial_insert`, and
+        // `spatial_delete`. Spilling here (async, ~every second) keeps the
+        // staging buffers bounded and ensures entries are durable before the
+        // next sync transport drain.
+        #[cfg(not(target_arch = "wasm32"))]
+        if let Some(q) = &self.fts_outbound
+            && let Err(e) = q.flush_staging().await
+        {
+            tracing::warn!(error = %e, "fts outbound flush_staging failed; \
+                    staged entries remain and will be retried on next flush");
+        }
+        #[cfg(not(target_arch = "wasm32"))]
+        if let Some(q) = &self.spatial_outbound
+            && let Err(e) = q.flush_staging().await
+        {
+            tracing::warn!(error = %e, "spatial outbound flush_staging failed; \
+                    staged entries remain and will be retried on next flush");
+        }
+
         Ok(())
     }
 }