feat(data): add per-core write-version index keyed by WAL LSN

Thread the WAL LSN the Control Plane allocates at wal-dispatch time
through the Request envelope and ExecutionTask onto every Data Plane
apply chokepoint (point/bulk KV, document, graph, columnar family,
transaction handlers). Each chokepoint records the committed write
into a new per-core WriteVersionIndex, tracking the last write LSN
per key and per collection, with horizon GC bounding the per-key map.

This is the shard-local write-version substrate an upcoming
optimistic-concurrency commit path will validate transaction read-sets
against; nothing reads the index yet.

Author: farhan-syah

nodedb-test-support/src/cluster_harness/node/inspect/crdt.rs

@@ -55,6 +55,7 @@ impl TestClusterNode {
             user_id: None,
             statement_digest: None,
             txn_id: None,
+            wal_lsn: None,
         };
 
         // Register for response routing before dispatching, then submit through
@@ -124,6 +125,7 @@ impl TestClusterNode {
             user_id: None,
             statement_digest: None,
             txn_id: None,
+            wal_lsn: None,
         };
 
         let mut rx = self.shared.tracker.register(request_id);

nodedb-test-support/src/cluster_harness/node/inspect/snapshot.rs

@@ -55,6 +55,7 @@ impl TestClusterNode {
             user_id: None,
             statement_digest: None,
             txn_id: None,
+            wal_lsn: None,
         };
 
         let mut rx = self.shared.tracker.register(request_id);
@@ -115,6 +116,7 @@ impl TestClusterNode {
             user_id: None,
             statement_digest: None,
             txn_id: None,
+            wal_lsn: None,
         };
 
         let mut rx = self.shared.tracker.register(request_id);

nodedb-test-support/src/tx_batch_helpers.rs

@@ -56,6 +56,7 @@ pub fn make_request(plan: PhysicalPlan) -> Request {
         user_id: None,
         statement_digest: None,
         txn_id: None,
+        wal_lsn: None,
     }
 }
 

nodedb/src/bridge/dispatch.rs

@@ -509,6 +509,7 @@ mod tests {
             user_id: None,
             statement_digest: None,
             txn_id: None,
+            wal_lsn: None,
         }
     }
 
@@ -537,6 +538,7 @@ mod tests {
             user_id: None,
             statement_digest: None,
             txn_id: None,
+            wal_lsn: None,
         }
     }
 

nodedb/src/bridge/envelope.rs

@@ -140,6 +140,17 @@ pub struct Request {
     /// keys the per-transaction staging overlay. `None` for autocommit /
     /// non-transactional / system requests.
     pub txn_id: Option<TxnId>,
+
+    /// WAL LSN the Control Plane allocated for this write at wal-dispatch time.
+    /// The committed write-LSN is part of the cross-plane write contract: the
+    /// Data Plane copies it onto the [`ExecutionTask`] so the apply chokepoint
+    /// records the per-key / per-collection write version (see
+    /// `data::executor::core_loop::write_index`). `None` for reads, control
+    /// ops, and writes whose LSN is not (yet) threaded — the version index is
+    /// skipped rather than advanced with a wrong value.
+    ///
+    /// [`ExecutionTask`]: crate::data::executor::task::ExecutionTask
+    pub wal_lsn: Option<Lsn>,
 }
 
 /// Response envelope: Data Plane -> Control Plane.
@@ -379,6 +390,7 @@ mod tests {
             user_id: None,
             statement_digest: None,
             txn_id: None,
+            wal_lsn: None,
         }
     }
 
@@ -447,6 +459,7 @@ mod tests {
             user_id: None,
             statement_digest: None,
             txn_id: None,
+            wal_lsn: None,
         };
         match req.plan {
             PhysicalPlan::Meta(MetaOp::Cancel { target_request_id }) => {

nodedb/src/control/array_sync/raft_apply.rs

@@ -424,6 +424,7 @@ fn build_array_request(
         user_id: None,
         statement_digest: None,
         txn_id: None,
+        wal_lsn: None,
     }
 }
 

nodedb/src/control/catalog_entry/post_apply/async_dispatch/continuous_aggregate.rs

@@ -102,6 +102,7 @@ async fn dispatch_meta(
                 user_id: None,
                 statement_digest: None,
                 txn_id: None,
+                wal_lsn: None,
             };
             let rx = shared.tracker.register(request_id);
             if d.dispatch_to_core(core_id, request).is_err() {

nodedb/src/control/catalog_entry/post_apply/async_dispatch/materialized_view.rs

@@ -50,6 +50,7 @@ pub async fn delete_async(tenant_id: u64, name: String, shared: Arc<SharedState>
                 user_id: None,
                 statement_digest: None,
                 txn_id: None,
+                wal_lsn: None,
             };
             let rx = shared.tracker.register(request_id);
             if d.dispatch_to_core(core_id, request).is_err() {

nodedb/src/control/checkpoint_manager.rs

@@ -124,6 +124,7 @@ pub async fn run_checkpoint_cycle(
                 user_id: None,
                 statement_digest: None,
                 txn_id: None,
+                wal_lsn: None,
             };
 
             let rx = tracker.register(request_id);

nodedb/src/control/clone/copyup.rs

@@ -95,6 +95,7 @@ pub async fn perform_kv_clone_copyup(params: KvCopyUpParams<'_>) -> crate::Resul
         user_id: None,
         statement_digest: None,
         txn_id: None,
+        wal_lsn: None,
     };
 
     let mut rx = state.tracker.register(req_id);
@@ -248,6 +249,7 @@ pub async fn perform_clone_copyup(params: CopyUpParams<'_>) -> crate::Result<Sur
         user_id: None,
         statement_digest: None,
         txn_id: None,
+        wal_lsn: None,
     };
 
     let mut rx = state.tracker.register(req_id);