4.2.0 (#597)

* remove TODO for old handshake version (this HELLO is not sent now)

* deduplicate connections in responses and verify server in the list of subscribed queues

* log transport and LargeMsg in tPut (the results it returns are only used in the tests)

* refactor

* refactor

CHANGELOG.md

@@ -1,3 +1,26 @@
+# 4.2.0
+
+SMP agent and server:
+
+- reduce sender traffic in cases when queue quota is exceeded:
+  - server sends quota exceeded message to the recipient when sender receives ERR QUOTA.
+  - recipient sends QCONT message to the send once the queue is drained (via reply queue).
+  - sender retry delays are increased, reducing traffic, but sender instantly resumes delivery where QCONT is received.
+
+SMP server:
+
+- increase internal queue sizes.
+
+SMP agent:
+
+- deduplicate connection IDs in connect/disconnect responses.
+- unit tests for Crypto.hs.
+- fix connection switch to another queue: correctly set primary send queue.
+
+Notification server (v1.3.0):
+
+- check token status when sending verification notificaiton.
+
 # 4.1.0
 
 SMP agent and server:

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 4.1.0
+version: 4.2.0
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,

simplexmq.cabal

@@ -1,11 +1,11 @@
 cabal-version: 1.12
 
--- This file has been generated from package.yaml by hpack version 0.34.4.
+-- This file has been generated from package.yaml by hpack version 0.35.0.
 --
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        4.1.0
+version:        4.2.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -152,9 +152,9 @@ library
     , x509 ==1.7.*
     , x509-store ==1.6.*
     , x509-validation ==1.6.*
+  default-language: Haskell2010
   if flag(swift)
     cpp-options: -DswiftJSON
-  default-language: Haskell2010
 
 executable ntf-server
   main-is: Main.hs
@@ -214,9 +214,9 @@ executable ntf-server
     , x509 ==1.7.*
     , x509-store ==1.6.*
     , x509-validation ==1.6.*
+  default-language: Haskell2010
   if flag(swift)
     cpp-options: -DswiftJSON
-  default-language: Haskell2010
 
 executable smp-agent
   main-is: Main.hs
@@ -276,9 +276,9 @@ executable smp-agent
     , x509 ==1.7.*
     , x509-store ==1.6.*
     , x509-validation ==1.6.*
+  default-language: Haskell2010
   if flag(swift)
     cpp-options: -DswiftJSON
-  default-language: Haskell2010
 
 executable smp-server
   main-is: Main.hs
@@ -338,9 +338,9 @@ executable smp-server
     , x509 ==1.7.*
     , x509-store ==1.6.*
     , x509-validation ==1.6.*
+  default-language: Haskell2010
   if flag(swift)
     cpp-options: -DswiftJSON
-  default-language: Haskell2010
 
 test-suite smp-server-test
   type: exitcode-stdio-1.0
@@ -357,6 +357,7 @@ test-suite smp-server-test
       CoreTests.CryptoTests
       CoreTests.EncodingTests
       CoreTests.ProtocolErrorTests
+      CoreTests.RetryIntervalTests
       CoreTests.VersionRangeTests
       NtfClient
       NtfServerTests
@@ -424,6 +425,6 @@ test-suite smp-server-test
     , x509 ==1.7.*
     , x509-store ==1.6.*
     , x509-validation ==1.6.*
+  default-language: Haskell2010
   if flag(swift)
     cpp-options: -DswiftJSON
-  default-language: Haskell2010

src/Simplex/Messaging/Agent.hs

@@ -504,7 +504,7 @@ joinConnSrv c connId asyncMode enableNtfs (CRInvitationUri ConnReqUriData {crAge
           unless duplexHS . void $ enqueueMessage c cData' sq SMP.noMsgFlags HELLO
           pure connId'
         Left e -> do
-          -- TODO recovery for failure on network timeout, see rfcs/2022-04-20-smp-conf-timeout-recovery.md
+          -- possible improvement: recovery for failure on network timeout, see rfcs/2022-04-20-smp-conf-timeout-recovery.md
           unless asyncMode $ withStore' c (`deleteConn` connId')
           throwError e
       where
@@ -791,7 +791,7 @@ runCommandProcessing c@AgentClient {subQ} server_ = do
     atomically $ beginAgentOperation c AOSndNetwork
     E.try (withStore c $ \db -> getPendingCommand db cmdId) >>= \case
       Left (e :: E.SomeException) -> atomically $ writeTBQueue subQ ("", "", ERR . INTERNAL $ show e)
-      Right (corrId, connId, cmd) -> processCmd ri corrId connId cmdId cmd
+      Right (corrId, connId, cmd) -> processCmd (riFast ri) corrId connId cmdId cmd
   where
     processCmd :: RetryInterval -> ACorrId -> ConnId -> AsyncCmdId -> AgentCommand -> m ()
     processCmd ri corrId connId cmdId command = case command of
@@ -964,22 +964,22 @@ queuePendingMsgs c sq msgIds = atomically $ do
   modifyTVar' (msgDeliveryOp c) $ \s -> s {opsInProgress = opsInProgress s + length msgIds}
   -- s <- readTVar (msgDeliveryOp c)
   -- unsafeIOToSTM $ putStrLn $ "msgDeliveryOp: " <> show (opsInProgress s)
-  q <- getPendingMsgQ c sq
-  mapM_ (writeTQueue q) msgIds
+  (mq, _) <- getPendingMsgQ c sq
+  mapM_ (writeTQueue mq) msgIds
 
-getPendingMsgQ :: AgentClient -> SndQueue -> STM (TQueue InternalId)
+getPendingMsgQ :: AgentClient -> SndQueue -> STM (TQueue InternalId, TMVar ())
 getPendingMsgQ c SndQueue {server, sndId} = do
   let qKey = (server, sndId)
   maybe (newMsgQueue qKey) pure =<< TM.lookup qKey (smpQueueMsgQueues c)
   where
     newMsgQueue qKey = do
-      mq <- newTQueue
-      TM.insert qKey mq $ smpQueueMsgQueues c
-      pure mq
+      q <- (,) <$> newTQueue <*> newEmptyTMVar
+      TM.insert qKey q $ smpQueueMsgQueues c
+      pure q
 
 runSmpQueueMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> m ()
 runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandshake} sq = do
-  mq <- atomically $ getPendingMsgQ c sq
+  (mq, qLock) <- atomically $ getPendingMsgQ c sq
   ri <- asks $ messageRetryInterval . config
   forever $ do
     atomically $ endAgentOperation c AOSndNetwork
@@ -993,7 +993,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
       Left (e :: E.SomeException) ->
         notify $ MERR mId (INTERNAL $ show e)
       Right (rq_, PendingMsgData {msgType, msgBody, msgFlags, internalTs}) ->
-        withRetryInterval ri $ \loop -> do
+        withRetryLock2 ri qLock $ \loop -> do
           resp <- tryError $ case msgType of
             AM_CONN_INFO -> sendConfirmation c sq msgBody
             _ -> sendAgentMessage c sq msgFlags msgBody
@@ -1004,7 +1004,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                 SMP SMP.QUOTA -> case msgType of
                   AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                   AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
-                  _ -> retrySndOp c loop
+                  _ -> retrySndOp c $ loop RISlow
                 SMP SMP.AUTH -> case msgType of
                   AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                   AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
@@ -1013,7 +1013,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                     -- because the queue must be secured by the time the confirmation or the first HELLO is received
                     | duplexHandshake == Just True -> connErr
                     | otherwise ->
-                      ifM (msgExpired helloTimeout) connErr (retrySndOp c loop)
+                      ifM (msgExpired helloTimeout) connErr (retrySndOp c $ loop RIFast)
                     where
                       connErr = case rq_ of
                         -- party initiating connection
@@ -1022,6 +1022,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                         _ -> connError msgId NOT_ACCEPTED
                   AM_REPLY_ -> notifyDel msgId err
                   AM_A_MSG_ -> notifyDel msgId err
+                  AM_QCONT_ -> notifyDel msgId err
                   AM_QADD_ -> qError msgId "QADD: AUTH"
                   AM_QKEY_ -> qError msgId "QKEY: AUTH"
                   AM_QUSE_ -> qError msgId "QUSE: AUTH"
@@ -1031,7 +1032,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                   -- the message sending would be retried
                   | temporaryOrHostError e -> do
                     let timeoutSel = if msgType == AM_HELLO_ then helloTimeout else messageTimeout
-                    ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySndOp c loop)
+                    ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySndOp c $ loop RIFast)
                   | otherwise -> notifyDel msgId err
               where
                 msgExpired timeoutSel = do
@@ -1044,7 +1045,6 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                   withStore' c $ \db -> do
                     setSndQueueStatus db sq Confirmed
                     when (isJust rq_) $ removeConfirmations db connId
-                  -- TODO possibly notification flag should be ON for one of the parties, to result in contact connected notification
                   unless (duplexHandshake == Just True) . void $ enqueueMessage c cData sq SMP.noMsgFlags HELLO
                 AM_CONN_INFO_REPLY -> pure ()
                 AM_REPLY_ -> pure ()
@@ -1071,6 +1071,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                       qInfo <- createReplyQueue c cData sq srv
                       void . enqueueMessage c cData sq SMP.noMsgFlags $ REPLY [qInfo]
                 AM_A_MSG_ -> notify $ SENT mId
+                AM_QCONT_ -> pure ()
                 AM_QADD_ -> pure ()
                 AM_QKEY_ -> pure ()
                 AM_QUSE_ -> pure ()
@@ -1080,16 +1081,18 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {connId, duplexHandsh
                   case conn of
                     DuplexConnection cData' rqs sqs -> do
                       -- remove old snd queue from connection once QTEST is sent to the new queue
-                      case findQ (qAddress sq) sqs of
+                      let addr = qAddress sq
+                      case findQ addr sqs of
                         -- this is the same queue where this loop delivers messages to but with updated state
                         Just SndQueue {dbReplaceQueueId = Just replacedId, primary} ->
-                          case removeQP (\SndQueue {dbQueueId} -> dbQueueId == replacedId) sqs of
+                          -- second part of this condition is a sanity check because dbReplaceQueueId cannot point to the same queue, see switchConnection'
+                          case removeQP (\sq'@SndQueue {dbQueueId} -> dbQueueId == replacedId && not (sameQueue addr sq')) sqs of
                             Nothing -> internalErr msgId "sent QTEST: queue not found in connection"
                             Just (sq', sq'' : sqs') -> do
                               -- remove the delivery from the map to stop the thread when the delivery loop is complete
                               atomically $ TM.delete (qAddress sq') $ smpQueueMsgQueues c
                               withStore' c $ \db -> do
-                                when primary $ setSndQueuePrimary db connId sq'
+                                when primary $ setSndQueuePrimary db connId sq
                                 deletePendingMsgs db connId sq'
                                 deleteConnSndQueue db connId sq'
                               let sqs'' = sq'' :| sqs'
@@ -1225,7 +1228,7 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
         (Nothing, Just NTARegister) -> do
           when (savedDeviceToken /= suppliedDeviceToken) $ withStore' c $ \db -> updateDeviceToken db tkn suppliedDeviceToken
           registerToken tkn $> NTRegistered
-        -- TODO minimal time before repeat registration
+        -- possible improvement: add minimal time before repeat registration
         (Just tknId, Nothing)
           | savedDeviceToken == suppliedDeviceToken ->
             when (ntfTknStatus == NTRegistered) (registerToken tkn) $> NTRegistered
@@ -1243,8 +1246,8 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
               agentNtfEnableCron c tknId tkn cron
               when (suppliedNtfMode == NMInstant) $ initializeNtfSubs c
               when (suppliedNtfMode == NMPeriodic && savedNtfMode == NMInstant) $ deleteNtfSubs c NSCDelete
-            pure ntfTknStatus -- TODO
-            -- agentNtfCheckToken c tknId tkn >>= \case
+            -- possible improvement: get updated token status from the server, or maybe TCRON could return the current status
+            pure ntfTknStatus
           | otherwise -> replaceToken tknId
         (Just tknId, Just NTADelete) -> do
           agentNtfDeleteToken c tknId tkn
@@ -1411,11 +1414,6 @@ sendNtfConnCommands c cmd = do
       _ ->
         atomically $ writeTBQueue (subQ c) ("", connId, ERR $ INTERNAL "no connection data")
 
--- TODO
--- There should probably be another function to cancel all subscriptions that would flush the queue first,
--- so that supervisor stops processing pending commands?
--- It is an optimization, but I am thinking how it would behave if a user were to flip on/off quickly several times.
-
 setNtfServers' :: AgentMonad m => AgentClient -> [NtfServer] -> m ()
 setNtfServers' c = atomically . writeTVar (ntfServers c)
 
@@ -1492,88 +1490,96 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
     processSMP :: RcvQueue -> Connection c -> ConnData -> m ()
     processSMP rq@RcvQueue {e2ePrivKey, e2eDhSecret, status} conn cData@ConnData {connId, duplexHandshake} = withConnLock c connId "processSMP" $
       case cmd of
-        SMP.MSG msg@SMP.RcvMessage {msgId = srvMsgId} -> handleNotifyAck $ do
-          SMP.ClientRcvMsgBody {msgTs = srvTs, msgFlags, msgBody} <- decryptSMPMessage v rq msg
-          clientMsg@SMP.ClientMsgEnvelope {cmHeader = SMP.PubHeader phVer e2ePubKey_} <-
-            parseMessage msgBody
-          clientVRange <- asks $ smpClientVRange . config
-          unless (phVer `isCompatible` clientVRange) . throwError $ AGENT A_VERSION
-          case (e2eDhSecret, e2ePubKey_) of
-            (Nothing, Just e2ePubKey) -> do
-              let e2eDh = C.dh' e2ePubKey e2ePrivKey
-              decryptClientMessage e2eDh clientMsg >>= \case
-                (SMP.PHConfirmation senderKey, AgentConfirmation {e2eEncryption, encConnInfo, agentVersion}) ->
-                  smpConfirmation senderKey e2ePubKey e2eEncryption encConnInfo phVer agentVersion >> ack
-                (SMP.PHEmpty, AgentInvitation {connReq, connInfo}) ->
-                  smpInvitation connReq connInfo >> ack
-                _ -> prohibited >> ack
-            (Just e2eDh, Nothing) -> do
-              decryptClientMessage e2eDh clientMsg >>= \case
-                (SMP.PHEmpty, AgentMsgEnvelope _ encAgentMsg) -> do
-                  -- primary queue is set as Active in helloMsg, below is to set additional queues Active
-                  let RcvQueue {primary, dbReplaceQueueId} = rq
-                  unless (status == Active) . withStore' c $ \db -> setRcvQueueStatus db rq Active
-                  case (conn, dbReplaceQueueId) of
-                    (DuplexConnection _ rqs _, Just replacedId) -> do
-                      when primary . withStore' c $ \db -> setRcvQueuePrimary db connId rq
-                      case find (\RcvQueue {dbQueueId} -> dbQueueId == replacedId) rqs of
-                        Just RcvQueue {server, rcvId} -> do
-                          enqueueCommand c "" connId (Just server) $ AInternalCommand $ ICQDelete rcvId
-                        _ -> notify . ERR . AGENT $ A_QUEUE "replaced RcvQueue not found in connection"
-                    _ -> pure ()
-                  tryError agentClientMsg >>= \case
-                    Right (Just (msgId, msgMeta, aMessage)) -> case aMessage of
-                      HELLO -> helloMsg >> ackDel msgId
-                      REPLY cReq -> replyMsg cReq >> ackDel msgId
-                      -- note that there is no ACK sent for A_MSG, it is sent with agent's user ACK command
-                      A_MSG body -> do
-                        logServer "<--" c srv rId "MSG <MSG>"
-                        notify $ MSG msgMeta msgFlags body
-                      QADD qs -> qDuplex "QADD" $ qAddMsg qs
-                      QKEY qs -> qDuplex "QKEY" $ qKeyMsg qs
-                      QUSE qs -> qDuplex "QUSE" $ qUseMsg qs
-                      -- no action needed for QTEST
-                      -- any message in the new queue will mark it active and trigger deletion of the old queue
-                      QTEST _ -> logServer "<--" c srv rId "MSG <QTEST>" >> ackDel msgId
-                      where
-                        qDuplex :: String -> (Connection 'CDuplex -> m ()) -> m ()
-                        qDuplex name a = case conn of
-                          DuplexConnection {} -> a conn >> ackDel msgId
-                          _ -> qError $ name <> ": message must be sent to duplex connection"
-                    Right _ -> prohibited >> ack
-                    Left e@(AGENT A_DUPLICATE) -> do
-                      withStore' c (\db -> getLastMsg db connId srvMsgId) >>= \case
-                        Just RcvMsg {internalId, msgMeta, msgBody = agentMsgBody, userAck}
-                          | userAck -> ackDel internalId
-                          | otherwise -> do
-                            liftEither (parse smpP (AGENT A_MESSAGE) agentMsgBody) >>= \case
-                              AgentMessage _ (A_MSG body) -> do
-                                logServer "<--" c srv rId "MSG <MSG>"
-                                notify $ MSG msgMeta msgFlags body
-                              _ -> pure ()
-                        _ -> throwError e
-                    Left e -> throwError e
-                  where
-                    agentClientMsg :: m (Maybe (InternalId, MsgMeta, AMessage))
-                    agentClientMsg = withStore c $ \db -> runExceptT $ do
-                      agentMsgBody <- agentRatchetDecrypt db connId encAgentMsg
-                      liftEither (parse smpP (SEAgentError $ AGENT A_MESSAGE) agentMsgBody) >>= \case
-                        agentMsg@(AgentMessage APrivHeader {sndMsgId, prevMsgHash} aMessage) -> do
-                          let msgType = agentMessageType agentMsg
-                              internalHash = C.sha256Hash agentMsgBody
-                          internalTs <- liftIO getCurrentTime
-                          (internalId, internalRcvId, prevExtSndId, prevRcvMsgHash) <- liftIO $ updateRcvIds db connId
-                          let integrity = checkMsgIntegrity prevExtSndId sndMsgId prevRcvMsgHash prevMsgHash
-                              recipient = (unId internalId, internalTs)
-                              broker = (srvMsgId, systemToUTCTime srvTs)
-                              msgMeta = MsgMeta {integrity, recipient, broker, sndMsgId}
-                              rcvMsg = RcvMsgData {msgMeta, msgType, msgFlags, msgBody = agentMsgBody, internalRcvId, internalHash, externalPrevSndHash = prevMsgHash}
-                          liftIO $ createRcvMsg db connId rq rcvMsg
-                          pure $ Just (internalId, msgMeta, aMessage)
-                        _ -> pure Nothing
-                _ -> prohibited >> ack
-            _ -> prohibited >> ack
+        SMP.MSG msg@SMP.RcvMessage {msgId = srvMsgId} ->
+          handleNotifyAck $
+            decryptSMPMessage v rq msg >>= \case
+              SMP.ClientRcvMsgBody {msgTs = srvTs, msgFlags, msgBody} -> processClientMsg srvTs msgFlags msgBody
+              SMP.ClientRcvMsgQuota {} -> queueDrained >> ack
           where
+            queueDrained = case conn of
+              DuplexConnection _ _ sqs -> void $ enqueueMessages c cData sqs SMP.noMsgFlags $ QCONT (sndAddress rq)
+              _ -> pure ()
+            processClientMsg srvTs msgFlags msgBody = do
+              clientMsg@SMP.ClientMsgEnvelope {cmHeader = SMP.PubHeader phVer e2ePubKey_} <-
+                parseMessage msgBody
+              clientVRange <- asks $ smpClientVRange . config
+              unless (phVer `isCompatible` clientVRange) . throwError $ AGENT A_VERSION
+              case (e2eDhSecret, e2ePubKey_) of
+                (Nothing, Just e2ePubKey) -> do
+                  let e2eDh = C.dh' e2ePubKey e2ePrivKey
+                  decryptClientMessage e2eDh clientMsg >>= \case
+                    (SMP.PHConfirmation senderKey, AgentConfirmation {e2eEncryption, encConnInfo, agentVersion}) ->
+                      smpConfirmation senderKey e2ePubKey e2eEncryption encConnInfo phVer agentVersion >> ack
+                    (SMP.PHEmpty, AgentInvitation {connReq, connInfo}) ->
+                      smpInvitation connReq connInfo >> ack
+                    _ -> prohibited >> ack
+                (Just e2eDh, Nothing) -> do
+                  decryptClientMessage e2eDh clientMsg >>= \case
+                    (SMP.PHEmpty, AgentMsgEnvelope _ encAgentMsg) -> do
+                      -- primary queue is set as Active in helloMsg, below is to set additional queues Active
+                      let RcvQueue {primary, dbReplaceQueueId} = rq
+                      unless (status == Active) . withStore' c $ \db -> setRcvQueueStatus db rq Active
+                      case (conn, dbReplaceQueueId) of
+                        (DuplexConnection _ rqs _, Just replacedId) -> do
+                          when primary . withStore' c $ \db -> setRcvQueuePrimary db connId rq
+                          case find (\RcvQueue {dbQueueId} -> dbQueueId == replacedId) rqs of
+                            Just RcvQueue {server, rcvId} -> do
+                              enqueueCommand c "" connId (Just server) $ AInternalCommand $ ICQDelete rcvId
+                            _ -> notify . ERR . AGENT $ A_QUEUE "replaced RcvQueue not found in connection"
+                        _ -> pure ()
+                      tryError agentClientMsg >>= \case
+                        Right (Just (msgId, msgMeta, aMessage)) -> case aMessage of
+                          HELLO -> helloMsg >> ackDel msgId
+                          REPLY cReq -> replyMsg cReq >> ackDel msgId
+                          -- note that there is no ACK sent for A_MSG, it is sent with agent's user ACK command
+                          A_MSG body -> do
+                            logServer "<--" c srv rId "MSG <MSG>"
+                            notify $ MSG msgMeta msgFlags body
+                          QCONT addr -> qDuplex "QCONT" $ continueSending addr
+                          QADD qs -> qDuplex "QADD" $ qAddMsg qs
+                          QKEY qs -> qDuplex "QKEY" $ qKeyMsg qs
+                          QUSE qs -> qDuplex "QUSE" $ qUseMsg qs
+                          -- no action needed for QTEST
+                          -- any message in the new queue will mark it active and trigger deletion of the old queue
+                          QTEST _ -> logServer "<--" c srv rId "MSG <QTEST>" >> ackDel msgId
+                          where
+                            qDuplex :: String -> (Connection 'CDuplex -> m ()) -> m ()
+                            qDuplex name a = case conn of
+                              DuplexConnection {} -> a conn >> ackDel msgId
+                              _ -> qError $ name <> ": message must be sent to duplex connection"
+                        Right _ -> prohibited >> ack
+                        Left e@(AGENT A_DUPLICATE) -> do
+                          withStore' c (\db -> getLastMsg db connId srvMsgId) >>= \case
+                            Just RcvMsg {internalId, msgMeta, msgBody = agentMsgBody, userAck}
+                              | userAck -> ackDel internalId
+                              | otherwise -> do
+                                liftEither (parse smpP (AGENT A_MESSAGE) agentMsgBody) >>= \case
+                                  AgentMessage _ (A_MSG body) -> do
+                                    logServer "<--" c srv rId "MSG <MSG>"
+                                    notify $ MSG msgMeta msgFlags body
+                                  _ -> pure ()
+                            _ -> throwError e
+                        Left e -> throwError e
+                      where
+                        agentClientMsg :: m (Maybe (InternalId, MsgMeta, AMessage))
+                        agentClientMsg = withStore c $ \db -> runExceptT $ do
+                          agentMsgBody <- agentRatchetDecrypt db connId encAgentMsg
+                          liftEither (parse smpP (SEAgentError $ AGENT A_MESSAGE) agentMsgBody) >>= \case
+                            agentMsg@(AgentMessage APrivHeader {sndMsgId, prevMsgHash} aMessage) -> do
+                              let msgType = agentMessageType agentMsg
+                                  internalHash = C.sha256Hash agentMsgBody
+                              internalTs <- liftIO getCurrentTime
+                              (internalId, internalRcvId, prevExtSndId, prevRcvMsgHash) <- liftIO $ updateRcvIds db connId
+                              let integrity = checkMsgIntegrity prevExtSndId sndMsgId prevRcvMsgHash prevMsgHash
+                                  recipient = (unId internalId, internalTs)
+                                  broker = (srvMsgId, systemToUTCTime srvTs)
+                                  msgMeta = MsgMeta {integrity, recipient, broker, sndMsgId}
+                                  rcvMsg = RcvMsgData {msgMeta, msgType, msgFlags, msgBody = agentMsgBody, internalRcvId, internalHash, externalPrevSndHash = prevMsgHash}
+                              liftIO $ createRcvMsg db connId rq rcvMsg
+                              pure $ Just (internalId, msgMeta, aMessage)
+                            _ -> pure Nothing
+                    _ -> prohibited >> ack
+                _ -> prohibited >> ack
             ack :: m ()
             ack = enqueueCmd $ ICAck rId srvMsgId
             ackDel :: InternalId -> m ()
@@ -1698,6 +1704,16 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (srv, v, sessId, rId, cm
                 connectReplyQueues c cData ownConnInfo smpQueues `catchError` (notify . ERR)
               _ -> prohibited
 
+        continueSending :: (SMPServer, SMP.SenderId) -> Connection 'CDuplex -> m ()
+        continueSending addr (DuplexConnection _ _ sqs) =
+          case findQ addr sqs of
+            Just sq -> do
+              logServer "<--" c srv rId "MSG <QCONT>"
+              atomically $ do
+                (_, qLock) <- getPendingMsgQ c sq
+                void $ tryPutTMVar qLock ()
+            Nothing -> qError "QCONT: queue address not found"
+
         -- processed by queue sender
         qAddMsg :: NonEmpty (SMPQueueUri, Maybe SndQAddr) -> Connection 'CDuplex -> m ()
         qAddMsg ((_, Nothing) :| _) _ = qError "adding queue without switching is not supported"

src/Simplex/Messaging/Agent/Client.hs

@@ -98,7 +98,7 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (isRight, partitionEithers)
 import Data.Functor (($>))
-import Data.List (partition, (\\))
+import Data.List (partition)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
@@ -180,7 +180,7 @@ data AgentClient = AgentClient
     activeSubs :: TRcvQueues,
     pendingSubs :: TRcvQueues,
     pendingMsgsQueued :: TMap SndQAddr Bool,
-    smpQueueMsgQueues :: TMap SndQAddr (TQueue InternalId),
+    smpQueueMsgQueues :: TMap SndQAddr (TQueue InternalId, TMVar ()),
     smpQueueMsgDeliveries :: TMap SndQAddr (Async ()),
     connCmdsQueued :: TMap ConnId Bool,
     asyncCmdQueues :: TMap (Maybe SMPServer) (TQueue AsyncCmdId),
@@ -303,12 +303,9 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} srv = do
         removeClientAndSubs :: IO ([RcvQueue], [ConnId])
         removeClientAndSubs = atomically $ do
           TM.delete srv smpClients
-          qs <- RQ.getDelSrvQueues srv $ activeSubs c
+          (qs, conns) <- RQ.getDelSrvQueues srv $ activeSubs c
           mapM_ (`RQ.addQueue` pendingSubs c) qs
-          cs <- RQ.getConns (activeSubs c)
-          -- TODO deduplicate conns
-          let conns = map (connId :: RcvQueue -> ConnId) qs \\ S.toList cs
-          pure (qs, conns)
+          pure (qs, S.toList conns)
 
         serverDown :: ([RcvQueue], [ConnId]) -> IO ()
         serverDown (qs, conns) = whenM (readTVarIO active) $ do
@@ -345,8 +342,7 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} srv = do
             unless connected . forM_ client_ $ \cl -> do
               incClientStat c cl "CONNECT" ""
               notifySub "" $ hostEvent CONNECT cl
-            -- TODO deduplicate okConns
-            let conns = okConns \\ S.toList cs
+            let conns = S.toList $ S.fromList okConns `S.difference` cs
             unless (null conns) $ notifySub "" $ UP srv conns
           let (tempErrs, finalErrs) = partition (temporaryAgentError . snd) errs
           liftIO $ mapM_ (\(connId, e) -> notifySub connId $ ERR e) finalErrs
@@ -647,8 +643,7 @@ temporaryOrHostError = \case
 subscribeQueues :: AgentMonad m => AgentClient -> SMPServer -> [RcvQueue] -> m (Maybe SMPClient, [(RcvQueue, Either AgentErrorType ())])
 subscribeQueues c srv qs = do
   (errs, qs_) <- partitionEithers <$> mapM checkQueue qs
-  forM_ qs_ $ \rq@RcvQueue {connId, server = _server} -> atomically $ do
-    -- TODO check server is correct
+  forM_ qs_ $ \rq@RcvQueue {connId} -> atomically $ do
     modifyTVar (subscrConns c) $ S.insert connId
     RQ.addQueue rq $ pendingSubs c
   case L.nonEmpty qs_ of
@@ -667,9 +662,11 @@ subscribeQueues c srv qs = do
             pure $ map (second . first $ protocolClientError SMP $ clientServer smp) rs
     _ -> pure (Nothing, errs)
   where
-    checkQueue rq@RcvQueue {rcvId, server} = do
-      prohibited <- atomically . TM.member (server, rcvId) $ getMsgLocks c
-      pure $ if prohibited || srv /= server then Left (rq, Left $ CMD PROHIBITED) else Right rq
+    checkQueue rq@RcvQueue {rcvId, server}
+      | server == srv = do
+        prohibited <- atomically . TM.member (server, rcvId) $ getMsgLocks c
+        pure $ if prohibited || srv /= server then Left (rq, Left $ CMD PROHIBITED) else Right rq
+      | otherwise = pure $ Left (rq, Left $ INTERNAL "queue server does not match parameter")
     queueCreds RcvQueue {rcvPrivateKey, rcvId} = (rcvPrivateKey, rcvId)
 
 addSubscription :: MonadIO m => AgentClient -> RcvQueue -> m ()

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

@@ -83,7 +83,7 @@ data AgentConfig = AgentConfig
     smpCfg :: ProtocolClientConfig,
     ntfCfg :: ProtocolClientConfig,
     reconnectInterval :: RetryInterval,
-    messageRetryInterval :: RetryInterval,
+    messageRetryInterval :: RetryInterval2,
     messageTimeout :: NominalDiffTime,
     helloTimeout :: NominalDiffTime,
     ntfCron :: Word16,
@@ -108,12 +108,24 @@ defaultReconnectInterval =
       maxInterval = 180_000000
     }
 
-defaultMessageRetryInterval :: RetryInterval
+defaultMessageRetryInterval :: RetryInterval2
 defaultMessageRetryInterval =
-  RetryInterval
-    { initialInterval = 1_000000,
-      increaseAfter = 10_000000,
-      maxInterval = 60_000000
+  RetryInterval2
+    { riFast =
+        RetryInterval
+          { initialInterval = 1_000000,
+            increaseAfter = 10_000000,
+            maxInterval = 60_000000
+          },
+      riSlow =
+        -- TODO: these timeouts can be increased in v4.6 once most clients are updated
+        -- to resume sending on QCONT messages.
+        -- After that local message expiration period should be also increased.
+        RetryInterval
+          { initialInterval = 10_000000,
+            increaseAfter = 30_000000,
+            maxInterval = 300_000000
+          }
     }
 
 defaultAgentConfig :: AgentConfig

src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

@@ -191,7 +191,7 @@ runNtfWorker c srv doWork = do
                 case clientNtfCreds of
                   Just ClientNtfCreds {ntfPrivateKey, notifierId} -> do
                     nSubId <- agentNtfCreateSubscription c tknId tkn (SMPQueueNtf smpServer notifierId) ntfPrivateKey
-                    -- TODO smaller retry until Active, less frequently (daily?) once Active
+                    -- possible improvement: smaller retry until Active, less frequently (daily?) once Active
                     let actionTs' = addUTCTime 30 ts
                     withStore' c $ \db ->
                       updateNtfSubscription db sub {ntfSubId = Just nSubId, ntfSubStatus = NASCreated NSNew} (NtfSubNTFAction NSACheck) actionTs'

src/Simplex/Messaging/Agent/Protocol.hs

@@ -577,6 +577,7 @@ data AgentMessageType
   | AM_HELLO_
   | AM_REPLY_
   | AM_A_MSG_
+  | AM_QCONT_
   | AM_QADD_
   | AM_QKEY_
   | AM_QUSE_
@@ -590,6 +591,7 @@ instance Encoding AgentMessageType where
     AM_HELLO_ -> "H"
     AM_REPLY_ -> "R"
     AM_A_MSG_ -> "M"
+    AM_QCONT_ -> "QC"
     AM_QADD_ -> "QA"
     AM_QKEY_ -> "QK"
     AM_QUSE_ -> "QU"
@@ -603,6 +605,7 @@ instance Encoding AgentMessageType where
       'M' -> pure AM_A_MSG_
       'Q' ->
         A.anyChar >>= \case
+          'C' -> pure AM_QCONT_
           'A' -> pure AM_QADD_
           'K' -> pure AM_QKEY_
           'U' -> pure AM_QUSE_
@@ -623,6 +626,7 @@ agentMessageType = \case
     -- REPLY is only used in v1
     REPLY _ -> AM_REPLY_
     A_MSG _ -> AM_A_MSG_
+    QCONT _ -> AM_QCONT_
     QADD _ -> AM_QADD_
     QKEY _ -> AM_QKEY_
     QUSE _ -> AM_QUSE_
@@ -645,6 +649,7 @@ data AMsgType
   = HELLO_
   | REPLY_
   | A_MSG_
+  | QCONT_
   | QADD_
   | QKEY_
   | QUSE_
@@ -656,6 +661,7 @@ instance Encoding AMsgType where
     HELLO_ -> "H"
     REPLY_ -> "R"
     A_MSG_ -> "M"
+    QCONT_ -> "QC"
     QADD_ -> "QA"
     QKEY_ -> "QK"
     QUSE_ -> "QU"
@@ -667,6 +673,7 @@ instance Encoding AMsgType where
       'M' -> pure A_MSG_
       'Q' ->
         A.anyChar >>= \case
+          'C' -> pure QCONT_
           'A' -> pure QADD_
           'K' -> pure QKEY_
           'U' -> pure QUSE_
@@ -684,6 +691,8 @@ data AMessage
     REPLY (L.NonEmpty SMPQueueInfo)
   | -- | agent envelope for the client message
     A_MSG MsgBody
+  | -- | the message instructing the client to continue sending messages (after ERR QUOTA)
+    QCONT SndQAddr
   | -- add queue to connection (sent by recipient), with optional address of the replaced queue
     QADD (L.NonEmpty (SMPQueueUri, Maybe SndQAddr))
   | -- key to secure the added queues and agree e2e encryption key (sent by sender)
@@ -701,6 +710,7 @@ instance Encoding AMessage where
     HELLO -> smpEncode HELLO_
     REPLY smpQueues -> smpEncode (REPLY_, smpQueues)
     A_MSG body -> smpEncode (A_MSG_, Tail body)
+    QCONT addr -> smpEncode (QCONT_, addr)
     QADD qs -> smpEncode (QADD_, qs)
     QKEY qs -> smpEncode (QKEY_, qs)
     QUSE qs -> smpEncode (QUSE_, qs)
@@ -711,6 +721,7 @@ instance Encoding AMessage where
         HELLO_ -> pure HELLO
         REPLY_ -> REPLY <$> smpP
         A_MSG_ -> A_MSG . unTail <$> smpP
+        QCONT_ -> QCONT <$> smpP
         QADD_ -> QADD <$> smpP
         QKEY_ -> QKEY <$> smpP
         QUSE_ -> QUSE <$> smpP
@@ -1119,7 +1130,6 @@ instance ToJSON BrokerErrorType where
   toEncoding = J.genericToEncoding $ sumTypeJSON id
 
 -- | Errors of another SMP agent.
--- TODO encode/decode without A prefix
 data SMPAgentError
   = -- | client or agent message that failed to parse
     A_MESSAGE

src/Simplex/Messaging/Agent/RetryInterval.hs

@@ -1,10 +1,21 @@
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 
-module Simplex.Messaging.Agent.RetryInterval where
+module Simplex.Messaging.Agent.RetryInterval
+  ( RetryInterval (..),
+    RetryInterval2 (..),
+    RetryIntervalMode (..),
+    withRetryInterval,
+    withRetryLock2,
+  )
+where
 
-import Control.Concurrent (threadDelay)
+import Control.Concurrent (forkIO, threadDelay)
+import Control.Monad (void)
 import Control.Monad.IO.Class (MonadIO, liftIO)
+import Simplex.Messaging.Util (whenM)
+import UnliftIO.STM
 
 data RetryInterval = RetryInterval
   { initialInterval :: Int,
@@ -12,17 +23,51 @@ data RetryInterval = RetryInterval
     maxInterval :: Int
   }
 
+data RetryInterval2 = RetryInterval2
+  { riSlow :: RetryInterval,
+    riFast :: RetryInterval
+  }
+
+data RetryIntervalMode = RISlow | RIFast
+  deriving (Eq)
+
 withRetryInterval :: forall m. MonadIO m => RetryInterval -> (m () -> m ()) -> m ()
-withRetryInterval RetryInterval {initialInterval, increaseAfter, maxInterval} action =
-  callAction 0 initialInterval
+withRetryInterval ri action = callAction 0 $ initialInterval ri
   where
     callAction :: Int -> Int -> m ()
-    callAction elapsedTime delay = action loop
+    callAction elapsed delay = action loop
       where
         loop = do
-          let newDelay =
-                if elapsedTime < increaseAfter || delay == maxInterval
-                  then delay
-                  else min (delay * 3 `div` 2) maxInterval
           liftIO $ threadDelay delay
-          callAction (elapsedTime + delay) newDelay
+          let elapsed' = elapsed + delay
+          callAction elapsed' $ nextDelay elapsed' delay ri
+
+-- This function allows action to toggle between slow and fast retry intervals.
+withRetryLock2 :: forall m. MonadIO m => RetryInterval2 -> TMVar () -> ((RetryIntervalMode -> m ()) -> m ()) -> m ()
+withRetryLock2 RetryInterval2 {riSlow, riFast} lock action =
+  callAction (0, initialInterval riSlow) (0, initialInterval riFast)
+  where
+    callAction :: (Int, Int) -> (Int, Int) -> m ()
+    callAction slow fast = action loop
+      where
+        loop = \case
+          RISlow -> run slow riSlow (`callAction` fast)
+          RIFast -> run fast riFast (callAction slow)
+        run (elapsed, delay) ri call = do
+          wait delay
+          let elapsed' = elapsed + delay
+          call (elapsed', nextDelay elapsed' delay ri)
+        wait delay = do
+          waiting <- newTVarIO True
+          _ <- liftIO . forkIO $ do
+            threadDelay delay
+            atomically $ whenM (readTVar waiting) $ void $ tryPutTMVar lock ()
+          atomically $ do
+            takeTMVar lock
+            writeTVar waiting False
+
+nextDelay :: Int -> Int -> RetryInterval -> Int
+nextDelay elapsed delay RetryInterval {increaseAfter, maxInterval} =
+  if elapsed < increaseAfter || delay == maxInterval
+    then delay
+    else min (delay * 3 `div` 2) maxInterval

src/Simplex/Messaging/Agent/TRcvQueues.hs

@@ -40,9 +40,9 @@ getSrvQueues srv (TRcvQueues qs) = M.foldl' addQ [] <$> readTVar qs
   where
     addQ qs' rq@RcvQueue {server} = if srv == server then rq : qs' else qs'
 
-getDelSrvQueues :: SMPServer -> TRcvQueues -> STM [RcvQueue]
-getDelSrvQueues srv (TRcvQueues qs) = stateTVar qs $ M.foldl' addQ ([], M.empty)
+getDelSrvQueues :: SMPServer -> TRcvQueues -> STM ([RcvQueue], Set ConnId)
+getDelSrvQueues srv (TRcvQueues qs) = stateTVar qs $ M.foldl' addQ (([], S.empty), M.empty)
   where
-    addQ (removed, qs') rq@RcvQueue {server, rcvId}
-      | srv == server = (rq : removed, qs')
+    addQ (removed@(remQs, remConns), qs') rq@RcvQueue {connId, server, rcvId}
+      | srv == server = ((rq : remQs, S.insert connId remConns), qs')
       | otherwise = (removed, M.insert (server, rcvId) rq qs')

src/Simplex/Messaging/Client.hs

@@ -1,7 +1,6 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE DeriveGeneric #-}
-{-# LANGUAGE DisambiguateRecordFields #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE GADTs #-}
@@ -94,7 +93,7 @@ import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client (SocksProxy, TransportClientConfig (..), TransportHost (..), runTransportClient)
 import Simplex.Messaging.Transport.KeepAlive
 import Simplex.Messaging.Transport.WebSockets (WS)
-import Simplex.Messaging.Util (bshow, liftError, raceAny_)
+import Simplex.Messaging.Util (bshow, raceAny_)
 import Simplex.Messaging.Version
 import System.Timeout (timeout)
 
@@ -541,7 +540,7 @@ sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ, tcpTimeout}} pKey
 mkTransmission :: forall msg. ProtocolEncoding (ProtoCommand msg) => ProtocolClient msg -> ClientCommand msg -> ExceptT ProtocolClientError IO (SentRawTransmission, TMVar (Response msg))
 mkTransmission ProtocolClient {sessionId, thVersion, client_ = PClient {clientCorrId, sentCommands}} (pKey, qId, cmd) = do
   corrId <- liftIO $ atomically getNextCorrId
-  t <- signTransmission $ encodeTransmission thVersion sessionId (corrId, qId, cmd)
+  let t = signTransmission $ encodeTransmission thVersion sessionId (corrId, qId, cmd)
   r <- liftIO . atomically $ mkRequest corrId
   pure (t, r)
   where
@@ -549,12 +548,8 @@ mkTransmission ProtocolClient {sessionId, thVersion, client_ = PClient {clientCo
     getNextCorrId = do
       i <- stateTVar clientCorrId $ \i -> (i, i + 1)
       pure . CorrId $ bshow i
-    signTransmission :: ByteString -> ExceptT ProtocolClientError IO SentRawTransmission
-    signTransmission t = case pKey of
-      Nothing -> pure (Nothing, t)
-      Just pk -> do
-        sig <- liftError PCESignatureError $ C.sign pk t
-        return (Just sig, t)
+    signTransmission :: ByteString -> SentRawTransmission
+    signTransmission t = ((`C.sign` t) <$> pKey, t)
     mkRequest :: CorrId -> STM (TMVar (Response msg))
     mkRequest corrId = do
       r <- newEmptyTMVar

src/Simplex/Messaging/Crypto.hs

@@ -67,7 +67,9 @@ module Simplex.Messaging.Crypto
 
     -- * key encoding/decoding
     encodePubKey,
+    decodePubKey,
     encodePrivKey,
+    decodePrivKey,
     pubKeyBytes,
 
     -- * sign/verify
@@ -888,12 +890,12 @@ cryptoFailable = liftEither . first AESCipherError . CE.eitherCryptoError
 -- | Message signing.
 --
 -- Used by SMP clients to sign SMP commands and by SMP agents to sign messages.
-sign' :: SignatureAlgorithm a => PrivateKey a -> ByteString -> ExceptT CryptoError IO (Signature a)
-sign' (PrivateKeyEd25519 pk k) msg = pure . SignatureEd25519 $ Ed25519.sign pk k msg
-sign' (PrivateKeyEd448 pk k) msg = pure . SignatureEd448 $ Ed448.sign pk k msg
+sign' :: SignatureAlgorithm a => PrivateKey a -> ByteString -> Signature a
+sign' (PrivateKeyEd25519 pk k) msg = SignatureEd25519 $ Ed25519.sign pk k msg
+sign' (PrivateKeyEd448 pk k) msg = SignatureEd448 $ Ed448.sign pk k msg
 
-sign :: APrivateSignKey -> ByteString -> ExceptT CryptoError IO ASignature
-sign (APrivateSignKey a k) = fmap (ASignature a) . sign' k
+sign :: APrivateSignKey -> ByteString -> ASignature
+sign (APrivateSignKey a k) = ASignature a . sign' k
 
 -- | Signature verification.
 --

src/Simplex/Messaging/Notifications/Server.hs

@@ -257,24 +257,27 @@ ntfPush s@NtfPushServer {pushQ} = forever $ do
   (tkn@NtfTknData {ntfTknId, token = DeviceToken pp _, tknStatus}, ntf) <- atomically (readTBQueue pushQ)
   liftIO $ logDebug $ "sending push notification to " <> T.pack (show pp)
   status <- readTVarIO tknStatus
-  case (status, ntf) of
-    (_, PNVerification _) ->
-      -- TODO check token status
-      deliverNotification pp tkn ntf >>= \case
-        Right _ -> do
-          status_ <- atomically $ stateTVar tknStatus $ \status' -> if status' == NTActive then (Nothing, NTActive) else (Just NTConfirmed, NTConfirmed)
-          forM_ status_ $ \status' -> withNtfLog $ \sl -> logTokenStatus sl ntfTknId status'
-        _ -> pure ()
-    (NTActive, PNCheckMessages) ->
+  case ntf of
+    PNVerification _
+      | status /= NTInvalid && status /= NTExpired ->
+        deliverNotification pp tkn ntf >>= \case
+          Right _ -> do
+            status_ <- atomically $ stateTVar tknStatus $ \status' -> if status' == NTActive then (Nothing, NTActive) else (Just NTConfirmed, NTConfirmed)
+            forM_ status_ $ \status' -> withNtfLog $ \sl -> logTokenStatus sl ntfTknId status'
+          _ -> pure ()
+      | otherwise -> logError "bad notification token status"
+    PNCheckMessages -> checkActiveTkn status $ do
       void $ deliverNotification pp tkn ntf
-    (NTActive, PNMessage {}) -> do
+    PNMessage {} -> checkActiveTkn status $ do
       stats <- asks serverStats
       atomically $ updatePeriodStats (activeTokens stats) ntfTknId
       void $ deliverNotification pp tkn ntf
       incNtfStat ntfDelivered
-    _ ->
-      liftIO $ logError "bad notification token status"
   where
+    checkActiveTkn :: NtfTknStatus -> M () -> M ()
+    checkActiveTkn status action
+      | status == NTActive = action
+      | otherwise = liftIO $ logError "bad notification token status"
     deliverNotification :: PushProvider -> NtfTknData -> PushNotification -> M (Either PushProviderError ())
     deliverNotification pp tkn@NtfTknData {ntfTknId, tknStatus} ntf = do
       deliver <- liftIO $ getPushClient s pp
@@ -361,13 +364,11 @@ verifyNtfTransmission (sig_, signed, (corrId, entId, _)) cmd = do
       s_ <- atomically $ findNtfSubscription st smpQueue
       case s_ of
         Nothing -> do
-          -- TODO move active token check here to differentiate error
           t_ <- atomically $ getActiveNtfToken st tknId
           verifyToken' t_ $ VRVerified (NtfReqNew corrId (ANE SSubscription sub))
         Just s@NtfSubData {tokenId = subTknId} ->
           if subTknId == tknId
             then do
-              -- TODO move active token check here to differentiate error
               t_ <- atomically $ getActiveNtfToken st subTknId
               verifyToken' t_ $ verifiedSubCmd s c
             else pure $ maybe False (dummyVerifyCmd signed) sig_ `seq` VRFailed
@@ -375,7 +376,6 @@ verifyNtfTransmission (sig_, signed, (corrId, entId, _)) cmd = do
       s_ <- atomically $ getNtfSubscription st entId
       case s_ of
         Just s@NtfSubData {tokenId = subTknId} -> do
-          -- TODO move active token check here to differentiate error
           t_ <- atomically $ getActiveNtfToken st subTknId
           verifyToken' t_ $ verifiedSubCmd s c
         _ -> pure $ maybe False (dummyVerifyCmd signed) sig_ `seq` VRFailed
@@ -512,7 +512,7 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
         (corrId,subId,) <$> case cmd of
           SNEW (NewNtfSub _ _ notifierKey) -> do
             logDebug "SNEW - existing subscription"
-            -- TODO retry if subscription failed, if pending or AUTH do nothing
+            -- possible improvement: retry if subscription failed, if pending or AUTH do nothing
             pure $
               if notifierKey == registeredNKey
                 then NRSubId subId

src/Simplex/Messaging/Notifications/Server/Main.hs

@@ -27,6 +27,9 @@ import System.FilePath (combine)
 import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
 import Text.Read (readMaybe)
 
+ntfServerVersion :: String
+ntfServerVersion = "1.3.0"
+
 ntfServerCLI :: FilePath -> FilePath -> IO ()
 ntfServerCLI cfgPath logPath =
   getCliCommand' (cliCommandP cfgPath logPath iniFile) serverVersion >>= \case
@@ -45,7 +48,7 @@ ntfServerCLI cfgPath logPath =
       putStrLn "Deleted configuration and log files"
   where
     iniFile = combine cfgPath "ntf-server.ini"
-    serverVersion = "SMP notifications server v1.2.0"
+    serverVersion = "SMP notifications server v" <> ntfServerVersion
     defaultServerPort = "443"
     executableName = "ntf-server"
     storeLogFilePath = combine logPath "ntf-server-store.log"

src/Simplex/Messaging/Notifications/Server/Push/APNS.hs

@@ -97,7 +97,7 @@ readECPrivateKey f = do
 data PushNotification
   = PNVerification NtfRegCode
   | PNMessage PNMessageData
-  | PNAlert Text
+  -- | PNAlert Text
   | PNCheckMessages
   deriving (Show)
 
@@ -287,14 +287,14 @@ apnsNotification NtfTknData {tknDhSecret} nonce paddedLen = \case
   PNMessage pnMessageData ->
     encrypt (strEncode pnMessageData) $ \ntfData ->
       apn apnMutableContent . Just $ J.object ["nonce" .= nonce, "message" .= ntfData]
-  PNAlert text -> Right $ apn (apnAlert $ APNSAlertText text) Nothing
+  -- PNAlert text -> Right $ apn (apnAlert $ APNSAlertText text) Nothing
   PNCheckMessages -> Right $ apn APNSBackground {contentAvailable = 1} . Just $ J.object ["checkMessages" .= True]
   where
     encrypt :: ByteString -> (Text -> APNSNotification) -> Either C.CryptoError APNSNotification
     encrypt ntfData f = f . safeDecodeUtf8 . U.encode <$> C.cbEncrypt tknDhSecret nonce ntfData paddedLen
     apn aps notificationData = APNSNotification {aps, notificationData}
     apnMutableContent = APNSMutableContent {mutableContent = 1, alert = APNSAlertText "Encrypted message or another app event", category = Just ntfCategoryCheckMessage}
-    apnAlert alert = APNSAlert {alert, badge = Nothing, sound = Nothing, category = Nothing}
+    -- apnAlert alert = APNSAlert {alert, badge = Nothing, sound = Nothing, category = Nothing}
 
 apnsRequest :: APNSPushClient -> ByteString -> APNSNotification -> IO Request
 apnsRequest c tkn ntf@APNSNotification {aps} = do

src/Simplex/Messaging/Protocol.hs

@@ -141,6 +141,7 @@ import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Char (isPrint, isSpace)
+import Data.Functor (($>))
 import Data.Kind
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
@@ -303,12 +304,17 @@ data RcvMessage = RcvMessage
   deriving (Eq, Show)
 
 -- | received message without server/recipient encryption
-data Message = Message
-  { msgId :: MsgId,
-    msgTs :: SystemTime,
-    msgFlags :: MsgFlags,
-    msgBody :: C.MaxLenBS MaxMessageLen
-  }
+data Message
+  = Message
+      { msgId :: MsgId,
+        msgTs :: SystemTime,
+        msgFlags :: MsgFlags,
+        msgBody :: C.MaxLenBS MaxMessageLen
+      }
+  | MessageQuota
+      { msgId :: MsgId,
+        msgTs :: SystemTime
+      }
 
 instance StrEncoding RcvMessage where
   strEncode RcvMessage {msgId, msgTs, msgFlags, msgBody = EncRcvMsgBody body} =
@@ -328,44 +334,72 @@ instance StrEncoding RcvMessage where
 newtype EncRcvMsgBody = EncRcvMsgBody ByteString
   deriving (Eq, Show)
 
-data RcvMsgBody = RcvMsgBody
-  { msgTs :: SystemTime,
-    msgFlags :: MsgFlags,
-    msgBody :: C.MaxLenBS MaxMessageLen
-  }
+data RcvMsgBody
+  = RcvMsgBody
+      { msgTs :: SystemTime,
+        msgFlags :: MsgFlags,
+        msgBody :: C.MaxLenBS MaxMessageLen
+      }
+  | RcvMsgQuota
+      { msgTs :: SystemTime
+      }
+
+msgQuotaTag :: ByteString
+msgQuotaTag = "QUOTA"
 
 encodeRcvMsgBody :: RcvMsgBody -> C.MaxLenBS MaxRcvMessageLen
-encodeRcvMsgBody RcvMsgBody {msgTs, msgFlags, msgBody} =
-  let rcvMeta :: C.MaxLenBS 16 = C.unsafeMaxLenBS $ smpEncode (msgTs, msgFlags, ' ')
-   in C.appendMaxLenBS rcvMeta msgBody
+encodeRcvMsgBody = \case
+  RcvMsgBody {msgTs, msgFlags, msgBody} ->
+    let rcvMeta :: C.MaxLenBS 16 = C.unsafeMaxLenBS $ smpEncode (msgTs, msgFlags, ' ')
+     in C.appendMaxLenBS rcvMeta msgBody
+  RcvMsgQuota {msgTs} ->
+    C.unsafeMaxLenBS $ msgQuotaTag <> " " <> smpEncode msgTs
 
-data ClientRcvMsgBody = ClientRcvMsgBody
-  { msgTs :: SystemTime,
-    msgFlags :: MsgFlags,
-    msgBody :: ByteString
-  }
+data ClientRcvMsgBody
+  = ClientRcvMsgBody
+      { msgTs :: SystemTime,
+        msgFlags :: MsgFlags,
+        msgBody :: ByteString
+      }
+  | ClientRcvMsgQuota
+      { msgTs :: SystemTime
+      }
 
 clientRcvMsgBodyP :: Parser ClientRcvMsgBody
-clientRcvMsgBodyP = do
-  msgTs <- smpP
-  msgFlags <- smpP
-  Tail msgBody <- _smpP
-  pure ClientRcvMsgBody {msgTs, msgFlags, msgBody}
+clientRcvMsgBodyP = msgQuotaP <|> msgBodyP
+  where
+    msgQuotaP = A.string msgQuotaTag *> (ClientRcvMsgQuota <$> _smpP)
+    msgBodyP = do
+      msgTs <- smpP
+      msgFlags <- smpP
+      Tail msgBody <- _smpP
+      pure ClientRcvMsgBody {msgTs, msgFlags, msgBody}
 
 instance StrEncoding Message where
-  strEncode Message {msgId, msgTs, msgFlags, msgBody} =
-    B.unwords
-      [ strEncode msgId,
-        strEncode msgTs,
-        "flags=" <> strEncode msgFlags,
-        strEncode msgBody
-      ]
+  strEncode = \case
+    Message {msgId, msgTs, msgFlags, msgBody} ->
+      B.unwords
+        [ strEncode msgId,
+          strEncode msgTs,
+          "flags=" <> strEncode msgFlags,
+          strEncode msgBody
+        ]
+    MessageQuota {msgId, msgTs} ->
+      B.unwords
+        [ strEncode msgId,
+          strEncode msgTs,
+          "quota"
+        ]
   strP = do
     msgId <- strP_
     msgTs <- strP_
-    msgFlags <- ("flags=" *> strP_) <|> pure noMsgFlags
-    msgBody <- strP
-    pure Message {msgId, msgTs, msgFlags, msgBody}
+    msgQuotaP msgId msgTs <|> msgP msgId msgTs
+    where
+      msgQuotaP msgId msgTs = "quota" $> MessageQuota {msgId, msgTs}
+      msgP msgId msgTs = do
+        msgFlags <- ("flags=" *> strP_) <|> pure noMsgFlags
+        msgBody <- strP
+        pure Message {msgId, msgTs, msgFlags, msgBody}
 
 type EncNMsgMeta = ByteString
 
@@ -377,7 +411,9 @@ data SMPMsgMeta = SMPMsgMeta
   deriving (Show)
 
 rcvMessageMeta :: MsgId -> ClientRcvMsgBody -> SMPMsgMeta
-rcvMessageMeta msgId ClientRcvMsgBody {msgTs, msgFlags} = SMPMsgMeta {msgId, msgTs, msgFlags}
+rcvMessageMeta msgId = \case
+  ClientRcvMsgBody {msgTs, msgFlags} -> SMPMsgMeta {msgId, msgTs, msgFlags}
+  ClientRcvMsgQuota {msgTs} -> SMPMsgMeta {msgId, msgTs, msgFlags = noMsgFlags}
 
 data NMsgMeta = NMsgMeta
   { msgId :: MsgId,
@@ -852,7 +888,7 @@ data ErrorType
   | -- | internal server error
     INTERNAL
   | -- | used internally, never returned by the server (to be removed)
-    DUPLICATE_ -- TODO remove, not part of SMP protocol
+    DUPLICATE_ -- not part of SMP protocol, used internally
   deriving (Eq, Generic, Read, Show)
 
 instance ToJSON ErrorType where
@@ -1124,26 +1160,33 @@ instance Encoding CommandError where
       _ -> fail "bad command error type"
 
 -- | Send signed SMP transmission to TCP transport.
-tPut :: Transport c => THandle c -> NonEmpty SentRawTransmission -> IO (NonEmpty (Either TransportError ()))
+tPut :: Transport c => THandle c -> NonEmpty SentRawTransmission -> IO [Either TransportError ()]
 tPut th trs
   | batch th = tPutBatch [] $ L.map tEncode trs
-  | otherwise = forM trs $ tPutBlock th . tEncode
+  | otherwise = forM (L.toList trs) $ tPutLog . tEncode
   where
-    tPutBatch :: [Either TransportError ()] -> NonEmpty ByteString -> IO (NonEmpty (Either TransportError ()))
+    tPutBatch :: [Either TransportError ()] -> NonEmpty ByteString -> IO [Either TransportError ()]
     tPutBatch rs ts = do
       let (n, s, ts_) = encodeBatch 0 "" ts
-      r <- if n == 0 then pure [Left TELargeMsg] else replicate n <$> tPutBlock th (lenEncode n `B.cons` s)
+      r <- if n == 0 then largeMsg else replicate n <$> tPutLog (lenEncode n `B.cons` s)
       let rs' = rs <> r
       case ts_ of
         Just ts' -> tPutBatch rs' ts'
-        _ -> pure $ L.fromList rs'
+        _ -> pure rs'
+    largeMsg = putStrLn "tPut error: large message" >> pure [Left TELargeMsg]
+    tPutLog s = do
+      r <- tPutBlock th s
+      case r of
+        Left e -> putStrLn ("tPut error: " <> show e)
+        _ -> pure ()
+      pure r
     encodeBatch :: Int -> ByteString -> NonEmpty ByteString -> (Int, ByteString, Maybe (NonEmpty ByteString))
     encodeBatch n s ts@(t :| ts_)
       | n == 255 = (n, s, Just ts)
       | otherwise =
         let s' = s <> smpEncode (Large t)
             n' = n + 1
-         in if B.length s' > blockSize th - 1
+         in if B.length s' > blockSize th - 1 -- one byte is reserved for the number of messages in the batch
               then (n,s,) $ if n == 0 then L.nonEmpty ts_ else Just ts
               else case L.nonEmpty ts_ of
                 Just ts' -> encodeBatch n' s' ts'

src/Simplex/Messaging/Server.hs

@@ -256,7 +256,6 @@ receive th Client {rcvQ, sndQ, activeAt} = forever $ do
 send :: Transport c => THandle c -> Client -> IO ()
 send h@THandle {thVersion = v} Client {sndQ, sessionId, activeAt} = forever $ do
   ts <- atomically $ L.sortWith tOrder <$> readTBQueue sndQ
-  -- TODO the line below can return Lefts, but we ignore it and do not disconnect the client
   void . liftIO . tPut h $ L.map ((Nothing,) . encodeTransmission v sessionId) ts
   atomically . writeTVar activeAt =<< liftIO getSystemTime
   where
@@ -538,20 +537,22 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
           | otherwise = case status qr of
             QueueOff -> return $ err AUTH
             QueueActive ->
-              mapM mkMessage (C.maxLenBS msgBody) >>= \case
+              case C.maxLenBS msgBody of
                 Left _ -> pure $ err LARGE_MSG
-                Right msg -> do
-                  resp@(_, _, sent) <- time "SEND" $ do
+                Right body -> do
+                  msg_ <- time "SEND" $ do
                     q <- getStoreMsgQueue "SEND" $ recipientId qr
                     expireMessages q
-                    atomically $ ifM (isFull q) (pure $ err QUOTA) (writeMsg q msg $> ok)
-                  when (sent == OK) . time "SEND ok" $ do
-                    when (notification msgFlags) $
-                      atomically . trySendNotification msg =<< asks idsDrg
-                    stats <- asks serverStats
-                    atomically $ modifyTVar (msgSent stats) (+ 1)
-                    atomically $ updatePeriodStats (activeQueues stats) (recipientId qr)
-                  pure resp
+                    atomically . writeMsg q =<< mkMessage body
+                  case msg_ of
+                    Nothing -> pure $ err QUOTA
+                    Just msg -> time "SEND ok" $ do
+                      when (notification msgFlags) $
+                        atomically . trySendNotification msg =<< asks idsDrg
+                      stats <- asks serverStats
+                      atomically $ modifyTVar (msgSent stats) (+ 1)
+                      atomically $ updatePeriodStats (activeQueues stats) (recipientId qr)
+                      pure ok
           where
             mkMessage :: C.MaxLenBS MaxMessageLen -> m Message
             mkMessage body = do
@@ -572,12 +573,14 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
 
             writeNtf :: NotifierId -> Message -> RcvNtfDhSecret -> TVar ChaChaDRG -> Client -> STM ()
             writeNtf nId msg rcvNtfDhSecret ntfNonceDrg Client {sndQ = q} =
-              unlessM (isFullTBQueue q) $ do
-                (nmsgNonce, encNMsgMeta) <- mkMessageNotification msg rcvNtfDhSecret ntfNonceDrg
-                writeTBQueue q [(CorrId "", nId, NMSG nmsgNonce encNMsgMeta)]
+              unlessM (isFullTBQueue q) $ case msg of
+                Message {msgId, msgTs} -> do
+                  (nmsgNonce, encNMsgMeta) <- mkMessageNotification msgId msgTs rcvNtfDhSecret ntfNonceDrg
+                  writeTBQueue q [(CorrId "", nId, NMSG nmsgNonce encNMsgMeta)]
+                _ -> pure ()
 
-            mkMessageNotification :: Message -> RcvNtfDhSecret -> TVar ChaChaDRG -> STM (C.CbNonce, EncNMsgMeta)
-            mkMessageNotification Message {msgId, msgTs} rcvNtfDhSecret ntfNonceDrg = do
+            mkMessageNotification :: ByteString -> SystemTime -> RcvNtfDhSecret -> TVar ChaChaDRG -> STM (C.CbNonce, EncNMsgMeta)
+            mkMessageNotification msgId msgTs rcvNtfDhSecret ntfNonceDrg = do
               cbNonce <- C.pseudoRandomCbNonce ntfNonceDrg
               let msgMeta = NMsgMeta {msgId, msgTs}
                   encNMsgMeta = C.cbEncrypt rcvNtfDhSecret cbNonce (smpEncode msgMeta) 128
@@ -615,17 +618,22 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
         time name = timed name queueId
 
         encryptMsg :: QueueRec -> Message -> RcvMessage
-        encryptMsg qr Message {msgId, msgTs, msgFlags, msgBody}
-          | thVersion == 1 || thVersion == 2 = encrypt msgBody
-          | otherwise = encrypt $ encodeRcvMsgBody RcvMsgBody {msgTs, msgFlags, msgBody}
+        encryptMsg qr msg = case msg of
+          Message {msgFlags, msgBody}
+            | thVersion == 1 || thVersion == 2 -> encrypt msgFlags msgBody
+            | otherwise -> encrypt msgFlags $ encodeRcvMsgBody RcvMsgBody {msgTs = msgTs', msgFlags, msgBody}
+          MessageQuota {} ->
+            encrypt noMsgFlags $ encodeRcvMsgBody (RcvMsgQuota msgTs')
           where
-            encrypt :: KnownNat i => C.MaxLenBS i -> RcvMessage
-            encrypt body =
-              let encBody = EncRcvMsgBody $ C.cbEncryptMaxLenBS (rcvDhSecret qr) (C.cbNonce msgId) body
-               in RcvMessage msgId msgTs msgFlags encBody
+            encrypt :: KnownNat i => MsgFlags -> C.MaxLenBS i -> RcvMessage
+            encrypt msgFlags body =
+              let encBody = EncRcvMsgBody $ C.cbEncryptMaxLenBS (rcvDhSecret qr) (C.cbNonce msgId') body
+               in RcvMessage msgId' msgTs' msgFlags encBody
+            msgId' = msgId (msg :: Message)
+            msgTs' = msgTs (msg :: Message)
 
         setDelivered :: Sub -> Message -> STM Bool
-        setDelivered s Message {msgId} = tryPutTMVar (delivered s) msgId
+        setDelivered s msg = tryPutTMVar (delivered s) $ msgId (msg :: Message)
 
         getStoreMsgQueue :: T.Text -> RecipientId -> m MsgQueue
         getStoreMsgQueue name rId = time (name <> " getMsgQueue") $ do
@@ -717,7 +725,7 @@ restoreServerMessages = asks (storeMsgsFile . config) >>= mapM_ restoreMessages
             addToMsgQueue rId msg = do
               full <- atomically $ do
                 q <- getMsgQueue ms rId quota
-                ifM (isFull q) (pure True) (writeMsg q msg $> False)
+                isNothing <$> writeMsg q msg
               when full . logError . decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (msgId (msg :: Message))
             updateMsgV1toV3 QueueRec {rcvDhSecret} RcvMessage {msgId, msgTs, msgFlags, msgBody = EncRcvMsgBody body} = do
               let nonce = C.cbNonce msgId

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

@@ -39,7 +39,7 @@ data ServerConfig = ServerConfig
   { transports :: [(ServiceName, ATransport)],
     tbqSize :: Natural,
     serverTbqSize :: Natural,
-    msgQueueQuota :: Natural,
+    msgQueueQuota :: Int,
     queueIdBytes :: Int,
     msgIdBytes :: Int,
     storeLogFile :: Maybe FilePath,

src/Simplex/Messaging/Server/MsgStore.hs

@@ -6,7 +6,6 @@ module Simplex.Messaging.Server.MsgStore where
 
 import Control.Applicative ((<|>))
 import Data.Int (Int64)
-import Numeric.Natural
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (Message (..), MsgId, RcvMessage (..), RecipientId)
 
@@ -19,13 +18,12 @@ instance StrEncoding MsgLogRecord where
   strP = "v3 " *> (MLRv3 <$> strP_ <*> strP) <|> MLRv1 <$> strP_ <*> strP
 
 class MonadMsgStore s q m | s -> q where
-  getMsgQueue :: s -> RecipientId -> Natural -> m q
+  getMsgQueue :: s -> RecipientId -> Int -> m q
   delMsgQueue :: s -> RecipientId -> m ()
   flushMsgQueue :: s -> RecipientId -> m [Message]
 
 class MonadMsgQueue q m where
-  isFull :: q -> m Bool
-  writeMsg :: q -> Message -> m () -- non blocking
+  writeMsg :: q -> Message -> m (Maybe Message) -- non blocking
   tryPeekMsg :: q -> m (Maybe Message) -- non blocking
   peekMsg :: q -> m Message -- blocking
   tryDelMsg :: q -> MsgId -> m Bool -- non blocking

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

@@ -7,22 +7,31 @@
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE TupleSections #-}
 
-module Simplex.Messaging.Server.MsgStore.STM where
+module Simplex.Messaging.Server.MsgStore.STM
+  ( STMMsgStore,
+    MsgQueue,
+    newMsgStore,
+  )
+where
 
-import Control.Concurrent.STM.TBQueue (flushTBQueue)
+import Control.Concurrent.STM.TQueue (flushTQueue)
 import Control.Monad (when)
 import qualified Data.ByteString.Char8 as B
 import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.Time.Clock.System (SystemTime (systemSeconds))
-import Numeric.Natural
 import Simplex.Messaging.Protocol (Message (..), MsgId, RecipientId)
 import Simplex.Messaging.Server.MsgStore
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import UnliftIO.STM
 
-newtype MsgQueue = MsgQueue {msgQueue :: TBQueue Message}
+data MsgQueue = MsgQueue
+  { msgQueue :: TQueue Message,
+    quota :: Int,
+    canWrite :: TVar Bool,
+    size :: TVar Int
+  }
 
 type STMMsgStore = TMap RecipientId MsgQueue
 
@@ -30,54 +39,77 @@ newMsgStore :: STM STMMsgStore
 newMsgStore = TM.empty
 
 instance MonadMsgStore STMMsgStore MsgQueue STM where
-  getMsgQueue :: STMMsgStore -> RecipientId -> Natural -> STM MsgQueue
+  getMsgQueue :: STMMsgStore -> RecipientId -> Int -> STM MsgQueue
   getMsgQueue st rId quota = maybe newQ pure =<< TM.lookup rId st
     where
       newQ = do
-        q <- MsgQueue <$> newTBQueue quota
+        msgQueue <- newTQueue
+        canWrite <- newTVar True
+        size <- newTVar 0
+        let q = MsgQueue {msgQueue, quota, canWrite, size}
         TM.insert rId q st
-        return q
+        pure q
 
   delMsgQueue :: STMMsgStore -> RecipientId -> STM ()
   delMsgQueue st rId = TM.delete rId st
 
   flushMsgQueue :: STMMsgStore -> RecipientId -> STM [Message]
-  flushMsgQueue st rId = TM.lookup rId st >>= maybe (pure []) (flushTBQueue . msgQueue)
+  flushMsgQueue st rId = TM.lookupDelete rId st >>= maybe (pure []) (flushTQueue . msgQueue)
 
 instance MonadMsgQueue MsgQueue STM where
-  isFull :: MsgQueue -> STM Bool
-  isFull = isFullTBQueue . msgQueue
-
-  writeMsg :: MsgQueue -> Message -> STM ()
-  writeMsg = writeTBQueue . msgQueue
+  writeMsg :: MsgQueue -> Message -> STM (Maybe Message)
+  writeMsg MsgQueue {msgQueue = q, quota, canWrite, size} msg = do
+    canWrt <- readTVar canWrite
+    empty <- isEmptyTQueue q
+    if canWrt || empty
+      then do
+        canWrt' <- (quota >) <$> readTVar size
+        writeTVar canWrite canWrt'
+        modifyTVar' size (+ 1)
+        if canWrt'
+          then writeTQueue q msg $> Just msg
+          else writeTQueue q msgQuota $> Nothing
+      else pure Nothing
+    where
+      msgQuota = MessageQuota {msgId = msgId msg, msgTs = msgTs msg}
 
   tryPeekMsg :: MsgQueue -> STM (Maybe Message)
-  tryPeekMsg = tryPeekTBQueue . msgQueue
+  tryPeekMsg = tryPeekTQueue . msgQueue
+  {-# INLINE tryPeekMsg #-}
 
   peekMsg :: MsgQueue -> STM Message
-  peekMsg = peekTBQueue . msgQueue
+  peekMsg = peekTQueue . msgQueue
+  {-# INLINE peekMsg #-}
 
   tryDelMsg :: MsgQueue -> MsgId -> STM Bool
-  tryDelMsg (MsgQueue q) msgId' =
-    tryPeekTBQueue q >>= \case
-      Just Message {msgId}
-        | msgId == msgId' || B.null msgId' -> tryReadTBQueue q $> True
+  tryDelMsg mq msgId' =
+    tryPeekMsg mq >>= \case
+      Just msg
+        | msgId msg == msgId' || B.null msgId' -> tryDeleteMsg mq >> pure True
         | otherwise -> pure False
       _ -> pure False
 
   -- atomic delete (== read) last and peek next message if available
   tryDelPeekMsg :: MsgQueue -> MsgId -> STM (Bool, Maybe Message)
-  tryDelPeekMsg (MsgQueue q) msgId' =
-    tryPeekTBQueue q >>= \case
-      msg_@(Just Message {msgId})
-        | msgId == msgId' || B.null msgId' -> (True,) <$> (tryReadTBQueue q >> tryPeekTBQueue q)
+  tryDelPeekMsg mq msgId' =
+    tryPeekMsg mq >>= \case
+      msg_@(Just msg)
+        | msgId msg == msgId' || B.null msgId' -> (True,) <$> (tryDeleteMsg mq >> tryPeekMsg mq)
         | otherwise -> pure (False, msg_)
       _ -> pure (False, Nothing)
 
   deleteExpiredMsgs :: MsgQueue -> Int64 -> STM ()
-  deleteExpiredMsgs (MsgQueue q) old = loop
+  deleteExpiredMsgs mq old = loop
     where
-      loop = tryPeekTBQueue q >>= mapM_ delOldMsg
-      delOldMsg Message {msgTs} =
-        when (systemSeconds msgTs < old) $
-          tryReadTBQueue q >> loop
+      loop = tryPeekMsg mq >>= mapM_ delOldMsg
+      delOldMsg = \case
+        Message {msgTs} ->
+          when (systemSeconds msgTs < old) $
+            tryDeleteMsg mq >> loop
+        _ -> pure ()
+
+tryDeleteMsg :: MsgQueue -> STM ()
+tryDeleteMsg MsgQueue {msgQueue = q, size} =
+  tryReadTQueue q >>= \case
+    Just _ -> modifyTVar' size (subtract 1)
+    _ -> pure ()

src/Simplex/Messaging/Transport.hs

@@ -100,7 +100,7 @@ supportedSMPServerVRange :: VersionRange
 supportedSMPServerVRange = mkVersionRange 1 5
 
 simplexMQVersion :: String
-simplexMQVersion = "4.1.0"
+simplexMQVersion = "4.2.0"
 
 -- * Transport connection class
 

tests/AgentTests.hs

@@ -78,6 +78,8 @@ agentTests (ATransport t) = do
       smpAgentTest2_2_1 $ testConcurrentMsgDelivery t
     it "should deliver messages if one of connections has quota exceeded" $
       smpAgentTest2_2_1 $ testMsgDeliveryQuotaExceeded t
+    it "should resume delivering messages after exceeding quota once all messages are received" $
+      smpAgentTest2_2_1 $ testResumeDeliveryQuotaExceeded t
 
 tGetAgent :: Transport c => c -> IO (ATransmissionOrError 'Agent)
 tGetAgent h = do
@@ -430,6 +432,32 @@ testMsgDeliveryQuotaExceeded _ alice bob = do
   -- if delivery is blocked it won't go further
   alice <# ("", "bob2", SENT 4)
 
+testResumeDeliveryQuotaExceeded :: Transport c => TProxy c -> c -> c -> IO ()
+testResumeDeliveryQuotaExceeded _ alice bob = do
+  connect (alice, "alice") (bob, "bob")
+  forM_ [1 .. 4 :: Int] $ \i -> do
+    let corrId = bshow i
+        msg = "message " <> bshow i
+    (_, "bob", Right (MID mId)) <- alice #: (corrId, "bob", "SEND F :" <> msg)
+    alice <#= \case ("", "bob", SENT m) -> m == mId; _ -> False
+  ("5", "bob", Right (MID 8)) <- alice #: ("5", "bob", "SEND F :over quota")
+  alice #:# "the last message not sent yet"
+  bob <#= \case ("", "alice", Msg "message 1") -> True; _ -> False
+  bob #: ("1", "alice", "ACK 4") #> ("1", "alice", OK)
+  alice #:# "the last message not sent"
+  bob <#= \case ("", "alice", Msg "message 2") -> True; _ -> False
+  bob #: ("2", "alice", "ACK 5") #> ("2", "alice", OK)
+  alice #:# "the last message not sent"
+  bob <#= \case ("", "alice", Msg "message 3") -> True; _ -> False
+  bob #: ("3", "alice", "ACK 6") #> ("3", "alice", OK)
+  alice #:# "the last message not sent"
+  bob <#= \case ("", "alice", Msg "message 4") -> True; _ -> False
+  bob #: ("4", "alice", "ACK 7") #> ("4", "alice", OK)
+  alice <# ("", "bob", SENT 8)
+  bob <#= \case ("", "alice", Msg "over quota") -> True; _ -> False
+  -- message 8 is skipped because of alice agent sending "QCONT" message
+  bob #: ("5", "alice", "ACK 9") #> ("5", "alice", OK)
+
 connect :: forall c. Transport c => (c, ByteString) -> (c, ByteString) -> IO ()
 connect (h1, name1) (h2, name2) = do
   ("c1", _, Right (INV cReq)) <- h1 #: ("c1", name2, "NEW T INV")

tests/CLITests.hs

@@ -6,6 +6,7 @@ import Data.Ini (lookupValue, readIniFile)
 import Data.List (isPrefixOf)
 import Simplex.Messaging.Notifications.Server.Main
 import Simplex.Messaging.Server.Main
+import Simplex.Messaging.Transport (simplexMQVersion)
 import Simplex.Messaging.Util (catchAll_)
 import System.Directory (doesFileExist)
 import System.Environment (withArgs)
@@ -51,7 +52,7 @@ smpServerTest storeLog basicAuth = do
   lookupValue "INACTIVE_CLIENTS" "disconnect" ini `shouldBe` Right "off"
   doesFileExist (cfgPath <> "/ca.key") `shouldReturn` True
   r <- lines <$> capture_ (withArgs ["start"] $ (100000 `timeout` smpServerCLI cfgPath logPath) `catchAll_` pure (Just ()))
-  r `shouldContain` ["SMP server v4.0.0"]
+  r `shouldContain` ["SMP server v" <> simplexMQVersion]
   r `shouldContain` (if storeLog then ["Store log: " <> logPath <> "/smp-server-store.log"] else ["Store log disabled."])
   r `shouldContain` ["Listening on port 5223 (TLS)..."]
   r `shouldContain` ["not expiring inactive clients"]
@@ -71,7 +72,7 @@ ntfServerTest storeLog = do
   lookupValue "TRANSPORT" "websockets" ini `shouldBe` Right "off"
   doesFileExist (ntfCfgPath <> "/ca.key") `shouldReturn` True
   r <- lines <$> capture_ (withArgs ["start"] $ (100000 `timeout` ntfServerCLI ntfCfgPath ntfLogPath) `catchAll_` pure (Just ()))
-  r `shouldContain` ["SMP notifications server v1.2.0"]
+  r `shouldContain` ["SMP notifications server v" <> ntfServerVersion]
   r `shouldContain` (if storeLog then ["Store log: " <> ntfLogPath <> "/ntf-server-store.log"] else ["Store log disabled."])
   r `shouldContain` ["Listening on port 443 (TLS)..."]
   capture_ (withStdin "Y" . withArgs ["delete"] $ ntfServerCLI ntfCfgPath ntfLogPath)

tests/CoreTests/CryptoTests.hs

@@ -1,18 +1,20 @@
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
 
 module CoreTests.CryptoTests (cryptoTests) where
 
+import qualified Data.ByteString.Char8 as B
+import Data.Either (isRight)
+import qualified Data.Text as T
+import Data.Text.Encoding (encodeUtf8)
 import qualified Simplex.Messaging.Crypto as C
 import Test.Hspec
 import Test.Hspec.QuickCheck (modifyMaxSuccess)
 import Test.QuickCheck
-import qualified Data.Text as T
-import qualified Data.ByteString.Char8 as B
-import Data.Text.Encoding (encodeUtf8)
 
 cryptoTests :: Spec
-cryptoTests = modifyMaxSuccess (const 10000) $ do
-  describe "padding / unpadding" $ do
+cryptoTests = do
+  modifyMaxSuccess (const 10000) . describe "padding / unpadding" $ do
     it "should pad / unpad string" . property $ \(s, paddedLen) ->
       let b = encodeUtf8 $ T.pack s
           len = B.length b
@@ -34,3 +36,37 @@ cryptoTests = modifyMaxSuccess (const 10000) $ do
     it "unpad should fail on shorter string" $ do
       C.unPad "\000\003abc" `shouldBe` Right "abc"
       C.unPad "\000\003ab" `shouldBe` Left C.CryptoInvalidMsgError
+  describe "Ed signatures" $ do
+    describe "Ed25519" $ testSignature C.SEd25519
+    describe "Ed448" $ testSignature C.SEd448
+  describe "DH X25519 + cryptobox" $
+    testDHCryptoBox
+  describe "X509 key encoding" $ do
+    describe "Ed25519" $ testEncoding C.SEd25519
+    describe "Ed448" $ testEncoding C.SEd448
+    describe "X25519" $ testEncoding C.SX25519
+    describe "X448" $ testEncoding C.SX448
+
+testSignature :: (C.AlgorithmI a, C.SignatureAlgorithm a) => C.SAlgorithm a -> Spec
+testSignature alg = it "should sign / verify string" . ioProperty $ do
+  (k, pk) <- C.generateSignatureKeyPair alg
+  pure $ \s -> let b = encodeUtf8 $ T.pack s in C.verify k (C.sign pk b) b
+
+testDHCryptoBox :: Spec
+testDHCryptoBox = it "should encrypt / decrypt string" . ioProperty $ do
+  (sk, spk) <- C.generateKeyPair'
+  (rk, rpk) <- C.generateKeyPair'
+  nonce <- C.randomCbNonce
+  pure $ \(s, pad) ->
+    let b = encodeUtf8 $ T.pack s
+        paddedLen = B.length b + abs pad + 2
+        cipher = C.cbEncrypt (C.dh' rk spk) nonce b paddedLen
+        plain = C.cbDecrypt (C.dh' sk rpk) nonce =<< cipher
+     in isRight cipher && cipher /= plain && Right b == plain
+
+testEncoding :: (C.AlgorithmI a) => C.SAlgorithm a -> Spec
+testEncoding alg = it "should encode / decode key" . ioProperty $ do
+  (k, pk) <- C.generateKeyPair alg
+  pure $ \(_ :: Int) ->
+    C.decodePubKey (C.encodePubKey k) == Right k
+      && C.decodePrivKey (C.encodePrivKey pk) == Right pk

tests/CoreTests/RetryIntervalTests.hs

@@ -0,0 +1,61 @@
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module CoreTests.RetryIntervalTests where
+
+import Control.Concurrent.STM
+import Control.Monad (when)
+import Data.Time.Clock (UTCTime, diffUTCTime, getCurrentTime, nominalDiffTimeToSeconds)
+import Simplex.Messaging.Agent.RetryInterval
+import Test.Hspec
+
+retryIntervalTests :: Spec
+retryIntervalTests = do
+  describe "Retry interval with 2 modes and lock" $ do
+    testRetryIntervalSameMode
+    testRetryIntervalSwitchMode
+
+testRI :: RetryInterval2
+testRI =
+  RetryInterval2
+    { riSlow =
+        RetryInterval
+          { initialInterval = 20000,
+            increaseAfter = 40000,
+            maxInterval = 40000
+          },
+      riFast =
+        RetryInterval
+          { initialInterval = 10000,
+            increaseAfter = 20000,
+            maxInterval = 40000
+          }
+    }
+
+testRetryIntervalSameMode :: Spec
+testRetryIntervalSameMode =
+  it "should increase elapased time and interval when the mode stays the same" $ do
+    lock <- newEmptyTMVarIO
+    intervals <- newTVarIO []
+    ts <- newTVarIO =<< getCurrentTime
+    withRetryLock2 testRI lock $ \loop -> do
+      ints <- addInterval intervals ts
+      when (length ints < 9) $ loop RIFast
+    (reverse <$> readTVarIO intervals) `shouldReturn` [0, 1, 1, 1, 2, 3, 4, 4, 4]
+
+testRetryIntervalSwitchMode :: Spec
+testRetryIntervalSwitchMode =
+  it "should increase elapased time and interval when the mode stays the same" $ do
+    lock <- newEmptyTMVarIO
+    intervals <- newTVarIO []
+    ts <- newTVarIO =<< getCurrentTime
+    withRetryLock2 testRI lock $ \loop -> do
+      ints <- addInterval intervals ts
+      when (length ints < 11) $ loop $ if length ints <= 5 then RIFast else RISlow
+    (reverse <$> readTVarIO intervals) `shouldReturn` [0, 1, 1, 1, 2, 3, 2, 2, 3, 4, 4]
+
+addInterval :: TVar [Int] -> TVar UTCTime -> IO [Int]
+addInterval intervals ts = do
+  ts' <- getCurrentTime
+  atomically $ do
+    int :: Int <- truncate . (* 100) . nominalDiffTimeToSeconds <$> stateTVar ts (\t -> (diffUTCTime ts' t, ts'))
+    stateTVar intervals $ \ints -> (int : ints, int : ints)

tests/NtfServerTests.hs

@@ -10,7 +10,6 @@
 module NtfServerTests where
 
 import Control.Concurrent (threadDelay)
-import Control.Monad.Except (runExceptT)
 import qualified Data.Aeson as J
 import qualified Data.Aeson.Types as JT
 import Data.Bifunctor (first)
@@ -77,8 +76,7 @@ sendRecvNtf h@THandle {thVersion, sessionId} (sgn, corrId, qId, cmd) = do
 signSendRecvNtf :: forall c e. (Transport c, NtfEntityI e) => THandle c -> C.APrivateSignKey -> (ByteString, ByteString, NtfCommand e) -> IO (SignedTransmission NtfResponse)
 signSendRecvNtf h@THandle {thVersion, sessionId} pk (corrId, qId, cmd) = do
   let t = encodeTransmission thVersion sessionId (CorrId corrId, qId, cmd)
-  Right sig <- runExceptT $ C.sign pk t
-  Right () <- tPut1 h (Just sig, t)
+  Right () <- tPut1 h (Just $ C.sign pk t, t)
   tGet1 h
 
 (.->) :: J.Value -> J.Key -> Either String ByteString

tests/ServerTests.hs

@@ -14,7 +14,7 @@ module ServerTests where
 import Control.Concurrent (ThreadId, killThread, threadDelay)
 import Control.Concurrent.STM
 import Control.Exception (SomeException, try)
-import Control.Monad.Except (forM, forM_, runExceptT)
+import Control.Monad.Except (forM, forM_)
 import Control.Monad.IO.Class
 import Data.Bifunctor (first)
 import Data.ByteString.Base64
@@ -49,6 +49,7 @@ serverTests t@(ATransport t') = do
     describe "switch subscription to another TCP connection" $ testSwitchSub t
     describe "GET command" $ testGetCommand t'
     describe "GET & SUB commands" $ testGetSubCommands t'
+    describe "Exceeding queue quota" $ testExceedQueueQuota t'
   describe "Store log" $ testWithStoreLog t
   describe "Restore messages" $ testRestoreMessages t
   describe "Restore messages (v2)" $ testRestoreMessagesV2 t
@@ -77,8 +78,7 @@ sendRecv h@THandle {thVersion, sessionId} (sgn, corrId, qId, cmd) = do
 signSendRecv :: forall c p. (Transport c, PartyI p) => THandle c -> C.APrivateSignKey -> (ByteString, ByteString, Command p) -> IO (SignedTransmission BrokerMsg)
 signSendRecv h@THandle {thVersion, sessionId} pk (corrId, qId, cmd) = do
   let t = encodeTransmission thVersion sessionId (CorrId corrId, qId, cmd)
-  Right sig <- runExceptT $ C.sign pk t
-  Right () <- tPut1 h (Just sig, t)
+  Right () <- tPut1 h (Just $ C.sign pk t, t)
   tGet1 h
 
 tPut1 :: Transport c => THandle c -> SentRawTransmission -> IO (Either TransportError ())
@@ -104,9 +104,11 @@ decryptMsgV2 :: C.DhSecret 'C.X25519 -> ByteString -> ByteString -> Either C.Cry
 decryptMsgV2 dhShared = C.cbDecrypt dhShared . C.cbNonce
 
 decryptMsgV3 :: C.DhSecret 'C.X25519 -> ByteString -> ByteString -> Either String MsgBody
-decryptMsgV3 dhShared nonce body = do
-  ClientRcvMsgBody {msgBody} <- parseAll clientRcvMsgBodyP =<< first show (C.cbDecrypt dhShared (C.cbNonce nonce) body)
-  pure msgBody
+decryptMsgV3 dhShared nonce body =
+  case parseAll clientRcvMsgBodyP =<< first show (C.cbDecrypt dhShared (C.cbNonce nonce) body) of
+    Right ClientRcvMsgBody {msgBody} -> Right msgBody
+    Right ClientRcvMsgQuota {} -> Left "ClientRcvMsgQuota"
+    Left e -> Left e
 
 testCreateSecureV2 :: forall c. Transport c => TProxy c -> Spec
 testCreateSecureV2 _ =
@@ -494,6 +496,32 @@ testGetSubCommands t =
       Resp "12" _ OK <- signSendRecv rh2 rKey ("12", rId, GET)
       pure ()
 
+testExceedQueueQuota :: forall c. Transport c => TProxy c -> Spec
+testExceedQueueQuota t =
+  it "should reply with ERR QUOTA to sender and send QUOTA message to the recipient" $ do
+    withSmpServerConfigOn (ATransport t) cfg {msgQueueQuota = 2} testPort $ \_ ->
+      testSMPClient @c $ \sh -> testSMPClient @c $ \rh -> do
+        (sPub, sKey) <- C.generateSignatureKeyPair C.SEd25519
+        (sId, rId, rKey, dhShared) <- createAndSecureQueue rh sPub
+        let dec = decryptMsgV3 dhShared
+        Resp "1" _ OK <- signSendRecv sh sKey ("1", sId, _SEND "hello 1")
+        Resp "2" _ OK <- signSendRecv sh sKey ("2", sId, _SEND "hello 2")
+        Resp "3" _ (ERR QUOTA) <- signSendRecv sh sKey ("3", sId, _SEND "hello 3")
+        Resp "" _ (Msg mId1 msg1) <- tGet1 rh
+        (dec mId1 msg1, Right "hello 1") #== "hello 1"
+        Resp "4" _ (Msg mId2 msg2) <- signSendRecv rh rKey ("4", rId, ACK mId1)
+        (dec mId2 msg2, Right "hello 2") #== "hello 2"
+        Resp "5" _ (ERR QUOTA) <- signSendRecv sh sKey ("5", sId, _SEND "hello 3")
+        Resp "6" _ (Msg mId3 msg3) <- signSendRecv rh rKey ("6", rId, ACK mId2)
+        (dec mId3 msg3, Left "ClientRcvMsgQuota") #== "ClientRcvMsgQuota"
+        Resp "7" _ (ERR QUOTA) <- signSendRecv sh sKey ("7", sId, _SEND "hello 3")
+        Resp "8" _ OK <- signSendRecv rh rKey ("8", rId, ACK mId3)
+        Resp "9" _ OK <- signSendRecv sh sKey ("9", sId, _SEND "hello 3")
+        Resp "" _ (Msg mId4 msg4) <- tGet1 rh
+        (dec mId4 msg4, Right "hello 3") #== "hello 3"
+        Resp "10" _ OK <- signSendRecv rh rKey ("10", rId, ACK mId4)
+        pure ()
+
 testWithStoreLog :: ATransport -> Spec
 testWithStoreLog at@(ATransport t) =
   it "should store simplex queues to log and restore them after server restart" $ do

tests/Test.hs

@@ -6,6 +6,7 @@ import CLITests
 import CoreTests.CryptoTests
 import CoreTests.EncodingTests
 import CoreTests.ProtocolErrorTests
+import CoreTests.RetryIntervalTests
 import CoreTests.VersionRangeTests
 import NtfServerTests (ntfServerTests)
 import ServerTests
@@ -31,6 +32,7 @@ main = do
       describe "Protocol error tests" protocolErrorTests
       describe "Version range" versionRangeTests
       describe "Encryption tests" cryptoTests
+      describe "Retry interval tests" retryIntervalTests
     describe "SMP server via TLS" $ serverTests (transport @TLS)
     describe "SMP server via WebSockets" $ serverTests (transport @WS)
     describe "Notifications server" $ ntfServerTests (transport @TLS)