diff --git a/rust/crates/dataplane/src/gate.rs b/rust/crates/dataplane/src/gate.rs index ffb4ffdb1..218056f1f 100644 --- a/rust/crates/dataplane/src/gate.rs +++ b/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 { + 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, 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); diff --git a/rust/crates/dataplane/src/session_engine.rs b/rust/crates/dataplane/src/session_engine.rs index 36954561f..6add54773 100644 --- a/rust/crates/dataplane/src/session_engine.rs +++ b/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 { + async fn response_rounds( + &mut self, + pending: &PendingCommand, + fwd_permit: &crate::gate::GatePermit, + ) -> Option { 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 => {}