feat(control): record per-shard watermarks in transaction read-set

Thread per-core watermark LSNs through Gathered/GatherOutcome instead of
collapsing them to a single max, and route every read seam (native SQL
loop, native direct-ops, pgwire dispatch) through one protocol-neutral
read_set::record_read_set capture point. A predicate read fanned over N
shards now records one read-set entry per shard, each validated against
that shard's own watermark, rather than a single collapsed value —
closing a phantom-safety gap where a read from a stale shard could pass
commit-time validation against another shard's higher watermark.

Point reads now capture exact row identity via the shared KeyRepr type
(extracted from the Data Plane's write-version index into a plane-neutral
types::identity module) so read keys and write keys compare directly;
everything else falls back to a collection-scoped predicate observation.

Author: farhan-syah

nodedb-physical/src/physical_plan/kv.rs

@@ -370,3 +370,46 @@ pub enum KvOp {
         count: usize,
     },
 }
+
+impl KvOp {
+    /// The user collection this op targets, if any. Sorted-index ops keyed only
+    /// by an index name (and no direct collection) return `None`; `TransferItem`
+    /// reports its source collection.
+    pub fn collection(&self) -> Option<&str> {
+        match self {
+            KvOp::Get { collection, .. }
+            | KvOp::Put { collection, .. }
+            | KvOp::Insert { collection, .. }
+            | KvOp::InsertIfAbsent { collection, .. }
+            | KvOp::InsertOnConflictUpdate { collection, .. }
+            | KvOp::Delete { collection, .. }
+            | KvOp::Scan { collection, .. }
+            | KvOp::Expire { collection, .. }
+            | KvOp::Persist { collection, .. }
+            | KvOp::GetTtl { collection, .. }
+            | KvOp::BatchGet { collection, .. }
+            | KvOp::BatchPut { collection, .. }
+            | KvOp::RegisterIndex { collection, .. }
+            | KvOp::DropIndex { collection, .. }
+            | KvOp::FieldGet { collection, .. }
+            | KvOp::FieldSet { collection, .. }
+            | KvOp::Truncate { collection, .. }
+            | KvOp::Incr { collection, .. }
+            | KvOp::IncrFloat { collection, .. }
+            | KvOp::Cas { collection, .. }
+            | KvOp::GetSet { collection, .. }
+            | KvOp::Transfer { collection, .. }
+            | KvOp::RegisterSortedIndex { collection, .. }
+            | KvOp::MaterializeScan { collection, .. } => Some(collection.as_str()),
+            KvOp::TransferItem {
+                source_collection, ..
+            } => Some(source_collection.as_str()),
+            KvOp::DropSortedIndex { .. }
+            | KvOp::SortedIndexRank { .. }
+            | KvOp::SortedIndexTopK { .. }
+            | KvOp::SortedIndexRange { .. }
+            | KvOp::SortedIndexCount { .. }
+            | KvOp::SortedIndexScore { .. } => None,
+        }
+    }
+}

nodedb/src/control/gateway/dispatcher.rs

@@ -233,7 +233,7 @@ async fn dispatch_remote(args: RemoteDispatchArgs<'_>) -> Result<Vec<Vec<u8>>, E
     {
         // A root-level Gather resolved entirely at the coordinator — its merged
         // response is ready; return it instead of shipping anything.
-        crate::control::server::exchange::Resolved::Gathered(resp) => {
+        crate::control::server::exchange::Resolved::Gathered(resp, _shard_watermarks) => {
             return Ok(vec![resp.payload.to_vec()]);
         }
         crate::control::server::exchange::Resolved::Plan(p) => p,
@@ -359,7 +359,7 @@ async fn dispatch_remote_stream(args: RemoteDispatchArgs<'_>) -> Result<ResultSt
         // ready response/stream — re-emit it as a single-batch / forwarded
         // stream. These do not occur for the streamable-scan plans routed here,
         // but handle exhaustively and behaviour-preservingly.
-        crate::control::server::exchange::Resolved::Gathered(resp) => {
+        crate::control::server::exchange::Resolved::Gathered(resp, _shard_watermarks) => {
             let batch = RowBatch {
                 payload: resp.payload.to_vec(),
                 watermark_lsn: resp.watermark_lsn,

nodedb/src/control/server/dispatch_utils.rs

@@ -267,7 +267,9 @@ async fn dispatch_to_data_plane_inner(
     )
     .await?
     {
-        crate::control::server::exchange::Resolved::Gathered(resp) => return Ok(resp),
+        crate::control::server::exchange::Resolved::Gathered(resp, _shard_watermarks) => {
+            return Ok(resp);
+        }
         crate::control::server::exchange::Resolved::Plan(p) => p,
         // Internal funnel callers want a fully-collected Response, not a lazy
         // stream: materialize the stream into one merged-array Response,

nodedb/src/control/server/exchange/gather.rs

@@ -108,8 +108,15 @@ pub struct GatherOutcome {
     /// Single merged msgpack array of all row elements.
     /// Consumed by the pgwire/native response path and `ProviderScan` embedding.
     pub merged_array: Vec<u8>,
-    /// Maximum watermark LSN seen across all responding cores.
+    /// Maximum watermark LSN seen across all responding cores. Retained as the
+    /// scalar fence value for Strong-consistency callers that need one LSN.
     pub watermark_lsn: Lsn,
+    /// Per-shard watermark LSNs — one `(vshard, watermark_lsn)` per responding
+    /// core, NOT collapsed to the max. The transaction read-set records one
+    /// entry per participating shard from this, so a predicate read fanned over
+    /// N cores is validated against each core's own version rather than a
+    /// single global max.
+    pub shard_watermarks: Vec<(VShardId, Lsn)>,
 }
 
 /// Fan `plan` to every Data-Plane core in parallel and gather the results.
@@ -149,43 +156,46 @@ pub async fn gather_all_cores(
     let deadline = Duration::from_secs(deadline_secs);
     let max_result_bytes = state.tuning.network.max_query_result_bytes as usize;
     let response_futures = receivers.into_iter().map(|(core_id, mut rx)| async move {
-        match tokio::time::timeout(
+        let result = match tokio::time::timeout(
             deadline,
             crate::control::server::dispatch_utils::collect_bounded_response(
                 &mut rx,
                 max_result_bytes,
             ),
         )
         .await
-        .map_err(|_| crate::Error::Dispatch {
-            detail: format!("gather timeout on core {core_id}"),
-        })? {
-            Ok(resp) => Ok(resp),
-            Err(crate::control::server::dispatch_utils::DispatchCollectError::OverBudget {
+        {
+            Err(_) => Err(crate::Error::Dispatch {
+                detail: format!("gather timeout on core {core_id}"),
+            }),
+            Ok(Ok(resp)) => Ok(resp),
+            Ok(Err(crate::control::server::dispatch_utils::DispatchCollectError::OverBudget {
                 bytes,
-            }) => Err(crate::Error::ExecutionLimitExceeded {
+            })) => Err(crate::Error::ExecutionLimitExceeded {
                 detail: format!(
                     "gather on core {core_id} exceeded max_query_result_bytes \
                      ({bytes} > {max_result_bytes} bytes)"
                 ),
             }),
-            Err(crate::control::server::dispatch_utils::DispatchCollectError::ChannelClosed) => {
-                Err(crate::Error::Dispatch {
-                    detail: format!("gather channel closed on core {core_id}"),
-                })
-            }
-        }
+            Ok(Err(
+                crate::control::server::dispatch_utils::DispatchCollectError::ChannelClosed,
+            )) => Err(crate::Error::Dispatch {
+                detail: format!("gather channel closed on core {core_id}"),
+            }),
+        };
+        (core_id, result)
     });
 
-    let results: Vec<crate::Result<Response>> = join_all(response_futures).await;
+    let results: Vec<(usize, crate::Result<Response>)> = join_all(response_futures).await;
 
     let mut raw = Vec::new();
     let mut all_elements: Vec<Vec<u8>> = Vec::new();
     let mut max_lsn = Lsn::ZERO;
+    let mut shard_watermarks: Vec<(VShardId, Lsn)> = Vec::new();
     let mut had_error = false;
     let mut error_msg = String::new();
 
-    for result in results {
+    for (core_id, result) in results {
         let resp = match result {
             Ok(r) => r,
             Err(e) => {
@@ -208,6 +218,11 @@ pub async fn gather_all_cores(
             continue;
         }
 
+        // Record this core's own watermark as a participating-shard version,
+        // even when its payload is empty — an empty scan slice is still a
+        // validatable observation at that shard's version (phantom safety).
+        shard_watermarks.push((VShardId::new(core_id as u32), resp.watermark_lsn));
+
         if resp.watermark_lsn > max_lsn {
             max_lsn = resp.watermark_lsn;
         }
@@ -231,6 +246,7 @@ pub async fn gather_all_cores(
         raw,
         merged_array,
         watermark_lsn: max_lsn,
+        shard_watermarks,
     })
 }
 
@@ -381,8 +397,12 @@ pub async fn gather_all_vshards(
         // watermark LSNs back through the gateway response, so Strong-consistency
         // LSN fencing degrades to pass-through on this path.  This is consistent
         // with existing gateway behavior (gateway.execute returns no LSN metadata).
-        // Tracked as a follow-up: propagate watermark_lsn through GatewayResponse.
+        // The cross-node per-shard watermark folds into the gateway wire change
+        // (co-located with cross-shard `txn_id`/`database_id` threading), so no
+        // per-shard watermarks are surfaced here yet — a ZERO read version is
+        // over-conservative (never unsafe) and nothing validates it until then.
         watermark_lsn: Lsn::ZERO,
+        shard_watermarks: Vec::new(),
     })
 }
 

nodedb/src/control/server/exchange/resolve/exchange.rs

@@ -20,7 +20,7 @@ use crate::bridge::envelope::Response;
 use crate::control::security::identity::AuthenticatedIdentity;
 use crate::control::state::SharedState;
 use crate::data::executor::response_codec::flatten_to_relational_rows;
-use crate::types::{DatabaseId, TenantId, TraceId, TxnId};
+use crate::types::{DatabaseId, Lsn, TenantId, TraceId, TxnId, VShardId};
 
 use crate::control::server::exchange::gather::{
     GatherOutcome, finalize_aggregate, gather_all_cores, gather_all_cores_stream,
@@ -33,8 +33,13 @@ use super::materialize::materialize_providers;
 /// Result of `resolve_and_materialize`.
 pub enum Resolved {
     /// The plan was a root-level `Gather` — the coordinator has already
-    /// executed it and the response is ready to return to the client.
-    Gathered(Response),
+    /// executed it and the response is ready to return to the client. The
+    /// second field carries the per-shard watermark LSNs the gather observed
+    /// (one `(vshard, watermark_lsn)` per responding core), so an in-transaction
+    /// read can record one read-set entry per participating shard rather than a
+    /// single collapsed max. Empty for cross-node gathers (per-shard watermarks
+    /// are not yet threaded through the gateway) and for shuffle joins.
+    Gathered(Response, Vec<(VShardId, Lsn)>),
     /// The plan (possibly mutated by catalog materialization or Broadcast
     /// embedding) is self-contained and should be dispatched normally.
     Plan(PhysicalPlan),
@@ -131,7 +136,7 @@ async fn resolve_exchange(
             .await?
             {
                 Resolved::Plan(p) => p,
-                Resolved::Gathered(resp) => return Ok(Resolved::Gathered(resp)),
+                Resolved::Gathered(resp, wms) => return Ok(Resolved::Gathered(resp, wms)),
                 // A nested Exchange that itself resolved to a stream cannot be
                 // re-wrapped by an outer Gather without materializing first;
                 // surface it as the stream (the outer Gather is redundant —
@@ -151,7 +156,14 @@ async fn resolve_exchange(
             //   and merges the per-route streams with the same `select_all`.
             //
             // Aggregate gathers keep the materialize-then-merge behaviour.
-            if !as_aggregate && child.is_streamable_unordered_scan() {
+            //
+            // An in-transaction read (`txn_id.is_some()`) also keeps the
+            // materialize path: streaming collapses per-core watermarks into one
+            // value, but a transaction must record each participating shard's own
+            // read version for optimistic-concurrency validation, so it takes the
+            // `gather_all_vshards` branch below whose `GatherOutcome` preserves
+            // `shard_watermarks`.
+            if !as_aggregate && txn_id.is_none() && child.is_streamable_unordered_scan() {
                 let stream = if let Some(gw) = state.gateway.as_ref() {
                     let ctx = crate::control::gateway::core::QueryContext {
                         tenant_id,
@@ -176,10 +188,10 @@ async fn resolve_exchange(
             } else {
                 outcome.merged_array
             };
-            Ok(Resolved::Gathered(outcome_to_response(
-                payload,
-                outcome.watermark_lsn,
-            )))
+            Ok(Resolved::Gathered(
+                outcome_to_response(payload, outcome.watermark_lsn),
+                outcome.shard_watermarks,
+            ))
         }
 
         // Root-level Broadcast: unusual but treat as Gather without merge.
@@ -189,10 +201,10 @@ async fn resolve_exchange(
         })) => {
             let outcome =
                 gather_all_vshards(state, tenant_id, database_id, *child, trace_id, txn_id).await?;
-            Ok(Resolved::Gathered(outcome_to_response(
-                outcome.merged_array,
-                outcome.watermark_lsn,
-            )))
+            Ok(Resolved::Gathered(
+                outcome_to_response(outcome.merged_array, outcome.watermark_lsn),
+                outcome.shard_watermarks,
+            ))
         }
 
         // Root-level Shuffle: orchestrate a real cross-node grace hash join.

nodedb/src/control/server/exchange/resolve/shuffle.rs

@@ -298,7 +298,12 @@ pub async fn resolve_shuffle_join(
     // staged sides are consumed; there is no coordinator-side cleanup hook to
     // call here. No leak: each `(shuffle_id, part, side)` inbox is dropped by
     // the part-owner after its consume completes.
-    Ok(Resolved::Gathered(outcome_to_response(merged, Lsn::ZERO)))
+    // Cross-node shuffle join: per-shard watermarks are not threaded through
+    // the shuffle transport, so no per-shard read versions are surfaced here.
+    Ok(Resolved::Gathered(
+        outcome_to_response(merged, Lsn::ZERO),
+        Vec::new(),
+    ))
 }
 
 /// Send one `ShuffleProduceRequest` and map the reply / RPC error to a typed

nodedb/src/control/server/exchange/resolve/shuffle_aggregate.rs

@@ -311,7 +311,12 @@ pub async fn resolve_shuffle_aggregate(
     }
     let merged = encode_msgpack_array(&elements);
 
-    Ok(Resolved::Gathered(outcome_to_response(merged, Lsn::ZERO)))
+    // Cross-node shuffle aggregate: per-shard watermarks are not threaded
+    // through the shuffle transport, so no per-shard read versions here.
+    Ok(Resolved::Gathered(
+        outcome_to_response(merged, Lsn::ZERO),
+        Vec::new(),
+    ))
 }
 
 /// Send one `ShuffleProduceRequest` and map the reply / RPC error to a typed

nodedb/src/control/server/native/dispatch/direct_ops.rs

@@ -226,6 +226,23 @@ pub(crate) async fn handle_graph_match(
         Err(e) => return error_to_native(seq, &e),
     };
 
+    // A MATCH issued inside a native transaction records a collection-scoped
+    // predicate read at the shard's watermark, identical to every other read
+    // seam. Single-shard direct op → one watermark, one entry.
+    if (resp.status == Status::Ok
+        || resp.error_code == Some(crate::bridge::envelope::ErrorCode::NotFound))
+        && ctx.sessions.transaction_state(ctx.peer_addr)
+            == crate::control::server::shared::session::TransactionState::InBlock
+    {
+        crate::control::server::shared::session::record_read_set(
+            ctx.sessions,
+            ctx.peer_addr,
+            ctx.tenant_id(),
+            &plan_for_response,
+            &[(vshard_id, resp.watermark_lsn)],
+        );
+    }
+
     if resp.status == Status::Error {
         return data_plane_response_to_native(ctx, seq, &plan_for_response, &resp);
     }
@@ -326,10 +343,33 @@ async fn dispatch_single_task(
     };
 
     let plan_for_response = task.plan.clone();
+    let task_vshard = task.vshard_id;
     match dispatch_single_task_raw(ctx, task.tenant_id, task.vshard_id, task.plan, task.txn_id)
         .await
     {
-        Ok(resp) => data_plane_response_to_native(ctx, seq, &plan_for_response, &resp),
+        Ok(resp) => {
+            // Track direct-op reads (PointGet / RangeScan / VectorSearch / KV
+            // Get) for conflict detection at the protocol-neutral layer, so
+            // native direct-ops record identically to native SQL and pgwire.
+            // Absent-key reads record too (a `NotFound` is a validatable phantom
+            // observation). Direct ops are single-shard, so one watermark → one
+            // entry.
+            let records_read = resp.status == Status::Ok
+                || resp.error_code == Some(crate::bridge::envelope::ErrorCode::NotFound);
+            if records_read
+                && ctx.sessions.transaction_state(ctx.peer_addr)
+                    == crate::control::server::shared::session::TransactionState::InBlock
+            {
+                crate::control::server::shared::session::record_read_set(
+                    ctx.sessions,
+                    ctx.peer_addr,
+                    ctx.tenant_id(),
+                    &plan_for_response,
+                    &[(task_vshard, resp.watermark_lsn)],
+                );
+            }
+            data_plane_response_to_native(ctx, seq, &plan_for_response, &resp)
+        }
         Err(e) => error_to_native(seq, &e),
     }
 }