feat(control): add write-admission gate and enforce it at the SPSC chokepoint

Introduce a neutral write-admission gate as the single seam every
write-class PhysicalPlan passes through before being enqueued to a
Data-Plane core, and a consolidated plan_is_write predicate derived
from the exhaustive required_permission mapping so a new PhysicalPlan
variant cannot silently miss the write/read split.

Add a required Admission field to Request (Admitted, or Exempt with
a reason of Read or AlreadyOrdered) so every construction site makes
an explicit choice, and wire a chokepoint guard into both SPSC
dispatch paths that flags any write-class plan reaching a core marked
exempt-as-read. Lock acquisition against the per-vShard deterministic
lock manager is a no-op for now; this change only funnels writes
through one place and makes the invariant enforceable ahead of that
follow-up.

Author: farhan-syah

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

@@ -56,6 +56,9 @@ impl TestClusterNode {
             statement_digest: None,
             txn_id: None,
             wal_lsn: None,
+            admission: nodedb::bridge::envelope::Admission::Exempt(
+                nodedb::bridge::envelope::ExemptReason::Read,
+            ),
         };
 
         // Register for response routing before dispatching, then submit through
@@ -126,6 +129,7 @@ impl TestClusterNode {
             statement_digest: None,
             txn_id: None,
             wal_lsn: None,
+            admission: nodedb::bridge::envelope::Admission::Admitted,
         };
 
         let mut rx = self.shared.tracker.register(request_id);

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

@@ -56,6 +56,9 @@ impl TestClusterNode {
             statement_digest: None,
             txn_id: None,
             wal_lsn: None,
+            admission: nodedb::bridge::envelope::Admission::Exempt(
+                nodedb::bridge::envelope::ExemptReason::Read,
+            ),
         };
 
         let mut rx = self.shared.tracker.register(request_id);
@@ -117,6 +120,9 @@ impl TestClusterNode {
             statement_digest: None,
             txn_id: None,
             wal_lsn: None,
+            admission: nodedb::bridge::envelope::Admission::Exempt(
+                nodedb::bridge::envelope::ExemptReason::Read,
+            ),
         };
 
         let mut rx = self.shared.tracker.register(request_id);

nodedb-test-support/src/tx_batch_helpers.rs

@@ -57,6 +57,7 @@ pub fn make_request(plan: PhysicalPlan) -> Request {
         statement_digest: None,
         txn_id: None,
         wal_lsn: None,
+        admission: nodedb::bridge::envelope::Admission::Admitted,
     }
 }
 

nodedb/src/bridge/admission_chokepoint.rs

@@ -0,0 +1,172 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! Write-admission chokepoint guard for the SPSC enqueue points.
+//!
+//! Both [`crate::bridge::dispatch::Dispatcher::dispatch`] and
+//! `dispatch_to_core` funnel every request into a Data-Plane core here: no
+//! write-class plan may reach a core marked exempt-as-read — that marker means
+//! the plan claimed to be a non-write and skipped the write-admission gate.
+//! Debug builds trip loudly; release builds increment a counter so a future
+//! write path that bypasses the write-admission gate is caught even when the
+//! assertion is compiled out — turning "a missed write path is a silent
+//! serializability hole" into a loud, observable regression.
+
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use crate::bridge::envelope::Request;
+
+/// Count of write-class Requests that reached an SPSC enqueue point marked
+/// [`Admission::Exempt`]`(`[`ExemptReason::Read`]`)` — i.e. hand-labeled a
+/// non-write and thus bypassed the write-admission gate.
+///
+/// [`Admission::Exempt`]: crate::bridge::envelope::Admission::Exempt
+/// [`ExemptReason::Read`]: crate::bridge::envelope::ExemptReason::Read
+///
+/// Stays at zero by construction: a base-state write is either
+/// [`Admission::Admitted`] by the gate or [`Admission::Exempt`] with
+/// [`ExemptReason::AlreadyOrdered`]. A non-zero value signals a regressed write
+/// path that was marked exempt-as-read.
+///
+/// [`Admission::Admitted`]: crate::bridge::envelope::Admission::Admitted
+/// [`ExemptReason::AlreadyOrdered`]: crate::bridge::envelope::ExemptReason::AlreadyOrdered
+static WRITES_BYPASSED_ADMISSION_GATE: AtomicU64 = AtomicU64::new(0);
+
+/// Read the count of writes that reached an SPSC enqueue point marked
+/// exempt-as-read (bypassing the write-admission gate). Exposed for the metrics
+/// exporter and tests.
+pub fn writes_bypassed_admission_gate() -> u64 {
+    WRITES_BYPASSED_ADMISSION_GATE.load(Ordering::Relaxed)
+}
+
+/// Chokepoint guard shared by both SPSC enqueue points: catch a write-class
+/// plan that reached a Data-Plane core marked exempt-as-read — i.e. bypassed
+/// the write-admission gate while claiming to be a non-write.
+#[inline]
+pub(crate) fn assert_write_admitted(request: &Request) {
+    let bypassed = crate::control::server::shared::write_admission::plan_is_write(&request.plan)
+        && request.admission.is_exempt_as_read();
+    if bypassed {
+        WRITES_BYPASSED_ADMISSION_GATE.fetch_add(1, Ordering::Relaxed);
+    }
+    debug_assert!(
+        !bypassed,
+        "a write-class plan reached the SPSC enqueue marked Exempt(Read) — it bypassed the write-admission gate"
+    );
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::bridge::envelope::{Admission, ExemptReason, PhysicalPlan, Priority};
+    use crate::control::server::shared::write_admission::plan_is_write;
+    use crate::types::{DatabaseId, ReadConsistency, RequestId, TenantId, TraceId, VShardId};
+    use nodedb_physical::physical_plan::{DocumentOp, KvOp};
+    use std::time::{Duration, Instant};
+
+    /// A write-class plan: `plan_is_write` returns true for `KvOp::Put`.
+    fn write_plan() -> PhysicalPlan {
+        PhysicalPlan::Kv(KvOp::Put {
+            collection: "c".into(),
+            key: b"k".to_vec(),
+            value: b"v".to_vec(),
+            ttl_ms: 0,
+            surrogate: nodedb_types::Surrogate::ZERO,
+        })
+    }
+
+    /// A read-class plan: `plan_is_write` returns false for `DocumentOp::PointGet`.
+    fn read_plan() -> PhysicalPlan {
+        PhysicalPlan::Document(DocumentOp::PointGet {
+            collection: "c".into(),
+            document_id: "d".into(),
+            surrogate: nodedb_types::Surrogate::ZERO,
+            pk_bytes: Vec::new(),
+            rls_filters: Vec::new(),
+            system_time: nodedb_types::SystemTimeScope::Current,
+            valid_at_ms: None,
+        })
+    }
+
+    fn request_with(plan: PhysicalPlan, admission: Admission) -> Request {
+        Request {
+            request_id: RequestId::new(1),
+            tenant_id: TenantId::new(1),
+            database_id: DatabaseId::DEFAULT,
+            vshard_id: VShardId::new(0),
+            plan,
+            deadline: Instant::now() + Duration::from_secs(5),
+            priority: Priority::Normal,
+            trace_id: TraceId::generate(),
+            consistency: ReadConsistency::Strong,
+            idempotency_key: None,
+            event_source: crate::event::EventSource::User,
+            user_roles: Vec::new(),
+            user_id: None,
+            statement_digest: None,
+            txn_id: None,
+            wal_lsn: None,
+            admission,
+        }
+    }
+
+    /// The guard BITES: a write-class plan marked `Exempt(Read)` bypassed the
+    /// write-admission gate, so the debug assertion must fire.
+    #[test]
+    #[should_panic(expected = "bypassed the write-admission gate")]
+    fn write_marked_exempt_as_read_trips_guard() {
+        let req = request_with(write_plan(), Admission::Exempt(ExemptReason::Read));
+        assert_write_admitted(&req);
+    }
+
+    /// A write that passed the gate (`Admitted`) does not trip the guard.
+    #[test]
+    fn admitted_write_does_not_trip() {
+        let req = request_with(write_plan(), Admission::Admitted);
+        assert_write_admitted(&req);
+    }
+
+    /// A legitimate read marked `Exempt(Read)` does not trip the guard.
+    #[test]
+    fn read_marked_exempt_as_read_does_not_trip() {
+        let req = request_with(read_plan(), Admission::Exempt(ExemptReason::Read));
+        assert_write_admitted(&req);
+    }
+
+    /// An already-ordered write (Calvin/Raft-follower/replay) is legitimately
+    /// exempt and does not trip the guard.
+    #[test]
+    fn write_marked_already_ordered_does_not_trip() {
+        let req = request_with(
+            write_plan(),
+            Admission::Exempt(ExemptReason::AlreadyOrdered),
+        );
+        assert_write_admitted(&req);
+    }
+
+    /// Sanity: the write plan is classified as a write and the read plan is not.
+    /// The bypass predicate the counter keys on is
+    /// `plan_is_write(&plan) && admission.is_exempt_as_read()`; asserting the
+    /// predicate directly (rather than calling `assert_write_admitted`, which
+    /// panics in debug) verifies the counter-increment condition without the
+    /// panic.
+    #[test]
+    fn plan_is_write_classification_and_bypass_predicate() {
+        let w = write_plan();
+        let r = read_plan();
+        assert!(plan_is_write(&w));
+        assert!(!plan_is_write(&r));
+
+        let exempt_read = Admission::Exempt(ExemptReason::Read);
+        // The exact condition under which the counter increments and the guard trips.
+        assert!(plan_is_write(&w) && exempt_read.is_exempt_as_read());
+        // A read marked exempt-as-read is NOT a bypass.
+        assert!(!(plan_is_write(&r) && exempt_read.is_exempt_as_read()));
+        // A write marked Admitted is NOT a bypass.
+        assert!(!(plan_is_write(&w) && Admission::Admitted.is_exempt_as_read()));
+        // A write marked AlreadyOrdered is NOT a bypass.
+        assert!(
+            !(plan_is_write(&w)
+                && Admission::Exempt(ExemptReason::AlreadyOrdered).is_exempt_as_read())
+        );
+    }
+}

nodedb/src/bridge/dispatch.rs

@@ -14,6 +14,8 @@ use crate::bridge::envelope;
 use crate::control::router::vshard::VShardRouter;
 use crate::data::eventfd::EventFdNotifier;
 
+use crate::bridge::admission_chokepoint::assert_write_admitted;
+
 /// Serialized form of a request that goes through the SPSC ring buffer.
 ///
 /// The bridge crate is generic over `T` — we serialize our typed `Request`
@@ -273,6 +275,7 @@ impl Dispatcher {
     /// then flushes WFQ → physical ring. Returns `Err` when the WFQ itself is
     /// full (total capacity reached across all active databases on that core).
     pub fn dispatch(&mut self, request: envelope::Request) -> crate::Result<()> {
+        assert_write_admitted(&request);
         let tenant_id = request.tenant_id.as_u64();
         let req_id = request.request_id.as_u64();
         let database_id = request.database_id.as_u64();
@@ -373,6 +376,7 @@ impl Dispatcher {
         core_id: usize,
         request: envelope::Request,
     ) -> crate::Result<()> {
+        assert_write_admitted(&request);
         if core_id >= self.cores.len() {
             return Err(crate::Error::Dispatch {
                 detail: format!("core {core_id} out of range (have {})", self.cores.len()),
@@ -510,6 +514,7 @@ mod tests {
             statement_digest: None,
             txn_id: None,
             wal_lsn: None,
+            admission: Admission::Exempt(ExemptReason::Read),
         }
     }
 
@@ -539,6 +544,7 @@ mod tests {
             statement_digest: None,
             txn_id: None,
             wal_lsn: None,
+            admission: Admission::Exempt(ExemptReason::Read),
         }
     }
 

nodedb/src/bridge/envelope.rs

@@ -151,6 +151,73 @@ pub struct Request {
     ///
     /// [`ExecutionTask`]: crate::data::executor::task::ExecutionTask
     pub wal_lsn: Option<Lsn>,
+
+    /// Write-admission decision for this request.
+    ///
+    /// Every write-class [`PhysicalPlan`] MUST pass the neutral write-admission
+    /// gate (`crate::control::server::shared::write_admission`) before it is
+    /// enqueued to a Data-Plane core; the gate stamps [`Admission::Admitted`].
+    /// Requests that do not re-enter the gate carry [`Admission::Exempt`] with
+    /// an [`ExemptReason`] — [`ExemptReason::Read`] for reads / savepoint /
+    /// overlay meta ops, [`ExemptReason::AlreadyOrdered`] for writes already
+    /// serialized elsewhere (Calvin-scheduled applies, Raft-follower / replay /
+    /// clone / checkpoint).
+    ///
+    /// The field is REQUIRED (no `Default`, no `#[serde(default)]`) so every
+    /// `Request` construction site makes an explicit choice — that is the
+    /// write-ingress completeness enforcement. The SPSC enqueue chokepoint
+    /// (`crate::bridge::dispatch`) asserts no write-class plan reaches a core
+    /// with the decision unmade.
+    pub admission: Admission,
+}
+
+/// Write-admission marker carried by every [`Request`].
+///
+/// A write-class plan becomes [`Admission::Admitted`] only by passing the
+/// neutral write-admission gate. Everything that does not re-enter the gate's
+/// OCC fence carries [`Admission::Exempt`] with an explicit [`ExemptReason`].
+/// There is intentionally no "unresolved" variant: the required field makes an
+/// unmade decision unrepresentable, so a missed write path is a compile error
+/// at the construction site rather than a silent serializability hole at
+/// runtime.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Admission {
+    /// Passed the write-admission gate.
+    Admitted,
+    /// Does not re-enter the write-admission gate — either a non-write, or a
+    /// write whose ordering was already decided elsewhere. The [`ExemptReason`]
+    /// records which, so the SPSC chokepoint can tell a legitimately exempt
+    /// write apart from a base-state write that bypassed the gate.
+    Exempt(ExemptReason),
+}
+
+/// Why a [`Request`] is exempt from the write-admission gate.
+///
+/// The distinction is load-bearing at the SPSC chokepoint: a write-class plan
+/// marked [`ExemptReason::Read`] is a bug (a write that bypassed the gate),
+/// whereas [`ExemptReason::AlreadyOrdered`] is a legitimately exempt write.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum ExemptReason {
+    /// The plan is not a base-state write — a read / query, or an overlay /
+    /// savepoint meta-op. It never needs the write fence.
+    Read,
+    /// A write-class plan whose ordering was ALREADY decided elsewhere and does
+    /// NOT re-enter the OCC fence: Calvin-scheduler applies (the scheduler
+    /// already holds the locks), replicated / Raft-follower applies, recovery /
+    /// replay, clone / copy-up materialization, and checkpoint. These are
+    /// legitimately exempt writes.
+    AlreadyOrdered,
+}
+
+impl Admission {
+    /// Whether this marker is [`Admission::Exempt`] with reason
+    /// [`ExemptReason::Read`] — i.e. claims the plan is not a base-state write.
+    ///
+    /// The SPSC chokepoint uses this to catch a write-class plan wrongly marked
+    /// exempt-as-read: such a plan bypassed the write-admission gate.
+    pub fn is_exempt_as_read(&self) -> bool {
+        matches!(self, Admission::Exempt(ExemptReason::Read))
+    }
 }
 
 /// Response envelope: Data Plane -> Control Plane.
@@ -391,6 +458,7 @@ mod tests {
             statement_digest: None,
             txn_id: None,
             wal_lsn: None,
+            admission: Admission::Exempt(ExemptReason::Read),
         }
     }
 
@@ -460,6 +528,7 @@ mod tests {
             statement_digest: None,
             txn_id: None,
             wal_lsn: None,
+            admission: Admission::Exempt(ExemptReason::Read),
         };
         match req.plan {
             PhysicalPlan::Meta(MetaOp::Cancel { target_request_id }) => {

nodedb/src/bridge/mod.rs

@@ -1,5 +1,6 @@
 // SPDX-License-Identifier: BUSL-1.1
 
+pub mod admission_chokepoint;
 pub mod dispatch;
 pub mod envelope;
 pub mod quiesce;
@@ -19,5 +20,6 @@ pub mod window_func {
     pub use nodedb_query::window::*;
 }
 
+pub use admission_chokepoint::writes_bypassed_admission_gate;
 pub use dispatch::Dispatcher;
 pub use envelope::{Request, Response, Status};

nodedb/src/control/array_sync/raft_apply.rs

@@ -425,6 +425,9 @@ fn build_array_request(
         statement_digest: None,
         txn_id: None,
         wal_lsn: None,
+        admission: crate::bridge::envelope::Admission::Exempt(
+            crate::bridge::envelope::ExemptReason::AlreadyOrdered,
+        ),
     }
 }
 

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

@@ -103,6 +103,9 @@ async fn dispatch_meta(
                 statement_digest: None,
                 txn_id: None,
                 wal_lsn: None,
+                admission: crate::bridge::envelope::Admission::Exempt(
+                    crate::bridge::envelope::ExemptReason::AlreadyOrdered,
+                ),
             };
             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

@@ -51,6 +51,9 @@ pub async fn delete_async(tenant_id: u64, name: String, shared: Arc<SharedState>
                 statement_digest: None,
                 txn_id: None,
                 wal_lsn: None,
+                admission: crate::bridge::envelope::Admission::Exempt(
+                    crate::bridge::envelope::ExemptReason::AlreadyOrdered,
+                ),
             };
             let rx = shared.tracker.register(request_id);
             if d.dispatch_to_core(core_id, request).is_err() {