smp server: refactor message delivery to always respond SOK to subscriptions

src/Simplex/Messaging/Protocol.hs

@@ -114,6 +114,7 @@ module Simplex.Messaging.Protocol
     BasicAuth (..),
     SrvLoc (..),
     CorrId (..),
+    pattern NoCorrId,
     EntityId (..),
     pattern NoEntity,
     QueueId,
@@ -1370,6 +1371,9 @@ newtype CorrId = CorrId {bs :: ByteString}
   deriving (Eq, Ord, Show)
   deriving newtype (Encoding)
 
+pattern NoCorrId :: CorrId
+pattern NoCorrId = CorrId ""
+
 instance IsString CorrId where
   fromString = CorrId . fromString
   {-# INLINE fromString #-}

src/Simplex/Messaging/Server.hs

@@ -66,12 +66,13 @@ import Data.IORef
 import Data.Int (Int64)
 import qualified Data.IntMap.Strict as IM
 import qualified Data.IntSet as IS
-import Data.List (foldl', intercalate, mapAccumR)
+import Data.Foldable (foldrM)
+import Data.List (foldl', intercalate)
 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 (catMaybes, fromMaybe, isJust, isNothing)
+import Data.Maybe (fromMaybe, isJust, isNothing)
 import Data.Semigroup (Sum (..))
 import qualified Data.Set as S
 import Data.Text (Text)
@@ -382,7 +383,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
                 let cancelNtfs s = throwSTM $ userError $ s <> ", " <> show (length ts_) <> " ntfs kept"
                 unlessM (currentClient readTVar c) $ cancelNtfs "not current client"
                 whenM (isFullTBQueue sndQ) $ cancelNtfs "sending queue full"
-                writeTBQueue sndQ ts
+                writeTBQueue sndQ (ts, [])
               pure $ length ts_
             currentClient :: Monad m => (forall a. TVar a -> m a) -> Client s' -> m Bool
             currentClient rd Client {clientId, connected} = (&&) <$> rd connected <*> (IS.member clientId <$> rd subClients)
@@ -393,7 +394,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
                 ntfs -> do
                   writeTVar v []
                   pure $ foldl' (\acc' ntf -> nmsg nId ntf : acc') acc ntfs -- reverses, to order by time
-            nmsg nId MsgNtf {ntfNonce, ntfEncMeta} = (CorrId "", nId, NMSG ntfNonce ntfEncMeta)
+            nmsg nId MsgNtf {ntfNonce, ntfEncMeta} = (NoCorrId, nId, NMSG ntfNonce ntfEncMeta)
             updateNtfStats :: Client s' -> Either SomeException Int -> IO ()
             updateNtfStats Client {clientId} = \case
               Right 0 -> pure ()
@@ -419,14 +420,14 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg, startOpt
           where
             enqueueEvts evts c@Client {connected, sndQ} =
               whenM (readTVarIO connected) $ do
-                sent <- atomically $ tryWriteTBQueue sndQ ts
+                sent <- atomically $ tryWriteTBQueue sndQ (ts, [])
                 if sent
                   then updateEndStats
                   else -- if queue is full it can block
                     forkClient c ("sendPendingEvtsThread.queueEvts") $
-                      atomically (writeTBQueue sndQ ts) >> updateEndStats
+                      atomically (writeTBQueue sndQ (ts, [])) >> updateEndStats
               where
-                ts = L.map (\(entId, evt) -> (CorrId "", entId, evt)) evts
+                ts = L.map (\(entId, evt) -> (NoCorrId, entId, evt)) evts
                 -- this accounts for both END and DELD events
                 updateEndStats = do
                   let len = L.length evts
@@ -1082,7 +1083,7 @@ receive h@THandle {params = THandleParams {thAuth, sessionId}} ms Client {rcvQ,
     atomically . (writeTVar rcvActiveAt $!) =<< getSystemTime
     let (es, ts') = partitionEithers $ L.toList ts
         errs = map (second ERR) es
-    case ts' of
+    errs' <- case ts' of
       (_, _, (_, _, Cmd p cmd)) : rest -> do
         let service = peerClientService =<< thAuth
         (errs', cmds) <- partitionEithers <$> case batchParty p of
@@ -1092,9 +1093,10 @@ receive h@THandle {params = THandleParams {thAuth, sessionId}} ms Client {rcvQ,
             qs <- getQueueRecs ms p $ map queueId ts'
             zipWithM (\t -> verified stats t . verifyLoadedQueue service thAuth t) ts' qs
           _ -> mapM (\t -> verified stats t =<< verifyTransmission ms service thAuth t) ts'
-        write rcvQ cmds
-        write sndQ $ errs ++ errs'
-      [] -> write sndQ errs
+        mapM_ (atomically . writeTBQueue rcvQ) $ L.nonEmpty cmds
+        pure $ errs ++ errs'
+      [] -> pure errs
+    mapM_ (atomically . writeTBQueue sndQ . (,[])) $ L.nonEmpty errs'
   where
     sameParty :: SParty p -> SignedTransmission Cmd -> Bool
     sameParty p (_, _, (_, _, Cmd p' _)) = isJust $ testEquality p p'
@@ -1116,37 +1118,32 @@ receive h@THandle {params = THandleParams {thAuth, sessionId}} ms Client {rcvQ,
             NSUB -> incStat $ ntfSubAuth stats
             GET -> incStat $ msgGetAuth stats
             _ -> pure ()
-    write q = mapM_ (atomically . writeTBQueue q) . L.nonEmpty
 
 send :: Transport c => MVar (THandleSMP c 'TServer) -> Client s -> IO ()
 send th c@Client {sndQ, msgQ, clientTHParams = THandleParams {sessionId}} = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " send"
   forever $ atomically (readTBQueue sndQ) >>= sendTransmissions
   where
-    sendTransmissions :: NonEmpty (Transmission BrokerMsg) -> IO ()
-    sendTransmissions ts
-      | L.length ts <= 2 = tSend th c ts
+    -- If the request had batched subscriptions
+    -- this will reply SOKs to all SUBs in the first batched transmission,
+    -- to reduce client timeouts.
+    -- After that all messages will be sent in separate transmissions,
+    -- without any client response timeouts, and allowing them to interleave
+    -- with other requests responses.
+    sendTransmissions :: (NonEmpty (Transmission BrokerMsg), [(RecipientId, RcvMessage)]) -> IO ()
+    sendTransmissions (ts, []) = tSend th c ts
+    sendTransmissions (ts, msg@(rId, encMsg) : msgs)
+      | length ts <= 4 = do -- up to 4 SOKs can be in one block with MSG (see testBatchSubResponses test)
+          tSend th c $ ts <> [(NoCorrId, rId, MSG encMsg)]
+          mapM_ (atomically . writeTBQueue msgQ) $ L.nonEmpty msgs
       | otherwise = do
-          let (msgs_, ts') = mapAccumR splitMessages [] ts
-          -- If the request had batched subscriptions and L.length ts > 2
-          -- this will reply OK to all SUBs in the first batched transmission,
-          -- to reduce client timeouts.
-          tSend th c ts'
-          -- After that all messages will be sent in separate transmissions,
-          -- without any client response timeouts, and allowing them to interleave
-          -- with other requests responses.
-          mapM_ (atomically . writeTBQueue msgQ) $ L.nonEmpty msgs_
-      where
-        splitMessages :: [Transmission BrokerMsg] -> Transmission BrokerMsg -> ([Transmission BrokerMsg], Transmission BrokerMsg)
-        splitMessages msgs t@(corrId, entId, cmd) = case cmd of
-          -- replace MSG response with OK, accumulating MSG in a separate list.
-          MSG {} -> ((CorrId "", entId, cmd) : msgs, (corrId, entId, OK))
-          _ -> (msgs, t)
+          tSend th c ts
+          atomically $ writeTBQueue msgQ (msg :| msgs)
 
 sendMsg :: Transport c => MVar (THandleSMP c 'TServer) -> Client s -> IO ()
 sendMsg th c@Client {msgQ, clientTHParams = THandleParams {sessionId}} = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " sendMsg"
-  forever $ atomically (readTBQueue msgQ) >>= mapM_ (\t -> tSend th c [t])
+  forever $ atomically (readTBQueue msgQ) >>= mapM_ (\(rId, msg) -> tSend th c [(NoCorrId, rId, MSG msg)])
 
 tSend :: Transport c => MVar (THandleSMP c 'TServer) -> Client s -> NonEmpty (Transmission BrokerMsg) -> IO ()
 tSend th Client {sndActiveAt} ts = do
@@ -1312,15 +1309,16 @@ client
     labelMyThread . B.unpack $ "client $" <> encode sessionId <> " commands"
     let THandleParams {thVersion} = thParams'
         service = peerClientService =<< thAuth thParams'
+        process t acc@(rs, msgs) =
+          (maybe acc (\(!r, !msg_) -> (r : rs, maybe msgs (: msgs) msg_)))
+            <$> processCommand service thVersion t
     forever $
       atomically (readTBQueue rcvQ)
-        >>= mapM (processCommand service thVersion)
-        >>= mapM_ reply . L.nonEmpty . catMaybes . L.toList
+        >>= foldrM process ([], [])
+        >>= \(rs_, msgs) -> mapM_ (atomically . writeTBQueue sndQ . (,msgs)) (L.nonEmpty rs_)
   where
-    reply :: MonadIO m => NonEmpty (Transmission BrokerMsg) -> m ()
-    reply = atomically . writeTBQueue sndQ
-    processProxiedCmd :: Transmission (Command 'ProxiedClient) -> M s (Maybe (Transmission BrokerMsg))
-    processProxiedCmd (corrId, EntityId sessId, command) = (corrId,EntityId sessId,) <$$> case command of
+    processProxiedCmd :: Transmission (Command 'ProxiedClient) -> M s (Maybe ResponseAndMessage)
+    processProxiedCmd (corrId, EntityId sessId, command) = (\t -> ((corrId, EntityId sessId, t), Nothing)) <$$> case command of
       PRXY srv auth -> ifM allowProxy getRelay (pure $ Just $ ERR $ PROXY BASIC_AUTH)
         where
           allowProxy = do
@@ -1383,7 +1381,7 @@ client
         forkProxiedCmd cmdAction = do
           bracket_ wait signal . forkClient clnt (B.unpack $ "client $" <> encode sessionId <> " proxy") $ do
             -- commands MUST be processed under a reasonable timeout or the client would halt
-            cmdAction >>= \t -> reply [(corrId, EntityId sessId, t)]
+            cmdAction >>= \t -> atomically $ writeTBQueue sndQ ([(corrId, EntityId sessId, t)], [])
           pure Nothing
           where
             wait = do
@@ -1399,32 +1397,32 @@ client
     mkIncProxyStats ps psOwn own sel = do
       incStat $ sel ps
       when own $ incStat $ sel psOwn
-    processCommand :: Maybe THPeerClientService -> VersionSMP -> VerifiedTransmission s -> M s (Maybe (Transmission BrokerMsg))
+    processCommand :: Maybe THPeerClientService -> VersionSMP -> VerifiedTransmission s -> M s (Maybe ResponseAndMessage)
     processCommand service clntVersion (q_, (corrId, entId, cmd)) = case cmd of
       Cmd SProxiedClient command -> processProxiedCmd (corrId, entId, command)
-      Cmd SSender command -> Just <$> case command of
+      Cmd SSender command -> response <$> case command of
         SKEY k -> withQueue $ \q qr -> checkMode QMMessaging qr $ secureQueue_ q k
-        SEND flags msgBody -> withQueue_ False $ sendMessage flags msgBody
-      Cmd SIdleClient PING -> pure $ Just (corrId, NoEntity, PONG)
-      Cmd SProxyService (RFWD encBlock) -> Just . (corrId,NoEntity,) <$> processForwardedCommand encBlock
-      Cmd SSenderLink command -> Just <$> case command of
+        SEND flags msgBody -> withQueue_ False err $ sendMessage flags msgBody
+      Cmd SIdleClient PING -> pure $ response (corrId, NoEntity, PONG)
+      Cmd SProxyService (RFWD encBlock) -> (response . (corrId,NoEntity,)) <$> processForwardedCommand encBlock
+      Cmd SSenderLink command -> response <$> case command of
         LKEY k -> withQueue $ \q qr -> checkMode QMMessaging qr $ secureQueue_ q k $>> getQueueLink_ q qr
         LGET -> withQueue $ \q qr -> checkContact qr $ getQueueLink_ q qr
-      Cmd SNotifier NSUB -> Just . (corrId,entId,) <$> case q_ of
+      Cmd SNotifier NSUB -> response . (corrId,entId,) <$> case q_ of
         Just (q, QueueRec {notifier = Just ntfCreds}) -> subscribeNotifications q ntfCreds
         _ -> pure $ ERR INTERNAL
-      Cmd SNotifierService NSUBS -> Just . (corrId,entId,) <$> case service of
+      Cmd SNotifierService NSUBS -> response . (corrId,entId,) <$> case service of
         Just s -> subscribeServiceNotifications s
         Nothing -> pure $ ERR INTERNAL
       Cmd SCreator (NEW nqr@NewQueueReq {auth_}) ->
-        Just <$> ifM allowNew (createQueue nqr) (pure (corrId, entId, ERR AUTH))
+        response <$> ifM allowNew (createQueue nqr) (pure (corrId, entId, ERR AUTH))
           where
             allowNew = do
               ServerConfig {allowNewQueues, newQueueBasicAuth} <- asks config
               pure $ allowNewQueues && maybe True ((== auth_) . Just) newQueueBasicAuth
+      Cmd SRecipient SUB -> Just <$> withQueue' subscribeQueueAndDeliver
       Cmd SRecipient command ->
-        Just <$> case command of
-          SUB -> withQueue subscribeQueue
+        response <$> case command of
           GET -> withQueue getMessage
           ACK msgId -> withQueue $ acknowledgeMsg msgId
           KEY sKey -> withQueue $ \q _ -> either err (corrId,entId,) <$> secureQueue_ q sKey
@@ -1443,12 +1441,12 @@ client
           OFF -> maybe (pure $ err INTERNAL) suspendQueue_ q_
           DEL -> maybe (pure $ err INTERNAL) delQueueAndMsgs q_
           QUE -> withQueue $ \q qr -> (corrId,entId,) <$> getQueueInfo q qr
-      Cmd SRecipientService SUBS -> pure $ Just $ err (CMD PROHIBITED) -- "TODO [certs rcv]"
+      Cmd SRecipientService SUBS -> pure $ response $ err (CMD PROHIBITED) -- "TODO [certs rcv]"
       where
         createQueue :: NewQueueReq -> M s (Transmission BrokerMsg)
         createQueue NewQueueReq {rcvAuthKey, rcvDhKey, subMode, queueReqData}
           | isJust service && subMode == SMOnlyCreate = pure (corrId, entId, ERR $ CMD PROHIBITED)
-          | otherwise = time "NEW" $ do
+          | otherwise = do
               g <- asks random
               idSize <- asks $ queueIdBytes . config
               updatedAt <- Just <$> liftIO getSystemDate
@@ -1506,7 +1504,7 @@ client
                             -- when (isJust ntf) $ incStat $ ntfCreated stats
                             case subMode of
                               SMOnlyCreate -> pure ()
-                              SMSubscribe -> void $ subscribeNewQueue rcvId qr -- no need to check if message is available, it's a new queue
+                              SMSubscribe -> void $ subscribeQueue rcvId qr -- no need to check if message is available, it's a new queue
                             pure $ IDS QIK {rcvId, sndId, rcvPublicDhKey, queueMode, linkId = fst <$> queueData, serviceId = rcvServiceId} -- , serverNtfCreds = snd <$> ntf
               (corrId,entId,) <$> tryCreate (3 :: Int)
 
@@ -1531,7 +1529,7 @@ client
         getQueueLink_ q qr = liftIO $ LNK (senderId qr) <$$> getQueueLinkData (queueStore ms) q entId
 
         addQueueNotifier_ :: StoreQueue s -> NtfPublicAuthKey -> RcvNtfPublicDhKey -> M s (Transmission BrokerMsg)
-        addQueueNotifier_ q notifierKey dhKey = time "NKEY" $ do
+        addQueueNotifier_ q notifierKey dhKey = do
           (rcvPublicDhKey, privDhKey) <- atomically . C.generateKeyPair =<< asks random
           let rcvNtfDhSecret = C.dh' dhKey privDhKey
           (corrId,entId,) <$> addNotifierRetry 3 rcvPublicDhKey rcvNtfDhSecret
@@ -1568,32 +1566,32 @@ client
         suspendQueue_ (q, _) = liftIO $ either err (const ok) <$> suspendQueue (queueStore ms) q
 
         -- TODO [certs rcv] if serviceId is passed, associate with the service and respond with SOK
-        subscribeQueue :: StoreQueue s -> QueueRec -> M s (Transmission BrokerMsg)
-        subscribeQueue q qr =
+        subscribeQueueAndDeliver :: StoreQueue s -> QueueRec -> M s ResponseAndMessage
+        subscribeQueueAndDeliver q qr =
           liftIO (TM.lookupIO rId subscriptions) >>= \case
-            Nothing -> subscribeNewQueue rId qr >>= deliver True
+            Nothing -> subscribeQueue rId qr >>= deliver True
             Just s@Sub {subThread} -> do
               stats <- asks serverStats
               case subThread of
                 ProhibitSub -> do
                   -- cannot use SUB in the same connection where GET was used
                   incStat $ qSubProhibited stats
-                  pure (corrId, rId, ERR $ CMD PROHIBITED)
+                  pure ((corrId, rId, ERR $ CMD PROHIBITED), Nothing)
                 _ -> do
                   incStat $ qSubDuplicate stats
                   atomically (tryTakeTMVar $ delivered s) >> deliver False s
           where
             rId = recipientId q
-            deliver :: Bool -> Sub -> M s (Transmission BrokerMsg)
+            deliver :: Bool -> Sub -> M s ResponseAndMessage
             deliver inc sub = do
               stats <- asks serverStats
-              fmap (either (\e -> (corrId, rId, ERR e)) id) $ liftIO $ runExceptT $ do
+              fmap (either (\e -> ((corrId, rId, ERR e), Nothing)) id) $ liftIO $ runExceptT $ do
                 msg_ <- tryPeekMsg ms q
-                liftIO $ when (inc && isJust msg_) $ incStat (qSub stats)
-                liftIO $ deliverMessage "SUB" qr rId sub msg_
+                mapM_ (\msg -> atomically (setDelivered sub msg) >> when inc (incStat $ qSub stats)) msg_
+                pure ((corrId, rId, SOK Nothing), (rId,) . encryptMsg qr <$> msg_)
 
-        subscribeNewQueue :: RecipientId -> QueueRec -> M s Sub
-        subscribeNewQueue rId QueueRec {rcvServiceId} = time "SUB newSub" . atomically $ do
+        subscribeQueue :: RecipientId -> QueueRec -> M s Sub
+        subscribeQueue rId QueueRec {rcvServiceId} = atomically $ do
           writeTQueue (subQ subscribers) (CSClient rId rcvServiceId Nothing, clientId)
           sub <- newSubscription NoSub
           TM.insert rId sub subscriptions
@@ -1601,7 +1599,7 @@ client
 
         -- clients that use GET are not added to server subscribers
         getMessage :: StoreQueue s -> QueueRec -> M s (Transmission BrokerMsg)
-        getMessage q qr = time "GET" $ do
+        getMessage q qr = do
           atomically (TM.lookup entId subscriptions) >>= \case
             Nothing ->
               atomically newSub >>= (`getMessage_` Nothing)
@@ -1637,19 +1635,22 @@ client
                   Nothing -> incStat (msgGetNoMsg stats) $> ok
 
         withQueue :: (StoreQueue s -> QueueRec -> M s (Transmission BrokerMsg)) -> M s (Transmission BrokerMsg)
-        withQueue = withQueue_ True
+        withQueue = withQueue_ True err
+
+        withQueue' :: (StoreQueue s -> QueueRec -> M s ResponseAndMessage) -> M s ResponseAndMessage
+        withQueue' = withQueue_ True ((,Nothing) . err)
 
         -- SEND passes queueNotBlocked False here to update time, but it fails anyway on blocked queues (see code for SEND).
-        withQueue_ :: Bool -> (StoreQueue s -> QueueRec -> M s (Transmission BrokerMsg)) -> M s (Transmission BrokerMsg)
-        withQueue_ queueNotBlocked action = case q_ of
-          Nothing -> pure $ err INTERNAL
+        withQueue_ :: Bool -> (ErrorType -> r) -> (StoreQueue s -> QueueRec -> M s r) -> M s r
+        withQueue_ queueNotBlocked err' action = case q_ of
+          Nothing -> pure $ err' INTERNAL
           Just (q, qr@QueueRec {status, updatedAt}) -> case status of
-            EntityBlocked info | queueNotBlocked -> pure $ err $ BLOCKED info
+            EntityBlocked info | queueNotBlocked -> pure $ err' $ BLOCKED info
             _ -> do
               t <- liftIO getSystemDate
               if updatedAt == Just t
                 then action q qr
-                else liftIO (updateQueueTime (queueStore ms) q t) >>= either (pure . err) (action q)
+                else liftIO (updateQueueTime (queueStore ms) q t) >>= either (pure . err') (action q)
 
         subscribeNotifications :: StoreQueue s -> NtfCreds -> M s BrokerMsg
         subscribeNotifications q NtfCreds {ntfServiceId} = do
@@ -1721,7 +1722,7 @@ client
             else pure $ SOKS srvSubs
 
         acknowledgeMsg :: MsgId -> StoreQueue s -> QueueRec -> M s (Transmission BrokerMsg)
-        acknowledgeMsg msgId q qr = time "ACK" $ do
+        acknowledgeMsg msgId q qr = do
           liftIO (TM.lookupIO entId subscriptions) >>= \case
             Nothing -> pure $ err NO_MSG
             Just sub ->
@@ -1736,8 +1737,10 @@ client
                         pure ok
                       _ -> do
                         (deletedMsg_, msg_) <- tryDelPeekMsg ms q msgId
-                        liftIO $ mapM_ (updateStats stats False) deletedMsg_
-                        liftIO $ deliverMessage "ACK" qr entId sub msg_
+                        liftIO $ do
+                          mapM_ (updateStats stats False) deletedMsg_
+                          mapM_ (atomically . setDelivered sub) msg_
+                          pure (corrId, entId, maybe OK (MSG . encryptMsg qr) msg_)
                 _ -> pure $ err NO_MSG
           where
             getDelivered :: Sub -> STM (Maybe ServerSub)
@@ -1788,7 +1791,7 @@ client
                         deleteQueueLinkData (queueStore ms) q
                       ServerConfig {messageExpiration, msgIdBytes} <- asks config
                       msgId <- randomId' msgIdBytes
-                      msg_ <- liftIO $ time "SEND" $ runExceptT $ do
+                      msg_ <- liftIO $ runExceptT $ do
                         expireMessages messageExpiration stats
                         msg <- liftIO $ mkMessage msgId body
                         writeMsg ms q True msg
@@ -1797,7 +1800,7 @@ client
                         Right Nothing -> do
                           incStat $ msgSentQuota stats
                           pure $ err QUOTA
-                        Right (Just (msg, wasEmpty)) -> time "SEND ok" $ do
+                        Right (Just (msg, wasEmpty)) -> do
                           when wasEmpty $ liftIO $ tryDeliverMessage msg
                           when (notification msgFlags) $ do
                             mapM_ (`enqueueNotification` msg) (notifier qr)
@@ -1856,7 +1859,7 @@ client
                         _ -> pure Nothing
                 deliver sndQ' s = do
                   let encMsg = encryptMsg qr msg
-                  writeTBQueue sndQ' [(CorrId "", rId, MSG encMsg)]
+                  writeTBQueue sndQ' ([(NoCorrId, rId, MSG encMsg)], [])
                   void $ setDelivered s msg
                 forkDeliver (rc@Client {sndQ = sndQ'}, s@Sub {delivered}, st) = do
                   t <- mkWeakThreadId =<< forkIO deliverThread
@@ -1870,7 +1873,7 @@ client
                       -- lookup can be outside of STM transaction,
                       -- as long as the check that it is the same client is inside.
                       getSubscribedClient rId (queueSubscribers subscribers) >>= mapM_ deliverIfSame
-                    deliverIfSame rcv = time "deliver" . atomically $
+                    deliverIfSame rcv = atomically $
                       whenM (sameClient rc rcv) $
                         tryTakeTMVar delivered >>= \case
                           Just _ -> pure () -- if a message was already delivered, should not deliver more
@@ -1916,7 +1919,8 @@ client
               Left r -> pure r
               -- rejectOrVerify filters allowed commands, no need to repeat it here.
               -- INTERNAL is used because processCommand never returns Nothing for sender commands (could be extracted for better types).
-              Right t''@(_, (corrId', entId', _)) -> fromMaybe (corrId', entId', ERR INTERNAL) <$> lift (processCommand Nothing fwdVersion t'')
+              -- `fst` removes empty message that is only returned for `SUB` command
+              Right t''@(_, (corrId', entId', _)) -> maybe (corrId', entId', ERR INTERNAL) fst <$> lift (processCommand Nothing fwdVersion t'')
           -- encode response
           r' <- case batchTransmissions clntTHParams [Right (Nothing, encodeTransmission clntTHParams r)] of
             [] -> throwE INTERNAL -- at least 1 item is guaranteed from NonEmpty/Right
@@ -1949,21 +1953,6 @@ client
                     VRVerified q -> Right (q, t'')
                     VRFailed e -> Left (corrId', entId', ERR e)
 
-        deliverMessage :: T.Text -> QueueRec -> RecipientId -> Sub -> Maybe Message -> IO (Transmission BrokerMsg)
-        deliverMessage name qr rId s@Sub {subThread} msg_ = time (name <> " deliver") . atomically $
-          case subThread of
-            ProhibitSub -> pure resp
-            _ -> case msg_ of
-              Just msg ->
-                let encMsg = encryptMsg qr msg
-                 in setDelivered s msg $> (corrId, rId, MSG encMsg)
-              _ -> pure resp
-          where
-            resp = (corrId, rId, OK)
-
-        time :: MonadIO m => T.Text -> m a -> m a
-        time name = timed name entId
-
         encryptMsg :: QueueRec -> Message -> RcvMessage
         encryptMsg qr msg = encrypt . encodeRcvMsgBody $ case msg of
           Message {msgFlags, msgBody} -> RcvMsgBody {msgTs = msgTs', msgFlags, msgBody}
@@ -2021,9 +2010,15 @@ client
 
         ok :: Transmission BrokerMsg
         ok = (corrId, entId, OK)
+        {-# INLINE ok #-}
 
         err :: ErrorType -> Transmission BrokerMsg
         err e = (corrId, entId, ERR e)
+        {-# INLINE err #-}
+
+        response :: Transmission BrokerMsg -> Maybe ResponseAndMessage
+        response = Just . (,Nothing)
+        {-# INLINE response #-}
 
 updateDeletedStats :: QueueRec -> M s ()
 updateDeletedStats q = do
@@ -2037,18 +2032,6 @@ incStat :: MonadIO m => IORef Int -> m ()
 incStat r = liftIO $ atomicModifyIORef'_ r (+ 1)
 {-# INLINE incStat #-}
 
-timed :: MonadIO m => T.Text -> RecipientId -> m a -> m a
-timed name (EntityId qId) a = do
-  t <- liftIO getSystemTime
-  r <- a
-  t' <- liftIO getSystemTime
-  let int = diff t t'
-  when (int > sec) . logDebug $ T.unwords [name, tshow $ encode qId, tshow int]
-  pure r
-  where
-    diff t t' = (systemSeconds t' - systemSeconds t) * sec + fromIntegral (systemNanoseconds t' - systemNanoseconds t)
-    sec = 1000_000000
-
 randomId' :: Int -> M s ByteString
 randomId' n = atomically . C.randomBytes n =<< asks random
 

src/Simplex/Messaging/Server/Env/STM.hs

@@ -40,6 +40,7 @@ module Simplex.Messaging.Server.Env.STM
     MsgStore (..),
     AStoreType (..),
     VerifiedTransmission,
+    ResponseAndMessage,
     newEnv,
     mkJournalStoreConfig,
     msgStore,
@@ -392,8 +393,8 @@ data Client s = Client
     serviceSubsCount :: TVar Int64, -- only one service can be subscribed, based on its certificate, this is subscription count
     ntfServiceSubsCount :: TVar Int64, -- only one service can be subscribed, based on its certificate, this is subscription count
     rcvQ :: TBQueue (NonEmpty (VerifiedTransmission s)),
-    sndQ :: TBQueue (NonEmpty (Transmission BrokerMsg)),
-    msgQ :: TBQueue (NonEmpty (Transmission BrokerMsg)),
+    sndQ :: TBQueue (NonEmpty (Transmission BrokerMsg), [(RecipientId, RcvMessage)]),
+    msgQ :: TBQueue (NonEmpty (RecipientId, RcvMessage)),
     procThreads :: TVar Int,
     endThreads :: TVar (IntMap (Weak ThreadId)),
     endThreadSeq :: TVar Int,
@@ -406,6 +407,8 @@ data Client s = Client
 
 type VerifiedTransmission s = (Maybe (StoreQueue s, QueueRec), Transmission Cmd)
 
+type ResponseAndMessage = (Transmission BrokerMsg, Maybe (RecipientId, RcvMessage))
+
 data ServerSub = ServerSub (TVar SubscriptionThread) | ProhibitSub
 
 data SubscriptionThread = NoSub | SubPending | SubThread (Weak ThreadId)

src/Simplex/Messaging/Server/QueueStore/Postgres.hs

@@ -43,14 +43,13 @@ import qualified Data.ByteString.Builder as BB
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Lazy as LB
 import Data.Bitraversable (bimapM)
-import Data.Either (fromRight, lefts, rights)
+import Data.Either (fromRight, lefts)
 import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.List (foldl', intersperse, partition)
 import Data.List.NonEmpty (NonEmpty)
-import qualified Data.List.NonEmpty as L
 import qualified Data.Map.Strict as M
-import Data.Maybe (catMaybes, fromMaybe, mapMaybe)
+import Data.Maybe (catMaybes, fromMaybe)
 import qualified Data.Set as S
 import Data.Text (Text)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
@@ -64,7 +63,7 @@ import Database.PostgreSQL.Simple.ToField (Action (..), ToField (..))
 import Database.PostgreSQL.Simple.Errors (ConstraintViolation (..), constraintViolation)
 import Database.PostgreSQL.Simple.SqlQQ (sql)
 import GHC.IO (catchAny)
-import Simplex.Messaging.Agent.Client (withLockMap, withLocksMap)
+import Simplex.Messaging.Agent.Client (withLockMap)
 import Simplex.Messaging.Agent.Lock (Lock)
 import Simplex.Messaging.Agent.Store.AgentStore ()
 import Simplex.Messaging.Agent.Store.Postgres (createDBStore, closeDBStore)
@@ -83,7 +82,7 @@ import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (SMPServiceRole (..))
-import Simplex.Messaging.Util (eitherToMaybe, firstRow, ifM, maybeFirstRow, tshow, (<$$>), ($>>=))
+import Simplex.Messaging.Util (eitherToMaybe, firstRow, ifM, maybeFirstRow, tshow, (<$$>))
 import System.Exit (exitFailure)
 import System.IO (IOMode (..), hFlush, stdout)
 import UnliftIO.STM