First cut of refactor to return confirmation for sent app messages

Author: davidsteiner

crates/hotfix/src/initiator.rs

@@ -15,6 +15,7 @@ use tracing::{debug, warn};
 use crate::application::Application;
 use crate::config::SessionConfig;
 use crate::message::{InboundMessage, OutboundMessage};
+use crate::session::error::{SendError, SendOutcome};
 use crate::session::{InternalSessionRef, SessionHandle};
 use crate::store::MessageStore;
 use crate::transport::connect;
@@ -50,6 +51,25 @@ impl<Outbound: OutboundMessage> Initiator<Outbound> {
         Ok(initiator)
     }
 
+    /// Sends a message and waits for confirmation that it was persisted.
+    ///
+    /// Returns `SendOutcome::Sent` with the sequence number if the message was
+    /// successfully persisted and sent, or `SendOutcome::Dropped` if the application
+    /// callback chose to drop the message.
+    pub async fn send(&self, msg: Outbound) -> Result<SendOutcome, SendError> {
+        self.session_handle.send(msg).await
+    }
+
+    /// Sends a message without waiting for confirmation.
+    ///
+    /// This is a fire-and-forget operation. The message will be queued for sending
+    /// but no confirmation is provided about whether it was actually sent.
+    pub async fn send_forget(&self, msg: Outbound) -> Result<(), SendError> {
+        self.session_handle.send_forget(msg).await
+    }
+
+    #[deprecated(since = "0.7.2", note = "use `send` or `send_forget` instead")]
+    #[allow(deprecated)]
     pub async fn send_message(&self, msg: Outbound) -> Result<()> {
         self.session_handle.send_message(msg).await?;
 

crates/hotfix/src/session.rs

@@ -36,12 +36,14 @@ use crate::message::verification::verify_message;
 use crate::message::verification_error::{CompIdType, MessageVerificationError};
 use crate::message_utils::{is_admin, prepare_message_for_resend};
 use crate::session::admin_request::AdminRequest;
+pub use crate::session::error::{SendError, SendOutcome};
 pub use crate::session::info::{SessionInfo, Status};
 pub use crate::session::session_handle::SessionHandle;
 #[cfg(not(feature = "test-utils"))]
 pub(crate) use crate::session::session_ref::InternalSessionRef;
 #[cfg(feature = "test-utils")]
 pub use crate::session::session_ref::InternalSessionRef;
+use crate::session::session_ref::OutboundRequest;
 use crate::session::state::SessionState;
 use crate::session::state::{AwaitingResendTransitionOutcome, TestRequestId};
 use crate::session_schedule::SessionSchedule;
@@ -819,6 +821,66 @@ where
         Ok(())
     }
 
+    async fn send_app_message_with_confirmation(
+        &mut self,
+        message: Outbound,
+    ) -> Result<SendOutcome, SendError> {
+        if !self.state.is_connected() {
+            return Err(SendError::Disconnected);
+        }
+
+        match self.application.on_outbound_message(&message).await {
+            OutboundDecision::Send => {
+                let sequence_number = self.send_message_returning_seq(message).await?;
+                Ok(SendOutcome::Sent { sequence_number })
+            }
+            OutboundDecision::Drop => {
+                debug!("dropped outbound message as instructed by the application");
+                Ok(SendOutcome::Dropped)
+            }
+            OutboundDecision::TerminateSession => {
+                warn!("the application indicated we should terminate the session");
+                self.state.disconnect_writer().await;
+                Err(SendError::SessionTerminated)
+            }
+        }
+    }
+
+    async fn send_message_returning_seq(
+        &mut self,
+        message: impl OutboundMessage,
+    ) -> Result<u64, SendError> {
+        let seq_num = self.store.next_sender_seq_number();
+        let msg_type = message.message_type().as_bytes().to_vec();
+        let msg = generate_message(
+            &self.config.begin_string,
+            &self.config.sender_comp_id,
+            &self.config.target_comp_id,
+            seq_num,
+            message,
+        )
+        .map_err(|e| {
+            SendError::Persist(crate::store::StoreError::PersistMessage {
+                sequence_number: seq_num,
+                source: e.into(),
+            })
+        })?;
+
+        self.store
+            .increment_sender_seq_number()
+            .await
+            .map_err(SendError::SequenceNumber)?;
+
+        self.store
+            .add(seq_num, &msg)
+            .await
+            .map_err(SendError::Persist)?;
+
+        self.send_raw(&msg_type, msg).await;
+
+        Ok(seq_num)
+    }
+
     async fn send_message(&mut self, message: impl OutboundMessage) -> Result<()> {
         let seq_num = self.store.next_sender_seq_number();
         let msg_type = message.message_type().as_bytes().to_vec();
@@ -957,9 +1019,19 @@ where
         }
     }
 
-    async fn handle_outbound_message(&mut self, message: Outbound) {
-        if let Err(err) = self.send_app_message(message).await {
-            error!(err = ?err, "failed to send app message: {err}");
+    async fn handle_outbound_message(&mut self, request: OutboundRequest<Outbound>) {
+        let OutboundRequest { message, confirm } = request;
+        match confirm {
+            Some(tx) => {
+                let result = self.send_app_message_with_confirmation(message).await;
+                // Ignore send errors - receiver may have been dropped
+                let _ = tx.send(result);
+            }
+            None => {
+                if let Err(err) = self.send_app_message(message).await {
+                    error!(err = ?err, "failed to send app message: {err}");
+                }
+            }
         }
     }
 
@@ -1069,7 +1141,7 @@ where
 async fn run_session<App, Inbound, Outbound, Store>(
     mut session: Session<App, Inbound, Outbound, Store>,
     mut event_receiver: mpsc::Receiver<SessionEvent>,
-    mut outbound_message_receiver: mpsc::Receiver<Outbound>,
+    mut outbound_message_receiver: mpsc::Receiver<OutboundRequest<Outbound>>,
     mut admin_request_receiver: mpsc::Receiver<AdminRequest>,
 ) where
     App: Application<Inbound, Outbound>,
@@ -1094,9 +1166,9 @@ async fn run_session<App, Inbound, Outbound, Store>(
                     None => break,
                 }
             }
-            next_outbound_message = outbound_message_receiver.recv() => {
-                match next_outbound_message {
-                    Some(message) => session.handle_outbound_message(message).await,
+            next_outbound_request = outbound_message_receiver.recv() => {
+                match next_outbound_request {
+                    Some(request) => session.handle_outbound_message(request).await,
                     None => break,
                 }
             }

crates/hotfix/src/session/error.rs

@@ -11,3 +11,31 @@ pub enum SessionError {
 }
 
 pub type Result<T> = std::result::Result<T, SessionError>;
+
+/// Outcome of a successful message send operation.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SendOutcome {
+    /// Message was persisted and sent with the given sequence number.
+    Sent { sequence_number: u64 },
+    /// Message was dropped by the application callback.
+    Dropped,
+}
+
+/// Error that can occur when sending a message.
+#[derive(Debug, Error)]
+pub enum SendError {
+    #[error("session is disconnected")]
+    Disconnected,
+
+    #[error("failed to persist message")]
+    Persist(#[source] StoreError),
+
+    #[error("failed to update sequence number")]
+    SequenceNumber(#[source] StoreError),
+
+    #[error("session terminated by application")]
+    SessionTerminated,
+
+    #[error("confirmation channel closed")]
+    ConfirmationLost,
+}

crates/hotfix/src/session/session_handle.rs

@@ -1,4 +1,6 @@
 use crate::session::admin_request::AdminRequest;
+use crate::session::error::{SendError, SendOutcome};
+use crate::session::session_ref::OutboundRequest;
 use crate::session::{InternalSessionRef, SessionInfo};
 use anyhow::anyhow;
 use tokio::sync::{mpsc, oneshot};
@@ -11,7 +13,7 @@ use tokio::sync::{mpsc, oneshot};
 /// and only exposes APIs intended for consumers of the engine.
 #[derive(Clone, Debug)]
 pub struct SessionHandle<Outbound> {
-    outbound_message_sender: mpsc::Sender<Outbound>,
+    outbound_message_sender: mpsc::Sender<OutboundRequest<Outbound>>,
     admin_request_sender: mpsc::Sender<AdminRequest>,
 }
 
@@ -24,9 +26,49 @@ impl<Outbound> SessionHandle<Outbound> {
         Ok(receiver.await?)
     }
 
+    /// Sends a message and waits for confirmation that it was persisted.
+    ///
+    /// Returns `SendOutcome::Sent` with the sequence number if the message was
+    /// successfully persisted and sent, or `SendOutcome::Dropped` if the application
+    /// callback chose to drop the message.
+    pub async fn send(&self, msg: Outbound) -> Result<SendOutcome, SendError> {
+        let (tx, rx) = oneshot::channel();
+        let request = OutboundRequest {
+            message: msg,
+            confirm: Some(tx),
+        };
+        self.outbound_message_sender
+            .send(request)
+            .await
+            .map_err(|_| SendError::Disconnected)?;
+
+        rx.await.map_err(|_| SendError::ConfirmationLost)?
+    }
+
+    /// Sends a message without waiting for confirmation.
+    ///
+    /// This is a fire-and-forget operation. The message will be queued for sending
+    /// but no confirmation is provided about whether it was actually sent.
+    pub async fn send_forget(&self, msg: Outbound) -> Result<(), SendError> {
+        let request = OutboundRequest {
+            message: msg,
+            confirm: None,
+        };
+        self.outbound_message_sender
+            .send(request)
+            .await
+            .map_err(|_| SendError::Disconnected)?;
+
+        Ok(())
+    }
+
+    #[deprecated(since = "0.5.0", note = "use `send` or `send_forget` instead")]
     pub async fn send_message(&self, msg: Outbound) -> anyhow::Result<()> {
         self.outbound_message_sender
-            .send(msg)
+            .send(OutboundRequest {
+                message: msg,
+                confirm: None,
+            })
             .await
             .map_err(|_| anyhow!("failed to send message"))?;
 

crates/hotfix/src/session/session_ref.rs

@@ -7,15 +7,22 @@ use crate::config::SessionConfig;
 use crate::message::{InboundMessage, OutboundMessage, RawFixMessage};
 use crate::session::Session;
 use crate::session::admin_request::AdminRequest;
+use crate::session::error::{SendError, SendOutcome};
 use crate::session::event::{AwaitingActiveSessionResponse, SessionEvent};
 use crate::store::MessageStore;
 use crate::transport::writer::WriterRef;
 use crate::{Application, session};
 
+/// A request to send an outbound message, optionally with confirmation.
+pub(crate) struct OutboundRequest<M> {
+    pub message: M,
+    pub confirm: Option<oneshot::Sender<Result<SendOutcome, SendError>>>,
+}
+
 #[derive(Clone)]
 pub struct InternalSessionRef<Outbound> {
     pub(crate) event_sender: mpsc::Sender<SessionEvent>,
-    pub(crate) outbound_message_sender: mpsc::Sender<Outbound>,
+    pub(crate) outbound_message_sender: mpsc::Sender<OutboundRequest<Outbound>>,
     pub(crate) admin_request_sender: mpsc::Sender<AdminRequest>,
 }
 
@@ -26,7 +33,8 @@ impl<Outbound: OutboundMessage> InternalSessionRef<Outbound> {
         store: impl MessageStore + 'static,
     ) -> Result<Self> {
         let (event_sender, event_receiver) = mpsc::channel::<SessionEvent>(100);
-        let (outbound_message_sender, outbound_message_receiver) = mpsc::channel::<Outbound>(10);
+        let (outbound_message_sender, outbound_message_receiver) =
+            mpsc::channel::<OutboundRequest<Outbound>>(10);
         let (admin_request_sender, admin_request_receiver) = mpsc::channel::<AdminRequest>(10);
         let session = Session::new(config, application, store)?;
         tokio::spawn(session::run_session(

crates/hotfix/src/transport/socket/socket_reader.rs

@@ -87,6 +87,7 @@ mod tests {
     use crate::message::Message;
     use crate::session::admin_request::AdminRequest;
     use crate::session::event::SessionEvent;
+    use crate::session::session_ref::OutboundRequest;
     use tokio::io::{AsyncWriteExt, duplex};
     use tokio::sync::mpsc;
 
@@ -107,7 +108,8 @@ mod tests {
         mpsc::Receiver<SessionEvent>,
     ) {
         let (event_sender, event_receiver) = mpsc::channel::<SessionEvent>(100);
-        let (outbound_message_sender, _outbound_receiver) = mpsc::channel::<TestMessage>(10);
+        let (outbound_message_sender, _outbound_receiver) =
+            mpsc::channel::<OutboundRequest<TestMessage>>(10);
         let (admin_request_sender, _admin_receiver) = mpsc::channel::<AdminRequest>(10);
 
         let session_ref = InternalSessionRef {

crates/hotfix/tests/session_test_cases/common/actions.rs

@@ -1,6 +1,7 @@
 use crate::common::fakes::{FakeCounterparty, SessionSpy};
 use crate::common::test_messages::TestMessage;
 use hotfix::message::OutboundMessage;
+use hotfix::session::{SendError, SendOutcome};
 use std::time::Duration;
 
 pub struct When<T> {
@@ -19,10 +20,17 @@ impl When<&SessionSpy> {
     pub async fn sends_message(self, message: TestMessage) {
         self.target
             .session_handle()
-            .send_message(message)
+            .send(message)
             .await
             .expect("message to be sent successfully");
     }
+
+    pub async fn sends_message_with_confirmation(
+        self,
+        message: TestMessage,
+    ) -> Result<SendOutcome, SendError> {
+        self.target.session_handle().send(message).await
+    }
 }
 
 impl When<&mut FakeCounterparty<TestMessage>> {

crates/hotfix/tests/session_test_cases/mod.rs

@@ -8,3 +8,4 @@ mod logout_tests;
 mod received_gap_fill_tests;
 mod received_reset_tests;
 mod resend_tests;
+mod send_confirmation_tests;