simplexmq 0.5.1 → 0.5.2
raw patch · 9 files changed
+147/−95 lines, 9 filesdep ~file-embed
Dependency ranges changed: file-embed
Files
- CHANGELOG.md +4/−0
- README.md +3/−1
- simplexmq.cabal +6/−6
- src/Simplex/Messaging/Agent.hs +44/−55
- src/Simplex/Messaging/Agent/Client.hs +9/−11
- src/Simplex/Messaging/Agent/Store.hs +2/−2
- src/Simplex/Messaging/Agent/Store/SQLite.hs +14/−20
- src/Simplex/Messaging/Util.hs +8/−0
- tests/AgentTests.hs +57/−0
CHANGELOG.md view
@@ -1,3 +1,7 @@+# 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.
README.md view
@@ -6,7 +6,9 @@ 📢 **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 && chmod +x /opt/simplex/bin/smp-server && systemctl start smp-server+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.cabal view
@@ -4,10 +4,10 @@ -- -- see: https://github.com/sol/hpack ----- hash: ff75c96b1b4f19821ceda01f32e8759e89ad661380f324f1ab9eec3d5c223425+-- hash: f5e649e0c359792d9528dec6e6cf07af772a521d746e66b8cc2b86d8c604afb2 name: simplexmq-version: 0.5.1+version: 0.5.2 synopsis: SimpleXMQ message broker description: This package includes <./docs/Simplex-Messaging-Server.html server>, <./docs/Simplex-Messaging-Client.html client> and@@ -79,7 +79,7 @@ , 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.*@@ -125,7 +125,7 @@ , 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.*@@ -173,7 +173,7 @@ , 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.*@@ -232,7 +232,7 @@ , 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.*
src/Simplex/Messaging/Agent.hs view
@@ -69,7 +69,6 @@ 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 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' 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 @@ 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 @@ 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 @@ _ -> 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 @@ 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 view
@@ -60,6 +60,7 @@ 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 +74,9 @@ 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 +94,13 @@ 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 +176,7 @@ 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/Store.hs view
@@ -63,8 +63,8 @@ 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 view
@@ -461,34 +461,28 @@ ":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/Util.hs view
@@ -53,3 +53,11 @@ 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 #-}
tests/AgentTests.hs view
@@ -14,6 +14,7 @@ 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 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 @@ 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 @@ 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")@@ -335,6 +382,16 @@ h2 <# ("", name1, INFO "info1") 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