test: add encryption, auto-flush, and dedup tests; wire nextest setup script for sync tests

New test files:
- `encryption_roundtrip.rs`: verifies passphrase and raw-key encryption
  round-trip across reopen.
- `auto_flush.rs`: verifies that the background flush task persists
  writes within the configured interval.
- `semantics/dedup.rs`: verifies that reconnect re-sends are
  deduplicated by Origin (idempotent-producer gate).

Nextest configuration:
- Add `scripts/ensure-origin.sh` setup script that builds the Origin
  binary before sync-interop and SQL parity test runs.
- Wire the setup script into both the `default` and `ci` profiles for
  all `sync_interop_*`, `sync_load`, `definition_sync_interop`, and
  `sql_parity` test binaries.

Existing integration and sync-interop tests updated throughout to match
the new async delegate API, durable outbound queue signatures, and
`Encryption` parameter on `PagedbStorage::open`.

Author: farhan-syah

.config/nextest.toml

@@ -5,6 +5,13 @@
 #
 # Run with: cargo nextest run -p nodedb-lite
 
+experimental = ["setup-scripts"]
+
+[scripts.setup.build-origin]
+command = { command-line = "scripts/ensure-origin.sh", relative-to = "workspace-root" }
+slow-timeout = { period = "600s", terminate-after = 1 }
+capture-stderr = true
+
 [profile.default]
 # Hard ceiling per test. Anything above this is a bug, not a slow test.
 slow-timeout = { period = "30s", terminate-after = 4 }
@@ -47,6 +54,10 @@ binary(/concurrency/)
 test-group = 'heavy'
 threads-required = 'num-test-threads'
 
+[[profile.default.scripts]]
+filter = 'binary(/sync_interop/) | binary(/sync_load/) | binary(/definition_sync_interop/) | binary(/sql_parity/)'
+setup = 'build-origin'
+
 [test-groups]
 heavy = { max-threads = 1 }
 
@@ -56,3 +67,7 @@ fail-fast = false
 
 [profile.ci.junit]
 path = "junit.xml"
+
+[[profile.ci.scripts]]
+filter = 'binary(/sync_interop/) | binary(/sync_load/) | binary(/definition_sync_interop/) | binary(/sql_parity/)'
+setup = 'build-origin'

nodedb-lite/tests/array_sync_catchup.rs

@@ -195,8 +195,8 @@ async fn snapshot_stream_applies_all_ops() {
 /// `CatchupTracker::record` persists across a reload from the same storage.
 #[tokio::test(flavor = "multi_thread")]
 async fn catchup_last_seen_persists_across_reload() {
-    use nodedb_lite::PagedbStorageDefault;
     use nodedb_lite::sync::array::catchup::CatchupTracker;
+    use nodedb_lite::{Encryption, PagedbStorageDefault};
     use std::sync::Arc;
 
     let dir = tempfile::tempdir().expect("tempdir");
@@ -205,15 +205,23 @@ async fn catchup_last_seen_persists_across_reload() {
     let target_hlc = common::hlc1(77_000);
 
     {
-        let storage = Arc::new(PagedbStorageDefault::open(&path).await.expect("open"));
+        let storage = Arc::new(
+            PagedbStorageDefault::open(&path, Encryption::Plaintext)
+                .await
+                .expect("open"),
+        );
         let tracker = CatchupTracker::load(Arc::clone(&storage))
             .await
             .expect("load");
         tracker.record("arr", target_hlc).await.expect("record");
     }
 
     {
-        let storage = Arc::new(PagedbStorageDefault::open(&path).await.expect("reopen"));
+        let storage = Arc::new(
+            PagedbStorageDefault::open(&path, Encryption::Plaintext)
+                .await
+                .expect("reopen"),
+        );
         let tracker = CatchupTracker::load(storage)
             .await
             .expect("load after restart");

nodedb-lite/tests/array_sync_interop.rs

@@ -17,7 +17,10 @@ mod common;
 #[test]
 #[ignore = "array sync over real Origin transport not yet wired; see module doc"]
 fn array_interop_put_roundtrip() {
-    let _origin = common::origin::OriginServer::spawn();
+    let Some(_origin) = common::origin::OriginServer::try_spawn() else {
+        eprintln!("SKIP: Origin binary unavailable (set NODEDB_BIN or run via `cargo nextest`)");
+        return;
+    };
     // When unignored this test should:
     // 1. Connect a Lite sync client to _origin.sync_addr().
     // 2. Subscribe to an array shape.
@@ -36,7 +39,10 @@ fn array_interop_put_roundtrip() {
 #[test]
 #[ignore = "array sync over real Origin transport not yet wired; see module doc"]
 fn array_interop_catchup_after_gap() {
-    let _origin = common::origin::OriginServer::spawn();
+    let Some(_origin) = common::origin::OriginServer::try_spawn() else {
+        eprintln!("SKIP: Origin binary unavailable (set NODEDB_BIN or run via `cargo nextest`)");
+        return;
+    };
     // When unignored this test should:
     // 1. Seed Origin with array ops via a Lite client that then disconnects.
     // 2. Connect a fresh Lite client with a stale cursor.

nodedb-lite/tests/array_sync_interop_real.rs

@@ -82,6 +82,9 @@ async fn array_delta_apply_and_ack() {
     let msg = ArrayDeltaMsg {
         array: "arr".into(),
         op_payload: payload,
+        producer_id: 0,
+        epoch: 0,
+        seq: 0,
     };
 
     // This is the exact call `dispatch_frame` makes.
@@ -145,6 +148,9 @@ async fn array_delta_idempotent_no_ack() {
     let msg = ArrayDeltaMsg {
         array: "idem".into(),
         op_payload: payload,
+        producer_id: 0,
+        epoch: 0,
+        seq: 0,
     };
 
     // First application — ack expected.
@@ -242,6 +248,9 @@ fn array_delta_frame_roundtrip() {
     let msg = ArrayDeltaMsg {
         array: "rt_arr".into(),
         op_payload: vec![0xDE, 0xAD, 0xBE, 0xEF],
+        producer_id: 0,
+        epoch: 0,
+        seq: 0,
     };
 
     let frame = SyncFrame::try_encode(SyncMessageType::ArrayDelta, &msg)

nodedb-lite/tests/array_sync_reject.rs

@@ -159,6 +159,9 @@ async fn schema_too_new_surfaces_as_rejected_outcome() {
     let msg = ArrayDeltaMsg {
         array: "schema_rej".into(),
         op_payload: payload,
+        producer_id: 0,
+        epoch: 0,
+        seq: 0,
     };
 
     let outcome = harness.inbound.handle_delta(&msg).expect("handle_delta");

nodedb-lite/tests/auto_flush.rs

@@ -0,0 +1,111 @@
+// SPDX-License-Identifier: Apache-2.0
+
+//! Integration tests for the auto-flush background task.
+//!
+//! Verifies the bounded-durability contract: writes are durable within
+//! `auto_flush_ms` milliseconds even without an explicit `flush()` call.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use nodedb_lite::{Encryption, LiteConfig, NodeDbLite, PagedbStorageDefault};
+
+// ---------------------------------------------------------------------------
+// auto_flush_persists_without_explicit_flush
+// ---------------------------------------------------------------------------
+
+/// A key written while auto-flush is active (interval 200 ms) survives a
+/// drop + reopen without any explicit `flush()` call, provided we wait long
+/// enough for at least one tick to fire.
+#[tokio::test]
+async fn auto_flush_persists_without_explicit_flush() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let path = dir.path().join("auto_flush_persist.pagedb");
+
+    {
+        let storage = PagedbStorageDefault::open(&path, Encryption::Plaintext)
+            .await
+            .expect("open storage");
+        let config = LiteConfig {
+            auto_flush_ms: 200,
+            ..LiteConfig::default()
+        };
+        let db = Arc::new(
+            NodeDbLite::open_with_config(storage, 1, config)
+                .await
+                .expect("open db"),
+        );
+        db.start_auto_flush(200);
+
+        db.kv_put("col", "key", b"auto_flushed")
+            .await
+            .expect("kv_put");
+
+        // Wait long enough for at least one auto-flush tick (200 ms interval,
+        // first tick is immediate on native Tokio; second tick fires at ~200 ms).
+        tokio::time::sleep(Duration::from_millis(450)).await;
+
+        // Drop without explicit flush — the auto-flush task already ran.
+    }
+
+    {
+        let storage = PagedbStorageDefault::open(&path, Encryption::Plaintext)
+            .await
+            .expect("reopen storage");
+        let db = NodeDbLite::open(storage, 1).await.expect("reopen db");
+        let got = db.kv_get("col", "key").await.expect("kv_get after reopen");
+        assert_eq!(
+            got.as_deref(),
+            Some(b"auto_flushed".as_slice()),
+            "key must survive reopen when auto-flush fired before drop"
+        );
+    }
+}
+
+// ---------------------------------------------------------------------------
+// disabled_auto_flush_does_not_persist
+// ---------------------------------------------------------------------------
+
+/// With `auto_flush_ms: 0` (disabled) and no explicit `flush()`, a write is
+/// NOT durable — a drop + immediate reopen finds nothing. This documents the
+/// bounded-window contract: callers must either enable auto-flush or call
+/// `flush()` explicitly.
+#[tokio::test]
+async fn disabled_auto_flush_does_not_persist() {
+    let dir = tempfile::tempdir().expect("tempdir");
+    let path = dir.path().join("auto_flush_disabled.pagedb");
+
+    {
+        let storage = PagedbStorageDefault::open(&path, Encryption::Plaintext)
+            .await
+            .expect("open storage");
+        let config = LiteConfig {
+            auto_flush_ms: 0,
+            ..LiteConfig::default()
+        };
+        let db = Arc::new(
+            NodeDbLite::open_with_config(storage, 1, config)
+                .await
+                .expect("open db"),
+        );
+        // auto_flush_ms=0 → start_auto_flush is a no-op.
+        db.start_auto_flush(0);
+
+        db.kv_put("col", "key", b"unflushed").await.expect("kv_put");
+
+        // Drop immediately without flush — no auto-flush task was started.
+    }
+
+    {
+        let storage = PagedbStorageDefault::open(&path, Encryption::Plaintext)
+            .await
+            .expect("reopen storage");
+        let db = NodeDbLite::open(storage, 1).await.expect("reopen db");
+        let got = db.kv_get("col", "key").await.expect("kv_get after reopen");
+        assert!(
+            got.is_none(),
+            "key must NOT survive reopen when auto-flush is disabled and flush() was not called; \
+             got: {got:?}"
+        );
+    }
+}

nodedb-lite/tests/common/harness.rs

@@ -155,6 +155,9 @@ impl SyncHarness {
         let msg = ArrayDeltaMsg {
             array: op.header.array.clone(),
             op_payload: payload,
+            producer_id: 0,
+            epoch: 0,
+            seq: 0,
         };
         self.inbound.handle_delta(&msg).expect("handle_delta")
     }

nodedb-lite/tests/common/origin.rs

@@ -3,8 +3,13 @@
 //! Tests that need a live Origin sync endpoint use [`OriginServer`].
 //! The guard kills the process on drop.
 //!
-//! The Origin binary is located relative to the nodedb workspace target dir.
-//! If the binary is not present the test fails immediately with a clear message.
+//! The Origin binary is located in one of three ways (in priority order):
+//! 1. `NODEDB_BIN` env var — set by the nextest setup script.
+//! 2. `<project-root>/nodedb/target/release/nodedb` (pre-built release).
+//! 3. `<project-root>/nodedb/target/debug/nodedb` (pre-built debug).
+//!
+//! If no binary is found, [`OriginServer::try_spawn`] returns `None` and
+//! the calling test should print a skip message and return early.
 //!
 //! The sync WebSocket listener always binds to `0.0.0.0:9090` (the
 //! `SyncListenerConfig` default). All interop test files are placed in the
@@ -19,47 +24,46 @@ use std::path::{Path, PathBuf};
 use std::process::{Child, Command, Stdio};
 use std::time::{Duration, Instant};
 
-/// Locate the nodedb Origin binary.
+/// Locate the nodedb Origin binary, if available.
+///
+/// Returns `None` when no binary can be found — interop tests treat that as a
+/// skip (see [`OriginServer::try_spawn`]), so a Lite-only checkout still passes.
 ///
 /// Search order:
-/// 1. `NODEDB_BIN` env var (CI override).
-/// 2. `<project-root>/nodedb/target/release/nodedb`
-/// 3. `<project-root>/nodedb/target/debug/nodedb`
+/// 1. `NODEDB_BIN` env var — exported by the `build-origin` nextest setup
+///    script (`scripts/ensure-origin.sh`), which runs `cargo build -p nodedb`
+///    against this workspace's dev-dependency. This is the normal path under
+///    `cargo nextest run`.
+/// 2. This workspace's own `target/{debug,release}/nodedb` — for a manual
+///    `cargo build -p nodedb` outside nextest.
 ///
-/// Project root is inferred by walking up from `CARGO_MANIFEST_DIR`.
-pub fn find_origin_binary() -> PathBuf {
-    if let Ok(path) = env::var("NODEDB_BIN") {
-        return PathBuf::from(path);
-    }
-
-    // Walk up from CARGO_MANIFEST_DIR (nodedb-lite/nodedb-lite/).
-    let manifest =
-        env::var("CARGO_MANIFEST_DIR").expect("CARGO_MANIFEST_DIR must be set in test environment");
-    let manifest = Path::new(&manifest);
-    // Up two levels: nodedb-lite/nodedb-lite → nodedb-lite → project root.
-    let project_root = manifest
-        .parent()
-        .and_then(|p| p.parent())
-        .expect("could not determine project root from CARGO_MANIFEST_DIR");
-
-    let release = project_root.join("nodedb/target/release/nodedb");
-    if release.exists() {
-        return release;
+/// There is deliberately NO hardcoded sibling-repo path: the Origin crate is
+/// resolved through cargo (crates.io or the local `[patch.crates-io]`), so its
+/// binary always lands in this workspace's own target directory.
+pub fn find_origin_binary() -> Option<PathBuf> {
+    if let Ok(val) = env::var("NODEDB_BIN") {
+        let p = PathBuf::from(&val);
+        if p.is_file() {
+            return Some(p);
+        }
     }
 
-    let debug = project_root.join("nodedb/target/debug/nodedb");
-    if debug.exists() {
-        return debug;
+    // This workspace's own cargo target dir. CARGO_MANIFEST_DIR is
+    // nodedb-lite/nodedb-lite/; its parent is the workspace root nodedb-lite/.
+    let manifest = env::var("CARGO_MANIFEST_DIR").ok()?;
+    let workspace_root = Path::new(&manifest).parent()?.to_path_buf();
+    let target = env::var("CARGO_TARGET_DIR")
+        .map(PathBuf::from)
+        .unwrap_or_else(|_| workspace_root.join("target"));
+
+    for profile in ["debug", "release"] {
+        let candidate = target.join(profile).join("nodedb");
+        if candidate.is_file() {
+            return Some(candidate);
+        }
     }
 
-    panic!(
-        "Origin binary not found. Expected one of:\n  {}\n  {}\n\
-         Build with: cd {}/nodedb && cargo build -p nodedb\n\
-         Or set NODEDB_BIN=/path/to/nodedb",
-        release.display(),
-        debug.display(),
-        project_root.display(),
-    )
+    None
 }
 
 /// The sync WebSocket URL that Origin always listens on.
@@ -71,7 +75,7 @@ pub const ORIGIN_PGWIRE_ADDR: &str = "127.0.0.1:6432";
 /// Guard for a running Origin server process.
 ///
 /// Kills the process on drop. Tests obtain an instance via
-/// [`OriginServer::spawn`] or [`OriginServer::spawn_with_pgwire`].
+/// [`OriginServer::try_spawn`] or [`OriginServer::try_spawn_with_pgwire`].
 ///
 /// Each instance has its own temporary data directory so WAL / storage
 /// state from previous runs cannot interfere.
@@ -88,22 +92,28 @@ pub struct OriginServer {
 impl OriginServer {
     /// Spawn a fresh Origin server with a private temp data directory.
     ///
+    /// Returns `None` if the Origin binary cannot be found (Origin repo absent
+    /// or not built). The caller should print a skip message and return early.
+    ///
     /// Blocks (up to 30 s) until the sync WebSocket port is accepting TCP
     /// connections.
-    pub fn spawn() -> Self {
-        Self::spawn_inner(false)
+    pub fn try_spawn() -> Option<Self> {
+        let binary = find_origin_binary()?;
+        Some(Self::spawn_inner(binary, false))
     }
 
     /// Spawn a fresh Origin server with both the sync WebSocket (port 9090)
     /// and the pgwire listener (port 6432) enabled in trust auth mode.
     ///
+    /// Returns `None` if the Origin binary cannot be found.
+    ///
     /// Blocks until **both** ports are accepting TCP connections (up to 30 s).
-    pub fn spawn_with_pgwire() -> Self {
-        Self::spawn_inner(true)
+    pub fn try_spawn_with_pgwire() -> Option<Self> {
+        let binary = find_origin_binary()?;
+        Some(Self::spawn_inner(binary, true))
     }
 
-    fn spawn_inner(with_pgwire: bool) -> Self {
-        let binary = find_origin_binary();
+    fn spawn_inner(binary: PathBuf, with_pgwire: bool) -> Self {
         let data_dir = tempfile::tempdir().expect("create temp data dir for Origin");
 
         let (mut cmd, config_dir) = if with_pgwire {