feat(control): add flag-gated single-node Calvin deployment

Standalone servers can now synthesize a one-node cluster (self-seeded,
replication factor 1) via `server.single_node_calvin`, driving the same
cluster startup a real deployment uses so the sequencer Raft group and
per-vShard schedulers come up and cross-core transactions take the
deterministic Calvin path without a multi-node cluster. Off by default,
leaving the existing standalone boot path unchanged.

Extends the cluster test harness to spawn nodes through this path and
adds coverage asserting the flag toggles calvin_available and that a
cross-shard write is admitted through the sequencer while single-shard
writes keep working with the flag off.

Author: farhan-syah

nodedb-cluster-tests/tests/single_node_calvin.rs

@@ -0,0 +1,233 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Single-node Calvin always-on: a STANDALONE (non-cluster) server can run the
+//! full Calvin stack — sequencer Raft group + per-vShard schedulers — so a
+//! cross-core (cross-vShard) transaction traverses the SAME deterministic
+//! Calvin path it would in a multi-node cluster.
+//!
+//! Gated behind `server.single_node_calvin` (default `false`). When the flag is
+//! off, the standalone server starts no Calvin stack and every write stays on
+//! the existing single-node path; when it is on, the server synthesizes a
+//! one-node cluster (self-seeded, replication factor 1) via
+//! `init_single_node_calvin`.
+//!
+//! ## What each test proves
+//!
+//! - `flag_on`: with the flag set, `calvin_available` becomes true
+//!   (`cluster_transport` + `sequencer_inbox` are both wired) and an AUTOCOMMIT
+//!   cross-shard write — a `GRAPH INSERT EDGE` whose endpoints home to DISTINCT
+//!   vShards — is admitted to a Calvin epoch (the sequencer's `admitted_total`
+//!   advances) and dual-homes atomically (a reverse/IN traversal from the
+//!   destination reaches the source). That is the sequencer→scheduler path, not
+//!   the single-home fast path and not a `SequencerUnavailable` error.
+//! - `flag_off`: a standalone server with no Calvin stack reports
+//!   `calvin_available == false` and single-shard writes still commit on the
+//!   fast path.
+
+mod common;
+
+use std::collections::HashSet;
+use std::sync::atomic::Ordering;
+use std::time::Duration;
+
+use nodedb_cluster::calvin::SEQUENCER_GROUP_ID;
+use nodedb_types::id::VShardId;
+
+use common::cluster_harness::{TestClusterNode, wait_for};
+use common::pgwire_harness::TestServer;
+
+/// Observed sequencer-group leader id from a node's local Raft status, or `0`
+/// if no leader is known yet.
+fn sequencer_leader(node: &TestClusterNode) -> u64 {
+    let Some(status_fn) = node.shared.raft_status_fn.get() else {
+        return 0;
+    };
+    status_fn()
+        .into_iter()
+        .find(|g| g.group_id == SEQUENCER_GROUP_ID)
+        .map(|g| g.leader_id)
+        .unwrap_or(0)
+}
+
+/// Count of transactions the single-node sequencer has admitted to an epoch, or
+/// `0` if the sequencer metrics handle is not installed yet.
+fn sequencer_admitted(node: &TestClusterNode) -> u64 {
+    node.shared
+        .sequencer_metrics
+        .get()
+        .map(|m| m.admitted_total.load(Ordering::Relaxed))
+        .unwrap_or(0)
+}
+
+/// A `(src, dst)` pair of graph node keys whose home vShards differ, so an edge
+/// between them is genuinely cross-shard (`VShardId::from_key` is how
+/// `insert_edge` homes each endpoint). Deterministic: the mapping is a pure
+/// function of the key bytes.
+fn distinct_vshard_node_keys() -> (String, String) {
+    let dst = "sncalvin_hub".to_string();
+    let vdst = VShardId::from_key(dst.as_bytes()).as_u32();
+    for i in 0u32..4096 {
+        let src = format!("sncalvin_src_{i}");
+        if VShardId::from_key(src.as_bytes()).as_u32() != vdst {
+            return (src, dst);
+        }
+    }
+    panic!("could not find a node key on a distinct vShard from the hub in 4096 tries");
+}
+
+/// Run a `GRAPH TRAVERSE` query that yields a single `result` JSON row.
+async fn query_graph_json(client: &tokio_postgres::Client, sql: &str) -> serde_json::Value {
+    let msgs = client.simple_query(sql).await.expect("simple_query");
+    let row = msgs
+        .iter()
+        .find_map(|m| match m {
+            tokio_postgres::SimpleQueryMessage::Row(r) => Some(r),
+            _ => None,
+        })
+        .expect("graph query returned no result row");
+    let raw = row.get("result").expect("result column present");
+    serde_json::from_str(raw).expect("result column is valid JSON")
+}
+
+/// The set of reached node ids from a `GRAPH TRAVERSE` result JSON.
+fn traversed_node_ids(v: &serde_json::Value) -> HashSet<String> {
+    v.get("nodes")
+        .and_then(|n| n.as_array())
+        .expect("traverse result has a nodes array")
+        .iter()
+        .map(|n| {
+            n.get("id")
+                .and_then(|id| id.as_str())
+                .expect("node has string id")
+                .to_string()
+        })
+        .collect()
+}
+
+/// Flag ON: a standalone server with `single_node_calvin = true` runs the
+/// Calvin stack and routes an autocommit cross-shard edge insert through the
+/// single-node sequencer→scheduler path.
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn single_node_calvin_flag_on_routes_cross_shard_write_through_sequencer() {
+    // 4 Data-Plane cores so distinct vShards land on distinct cores — a genuine
+    // cross-core transaction.
+    let node = TestClusterNode::spawn_single_node_calvin(4)
+        .await
+        .expect("spawn standalone single-node-calvin server");
+
+    // The lone sequencer voter self-elects; wait for it before submitting.
+    wait_for(
+        "single-node sequencer leader elected",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || sequencer_leader(&node) == node.node_id,
+    )
+    .await;
+
+    // calvin_available == true: both fields the neutral gate checks are wired.
+    assert!(
+        node.shared.cluster_transport.is_some(),
+        "single-node calvin must install cluster_transport"
+    );
+    assert!(
+        node.shared.sequencer_inbox.get().is_some(),
+        "single-node calvin must install sequencer_inbox → calvin_available"
+    );
+
+    // A graph-capable collection.
+    node.client
+        .simple_query("CREATE COLLECTION sncalvin_graph")
+        .await
+        .expect("CREATE COLLECTION sncalvin_graph");
+    wait_for(
+        "collection visible on the node",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || node.cached_collection_count() >= 1,
+    )
+    .await;
+
+    let (src, dst) = distinct_vshard_node_keys();
+    let admitted_before = sequencer_admitted(&node);
+
+    // AUTOCOMMIT cross-shard edge insert. Because `calvin_available` is true and
+    // the endpoints home to distinct vShards, `insert_edge` dual-homes the edge
+    // atomically through Calvin (`submit_calvin_routed`) instead of taking the
+    // single-home fast path. If the single-node sequencer stack were not
+    // operational this would error (`SequencerUnavailable`) or time out.
+    node.client
+        .simple_query(&format!(
+            "GRAPH INSERT EDGE IN 'sncalvin_graph' FROM '{src}' TO '{dst}' TYPE 'l'"
+        ))
+        .await
+        .expect("cross-shard edge insert must commit via the single-node Calvin path");
+
+    // Proof it traversed the sequencer→scheduler path: the transaction was
+    // admitted to a Calvin epoch.
+    let admitted_after = sequencer_admitted(&node);
+    assert!(
+        admitted_after > admitted_before,
+        "cross-shard edge must be admitted to a Calvin epoch (before={admitted_before}, \
+         after={admitted_after})"
+    );
+
+    // Functional proof the edge dual-homed atomically: a reverse (IN) traversal
+    // from the destination reaches the source, so the REVERSE_EDGES row landed
+    // on the destination's home vShard.
+    let v = query_graph_json(
+        &node.client,
+        &format!("GRAPH TRAVERSE FROM '{dst}' DEPTH 1 LABEL 'l' DIRECTION in"),
+    )
+    .await;
+    let reached = traversed_node_ids(&v);
+    assert!(
+        reached.contains(&src),
+        "reverse traversal from '{dst}' must reach '{src}' (edge dual-homed via Calvin); \
+         got {reached:?}"
+    );
+
+    node.shutdown().await;
+}
+
+/// Flag OFF: a standalone server starts no Calvin stack, so `calvin_available`
+/// is false and single-shard writes commit on the existing fast path.
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn single_node_calvin_flag_off_keeps_fast_path() {
+    let server = TestServer::start().await;
+
+    // No cluster/Calvin stack was started on a standalone server.
+    assert!(
+        server.shared.cluster_transport.is_none(),
+        "standalone server (flag off) must not install cluster_transport"
+    );
+    assert!(
+        server.shared.sequencer_inbox.get().is_none(),
+        "standalone server (flag off) must leave sequencer_inbox unset → calvin_available false"
+    );
+
+    // A single-shard write commits and reads back on the fast path.
+    server
+        .client
+        .simple_query("CREATE COLLECTION sncalvin_fastpath (id TEXT PRIMARY KEY, v TEXT)")
+        .await
+        .expect("CREATE COLLECTION sncalvin_fastpath");
+    server
+        .client
+        .simple_query("INSERT INTO sncalvin_fastpath (id, v) VALUES ('k1', 'hello')")
+        .await
+        .expect("single-shard INSERT on the fast path");
+
+    let rows = server
+        .client
+        .simple_query("SELECT v FROM sncalvin_fastpath WHERE id = 'k1'")
+        .await
+        .expect("SELECT sncalvin_fastpath");
+    let count = rows
+        .iter()
+        .filter(|m| matches!(m, tokio_postgres::SimpleQueryMessage::Row(_)))
+        .count();
+    assert_eq!(
+        count, 1,
+        "single-shard row must be present after the fast-path write"
+    );
+}

nodedb-test-support/src/cluster_harness/cluster/bringup.rs

@@ -34,6 +34,7 @@ impl TestCluster {
             num_cores,
             log_compaction_threshold,
             replication_factor,
+            single_node_calvin: false,
         };
 
         let node1 = TestClusterNode::spawn_with_full_config(1, vec![], &config).await?;

nodedb-test-support/src/cluster_harness/cluster/types.rs

@@ -25,6 +25,13 @@ pub(crate) struct ClusterSpawnConfig {
     /// `min(replication_factor, node_count)`). Defaults to 3 for every
     /// spawn entry point except [`TestCluster::spawn_three_with_compaction_threshold_and_rf`].
     pub(crate) replication_factor: usize,
+    /// When `true`, the node acquires its cluster handle from
+    /// `init_single_node_calvin` (the flag-gated standalone Calvin
+    /// synthesis) instead of building explicit `ClusterSettings` and
+    /// calling `init_cluster_with_transport`. Used only by
+    /// [`TestClusterNode::spawn_single_node_calvin`]. Defaults to `false`
+    /// for every multi-node spawn path.
+    pub(crate) single_node_calvin: bool,
 }
 
 /// An in-process cluster of `TestClusterNode`s.

nodedb-test-support/src/cluster_harness/node/lifecycle/spawn_full.rs

@@ -58,37 +58,6 @@ impl TestClusterNode {
         let data_dir = tempfile::tempdir()?;
         let data_dir_path: PathBuf = data_dir.path().to_path_buf();
 
-        // Pre-bind the QUIC transport on a random port so we know the
-        // listen address before wiring seeds / cluster settings.
-        let transport = Arc::new(nodedb_cluster::NexarTransport::new(
-            node_id,
-            "127.0.0.1:0".parse()?,
-            nodedb_cluster::TransportCredentials::Insecure,
-        )?);
-        let listen_addr = transport.local_addr();
-
-        // Build cluster settings. Empty `seed_nodes` → single-node
-        // bootstrap by listing only our own address.
-        let seeds = if seed_nodes.is_empty() {
-            vec![listen_addr]
-        } else {
-            seed_nodes
-        };
-        let cluster_settings = ClusterSettings {
-            node_id,
-            listen: listen_addr,
-            seed_nodes: seeds,
-            num_groups: 2,
-            replication_factor,
-            force_bootstrap: false,
-            tls: None,
-            max_active_sessions: 0,
-            login_attempts_per_ip_per_min: 30,
-            login_attempts_per_user_per_min: 10,
-            insecure_transport: true,
-            log_compaction_threshold,
-        };
-
         // Open WAL + dispatcher + event bus.
         let wal = Arc::new(WalManager::open_for_testing(
             &data_dir_path.join("test.wal"),
@@ -106,14 +75,57 @@ impl TestClusterNode {
         let mut shared =
             SharedState::new_with_credentials(dispatcher, Arc::clone(&wal), credentials)?;
 
-        // Initialise the cluster using the pre-bound transport.
-        let handle = nodedb::control::cluster::init_cluster_with_transport(
-            &cluster_settings,
-            transport.clone(),
-            &data_dir_path,
-            tuning,
-        )
-        .await?;
+        // Acquire the cluster handle. The single-node-Calvin path drives the
+        // production `init_single_node_calvin` synthesis (which binds its own
+        // loopback transport); every multi-node path pre-binds a transport and
+        // builds explicit `ClusterSettings`.
+        let (handle, listen_addr) = if config.single_node_calvin {
+            let handle =
+                nodedb::control::cluster::init_single_node_calvin(&data_dir_path, tuning).await?;
+            let listen_addr = handle.transport.local_addr();
+            (handle, listen_addr)
+        } else {
+            // Pre-bind the QUIC transport on a random port so we know the
+            // listen address before wiring seeds / cluster settings.
+            let transport = Arc::new(nodedb_cluster::NexarTransport::new(
+                node_id,
+                "127.0.0.1:0".parse()?,
+                nodedb_cluster::TransportCredentials::Insecure,
+            )?);
+            let listen_addr = transport.local_addr();
+
+            // Build cluster settings. Empty `seed_nodes` → single-node
+            // bootstrap by listing only our own address.
+            let seeds = if seed_nodes.is_empty() {
+                vec![listen_addr]
+            } else {
+                seed_nodes
+            };
+            let cluster_settings = ClusterSettings {
+                node_id,
+                listen: listen_addr,
+                seed_nodes: seeds,
+                num_groups: 2,
+                replication_factor,
+                force_bootstrap: false,
+                tls: None,
+                max_active_sessions: 0,
+                login_attempts_per_ip_per_min: 30,
+                login_attempts_per_user_per_min: 10,
+                insecure_transport: true,
+                log_compaction_threshold,
+            };
+
+            // Initialise the cluster using the pre-bound transport.
+            let handle = nodedb::control::cluster::init_cluster_with_transport(
+                &cluster_settings,
+                transport.clone(),
+                &data_dir_path,
+                tuning,
+            )
+            .await?;
+            (handle, listen_addr)
+        };
 
         // Wire cluster handles into SharedState (mirrors main.rs).
         // `Arc::get_mut` is valid here: `shared` has not been cloned.

nodedb-test-support/src/cluster_harness/node/lifecycle/spawn_variants.rs

@@ -102,7 +102,39 @@ impl TestClusterNode {
             num_cores,
             log_compaction_threshold: None,
             replication_factor: 3,
+            single_node_calvin: false,
         };
         Self::spawn_with_full_config(node_id, seed_nodes, &config).await
     }
+
+    /// Spawn a standalone node with the flag-gated single-node Calvin stack
+    /// (`server.single_node_calvin = true`), exercising the same
+    /// `init_single_node_calvin` synthesis the production standalone boot uses.
+    ///
+    /// The node stands up its own sequencer Raft group and per-vShard
+    /// schedulers, so `calvin_available` becomes true and a cross-vShard
+    /// transaction traverses the deterministic Calvin path — all on one node.
+    /// `num_cores` should be `>= 2` so distinct vShards map to distinct cores.
+    pub async fn spawn_single_node_calvin(
+        num_cores: usize,
+    ) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
+        // Sub-second election so the lone sequencer voter self-elects quickly
+        // under the shared CI CPU pool without spurious re-elections.
+        let tuning = ClusterTransportTuning {
+            health_ping_interval_secs: 1,
+            election_timeout_min_ms: 500,
+            election_timeout_max_ms: 1000,
+            ..ClusterTransportTuning::default()
+        };
+        let config = ClusterSpawnConfig {
+            tuning,
+            graph_tuning: nodedb_types::config::tuning::GraphTuning::default(),
+            query_tuning: nodedb_types::config::tuning::QueryTuning::default(),
+            num_cores,
+            log_compaction_threshold: None,
+            replication_factor: 1,
+            single_node_calvin: true,
+        };
+        Self::spawn_with_full_config(1, vec![], &config).await
+    }
 }