perf(dataplane): steady-state fast path at the response point (v5 S2c, default-off)

rust/crates/dataplane/src/gate.rs

@@ -170,6 +170,52 @@ impl Gate {
         }
     }
 
+    /// Engine fast path for the response-ACK point: commit one steady transition
+    /// inline when the *only* in-flight registration is the caller's own command
+    /// `fwd_permit`.
+    ///
+    /// Unlike [`Gate::try_commit_steady`] this registers **no** new permit: the
+    /// response packet is already flushed to the client before this point, so
+    /// the skipped `events`→loop→`await_effect` round-trip carries no I/O. The
+    /// existing `fwd_permit` keeps the `pending` floor across the response phase
+    /// exactly as before.
+    ///
+    /// `fwd_permit` is a capability/liveness witness. Rust's borrow proves it is
+    /// not dropped for the duration of the call, and [`Arc::ptr_eq`] proves it
+    /// belongs to *this* core — so `pending == 1` means that single registration
+    /// is exactly this permit and nothing else (no redirect/drain/control/
+    /// in-flight sync) is outstanding. The identity check, the `open`/shutdown
+    /// read, the speculative transition, the effect match, and the write-back
+    /// all run in one lock critical section; no raw `pending` escapes, so a
+    /// control intent cannot slip in between a check and the commit (no TOCTOU).
+    /// A same-core identity mismatch falls back rather than asserting.
+    pub(crate) fn try_inline_commit_excluding(
+        &self,
+        fwd_permit: &GatePermit,
+        event: SessionEvent,
+        expected_effect: SessionEffect,
+        shutdown: &watch::Receiver<bool>,
+    ) -> bool {
+        let mut inner = lock(&self.inner);
+        let closed = shutdown.has_changed().map_or(true, |_| *shutdown.borrow());
+        if !Arc::ptr_eq(&fwd_permit.inner, &self.inner)
+            || !inner.open
+            || closed
+            || inner.pending != 1
+        {
+            return false;
+        }
+        let mut speculative = inner.fsm.clone();
+        match speculative.on_event(event) {
+            Ok(effects) if effects.len() == 1 && effects[0] == expected_effect => {
+                inner.fsm = speculative;
+                inner.fast_transitions += 1;
+                true
+            }
+            _ => false,
+        }
+    }
+
     /// Admits a quiescent loop-side operation (currently the backend probe)
     /// atomically with the core: only when the core is `open`, `pending == 0`,
     /// and `state_ok` holds for the current FSM state does it register an
@@ -606,6 +652,154 @@ mod tests {
         assert_eq!(core.state_snapshot(), SessionState::Ready);
     }
 
+    /// Drives a core to `Command` (after a `ClientCommand`) — the state where a
+    /// backend response part produces `ForwardResponseToClient`.
+    fn core_in_command() -> Gate {
+        let core = core_in_ready();
+        assert!(
+            matches!(
+                core.apply_slow(SessionEvent::ClientCommand, true),
+                SlowOutcome::Applied { .. }
+            ),
+            "Ready ClientCommand must apply"
+        );
+        assert_eq!(core.state_snapshot(), SessionState::Command);
+        core
+    }
+
+    #[test]
+    fn try_inline_commit_excluding_commits_the_response_point_without_a_new_permit() -> R {
+        let (_tx, shutdown) = watch::channel(false);
+        let core = core_in_command();
+        // The command's own forward permit is the sole in-flight registration.
+        let own = core.operation_permit();
+        assert_eq!(core.pending(), 1);
+
+        // Reference: what the slow path would do for the same response event.
+        let reference = core_in_command();
+        let SlowOutcome::Applied { result, post_state } =
+            reference.apply_slow(SessionEvent::BackendResponsePart, true)
+        else {
+            return Err("slow path must apply the response part".into());
+        };
+        assert_eq!(result, Ok(vec![SessionEffect::ForwardResponseToClient]));
+
+        // Fast path: commits the same transition inline, registers NO new permit
+        // (the already-flushed packet means no post-commit I/O), bumps the
+        // counter, and lands on the same FSM state.
+        assert!(
+            core.try_inline_commit_excluding(
+                &own,
+                SessionEvent::BackendResponsePart,
+                SessionEffect::ForwardResponseToClient,
+                &shutdown,
+            ),
+            "a quiescent response part whose sole blocker is the own permit commits inline"
+        );
+        assert_eq!(core.fast_transitions(), 1);
+        assert_eq!(
+            core.pending(),
+            1,
+            "no new permit is registered; the own permit still holds the floor"
+        );
+        assert_eq!(
+            core.state_snapshot(),
+            post_state,
+            "fast commit lands on the same FSM state as the slow transition"
+        );
+
+        // The terminal response part also commits inline (Response -> Ready).
+        assert!(
+            core.try_inline_commit_excluding(
+                &own,
+                SessionEvent::BackendResponseTxnDone,
+                SessionEffect::ForwardResponseToClient,
+                &shutdown,
+            ),
+            "the terminal response part also commits inline"
+        );
+        assert_eq!(core.fast_transitions(), 2);
+        assert_eq!(core.state_snapshot(), SessionState::Ready);
+
+        drop(own);
+        assert_eq!(core.pending(), 0, "dropping the own permit frees the floor");
+        Ok(())
+    }
+
+    #[test]
+    fn try_inline_commit_excluding_falls_back_unless_the_own_permit_is_the_sole_blocker() {
+        let (_open_tx, open) = watch::channel(false);
+        let assert_fallback =
+            |core: &Gate, own: &GatePermit, shutdown: &watch::Receiver<bool>, label: &str| {
+                let state = core.state_snapshot();
+                let fast = core.fast_transitions();
+                let pending = core.pending();
+                assert!(
+                    !core.try_inline_commit_excluding(
+                        own,
+                        SessionEvent::BackendResponsePart,
+                        SessionEffect::ForwardResponseToClient,
+                        shutdown,
+                    ),
+                    "{label}: must fall back"
+                );
+                assert_eq!(core.state_snapshot(), state, "{label}: FSM untouched");
+                assert_eq!(core.fast_transitions(), fast, "{label}: no fast count");
+                assert_eq!(core.pending(), pending, "{label}: pending untouched");
+            };
+
+        // pending == 2: own permit plus an extra in-flight op (a concurrent
+        // redirect/drain/control registration) — the own permit is no longer the
+        // sole blocker.
+        let core = core_in_command();
+        let own = core.operation_permit();
+        let extra = core.operation_permit();
+        assert_eq!(core.pending(), 2);
+        assert_fallback(&core, &own, &open, "pending==2");
+        drop(extra);
+
+        // Wrong-gate witness: a permit from a DIFFERENT core fails `Arc::ptr_eq`
+        // even though the host core is quiescent with pending == 1.
+        let other = core_in_command();
+        let foreign = other.operation_permit();
+        let host = core_in_command();
+        let _host_own = host.operation_permit();
+        assert_eq!(host.pending(), 1);
+        assert_fallback(&host, &foreign, &open, "wrong-gate witness");
+
+        // Sealed closed.
+        let sealed = core_in_command();
+        let sealed_own = sealed.operation_permit();
+        sealed.seal_closed();
+        assert_fallback(&sealed, &sealed_own, &open, "sealed");
+
+        // Shutdown observed (value true).
+        let (_tx, closed) = watch::channel(true);
+        let s2 = core_in_command();
+        let s2_own = s2.operation_permit();
+        assert_fallback(&s2, &s2_own, &closed, "shutdown-true");
+
+        // Effect mismatch: correct event/state, wrong expected effect.
+        let mism = core_in_command();
+        let mism_own = mism.operation_permit();
+        let state = mism.state_snapshot();
+        assert!(
+            !mism.try_inline_commit_excluding(
+                &mism_own,
+                SessionEvent::BackendResponsePart,
+                SessionEffect::ForwardCommandToBackend,
+                &open,
+            ),
+            "a non-matching expected effect must fall back"
+        );
+        assert_eq!(
+            mism.state_snapshot(),
+            state,
+            "mismatch leaves the FSM untouched"
+        );
+        assert_eq!(mism.fast_transitions(), 0);
+    }
+
     #[test]
     fn probe_permit_is_quiescent_and_blocks_the_fast_path_across_its_await() {
         let (_tx, shutdown) = watch::channel(false);

rust/crates/dataplane/src/session_engine.rs

@@ -2858,8 +2858,7 @@ impl Engine {
             // The permit (fast) or the awaited effect's permit (slow) is held to
             // the end of this iteration, keeping `pending >= 1` across the
             // forward exactly as the slow path does.
-            let _fwd_cmd_permit = if self.fast_path_enabled && event == SessionEvent::ClientCommand
-            {
+            let fwd_cmd_permit = if self.fast_path_enabled && event == SessionEvent::ClientCommand {
                 match self.core.try_commit_steady(
                     SessionEvent::ClientCommand,
                     SessionEffect::ForwardCommandToBackend,
@@ -2992,7 +2991,7 @@ impl Engine {
             let response_source = if pending.expected == ExpectedResponse::Prepare {
                 self.prepare_response_rounds(&pending).await
             } else {
-                self.response_rounds(&pending).await
+                self.response_rounds(&pending, &fwd_cmd_permit).await
             };
             self.record_command(&pending);
             if let Some(source) = response_source {
@@ -3510,7 +3509,11 @@ impl Engine {
 
     /// Streams one command's backend response(s) to the client.
     #[allow(clippy::too_many_lines)] // hot response loop; opt#5 bypass adds a branch
-    async fn response_rounds(&mut self, pending: &PendingCommand) -> Option<WireErrorSource> {
+    async fn response_rounds(
+        &mut self,
+        pending: &PendingCommand,
+        fwd_permit: &crate::gate::GatePermit,
+    ) -> Option<WireErrorSource> {
         let Ok(mut observer) = ResponseObserver::new(
             pending.expected,
             self.negotiated,
@@ -3691,16 +3694,39 @@ impl Engine {
             }
             bypass_since_yield = 0;
             let event = effect.session_event();
-            if self.events.send(event).await.is_err() {
-                return Some(WireErrorSource::Proxy);
-            }
             let expected_ack = match effect.disposition {
                 ResponseDisposition::LocalInfile => SessionEffect::RequestLocalInfileFromClient,
                 _ => SessionEffect::ForwardResponseToClient,
             };
-            let _resp_ack_permit = match self.await_effect(expected_ack).await {
-                EffectAck::Got(permit) => permit,
-                EffectAck::Closing => return None,
+            // S2c response-point fast path: the response packet is already
+            // flushed to the client above, so this `events`→loop→`await_effect`
+            // round-trip is pure FSM bookkeeping with no I/O. When enabled and
+            // the only in-flight registration is this command's own forward
+            // permit (`pending == 1`), commit the response transition inline and
+            // skip the hop. `LocalInfile` (which expects a different ack and
+            // then reads from the client) always takes the slow path. Default-off.
+            let committed_inline = self.fast_path_enabled
+                && expected_ack == SessionEffect::ForwardResponseToClient
+                && self.core.try_inline_commit_excluding(
+                    fwd_permit,
+                    event,
+                    SessionEffect::ForwardResponseToClient,
+                    &self.shutdown,
+                );
+            // On the slow path the ack permit is held to the end of the loop
+            // iteration exactly as before (so flag-off timing is byte-identical);
+            // on the inline path there is no new permit and `fwd_permit` keeps
+            // the floor.
+            let _resp_ack_permit = if committed_inline {
+                None
+            } else {
+                if self.events.send(event).await.is_err() {
+                    return Some(WireErrorSource::Proxy);
+                }
+                match self.await_effect(expected_ack).await {
+                    EffectAck::Got(permit) => Some(permit),
+                    EffectAck::Closing => return None,
+                }
             };
             match effect.disposition {
                 ResponseDisposition::Continue | ResponseDisposition::MoreResults => {}