asynchronous message sending to allow accepting messages when agent (or server) is offline (#179)

* asynchronous message sending to allow accepting messages when agent (or server) is offline

* send pending messages after agent restart; test for pending messages delivery when server restarted

* test pending message delivery after agent restart

* message delivery process per server

* test layout

migrations/20210809_snd_messages.sql

@@ -0,0 +1,3 @@
+ALTER TABLE messages ADD msg_body BLOB NOT NULL DEFAULT x'';  -- this field replaces body TEXT
+-- TODO possibly migrate the data from body if it is possible in migration
+ALTER TABLE snd_messages ADD previous_msg_hash BLOB NOT NULL DEFAULT x'';

src/Simplex/Messaging/Agent.hs

@@ -62,6 +62,9 @@ import Data.Composition ((.:), (.:.))
 import Data.Functor (($>))
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
+import Data.Map.Strict (Map)
+import qualified Data.Map.Strict as M
+import Data.Maybe (isJust)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeUtf8)
 import Data.Time.Clock
@@ -69,6 +72,7 @@ import Database.SQLite.Simple (SQLError)
 import Simplex.Messaging.Agent.Client
 import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Protocol
+import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore)
 import Simplex.Messaging.Client (SMPServerTransmission)
@@ -78,7 +82,7 @@ import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), runTransportServer)
 import Simplex.Messaging.Util (bshow)
 import System.Random (randomR)
-import UnliftIO.Async (async, race_)
+import UnliftIO.Async (Async, async, race_)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
@@ -226,7 +230,7 @@ processCommand c (connId, cmd) = case cmd of
   JOIN smpQueueInfo connInfo -> (,OK) <$> joinConn c connId smpQueueInfo connInfo
   ACPT confId ownConnInfo -> acceptConnection' c connId confId ownConnInfo $> (connId, OK)
   SUB -> subscribeConnection' c connId $> (connId, OK)
-  SEND msgBody -> (connId,) . SENT . unId <$> sendMessage' c connId msgBody
+  SEND msgBody -> (connId,) . MID . unId <$> sendMessage' c connId msgBody
   OFF -> suspendConnection' c connId $> (connId, OK)
   DEL -> deleteConnection' c connId $> (connId, OK)
 
@@ -282,22 +286,32 @@ processConfirmation c rq sndKey = do
 subscribeConnection' :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()
 subscribeConnection' c connId =
   withStore (`getConn` connId) >>= \case
-    SomeConn _ (DuplexConnection _ rq sq) -> case status (sq :: SndQueue) of
-      Confirmed -> withVerifyKey sq $ \verifyKey -> do
-        conf <- withStore (`getAcceptedConfirmation` connId)
-        secureQueue c rq $ senderKey (conf :: AcceptedConfirmation)
-        withStore $ \st -> setRcvQueueStatus st rq Secured
-        activateSecuredQueue rq sq verifyKey
-      Secured -> withVerifyKey sq $ activateSecuredQueue rq sq
-      Active -> subscribeQueue c rq connId
-      _ -> throwError $ INTERNAL "unexpected queue status"
-    SomeConn _ (SndConnection _ sq) -> case status (sq :: SndQueue) of
-      Confirmed -> withVerifyKey sq $ \verifyKey ->
-        activateQueueJoining c connId sq verifyKey =<< resumeInterval
-      Active -> throwError $ CONN SIMPLEX
-      _ -> throwError $ INTERNAL "unexpected queue status"
+    SomeConn _ (DuplexConnection _ rq sq) -> do
+      resumeDelivery sq
+      case status (sq :: SndQueue) of
+        Confirmed -> withVerifyKey sq $ \verifyKey -> do
+          conf <- withStore (`getAcceptedConfirmation` connId)
+          secureQueue c rq $ senderKey (conf :: AcceptedConfirmation)
+          withStore $ \st -> setRcvQueueStatus st rq Secured
+          activateSecuredQueue rq sq verifyKey
+        Secured -> withVerifyKey sq $ activateSecuredQueue rq sq
+        Active -> subscribeQueue c rq connId
+        _ -> throwError $ INTERNAL "unexpected queue status"
+    SomeConn _ (SndConnection _ sq) -> do
+      resumeDelivery sq
+      case status (sq :: SndQueue) of
+        Confirmed -> withVerifyKey sq $ \verifyKey ->
+          activateQueueJoining c connId sq verifyKey =<< resumeInterval
+        Active -> throwError $ CONN SIMPLEX
+        _ -> throwError $ INTERNAL "unexpected queue status"
     SomeConn _ (RcvConnection _ rq) -> subscribeQueue c rq connId
   where
+    resumeDelivery :: SndQueue -> m ()
+    resumeDelivery SndQueue {server} = do
+      wasDelivering <- resumeMsgDelivery c connId server
+      unless wasDelivering $ do
+        pending <- withStore (`getPendingMsgs` connId)
+        queuePendingMsgs c connId pending
     withVerifyKey :: SndQueue -> (C.PublicKey -> m ()) -> m ()
     withVerifyKey sq action =
       let err = throwError $ INTERNAL "missing signing key public counterpart"
@@ -313,30 +327,96 @@ subscribeConnection' c connId =
 
 -- | Send message to the connection (SEND command) in Reader monad
 sendMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> MsgBody -> m InternalId
-sendMessage' c connId msgBody =
+sendMessage' c connId msg =
   withStore (`getConn` connId) >>= \case
-    SomeConn _ (DuplexConnection _ _ sq) -> sendMsg_ sq
-    SomeConn _ (SndConnection _ sq) -> sendMsg_ sq
+    SomeConn _ (DuplexConnection _ _ sq) -> enqueueMessage sq
+    SomeConn _ (SndConnection _ sq) -> enqueueMessage sq
     _ -> throwError $ CONN SIMPLEX
   where
-    sendMsg_ :: SndQueue -> m InternalId
-    sendMsg_ sq = do
-      internalTs <- liftIO getCurrentTime
-      (internalId, internalSndId, previousMsgHash) <- withStore (`updateSndIds` connId)
-      let msgStr =
-            serializeSMPMessage
-              SMPMessage
-                { senderMsgId = unSndId internalSndId,
-                  senderTimestamp = internalTs,
-                  previousMsgHash,
-                  agentMessage = A_MSG msgBody
-                }
-          msgHash = C.sha256Hash msgStr
-      withStore $ \st ->
-        createSndMsg st connId $
-          SndMsgData {internalId, internalSndId, internalTs, msgBody, internalHash = msgHash}
-      sendAgentMessage c sq msgStr
-      pure internalId
+    enqueueMessage :: SndQueue -> m InternalId
+    enqueueMessage SndQueue {server} = do
+      msgId <- storeSentMsg
+      wasDelivering <- resumeMsgDelivery c connId server
+      pending <-
+        if wasDelivering
+          then pure [PendingMsg {connId, msgId}]
+          else withStore (`getPendingMsgs` connId)
+      queuePendingMsgs c connId pending
+      pure msgId
+      where
+        storeSentMsg :: m InternalId
+        storeSentMsg = do
+          internalTs <- liftIO getCurrentTime
+          withStore $ \st -> do
+            (internalId, internalSndId, previousMsgHash) <- updateSndIds st connId
+            let msgBody =
+                  serializeSMPMessage
+                    SMPMessage
+                      { senderMsgId = unSndId internalSndId,
+                        senderTimestamp = internalTs,
+                        previousMsgHash,
+                        agentMessage = A_MSG msg
+                      }
+                msgData = SndMsgData {internalId, internalSndId, internalTs, msgBody, internalHash = C.sha256Hash msgBody, previousMsgHash}
+            createSndMsg st connId msgData
+            pure internalId
+
+resumeMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnId -> SMPServer -> m Bool
+resumeMsgDelivery c connId srv = do
+  void $ resume srv (srvMsgDeliveries c) $ runSrvMsgDelivery c srv
+  resume connId (connMsgDeliveries c) $ runMsgDelivery c connId srv
+  where
+    resume :: Ord a => a -> TVar (Map a (Async ())) -> m () -> m Bool
+    resume key actionMap actionProcess = do
+      isDelivering <- isJust . M.lookup key <$> readTVarIO actionMap
+      unless isDelivering $
+        async actionProcess
+          >>= atomically . modifyTVar actionMap . M.insert key
+      pure isDelivering
+
+queuePendingMsgs :: AgentMonad m => AgentClient -> ConnId -> [PendingMsg] -> m ()
+queuePendingMsgs c connId pending =
+  atomically $ getPendingMsgQ connId (connMsgQueues c) >>= forM_ pending . writeTQueue
+
+getPendingMsgQ :: Ord a => a -> TVar (Map a (TQueue PendingMsg)) -> STM (TQueue PendingMsg)
+getPendingMsgQ key queueMap = do
+  maybe newMsgQueue pure . M.lookup key =<< readTVar queueMap
+  where
+    newMsgQueue :: STM (TQueue PendingMsg)
+    newMsgQueue = do
+      mq <- newTQueue
+      modifyTVar queueMap $ M.insert key mq
+      pure mq
+
+runMsgDelivery :: AgentMonad m => AgentClient -> ConnId -> SMPServer -> m ()
+runMsgDelivery c connId srv = do
+  mq <- atomically . getPendingMsgQ connId $ connMsgQueues c
+  smq <- atomically . getPendingMsgQ srv $ srvMsgQueues c
+  forever . atomically $ readTQueue mq >>= writeTQueue smq
+
+runSrvMsgDelivery :: forall m. AgentMonad m => AgentClient -> SMPServer -> m ()
+runSrvMsgDelivery c@AgentClient {subQ} srv = do
+  mq <- atomically . getPendingMsgQ srv $ srvMsgQueues c
+  ri <- asks $ reconnectInterval . config
+  forever $ do
+    PendingMsg {connId, msgId} <- atomically $ readTQueue mq
+    let mId = unId msgId
+    r <- withStore $ \st ->
+      (Right <$> getPendingMsgData st connId msgId)
+        `E.catch` \(e :: E.SomeException) -> pure $ Left e
+    case r of
+      Left e -> notify connId $ MERR mId (INTERNAL $ show e)
+      Right (sq, msgBody) -> do
+        withRetryInterval ri $ \loop -> do
+          sendAgentMessage c sq msgBody
+            `catchError` \case
+              e@SMP {} -> notify connId $ MERR mId e
+              _ -> loop
+          notify connId $ SENT mId
+          withStore $ \st -> updateSndMsgStatus st connId msgId SndMsgSent
+  where
+    notify :: ConnId -> ACommand 'Agent -> m ()
+    notify connId cmd = atomically $ writeTBQueue subQ ("", connId, cmd)
 
 -- | Suspend SMP agent connection (OFF command) in Reader monad
 suspendConnection' :: AgentMonad m => AgentClient -> ConnId -> m ()

src/Simplex/Messaging/Agent/Client.hs

@@ -72,6 +72,10 @@ data AgentClient = AgentClient
     subscrSrvrs :: TVar (Map SMPServer (Map ConnId RcvQueue)),
     subscrConns :: TVar (Map ConnId SMPServer),
     activations :: TVar (Map ConnId (Async ())), -- activations of send queues in progress
+    connMsgQueues :: TVar (Map ConnId (TQueue PendingMsg)),
+    connMsgDeliveries :: TVar (Map ConnId (Async ())),
+    srvMsgQueues :: TVar (Map SMPServer (TQueue PendingMsg)),
+    srvMsgDeliveries :: TVar (Map SMPServer (Async ())),
     reconnections :: TVar [Async ()],
     clientId :: Int,
     agentEnv :: Env,
@@ -89,10 +93,14 @@ newAgentClient agentEnv = do
   subscrSrvrs <- newTVar M.empty
   subscrConns <- newTVar M.empty
   activations <- newTVar M.empty
+  connMsgQueues <- newTVar M.empty
+  connMsgDeliveries <- newTVar M.empty
+  srvMsgQueues <- newTVar M.empty
+  srvMsgDeliveries <- newTVar M.empty
   reconnections <- newTVar []
   clientId <- stateTVar (clientCounter agentEnv) $ \i -> (i + 1, i + 1)
   lock <- newTMVar ()
-  return AgentClient {rcvQ, subQ, msgQ, smpClients, subscrSrvrs, subscrConns, activations, reconnections, clientId, agentEnv, smpSubscriber = undefined, lock}
+  return AgentClient {rcvQ, subQ, msgQ, smpClients, subscrSrvrs, subscrConns, activations, connMsgQueues, connMsgDeliveries, srvMsgQueues, srvMsgDeliveries, reconnections, clientId, agentEnv, smpSubscriber = undefined, lock}
 
 -- | Agent monad with MonadReader Env and MonadError AgentErrorType
 type AgentMonad m = (MonadUnliftIO m, MonadReader Env m, MonadError AgentErrorType m)
@@ -168,6 +176,8 @@ closeAgentClient c = liftIO $ do
   closeSMPServerClients c
   cancelActions $ activations c
   cancelActions $ reconnections c
+  cancelActions $ connMsgDeliveries c
+  cancelActions $ srvMsgDeliveries c
 
 closeSMPServerClients :: AgentClient -> IO ()
 closeSMPServerClients c = readTVarIO (smpClients c) >>= mapM_ closeSMPClient

src/Simplex/Messaging/Agent/Protocol.hs

@@ -169,6 +169,7 @@ data ACommand (p :: AParty) where
   SEND :: MsgBody -> ACommand Client
   MID :: AgentMsgId -> ACommand Agent
   SENT :: AgentMsgId -> ACommand Agent
+  MERR :: AgentMsgId -> AgentErrorType -> ACommand Agent
   MSG :: MsgMeta -> MsgBody -> ACommand Agent
   -- ACK :: AgentMsgId -> ACommand Client
   -- RCVD :: AgentMsgId -> ACommand Agent
@@ -463,6 +464,7 @@ commandP =
     <|> "SEND " *> sendCmd
     <|> "MID " *> msgIdResp
     <|> "SENT " *> sentResp
+    <|> "MERR " *> msgErrResp
     <|> "MSG " *> message
     <|> "OFF" $> ACmd SClient OFF
     <|> "DEL" $> ACmd SClient DEL
@@ -478,6 +480,7 @@ commandP =
     sendCmd = ACmd SClient . SEND <$> A.takeByteString
     msgIdResp = ACmd SAgent . MID <$> A.decimal
     sentResp = ACmd SAgent . SENT <$> A.decimal
+    msgErrResp = ACmd SAgent <$> (MERR <$> A.decimal <* A.space <*> agentErrorTypeP)
     message = ACmd SAgent <$> (MSG <$> msgMetaP <* A.space <*> A.takeByteString)
     msgMetaP = do
       integrity <- msgIntegrityP
@@ -517,6 +520,7 @@ serializeCommand = \case
   SEND msgBody -> "SEND " <> serializeBinary msgBody
   MID mId -> "MID " <> bshow mId
   SENT mId -> "SENT " <> bshow mId
+  MERR mId e -> "MERR " <> bshow mId <> " " <> serializeAgentError e
   MSG msgMeta msgBody ->
     "MSG " <> serializeMsgMeta msgMeta <> " " <> serializeBinary msgBody
   OFF -> "OFF"

src/Simplex/Messaging/Agent/Store.hs

@@ -56,6 +56,9 @@ class Monad m => MonadAgentStore s m where
   createRcvMsg :: s -> ConnId -> RcvMsgData -> m ()
   updateSndIds :: s -> ConnId -> m (InternalId, InternalSndId, PrevSndMsgHash)
   createSndMsg :: s -> ConnId -> SndMsgData -> m ()
+  updateSndMsgStatus :: s -> ConnId -> InternalId -> SndMsgStatus -> m ()
+  getPendingMsgData :: s -> ConnId -> InternalId -> m (SndQueue, MsgBody)
+  getPendingMsgs :: s -> ConnId -> m [PendingMsg]
   getMsg :: s -> ConnId -> InternalId -> m Msg
 
 -- * Queue types
@@ -187,9 +190,16 @@ data SndMsgData = SndMsgData
     internalSndId :: InternalSndId,
     internalTs :: InternalTs,
     msgBody :: MsgBody,
-    internalHash :: MsgHash
+    internalHash :: MsgHash,
+    previousMsgHash :: MsgHash
   }
 
+data PendingMsg = PendingMsg
+  { connId :: ConnId,
+    msgId :: InternalId
+  }
+  deriving (Show)
+
 -- * Broadcast types
 
 type BroadcastId = ByteString
@@ -310,6 +320,8 @@ data StoreError
     SEBadConnType ConnType
   | -- | Confirmation not found.
     SEConfirmationNotFound
+  | -- | Message not found
+    SEMsgNotFound
   | -- | Currently not used. The intention was to pass current expected queue status in methods,
     -- as we always know what it should be at any stage of the protocol,
     -- and in case it does not match use this error.

src/Simplex/Messaging/Agent/Store/SQLite.hs

@@ -29,8 +29,7 @@ where
 import Control.Concurrent (threadDelay)
 import Control.Concurrent.STM
 import Control.Exception (bracket)
-import Control.Monad (replicateM_, unless, when)
-import Control.Monad.Except (MonadError (throwError), MonadIO (liftIO))
+import Control.Monad.Except
 import Control.Monad.IO.Unlift (MonadUnliftIO)
 import Crypto.Random (ChaChaDRG, randomBytesGenerate)
 import Data.ByteString (ByteString)
@@ -40,7 +39,6 @@ import Data.List (find)
 import Data.Maybe (fromMaybe)
 import Data.Text (Text)
 import qualified Data.Text as T
-import Data.Text.Encoding (decodeUtf8)
 import Database.SQLite.Simple (FromRow, NamedParam (..), Only (..), SQLData (..), SQLError, field)
 import qualified Database.SQLite.Simple as DB
 import Database.SQLite.Simple.FromField
@@ -54,6 +52,7 @@ import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite.Migrations (Migration)
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
 import Simplex.Messaging.Parsers (blobFieldParser)
+import Simplex.Messaging.Protocol (MsgBody)
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Util (bshow, liftIOEither)
 import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
@@ -392,6 +391,51 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
       insertSndMsgDetails_ db connId sndMsgData
       updateHashSnd_ db connId sndMsgData
 
+  updateSndMsgStatus :: SQLiteStore -> ConnId -> InternalId -> SndMsgStatus -> m ()
+  updateSndMsgStatus st connId msgId msgStatus =
+    liftIO . withTransaction st $ \db ->
+      DB.executeNamed
+        db
+        [sql|
+          UPDATE snd_messages
+          SET snd_status = :snd_status
+          WHERE conn_alias = :conn_alias AND internal_id = :internal_id
+        |]
+        [ ":conn_alias" := connId,
+          ":internal_id" := msgId,
+          ":snd_status" := msgStatus
+        ]
+
+  getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m (SndQueue, MsgBody)
+  getPendingMsgData st connId msgId =
+    liftIOEither . withTransaction st $ \db -> runExceptT $ do
+      sq <- ExceptT $ sndQueue <$> getSndQueueByConnAlias_ db connId
+      msgBody <-
+        ExceptT $
+          sndMsgData
+            <$> DB.query
+              db
+              [sql|
+                SELECT m.msg_body
+                FROM messages m
+                JOIN snd_messages s ON s.conn_alias = m.conn_alias AND s.internal_id = m.internal_id
+                WHERE m.conn_alias = ? AND m.internal_id = ?
+              |]
+              (connId, msgId)
+      pure (sq, msgBody)
+    where
+      sndMsgData :: [Only MsgBody] -> Either StoreError MsgBody
+      sndMsgData [Only msgBody] = Right msgBody
+      sndMsgData _ = Left SEMsgNotFound
+      sndQueue :: Maybe SndQueue -> Either StoreError SndQueue
+      sndQueue = maybe (Left SEConnNotFound) Right
+
+  getPendingMsgs :: SQLiteStore -> ConnId -> m [PendingMsg]
+  getPendingMsgs st connId =
+    liftIO . withTransaction st $ \db ->
+      map (PendingMsg connId . fromOnly)
+        <$> DB.query db "SELECT internal_id FROM snd_messages WHERE conn_alias = ? AND snd_status = ?" (connId, SndMsgCreated)
+
   getMsg :: SQLiteStore -> ConnId -> InternalId -> m Msg
   getMsg _st _connAlias _id = throwError SENotImplemented
 
@@ -676,15 +720,15 @@ insertRcvMsgBase_ dbConn connId RcvMsgData {msgMeta, msgBody, internalRcvId} = d
     dbConn
     [sql|
       INSERT INTO messages
-        ( conn_alias, internal_id, internal_ts, internal_rcv_id, internal_snd_id, body)
+        ( conn_alias, internal_id, internal_ts, internal_rcv_id, internal_snd_id, body, msg_body)
       VALUES
-        (:conn_alias,:internal_id,:internal_ts,:internal_rcv_id,            NULL,:body);
+        (:conn_alias,:internal_id,:internal_ts,:internal_rcv_id,            NULL,   '', :msg_body);
     |]
     [ ":conn_alias" := connId,
       ":internal_id" := internalId,
       ":internal_ts" := internalTs,
       ":internal_rcv_id" := internalRcvId,
-      ":body" := decodeUtf8 msgBody
+      ":msg_body" := msgBody
     ]
 
 insertRcvMsgDetails_ :: DB.Connection -> ConnId -> RcvMsgData -> IO ()
@@ -771,15 +815,15 @@ insertSndMsgBase_ dbConn connId SndMsgData {..} = do
     dbConn
     [sql|
       INSERT INTO messages
-        ( conn_alias, internal_id, internal_ts, internal_rcv_id, internal_snd_id, body)
+        ( conn_alias, internal_id, internal_ts, internal_rcv_id, internal_snd_id, body, msg_body)
       VALUES
-        (:conn_alias,:internal_id,:internal_ts,            NULL,:internal_snd_id,:body);
+        (:conn_alias,:internal_id,:internal_ts,            NULL,:internal_snd_id,   '',:msg_body);
     |]
     [ ":conn_alias" := connId,
       ":internal_id" := internalId,
       ":internal_ts" := internalTs,
       ":internal_snd_id" := internalSndId,
-      ":body" := decodeUtf8 msgBody
+      ":msg_body" := msgBody
     ]
 
 insertSndMsgDetails_ :: DB.Connection -> ConnId -> SndMsgData -> IO ()
@@ -788,15 +832,16 @@ insertSndMsgDetails_ dbConn connId SndMsgData {..} =
     dbConn
     [sql|
       INSERT INTO snd_messages
-        ( conn_alias, internal_snd_id, internal_id, snd_status, sent_ts, delivered_ts, internal_hash)
+        ( conn_alias, internal_snd_id, internal_id, snd_status, sent_ts, delivered_ts, internal_hash, previous_msg_hash)
       VALUES
-        (:conn_alias,:internal_snd_id,:internal_id,:snd_status,    NULL,         NULL,:internal_hash);
+        (:conn_alias,:internal_snd_id,:internal_id,:snd_status,    NULL,         NULL,:internal_hash,:previous_msg_hash);
     |]
     [ ":conn_alias" := connId,
       ":internal_snd_id" := internalSndId,
       ":internal_id" := internalId,
       ":snd_status" := SndMsgCreated,
-      ":internal_hash" := internalHash
+      ":internal_hash" := internalHash,
+      ":previous_msg_hash" := previousMsgHash
     ]
 
 updateHashSnd_ :: DB.Connection -> ConnId -> SndMsgData -> IO ()

tests/AgentTests.hs

@@ -16,10 +16,11 @@ import Control.Concurrent
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import SMPAgentClient
-import SMPClient (withSmpServer)
+import SMPClient (testPort, testPort2, testStoreLogFile, withSmpServer, withSmpServerStoreLogOn)
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Protocol (ErrorType (..), MsgBody)
 import Simplex.Messaging.Transport (ATransport (..), TProxy (..), Transport (..))
+import System.Directory (removeFile)
 import System.Timeout
 import Test.Hspec
 
@@ -46,6 +47,11 @@ agentTests (ATransport t) = do
       smpAgentTest3_1_1 $ testSubscription t
     it "should send notifications to client when server disconnects" $
       smpAgentServerTest $ testSubscrNotification t
+  describe "Message delivery" do
+    it "should deliver messages after losing server connection and re-connecting" $
+      smpAgentTest2_2_2_needs_server $ testMsgDeliveryServerRestart t
+    it "should deliver pending messages after agent restarting" $
+      smpAgentTest1_1_1 $ testMsgDeliveryAgentRestart t
 
 -- | receive message to handle `h`
 (<#:) :: Transport c => c -> IO (ATransmissionOrError 'Agent)
@@ -100,16 +106,21 @@ testDuplexConnection _ alice bob = do
   bob <# ("", "alice", INFO "alice's connInfo")
   bob <# ("", "alice", CON)
   alice <# ("", "bob", CON)
-  alice #: ("3", "bob", "SEND :hello") #> ("3", "bob", SENT 1)
-  alice #: ("4", "bob", "SEND :how are you?") #> ("4", "bob", SENT 2)
+  alice #: ("3", "bob", "SEND :hello") #> ("3", "bob", MID 1)
+  alice <# ("", "bob", SENT 1)
+  alice #: ("4", "bob", "SEND :how are you?") #> ("4", "bob", MID 2)
+  alice <# ("", "bob", SENT 2)
   bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False
   bob <#= \case ("", "alice", Msg "how are you?") -> True; _ -> False
-  bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", SENT 3)
+  bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", MID 3)
+  bob <# ("", "alice", SENT 3)
   alice <#= \case ("", "bob", Msg "hello too") -> True; _ -> False
-  bob #: ("15", "alice", "SEND 9\nmessage 1") #> ("15", "alice", SENT 4)
+  bob #: ("15", "alice", "SEND 9\nmessage 1") #> ("15", "alice", MID 4)
+  bob <# ("", "alice", SENT 4)
   alice <#= \case ("", "bob", Msg "message 1") -> True; _ -> False
   alice #: ("5", "bob", "OFF") #> ("5", "bob", OK)
-  bob #: ("17", "alice", "SEND 9\nmessage 3") #> ("17", "alice", ERR (SMP AUTH))
+  bob #: ("17", "alice", "SEND 9\nmessage 3") #> ("17", "alice", MID 5)
+  bob <# ("", "alice", MERR 5 (SMP AUTH))
   alice #: ("6", "bob", "DEL") #> ("6", "bob", OK)
   alice #:# "nothing else should be delivered to alice"
 
@@ -124,29 +135,37 @@ testDuplexConnRandomIds _ alice bob = do
   bob <# ("", aliceConn, INFO "alice's connInfo")
   bob <# ("", aliceConn, CON)
   alice <# ("", bobConn, CON)
-  alice #: ("2", bobConn, "SEND :hello") #> ("2", bobConn, SENT 1)
-  alice #: ("3", bobConn, "SEND :how are you?") #> ("3", bobConn, SENT 2)
+  alice #: ("2", bobConn, "SEND :hello") #> ("2", bobConn, MID 1)
+  alice <# ("", bobConn, SENT 1)
+  alice #: ("3", bobConn, "SEND :how are you?") #> ("3", bobConn, MID 2)
+  alice <# ("", bobConn, SENT 2)
   bob <#= \case ("", c, Msg "hello") -> c == aliceConn; _ -> False
   bob <#= \case ("", c, Msg "how are you?") -> c == aliceConn; _ -> False
-  bob #: ("14", aliceConn, "SEND 9\nhello too") #> ("14", aliceConn, SENT 3)
+  bob #: ("14", aliceConn, "SEND 9\nhello too") #> ("14", aliceConn, MID 3)
+  bob <# ("", aliceConn, SENT 3)
   alice <#= \case ("", c, Msg "hello too") -> c == bobConn; _ -> False
-  bob #: ("15", aliceConn, "SEND 9\nmessage 1") #> ("15", aliceConn, SENT 4)
+  bob #: ("15", aliceConn, "SEND 9\nmessage 1") #> ("15", aliceConn, MID 4)
+  bob <# ("", aliceConn, SENT 4)
   alice <#= \case ("", c, Msg "message 1") -> c == bobConn; _ -> False
   alice #: ("5", bobConn, "OFF") #> ("5", bobConn, OK)
-  bob #: ("17", aliceConn, "SEND 9\nmessage 3") #> ("17", aliceConn, ERR (SMP AUTH))
+  bob #: ("17", aliceConn, "SEND 9\nmessage 3") #> ("17", aliceConn, MID 5)
+  bob <# ("", aliceConn, MERR 5 (SMP AUTH))
   alice #: ("6", bobConn, "DEL") #> ("6", bobConn, OK)
   alice #:# "nothing else should be delivered to alice"
 
 testSubscription :: Transport c => TProxy c -> c -> c -> c -> IO ()
 testSubscription _ alice1 alice2 bob = do
   (alice1, "alice") `connect` (bob, "bob")
-  bob #: ("12", "alice", "SEND 5\nhello") #> ("12", "alice", SENT 1)
-  bob #: ("13", "alice", "SEND 11\nhello again") #> ("13", "alice", SENT 2)
+  bob #: ("12", "alice", "SEND 5\nhello") #> ("12", "alice", MID 1)
+  bob <# ("", "alice", SENT 1)
+  bob #: ("13", "alice", "SEND 11\nhello again") #> ("13", "alice", MID 2)
+  bob <# ("", "alice", SENT 2)
   alice1 <#= \case ("", "bob", Msg "hello") -> True; _ -> False
   alice1 <#= \case ("", "bob", Msg "hello again") -> True; _ -> False
   alice2 #: ("21", "bob", "SUB") #> ("21", "bob", OK)
   alice1 <# ("", "bob", END)
-  bob #: ("14", "alice", "SEND 2\nhi") #> ("14", "alice", SENT 3)
+  bob #: ("14", "alice", "SEND 2\nhi") #> ("14", "alice", MID 3)
+  bob <# ("", "alice", SENT 3)
   alice2 <#= \case ("", "bob", Msg "hi") -> True; _ -> False
   alice1 #:# "nothing else should be delivered to alice1"
 
@@ -159,6 +178,61 @@ testSubscrNotification t (server, _) client = do
   withSmpServer (ATransport t) $
     client <# ("", "conn1", ERR (SMP AUTH)) -- this new server does not have the queue
 
+testMsgDeliveryServerRestart :: Transport c => TProxy c -> c -> c -> IO ()
+testMsgDeliveryServerRestart t alice bob = do
+  withServer $ do
+    connect (alice, "alice") (bob, "bob")
+    bob #: ("1", "alice", "SEND 2\nhi") #> ("1", "alice", MID 1)
+    bob <# ("", "alice", SENT 1)
+    alice <#= \case ("", "bob", Msg "hi") -> True; _ -> False
+    alice #:# "nothing else delivered before the server is killed"
+
+  alice <# ("", "bob", DOWN)
+  bob #: ("2", "alice", "SEND 11\nhello again") #> ("2", "alice", MID 2)
+  bob #:# "nothing else delivered before the server is restarted"
+  alice #:# "nothing else delivered before the server is restarted"
+
+  withServer $ do
+    bob <# ("", "alice", SENT 2)
+    alice <# ("", "bob", UP)
+    alice <#= \case ("", "bob", Msg "hello again") -> True; _ -> False
+
+  removeFile testStoreLogFile
+  where
+    withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
+
+testMsgDeliveryAgentRestart :: Transport c => TProxy c -> c -> IO ()
+testMsgDeliveryAgentRestart t bob = do
+  withAgent $ \alice -> do
+    withServer $ do
+      connect (bob, "bob") (alice, "alice")
+      alice #: ("1", "bob", "SEND 5\nhello") #> ("1", "bob", MID 1)
+      alice <# ("", "bob", SENT 1)
+      bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False
+      bob #:# "nothing else delivered before the server is down"
+
+    bob <# ("", "alice", DOWN)
+    alice #: ("2", "bob", "SEND 11\nhello again") #> ("2", "bob", MID 2)
+    alice #:# "nothing else delivered before the server is restarted"
+    bob #:# "nothing else delivered before the server is restarted"
+
+  withAgent $ \alice -> do
+    withServer $ do
+      tPutRaw alice ("3", "bob", "SUB")
+      alice <#= \case
+        (corrId, "bob", cmd) ->
+          (corrId == "3" && cmd == OK)
+            || (corrId == "" && cmd == SENT 2)
+        _ -> False
+      bob <# ("", "alice", UP)
+      bob <#= \case ("", "alice", Msg "hello again") -> True; _ -> False
+
+  removeFile testStoreLogFile
+  removeFile testDB
+  where
+    withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
+    withAgent = withSmpAgentThreadOn_ (ATransport t) (agentTestPort, testPort, testDB) (pure ()) . const . testSMPAgentClientOn agentTestPort
+
 connect :: forall c. Transport c => (c, ByteString) -> (c, ByteString) -> IO ()
 connect (h1, name1) (h2, name2) = do
   ("c1", _, Right (INV qInfo)) <- h1 #: ("c1", name2, "NEW")
@@ -170,17 +244,17 @@ connect (h1, name1) (h2, name2) = do
   h2 <# ("", name1, CON)
   h1 <# ("", name2, CON)
 
-connect' :: forall c. Transport c => c -> c -> IO (ByteString, ByteString)
-connect' h1 h2 = do
-  ("c1", conn2, Right (INV qInfo)) <- h1 #: ("c1", "", "NEW")
-  let qInfo' = serializeSmpQueueInfo qInfo
-  ("c2", conn1, Right OK) <- h2 #: ("c2", "", "JOIN " <> qInfo' <> " 5\ninfo2")
-  ("", _, Right (REQ connId "info2")) <- (h1 <#:)
-  h1 #: ("c3", conn2, "ACPT " <> connId <> " 5\ninfo1") =#> \case ("c3", c, OK) -> c == conn2; _ -> False
-  h2 <# ("", conn1, INFO "info1")
-  h2 <# ("", conn1, CON)
-  h1 <# ("", conn2, CON)
-  pure (conn1, conn2)
+-- connect' :: forall c. Transport c => c -> c -> IO (ByteString, ByteString)
+-- connect' h1 h2 = do
+--   ("c1", conn2, Right (INV qInfo)) <- h1 #: ("c1", "", "NEW")
+--   let qInfo' = serializeSmpQueueInfo qInfo
+--   ("c2", conn1, Right OK) <- h2 #: ("c2", "", "JOIN " <> qInfo' <> " 5\ninfo2")
+--   ("", _, Right (REQ connId "info2")) <- (h1 <#:)
+--   h1 #: ("c3", conn2, "ACPT " <> connId <> " 5\ninfo1") =#> \case ("c3", c, OK) -> c == conn2; _ -> False
+--   h2 <# ("", conn1, INFO "info1")
+--   h2 <# ("", conn1, CON)
+--   h1 <# ("", conn2, CON)
+--   pure (conn1, conn2)
 
 samplePublicKey :: ByteString
 samplePublicKey = "rsa:MIIBoDANBgkqhkiG9w0BAQEFAAOCAY0AMIIBiAKCAQEAtn1NI2tPoOGSGfad0aUg0tJ0kG2nzrIPGLiz8wb3dQSJC9xkRHyzHhEE8Kmy2cM4q7rNZIlLcm4M7oXOTe7SC4x59bLQG9bteZPKqXu9wk41hNamV25PWQ4zIcIRmZKETVGbwN7jFMpH7wxLdI1zzMArAPKXCDCJ5ctWh4OWDI6OR6AcCtEj+toCI6N6pjxxn5VigJtwiKhxYpoUJSdNM60wVEDCSUrZYBAuDH8pOxPfP+Tm4sokaFDTIG3QJFzOjC+/9nW4MUjAOFll9PCp9kaEFHJ/YmOYKMWNOCCPvLS6lxA83i0UaardkNLNoFS5paWfTlroxRwOC2T6PwO2ywKBgDjtXcSED61zK1seocQMyGRINnlWdhceD669kIHju/f6kAayvYKW3/lbJNXCmyinAccBosO08/0sUxvtuniIo18kfYJE0UmP1ReCjhMP+O+yOmwZJini/QelJk/Pez8IIDDWnY1qYQsN/q7ocjakOYrpGG7mig6JMFpDJtD6istR"

tests/AgentTests/FunctionalAPITests.hs

@@ -7,7 +7,7 @@
 
 module AgentTests.FunctionalAPITests (functionalAPITests) where
 
-import Control.Monad.Except (ExceptT, catchError, runExceptT)
+import Control.Monad.Except (ExceptT, runExceptT)
 import Control.Monad.IO.Unlift
 import SMPAgentClient
 import SMPClient (withSmpServer)
@@ -62,15 +62,20 @@ testAgentClient = do
     get bob ##> ("", aliceId, INFO "alice's connInfo")
     get bob ##> ("", aliceId, CON)
     InternalId 1 <- sendMessage alice bobId "hello"
+    get alice ##> ("", bobId, SENT 1)
     InternalId 2 <- sendMessage alice bobId "how are you?"
+    get alice ##> ("", bobId, SENT 2)
     get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
     get bob =##> \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False
     InternalId 3 <- sendMessage bob aliceId "hello too"
+    get bob ##> ("", aliceId, SENT 3)
     InternalId 4 <- sendMessage bob aliceId "message 1"
+    get bob ##> ("", aliceId, SENT 4)
     get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False
     get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False
     suspendConnection alice bobId
-    InternalId 0 <- sendMessage bob aliceId "message 2" `catchError` \(SMP AUTH) -> pure $ InternalId 0
+    InternalId 5 <- sendMessage bob aliceId "message 2"
+    get bob ##> ("", aliceId, MERR 5 (SMP AUTH))
     deleteConnection alice bobId
     liftIO $ noMessages alice "nothing else should be delivered to alice"
   pure ()
@@ -146,6 +151,8 @@ testAsyncBothOffline = do
 exchangeGreetings :: AgentClient -> ConnId -> AgentClient -> ConnId -> ExceptT AgentErrorType IO ()
 exchangeGreetings alice bobId bob aliceId = do
   InternalId 1 <- sendMessage alice bobId "hello"
+  get alice ##> ("", bobId, SENT 1)
   get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
   InternalId 2 <- sendMessage bob aliceId "hello too"
+  get bob ##> ("", aliceId, SENT 2)
   get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False

tests/AgentTests/SQLiteTests.hs

@@ -438,7 +438,8 @@ mkSndMsgData internalId internalSndId internalHash =
       internalSndId,
       internalTs = ts,
       msgBody = hw,
-      internalHash
+      internalHash,
+      previousMsgHash = internalHash
     }
 
 testCreateSndMsg' :: SQLiteStore -> PrevSndMsgHash -> ConnId -> SndMsgData -> Expectation

tests/SMPAgentClient.hs

@@ -96,11 +96,15 @@ smpAgentTestN_1 n test' = runSmpAgentTestN_1 n test' `shouldReturn` ()
 smpAgentTest2_2_2 :: forall c. Transport c => (c -> c -> IO ()) -> Expectation
 smpAgentTest2_2_2 test' =
   withSmpServerOn (transport @c) testPort2 $
-    smpAgentTestN
-      [ (agentTestPort, testPort, testDB),
-        (agentTestPort2, testPort2, testDB2)
-      ]
-      _test
+    smpAgentTest2_2_2_needs_server test'
+
+smpAgentTest2_2_2_needs_server :: forall c. Transport c => (c -> c -> IO ()) -> Expectation
+smpAgentTest2_2_2_needs_server test' =
+  smpAgentTestN
+    [ (agentTestPort, testPort, testDB),
+      (agentTestPort2, testPort2, testDB2)
+    ]
+    _test
   where
     _test [h1, h2] = test' h1 h2
     _test _ = error "expected 2 handles"
@@ -140,6 +144,15 @@ smpAgentTest3_1_1 test' = smpAgentTestN_1 3 _test
     _test [h1, h2, h3] = test' h1 h2 h3
     _test _ = error "expected 3 handles"
 
+smpAgentTest1_1_1 :: forall c. Transport c => (c -> IO ()) -> Expectation
+smpAgentTest1_1_1 test' =
+  smpAgentTestN
+    [(agentTestPort2, testPort2, testDB2)]
+    _test
+  where
+    _test [h] = test' h
+    _test _ = error "expected 1 handle"
+
 cfg :: AgentConfig
 cfg =
   defaultAgentConfig
@@ -156,12 +169,15 @@ cfg =
       retryInterval = (retryInterval defaultAgentConfig) {initialInterval = 50_000}
     }
 
-withSmpAgentThreadOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> (ThreadId -> m a) -> m a
-withSmpAgentThreadOn t (port', smpPort', db') =
+withSmpAgentThreadOn_ :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> m () -> (ThreadId -> m a) -> m a
+withSmpAgentThreadOn_ t (port', smpPort', db') afterProcess =
   let cfg' = cfg {tcpPort = port', dbFile = db', smpServers = L.fromList [SMPServer "localhost" (Just smpPort') testKeyHash]}
    in serverBracket
         (\started -> runSMPAgentBlocking t started cfg')
-        (removeFile db')
+        afterProcess
+
+withSmpAgentThreadOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> (ThreadId -> m a) -> m a
+withSmpAgentThreadOn t a@(_, _, db') = withSmpAgentThreadOn_ t a $ removeFile db'
 
 withSmpAgentOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> m a -> m a
 withSmpAgentOn t (port', smpPort', db') = withSmpAgentThreadOn t (port', smpPort', db') . const