rename simplexmq-legacy into simplexmq-v0 (#270)

* rename simplexmq-legacy into simplexmq-v0

* split UnliftIO instance of ExceptT

.github/workflows/build.yml

@@ -4,6 +4,7 @@ on:
   push:
     branches:
       - master
+      - stable
     tags:
       - "v*"
   pull_request:

CHANGELOG.md

@@ -1,3 +1,11 @@
+# 0.5.2
+
+- Fix message delivery logic that blocked delivery of all server messages when server per-queue quota exceeded, making it concurrent per SMP queue, not per server.
+
+# 0.5.1
+
+- Fix server subscription logic bug that was leading to memory leak / resource exhaustion in some edge cases.
+
 # 0.5.0
 
 - No changes in SMP server implementation - it is backwards compatible with v0.4.1

README.md

@@ -3,6 +3,14 @@
 [![GitHub build](https://github.com/simplex-chat/simplexmq/workflows/build/badge.svg)](https://github.com/simplex-chat/simplexmq/actions?query=workflow%3Abuild)
 [![GitHub release](https://img.shields.io/github/v/release/simplex-chat/simplexmq)](https://github.com/simplex-chat/simplexmq/releases)
 
+📢 **v0.5.1 brings a hotfix to the server's subscription management logic, to apply it log in to your server via SSH and run the following command. If you have store log enabled for your server, information about already established queues will be preserved.** If you're doing a custom installation instead of Linode or DigitalOcean you may have to change the path for binary download.
+
+```sh
+systemctl stop smp-server
+curl -L -o /opt/simplex/bin/smp-server https://github.com/simplex-chat/simplexmq/releases/download/v0.5.1/smp-server-ubuntu-20_04-x86-64
+systemctl start smp-server
+```
+
 ## Message broker for unidirectional (simplex) queues
 
 SimpleXMQ is a message broker for managing message queues and sending messages over public network. It consists of SMP server, SMP client library and SMP agent that implement [SMP protocol](https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md) for client-server communication and [SMP agent protocol](https://github.com/simplex-chat/simplexmq/blob/master/protocol/agent-protocol.md) to manage duplex connections via simplex queues on multiple SMP servers.

apps/smp-server/Main.hs

@@ -41,6 +41,7 @@ serverConfig :: ServerConfig
 serverConfig =
   ServerConfig
     { tbqSize = 16,
+      serverTbqSize = 128,
       msgQueueQuota = 256,
       queueIdBytes = 12,
       msgIdBytes = 6,

package.yaml

@@ -1,5 +1,5 @@
-name: simplexmq
-version: 0.5.0
+name: simplexmq-v0
+version: 0.5.3
 synopsis: SimpleXMQ message broker
 description:  |
   This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -37,7 +37,7 @@ dependencies:
   - cryptonite >= 0.27 && < 0.30
   - direct-sqlite == 2.3.*
   - directory == 1.3.*
-  - file-embed == 0.0.14.*
+  - file-embed >= 0.0.14.0 && <= 0.0.15.0
   - filepath == 1.4.*
   - http-types == 0.12.*
   - generic-random >= 1.3 && < 1.5
@@ -71,7 +71,7 @@ executables:
       - cryptostore == 0.2.*
       - ini == 0.4.*
       - optparse-applicative >= 0.15 && < 0.17
-      - simplexmq
+      - simplexmq-v0
     ghc-options:
       - -threaded
 
@@ -79,7 +79,7 @@ executables:
     source-dirs: apps/smp-agent
     main: Main.hs
     dependencies:
-      - simplexmq
+      - simplexmq-v0
     ghc-options:
       - -threaded
 
@@ -88,7 +88,7 @@ tests:
     source-dirs: tests
     main: Test.hs
     dependencies:
-      - simplexmq
+      - simplexmq-v0
       - hspec == 2.7.*
       - hspec-core == 2.7.*
       - HUnit == 1.6.*

simplexmq-v0.cabal

@@ -3,11 +3,9 @@ cabal-version: 1.12
 -- This file has been generated from package.yaml by hpack version 0.34.4.
 --
 -- see: https://github.com/sol/hpack
---
--- hash: 3bdb491a0318dc4b53cba4131f192e4c4e17b42e88043495666d1688a1f95443
 
-name:           simplexmq
-version:        0.5.0
+name:           simplexmq-v0
+version:        0.5.3
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -39,6 +37,7 @@ library
       Simplex.Messaging.Agent
       Simplex.Messaging.Agent.Client
       Simplex.Messaging.Agent.Env.SQLite
+      Simplex.Messaging.Agent.ExceptT
       Simplex.Messaging.Agent.Protocol
       Simplex.Messaging.Agent.RetryInterval
       Simplex.Messaging.Agent.Store
@@ -59,7 +58,7 @@ library
       Simplex.Messaging.Transport.WebSockets
       Simplex.Messaging.Util
   other-modules:
-      Paths_simplexmq
+      Paths_simplexmq_v0
   hs-source-dirs:
       src
   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns
@@ -79,7 +78,7 @@ library
     , cryptonite >=0.27 && <0.30
     , direct-sqlite ==2.3.*
     , directory ==1.3.*
-    , file-embed ==0.0.14.*
+    , file-embed >=0.0.14.0 && <=0.0.15.0
     , filepath ==1.4.*
     , generic-random >=1.3 && <1.5
     , http-types ==0.12.*
@@ -105,7 +104,7 @@ library
 executable smp-agent
   main-is: Main.hs
   other-modules:
-      Paths_simplexmq
+      Paths_simplexmq_v0
   hs-source-dirs:
       apps/smp-agent
   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
@@ -125,7 +124,7 @@ executable smp-agent
     , cryptonite >=0.27 && <0.30
     , direct-sqlite ==2.3.*
     , directory ==1.3.*
-    , file-embed ==0.0.14.*
+    , file-embed >=0.0.14.0 && <=0.0.15.0
     , filepath ==1.4.*
     , generic-random >=1.3 && <1.5
     , http-types ==0.12.*
@@ -136,7 +135,7 @@ executable smp-agent
     , network-transport ==0.5.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
-    , simplexmq
+    , simplexmq-v0
     , sqlite-simple ==0.4.*
     , stm ==2.5.*
     , template-haskell ==2.16.*
@@ -152,7 +151,7 @@ executable smp-agent
 executable smp-server
   main-is: Main.hs
   other-modules:
-      Paths_simplexmq
+      Paths_simplexmq_v0
   hs-source-dirs:
       apps/smp-server
   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
@@ -173,7 +172,7 @@ executable smp-server
     , cryptostore ==0.2.*
     , direct-sqlite ==2.3.*
     , directory ==1.3.*
-    , file-embed ==0.0.14.*
+    , file-embed >=0.0.14.0 && <=0.0.15.0
     , filepath ==1.4.*
     , generic-random >=1.3 && <1.5
     , http-types ==0.12.*
@@ -186,7 +185,7 @@ executable smp-server
     , optparse-applicative >=0.15 && <0.17
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
-    , simplexmq
+    , simplexmq-v0
     , sqlite-simple ==0.4.*
     , stm ==2.5.*
     , template-haskell ==2.16.*
@@ -211,7 +210,7 @@ test-suite smp-server-test
       ServerTests
       SMPAgentClient
       SMPClient
-      Paths_simplexmq
+      Paths_simplexmq_v0
   hs-source-dirs:
       tests
   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns
@@ -232,7 +231,7 @@ test-suite smp-server-test
     , cryptonite >=0.27 && <0.30
     , direct-sqlite ==2.3.*
     , directory ==1.3.*
-    , file-embed ==0.0.14.*
+    , file-embed >=0.0.14.0 && <=0.0.15.0
     , filepath ==1.4.*
     , generic-random >=1.3 && <1.5
     , hspec ==2.7.*
@@ -245,7 +244,7 @@ test-suite smp-server-test
     , network-transport ==0.5.*
     , random >=1.1 && <1.3
     , simple-logger ==0.1.*
-    , simplexmq
+    , simplexmq-v0
     , sqlite-simple ==0.4.*
     , stm ==2.5.*
     , template-haskell ==2.16.*

src/Simplex/Messaging/Agent.hs

@@ -69,7 +69,6 @@ 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
@@ -87,9 +86,9 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol (MsgBody, SenderPublicKey)
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), currentSMPVersionStr, runTransportServer)
-import Simplex.Messaging.Util (bshow, tryError)
+import Simplex.Messaging.Util (bshow, tryError, unlessM)
 import System.Random (randomR)
-import UnliftIO.Async (Async, async, race_)
+import UnliftIO.Async (async, race_)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
@@ -334,7 +333,7 @@ subscribeConnection' :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()
 subscribeConnection' c connId =
   withStore (`getConn` connId) >>= \case
     SomeConn _ (DuplexConnection _ rq sq) -> do
-      resumeDelivery sq
+      resumeMsgDelivery c connId sq
       case status (sq :: SndQueue) of
         Confirmed -> withVerifyKey sq $ \verifyKey -> do
           conf <- withStore (`getAcceptedConfirmation` connId)
@@ -345,7 +344,7 @@ subscribeConnection' c connId =
         Active -> subscribeQueue c rq connId
         _ -> throwError $ INTERNAL "unexpected queue status"
     SomeConn _ (SndConnection _ sq) -> do
-      resumeDelivery sq
+      resumeMsgDelivery c connId sq
       case status (sq :: SndQueue) of
         Confirmed -> withVerifyKey sq $ \verifyKey ->
           activateQueueJoining c connId sq verifyKey =<< resumeInterval
@@ -354,12 +353,6 @@ subscribeConnection' c connId =
     SomeConn _ (RcvConnection _ rq) -> subscribeQueue c rq connId
     SomeConn _ (ContactConnection _ 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"
@@ -382,14 +375,10 @@ sendMessage' c connId msg =
     _ -> throwError $ CONN SIMPLEX
   where
     enqueueMessage :: SndQueue -> m AgentMsgId
-    enqueueMessage SndQueue {server} = do
+    enqueueMessage sq = do
+      resumeMsgDelivery c connId sq
       msgId <- storeSentMsg
-      wasDelivering <- resumeMsgDelivery c connId server
-      pending <-
-        if wasDelivering
-          then pure [PendingMsg {connId, msgId}]
-          else withStore (`getPendingMsgs` connId)
-      queuePendingMsgs c connId pending
+      queuePendingMsgs c connId sq [msgId]
       pure $ unId msgId
       where
         storeSentMsg :: m InternalId
@@ -410,63 +399,63 @@ sendMessage' c connId msg =
             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
+resumeMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnId -> SndQueue -> m ()
+resumeMsgDelivery c connId sq@SndQueue {server, sndId} = do
+  let qKey = (connId, server, sndId)
+  unlessM (queueDelivering qKey) $
+    async (runSmpQueueMsgDelivery c connId sq)
+      >>= atomically . modifyTVar (smpQueueMsgDeliveries c) . M.insert qKey
+  unlessM connQueued $
+    withStore (`getPendingMsgs` connId)
+      >>= queuePendingMsgs c connId sq
   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
+    queueDelivering qKey = isJust . M.lookup qKey <$> readTVarIO (smpQueueMsgDeliveries c)
+    connQueued =
+      atomically $
+        isJust
+          <$> stateTVar
+            (connMsgsQueued c)
+            (\m -> (M.lookup connId m, M.insert connId True m))
 
-queuePendingMsgs :: AgentMonad m => AgentClient -> ConnId -> [PendingMsg] -> m ()
-queuePendingMsgs c connId pending =
-  atomically $ getPendingMsgQ connId (connMsgQueues c) >>= forM_ pending . writeTQueue
+queuePendingMsgs :: AgentMonad m => AgentClient -> ConnId -> SndQueue -> [InternalId] -> m ()
+queuePendingMsgs c connId sq msgIds = atomically $ do
+  q <- getPendingMsgQ c connId sq
+  mapM_ (writeTQueue q) msgIds
 
-getPendingMsgQ :: Ord a => a -> TVar (Map a (TQueue PendingMsg)) -> STM (TQueue PendingMsg)
-getPendingMsgQ key queueMap = do
-  maybe newMsgQueue pure . M.lookup key =<< readTVar queueMap
+getPendingMsgQ :: AgentClient -> ConnId -> SndQueue -> STM (TQueue InternalId)
+getPendingMsgQ c connId SndQueue {server, sndId} = do
+  let qKey = (connId, server, sndId)
+  maybe (newMsgQueue qKey) pure . M.lookup qKey =<< readTVar (smpQueueMsgQueues c)
   where
-    newMsgQueue :: STM (TQueue PendingMsg)
-    newMsgQueue = do
+    newMsgQueue qKey = do
       mq <- newTQueue
-      modifyTVar queueMap $ M.insert key mq
+      modifyTVar (smpQueueMsgQueues c) $ M.insert qKey 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
+runSmpQueueMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnId -> SndQueue -> m ()
+runSmpQueueMsgDelivery c@AgentClient {subQ} connId sq = do
+  mq <- atomically $ getPendingMsgQ c connId sq
   ri <- asks $ reconnectInterval . config
   forever $ do
-    PendingMsg {connId, msgId} <- atomically $ readTQueue mq
+    msgId <- atomically $ readTQueue mq
     let mId = unId msgId
     withStore (\st -> E.try $ getPendingMsgData st connId msgId) >>= \case
       Left (e :: E.SomeException) ->
-        notify connId $ MERR mId (INTERNAL $ show e)
-      Right (sq, msgBody) -> do
+        notify $ MERR mId (INTERNAL $ show e)
+      Right msgBody -> do
         withRetryInterval ri $ \loop -> do
           tryError (sendAgentMessage c sq msgBody) >>= \case
             Left e -> case e of
               SMP SMP.QUOTA -> loop
-              SMP {} -> notify connId $ MERR mId e
-              CMD {} -> notify connId $ MERR mId e
+              SMP {} -> notify $ MERR mId e
+              CMD {} -> notify $ MERR mId e
               _ -> loop
             Right () -> do
-              notify connId $ SENT mId
+              notify $ SENT mId
               withStore $ \st -> updateSndMsgStatus st connId msgId SndMsgSent
   where
-    notify :: ConnId -> ACommand 'Agent -> m ()
-    notify connId cmd = atomically $ writeTBQueue subQ ("", connId, cmd)
+    notify :: ACommand 'Agent -> m ()
+    notify cmd = atomically $ writeTBQueue subQ ("", connId, cmd)
 
 ackMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
 ackMessage' c connId msgId = do

src/Simplex/Messaging/Agent/Client.hs

@@ -54,12 +54,14 @@ import qualified Data.Set as S
 import Data.Text.Encoding
 import Data.Time.Clock
 import Simplex.Messaging.Agent.Env.SQLite
+import Simplex.Messaging.Agent.ExceptT ()
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Client
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgBody, QueueId, SenderPublicKey)
+import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Util (bshow, liftEitherError, liftError)
 import UnliftIO.Exception (IOException)
 import qualified UnliftIO.Exception as E
@@ -73,10 +75,9 @@ 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 ())),
+    connMsgsQueued :: TVar (Map ConnId Bool),
+    smpQueueMsgQueues :: TVar (Map (ConnId, SMPServer, SMP.SenderId) (TQueue InternalId)),
+    smpQueueMsgDeliveries :: TVar (Map (ConnId, SMPServer, SMP.SenderId) (Async ())),
     reconnections :: TVar [Async ()],
     clientId :: Int,
     agentEnv :: Env,
@@ -94,14 +95,13 @@ 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
+  connMsgsQueued <- newTVar M.empty
+  smpQueueMsgQueues <- newTVar M.empty
+  smpQueueMsgDeliveries <- 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, connMsgQueues, connMsgDeliveries, srvMsgQueues, srvMsgDeliveries, reconnections, clientId, agentEnv, smpSubscriber = undefined, lock}
+  return AgentClient {rcvQ, subQ, msgQ, smpClients, subscrSrvrs, subscrConns, activations, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, 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)
@@ -177,8 +177,7 @@ closeAgentClient c = liftIO $ do
   closeSMPServerClients c
   cancelActions $ activations c
   cancelActions $ reconnections c
-  cancelActions $ connMsgDeliveries c
-  cancelActions $ srvMsgDeliveries c
+  cancelActions $ smpQueueMsgDeliveries c
 
 closeSMPServerClients :: AgentClient -> IO ()
 closeSMPServerClients c = readTVarIO (smpClients c) >>= mapM_ closeSMPClient

src/Simplex/Messaging/Agent/ExceptT.hs

@@ -0,0 +1,22 @@
+{-# LANGUAGE InstanceSigs #-}
+{-# LANGUAGE RankNTypes #-}
+{-# OPTIONS_GHC -fno-warn-orphans #-}
+
+module Simplex.Messaging.Agent.ExceptT where
+
+import Control.Monad.Except
+import Control.Monad.IO.Unlift
+import UnliftIO.Exception (Exception)
+import qualified UnliftIO.Exception as E
+
+newtype InternalException e = InternalException {unInternalException :: e}
+  deriving (Eq, Show)
+
+instance Exception e => Exception (InternalException e)
+
+instance (MonadUnliftIO m, Exception e) => MonadUnliftIO (ExceptT e m) where
+  withRunInIO :: ((forall a. ExceptT e m a -> IO a) -> IO b) -> ExceptT e m b
+  withRunInIO exceptToIO =
+    withExceptT unInternalException . ExceptT . E.try $
+      withRunInIO $ \run ->
+        exceptToIO $ run . (either (E.throwIO . InternalException) return <=< runExceptT)

src/Simplex/Messaging/Agent/Store.hs

@@ -63,8 +63,8 @@ class Monad m => MonadAgentStore s m where
   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]
+  getPendingMsgData :: s -> ConnId -> InternalId -> m MsgBody
+  getPendingMsgs :: s -> ConnId -> m [InternalId]
   getMsg :: s -> ConnId -> InternalId -> m Msg
   checkRcvMsg :: s -> ConnId -> InternalId -> m ()
   updateRcvMsgAck :: s -> ConnId -> InternalId -> m ()

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

@@ -461,34 +461,28 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
           ":snd_status" := msgStatus
         ]
 
-  getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m (SndQueue, MsgBody)
+  getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m 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)
+    liftIOEither . withTransaction st $ \db ->
+      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)
     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 :: SQLiteStore -> ConnId -> m [InternalId]
   getPendingMsgs st connId =
     liftIO . withTransaction st $ \db ->
-      map (PendingMsg connId . fromOnly)
+      map fromOnly
         <$> DB.query db "SELECT internal_id FROM snd_messages WHERE conn_alias = ? AND snd_status = ?" (connId, SndMsgCreated)
 
   getMsg :: SQLiteStore -> ConnId -> InternalId -> m Msg

src/Simplex/Messaging/Server.hs

@@ -79,13 +79,25 @@ runSMPServerBlocking started cfg@ServerConfig {transports} = do
     runServer (tcpPort, ATransport t) = runTransportServer started tcpPort (runClient t)
 
     serverThread :: MonadUnliftIO m' => Server -> m' ()
-    serverThread Server {subscribedQ, subscribers} = forever . atomically $ do
-      (rId, clnt) <- readTBQueue subscribedQ
-      cs <- readTVar subscribers
-      case M.lookup rId cs of
-        Just Client {rcvQ} -> writeTBQueue rcvQ (CorrId B.empty, rId, Cmd SBroker END)
-        Nothing -> return ()
-      writeTVar subscribers $ M.insert rId clnt cs
+    serverThread Server {subscribedQ, subscribers} = forever $ do
+      atomically updateSubscribers >>= \case
+        Just (rId, Client {rcvQ}) ->
+          void . forkIO . atomically $
+            writeTBQueue rcvQ (CorrId "", rId, Cmd SBroker END)
+        _ -> pure ()
+      where
+        updateSubscribers :: STM (Maybe (RecipientId, Client))
+        updateSubscribers = do
+          (rId, c) <- readTBQueue subscribedQ
+          stateTVar subscribers (\cs -> (M.lookup rId cs, M.insert rId c cs)) >>= \case
+            Just c' -> clientToBeNotified rId c c'
+            _ -> pure Nothing
+        clientToBeNotified :: RecipientId -> Client -> Client -> STM (Maybe (RecipientId, Client))
+        clientToBeNotified rId c c'@Client {connected}
+          | clientId c /= clientId c' = do
+            yes <- readTVar connected
+            pure $ if yes then Just (rId, c') else Nothing
+          | otherwise = pure Nothing
 
 runClient :: (Transport c, MonadUnliftIO m, MonadReader Env m) => TProxy c -> c -> m ()
 runClient _ h = do
@@ -98,14 +110,23 @@ runClient _ h = do
 runClientTransport :: (Transport c, MonadUnliftIO m, MonadReader Env m) => THandle c -> m ()
 runClientTransport th = do
   q <- asks $ tbqSize . config
-  c <- atomically $ newClient q
   s <- asks server
+  c <- atomically $ newClient s q
   raceAny_ [send th c, client c s, receive th c]
-    `finally` cancelSubscribers c
+    `finally` clientDisconnected c
 
-cancelSubscribers :: MonadUnliftIO m => Client -> m ()
-cancelSubscribers Client {subscriptions} =
-  readTVarIO subscriptions >>= mapM_ cancelSub
+clientDisconnected :: (MonadUnliftIO m, MonadReader Env m) => Client -> m ()
+clientDisconnected c@Client {subscriptions, connected} = do
+  atomically $ writeTVar connected False
+  subs <- readTVarIO subscriptions
+  mapM_ cancelSub subs
+  cs <- asks $ subscribers . server
+  atomically . mapM_ (modifyTVar cs . M.update deleteCurrentClient) $ M.keys subs
+  where
+    deleteCurrentClient :: Client -> Maybe Client
+    deleteCurrentClient c'
+      | clientId c == clientId c' = Nothing
+      | otherwise = Just c'
 
 cancelSub :: MonadUnliftIO m => Sub -> m ()
 cancelSub = \case
@@ -326,7 +347,7 @@ client clnt@Client {subscriptions, rcvQ, sndQ} Server {subscribedQ} =
             subscriber :: MsgQueue -> m ()
             subscriber q = atomically $ do
               msg <- peekMsg q
-              writeTBQueue sndQ $ mkResp (CorrId B.empty) rId (msgCmd msg)
+              writeTBQueue sndQ $ mkResp (CorrId "") rId (msgCmd msg)
               setSub (\s -> s {subThread = NoSub})
               void setDelivered
 

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

@@ -6,6 +6,7 @@
 module Simplex.Messaging.Server.Env.STM where
 
 import Control.Concurrent (ThreadId)
+import Control.Concurrent.STM (stateTVar)
 import Control.Monad.IO.Unlift
 import Crypto.Random
 import Data.Map.Strict (Map)
@@ -25,6 +26,7 @@ import UnliftIO.STM
 data ServerConfig = ServerConfig
   { transports :: [(ServiceName, ATransport)],
     tbqSize :: Natural,
+    serverTbqSize :: Natural,
     msgQueueQuota :: Natural,
     queueIdBytes :: Int,
     msgIdBytes :: Int,
@@ -46,13 +48,16 @@ data Env = Env
 
 data Server = Server
   { subscribedQ :: TBQueue (RecipientId, Client),
-    subscribers :: TVar (Map RecipientId Client)
+    subscribers :: TVar (Map RecipientId Client),
+    nextClientId :: TVar Natural
   }
 
 data Client = Client
   { subscriptions :: TVar (Map RecipientId Sub),
     rcvQ :: TBQueue Transmission,
-    sndQ :: TBQueue Transmission
+    sndQ :: TBQueue Transmission,
+    clientId :: Natural,
+    connected :: TVar Bool
   }
 
 data SubscriptionThread = NoSub | SubPending | SubThread ThreadId
@@ -66,14 +71,17 @@ newServer :: Natural -> STM Server
 newServer qSize = do
   subscribedQ <- newTBQueue qSize
   subscribers <- newTVar M.empty
-  return Server {subscribedQ, subscribers}
+  nextClientId <- newTVar 0
+  return Server {subscribedQ, subscribers, nextClientId}
 
-newClient :: Natural -> STM Client
-newClient qSize = do
+newClient :: Server -> Natural -> STM Client
+newClient Server {nextClientId} qSize = do
   subscriptions <- newTVar M.empty
   rcvQ <- newTBQueue qSize
   sndQ <- newTBQueue qSize
-  return Client {subscriptions, rcvQ, sndQ}
+  clientId <- stateTVar nextClientId $ \i -> (i, i + 1)
+  connected <- newTVar True
+  return Client {subscriptions, rcvQ, sndQ, clientId, connected}
 
 newSubscription :: STM Sub
 newSubscription = do
@@ -82,7 +90,7 @@ newSubscription = do
 
 newEnv :: forall m. (MonadUnliftIO m, MonadRandom m) => ServerConfig -> m Env
 newEnv config = do
-  server <- atomically $ newServer (tbqSize config)
+  server <- atomically $ newServer (serverTbqSize config)
   queueStore <- atomically newQueueStore
   msgStore <- atomically newMsgStore
   idsDrg <- drgNew >>= newTVarIO

src/Simplex/Messaging/Transport.hs

@@ -222,7 +222,7 @@ major :: SMPVersion -> (Int, Int)
 major (SMPVersion a b _ _) = (a, b)
 
 currentSMPVersion :: SMPVersion
-currentSMPVersion = "0.5.0.0"
+currentSMPVersion = "0.5.1.0"
 
 currentSMPVersionStr :: ByteString
 currentSMPVersionStr = serializeSMPVersion currentSMPVersion

src/Simplex/Messaging/Util.hs

@@ -1,4 +1,3 @@
-{-# LANGUAGE InstanceSigs #-}
 {-# LANGUAGE RankNTypes #-}
 {-# OPTIONS_GHC -fno-warn-orphans #-}
 
@@ -10,20 +9,6 @@ import Data.Bifunctor (first)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import UnliftIO.Async
-import UnliftIO.Exception (Exception)
-import qualified UnliftIO.Exception as E
-
-newtype InternalException e = InternalException {unInternalException :: e}
-  deriving (Eq, Show)
-
-instance Exception e => Exception (InternalException e)
-
-instance (MonadUnliftIO m, Exception e) => MonadUnliftIO (ExceptT e m) where
-  withRunInIO :: ((forall a. ExceptT e m a -> IO a) -> IO b) -> ExceptT e m b
-  withRunInIO exceptToIO =
-    withExceptT unInternalException . ExceptT . E.try $
-      withRunInIO $ \run ->
-        exceptToIO $ run . (either (E.throwIO . InternalException) return <=< runExceptT)
 
 raceAny_ :: MonadUnliftIO m => [m a] -> m ()
 raceAny_ = r []
@@ -53,3 +38,11 @@ liftEitherError f a = liftIOEither (first f <$> a)
 
 tryError :: MonadError e m => m a -> m (Either e a)
 tryError action = (Right <$> action) `catchError` (pure . Left)
+
+ifM :: Monad m => m Bool -> m a -> m a -> m a
+ifM ba t f = ba >>= \b -> if b then t else f
+{-# INLINE ifM #-}
+
+unlessM :: Monad m => m Bool -> m () -> m ()
+unlessM b = ifM b $ pure ()
+{-# INLINE unlessM #-}

stack.yaml

@@ -17,7 +17,7 @@
 #
 # resolver: ./custom-snapshot.yaml
 # resolver: https://example.com/snapshots/2018-01-01.yaml
-resolver: lts-18.0
+resolver: lts-18.21
 
 # User packages to be built.
 # Various formats can be used as shown in the example below.
@@ -36,9 +36,7 @@ packages:
 #
 extra-deps:
   - cryptostore-0.2.1.0@sha256:9896e2984f36a1c8790f057fd5ce3da4cbcaf8aa73eb2d9277916886978c5b19,3881
-  - direct-sqlite-2.3.26@sha256:04e835402f1508abca383182023e4e2b9b86297b8533afbd4e57d1a5652e0c23,3718
   - simple-logger-0.1.0@sha256:be8ede4bd251a9cac776533bae7fb643369ebd826eb948a9a18df1a8dd252ff8,1079
-  - sqlite-simple-0.4.18.0@sha256:3ceea56375c0a3590c814e411a4eb86943f8d31b93b110ca159c90689b6b39e5,3002
 # - network-run-0.2.4@sha256:7dbb06def522dab413bce4a46af476820bffdff2071974736b06f52f4ab57c96,885
 # - git: https://github.com/commercialhaskell/stack.git
 #   commit: e7b331f14bcffb8367cd58fbfc8b40ec7642100a

tests/AgentTests.hs

@@ -14,6 +14,7 @@ import AgentTests.ConnectionRequestTests
 import AgentTests.FunctionalAPITests (functionalAPITests)
 import AgentTests.SQLiteTests (storeTests)
 import Control.Concurrent
+import Control.Monad (forM_)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Network.HTTP.Types (urlEncode)
@@ -23,6 +24,7 @@ import Simplex.Messaging.Agent.Protocol
 import qualified Simplex.Messaging.Agent.Protocol as A
 import Simplex.Messaging.Protocol (ErrorType (..), MsgBody)
 import Simplex.Messaging.Transport (ATransport (..), TProxy (..), Transport (..))
+import Simplex.Messaging.Util (bshow)
 import System.Directory (removeFile)
 import System.Timeout
 import Test.Hspec
@@ -63,6 +65,10 @@ agentTests (ATransport t) = do
       smpAgentTest2_2_2_needs_server $ testMsgDeliveryServerRestart t
     it "should deliver pending messages after agent restarting" $
       smpAgentTest1_1_1 $ testMsgDeliveryAgentRestart t
+    it "should concurrently deliver messages to connections without blocking" $
+      smpAgentTest2_2_1 $ testConcurrentMsgDelivery t
+    it "should deliver messages if one of connections has quota exceeded" $
+      smpAgentTest2_2_1 $ testMsgDeliveryQuotaExceeded t
 
 -- | receive message to handle `h`
 (<#:) :: Transport c => c -> IO (ATransmissionOrError 'Agent)
@@ -325,6 +331,47 @@ testMsgDeliveryAgentRestart t bob = do
     withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
     withAgent = withSmpAgentThreadOn_ (ATransport t) (agentTestPort, testPort, testDB) (pure ()) . const . testSMPAgentClientOn agentTestPort
 
+testConcurrentMsgDelivery :: Transport c => TProxy c -> c -> c -> IO ()
+testConcurrentMsgDelivery _ alice bob = do
+  connect (alice, "alice") (bob, "bob")
+
+  ("1", "bob2", Right (INV cReq)) <- alice #: ("1", "bob2", "NEW INV")
+  let cReq' = serializeConnReq cReq
+  bob #: ("11", "alice2", "JOIN " <> cReq' <> " 14\nbob's connInfo") #> ("11", "alice2", OK)
+  ("", "bob2", Right (CONF _confId "bob's connInfo")) <- (alice <#:)
+  -- below commands would be needed to accept bob's connection, but alice does not
+  -- alice #: ("2", "bob", "LET " <> _confId <> " 16\nalice's connInfo") #> ("2", "bob", OK)
+  -- bob <# ("", "alice", INFO "alice's connInfo")
+  -- bob <# ("", "alice", CON)
+  -- alice <# ("", "bob", CON)
+
+  -- the first connection should not be blocked by the second one
+  sendMessage (alice, "alice") (bob, "bob") "hello"
+  -- alice #: ("2", "bob", "SEND :hello") #> ("2", "bob", MID 1)
+  -- alice <# ("", "bob", SENT 1)
+  -- bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False
+  -- bob #: ("12", "alice", "ACK 1") #> ("12", "alice", OK)
+  bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", MID 2)
+  bob <# ("", "alice", SENT 2)
+  -- if delivery is blocked it won't go further
+  alice <#= \case ("", "bob", Msg "hello too") -> True; _ -> False
+  alice #: ("3", "bob", "ACK 2") #> ("3", "bob", OK)
+
+testMsgDeliveryQuotaExceeded :: Transport c => TProxy c -> c -> c -> IO ()
+testMsgDeliveryQuotaExceeded _ alice bob = do
+  connect (alice, "alice") (bob, "bob")
+  connect (alice, "alice2") (bob, "bob2")
+  forM_ [1 .. 4 :: Int] $ \i -> do
+    let corrId = bshow i
+        msg = "message " <> bshow i
+    (_, "bob", Right (MID mId)) <- alice #: (corrId, "bob", "SEND :" <> msg)
+    alice <#= \case ("", "bob", SENT m) -> m == mId; _ -> False
+  (_, "bob", Right (MID _)) <- alice #: ("5", "bob", "SEND :over quota")
+
+  alice #: ("1", "bob2", "SEND :hello") #> ("1", "bob2", MID 1)
+  -- if delivery is blocked it won't go further
+  alice <# ("", "bob2", SENT 1)
+
 connect :: forall c. Transport c => (c, ByteString) -> (c, ByteString) -> IO ()
 connect (h1, name1) (h2, name2) = do
   ("c1", _, Right (INV cReq)) <- h1 #: ("c1", name2, "NEW INV")
@@ -336,6 +383,16 @@ connect (h1, name1) (h2, name2) = do
   h2 <# ("", name1, CON)
   h1 <# ("", name2, CON)
 
+sendMessage :: Transport c => (c, ConnId) -> (c, ConnId) -> ByteString -> IO ()
+sendMessage (h1, name1) (h2, name2) msg = do
+  ("m1", name2', Right (MID mId)) <- h1 #: ("m1", name2, "SEND :" <> msg)
+  name2' `shouldBe` name2
+  h1 <#= \case ("", n, SENT m) -> n == name2 && m == mId; _ -> False
+  ("", name1', Right (MSG MsgMeta {recipient = (msgId, _)} msg')) <- (h2 <#:)
+  name1' `shouldBe` name1
+  msg' `shouldBe` msg
+  h2 #: ("m2", name1, "ACK " <> bshow msgId) =#> \case ("m2", n, OK) -> n == name1; _ -> False
+
 -- connect' :: forall c. Transport c => c -> c -> IO (ByteString, ByteString)
 -- connect' h1 h2 = do
 --   ("c1", conn2, Right (INV cReq)) <- h1 #: ("c1", "", "NEW INV")

tests/AgentTests/SQLiteTests.hs

@@ -22,6 +22,7 @@ import Data.Word (Word32)
 import qualified Database.SQLite.Simple as DB
 import Database.SQLite.Simple.QQ (sql)
 import SMPClient (testKeyHash)
+import Simplex.Messaging.Agent.ExceptT ()
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite

tests/SMPClient.hs

@@ -60,6 +60,7 @@ cfg =
   ServerConfig
     { transports = undefined,
       tbqSize = 1,
+      serverTbqSize = 1,
       msgQueueQuota = 4,
       queueIdBytes = 12,
       msgIdBytes = 6,

tests/ServerTests.hs

@@ -11,7 +11,7 @@ module ServerTests where
 import Control.Concurrent (ThreadId, killThread)
 import Control.Concurrent.STM
 import Control.Exception (SomeException, try)
-import Control.Monad.Except (forM_, runExceptT)
+import Control.Monad.Except (forM, forM_, runExceptT)
 import Data.ByteString.Base64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
@@ -34,6 +34,7 @@ serverTests t = do
   describe "SMP queues" do
     describe "NEW and KEY commands, SEND messages" $ testCreateSecure t
     describe "NEW, OFF and DEL commands, SEND messages" $ testCreateDelete t
+    describe "Stress test" $ stressTest t
   describe "SMP messages" do
     describe "duplex communication over 2 SMP connections" $ testDuplex t
     describe "switch subscription to another SMP queue" $ testSwitchSub t
@@ -179,6 +180,22 @@ testCreateDelete (ATransport t) =
       Resp "cdab" _ err10 <- signSendRecv rh rKey ("cdab", rId, "SUB")
       (err10, ERR AUTH) #== "rejects SUB when deleted"
 
+stressTest :: ATransport -> Spec
+stressTest (ATransport t) =
+  it "should create many queues, disconnect and re-connect" $
+    smpTest3 t $ \h1 h2 h3 -> do
+      (rPub, rKey) <- C.generateKeyPair rsaKeySize
+      rIds <- forM [1 .. 50 :: Int] . const $ do
+        Resp "" "" (IDS rId _) <- signSendRecv h1 rKey ("", "", "NEW " <> C.serializePubKey rPub)
+        pure rId
+      let subscribeQueues h = forM_ rIds $ \rId -> do
+            Resp "" rId' OK <- signSendRecv h rKey ("", rId, "SUB")
+            rId' `shouldBe` rId
+      closeConnection $ connection h1
+      subscribeQueues h2
+      closeConnection $ connection h2
+      subscribeQueues h3
+
 testDuplex :: ATransport -> Spec
 testDuplex (ATransport t) =
   it "should create 2 simplex connections and exchange messages" $