Move resend completion into AwaitingResend state variant

Author: davidsteiner

crates/hotfix/src/session.rs

@@ -277,46 +277,42 @@ where
     }
 
     async fn check_end_of_resend(&mut self) -> Result<(), SessionOperationError> {
-        let backlog = if let SessionState::AwaitingResend(state) = &mut self.state {
-            if self.ctx.store.next_target_seq_number() > state.end_seq_number {
-                let inbound_queue = std::mem::take(&mut state.inbound_queue);
-                let new_state = SessionState::new_active(
-                    state.writer.clone(),
-                    self.ctx.config.heartbeat_interval,
-                );
-                self.apply_transition(TransitionResult::TransitionTo(new_state))
-                    .await;
-                Some(inbound_queue)
-            } else {
-                None
-            }
+        let completed = if let SessionState::AwaitingResend(state) = &mut self.state {
+            state.try_complete(
+                self.ctx.store.next_target_seq_number(),
+                self.ctx.config.heartbeat_interval,
+            )
         } else {
             None
         };
 
-        if let Some(mut inbound_queue) = backlog {
-            // we have reached the end of the resend,
-            // process queued messages and resume normal operation
-            debug!("resend is done, processing backlog");
-            while let Some(msg) = inbound_queue.pop_front() {
-                let seq_number: u64 = msg.get(MSG_SEQ_NUM).unwrap_or_else(|e| {
-                    error!("failed to get seq number: {:?}", e);
-                    0
-                });
-                let msg_type: &str = msg.header().get(MSG_TYPE).unwrap_or("");
-                debug!(seq_number, msg_type, "processing queued message");
-
-                if msg_type == ResendRequest::MSG_TYPE {
-                    // ResendRequest was already processed when it arrived (it bypasses
-                    // the queue in process_message). Just increment the target seq number
-                    // for sequence accounting purposes.
-                    self.ctx.store.increment_target_seq_number().await?;
-                } else {
-                    self.process_message(msg).await?;
-                }
+        let Some((new_state, mut backlog)) = completed else {
+            return Ok(());
+        };
+
+        self.apply_transition(TransitionResult::TransitionTo(new_state))
+            .await;
+
+        // Process queued messages and resume normal operation
+        debug!("resend is done, processing backlog");
+        while let Some(msg) = backlog.pop_front() {
+            let seq_number: u64 = msg.get(MSG_SEQ_NUM).unwrap_or_else(|e| {
+                error!("failed to get seq number: {:?}", e);
+                0
+            });
+            let msg_type: &str = msg.header().get(MSG_TYPE).unwrap_or("");
+            debug!(seq_number, msg_type, "processing queued message");
+
+            if msg_type == ResendRequest::MSG_TYPE {
+                // ResendRequest was already processed when it arrived (it bypasses
+                // the queue in process_message). Just increment the target seq number
+                // for sequence accounting purposes.
+                self.ctx.store.increment_target_seq_number().await?;
+            } else {
+                self.process_message(msg).await?;
             }
-            debug!("resend backlog is cleared, resuming normal operation");
         }
+        debug!("resend backlog is cleared, resuming normal operation");
 
         Ok(())
     }

crates/hotfix/src/session/ctx.rs

@@ -20,6 +20,7 @@ pub(crate) enum TransitionResult {
 ///
 /// Before verification and dispatch, the current state gets a chance to
 /// decide whether a message should be processed, queued, or rejected.
+#[allow(clippy::large_enum_variant)]
 pub(crate) enum PreProcessDecision {
     /// Continue processing this message through verification and dispatch.
     Accept(Message),

crates/hotfix/src/session/state/awaiting_resend.rs

@@ -42,6 +42,23 @@ impl AwaitingResendState {
         }
     }
 
+    /// Check whether the resend is complete. If the next expected target sequence number
+    /// exceeds the end of the gap, return the queued backlog for replay and transition
+    /// to Active. Otherwise return `None`.
+    pub(crate) fn try_complete(
+        &mut self,
+        next_target_seq: u64,
+        heartbeat_interval: u64,
+    ) -> Option<(SessionState, VecDeque<Message>)> {
+        if next_target_seq > self.end_seq_number {
+            let backlog = std::mem::take(&mut self.inbound_queue);
+            let new_state = SessionState::new_active(self.writer.clone(), heartbeat_interval);
+            Some((new_state, backlog))
+        } else {
+            None
+        }
+    }
+
     pub(crate) fn pre_process_inbound(&mut self, message: Message) -> PreProcessDecision {
         let dominated_by_resend = message
             .header()