feat(control): thread real read watermarks through cross-node gather

ExecuteResponse now carries watermark_lsn so a remote-homed read
reports the executing core's actual committed LSN instead of the
former hardcoded Lsn::ZERO. The gateway exposes execute_with_watermarks
to surface per-shard (vshard, lsn) pairs alongside the collected
payloads, and dispatch_route returns a DispatchOutcome so local and
remote dispatch report watermarks uniformly; remote dispatch is split
out into its own module to keep the outcome-returning path readable.
Graph traversal and cross-shard scatter/gather call sites are updated
to consume the new outcome shape.

Author: farhan-syah

nodedb-cluster-tests/tests/cluster_execute_request.rs

@@ -21,7 +21,7 @@ use nodedb_cluster::rpc_codec::{
     DescriptorVersionEntry, ExecuteRequest, RaftRpc, TypedClusterError,
 };
 use nodedb_physical::physical_plan::wire as plan_wire;
-use nodedb_physical::physical_plan::{KvOp, PhysicalPlan};
+use nodedb_physical::physical_plan::{DocumentOp, KvOp, PhysicalPlan};
 
 /// Build an `ExecuteRequest` wrapping a trivial `KvOp::Put`.
 fn make_kv_put_request(
@@ -102,6 +102,81 @@ async fn execute_request_deadline_exceeded_immediate() {
     node1.shutdown().await;
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
+async fn execute_request_read_carries_watermark_lsn() {
+    // Single-node: create a document_schemaless collection, commit a write, then
+    // ship a Document `Scan` via `ExecuteRequest` and assert the response body
+    // carries a non-zero read watermark LSN. Before the wire gained
+    // `ExecuteResponse.watermark_lsn`, the non-streaming read path implicitly
+    // reported 0 (the cross-node gather hardcoded `Lsn::ZERO`); this proves the
+    // producer captures the executing core's real committed LSN and it
+    // roundtrips over the wire.
+    let node1 = common::cluster_harness::TestClusterNode::spawn(1, vec![])
+        .await
+        .expect("spawn node 1");
+    tokio::time::sleep(Duration::from_millis(300)).await;
+
+    node1
+        .exec("CREATE COLLECTION watermark_scan_test WITH (engine='document_schemaless')")
+        .await
+        .expect("create collection");
+
+    // Commit a write so the leader's WAL LSN advances past zero.
+    node1
+        .exec("INSERT INTO watermark_scan_test (id, body) VALUES ('d1', 'hello')")
+        .await
+        .expect("insert document");
+
+    // Let the metadata + write commit and the Data Plane register the collection.
+    tokio::time::sleep(Duration::from_millis(300)).await;
+
+    let transport = node1
+        .shared
+        .cluster_transport
+        .as_ref()
+        .expect("cluster_transport");
+
+    // Empty descriptor_versions → no descriptor-version validation on the
+    // receiver, so the scan reaches the local executor unconditionally.
+    let scan = PhysicalPlan::Document(DocumentOp::Scan {
+        collection: "watermark_scan_test".into(),
+        limit: 100,
+        offset: 0,
+        sort_keys: vec![],
+        filters: vec![],
+        distinct: false,
+        projection: vec![],
+        computed_columns: vec![],
+        window_functions: vec![],
+        system_time: nodedb_types::SystemTimeScope::Current,
+        valid_at_ms: None,
+        prefilter: None,
+    });
+    let req = ExecuteRequest {
+        plan_bytes: plan_wire::encode(&scan).expect("encode scan plan"),
+        tenant_id: 0,
+        database_id: 0,
+        deadline_remaining_ms: 5000,
+        trace_id: [0u8; 16],
+        descriptor_versions: vec![],
+    };
+
+    let resp = send_execute_request(transport, node1.listen_addr, req).await;
+
+    assert!(
+        resp.success,
+        "scan should succeed, got error: {:?}",
+        resp.error
+    );
+    assert!(
+        resp.watermark_lsn > 0,
+        "read response must carry the executing core's real committed LSN, got {}",
+        resp.watermark_lsn
+    );
+
+    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

nodedb-cluster/src/rpc_codec/execute.rs

@@ -47,6 +47,10 @@ pub struct ExecuteResponse {
     /// Raw Data Plane response payloads, one per result set.
     pub payloads: Vec<Vec<u8>>,
     pub error: Option<TypedClusterError>,
+    /// Max read watermark LSN observed by the executing node's cores; 0 for
+    /// writes/errors. Mirrors [`ExecuteStreamChunk::watermark_lsn`]: raw `u64`
+    /// on the wire, converted to `Lsn` at the coordinator via `Lsn::new`.
+    pub watermark_lsn: u64,
 }
 
 /// Typed error returned by the remote executor.
@@ -97,18 +101,20 @@ pub struct ExecuteStreamEnd {
 }
 
 impl ExecuteResponse {
-    pub fn ok(payloads: Vec<Vec<u8>>) -> Self {
+    pub fn ok(payloads: Vec<Vec<u8>>, watermark_lsn: u64) -> Self {
         Self {
             success: true,
             payloads,
             error: None,
+            watermark_lsn,
         }
     }
     pub fn err(error: TypedClusterError) -> Self {
         Self {
             success: false,
             payloads: vec![],
             error: Some(error),
+            watermark_lsn: 0,
         }
     }
 }
@@ -268,12 +274,16 @@ mod tests {
 
     #[test]
     fn roundtrip_execute_response_success() {
-        let resp = ExecuteResponse::ok(vec![b"row1".to_vec(), b"row2".to_vec()]);
+        let resp = ExecuteResponse::ok(vec![b"row1".to_vec(), b"row2".to_vec()], 0xCAFE_1234);
         let decoded = roundtrip_resp(resp);
         assert!(decoded.success);
         assert_eq!(decoded.payloads.len(), 2);
         assert_eq!(decoded.payloads[0], b"row1");
         assert!(decoded.error.is_none());
+        assert_eq!(
+            decoded.watermark_lsn, 0xCAFE_1234,
+            "read watermark roundtrips on the response body"
+        );
     }
 
     #[test]
@@ -286,6 +296,10 @@ mod tests {
         });
         let decoded = roundtrip_resp(resp);
         assert!(!decoded.success);
+        assert_eq!(
+            decoded.watermark_lsn, 0,
+            "error responses carry no watermark"
+        );
         match decoded.error {
             Some(TypedClusterError::NotLeader {
                 group_id,

nodedb/src/control/exec_receiver/executor.rs

@@ -271,7 +271,8 @@ impl LocalPlanExecutor {
             )
             .await
             {
-                Ok(payload) => ExecuteResponse::ok(vec![payload]),
+                // Replicated writes carry no read watermark → 0.
+                Ok(payload) => ExecuteResponse::ok(vec![payload], 0),
                 Err(e) => ExecuteResponse::err(TypedClusterError::Internal {
                     code: PLAN_DECODE_FAILED,
                     message: e.to_string(),
@@ -285,7 +286,9 @@ impl LocalPlanExecutor {
         )
         .await
         {
-            Ok(Ok(result)) => ExecuteResponse::ok(vec![result.payload]),
+            Ok(Ok(result)) => {
+                ExecuteResponse::ok(vec![result.payload], result.watermark_lsn.as_u64())
+            }
             Ok(Err(e)) => ExecuteResponse::err(TypedClusterError::Internal {
                 code: PLAN_DECODE_FAILED,
                 message: e.to_string(),

nodedb/src/control/gateway/core.rs

@@ -27,7 +27,7 @@ use tracing::{Instrument, debug, info_span};
 use crate::Error;
 use crate::control::state::SharedState;
 use crate::control::trace_export::EmitSpanParams;
-use crate::types::{DatabaseId, TenantId, TraceId};
+use crate::types::{DatabaseId, Lsn, TenantId, TraceId, VShardId};
 use nodedb_physical::physical_plan::PhysicalPlan;
 
 use super::dispatcher::{default_deadline_ms, dispatch_route};
@@ -106,11 +106,33 @@ impl Gateway {
     ///
     /// Returns one `Vec<u8>` payload per vShard result. For point operations
     /// the returned Vec has exactly one element.
+    ///
+    /// Thin wrapper over [`Gateway::execute_with_watermarks`] that discards the
+    /// per-shard read watermarks — for the ~5 existing callers that only need
+    /// payloads.
     pub async fn execute(
         &self,
         ctx: &QueryContext,
         plan: PhysicalPlan,
     ) -> Result<Vec<Vec<u8>>, Error> {
+        self.execute_with_watermarks(ctx, plan)
+            .await
+            .map(|(payloads, _watermarks)| payloads)
+    }
+
+    /// Execute a pre-planned `PhysicalPlan`, returning both the raw payloads and
+    /// the per-shard read watermarks observed across every dispatched route.
+    ///
+    /// Each `(vshard, watermark_lsn)` entry is one participating shard's real
+    /// committed LSN (local SPSC response watermark or the remote's
+    /// `ExecuteResponse.watermark_lsn`). The cross-node gather consumer folds
+    /// these into the transaction read-set so a remote-homed read records the
+    /// remote's actual LSN instead of the former hardcoded `Lsn::ZERO`.
+    pub async fn execute_with_watermarks(
+        &self,
+        ctx: &QueryContext,
+        plan: PhysicalPlan,
+    ) -> Result<(Vec<Vec<u8>>, Vec<(VShardId, Lsn)>), Error> {
         let shared = self.shared()?;
         let span = info_span!(
             "gateway.execute",
@@ -201,7 +223,8 @@ impl Gateway {
                     debug!(sql = %sql, "gateway: plan cache hit (two-phase)");
                     return self
                         .execute_with_version_set(ctx, (*cached_plan).clone(), stored_vs)
-                        .await;
+                        .await
+                        .map(|(payloads, _watermarks)| payloads);
                 }
             }
             // Stored version set is stale or plan was evicted — fall through
@@ -225,16 +248,22 @@ impl Gateway {
             .insert_version_set(sql_key, actual_vs.clone());
         self.plan_cache.insert(actual_key, Arc::new(plan.clone()));
 
-        self.execute_with_version_set(ctx, plan, actual_vs).await
+        self.execute_with_version_set(ctx, plan, actual_vs)
+            .await
+            .map(|(payloads, _watermarks)| payloads)
     }
 
     /// Core execution path: route → dispatch with retry → fuse.
+    ///
+    /// Returns the fused/collected payloads alongside every route's per-shard
+    /// read watermarks (one `(vshard, watermark_lsn)` per participating shard,
+    /// accumulated across routes — never collapsed).
     async fn execute_with_version_set(
         &self,
         ctx: &QueryContext,
         plan: PhysicalPlan,
         version_set: GatewayVersionSet,
-    ) -> Result<Vec<Vec<u8>>, Error> {
+    ) -> Result<(Vec<Vec<u8>>, Vec<(VShardId, Lsn)>), Error> {
         let shared = self.shared()?;
         let routes = self.compute_routes(plan, ctx)?;
 
@@ -245,6 +274,7 @@ impl Gateway {
         // N × cap across routes.
         let max_total_bytes = shared.tuning.network.max_query_result_bytes as usize;
         let mut all_payloads: Vec<Vec<u8>> = Vec::new();
+        let mut all_shard_watermarks: Vec<(VShardId, Lsn)> = Vec::new();
         let mut accumulated_bytes: usize = 0;
 
         for route in routes {
@@ -258,7 +288,7 @@ impl Gateway {
             let retry_counter = Arc::clone(&self.not_leader_retry_count);
             let version_set_for_route = version_set.clone();
             let shared_for_route = Arc::clone(&shared);
-            let payloads = retry_not_leader(routing_ref, move |attempt| {
+            let outcome = retry_not_leader(routing_ref, move |attempt| {
                 if attempt > 0 {
                     retry_counter.fetch_add(1, Ordering::Relaxed);
                 }
@@ -324,7 +354,12 @@ impl Gateway {
                 e
             })?;
 
-            for p in payloads {
+            // Accumulate this route's per-shard watermarks — one entry per
+            // participating shard, never collapsed to a scalar, so a multi-route
+            // read produces one read-set entry per shard.
+            all_shard_watermarks.extend(outcome.shard_watermarks);
+
+            for p in outcome.payloads {
                 accumulated_bytes = accumulated_bytes.saturating_add(p.len());
                 if accumulated_bytes > max_total_bytes {
                     return Err(Error::ExecutionLimitExceeded {
@@ -338,12 +373,14 @@ impl Gateway {
             }
         }
 
-        // For broadcast scans, fuse all shard payloads into one.
+        // For broadcast scans, fuse all shard payloads into one. The per-shard
+        // watermarks are NOT fused — each participating shard keeps its own
+        // read-set entry.
         if all_payloads.len() > 1 {
             let fused = fuse_payloads(all_payloads)?;
-            Ok(vec![fused.payload])
+            Ok((vec![fused.payload], all_shard_watermarks))
         } else {
-            Ok(all_payloads)
+            Ok((all_payloads, all_shard_watermarks))
         }
     }