Introduce pre_process_inbound function to delegate pre-processing decisions to state variants

Author: davidsteiner

crates/hotfix/src/session.rs

@@ -36,7 +36,7 @@ use crate::message::sequence_reset::SequenceReset;
 use crate::message::test_request::TestRequest;
 use crate::message::verification::VerificationFlags;
 use crate::session::admin_request::AdminRequest;
-use crate::session::ctx::{SessionCtx, TransitionResult, VerificationResult};
+use crate::session::ctx::{PreProcessDecision, SessionCtx, TransitionResult, VerificationResult};
 use crate::session::error::SessionCreationError;
 use crate::session::error::{InternalSendError, InternalSendResultExt, SessionOperationError};
 pub use crate::session::error::{SendError, SendOutcome};
@@ -194,29 +194,20 @@ where
     }
 
     async fn process_message(&mut self, message: Message) -> Result<(), SessionOperationError> {
+        let message = match self.state.pre_process_inbound(message) {
+            PreProcessDecision::Accept(msg) => msg,
+            PreProcessDecision::Queued => return Ok(()),
+            PreProcessDecision::Disconnect => {
+                self.state.disconnect_writer().await;
+                return Ok(());
+            }
+        };
+
         let message_type: &str = message
             .header()
             .get(MSG_TYPE)
             .map_err(|_| SessionOperationError::MissingField("MSG_TYPE"))?;
 
-        if let SessionState::AwaitingResend(state) = &mut self.state {
-            let seq_number = get_msg_seq_num(&message);
-            if seq_number > state.end_seq_number && message_type != ResendRequest::MSG_TYPE {
-                state.inbound_queue.push_back(message);
-                return Ok(());
-            }
-        }
-
-        // TODO: add state-level pre-process check that validates whether the message type
-        // is acceptable in the current state (e.g. AwaitingLogon rejects non-Logon,
-        // unexpected Logon in Active should be rejected per FIX spec).
-        if let SessionState::AwaitingLogon(_) = &mut self.state
-            && message_type != Logon::MSG_TYPE
-        {
-            self.state.disconnect_writer().await;
-            return Ok(());
-        }
-
         let flags = VerificationFlags::for_message(&message)?;
         if let VerificationResult::Issue(result) = self
             .state

crates/hotfix/src/session/ctx.rs

@@ -9,12 +9,26 @@ use crate::message::parser::RawFixMessage;
 use crate::session::error::InternalSendError;
 use crate::session::state::SessionState;
 use crate::store::StoreError;
+use hotfix_message::message::Message;
 
 pub(crate) enum TransitionResult {
     Stay,
     TransitionTo(SessionState),
 }
 
+/// The result of the pre-processing step for an inbound message.
+///
+/// Before verification and dispatch, the current state gets a chance to
+/// decide whether a message should be processed, queued, or rejected.
+pub(crate) enum PreProcessDecision {
+    /// Continue processing this message through verification and dispatch.
+    Accept(Message),
+    /// The message has been queued for later processing (e.g. AwaitingResend backlog).
+    Queued,
+    /// The message is not acceptable in this state — disconnect the writer.
+    Disconnect,
+}
+
 /// The result of verifying an inbound message via a state variant's
 /// `handle_verification_issue` method.
 pub(crate) enum VerificationResult {

crates/hotfix/src/session/state.rs

@@ -15,7 +15,7 @@ use crate::message::OutboundMessage;
 use crate::message::logon::Logon;
 use crate::message::logout::Logout;
 use crate::message::verification::VerificationFlags;
-use crate::session::ctx::{SessionCtx, TransitionResult, VerificationResult};
+use crate::session::ctx::{PreProcessDecision, SessionCtx, TransitionResult, VerificationResult};
 use crate::session::error::{InternalSendError, InternalSendResultExt, SessionOperationError};
 use crate::session::event::ScheduleResponse;
 use crate::session::info::Status as SessionInfoStatus;
@@ -63,6 +63,16 @@ impl SessionState {
         })
     }
 
+    /// Let the current state decide whether an inbound message should be processed,
+    /// queued for later, or rejected before verification and dispatch.
+    pub(crate) fn pre_process_inbound(&mut self, message: Message) -> PreProcessDecision {
+        match self {
+            Self::AwaitingResend(state) => state.pre_process_inbound(message),
+            Self::AwaitingLogon(state) => state.pre_process_inbound(message),
+            _ => PreProcessDecision::Accept(message),
+        }
+    }
+
     pub fn should_reconnect(&self) -> bool {
         match self {
             SessionState::Disconnected(DisconnectedState { reconnect, .. }) => *reconnect,

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

@@ -1,13 +1,16 @@
 use crate::Application;
+use crate::message::logon::Logon;
 use crate::message::resend_request::ResendRequest;
 use crate::message::verification::VerificationFlags;
-use crate::session::ctx::{SessionCtx, TransitionResult, VerificationResult};
+use crate::session::ctx::{PreProcessDecision, SessionCtx, TransitionResult, VerificationResult};
 use crate::session::error::{InternalSendResultExt, SessionOperationError};
 use crate::session::inbound::{self, VerificationOutcome};
 use crate::session::outbound;
 use crate::session::state::{AwaitingResendState, SessionState};
 use crate::transport::writer::WriterRef;
+use hotfix_message::Part;
 use hotfix_message::message::Message;
+use hotfix_message::session_fields::MSG_TYPE;
 use hotfix_store::MessageStore;
 use tokio::time::Instant;
 use tracing::debug;
@@ -22,6 +25,19 @@ pub(crate) struct AwaitingLogonState {
 }
 
 impl AwaitingLogonState {
+    pub(crate) fn pre_process_inbound(&self, message: Message) -> PreProcessDecision {
+        let is_logon = message
+            .header()
+            .get::<&str>(MSG_TYPE)
+            .is_ok_and(|t| t == Logon::MSG_TYPE);
+
+        if is_logon {
+            PreProcessDecision::Accept(message)
+        } else {
+            PreProcessDecision::Disconnect
+        }
+    }
+
     pub(crate) async fn handle_verification_issue<A: Application, S: MessageStore>(
         &self,
         ctx: &mut SessionCtx<A, S>,

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

@@ -1,13 +1,16 @@
 use crate::Application;
 use crate::message::resend_request::ResendRequest;
 use crate::message::verification::VerificationFlags;
-use crate::session::ctx::{SessionCtx, TransitionResult, VerificationResult};
+use crate::session::ctx::{PreProcessDecision, SessionCtx, TransitionResult, VerificationResult};
 use crate::session::error::{InternalSendResultExt, SessionOperationError};
+use crate::session::get_msg_seq_num;
 use crate::session::inbound::{self, VerificationOutcome};
 use crate::session::outbound;
 use crate::session::state::SessionState;
 use crate::transport::writer::WriterRef;
+use hotfix_message::Part;
 use hotfix_message::message::Message;
+use hotfix_message::session_fields::MSG_TYPE;
 use hotfix_store::MessageStore;
 use std::collections::VecDeque;
 use tracing::debug;
@@ -39,6 +42,20 @@ impl AwaitingResendState {
         }
     }
 
+    pub(crate) fn pre_process_inbound(&mut self, message: Message) -> PreProcessDecision {
+        let dominated_by_resend = message
+            .header()
+            .get::<&str>(MSG_TYPE)
+            .is_ok_and(|t| t != ResendRequest::MSG_TYPE);
+
+        if get_msg_seq_num(&message) > self.end_seq_number && dominated_by_resend {
+            self.inbound_queue.push_back(message);
+            PreProcessDecision::Queued
+        } else {
+            PreProcessDecision::Accept(message)
+        }
+    }
+
     pub(crate) async fn handle_verification_issue<A: Application, S: MessageStore>(
         &mut self,
         ctx: &mut SessionCtx<A, S>,