test: add integration tests for gateway, startup gates, and shutdown phases

New test suites cover the gateway and shutdown work introduced in this
branch:

Gateway:
- gateway_execute: basic execute roundtrip through the Gateway
- cluster_execute_request: ExecuteRequest routing over QUIC in a 3-node cluster
- {http,ilp,native,pgwire,resp}_gateway_migration: verify each protocol
  handler falls back correctly from direct dispatch to gateway routing
- listeners_gateway_smoke: all listeners accept queries after gateway enable
- listeners_typed_not_leader: non-leader nodes return NOT_LEADER via gateway
- catalog_recovery_check: diverged in-memory registry is repaired on startup

Startup gates:
- startup_gate_{pgwire,http,native,ilp,resp}: each listener blocks connections
  before GatewayEnable and accepts them immediately after
- startup_failure: node in Failed state rejects all connections

Shutdown phases:
- shutdown_{idempotent,budget,in_flight,abort_offender,event_plane}: phased
  drain correctness, per-phase budget enforcement, and offender abort

Existing tests updated for renamed harness helpers and new startup/shutdown APIs.

Author: farhan-syah

nodedb/tests/catalog_recovery_check.rs

@@ -0,0 +1,221 @@
+//! Integration tests for `ExecuteRequest` / `ExecuteResponse` cross-node RPC.
+//!
+//! Tests the C-β physical-plan forwarding path end-to-end:
+//!   1. Happy path: encode a `PhysicalPlan`, ship it via `ExecuteRequest`,
+//!      get payloads back.
+//!   2. DescriptorMismatch: caller passes a stale version, receiver returns
+//!      `TypedClusterError::DescriptorMismatch`.
+//!   3. DeadlineExceeded: caller passes `deadline_remaining_ms = 0`, receiver
+//!      returns `DeadlineExceeded` immediately — no dispatch to Data Plane.
+//!
+//! These tests run in the `cluster` nextest group (max-threads = 1,
+//! threads-required = num-test-threads) because they bring up 3-node clusters.
+
+mod common;
+
+use std::time::Duration;
+
+use common::cluster_harness::TestCluster;
+use nodedb::bridge::physical_plan::wire as plan_wire;
+use nodedb::bridge::physical_plan::{KvOp, PhysicalPlan};
+use nodedb_cluster::rpc_codec::{
+    DescriptorVersionEntry, ExecuteRequest, RaftRpc, TypedClusterError,
+};
+
+/// Build an `ExecuteRequest` wrapping a trivial `KvOp::Put`.
+fn make_kv_put_request(
+    collection: &str,
+    descriptor_version: u64,
+    deadline_remaining_ms: u64,
+) -> ExecuteRequest {
+    // KvOp::Put expects binary-encoded value bytes (Binary Tuple / msgpack).
+    // Use a minimal msgpack-encoded string via zerompk.
+    let value_bytes = zerompk::to_msgpack_vec(&nodedb_types::Value::String("hello".into()))
+        .expect("encode value");
+    let plan = PhysicalPlan::Kv(KvOp::Put {
+        collection: collection.into(),
+        key: b"test-key".to_vec(),
+        value: value_bytes,
+        ttl_ms: 0,
+    });
+
+    let plan_bytes = plan_wire::encode(&plan).expect("encode plan");
+
+    ExecuteRequest {
+        plan_bytes,
+        tenant_id: 0,
+        deadline_remaining_ms,
+        trace_id: 0xDEAD_CAFE_1234,
+        descriptor_versions: vec![DescriptorVersionEntry {
+            collection: collection.into(),
+            version: descriptor_version,
+        }],
+    }
+}
+
+/// Send an `ExecuteRequest` to a specific node and decode the response.
+///
+/// Uses `send_rpc_to_addr` so the test doesn't need to know a node's ID in the
+/// transport routing table — it just sends directly to the QUIC listen address.
+async fn send_execute_request(
+    transport: &nodedb_cluster::NexarTransport,
+    target_addr: std::net::SocketAddr,
+    req: ExecuteRequest,
+) -> nodedb_cluster::rpc_codec::ExecuteResponse {
+    let rpc = RaftRpc::ExecuteRequest(req);
+    match transport.send_rpc_to_addr(target_addr, rpc).await {
+        Ok(RaftRpc::ExecuteResponse(resp)) => resp,
+        Ok(other) => panic!("expected ExecuteResponse, got {other:?}"),
+        Err(e) => panic!("transport error: {e}"),
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
+async fn execute_request_deadline_exceeded_immediate() {
+    // Simple test that doesn't need a 3-node cluster: a single node already
+    // has `LocalPlanExecutor` wired. Send with deadline_remaining_ms=0 and
+    // verify the receiver returns DeadlineExceeded without touching storage.
+    let node1 = common::cluster_harness::TestClusterNode::spawn(1, vec![])
+        .await
+        .expect("spawn node 1");
+
+    // Give the node a moment to finish startup.
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    let transport = node1
+        .shared
+        .cluster_transport
+        .as_ref()
+        .expect("cluster_transport");
+    let req = make_kv_put_request("deadlines_test", 1, 0 /* deadline = 0 */);
+    let resp = send_execute_request(transport, node1.listen_addr, req).await;
+
+    assert!(!resp.success, "expected failure for expired deadline");
+    match resp.error {
+        Some(TypedClusterError::DeadlineExceeded { .. }) => {}
+        other => panic!("expected DeadlineExceeded, got {other:?}"),
+    }
+
+    node1.shutdown().await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
+async fn execute_request_descriptor_mismatch() {
+    // Single-node: create a collection, then send an ExecuteRequest with
+    // a stale descriptor_version and verify DescriptorMismatch is returned.
+    let node1 = common::cluster_harness::TestClusterNode::spawn(1, vec![])
+        .await
+        .expect("spawn node 1");
+    tokio::time::sleep(Duration::from_millis(200)).await;
+
+    // Create the collection so the node has a real descriptor (version ≥ 1).
+    node1
+        .exec("CREATE COLLECTION schema_check_test KEY TEXT")
+        .await
+        .expect("create collection");
+
+    // Give the metadata applier a moment to commit.
+    tokio::time::sleep(Duration::from_millis(300)).await;
+
+    let transport = node1
+        .shared
+        .cluster_transport
+        .as_ref()
+        .expect("cluster_transport");
+
+    // Version 999 is deliberately stale — the actual version will be 1.
+    let req = make_kv_put_request("schema_check_test", 999, 5000);
+    let resp = send_execute_request(transport, node1.listen_addr, req).await;
+
+    assert!(!resp.success, "expected failure for stale descriptor");
+    match resp.error {
+        Some(TypedClusterError::DescriptorMismatch {
+            collection,
+            expected_version,
+            ..
+        }) => {
+            assert_eq!(collection, "schema_check_test");
+            assert_eq!(expected_version, 999);
+        }
+        other => panic!("expected DescriptorMismatch, got {other:?}"),
+    }
+
+    node1.shutdown().await;
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+async fn execute_request_cross_node_dispatch() {
+    // 3-node cluster: create a collection on the leader, then send an
+    // ExecuteRequest from node 2's transport directly to node 1 (the bootstrap
+    // leader). Verify the response indicates success or a known dispatch error.
+    //
+    // We use version 0 in the descriptor_versions list so any version matches
+    // (the catalog check only rejects when expected ≠ actual AND actual > 0).
+    // This lets the test succeed even if the applier hasn't flushed yet.
+    let cluster = TestCluster::spawn_three()
+        .await
+        .expect("3-node cluster spawn");
+
+    // Create a KV collection on whatever node is the DDL leader.
+    cluster
+        .exec_ddl_on_any_leader("CREATE COLLECTION cross_node_kv KEY TEXT")
+        .await
+        .expect("create collection");
+
+    // Give the metadata applier on all nodes a moment to replicate.
+    tokio::time::sleep(Duration::from_millis(400)).await;
+
+    // Node 2 sends the request; node 1 (bootstrap leader) receives it.
+    let sender_transport = cluster.nodes[1]
+        .shared
+        .cluster_transport
+        .as_ref()
+        .expect("node 2 transport");
+    let target_addr = cluster.nodes[0].listen_addr;
+
+    // Use version 0 to bypass the descriptor check (pre-bootstrap sentinel).
+    let req = ExecuteRequest {
+        plan_bytes: {
+            let value_bytes = zerompk::to_msgpack_vec(&nodedb_types::Value::String("v1".into()))
+                .expect("encode value");
+            let plan = PhysicalPlan::Kv(KvOp::Put {
+                collection: "cross_node_kv".into(),
+                key: b"k1".to_vec(),
+                value: value_bytes,
+                ttl_ms: 0,
+            });
+            plan_wire::encode(&plan).expect("encode plan")
+        },
+        tenant_id: 0,
+        deadline_remaining_ms: 5000,
+        trace_id: 0xBEEF_FACE,
+        descriptor_versions: vec![DescriptorVersionEntry {
+            collection: "cross_node_kv".into(),
+            version: 0, // Accept any version (pre-B.1 sentinel bypass)
+        }],
+    };
+
+    let resp = send_execute_request(sender_transport, target_addr, req).await;
+
+    // The response is either success (Data Plane executed the put) or an
+    // Internal error from the dispatcher (e.g. if no Data Plane core is
+    // registered for this vshard in the test harness). Both are acceptable
+    // outcomes for this path test — we're validating the RPC codec and
+    // handler wiring, not Data Plane correctness.
+    //
+    // What must NOT happen: an unexpected panic, a codec error, or a
+    // DescriptorMismatch (version 0 bypasses that check).
+    match resp.error {
+        Some(TypedClusterError::DescriptorMismatch { .. }) => {
+            panic!("DescriptorMismatch should not fire for version 0");
+        }
+        Some(TypedClusterError::DeadlineExceeded { .. }) => {
+            panic!("DeadlineExceeded should not fire with 5s deadline");
+        }
+        _ => {
+            // success or Internal — both acceptable
+        }
+    }
+
+    cluster.shutdown().await;
+}

nodedb/tests/cluster_execute_request.rs

@@ -47,7 +47,7 @@ pub struct TestClusterNode {
     pub shared: Arc<SharedState>,
     _data_dir: tempfile::TempDir,
     _conn_handle: tokio::task::JoinHandle<()>,
-    pg_shutdown_tx: tokio::sync::watch::Sender<bool>,
+    pg_shutdown_bus: nodedb::control::shutdown::ShutdownBus,
     poller_shutdown_tx: tokio::sync::watch::Sender<bool>,
     cluster_shutdown_tx: tokio::sync::watch::Sender<bool>,
     core_stop_tx: std::sync::mpsc::Sender<()>,
@@ -201,6 +201,7 @@ impl TestClusterNode {
             Arc::clone(&shared),
             trigger_dlq,
             Arc::clone(&shared.cdc_router),
+            Arc::clone(&shared.shutdown),
         );
 
         // Start Raft + install MetadataCommitApplier.
@@ -224,19 +225,54 @@ impl TestClusterNode {
             cluster_shutdown_rx,
         );
 
+        // Construct the gateway and install it (plus its DDL invalidator) on
+        // SharedState, mirroring what main.rs does before listeners bind.
+        //
+        // We use a raw-pointer write because `shared` has already been cloned
+        // by the response poller task, making `Arc::get_mut` return None.
+        // This is sound at this point in setup because:
+        //   1. The response poller only calls `poll_and_route_responses()`,
+        //      which never touches the `gateway` or `gateway_invalidator` fields.
+        //   2. No other concurrent task reads those fields before the pgwire
+        //      listener binds (a few lines below).
+        //   3. The write completes before the pgwire listener spawns, so the
+        //      happens-before relationship is guaranteed.
+        {
+            let shared_for_gw = Arc::clone(&shared);
+            let gateway = Arc::new(nodedb::control::gateway::Gateway::new(shared_for_gw));
+            let invalidator = Arc::new(nodedb::control::gateway::PlanCacheInvalidator::new(
+                &gateway.plan_cache,
+            ));
+            // SAFETY: no concurrent reads of `gateway` / `gateway_invalidator`
+            // at this point (see comment above). Fields start as `None` and
+            // are written once here before any listener starts.
+            unsafe {
+                let state = Arc::as_ptr(&shared) as *mut nodedb::control::state::SharedState;
+                (*state).gateway = Some(Arc::clone(&gateway));
+                (*state).gateway_invalidator = Some(invalidator);
+            }
+        }
+
         // pgwire listener.
+        // In the test harness, use the startup gate already on SharedState
+        // (a pre-fired placeholder from `new_inner`). This means the listener
+        // accepts immediately without a startup-phase delay.
         let pg_listener = PgListener::bind("127.0.0.1:0".parse()?).await?;
         let pg_addr = pg_listener.local_addr();
-        let (pg_shutdown_tx, pg_shutdown_rx) = tokio::sync::watch::channel(false);
+        let (pg_shutdown_bus, _) =
+            nodedb::control::shutdown::ShutdownBus::new(Arc::clone(&shared.shutdown));
         let shared_pg = Arc::clone(&shared);
+        let test_startup_gate = Arc::clone(&shared.startup);
+        let bus_pg = pg_shutdown_bus.clone();
         let pg_handle = tokio::spawn(async move {
             let _ = pg_listener
                 .run(
                     shared_pg,
                     AuthMode::Trust,
                     None,
                     Arc::new(tokio::sync::Semaphore::new(128)),
-                    pg_shutdown_rx,
+                    test_startup_gate,
+                    bus_pg,
                 )
                 .await;
         });
@@ -264,7 +300,7 @@ impl TestClusterNode {
             shared,
             _data_dir: data_dir,
             _conn_handle: conn_handle,
-            pg_shutdown_tx,
+            pg_shutdown_bus,
             poller_shutdown_tx,
             cluster_shutdown_tx,
             core_stop_tx,
@@ -633,6 +669,35 @@ impl TestClusterNode {
             .unwrap_or(false)
     }
 
+    /// Force the routing table on this node to point `group_id` at `fake_leader`,
+    /// creating a stale route.
+    ///
+    /// When the gateway on this node next dispatches to `group_id`, it will send
+    /// the request to `fake_leader` instead of the real leader. The remote node
+    /// (which is NOT the leader for that group) will return `TypedClusterError::NotLeader`,
+    /// causing `retry_not_leader` to update the routing table and retry against
+    /// the real leader. This is the canonical way to exercise the NotLeader retry
+    /// path in tests without needing a real leadership change (which is slow and
+    /// flaky).
+    pub fn force_stale_route_for_test(&self, group_id: u64, fake_leader: u64) {
+        if let Some(ref routing) = self.shared.cluster_routing {
+            let mut table = routing.write().unwrap_or_else(|p| p.into_inner());
+            table.set_leader(group_id, fake_leader);
+        }
+    }
+
+    /// Read the current `not_leader_retry_count` from this node's shared gateway.
+    ///
+    /// Returns 0 if the gateway has not been constructed yet (shouldn't happen
+    /// in tests since the harness wires the gateway during spawn).
+    pub fn not_leader_retry_count(&self) -> u64 {
+        self.shared
+            .gateway
+            .as_ref()
+            .map(|gw| gw.not_leader_retry_count())
+            .unwrap_or(0)
+    }
+
     /// Execute a simple query; returns an error message on SQL error.
     pub async fn exec(&self, sql: &str) -> Result<(), String> {
         match self.client.simple_query(sql).await {
@@ -643,7 +708,7 @@ impl TestClusterNode {
 
     /// Cooperatively shut down every background task this node owns.
     pub async fn shutdown(self) {
-        let _ = self.pg_shutdown_tx.send(true);
+        self.pg_shutdown_bus.initiate();
         let _ = self.cluster_shutdown_tx.send(true);
         let _ = self.poller_shutdown_tx.send(true);
         let _ = self.core_stop_tx.send(());
@@ -678,7 +743,7 @@ impl TestClusterNode {
 /// in milliseconds instead of minutes.
 impl Drop for TestClusterNode {
     fn drop(&mut self) {
-        let _ = self.pg_shutdown_tx.send(true);
+        self.pg_shutdown_bus.initiate();
         let _ = self.cluster_shutdown_tx.send(true);
         let _ = self.poller_shutdown_tx.send(true);
         // `core_stop_tx` is a std mpsc Sender; dropping it disconnects