fix: resolve deadlock when both sides send resendrequest simultaneously (#314)

* Bypass sequence number check when handling resend request

* Remove magic strings for admin message types

* Remove magic values for message type in session states

* Add test case for receiving heartbeats

* Add test for rejects

* Attempt to fix flaky tests due to timestamp precision

Author: davidsteiner

crates/hotfix/src/lib.rs

@@ -25,7 +25,6 @@ pub mod application;
 pub mod config;
 pub mod initiator;
 pub mod message;
-pub mod message_utils;
 pub mod session;
 mod session_schedule;
 pub mod store;

crates/hotfix/src/message.rs

@@ -2,7 +2,9 @@
 use hotfix_message::error::EncodingError as EncodeError;
 pub use hotfix_message::field_types::Timestamp;
 pub(crate) use hotfix_message::message::{Config, Message};
-use hotfix_message::session_fields::{MSG_SEQ_NUM, SENDER_COMP_ID, SENDING_TIME, TARGET_COMP_ID};
+use hotfix_message::session_fields::{
+    MSG_SEQ_NUM, ORIG_SENDING_TIME, POSS_DUP_FLAG, SENDER_COMP_ID, SENDING_TIME, TARGET_COMP_ID,
+};
 pub use hotfix_message::{Part, RepeatingGroup};
 
 pub mod business_reject;
@@ -20,12 +22,75 @@ pub mod verification_error;
 pub use parser::RawFixMessage;
 pub use resend_request::ResendRequest;
 
+use heartbeat::Heartbeat;
+use logon::Logon;
+use logout::Logout;
+use reject::Reject;
+use sequence_reset::SequenceReset;
+use test_request::TestRequest;
+
+static ADMIN_TYPES: [&str; 7] = [
+    Logon::MSG_TYPE,
+    Heartbeat::MSG_TYPE,
+    TestRequest::MSG_TYPE,
+    ResendRequest::MSG_TYPE,
+    Reject::MSG_TYPE,
+    SequenceReset::MSG_TYPE,
+    Logout::MSG_TYPE,
+];
+
+pub fn is_admin(message_type: &str) -> bool {
+    ADMIN_TYPES.contains(&message_type)
+}
+
 pub trait OutboundMessage: Clone + Send + 'static {
     fn write(&self, msg: &mut Message);
 
     fn message_type(&self) -> &str;
 }
 
+/// Prepares a FIX message for resend per the FIX spec (PossDupFlag logic).
+///
+/// Behaviour:
+/// - On first resend (no PossDupFlag Y / no OrigSendingTime):
+///   * Move current SendingTime(52) to OrigSendingTime(122)
+///   * Set SendingTime(52) to the current sending time (may be equal if clock granularity causes no change)
+///   * Set PossDupFlag(43)=Y
+/// - On subsequent resends (already marked possible duplicate and has OrigSendingTime):
+///   * Refresh SendingTime(52) to current time (value may or may not differ from previous)
+pub fn prepare_message_for_resend(msg: &mut Message) -> Result<(), &'static str> {
+    let header = msg.header_mut();
+
+    if header.get_raw(SENDING_TIME).is_none() {
+        return Err("Missing SendingTime (52)");
+    }
+
+    let already_poss_dup = header.get::<bool>(POSS_DUP_FLAG).unwrap_or(false);
+    let has_orig_sending_time = header.get_raw(ORIG_SENDING_TIME).is_some();
+
+    if already_poss_dup && has_orig_sending_time {
+        // Subsequent resend: refresh SendingTime only
+        return if header.pop(SENDING_TIME).is_some() {
+            header.set(SENDING_TIME, Timestamp::utc_now());
+            Ok(())
+        } else {
+            Err("Failed to extract previous SendingTime")
+        };
+    }
+
+    // First resend path
+    if let Some(original_sending_time_field) = header.pop(SENDING_TIME) {
+        let original_ts = Timestamp::parse(&original_sending_time_field.data)
+            .ok_or("Invalid original SendingTime format")?;
+        header.set(ORIG_SENDING_TIME, original_ts);
+        header.set(SENDING_TIME, Timestamp::utc_now());
+        header.set(POSS_DUP_FLAG, true);
+        Ok(())
+    } else {
+        Err("Failed to extract original SendingTime")
+    }
+}
+
 pub fn generate_message(
     begin_string: &str,
     sender_comp_id: &str,

crates/hotfix/src/message/business_reject.rs

@@ -49,6 +49,8 @@ pub(crate) struct BusinessReject {
 }
 
 impl BusinessReject {
+    pub(crate) const MSG_TYPE: &str = "j";
+
     pub(crate) fn new(ref_msg_type: &str, reason: BusinessRejectReason) -> Self {
         Self {
             ref_msg_type: ref_msg_type.to_string(),
@@ -100,7 +102,7 @@ impl OutboundMessage for BusinessReject {
     }
 
     fn message_type(&self) -> &str {
-        "j"
+        Self::MSG_TYPE
     }
 }
 

crates/hotfix/src/message/heartbeat.rs

@@ -9,6 +9,8 @@ pub struct Heartbeat {
 }
 
 impl Heartbeat {
+    pub const MSG_TYPE: &str = "0";
+
     pub fn for_request(test_req_id: String) -> Self {
         Self {
             test_req_id: Some(test_req_id),
@@ -24,6 +26,6 @@ impl OutboundMessage for Heartbeat {
     }
 
     fn message_type(&self) -> &str {
-        "0"
+        Self::MSG_TYPE
     }
 }

crates/hotfix/src/message/logon.rs

@@ -19,6 +19,8 @@ pub enum ResetSeqNumConfig {
 }
 
 impl Logon {
+    pub const MSG_TYPE: &str = "A";
+
     pub fn new(heartbeat_interval: u64, reset_config: ResetSeqNumConfig) -> Self {
         let (reset_seq_num_flag, next_expected_msg_seq_num) = match reset_config {
             ResetSeqNumConfig::Reset => (ResetSeqNumFlag::Yes, None),
@@ -45,7 +47,7 @@ impl OutboundMessage for Logon {
     }
 
     fn message_type(&self) -> &str {
-        "A"
+        Self::MSG_TYPE
     }
 }
 

crates/hotfix/src/message/logout.rs

@@ -9,6 +9,8 @@ pub struct Logout {
 }
 
 impl Logout {
+    pub const MSG_TYPE: &str = "5";
+
     pub fn with_reason(reason: String) -> Self {
         Self { text: Some(reason) }
     }
@@ -22,6 +24,6 @@ impl OutboundMessage for Logout {
     }
 
     fn message_type(&self) -> &str {
-        "5"
+        Self::MSG_TYPE
     }
 }

crates/hotfix/src/message/reject.rs

@@ -16,6 +16,8 @@ pub(crate) struct Reject {
 }
 
 impl Reject {
+    pub(crate) const MSG_TYPE: &str = "3";
+
     pub(crate) fn new(ref_seq_num: u64) -> Self {
         Self {
             ref_seq_num,
@@ -85,7 +87,7 @@ impl OutboundMessage for Reject {
     }
 
     fn message_type(&self) -> &str {
-        "3"
+        Self::MSG_TYPE
     }
 }
 

crates/hotfix/src/message/resend_request.rs

@@ -10,6 +10,8 @@ pub struct ResendRequest {
 }
 
 impl ResendRequest {
+    pub const MSG_TYPE: &str = "2";
+
     pub fn new(begin: u64, end: u64) -> Self {
         Self {
             begin_seq_no: begin,
@@ -25,6 +27,6 @@ impl OutboundMessage for ResendRequest {
     }
 
     fn message_type(&self) -> &str {
-        "2"
+        Self::MSG_TYPE
     }
 }

crates/hotfix/src/message/sequence_reset.rs

@@ -12,6 +12,10 @@ pub struct SequenceReset {
     pub new_seq_no: u64,
 }
 
+impl SequenceReset {
+    pub const MSG_TYPE: &str = "4";
+}
+
 impl OutboundMessage for SequenceReset {
     fn write(&self, msg: &mut Message) {
         msg.set(GAP_FILL_FLAG, self.gap_fill);
@@ -25,6 +29,6 @@ impl OutboundMessage for SequenceReset {
     }
 
     fn message_type(&self) -> &str {
-        "4"
+        Self::MSG_TYPE
     }
 }

crates/hotfix/src/message/test_request.rs

@@ -9,6 +9,8 @@ pub struct TestRequest {
 }
 
 impl TestRequest {
+    pub const MSG_TYPE: &str = "1";
+
     pub fn new(test_req_id: String) -> Self {
         Self { test_req_id }
     }
@@ -20,6 +22,6 @@ impl OutboundMessage for TestRequest {
     }
 
     fn message_type(&self) -> &str {
-        "1"
+        Self::MSG_TYPE
     }
 }