diff --git a/rust/crates/dataplane/src/gate.rs b/rust/crates/dataplane/src/gate.rs index de4e89dcd..7069af34a 100644 --- a/rust/crates/dataplane/src/gate.rs +++ b/rust/crates/dataplane/src/gate.rs @@ -1,12 +1,15 @@ -//! Fast-path admission gate (v5, slice S1 — plumbing only, no fast path yet). +//! Shared session core + fast-path admission gate (v5, slices S1–S2a — no fast +//! path yet). //! //! Every message that can change how the *next* client command is handled — //! engine→loop [`SessionEvent`]s, loop/owner→engine [`EngineCmd`]s, and //! owner→loop [`SessionControl`]s — is registered in a shared counter before it //! is enqueued and deregistered only after it (and every effect it spawns) has -//! been fully processed. A later slice reads `pending == 0 && open` to decide -//! whether the engine may inline a steady transition; S1 only installs the -//! counter and routes all three channels through it, with no behavior change. +//! been fully processed (S1). S2a folds the session FSM into that same locked +//! core, so a later slice can read `pending == 0 && open` together with a +//! speculative transition to decide whether the engine may inline a steady +//! transition. Through S2a the loop is still the FSM's only writer (via +//! [`Gate::apply_slow`]) and there is no behavior change. //! //! Registration is a RAII [`GatePermit`] carried alongside the message in the //! channel (`mpsc::Sender<(T, GatePermit)>`). The permit's `Drop` decrements @@ -18,24 +21,33 @@ use std::sync::{Arc, Mutex}; +use session_core::fsm::{Effects, SessionEvent, SessionFsm, SessionState, TransitionError}; use tokio::sync::mpsc; -/// Shared gate state. The mutex guards only the small counters; no `.await`, -/// send, or effect runs while it is held. +/// Shared session core: the session FSM together with the admission counters, +/// behind one std `Mutex` so a later slice can evaluate the fast-path predicate +/// (`open && pending == 0` with a speculative transition) atomically. The mutex +/// guards only pure computation — FSM transitions and counter arithmetic — +/// never an `.await`, channel send, or effect execution. #[derive(Debug)] -struct GateInner { +struct FsmCore { + /// The session state machine. In S2a the loop is still its only writer (via + /// [`Gate::apply_slow`]); a later slice lets the engine commit whitelisted + /// steady transitions under this same lock. + fsm: SessionFsm, /// Registered-but-not-yet-processed messages across all producers/channels. pending: u32, - /// Loop is running and has not entered terminal handling. (S1: always true - /// and not yet read; the terminal gate consumes it in a later slice.) + /// Loop is running and has not entered terminal handling. Sealed to `false` + /// under the lock before the loop takes any terminal path; once false it + /// never reopens. (S2a maintains it; a later slice's predicate reads it.) #[allow(dead_code)] open: bool, } -/// Handle to the shared admission gate for one session. +/// Handle to the shared session core (FSM + admission gate) for one session. #[derive(Clone, Debug)] pub struct Gate { - inner: Arc>, + inner: Arc>, } impl Default for Gate { @@ -45,17 +57,46 @@ impl Default for Gate { } impl Gate { - /// Creates a gate with zero in-flight messages and `open == true`. + /// Creates a core with a fresh FSM (starting at `Accept`), zero in-flight + /// messages, and `open == true`. #[must_use] pub fn new() -> Self { Self { - inner: Arc::new(Mutex::new(GateInner { + inner: Arc::new(Mutex::new(FsmCore { + fsm: SessionFsm::new(), pending: 0, open: true, })), } } + /// Reads the current FSM state (a `Copy` value) under the lock and releases + /// immediately. Never exposes the guard. + pub(crate) fn state_snapshot(&self) -> SessionState { + lock(&self.inner).fsm.state() + } + + /// Applies one event to the FSM on the slow path, under the lock, returning + /// the owned effects (or the rejection) together with the post-transition + /// state. The caller executes the effects *outside* the lock. The lock holds + /// only the pure `on_event` computation; no guard escapes. + pub(crate) fn apply_slow( + &self, + event: SessionEvent, + ) -> (Result, SessionState) { + let mut inner = lock(&self.inner); + let result = inner.fsm.on_event(event); + let post_state = inner.fsm.state(); + (result, post_state) + } + + /// Seals the loop closed under the lock before it takes a terminal path. + /// Idempotent; once closed it never reopens. A later slice's fast-path + /// predicate reads `open` to refuse committing once the loop is terminating. + pub(crate) fn seal_closed(&self) { + lock(&self.inner).open = false; + } + /// Registers a standalone "operation" permit not tied to a single message. /// Used to bridge a multi-message atomic operation (e.g. the redirect pair /// `PrepareRedirect` + `Redirect`): held across both sends so `pending` @@ -102,7 +143,7 @@ impl Gate { #[derive(Debug)] #[must_use = "a GatePermit must travel with its message; dropping it early frees the gate slot"] pub struct GatePermit { - inner: Arc>, + inner: Arc>, } impl Drop for GatePermit { @@ -211,6 +252,54 @@ mod tests { type R = Result<(), Box>; + #[test] + fn apply_slow_and_state_snapshot_drive_the_fsm_like_a_standalone() { + // The relocated FSM, driven through the core, must transition and emit + // effects identically to a standalone `SessionFsm` on the same events. + let core = Gate::new(); + let mut reference = SessionFsm::new(); + assert_eq!(core.state_snapshot(), reference.state()); + + let script = [ + SessionEvent::ConnectionAccepted, + SessionEvent::ClientHandshakeResponse, + SessionEvent::BackendGreetingReceived, + SessionEvent::BackendAuthOk, + SessionEvent::ClientCommand, + SessionEvent::BackendResponseTxnDone, + ]; + for event in script { + let (result, post_state) = core.apply_slow(event); + let expected = reference.on_event(event); + assert_eq!( + result, expected, + "core effects/rejection match the standalone FSM for {event:?}" + ); + assert_eq!( + post_state, + reference.state(), + "core post-state matches the standalone FSM for {event:?}" + ); + assert_eq!( + core.state_snapshot(), + reference.state(), + "a later snapshot still matches for {event:?}" + ); + } + } + + #[test] + fn seal_closed_is_sticky_and_idempotent() { + let core = Gate::new(); + assert!(core.is_open(), "a fresh core is open"); + core.seal_closed(); + assert!(!core.is_open(), "seal_closed closes the core"); + // The loop may reach several terminal checks; sealing again is a no-op + // and never reopens. + core.seal_closed(); + assert!(!core.is_open(), "once closed the core never reopens"); + } + #[tokio::test] async fn pending_visible_before_apply_and_cleared_after_drop() -> R { let gate = Gate::new(); diff --git a/rust/crates/dataplane/src/session.rs b/rust/crates/dataplane/src/session.rs index d410d2fb4..521c77c35 100644 --- a/rust/crates/dataplane/src/session.rs +++ b/rust/crates/dataplane/src/session.rs @@ -61,7 +61,7 @@ use std::pin::Pin; use std::time::Duration; -use session_core::fsm::{SessionEffect, SessionEvent, SessionFsm, SessionState, TransitionError}; +use session_core::fsm::{SessionEffect, SessionEvent, SessionState, TransitionError}; use tokio::sync::{mpsc, oneshot, watch}; use tokio::task::{JoinHandle, JoinSet}; use tokio::time::{Instant, Sleep, sleep_until, timeout_at}; @@ -248,7 +248,11 @@ const EVENT_PUMP_CAPACITY: usize = 1; /// The single owner of one session's mutable state. pub struct SessionLoop { - fsm: SessionFsm, + /// The shared session core (FSM + admission gate). The loop is the FSM's + /// only writer on the slow path, via [`crate::gate::Gate::apply_slow`]; a + /// later slice lets the engine commit whitelisted steady transitions under + /// the same lock. + core: crate::gate::Gate, source: Option, handler: E, control: crate::gate::GatedReceiver, @@ -322,9 +326,12 @@ impl Drop for AbortOnDrop { } impl SessionLoop { - /// Creates a session loop; the FSM starts at `Accept`. + /// Creates a session loop over the shared `core` (which carries the FSM, + /// starting at `Accept`). The loop is the FSM's only writer on the slow + /// path. #[must_use] pub fn new( + core: crate::gate::Gate, source: S, handler: E, control: crate::gate::GatedReceiver, @@ -332,7 +339,7 @@ impl SessionLoop { config: SessionLoopConfig, ) -> Self { Self { - fsm: SessionFsm::new(), + core, source: Some(source), handler, control, @@ -434,7 +441,7 @@ impl SessionLoop { let cleanup = self.cleanup_within(cleanup_by).await; SessionSummary { end, - final_state: self.fsm.state(), + final_state: self.core.state_snapshot(), effects_executed: self.effects_executed, rejected_events: self.rejected_events, control_detached: self.control_detached, @@ -442,6 +449,15 @@ impl SessionLoop { } } + /// Terminal fence: seals the core closed — so a later slice's engine fast + /// path refuses to commit once the loop is terminating — then returns the + /// loop's end reason. Called at every path that leaves the steady loop, + /// before the FSM is still `Ready` but the session is tearing down. + fn seal_and_end(&self, end: LoopEnd) -> LoopEnd { + self.core.seal_closed(); + end + } + async fn event_loop(&mut self, events: &mut LoopEventRx) -> LoopEnd { let handshake_deadline = Instant::now() + self.config.handshake_deadline; let mut armed_deadline: Option<(Instant, SessionEvent)> = @@ -480,10 +496,10 @@ impl SessionLoop { // redirect owns the command boundary (Go's processLock). let mut queued_command = None; loop { - if self.fsm.state() == SessionState::Closed { - return LoopEnd::FsmClosed; + if self.core.state_snapshot() == SessionState::Closed { + return self.seal_and_end(LoopEnd::FsmClosed); } - if self.fsm.state() == SessionState::Closing { + if self.core.state_snapshot() == SessionState::Closing { // The loop is the runtime: teardown effects have executed // and their children are tracked, so seal the FSM now; // the children drain in the terminal cleanup, under the @@ -515,8 +531,8 @@ impl SessionLoop { ) .await; match action { - LoopAction::ServerShutdown => return LoopEnd::Shutdown, - LoopAction::SourceExhausted => return LoopEnd::SourceExhausted, + LoopAction::ServerShutdown => return self.seal_and_end(LoopEnd::Shutdown), + LoopAction::SourceExhausted => return self.seal_and_end(LoopEnd::SourceExhausted), LoopAction::ControlDetached => { // Control-v1 last-good: losing the control channel never // tears down an established session. Redirect/drain @@ -528,7 +544,14 @@ impl SessionLoop { if let Some(probe) = next_probe.as_mut() { *probe = Instant::now() + self.config.backend_check_interval; } - if probe_safe(self.fsm.state()) && !self.handler.backend_active().await { + // Snapshot the state out of the lock *before* the await: a + // `self.core.state_snapshot()` sub-expression would otherwise + // keep the guard alive across `backend_active().await`. + // S2b: this read must be re-snapshotted after the await + // before acting, once the engine can commit concurrently — + // it is not commit authority. + let probe_state = self.core.state_snapshot(); + if probe_safe(probe_state) && !self.handler.backend_active().await { self.apply(SessionEvent::BackendIoError, &mut armed_deadline, None) .await; } @@ -538,7 +561,7 @@ impl SessionLoop { self.apply(event, &mut armed_deadline, None).await; } LoopAction::Event(event, permit) => { - if self.fsm.state() == SessionState::RedirectPending + if self.core.state_snapshot() == SessionState::RedirectPending && matches!( event, SessionEvent::ClientCommand | SessionEvent::ClientCommandQuit @@ -567,7 +590,7 @@ impl SessionLoop { // (any effect it spawned already holds its own cmd permit). } } - if self.fsm.state() == SessionState::Ready + if self.core.state_snapshot() == SessionState::Ready && let Some((event, permit)) = queued_command.take() { // apply() enqueues all terminal redirect effects first, so the @@ -639,7 +662,11 @@ impl SessionLoop { armed_deadline: &mut Option<(Instant, SessionEvent)>, drain_deadline: Option, ) { - match self.fsm.on_event(event) { + // The transition runs under the core lock and returns owned effects plus + // the post-transition state; the effects then execute *outside* the lock + // (the std mutex guards only pure computation). + let (result, post_state) = self.core.apply_slow(event); + match result { Ok(effects) => { for effect in effects { if effect == SessionEffect::BeginDrainTimer { @@ -654,10 +681,12 @@ impl SessionLoop { // The handshake deadline is judged on the post-transition // state: the very transition into an authenticated state // (`BackendAuthOk`) disarms it, with no dependency on any - // later event arriving. + // later event arriving. The post-state captured with the + // transition is authoritative here (the loop is the only + // writer of these non-steady transitions). if let Some((_, pending)) = *armed_deadline && pending == SessionEvent::HandshakeTimerExpired - && authenticated_phase(self.fsm.state()) + && authenticated_phase(post_state) { *armed_deadline = None; } @@ -674,11 +703,13 @@ impl SessionLoop { /// the machine. Rejections are tolerated (the FSM may already be /// closing or closed). async fn close_via_fsm(&mut self, close_event: SessionEvent) { + // Reached only from the terminal path in `run`, after the core is + // sealed closed; the loop is the sole writer here. let mut deadline = None; - if self.fsm.state() != SessionState::Closed { + if self.core.state_snapshot() != SessionState::Closed { self.apply(close_event, &mut deadline, None).await; } - if self.fsm.state() != SessionState::Closed { + if self.core.state_snapshot() != SessionState::Closed { self.apply(SessionEvent::TeardownComplete, &mut deadline, None) .await; } diff --git a/rust/crates/dataplane/src/session_engine.rs b/rust/crates/dataplane/src/session_engine.rs index 344789bcc..58b802cfc 100644 --- a/rust/crates/dataplane/src/session_engine.rs +++ b/rust/crates/dataplane/src/session_engine.rs @@ -840,6 +840,7 @@ async fn run_bound_session_observed( // fires — not only when the loop finishes its own cleanup. let mut owner_shutdown = shutdown.clone(); let session_loop = SessionLoop::new( + gate.clone(), EventRx { events: event_rx }, CmdTx { cmds: cmd_tx.clone(), diff --git a/rust/crates/dataplane/tests/session_loop.rs b/rust/crates/dataplane/tests/session_loop.rs index f6d7eefcf..b9506c989 100644 --- a/rust/crates/dataplane/tests/session_loop.rs +++ b/rust/crates/dataplane/tests/session_loop.rs @@ -276,14 +276,19 @@ impl EffectHandler for SpinUntilDropHandler { } fn channels() -> ( + dataplane::gate::Gate, dataplane::gate::GatedSender, dataplane::gate::GatedReceiver, watch::Sender, watch::Receiver, ) { - let (control_tx, control_rx) = dataplane::gate::channel_with_new_gate(8); + // One shared core per session, as production wires it: the control channel + // and the loop's FSM core are the same `Gate`, so control messages register + // on the core the engine's fast path reads. + let gate = dataplane::gate::Gate::new(); + let (control_tx, control_rx) = dataplane::gate::channel(gate.clone(), 8); let (shutdown_tx, shutdown_rx) = watch::channel(false); - (control_tx, control_rx, shutdown_tx, shutdown_rx) + (gate, control_tx, control_rx, shutdown_tx, shutdown_rx) } const HANDSHAKE: [SessionEvent; 4] = [ @@ -318,10 +323,11 @@ async fn scripted_lifecycle_completes_cleanly() { SessionEvent::ClientCommandQuit, SessionEvent::TeardownComplete, ]); - let (_control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, _control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); let handler = Recorder::default(); let effects = handler.effects(); let summary = SessionLoop::new( + fsm_core, FakeSource::scripted(&events), handler, control_rx, @@ -361,13 +367,14 @@ async fn scripted_lifecycle_completes_cleanly() { /// default. #[tokio::test(start_paused = true)] async fn coordinated_drain_uses_latest_deadline_override() { - let (control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); let mut events = HANDSHAKE.to_vec(); events.extend([ SessionEvent::ClientCommand, SessionEvent::BackendResponseTxnOpen, ]); let looped = SessionLoop::new( + fsm_core, FakeSource::parking(&events), Recorder::default(), control_rx, @@ -405,12 +412,13 @@ async fn eof_aborts_stuck_children_after_drain_window() { let mut events = HANDSHAKE.to_vec(); events.push(SessionEvent::ClientCommand); // Transport ends abruptly afterwards (client vanished). - let (_control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, _control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); let handler = Recorder { spawn_stuck_child_on_forward: true, ..Recorder::default() }; let summary = SessionLoop::new( + fsm_core, FakeSource::scripted(&events), handler, control_rx, @@ -438,13 +446,14 @@ async fn teardown_children_drain_to_completion() { events.push(SessionEvent::ClientCommandQuit); events.push(SessionEvent::TeardownComplete); let finished = Arc::new(AtomicBool::new(false)); - let (_control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, _control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); let handler = Recorder { // Inside the 5s cleanup deadline, but requires real draining. spawn_slow_teardown_child: Some((Duration::from_secs(2), Arc::clone(&finished))), ..Recorder::default() }; let summary = SessionLoop::new( + fsm_core, FakeSource::scripted(&events), handler, control_rx, @@ -489,10 +498,11 @@ async fn redirect_command_shutdown_interleavings_never_deadlock() { ]; for permutation in permutations { let (event_tx, source) = ChannelSource::new(); - let (control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); let handler = Recorder::default(); let effects = handler.effects(); let looped = SessionLoop::new( + fsm_core, source, handler, control_rx, @@ -557,7 +567,7 @@ async fn redirect_command_shutdown_interleavings_never_deadlock() { #[tokio::test(start_paused = true)] async fn pending_stimuli_shutdown_precheck_dominates() { let (event_tx, reads, source) = CountingSource::new(); - let (control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); let gate = Arc::new(tokio::sync::Semaphore::new(0)); let handler = Recorder { forward_gate: Some(Arc::clone(&gate)), @@ -565,6 +575,7 @@ async fn pending_stimuli_shutdown_precheck_dominates() { }; let effects = handler.effects(); let looped = SessionLoop::new( + fsm_core, source, handler, control_rx, @@ -616,7 +627,7 @@ async fn pending_stimuli_shutdown_precheck_dominates() { #[tokio::test(start_paused = true)] async fn three_select_arms_race_cleanly() { let (event_tx, reads, source) = CountingSource::new(); - let (control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); let gate = Arc::new(tokio::sync::Semaphore::new(0)); let child_done = Arc::new(AtomicBool::new(false)); let handler = Recorder { @@ -626,6 +637,7 @@ async fn three_select_arms_race_cleanly() { }; let effects = handler.effects(); let looped = SessionLoop::new( + fsm_core, source, handler, control_rx, @@ -695,7 +707,7 @@ async fn three_select_arms_race_cleanly() { #[tokio::test(start_paused = true)] async fn classifier_reads_at_most_one_ahead() { let (event_tx, reads, source) = CountingSource::new(); - let (_control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, _control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); let gate = Arc::new(tokio::sync::Semaphore::new(0)); let handler = Recorder { forward_gate: Some(Arc::clone(&gate)), @@ -703,6 +715,7 @@ async fn classifier_reads_at_most_one_ahead() { }; let effects = handler.effects(); let looped = SessionLoop::new( + fsm_core, source, handler, control_rx, @@ -771,7 +784,7 @@ async fn terminal_cleanup_budget_and_source_release_order() { inner: FakeSource::parking(&events), dropped: Arc::clone(&source_dropped), }; - let (_control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, _control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); let handler = SpinUntilDropHandler { inner: Recorder::default(), source_dropped: Arc::clone(&source_dropped), @@ -779,6 +792,7 @@ async fn terminal_cleanup_budget_and_source_release_order() { }; let start = tokio::time::Instant::now(); let summary = SessionLoop::new( + fsm_core, source, handler, control_rx, @@ -805,13 +819,14 @@ async fn terminal_cleanup_budget_and_source_release_order() { // cleanup_deadline. let mut events = HANDSHAKE.to_vec(); events.push(SessionEvent::ClientCommand); - let (_control_tx2, control_rx, _shutdown_tx2, shutdown_rx) = channels(); + let (fsm_core, _control_tx2, control_rx, _shutdown_tx2, shutdown_rx) = channels(); let handler = Recorder { spawn_stuck_child_on_forward: true, ..Recorder::default() }; let start = tokio::time::Instant::now(); let summary = SessionLoop::new( + fsm_core, FakeSource::scripted(&events), handler, control_rx, @@ -833,10 +848,11 @@ async fn terminal_cleanup_budget_and_source_release_order() { /// seals the FSM. #[tokio::test(start_paused = true)] async fn server_shutdown_closes_idle_session() { - let (_control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, _control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); let handler = Recorder::default(); let effects = handler.effects(); let looped = SessionLoop::new( + fsm_core, FakeSource::parking(&HANDSHAKE), handler, control_rx, @@ -863,11 +879,12 @@ async fn server_shutdown_closes_idle_session() { /// comes. #[tokio::test(start_paused = true)] async fn preexisting_shutdown_is_not_missed() { - let (_control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, _control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); let _ = shutdown_tx.send(true); let handler = Recorder::default(); let effects = handler.effects(); let summary = SessionLoop::new( + fsm_core, FakeSource::parking(&[]), handler, control_rx, @@ -890,8 +907,9 @@ async fn preexisting_shutdown_is_not_missed() { #[tokio::test(start_paused = true)] async fn handshake_deadline_only_before_authentication() { // Stalled pre-auth session: the deadline closes it. - let (_control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, _control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); let summary = SessionLoop::new( + fsm_core, FakeSource::parking(&[SessionEvent::ConnectionAccepted]), Recorder::default(), control_rx, @@ -910,8 +928,9 @@ async fn handshake_deadline_only_before_authentication() { backend_check_interval: Duration::ZERO, ..SessionLoopConfig::default() }; - let (_control_tx2, control_rx, shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, _control_tx2, control_rx, shutdown_tx, shutdown_rx) = channels(); let looped = SessionLoop::new( + fsm_core, FakeSource::parking(&HANDSHAKE), Recorder::default(), control_rx, @@ -936,7 +955,7 @@ async fn handshake_deadline_only_before_authentication() { /// down through the FSM. #[tokio::test(start_paused = true)] async fn dead_backend_probe_closes_session() { - let (_control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, _control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); let handler = Recorder::default(); match handler.backend_alive.lock() { Ok(mut alive) => { @@ -951,6 +970,7 @@ async fn dead_backend_probe_closes_session() { } let effects = handler.effects(); let summary = SessionLoop::new( + fsm_core, FakeSource::parking(&HANDSHAKE), handler, control_rx, @@ -975,14 +995,14 @@ async fn dead_backend_probe_closes_session() { #[tokio::test(start_paused = true)] async fn probe_skips_in_flight_commands() { let (event_tx, source) = ChannelSource::new(); - let (_control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, _control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); let handler = Recorder::default(); let probes = Arc::clone(&handler.probe_calls); let config = SessionLoopConfig { backend_check_interval: Duration::from_secs(1), ..SessionLoopConfig::default() }; - let looped = SessionLoop::new(source, handler, control_rx, shutdown_rx, config); + let looped = SessionLoop::new(fsm_core, source, handler, control_rx, shutdown_rx, config); let run = tokio::spawn(looped.run()); for event in HANDSHAKE { let _ = event_tx.send(event).await; @@ -1030,10 +1050,11 @@ async fn probe_skips_in_flight_commands() { #[tokio::test(start_paused = true)] async fn control_detach_keeps_session_serving() { let (event_tx, source) = ChannelSource::new(); - let (control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); let handler = Recorder::default(); let effects = handler.effects(); let looped = SessionLoop::new( + fsm_core, source, handler, control_rx, @@ -1085,10 +1106,11 @@ async fn multi_await_classifier_survives_select_noise() { halves, completed: Arc::clone(&completed), }; - let (_control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, _control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); let handler = Recorder::default(); let effects = handler.effects(); let looped = SessionLoop::new( + fsm_core, source, handler, control_rx, @@ -1153,11 +1175,12 @@ async fn next_command_waits_for_redirect_after_commit() { SessionEvent::ControlCloseImmediate, ] { let (event_tx, source) = ChannelSource::new(); - let (control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, control_tx, control_rx, _shutdown_tx, shutdown_rx) = channels(); let handler = Recorder::default(); let effects = handler.effects(); let run = tokio::spawn( SessionLoop::new( + fsm_core, source, handler, control_rx, @@ -1314,7 +1337,7 @@ impl EffectHandler for OrderRecorder { #[tokio::test(start_paused = true)] async fn due_probe_is_selected_before_a_queued_command_on_the_direct_channel() { let (event_tx, source) = DirectChannelSource::new(); - let (_control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); + let (fsm_core, _control_tx, control_rx, shutdown_tx, shutdown_rx) = channels(); let gate = Arc::new(tokio::sync::Semaphore::new(0)); let handler = OrderRecorder { marks: Arc::new(Mutex::new(Vec::new())), @@ -1326,8 +1349,9 @@ async fn due_probe_is_selected_before_a_queued_command_on_the_direct_channel() { backend_check_interval: Duration::from_secs(1), ..SessionLoopConfig::default() }; - let run = - tokio::spawn(SessionLoop::new(source, handler, control_rx, shutdown_rx, config).run()); + let run = tokio::spawn( + SessionLoop::new(fsm_core, source, handler, control_rx, shutdown_rx, config).run(), + ); for event in HANDSHAKE { let _ = event_tx.send(event).await; }