feat(config,core): add auto-flush background task and outbound queue cap config

Add `LiteConfig::auto_flush_ms` (default 1000 ms) that drives a new
`NodeDbLite::start_auto_flush` background task. The task calls
`flush()` on each tick, bounding the data-loss window uniformly across
all engines (KV buffer, CRDT deltas, HNSW id-map, CSR graph, FTS, spatial).

Add `LiteConfig::outbound_queue_cap` (default 100 000) that caps the
number of pending entries in each durable outbound sync queue. Writes
return `LiteError::Backpressure` when the cap is reached.

Both settings are also configurable via environment variables:
`NODEDB_LITE_AUTO_FLUSH_MS` and `NODEDB_LITE_OUTBOUND_QUEUE_CAP`.

Author: farhan-syah

nodedb-lite/src/config.rs

@@ -6,9 +6,11 @@
 //!
 //! ## Environment variables
 //!
-//! | Variable                | Description                                  | Default |
-//! |-------------------------|----------------------------------------------|---------|
-//! | `NODEDB_LITE_MEMORY_MB` | Total memory budget in mebibytes             | 100     |
+//! | Variable                      | Description                                        | Default |
+//! |-------------------------------|----------------------------------------------------|---------|
+//! | `NODEDB_LITE_MEMORY_MB`          | Total memory budget in mebibytes                   | 100     |
+//! | `NODEDB_LITE_AUTO_FLUSH_MS`      | Auto-flush interval in milliseconds (0 = disabled) | 1000    |
+//! | `NODEDB_LITE_OUTBOUND_QUEUE_CAP` | Max pending entries per durable outbound queue     | 100000  |
 
 use nodedb_types::error::{NodeDbError, NodeDbResult};
 use serde::{Deserialize, Serialize};
@@ -81,12 +83,47 @@ pub struct LiteConfig {
     /// A value of 0 is rejected at open time; use 1 as the effective minimum.
     #[serde(default = "default_kv_cache_capacity")]
     pub kv_cache_capacity: usize,
+
+    /// Maximum number of pending entries in each durable outbound queue
+    /// (columnar and timeseries). Default: 100_000.
+    ///
+    /// When a queue reaches this cap, write operations return
+    /// [`LiteError::Backpressure`] until the sync transport drains entries.
+    /// This bounds RAM usage to the key/pointer overhead regardless of how
+    /// long the device stays offline; the payloads themselves are on disk.
+    ///
+    /// Can also be set via the `NODEDB_LITE_OUTBOUND_QUEUE_CAP` environment
+    /// variable.
+    #[serde(default = "default_outbound_queue_cap")]
+    pub outbound_queue_cap: usize,
+
+    /// Interval between automatic background flushes, in milliseconds.
+    /// Default: 1000 (1 second).
+    ///
+    /// The auto-flush task calls the global `flush()` every `auto_flush_ms`
+    /// milliseconds, bounding the data-loss window uniformly across all engines
+    /// (KV buffer, vector id-map, CRDT deltas, CSR graph, spatial, FTS).
+    ///
+    /// **Durability contract**: `await`-ing a write operation (e.g. `kv_put`,
+    /// `vector_insert`) returning `Ok` does NOT guarantee on-disk durability.
+    /// Durability is bounded by `auto_flush_ms`. Set to 0 to disable the
+    /// background task; call `flush()` explicitly to guarantee durability.
+    #[serde(default = "default_auto_flush_ms")]
+    pub auto_flush_ms: u64,
+}
+
+fn default_outbound_queue_cap() -> usize {
+    100_000
 }
 
 fn default_kv_cache_capacity() -> usize {
     10_000
 }
 
+fn default_auto_flush_ms() -> u64 {
+    1_000
+}
+
 fn default_sync_enabled() -> bool {
     true
 }
@@ -112,10 +149,12 @@ impl Default for LiteConfig {
             loro_percent: 15,
             query_percent: 15,
             sync_enabled: true,
+            outbound_queue_cap: default_outbound_queue_cap(),
             argon2_m_cost: default_argon2_m_cost(),
             argon2_t_cost: default_argon2_t_cost(),
             argon2_p_cost: default_argon2_p_cost(),
             kv_cache_capacity: default_kv_cache_capacity(),
+            auto_flush_ms: default_auto_flush_ms(),
         }
     }
 }
@@ -126,6 +165,10 @@ impl LiteConfig {
     ///
     /// Handled variables:
     /// - `NODEDB_LITE_MEMORY_MB` — total memory budget in mebibytes (parsed as `usize`)
+    /// - `NODEDB_LITE_AUTO_FLUSH_MS` — auto-flush interval in milliseconds (parsed as `u64`;
+    ///   0 = disabled)
+    /// - `NODEDB_LITE_OUTBOUND_QUEUE_CAP` — max pending entries per durable outbound queue
+    ///   (parsed as `usize`; must be > 0)
     pub fn from_env() -> Self {
         let mut cfg = Self::default();
 
@@ -152,6 +195,54 @@ impl LiteConfig {
             }
         }
 
+        if let Ok(val) = std::env::var("NODEDB_LITE_OUTBOUND_QUEUE_CAP") {
+            match val.trim().parse::<usize>() {
+                Ok(cap) if cap > 0 => {
+                    tracing::info!(
+                        env_var = "NODEDB_LITE_OUTBOUND_QUEUE_CAP",
+                        value = cap,
+                        "environment variable override applied"
+                    );
+                    cfg.outbound_queue_cap = cap;
+                }
+                Ok(_) => {
+                    tracing::warn!(
+                        env_var = "NODEDB_LITE_OUTBOUND_QUEUE_CAP",
+                        "value must be > 0; using default 100_000"
+                    );
+                }
+                Err(_) => {
+                    tracing::warn!(
+                        env_var = "NODEDB_LITE_OUTBOUND_QUEUE_CAP",
+                        value = %val,
+                        "ignoring malformed environment variable (expected unsigned integer), \
+                         using default 100_000"
+                    );
+                }
+            }
+        }
+
+        if let Ok(val) = std::env::var("NODEDB_LITE_AUTO_FLUSH_MS") {
+            match val.trim().parse::<u64>() {
+                Ok(ms) => {
+                    tracing::info!(
+                        env_var = "NODEDB_LITE_AUTO_FLUSH_MS",
+                        value = ms,
+                        "environment variable override applied"
+                    );
+                    cfg.auto_flush_ms = ms;
+                }
+                Err(_) => {
+                    tracing::warn!(
+                        env_var = "NODEDB_LITE_AUTO_FLUSH_MS",
+                        value = %val,
+                        "ignoring malformed environment variable (expected unsigned integer), \
+                         using default 1000 ms"
+                    );
+                }
+            }
+        }
+
         cfg
     }
 
@@ -206,6 +297,7 @@ mod tests {
         assert_eq!(cfg.argon2_m_cost, 19_456);
         assert_eq!(cfg.argon2_t_cost, 2);
         assert_eq!(cfg.argon2_p_cost, 1);
+        assert_eq!(cfg.auto_flush_ms, 1_000);
     }
 
     #[test]
@@ -262,6 +354,37 @@ mod tests {
 
         // Cleanup.
         unsafe { std::env::remove_var("NODEDB_LITE_MEMORY_MB") };
+
+        // NODEDB_LITE_AUTO_FLUSH_MS cases.
+
+        // Case A: var absent → default 1000.
+        unsafe { std::env::remove_var("NODEDB_LITE_AUTO_FLUSH_MS") };
+        let cfg = LiteConfig::from_env();
+        assert_eq!(
+            cfg.auto_flush_ms, 1_000,
+            "absent var should give default 1000 ms"
+        );
+
+        // Case B: valid integer → applied.
+        unsafe { std::env::set_var("NODEDB_LITE_AUTO_FLUSH_MS", "500") };
+        let cfg = LiteConfig::from_env();
+        assert_eq!(cfg.auto_flush_ms, 500, "500 ms should be applied");
+
+        // Case C: 0 = disabled.
+        unsafe { std::env::set_var("NODEDB_LITE_AUTO_FLUSH_MS", "0") };
+        let cfg = LiteConfig::from_env();
+        assert_eq!(cfg.auto_flush_ms, 0, "0 should disable auto-flush");
+
+        // Case D: malformed → fallback to default.
+        unsafe { std::env::set_var("NODEDB_LITE_AUTO_FLUSH_MS", "not_a_number") };
+        let cfg = LiteConfig::from_env();
+        assert_eq!(
+            cfg.auto_flush_ms, 1_000,
+            "malformed var should fall back to default 1000 ms"
+        );
+
+        // Cleanup.
+        unsafe { std::env::remove_var("NODEDB_LITE_AUTO_FLUSH_MS") };
     }
 
     #[test]

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

@@ -0,0 +1,80 @@
+// SPDX-License-Identifier: Apache-2.0
+
+//! `NodeDbLite::start_auto_flush` — durable background flush task.
+
+use std::sync::{Arc, Weak};
+use std::time::Duration;
+
+use crate::storage::engine::StorageEngine;
+
+use super::types::NodeDbLite;
+
+impl<S: StorageEngine> NodeDbLite<S> {
+    /// Start a background task that calls the global `flush()` every
+    /// `interval_ms` milliseconds, bounding the data-loss window uniformly
+    /// across all engines (KV buffer, vector id-map, CRDT deltas, CSR graph,
+    /// spatial, FTS).
+    ///
+    /// # Durability contract
+    ///
+    /// `await`-ing a write operation (e.g. `kv_put`, `vector_insert`) returning
+    /// `Ok` does NOT guarantee on-disk durability. Durability is bounded by
+    /// `interval_ms`. For guaranteed durability, call `flush()` explicitly after
+    /// writes.
+    ///
+    /// # Usage
+    ///
+    /// Call this once after wrapping the database in `Arc`:
+    ///
+    /// ```ignore
+    /// let db = Arc::new(NodeDbLite::open(storage, peer_id).await?);
+    /// db.start_auto_flush(1_000); // flush every second
+    /// ```
+    ///
+    /// Direct library users (not using the FFI or WASM wrappers) must call
+    /// this themselves — the embedded `open*` constructors return `Self`, not
+    /// `Arc<Self>`, so the task cannot be spawned internally.
+    ///
+    /// # Task lifecycle
+    ///
+    /// The spawned task holds a `Weak` reference to the database. When the
+    /// `Arc<NodeDbLite>` is dropped, the `Weak` upgrade fails and the task
+    /// exits cleanly — no task leak.
+    ///
+    /// # Disabling
+    ///
+    /// Pass `interval_ms = 0` to skip spawning entirely (auto-flush disabled).
+    pub fn start_auto_flush(self: &Arc<Self>, interval_ms: u64) {
+        if interval_ms == 0 {
+            return;
+        }
+
+        let weak: Weak<Self> = Arc::downgrade(self);
+        let period = Duration::from_millis(interval_ms);
+
+        crate::runtime::spawn(async move {
+            let mut ticker = crate::runtime::interval(period);
+            // Consume the first tick so the initial period elapses before the
+            // first flush (matches Tokio's immediate-first-tick semantics on
+            // native; on WASM the first tick already waits one period).
+            ticker.tick().await;
+
+            loop {
+                ticker.tick().await;
+
+                let db = match weak.upgrade() {
+                    Some(db) => db,
+                    None => break,
+                };
+
+                if let Err(e) = db.flush().await {
+                    tracing::warn!(error = %e, "auto-flush failed");
+                }
+
+                // Drop the strong Arc before the next tick so the loop does
+                // not keep the database alive between ticks.
+                drop(db);
+            }
+        });
+    }
+}

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

@@ -1,4 +1,5 @@
 // SPDX-License-Identifier: Apache-2.0
+mod auto_flush;
 mod flush;
 mod open;
 mod ops;