version 1.1.0 (#358 )

* version 1.1.0 * increase test delays to stabilize
SMP confirmation timeout recovery rfc (#357 )
2026-08-31 07:28:23 +00:00 · 2022-04-21 15:00:30 +01:00 · 2022-04-20 22:17:51 +01:00 · 2022-04-20 08:14:26 +01:00 · 2022-04-18 08:56:08 +01:00 · 2022-04-12 11:35:28 +01:00
40 changed files with 1121 additions and 540 deletions
@@ -1,24 +1,58 @@
+# 1.1.0
+
+SMP server:
+
+- message TTL and periodic deletion of old messages
+- configuration to prevent creation of the new queues
+
+SMP agent:
+
+- asynchronous connection handshake
+- configurable SMP servers at run-time
+- use TCP keep-alive for connection stability
+- improve stability of connection subscriptions
+- auto-vacuum DB to remove deleted records
+
+# 1.0.3
+
+SMP server:
+
+- Reduce server message queue quota to 128 messages.
+
+SMP agent:
+
+- Add "yes to migrations" option.
+- Make new SMP client attempt to reconnect on network error.
+- Reduce connection handshake expiration to 2 days.
+
+JSON encoding of types used in simplex-chat, some other minor adjustments.
+
 # 1.0.2

 General:
+
 - Enable TLS 1.3 parameters for TLS handshake (server and client).
 - Switch from hs-tls fork to original repo now that it supports getFinished and getPeerFinished APIs for both TLS 1.2 and TLS 1.3.

 SMP server:
+
 - Perform TLS handshake in a separate thread per-connection.

 SMP agent:
+
 - Cease attempts to send HELLO after one week timeout.
 - Coalesce requests to connect to SMP servers, to have 1 connection per server.

 # 1.0.1

 SMP server:
+
 - Explicitly set line buffering in stdout/stderr to log each line when output is redirected to files.

 # 1.0.0

 Security and privacy improvements:
+
 - Faster and more secure 2-layer E2E encryption with additional encryption layer between servers and recipients:
  - application messages in each duplex connection (managed by SMP agents - see [overview](https://github.com/simplex-chat/simplexmq/blob/master/protocol/overview-tjr.md)) are encrypted using [double-ratchet algorithm](https://www.signal.org/docs/specifications/doubleratchet/), providing forward secrecy and break-in recovery. This layer uses two Curve448 keys per client for [X3DH key agreement](https://www.signal.org/docs/specifications/x3dh/), SHA512 based HKDFs and AES-GCM AEAD encryption.
  - SMP client messages are additionally E2E encrypted in each SMP queue to avoid cipher-text correlation of messages sent via multiple redundant queues (that will be supported soon). This and the next layer use [NaCl crypto_box algorithm](https://nacl.cr.yp.to/index.html) with XSalsa20Poly1305 cipher and Curve25519 keys for DH key agreement.
@@ -31,13 +65,16 @@ Security and privacy improvements:
 - Server identity verification via server offline certificate fingerprints included in SMP server addresses.

 New functionality:
+
 - Support for notification servers with new SMP commands: `NKEY`/`NID`, `NSUB`/`NMSG`.

 Efficiency improvements:
+
 - Binary protocol encodings to reduce overhead from circa 15% to approximately 3.7% of transmitted application message size, with only 2.2% overhead for SMP protocol messages.
 - More performant cryptographic algorithms.

 For more information about SimpleX:
+
 - [SimpleX overview](https://github.com/simplex-chat/simplexmq/blob/master/protocol/overview-tjr.md).
 - [SimpleX chat v1 announcement](https://github.com/simplex-chat/simplex-chat/blob/master/blog/20220112-simplex-chat-v1-released.md).

@@ -27,9 +27,9 @@ SimpleXMQ is implemented in Haskell - it benefits from robust software transacti

 ### SMP server

-[SMP server](https://github.com/simplex-chat/simplexmq/blob/master/apps/smp-server/Main.hs) can be run on any Linux distribution without any dependencies, including low power/low memory devices.
+[SMP server](https://github.com/simplex-chat/simplexmq/blob/master/apps/smp-server/Main.hs) can be run on any Linux distribution, including low power/low memory devices. OpenSSL library is required for initialization.

-To initialize the server use `smp-server init` command - it will generate keys and certificates for TLS transport. The fingerprint of offline certificate is used as part of the server address to protect client/server connection against man-in-the-middle attacks: `smp://<fingerprint>@<hostname>[:5223]`.
+To initialize the server use `smp-server init -n <fqdn>` (or `smp-server init --ip <ip>` for IP based address) command - it will generate keys and certificates for TLS transport. The fingerprint of offline certificate is used as part of the server address to protect client/server connection against man-in-the-middle attacks: `smp://<fingerprint>@<hostname>[:5223]`.

 SMP server uses in-memory persistence with an optional append-only log of created queues that allows to re-start the server without losing the connections. This log is compacted on every server restart, permanently removing suspended and removed queues.

@@ -87,6 +87,26 @@ You can either run your own SMP server locally or deploy using [Linode StackScri

 It's the easiest to try SMP agent via a prototype [simplex-chat](https://github.com/simplex-chat/simplex-chat) terminal UI.

+## Deploy SMP server on Linux
+
+You can run your SMP server as a Linux process, optionally using a service manager for booting and restarts.
+
+- For Ubuntu you can download a binary from [the latest release](https://github.com/simplex-chat/simplexmq/releases).
+
+  If you're using other Linux distribution and the binary is incompatible with it, you can build from source using [Haskell stack](https://docs.haskellstack.org/en/stable/README/):
+
+  ```shell
+  curl -sSL https://get.haskellstack.org/ | sh
+  ...
+  stack install
+  ```
+
+- Initialize SMP server with `smp-server init [-l] -n <fqdn>` or `smp-server init [-l] --ip <ip>` - depending on how you initialize it, either FQDN or IP will be used for server's address.
+
+- Run `smp-server start` to start SMP server, or you can configure a service manager to run it as a service.
+
+See [this section](#smp-server) for more information. Run `smp-server -h` and `smp-server init -h` for explanation of commands and options.
+
 [<img alt="Linode" src="https://raw.githubusercontent.com/simplex-chat/simplexmq/master/img/linode.svg" align="right" width="200">](https://cloud.linode.com/stackscripts/748014)

 ## Deploy SMP server on Linode
@@ -11,7 +11,7 @@ import Simplex.Messaging.Agent.Server (runSMPAgent)
 import Simplex.Messaging.Transport (TLS, Transport (..))

 cfg :: AgentConfig
-cfg = defaultAgentConfig {smpServers = L.fromList ["smp://bU0K-bRg24xWW__lS0umO1Zdw_SXqpJNtm1_RrPLViE=@localhost:5223"]}
+cfg = defaultAgentConfig {initialSMPServers = L.fromList ["smp://bU0K-bRg24xWW__lS0umO1Zdw_SXqpJNtm1_RrPLViE=@localhost:5223"]}

 logCfg :: LogConfig
 logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
@@ -3,6 +3,7 @@
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE TypeApplications #-}

@@ -131,8 +132,7 @@ cliCommandP =
                <*> strOption
                  ( long "ip"
                      <> help
-                        "Server IP address used as Subject Alternative Name for TLS online certificate, \
-                        \also used as Common Name if FQDN is not supplied"
+                        "Server IP address, used as Common Name for TLS online certificate if FQDN is not supplied"
                      <> value "127.0.0.1"
                      <> showDefault
                      <> metavar "IP"
@@ -140,7 +140,7 @@ cliCommandP =
                <*> (optional . strOption)
                  ( long "fqdn"
                      <> short 'n'
-                      <> help "Server FQDN used as Common Name and Subject Alternative Name for TLS online certificate"
+                      <> help "Server FQDN used as Common Name for TLS online certificate"
                      <> showDefault
                      <> metavar "FQDN"
                  )
@@ -277,14 +277,17 @@ runServer IniOptions {enableStoreLog, port, enableWebsockets} = do
      ServerConfig
        { transports = (port, transport @TLS) : [("80", transport @WS) | enableWebsockets],
          tbqSize = 16,
-          serverTbqSize = 128,
-          msgQueueQuota = 256,
+          serverTbqSize = 64,
+          msgQueueQuota = 128,
          queueIdBytes = 24,
          msgIdBytes = 24, -- must be at least 24 bytes, it is used as 192-bit nonce for XSalsa20
          caCertificateFile = caCrtFile,
          privateKeyFile = serverKeyFile,
          certificateFile = serverCrtFile,
-          storeLog
+          storeLog,
+          allowNewQueues = True,
+          messageTTL = Just $ 7 * 86400, -- 7 days
+          expireMessagesInterval = Just 21600_000000 -- microseconds, 6 hours
        }

    openStoreLog :: IO (Maybe (StoreLog 'ReadMode))
@@ -2,5 +2,5 @@ packages: .

 source-repository-package
    type: git
-    location: git://github.com/simplex-chat/aeson.git
+    location: https://github.com/simplex-chat/aeson.git
    tag: 3eb66f9a68f103b5f1489382aad89f5712a64db7
@@ -1,7 +1,7 @@
 name: simplexmq
-version: 1.0.2
+version: 1.1.0
 synopsis: SimpleXMQ message broker
-description:  |
+description: |
  This package includes <./docs/Simplex-Messaging-Server.html server>,
  <./docs/Simplex-Messaging-Client.html client> and
  <./docs/Simplex-Messaging-Agent.html agent> for SMP protocols:
@@ -45,7 +45,7 @@ dependencies:
  - iso8601-time == 0.1.*
  - memory == 0.15.*
  - mtl == 2.2.*
-  - network == 3.1.*
+  - network == 3.1.2.*
  - network-transport == 0.5.*
  - QuickCheck == 2.14.*
  - random >= 1.1 && < 1.3
@@ -64,6 +64,17 @@ dependencies:
  - x509-store == 1.6.*
  - x509-validation == 1.6.*

+flags:
+  swift:
+    description: Enable swift JSON format
+    manual: True
+    default: False
+
+when:
+  - condition: flag(swift)
+    ghc-options:
+      - -DswiftJSON
+
 library:
  source-dirs: src

@@ -0,0 +1,21 @@
+# SMP confirmation timeout recovery
+
+## Problem
+
+When sending an SMP confirmation a network timeout can lead to the following race condition:
+- server receives the confirmation while the joining party fails to receive the server's response;
+- joining party deletes the connection together with credentials sent in the confirmation for securing the queue;
+- initiating party will receive the confirmation from the server and secure the queue;
+- on subsequent attempt to join via the same invitation link initiating party will generate new credentials and fail authorization.
+
+This renders the joining party permanently unable to join via that invitation link and complete the connection.
+
+## Solution
+
+A possible solution is to keep and try to reuse same credentials on subsequent attempts:
+- joining party has to remember invitation link when saving the connection;
+- if SMP confirmation fails due to network timeout joining party doesn't delete the connection and keeps the credentials;
+- when joining, joining party checks whether such invitation link was already used for a connection, if yes:
+  - joining party tries to send SMP confirmation with the same credentials;
+  - if this SMP confirmation fails with authorization error (for example it can happen due to race condition explained above) joining party tries to send HELLO message;
+  - if HELLO message fails with authorization error (it can happen if connection was deleted or secured with different credentials), the recovery is no longer possible and connection can be deleted.
@@ -16,3 +16,8 @@ brew install hashicorp/tap/packer
 cd ./scripts/smp-server-digitalocean-droplet
 DIGITALOCEAN_TOKEN=$YOUR_TOKEN packer build -on-error=ask -color=false ./marketplace-image.json
 ```
+
+**TODO** (see Linode script)
+
+- Increase file descriptors limit
+- Configure Restart for systemd service
@@ -44,6 +44,12 @@ ufw allow ssh
 ufw allow https
 ufw allow 5223

+# Increase file descriptors limit
+echo 'fs.file-max = 1000000' >> /etc/sysctl.conf
+echo 'fs.inode-max = 1000000' >> /etc/sysctl.conf
+echo 'root soft nofile unlimited' >> /etc/security/limits.conf
+echo 'root hard nofile unlimited' >> /etc/security/limits.conf
+
 # Download latest release
 bin_dir="/opt/simplex/bin"
 binary="$bin_dir/smp-server"
@@ -151,6 +157,10 @@ Description=SMP server
 [Service]
 Type=simple
 ExecStart=/bin/sh -c "exec $binary start >> /var/opt/simplex/smp-server.log 2>&1"
+Restart=always
+RestartSec=10
+LimitNOFILE=1000000
+LimitNOFILESoft=1000000

 [Install]
 WantedBy=multi-user.target
@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack

 name:           simplexmq
-version:        1.0.2
+version:        1.1.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                <./docs/Simplex-Messaging-Client.html client> and
@@ -27,6 +27,11 @@ extra-source-files:
    README.md
    CHANGELOG.md

+flag swift
+  description: Enable swift JSON format
+  manual: True
+  default: False
+
 library
  exposed-modules:
      Simplex.Messaging.Agent
@@ -40,6 +45,7 @@ library
      Simplex.Messaging.Agent.Store.SQLite
      Simplex.Messaging.Agent.Store.SQLite.Migrations
      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys
      Simplex.Messaging.Client
      Simplex.Messaging.Crypto
      Simplex.Messaging.Crypto.Ratchet
@@ -54,8 +60,10 @@ library
      Simplex.Messaging.Server.QueueStore
      Simplex.Messaging.Server.QueueStore.STM
      Simplex.Messaging.Server.StoreLog
+      Simplex.Messaging.TMap
      Simplex.Messaging.Transport
      Simplex.Messaging.Transport.Client
+      Simplex.Messaging.Transport.KeepAlive
      Simplex.Messaging.Transport.Server
      Simplex.Messaging.Transport.WebSockets
      Simplex.Messaging.Util
@@ -90,7 +98,7 @@ library
    , iso8601-time ==0.1.*
    , memory ==0.15.*
    , mtl ==2.2.*
-    , network ==3.1.*
+    , network ==3.1.2.*
    , network-transport ==0.5.*
    , random >=1.1 && <1.3
    , simple-logger ==0.1.*
@@ -107,6 +115,8 @@ library
    , x509 ==1.7.*
    , x509-store ==1.6.*
    , x509-validation ==1.6.*
+  if flag(swift)
+    ghc-options: -DswiftJSON
  default-language: Haskell2010

 executable smp-agent
@@ -141,7 +151,7 @@ executable smp-agent
    , iso8601-time ==0.1.*
    , memory ==0.15.*
    , mtl ==2.2.*
-    , network ==3.1.*
+    , network ==3.1.2.*
    , network-transport ==0.5.*
    , random >=1.1 && <1.3
    , simple-logger ==0.1.*
@@ -159,6 +169,8 @@ executable smp-agent
    , x509 ==1.7.*
    , x509-store ==1.6.*
    , x509-validation ==1.6.*
+  if flag(swift)
+    ghc-options: -DswiftJSON
  default-language: Haskell2010

 executable smp-server
@@ -194,7 +206,7 @@ executable smp-server
    , iso8601-time ==0.1.*
    , memory ==0.15.*
    , mtl ==2.2.*
-    , network ==3.1.*
+    , network ==3.1.2.*
    , network-transport ==0.5.*
    , optparse-applicative >=0.15 && <0.17
    , process ==1.6.*
@@ -214,6 +226,8 @@ executable smp-server
    , x509 ==1.7.*
    , x509-store ==1.6.*
    , x509-validation ==1.6.*
+  if flag(swift)
+    ghc-options: -DswiftJSON
  default-language: Haskell2010

 test-suite smp-server-test
@@ -263,7 +277,7 @@ test-suite smp-server-test
    , iso8601-time ==0.1.*
    , memory ==0.15.*
    , mtl ==2.2.*
-    , network ==3.1.*
+    , network ==3.1.2.*
    , network-transport ==0.5.*
    , random >=1.1 && <1.3
    , simple-logger ==0.1.*
@@ -282,4 +296,6 @@ test-suite smp-server-test
    , x509 ==1.7.*
    , x509-store ==1.6.*
    , x509-validation ==1.6.*
+  if flag(swift)
+    ghc-options: -DswiftJSON
  default-language: Haskell2010
@@ -47,6 +47,7 @@ module Simplex.Messaging.Agent
    ackMessage,
    suspendConnection,
    deleteConnection,
+    setSMPServers,
    logConnection,
  )
 where
@@ -75,13 +76,14 @@ import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore)
-import Simplex.Messaging.Client (SMPServerTransmission)
+import Simplex.Messaging.Client (SMPClient (..), SMPServerTransmission)
 import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Crypto.Ratchet as CR
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Parsers (parse)
 import Simplex.Messaging.Protocol (MsgBody)
 import qualified Simplex.Messaging.Protocol as SMP
+import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Util (bshow, liftError, tryError, unlessM)
 import Simplex.Messaging.Version
 import System.Random (randomR)
@@ -143,6 +145,10 @@ suspendConnection c = withAgentEnv c . suspendConnection' c
 deleteConnection :: AgentErrorMonad m => AgentClient -> ConnId -> m ()
 deleteConnection c = withAgentEnv c . deleteConnection' c

+-- | Change servers to be used for creating new queues
+setSMPServers :: AgentErrorMonad m => AgentClient -> NonEmpty SMPServer -> m ()
+setSMPServers c = withAgentEnv c . setSMPServers' c
+
 withAgentEnv :: AgentClient -> ReaderT Env m a -> m a
 withAgentEnv c = (`runReaderT` agentEnv c)

@@ -209,7 +215,7 @@ processCommand c (connId, cmd) = case cmd of

 newConn :: AgentMonad m => AgentClient -> ConnId -> SConnectionMode c -> m (ConnId, ConnectionRequestUri c)
 newConn c connId cMode = do
-  srv <- getSMPServer
+  srv <- getSMPServer c
  (rq, qUri) <- newRcvQueue c srv
  g <- asks idsDrg
  let cData = ConnData {connId}
@@ -234,16 +240,21 @@ joinConn c connId (CRInvitationUri (ConnReqUriData _ agentVRange (qUri :| _)) e2
      (pk1, pk2, e2eSndParams) <- liftIO . CR.generateE2EParams $ version e2eRcvParams
      (_, rcDHRs) <- liftIO C.generateKeyPair'
      let rc = CR.initSndRatchet rcDHRr rcDHRs $ CR.x3dhSnd pk1 pk2 e2eRcvParams
-      (sq, smpConf) <- newSndQueue qInfo cInfo
+      sq <- newSndQueue qInfo
      g <- asks idsDrg
      let cData = ConnData {connId}
      connId' <- withStore $ \st -> do
        connId' <- createSndConn st g cData sq
        createRatchet st connId' rc
        pure connId'
-      confirmQueue c connId' sq smpConf $ Just e2eSndParams
-      void $ enqueueMessage c connId' sq HELLO
-      pure connId'
+      tryError (confirmQueue c connId' sq cInfo $ Just e2eSndParams) >>= \case
+        Right _ -> do
+          void $ enqueueMessage c connId' sq HELLO
+          pure connId'
+        Left e -> do
+          -- TODO recovery for failure on network timeout, see rfcs/2022-04-20-smp-conf-timeout-recovery.md
+          withStore (`deleteConn` connId')
+          throwError e
    _ -> throwError $ AGENT A_VERSION
 joinConn c connId (CRContactUri (ConnReqUriData _ agentVRange (qUri :| _))) cInfo =
  case ( qUri `compatibleVersion` SMP.smpClientVRange,
@@ -258,7 +269,7 @@ joinConn c connId (CRContactUri (ConnReqUriData _ agentVRange (qUri :| _))) cInf

 createReplyQueue :: AgentMonad m => AgentClient -> ConnId -> SndQueue -> m ()
 createReplyQueue c connId sq = do
-  srv <- getSMPServer
+  srv <- getSMPServer c
  (rq, qUri) <- newRcvQueue c srv
  -- TODO reply queue version should be the same as send queue, ignoring it in v1
  let qInfo = toVersionT qUri SMP.smpClientVersion
@@ -305,16 +316,6 @@ subscribeConnection' c connId =
    SomeConn _ (DuplexConnection _ rq sq) -> do
      resumeMsgDelivery c connId sq
      subscribeQueue c rq connId
-      case status (sq :: SndQueue) of
-        Confirmed -> do
-          -- TODO if there is no confirmation saved, just update the status without securing the queue
-          AcceptedConfirmation {senderConf = SMPConfirmation {senderKey}} <-
-            withStore (`getAcceptedConfirmation` connId)
-          secureQueue c rq senderKey
-          withStore $ \st -> setRcvQueueStatus st rq Secured
-        Secured -> pure ()
-        Active -> pure ()
-        _ -> throwError $ INTERNAL "unexpected queue status"
    SomeConn _ (SndConnection _ sq) -> do
      resumeMsgDelivery c connId sq
      case status (sq :: SndQueue) of
@@ -347,12 +348,12 @@ enqueueMessage c connId sq aMessage = do
      internalTs <- liftIO getCurrentTime
      (internalId, internalSndId, prevMsgHash) <- withStore (`updateSndIds` connId)
      let privHeader = APrivHeader (unSndId internalSndId) prevMsgHash
-          agentMessage = smpEncode $ AgentMessage privHeader aMessage
-          internalHash = C.sha256Hash agentMessage
-
-      encAgentMessage <- agentRatchetEncrypt connId agentMessage e2eEncUserMsgLength
+          agentMsg = AgentMessage privHeader aMessage
+          agentMsgStr = smpEncode agentMsg
+          internalHash = C.sha256Hash agentMsgStr
+      encAgentMessage <- agentRatchetEncrypt connId agentMsgStr e2eEncUserMsgLength
      let msgBody = smpEncode $ AgentMsgEnvelope {agentVersion = smpAgentVersion, encAgentMessage}
-          msgType = aMessageType aMessage
+          msgType = agentMessageType agentMsg
          msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, internalHash, prevMsgHash}
      withStore $ \st -> createSndMsg st connId msgData
      pure internalId
@@ -362,18 +363,13 @@ 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
+      >>= \a -> atomically (TM.insert qKey a $ smpQueueMsgDeliveries c)
  unlessM connQueued $
    withStore (`getPendingMsgs` connId)
      >>= queuePendingMsgs c connId sq
  where
-    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))
+    queueDelivering qKey = atomically $ isJust <$> TM.lookup qKey (smpQueueMsgDeliveries c)
+    connQueued = atomically $ isJust <$> TM.lookupInsert connId True (connMsgsQueued c)

 queuePendingMsgs :: AgentMonad m => AgentClient -> ConnId -> SndQueue -> [InternalId] -> m ()
 queuePendingMsgs c connId sq msgIds = atomically $ do
@@ -383,11 +379,11 @@ queuePendingMsgs c connId sq msgIds = atomically $ do
 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)
+  maybe (newMsgQueue qKey) pure =<< TM.lookup qKey (smpQueueMsgQueues c)
  where
    newMsgQueue qKey = do
      mq <- newTQueue
-      modifyTVar (smpQueueMsgQueues c) $ M.insert qKey mq
+      TM.insert qKey mq $ smpQueueMsgQueues c
      pure mq

 runSmpQueueMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnId -> SndQueue -> m ()
@@ -402,30 +398,38 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} connId sq = do
        notify $ MERR mId (INTERNAL $ show e)
      Right (rq_, (msgType, msgBody, internalTs)) ->
        withRetryInterval ri $ \loop ->
-          tryError (sendAgentMessage c sq msgBody) >>= \case
+          tryError (send msgType c sq msgBody) >>= \case
            Left e -> do
+              let err = if msgType == AM_CONN_INFO then ERR e else MERR mId e
              case e of
-                SMP SMP.QUOTA -> loop
+                SMP SMP.QUOTA -> case msgType of
+                  AM_CONN_INFO -> connError msgId NOT_AVAILABLE
+                  _ -> loop
                SMP SMP.AUTH -> case msgType of
-                  HELLO_ -> do
+                  AM_CONN_INFO -> connError msgId NOT_AVAILABLE
+                  AM_HELLO_ -> do
                    helloTimeout <- asks $ helloTimeout . config
                    currentTime <- liftIO getCurrentTime
                    if diffUTCTime currentTime internalTs > helloTimeout
                      then case rq_ of
                        -- party initiating connection
-                        Just _ -> notifyDel msgId . ERR $ CONN NOT_AVAILABLE
+                        Just _ -> connError msgId NOT_AVAILABLE
                        -- party joining connection
-                        _ -> notifyDel msgId . ERR $ CONN NOT_ACCEPTED
+                        _ -> connError msgId NOT_ACCEPTED
                      else loop
-                  REPLY_ -> notifyDel msgId $ ERR e
-                  A_MSG_ -> notifyDel msgId $ MERR mId e
-                SMP (SMP.CMD _) -> notifyDel msgId $ MERR mId e
-                SMP SMP.LARGE_MSG -> notifyDel msgId $ MERR mId e
-                SMP {} -> notify (MERR mId e) >> loop
+                  AM_REPLY_ -> notifyDel msgId $ ERR e
+                  AM_A_MSG_ -> notifyDel msgId $ MERR mId e
+                SMP (SMP.CMD _) -> notifyDel msgId err
+                SMP SMP.LARGE_MSG -> notifyDel msgId err
+                SMP {} -> notify err >> loop
                _ -> loop
            Right () -> do
              case msgType of
-                HELLO_ -> do
+                AM_CONN_INFO -> do
+                  withStore $ \st -> setSndQueueStatus st sq Confirmed
+                  when (isJust rq_) $ withStore (`removeConfirmations` connId)
+                  void $ enqueueMessage c connId sq HELLO
+                AM_HELLO_ -> do
                  withStore $ \st -> setSndQueueStatus st sq Active
                  case rq_ of
                    -- party initiating connection
@@ -434,16 +438,20 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} connId sq = do
                      notify CON
                    -- party joining connection
                    _ -> createReplyQueue c connId sq
-                A_MSG_ -> notify $ SENT mId
+                AM_A_MSG_ -> notify $ SENT mId
                _ -> pure ()
              delMsg msgId
  where
+    send = \case
+      AM_CONN_INFO -> sendConfirmation
+      _ -> sendAgentMessage
    delMsg :: InternalId -> m ()
    delMsg msgId = withStore $ \st -> deleteMsg st connId msgId
    notify :: ACommand 'Agent -> m ()
    notify cmd = atomically $ writeTBQueue subQ ("", connId, cmd)
    notifyDel :: InternalId -> ACommand 'Agent -> m ()
    notifyDel msgId cmd = notify cmd >> delMsg msgId
+    connError msgId = notifyDel msgId . ERR . CONN

 ackMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m ()
 ackMessage' c connId msgId = do
@@ -479,12 +487,18 @@ deleteConnection' c connId =
    delete :: RcvQueue -> m ()
    delete rq = do
      deleteQueue c rq
-      removeSubscription c connId
+      atomically $ removeSubscription c connId
      withStore (`deleteConn` connId)

-getSMPServer :: AgentMonad m => m SMPServer
-getSMPServer =
-  asks (smpServers . config) >>= \case
+-- | Change servers to be used for creating new queues, in Reader monad
+setSMPServers' :: forall m. AgentMonad m => AgentClient -> NonEmpty SMPServer -> m ()
+setSMPServers' c servers = do
+  atomically $ writeTVar (smpServers c) servers
+
+getSMPServer :: AgentMonad m => AgentClient -> m SMPServer
+getSMPServer c = do
+  smpServers <- readTVarIO $ smpServers c
+  case smpServers of
    srv :| [] -> pure srv
    servers -> do
      gen <- asks randomServer
@@ -499,7 +513,7 @@ subscriber c@AgentClient {msgQ} = forever $ do
    Right _ -> return ()

 processSMPTransmission :: forall m. AgentMonad m => AgentClient -> SMPServerTransmission -> m ()
-processSMPTransmission c@AgentClient {subQ} (srv, rId, cmd) = do
+processSMPTransmission c@AgentClient {smpClients, subQ} (srv, sessId, rId, cmd) = do
  withStore (\st -> getRcvConn st srv rId) >>= \case
    SomeConn SCDuplex (DuplexConnection cData rq _) -> processSMP SCDuplex cData rq
    SomeConn SCRcv (RcvConnection cData rq) -> processSMP SCRcv cData rq
@@ -529,8 +543,9 @@ processSMPTransmission c@AgentClient {subQ} (srv, rId, cmd) = do
                (SMP.PHEmpty, AgentMsgEnvelope _ encAgentMsg) -> do
                  agentMsgBody <- agentRatchetDecrypt connId encAgentMsg
                  parseMessage agentMsgBody >>= \case
-                    AgentMessage APrivHeader {sndMsgId, prevMsgHash} aMessage -> do
-                      (msgId, msgMeta) <- agentClientMsg prevMsgHash sndMsgId (srvMsgId, systemToUTCTime srvTs) agentMsgBody aMessage
+                    agentMsg@(AgentMessage APrivHeader {sndMsgId, prevMsgHash} aMessage) -> do
+                      let msgType = agentMessageType agentMsg
+                      (msgId, msgMeta) <- agentClientMsg prevMsgHash sndMsgId (srvMsgId, systemToUTCTime srvTs) agentMsgBody msgType
                      case aMessage of
                        HELLO -> helloMsg >> ack >> withStore (\st -> deleteMsg st connId msgId)
                        REPLY cReq -> replyMsg cReq >> ack >> withStore (\st -> deleteMsg st connId msgId)
@@ -539,10 +554,19 @@ processSMPTransmission c@AgentClient {subQ} (srv, rId, cmd) = do
                    _ -> prohibited >> ack
                _ -> prohibited >> ack
            _ -> prohibited >> ack
-        SMP.END -> do
-          removeSubscription c connId
-          logServer "<--" c srv rId "END"
-          notify END
+        SMP.END ->
+          atomically (TM.lookup srv smpClients >>= fmap join . mapM tryReadTMVar >>= processEND)
+            >>= logServer "<--" c srv rId
+          where
+            processEND = \case
+              Just (Right clnt)
+                | sessId == sessionId clnt -> do
+                  removeSubscription c connId
+                  writeTBQueue subQ ("", connId, END)
+                  pure "END"
+                | otherwise -> ignored
+              _ -> ignored
+            ignored = pure "END from disconnected client - ignored"
        _ -> do
          logServer "<--" c srv rId $ "unexpected: " <> bshow cmd
          notify . ERR $ BROKER UNEXPECTED
@@ -620,15 +644,13 @@ processSMPTransmission c@AgentClient {subQ} (srv, rId, cmd) = do
              case qInfo `proveCompatible` SMP.smpClientVRange of
                Nothing -> notify . ERR $ AGENT A_VERSION
                Just qInfo' -> do
-                  (sq, smpConf) <- newSndQueue qInfo' ownConnInfo
+                  sq <- newSndQueue qInfo'
                  withStore $ \st -> upgradeRcvConnToDuplex st connId sq
-                  confirmQueue c connId sq smpConf Nothing
-                  withStore (`removeConfirmations` connId)
-                  void $ enqueueMessage c connId sq HELLO
+                  enqueueConfirmation c connId sq ownConnInfo Nothing
            _ -> prohibited

-        agentClientMsg :: PrevRcvMsgHash -> ExternalSndId -> (BrokerId, BrokerTs) -> MsgBody -> AMessage -> m (InternalId, MsgMeta)
-        agentClientMsg externalPrevSndHash sndMsgId broker msgBody aMessage = do
+        agentClientMsg :: PrevRcvMsgHash -> ExternalSndId -> (BrokerId, BrokerTs) -> MsgBody -> AgentMessageType -> m (InternalId, MsgMeta)
+        agentClientMsg externalPrevSndHash sndMsgId broker msgBody msgType = do
          logServer "<--" c srv rId "MSG <MSG>"
          let internalHash = C.sha256Hash msgBody
          internalTs <- liftIO getCurrentTime
@@ -636,7 +658,6 @@ processSMPTransmission c@AgentClient {subQ} (srv, rId, cmd) = do
          let integrity = checkMsgIntegrity prevExtSndId sndMsgId prevRcvMsgHash externalPrevSndHash
              recipient = (unId internalId, internalTs)
              msgMeta = MsgMeta {integrity, recipient, broker, sndMsgId}
-              msgType = aMessageType aMessage
              rcvMsg = RcvMsgData {msgMeta, msgType, msgBody, internalRcvId, internalHash, externalPrevSndHash}
          withStore $ \st -> createRcvMsg st connId rcvMsg
          pure (internalId, msgMeta)
@@ -661,8 +682,9 @@ processSMPTransmission c@AgentClient {subQ} (srv, rId, cmd) = do
          | internalPrevMsgHash /= receivedPrevMsgHash = MsgError MsgBadHash
          | otherwise = MsgError MsgDuplicate -- this case is not possible

-confirmQueue :: forall m. AgentMonad m => AgentClient -> ConnId -> SndQueue -> SMPConfirmation -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()
-confirmQueue c connId sq SMPConfirmation {senderKey, e2ePubKey, connInfo} e2eEncryption = do
+confirmQueue :: forall m. AgentMonad m => AgentClient -> ConnId -> SndQueue -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()
+confirmQueue c connId sq connInfo e2eEncryption = do
+  _ <- withStore (`updateSndIds` connId)
  msg <- mkConfirmation
  sendConfirmation c sq msg
  withStore $ \st -> setSndQueueStatus st sq Confirmed
@@ -670,9 +692,27 @@ confirmQueue c connId sq SMPConfirmation {senderKey, e2ePubKey, connInfo} e2eEnc
    mkConfirmation :: m MsgBody
    mkConfirmation = do
      encConnInfo <- agentRatchetEncrypt connId (smpEncode $ AgentConnInfo connInfo) e2eEncConnInfoLength
-      let agentEnvelope = AgentConfirmation {agentVersion = smpAgentVersion, e2eEncryption, encConnInfo}
-      agentCbEncrypt sq (Just e2ePubKey) . smpEncode $
-        SMP.ClientMessage (SMP.PHConfirmation senderKey) $ smpEncode agentEnvelope
+      pure . smpEncode $ AgentConfirmation {agentVersion = smpAgentVersion, e2eEncryption, encConnInfo}
+
+enqueueConfirmation :: forall m. AgentMonad m => AgentClient -> ConnId -> SndQueue -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()
+enqueueConfirmation c connId sq connInfo e2eEncryption = do
+  resumeMsgDelivery c connId sq
+  msgId <- storeConfirmation
+  queuePendingMsgs c connId sq [msgId]
+  where
+    storeConfirmation :: m InternalId
+    storeConfirmation = do
+      internalTs <- liftIO getCurrentTime
+      (internalId, internalSndId, prevMsgHash) <- withStore (`updateSndIds` connId)
+      let agentMsg = AgentConnInfo connInfo
+          agentMsgStr = smpEncode agentMsg
+          internalHash = C.sha256Hash agentMsgStr
+      encConnInfo <- agentRatchetEncrypt connId agentMsgStr e2eEncConnInfoLength
+      let msgBody = smpEncode $ AgentConfirmation {agentVersion = smpAgentVersion, e2eEncryption, encConnInfo}
+          msgType = agentMessageType agentMsg
+          msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, internalHash, prevMsgHash}
+      withStore $ \st -> createSndMsg st connId msgData
+      pure internalId

 -- encoded AgentMessage -> encoded EncAgentMessage
 agentRatchetEncrypt :: AgentMonad m => ConnId -> ByteString -> Int -> m ByteString
@@ -694,27 +734,27 @@ agentRatchetDecrypt connId encAgentMsg = do
 notifyConnected :: AgentMonad m => AgentClient -> ConnId -> m ()
 notifyConnected c connId = atomically $ writeTBQueue (subQ c) ("", connId, CON)

-newSndQueue :: (MonadUnliftIO m, MonadReader Env m) => Compatible SMPQueueInfo -> ConnInfo -> m (SndQueue, SMPConfirmation)
-newSndQueue qInfo cInfo =
+newSndQueue :: (MonadUnliftIO m, MonadReader Env m) => Compatible SMPQueueInfo -> m SndQueue
+newSndQueue qInfo =
  asks (cmdSignAlg . config) >>= \case
-    C.SignAlg a -> newSndQueue_ a qInfo cInfo
+    C.SignAlg a -> newSndQueue_ a qInfo

 newSndQueue_ ::
  (C.SignatureAlgorithm a, C.AlgorithmI a, MonadUnliftIO m) =>
  C.SAlgorithm a ->
  Compatible SMPQueueInfo ->
-  ConnInfo ->
-  m (SndQueue, SMPConfirmation)
-newSndQueue_ a (Compatible (SMPQueueInfo _clientVersion smpServer senderId rcvE2ePubDhKey)) cInfo = do
+  m SndQueue
+newSndQueue_ a (Compatible (SMPQueueInfo _clientVersion smpServer senderId rcvE2ePubDhKey)) = do
  -- this function assumes clientVersion is compatible - it was tested before
-  (senderKey, sndPrivateKey) <- liftIO $ C.generateSignatureKeyPair a
+  (sndPublicKey, sndPrivateKey) <- liftIO $ C.generateSignatureKeyPair a
  (e2ePubKey, e2ePrivKey) <- liftIO C.generateKeyPair'
-  let sndQueue =
-        SndQueue
-          { server = smpServer,
-            sndId = senderId,
-            sndPrivateKey,
-            e2eDhSecret = C.dh' rcvE2ePubDhKey e2ePrivKey,
-            status = New
-          }
-  pure (sndQueue, SMPConfirmation senderKey e2ePubKey cInfo)
+  pure
+    SndQueue
+      { server = smpServer,
+        sndId = senderId,
+        sndPublicKey = Just sndPublicKey,
+        sndPrivateKey,
+        e2eDhSecret = C.dh' rcvE2ePubDhKey e2ePrivKey,
+        e2ePubKey = Just e2ePubKey,
+        status = New
+      }
@@ -37,7 +37,7 @@ module Simplex.Messaging.Agent.Client
 where

 import Control.Concurrent (forkIO)
-import Control.Concurrent.Async (Async, async, uninterruptibleCancel)
+import Control.Concurrent.Async (Async, uninterruptibleCancel)
 import Control.Concurrent.STM (stateTVar)
 import Control.Logger.Simple
 import Control.Monad.Except
@@ -47,11 +47,10 @@ import Data.Bifunctor (first)
 import Data.ByteString.Base64
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
+import Data.List.NonEmpty (NonEmpty)
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Data.Maybe (isNothing)
-import Data.Set (Set)
-import qualified Data.Set as S
 import Data.Text.Encoding
 import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Protocol
@@ -62,8 +61,12 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Protocol (QueueId, QueueIdsKeys (..), SndPublicVerifyKey)
 import qualified Simplex.Messaging.Protocol as SMP
-import Simplex.Messaging.Util (bshow, liftEitherError, liftError, liftIOEither, tryError)
+import Simplex.Messaging.TMap (TMap)
+import qualified Simplex.Messaging.TMap as TM
+import Simplex.Messaging.Util (bshow, liftEitherError, liftError, tryError, whenM)
 import Simplex.Messaging.Version
+import System.Timeout (timeout)
+import UnliftIO (async, forConcurrently_)
 import UnliftIO.Exception (Exception, IOException)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
@@ -74,13 +77,16 @@ data AgentClient = AgentClient
  { rcvQ :: TBQueue (ATransmission 'Client),
    subQ :: TBQueue (ATransmission 'Agent),
    msgQ :: TBQueue SMPServerTransmission,
-    smpClients :: TVar (Map SMPServer SMPClientVar),
-    subscrSrvrs :: TVar (Map SMPServer (Map ConnId RcvQueue)),
-    subscrConns :: TVar (Map ConnId SMPServer),
-    connMsgsQueued :: TVar (Map ConnId Bool),
-    smpQueueMsgQueues :: TVar (Map (ConnId, SMPServer, SMP.SenderId) (TQueue InternalId)),
-    smpQueueMsgDeliveries :: TVar (Map (ConnId, SMPServer, SMP.SenderId) (Async ())),
+    smpServers :: TVar (NonEmpty SMPServer),
+    smpClients :: TMap SMPServer SMPClientVar,
+    subscrSrvrs :: TMap SMPServer (TMap ConnId RcvQueue),
+    pendingSubscrSrvrs :: TMap SMPServer (TMap ConnId RcvQueue),
+    subscrConns :: TMap ConnId SMPServer,
+    connMsgsQueued :: TMap ConnId Bool,
+    smpQueueMsgQueues :: TMap (ConnId, SMPServer, SMP.SenderId) (TQueue InternalId),
+    smpQueueMsgDeliveries :: TMap (ConnId, SMPServer, SMP.SenderId) (Async ()),
    reconnections :: TVar [Async ()],
+    asyncClients :: TVar [Async ()],
    clientId :: Int,
    agentEnv :: Env,
    smpSubscriber :: Async (),
@@ -93,16 +99,19 @@ newAgentClient agentEnv = do
  rcvQ <- newTBQueue qSize
  subQ <- newTBQueue qSize
  msgQ <- newTBQueue qSize
-  smpClients <- newTVar M.empty
-  subscrSrvrs <- newTVar M.empty
-  subscrConns <- newTVar M.empty
-  connMsgsQueued <- newTVar M.empty
-  smpQueueMsgQueues <- newTVar M.empty
-  smpQueueMsgDeliveries <- newTVar M.empty
+  smpServers <- newTVar $ initialSMPServers (config agentEnv)
+  smpClients <- TM.empty
+  subscrSrvrs <- TM.empty
+  pendingSubscrSrvrs <- TM.empty
+  subscrConns <- TM.empty
+  connMsgsQueued <- TM.empty
+  smpQueueMsgQueues <- TM.empty
+  smpQueueMsgDeliveries <- TM.empty
  reconnections <- newTVar []
+  asyncClients <- newTVar []
  clientId <- stateTVar (clientCounter agentEnv) $ \i -> (i + 1, i + 1)
  lock <- newTMVar ()
-  return AgentClient {rcvQ, subQ, msgQ, smpClients, subscrSrvrs, subscrConns, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, reconnections, clientId, agentEnv, smpSubscriber = undefined, lock}
+  return AgentClient {rcvQ, subQ, msgQ, smpServers, smpClients, subscrSrvrs, pendingSubscrSrvrs, subscrConns, connMsgsQueued, smpQueueMsgQueues, smpQueueMsgDeliveries, reconnections, asyncClients, clientId, agentEnv, smpSubscriber = undefined, lock}

 -- | Agent monad with MonadReader Env and MonadError AgentErrorType
 type AgentMonad m = (MonadUnliftIO m, MonadReader Env m, MonadError AgentErrorType m)
@@ -124,29 +133,48 @@ getSMPServerClient c@AgentClient {smpClients, msgQ} srv =
  atomically getClientVar >>= either newSMPClient waitForSMPClient
  where
    getClientVar :: STM (Either SMPClientVar SMPClientVar)
-    getClientVar = maybe (Left <$> newClientVar) (pure . Right) . M.lookup srv =<< readTVar smpClients
+    getClientVar = maybe (Left <$> newClientVar) (pure . Right) =<< TM.lookup srv smpClients

    newClientVar :: STM SMPClientVar
    newClientVar = do
      smpVar <- newEmptyTMVar
-      modifyTVar smpClients $ M.insert srv smpVar
+      TM.insert srv smpVar smpClients
      pure smpVar

    waitForSMPClient :: TMVar (Either AgentErrorType SMPClient) -> m SMPClient
-    waitForSMPClient = liftIOEither . atomically . readTMVar
+    waitForSMPClient smpVar = do
+      SMPClientConfig {tcpTimeout} <- asks $ smpCfg . config
+      smpClient_ <- liftIO $ tcpTimeout `timeout` atomically (readTMVar smpVar)
+      liftEither $ case smpClient_ of
+        Just (Right smpClient) -> Right smpClient
+        Just (Left e) -> Left e
+        Nothing -> Left $ BROKER TIMEOUT

    newSMPClient :: TMVar (Either AgentErrorType SMPClient) -> m SMPClient
-    newSMPClient smpVar =
-      tryError connectClient >>= \r -> case r of
-        Right smp -> do
-          logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv
-          atomically $ putTMVar smpVar r
-          pure smp
-        Left e -> do
-          atomically $ do
-            putTMVar smpVar r
-            modifyTVar smpClients $ M.delete srv
-          throwError e
+    newSMPClient smpVar = tryConnectClient pure tryConnectAsync
+      where
+        tryConnectClient :: (SMPClient -> m a) -> m () -> m a
+        tryConnectClient successAction retryAction =
+          tryError connectClient >>= \r -> case r of
+            Right smp -> do
+              logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv
+              atomically $ putTMVar smpVar r
+              successAction smp
+            Left e -> do
+              if e == BROKER NETWORK || e == BROKER TIMEOUT
+                then retryAction
+                else atomically $ do
+                  putTMVar smpVar (Left e)
+                  TM.delete srv smpClients
+              throwError e
+        tryConnectAsync :: m ()
+        tryConnectAsync = do
+          a <- async connectAsync
+          atomically $ modifyTVar' (asyncClients c) (a :)
+        connectAsync :: m ()
+        connectAsync = do
+          ri <- asks $ reconnectInterval . config
+          withRetryInterval ri $ \loop -> void $ tryConnectClient (const reconnectClient) loop

    connectClient :: m SMPClient
    connectClient = do
@@ -160,44 +188,62 @@ getSMPServerClient c@AgentClient {smpClients, msgQ} srv =

    clientDisconnected :: UnliftIO m -> IO ()
    clientDisconnected u = do
-      removeClientSubs >>= (`forM_` serverDown u)
+      removeClientAndSubs >>= (`forM_` serverDown u)
      logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv

-    removeClientSubs :: IO (Maybe (Map ConnId RcvQueue))
-    removeClientSubs = atomically $ do
-      modifyTVar smpClients $ M.delete srv
-      cs <- M.lookup srv <$> readTVar (subscrSrvrs c)
-      modifyTVar (subscrSrvrs c) $ M.delete srv
-      modifyTVar (subscrConns c) $ maybe id (deleteKeys . M.keysSet) cs
-      return cs
+    removeClientAndSubs :: IO (Maybe (Map ConnId RcvQueue))
+    removeClientAndSubs = atomically $ do
+      TM.delete srv smpClients
+      cVar_ <- TM.lookupDelete srv $ subscrSrvrs c
+      forM cVar_ $ \cVar -> do
+        cs <- readTVar cVar
+        modifyTVar' (subscrConns c) (`M.withoutKeys` M.keysSet cs)
+        addPendingSubs cVar cs
+        pure cs
      where
-        deleteKeys :: Ord k => Set k -> Map k a -> Map k a
-        deleteKeys ks m = S.foldr' M.delete m ks
+        addPendingSubs cVar cs = do
+          let ps = pendingSubscrSrvrs c
+          TM.lookup srv ps >>= \case
+            Just v -> TM.union cs v
+            _ -> TM.insert srv cVar ps

    serverDown :: UnliftIO m -> Map ConnId RcvQueue -> IO ()
    serverDown u cs = unless (M.null cs) $ do
      mapM_ (notifySub DOWN) $ M.keysSet cs
-      a <- async . unliftIO u $ tryReconnectClient cs
-      atomically $ modifyTVar (reconnections c) (a :)
+      unliftIO u reconnectServer

-    tryReconnectClient :: Map ConnId RcvQueue -> m ()
-    tryReconnectClient cs = do
+    reconnectServer :: m ()
+    reconnectServer = do
+      a <- async tryReconnectClient
+      atomically $ modifyTVar' (reconnections c) (a :)
+
+    tryReconnectClient :: m ()
+    tryReconnectClient = do
      ri <- asks $ reconnectInterval . config
      withRetryInterval ri $ \loop ->
-        reconnectClient cs `catchError` const loop
+        reconnectClient `catchError` const loop

-    reconnectClient :: Map ConnId RcvQueue -> m ()
-    reconnectClient cs = do
+    reconnectClient :: m ()
+    reconnectClient =
      withAgentLock c . withSMP c srv $ \smp -> do
-        subs <- readTVarIO $ subscrConns c
-        forM_ (M.toList cs) $ \(connId, rq@RcvQueue {rcvPrivateKey, rcvId}) ->
-          when (isNothing $ M.lookup connId subs) $ do
-            subscribeSMPQueue smp rcvPrivateKey rcvId
-              `catchError` \case
-                SMPServerError e -> liftIO $ notifySub (ERR $ SMP e) connId
-                e -> throwError e
-            addSubscription c rq connId
-            liftIO $ notifySub UP connId
+        cs <- atomically $ mapM readTVar =<< TM.lookup srv (pendingSubscrSrvrs c)
+        forConcurrently_ (maybe [] M.toList cs) $ \sub@(connId, _) ->
+          whenM (atomically $ isNothing <$> TM.lookup connId (subscrConns c)) $
+            subscribe_ smp sub `catchError` handleError connId
+      where
+        subscribe_ :: SMPClient -> (ConnId, RcvQueue) -> ExceptT SMPClientError IO ()
+        subscribe_ smp (connId, rq@RcvQueue {rcvPrivateKey, rcvId}) = do
+          subscribeSMPQueue smp rcvPrivateKey rcvId
+          addSubscription c rq connId
+          liftIO $ notifySub UP connId
+
+        handleError :: ConnId -> SMPClientError -> ExceptT SMPClientError IO ()
+        handleError connId = \case
+          e@SMPResponseTimeout -> throwError e
+          e@SMPNetworkError -> throwError e
+          e -> do
+            liftIO $ notifySub (ERR $ smpClientError e) connId
+            atomically $ removePendingSubscription c srv connId

    notifySub :: ACommand 'Agent -> ConnId -> IO ()
    notifySub cmd connId = atomically $ writeTBQueue (subQ c) ("", connId, cmd)
@@ -206,6 +252,7 @@ closeAgentClient :: MonadUnliftIO m => AgentClient -> m ()
 closeAgentClient c = liftIO $ do
  closeSMPServerClients c
  cancelActions $ reconnections c
+  cancelActions $ asyncClients c
  cancelActions $ smpQueueMsgDeliveries c

 closeSMPServerClients :: AgentClient -> IO ()
@@ -294,31 +341,41 @@ newRcvQueue_ a c srv = do

 subscribeQueue :: AgentMonad m => AgentClient -> RcvQueue -> ConnId -> m ()
 subscribeQueue c rq@RcvQueue {server, rcvPrivateKey, rcvId} connId = do
-  withLogSMP c server rcvId "SUB" $ \smp ->
-    subscribeSMPQueue smp rcvPrivateKey rcvId
-  addSubscription c rq connId
+  atomically $ addPendingSubscription c rq connId
+  withLogSMP c server rcvId "SUB" $ \smp -> do
+    liftIO (runExceptT $ subscribeSMPQueue smp rcvPrivateKey rcvId) >>= \case
+      Left e -> do
+        atomically . when (e /= SMPNetworkError && e /= SMPResponseTimeout) $
+          removePendingSubscription c server connId
+        throwError e
+      Right _ -> addSubscription c rq connId

 addSubscription :: MonadUnliftIO m => AgentClient -> RcvQueue -> ConnId -> m ()
 addSubscription c rq@RcvQueue {server} connId = atomically $ do
-  modifyTVar (subscrConns c) $ M.insert connId server
-  modifyTVar (subscrSrvrs c) $ M.alter (Just . addSub) server
-  where
-    addSub :: Maybe (Map ConnId RcvQueue) -> Map ConnId RcvQueue
-    addSub (Just cs) = M.insert connId rq cs
-    addSub _ = M.singleton connId rq
+  TM.insert connId server $ subscrConns c
+  addSubs_ (subscrSrvrs c) rq connId
+  removePendingSubscription c server connId

-removeSubscription :: AgentMonad m => AgentClient -> ConnId -> m ()
-removeSubscription AgentClient {subscrConns, subscrSrvrs} connId = atomically $ do
-  cs <- readTVar subscrConns
-  writeTVar subscrConns $ M.delete connId cs
-  mapM_
-    (modifyTVar subscrSrvrs . M.alter (>>= delSub))
-    (M.lookup connId cs)
-  where
-    delSub :: Map ConnId RcvQueue -> Maybe (Map ConnId RcvQueue)
-    delSub cs =
-      let cs' = M.delete connId cs
-       in if M.null cs' then Nothing else Just cs'
+addPendingSubscription :: AgentClient -> RcvQueue -> ConnId -> STM ()
+addPendingSubscription = addSubs_ . pendingSubscrSrvrs
+
+addSubs_ :: TMap SMPServer (TMap ConnId RcvQueue) -> RcvQueue -> ConnId -> STM ()
+addSubs_ ss rq@RcvQueue {server} connId =
+  TM.lookup server ss >>= \case
+    Just m -> TM.insert connId rq m
+    _ -> TM.singleton connId rq >>= \m -> TM.insert server m ss
+
+removeSubscription :: AgentClient -> ConnId -> STM ()
+removeSubscription c@AgentClient {subscrConns} connId = do
+  server_ <- TM.lookupDelete connId subscrConns
+  mapM_ (\server -> removeSubs_ (subscrSrvrs c) server connId) server_
+
+removePendingSubscription :: AgentClient -> SMPServer -> ConnId -> STM ()
+removePendingSubscription = removeSubs_ . pendingSubscrSrvrs
+
+removeSubs_ :: TMap SMPServer (TMap ConnId RcvQueue) -> SMPServer -> ConnId -> STM ()
+removeSubs_ ss server connId =
+  TM.lookup server ss >>= mapM_ (TM.delete connId)

 logServer :: AgentMonad m => ByteString -> AgentClient -> SMPServer -> QueueId -> ByteString -> m ()
 logServer dir AgentClient {clientId} srv qId cmdStr =
@@ -331,11 +388,13 @@ showServer SMPServer {host, port} =
 logSecret :: ByteString -> ByteString
 logSecret bs = encode $ B.take 3 bs

-- TODO maybe package E2ERatchetParams into SMPConfirmation
 sendConfirmation :: forall m. AgentMonad m => AgentClient -> SndQueue -> ByteString -> m ()
-sendConfirmation c SndQueue {server, sndId} encConfirmation =
-  withLogSMP_ c server sndId "SEND <CONF>" $ \smp ->
-    liftSMP $ sendSMPMessage smp Nothing sndId encConfirmation
+sendConfirmation c sq@SndQueue {server, sndId, sndPublicKey = Just sndPublicKey, e2ePubKey = e2ePubKey@Just {}} agentConfirmation =
+  withLogSMP_ c server sndId "SEND <CONF>" $ \smp -> do
+    let clientMsg = SMP.ClientMessage (SMP.PHConfirmation sndPublicKey) agentConfirmation
+    msg <- agentCbEncrypt sq e2ePubKey $ smpEncode clientMsg
+    liftSMP $ sendSMPMessage smp Nothing sndId msg
+sendConfirmation _ _ _ = throwError $ INTERNAL "sendConfirmation called without snd_queue public key(s) in the database"

 sendInvitation :: forall m. AgentMonad m => AgentClient -> Compatible SMPQueueInfo -> ConnectionRequestUri 'CMInvitation -> ConnInfo -> m ()
 sendInvitation c (Compatible SMPQueueInfo {smpServer, senderId, dhPublicKey}) connReq connInfo =
@@ -370,7 +429,6 @@ deleteQueue c RcvQueue {server, rcvId, rcvPrivateKey} =
  withLogSMP c server rcvId "DEL" $ \smp ->
    deleteSMPQueue smp rcvPrivateKey rcvId

-- TODO this is just wrong
 sendAgentMessage :: forall m. AgentMonad m => AgentClient -> SndQueue -> ByteString -> m ()
 sendAgentMessage c sq@SndQueue {server, sndId, sndPrivateKey} agentMsg =
  withLogSMP_ c server sndId "SEND <MSG>" $ \smp -> do
@@ -29,12 +29,13 @@ import UnliftIO.STM

 data AgentConfig = AgentConfig
  { tcpPort :: ServiceName,
-    smpServers :: NonEmpty SMPServer,
+    initialSMPServers :: NonEmpty SMPServer,
    cmdSignAlg :: C.SignAlg,
    connIdBytes :: Int,
    tbqSize :: Natural,
    dbFile :: FilePath,
    dbPoolSize :: Int,
+    yesToMigrations :: Bool,
    smpCfg :: SMPClientConfig,
    reconnectInterval :: RetryInterval,
    helloTimeout :: NominalDiffTime,
@@ -47,12 +48,13 @@ defaultAgentConfig :: AgentConfig
 defaultAgentConfig =
  AgentConfig
    { tcpPort = "5224",
-      smpServers = undefined, -- TODO move it elsewhere?
+      initialSMPServers = undefined, -- TODO move it elsewhere?
      cmdSignAlg = C.SignAlg C.SEd448,
      connIdBytes = 12,
-      tbqSize = 16,
+      tbqSize = 64,
      dbFile = "smp-agent.db",
      dbPoolSize = 4,
+      yesToMigrations = False,
      smpCfg = smpDefaultConfig,
      reconnectInterval =
        RetryInterval
@@ -60,7 +62,7 @@ defaultAgentConfig =
            increaseAfter = 10 * second,
            maxInterval = 10 * second
          },
-      helloTimeout = 7 * nominalDay,
+      helloTimeout = 2 * nominalDay,
      -- CA certificate private key is not needed for initialization
      -- ! we do not generate these
      caCertificateFile = "/etc/opt/simplex-agent/ca.crt",
@@ -79,9 +81,9 @@ data Env = Env
  }

 newSMPAgentEnv :: (MonadUnliftIO m, MonadRandom m) => AgentConfig -> m Env
-newSMPAgentEnv cfg = do
+newSMPAgentEnv config@AgentConfig {dbFile, dbPoolSize, yesToMigrations} = do
  idsDrg <- newTVarIO =<< drgNew
-  store <- liftIO $ createSQLiteStore (dbFile cfg) (dbPoolSize cfg) Migrations.app
+  store <- liftIO $ createSQLiteStore dbFile dbPoolSize Migrations.app yesToMigrations
  clientCounter <- newTVarIO 0
  randomServer <- newTVarIO =<< liftIO newStdGen
-  return Env {config = cfg, store, idsDrg, clientCounter, randomServer}
+  return Env {config, store, idsDrg, clientCounter, randomServer}
@@ -46,9 +46,9 @@ module Simplex.Messaging.Agent.Protocol
    SMPConfirmation (..),
    AgentMsgEnvelope (..),
    AgentMessage (..),
+    AgentMessageType (..),
    APrivHeader (..),
    AMessage (..),
-    AMsgType (..),
    SMPServer (..),
    SrvLoc (..),
    SMPQueueUri (..),
@@ -89,7 +89,7 @@ module Simplex.Messaging.Agent.Protocol
    connModeT,
    serializeQueueStatus,
    queueStatusT,
-    aMessageType,
+    agentMessageType,

    -- * TCP transport functions
    tPut,
@@ -343,6 +343,31 @@ instance Encoding AgentMessage where
      'M' -> AgentMessage <$> smpP <*> smpP
      _ -> fail "bad AgentMessage"

+data AgentMessageType = AM_CONN_INFO | AM_HELLO_ | AM_REPLY_ | AM_A_MSG_
+  deriving (Eq, Show)
+
+instance Encoding AgentMessageType where
+  smpEncode = \case
+    AM_CONN_INFO -> "C"
+    AM_HELLO_ -> "H"
+    AM_REPLY_ -> "R"
+    AM_A_MSG_ -> "M"
+  smpP =
+    A.anyChar >>= \case
+      'C' -> pure AM_CONN_INFO
+      'H' -> pure AM_HELLO_
+      'R' -> pure AM_REPLY_
+      'M' -> pure AM_A_MSG_
+      _ -> fail "bad AgentMessageType"
+
+agentMessageType :: AgentMessage -> AgentMessageType
+agentMessageType = \case
+  AgentConnInfo _ -> AM_CONN_INFO
+  AgentMessage _ aMsg -> case aMsg of
+    HELLO -> AM_HELLO_
+    REPLY _ -> AM_REPLY_
+    A_MSG _ -> AM_A_MSG_
+
 data APrivHeader = APrivHeader
  { -- | sequential ID assigned by the sending agent
    sndMsgId :: AgentMsgId,
@@ -371,12 +396,6 @@ instance Encoding AMsgType where
      'M' -> pure A_MSG_
      _ -> fail "bad AMsgType"

-aMessageType :: AMessage -> AMsgType
-aMessageType = \case
-  HELLO -> HELLO_
-  REPLY _ -> REPLY_
-  A_MSG _ -> A_MSG_
-
 -- | Messages sent between SMP agents once SMP queue is secured.
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/agent-protocol.md#messages-between-smp-agents
@@ -705,7 +724,7 @@ data ConnectionErrorType
    SIMPLEX
  | -- | connection not accepted on join HELLO after timeout
    NOT_ACCEPTED
-  | -- | connection not available on reply HELLO after timeout
+  | -- | connection not available on reply confirmation/HELLO after timeout
    NOT_AVAILABLE
  deriving (Eq, Generic, Read, Show, Exception)

@@ -62,7 +62,7 @@ class Monad m => MonadAgentStore s m where
  createRcvMsg :: s -> ConnId -> RcvMsgData -> m ()
  updateSndIds :: s -> ConnId -> m (InternalId, InternalSndId, PrevSndMsgHash)
  createSndMsg :: s -> ConnId -> SndMsgData -> m ()
-  getPendingMsgData :: s -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody, InternalTs))
+  getPendingMsgData :: s -> ConnId -> InternalId -> m (Maybe RcvQueue, (AgentMessageType, MsgBody, InternalTs))
  getPendingMsgs :: s -> ConnId -> m [InternalId]
  checkRcvMsg :: s -> ConnId -> InternalId -> m ()
  deleteMsg :: s -> ConnId -> InternalId -> m ()
@@ -102,8 +102,11 @@ data SndQueue = SndQueue
  { server :: SMPServer,
    -- | sender queue ID
    sndId :: SMP.SenderId,
-    -- | key used by the sender to sign transmissions
+    -- | key pair used by the sender to sign transmissions
+    sndPublicKey :: Maybe C.APublicVerifyKey,
    sndPrivateKey :: SndPrivateSignKey,
+    -- | DH public key used to negotiate per-queue e2e encryption
+    e2ePubKey :: Maybe C.PublicKeyX25519,
    -- | shared DH secret agreed for simple per-queue e2e encryption
    e2eDhSecret :: C.DhSecretX25519,
    -- | queue status
@@ -221,7 +224,7 @@ type PrevSndMsgHash = MsgHash

 data RcvMsgData = RcvMsgData
  { msgMeta :: MsgMeta,
-    msgType :: AMsgType,
+    msgType :: AgentMessageType,
    msgBody :: MsgBody,
    internalRcvId :: InternalRcvId,
    internalHash :: MsgHash,
@@ -232,7 +235,7 @@ data SndMsgData = SndMsgData
  { internalId :: InternalId,
    internalSndId :: InternalSndId,
    internalTs :: InternalTs,
-    msgType :: AMsgType,
+    msgType :: AgentMessageType,
    msgBody :: MsgBody,
    internalHash :: MsgHash,
    prevMsgHash :: MsgHash
@@ -76,13 +76,13 @@ data SQLiteStore = SQLiteStore
    dbNew :: Bool
  }

-createSQLiteStore :: FilePath -> Int -> [Migration] -> IO SQLiteStore
-createSQLiteStore dbFilePath poolSize migrations = do
+createSQLiteStore :: FilePath -> Int -> [Migration] -> Bool -> IO SQLiteStore
+createSQLiteStore dbFilePath poolSize migrations yesToMigrations = do
  let dbDir = takeDirectory dbFilePath
  createDirectoryIfMissing False dbDir
  st <- connectSQLiteStore dbFilePath poolSize
  checkThreadsafe st
-  migrateSchema st migrations
+  migrateSchema st migrations yesToMigrations
  pure st

 checkThreadsafe :: SQLiteStore -> IO ()
@@ -94,15 +94,16 @@ checkThreadsafe st = withConnection st $ \db -> do
    Nothing -> putStrLn "Warning: SQLite THREADSAFE compile option not found"
    _ -> return ()

-migrateSchema :: SQLiteStore -> [Migration] -> IO ()
-migrateSchema st migrations = withConnection st $ \db -> do
+migrateSchema :: SQLiteStore -> [Migration] -> Bool -> IO ()
+migrateSchema st migrations yesToMigrations = withConnection st $ \db -> do
  Migrations.initialize db
  Migrations.get db migrations >>= \case
    Left e -> confirmOrExit $ "Database error: " <> e
    Right [] -> pure ()
    Right ms -> do
      unless (dbNew st) $ do
-        confirmOrExit "The app has a newer version than the database - it will be backed up and upgraded."
+        unless yesToMigrations $
+          confirmOrExit "The app has a newer version than the database - it will be backed up and upgraded."
        let f = dbFilePath st
        copyFile f (f <> ".bak")
      Migrations.run db ms
@@ -126,9 +127,25 @@ connectSQLiteStore dbFilePath poolSize = do
 connectDB :: FilePath -> IO DB.Connection
 connectDB path = do
  dbConn <- DB.open path
-  DB.execute_ dbConn "PRAGMA foreign_keys = ON; PRAGMA journal_mode = WAL;"
+  DB.execute_ dbConn "PRAGMA foreign_keys = ON;"
+  -- DB.execute_ dbConn "PRAGMA trusted_schema = OFF;"
+  DB.execute_ dbConn "PRAGMA secure_delete = ON;"
+  DB.execute_ dbConn "PRAGMA auto_vacuum = FULL;"
+  -- _printPragmas dbConn path
  pure dbConn

+_printPragmas :: DB.Connection -> FilePath -> IO ()
+_printPragmas db path = do
+  foreign_keys <- DB.query_ db "PRAGMA foreign_keys;" :: IO [[Int]]
+  print $ path <> " foreign_keys: " <> show foreign_keys
+  -- when run via sqlite-simple query for trusted_schema seems to return empty list
+  trusted_schema <- DB.query_ db "PRAGMA trusted_schema;" :: IO [[Int]]
+  print $ path <> " trusted_schema: " <> show trusted_schema
+  secure_delete <- DB.query_ db "PRAGMA secure_delete;" :: IO [[Int]]
+  print $ path <> " secure_delete: " <> show secure_delete
+  auto_vacuum <- DB.query_ db "PRAGMA auto_vacuum;" :: IO [[Int]]
+  print $ path <> " auto_vacuum: " <> show auto_vacuum
+
 checkConstraint :: StoreError -> IO (Either StoreError a) -> IO (Either StoreError a)
 checkConstraint err action = action `E.catch` (pure . Left . handleSQLError err)

@@ -440,7 +457,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto
      insertSndMsgDetails_ db connId sndMsgData
      updateHashSnd_ db connId sndMsgData

-  getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody, InternalTs))
+  getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m (Maybe RcvQueue, (AgentMessageType, MsgBody, InternalTs))
  getPendingMsgData st connId msgId =
    liftIOEither . withTransaction st $ \db -> runExceptT $ do
      rq_ <- liftIO $ getRcvQueueByConnId_ db connId
@@ -565,9 +582,9 @@ instance ToField InternalId where toField (InternalId x) = toField x

 instance FromField InternalId where fromField x = InternalId <$> fromField x

-instance ToField AMsgType where toField = toField . smpEncode
+instance ToField AgentMessageType where toField = toField . smpEncode

-instance FromField AMsgType where fromField = blobFieldParser smpP
+instance FromField AgentMessageType where fromField = blobFieldParser smpP

 instance ToField MsgIntegrity where toField = toField . strEncode

@@ -655,46 +672,25 @@ upsertServer_ dbConn SMPServer {host, port, keyHash} = do

 insertRcvQueue_ :: DB.Connection -> ConnId -> RcvQueue -> IO ()
 insertRcvQueue_ dbConn connId RcvQueue {..} = do
-  DB.executeNamed
+  DB.execute
    dbConn
    [sql|
      INSERT INTO rcv_queues
-        ( host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, e2e_dh_secret, snd_id, status)
-      VALUES
-        (:host,:port,:rcv_id,:conn_id,:rcv_private_key,:rcv_dh_secret,:e2e_priv_key,:e2e_dh_secret,:snd_id,:status);
+        ( host, port, rcv_id, conn_id, rcv_private_key, rcv_dh_secret, e2e_priv_key, e2e_dh_secret, snd_id, status) VALUES (?,?,?,?,?,?,?,?,?,?);
    |]
-    [ ":host" := host server,
-      ":port" := port server,
-      ":rcv_id" := rcvId,
-      ":conn_id" := connId,
-      ":rcv_private_key" := rcvPrivateKey,
-      ":rcv_dh_secret" := rcvDhSecret,
-      ":e2e_priv_key" := e2ePrivKey,
-      ":e2e_dh_secret" := e2eDhSecret,
-      ":snd_id" := sndId,
-      ":status" := status
-    ]
+    (host server, port server, rcvId, connId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status)

 -- * createSndConn helpers

 insertSndQueue_ :: DB.Connection -> ConnId -> SndQueue -> IO ()
 insertSndQueue_ dbConn connId SndQueue {..} = do
-  DB.executeNamed
+  DB.execute
    dbConn
    [sql|
      INSERT INTO snd_queues
-        ( host, port, snd_id, conn_id, snd_private_key, e2e_dh_secret, status)
-      VALUES
-        (:host,:port,:snd_id,:conn_id,:snd_private_key,:e2e_dh_secret,:status);
+        (host, port, snd_id, conn_id, snd_public_key, snd_private_key, e2e_pub_key, e2e_dh_secret, status) VALUES (?,?,?,?,?, ?,?, ?,?);
    |]
-    [ ":host" := host server,
-      ":port" := port server,
-      ":snd_id" := sndId,
-      ":conn_id" := connId,
-      ":snd_private_key" := sndPrivateKey,
-      ":e2e_dh_secret" := e2eDhSecret,
-      ":status" := status
-    ]
+    (host server, port server, sndId, connId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status)

 -- * getConn helpers

@@ -745,16 +741,16 @@ getSndQueueByConnId_ dbConn connId =
    <$> DB.query
      dbConn
      [sql|
-        SELECT s.key_hash, q.host, q.port, q.snd_id, q.snd_private_key, q.e2e_dh_secret, q.status
+        SELECT s.key_hash, q.host, q.port, q.snd_id, q.snd_public_key, q.snd_private_key, q.e2e_pub_key, q.e2e_dh_secret, q.status
        FROM snd_queues q
        INNER JOIN servers s ON q.host = s.host AND q.port = s.port
        WHERE q.conn_id = ?;
      |]
      (Only connId)
  where
-    sndQueue [(keyHash, host, port, sndId, sndPrivateKey, e2eDhSecret, status)] =
+    sndQueue [(keyHash, host, port, sndId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status)] =
      let server = SMPServer host port keyHash
-       in Just SndQueue {server, sndId, sndPrivateKey, e2eDhSecret, status}
+       in Just SndQueue {server, sndId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status}
    sndQueue _ = Nothing

 -- * updateRcvIds helpers
@@ -25,13 +25,15 @@ import qualified Database.SQLite.Simple as DB
 import Database.SQLite.Simple.QQ (sql)
 import qualified Database.SQLite3 as SQLite3
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys

 data Migration = Migration {name :: String, up :: Text}
  deriving (Show)

 schemaMigrations :: [(String, Query)]
 schemaMigrations =
-  [ ("20220101_initial", m20220101_initial)
+  [ ("20220101_initial", m20220101_initial),
+    ("20220301_snd_queue_keys", m20220301_snd_queue_keys)
  ]

 -- | The list of migrations in ascending order by date
@@ -0,0 +1,13 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20220301_snd_queue_keys :: Query
+m20220301_snd_queue_keys =
+  [sql|
+ALTER TABLE snd_queues ADD COLUMN snd_public_key BLOB;
+ALTER TABLE snd_queues ADD COLUMN e2e_pub_key BLOB;
+|]
@@ -23,7 +23,7 @@
 -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md
 module Simplex.Messaging.Client
  ( -- * Connect (disconnect) client to (from) SMP server
-    SMPClient,
+    SMPClient (sessionId),
    getSMPClient,
    closeSMPClient,

@@ -56,15 +56,16 @@ import Control.Monad.Trans.Class
 import Control.Monad.Trans.Except
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import Data.Map.Strict (Map)
-import qualified Data.Map.Strict as M
 import Data.Maybe (fromMaybe)
 import Network.Socket (ServiceName)
 import Numeric.Natural
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol
-import Simplex.Messaging.Transport (ATransport (..), THandle (..), TLS, TProxy, Transport (..), TransportError, clientHandshake)
+import Simplex.Messaging.TMap (TMap)
+import qualified Simplex.Messaging.TMap as TM
+import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client (runTransportClient)
+import Simplex.Messaging.Transport.KeepAlive
 import Simplex.Messaging.Transport.WebSockets (WS)
 import Simplex.Messaging.Util (bshow, liftError, raceAny_)
 import System.Timeout (timeout)
@@ -78,18 +79,18 @@ import System.Timeout (timeout)
 data SMPClient = SMPClient
  { action :: Async (),
    connected :: TVar Bool,
-    sessionId :: ByteString,
+    sessionId :: SessionId,
    smpServer :: SMPServer,
    tcpTimeout :: Int,
    clientCorrId :: TVar Natural,
-    sentCommands :: TVar (Map CorrId Request),
+    sentCommands :: TMap CorrId Request,
    sndQ :: TBQueue SentRawTransmission,
    rcvQ :: TBQueue (SignedTransmission BrokerMsg),
    msgQ :: TBQueue SMPServerTransmission
  }

 -- | Type synonym for transmission from some SPM server queue.
-type SMPServerTransmission = (SMPServer, RecipientId, BrokerMsg)
+type SMPServerTransmission = (SMPServer, SessionId, RecipientId, BrokerMsg)

 -- | SMP client configuration.
 data SMPClientConfig = SMPClientConfig
@@ -99,6 +100,8 @@ data SMPClientConfig = SMPClientConfig
    defaultTransport :: (ServiceName, ATransport),
    -- | timeout of TCP commands (microseconds)
    tcpTimeout :: Int,
+    -- | TCP keep-alive options, Nothing to skip enabling keep-alive
+    tcpKeepAlive :: Maybe KeepAliveOpts,
    -- | period for SMP ping commands (microseconds)
    smpPing :: Int
  }
@@ -107,10 +110,11 @@ data SMPClientConfig = SMPClientConfig
 smpDefaultConfig :: SMPClientConfig
 smpDefaultConfig =
  SMPClientConfig
-    { qSize = 16,
+    { qSize = 64,
      defaultTransport = ("5223", transport @TLS),
-      tcpTimeout = 4_000_000,
-      smpPing = 30_000_000
+      tcpTimeout = 5_000_000,
+      tcpKeepAlive = Just defaultKeepAliveOpts,
+      smpPing = 600_000_000 -- 10min
    }

 data Request = Request
@@ -126,14 +130,14 @@ type Response = Either SMPClientError BrokerMsg
 -- A single queue can be used for multiple 'SMPClient' instances,
 -- as 'SMPServerTransmission' includes server information.
 getSMPClient :: SMPServer -> SMPClientConfig -> TBQueue SMPServerTransmission -> IO () -> IO (Either SMPClientError SMPClient)
-getSMPClient smpServer cfg@SMPClientConfig {qSize, tcpTimeout, smpPing} msgQ disconnected =
+getSMPClient smpServer cfg@SMPClientConfig {qSize, tcpTimeout, tcpKeepAlive, smpPing} msgQ disconnected =
  atomically mkSMPClient >>= runClient useTransport
  where
    mkSMPClient :: STM SMPClient
    mkSMPClient = do
      connected <- newTVar False
      clientCorrId <- newTVar 0
-      sentCommands <- newTVar M.empty
+      sentCommands <- TM.empty
      sndQ <- newTBQueue qSize
      rcvQ <- newTBQueue qSize
      return
@@ -155,7 +159,7 @@ getSMPClient smpServer cfg@SMPClientConfig {qSize, tcpTimeout, smpPing} msgQ dis
      thVar <- newEmptyTMVarIO
      action <-
        async $
-          runTransportClient (host smpServer) port' (keyHash smpServer) (client t c thVar)
+          runTransportClient (host smpServer) port' (keyHash smpServer) tcpKeepAlive (client t c thVar)
            `finally` atomically (putTMVar thVar $ Left SMPNetworkError)
      th_ <- tcpTimeout `timeout` atomically (takeTMVar thVar)
      pure $ case th_ of
@@ -193,16 +197,15 @@ getSMPClient smpServer cfg@SMPClientConfig {qSize, tcpTimeout, smpPing} msgQ dis
      runExceptT $ sendSMPCommand c Nothing "" PING

    process :: SMPClient -> IO ()
-    process SMPClient {rcvQ, sentCommands} = forever $ do
+    process SMPClient {sessionId, rcvQ, sentCommands} = forever $ do
      (_, _, (corrId, qId, respOrErr)) <- atomically $ readTBQueue rcvQ
      if B.null $ bs corrId
        then sendMsg qId respOrErr
        else do
-          cs <- readTVarIO sentCommands
-          case M.lookup corrId cs of
+          atomically (TM.lookup corrId sentCommands) >>= \case
            Nothing -> sendMsg qId respOrErr
            Just Request {queueId, responseVar} -> atomically $ do
-              modifyTVar sentCommands $ M.delete corrId
+              TM.delete corrId sentCommands
              putTMVar responseVar $
                if queueId == qId
                  then case respOrErr of
@@ -210,12 +213,12 @@ getSMPClient smpServer cfg@SMPClientConfig {qSize, tcpTimeout, smpPing} msgQ dis
                    Right (ERR e) -> Left $ SMPServerError e
                    Right r -> Right r
                  else Left SMPUnexpectedResponse
-
-    sendMsg :: QueueId -> Either ErrorType BrokerMsg -> IO ()
-    sendMsg qId = \case
-      Right cmd -> atomically $ writeTBQueue msgQ (smpServer, qId, cmd)
-      -- TODO send everything else to errQ and log in agent
-      _ -> return ()
+      where
+        sendMsg :: QueueId -> Either ErrorType BrokerMsg -> IO ()
+        sendMsg qId = \case
+          Right cmd -> atomically $ writeTBQueue msgQ (smpServer, sessionId, qId, cmd)
+          -- TODO send everything else to errQ and log in agent
+          _ -> return ()

 -- | Disconnects SMP client from the server and terminates client threads.
 closeSMPClient :: SMPClient -> IO ()
@@ -265,11 +268,11 @@ createSMPQueue c rpKey rKey dhKey =
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#subscribe-to-queue
 subscribeSMPQueue :: SMPClient -> RcvPrivateSignKey -> RecipientId -> ExceptT SMPClientError IO ()
-subscribeSMPQueue c@SMPClient {smpServer, msgQ} rpKey rId =
+subscribeSMPQueue c@SMPClient {smpServer, sessionId, msgQ} rpKey rId =
  sendSMPCommand c (Just rpKey) rId SUB >>= \case
    OK -> return ()
    cmd@MSG {} ->
-      lift . atomically $ writeTBQueue msgQ (smpServer, rId, cmd)
+      lift . atomically $ writeTBQueue msgQ (smpServer, sessionId, rId, cmd)
    _ -> throwE SMPUnexpectedResponse

 -- | Subscribe to the SMP queue notifications.
@@ -306,11 +309,11 @@ sendSMPMessage c spKey sId msg =
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#acknowledge-message-delivery
 ackSMPMessage :: SMPClient -> RcvPrivateSignKey -> QueueId -> ExceptT SMPClientError IO ()
-ackSMPMessage c@SMPClient {smpServer, msgQ} rpKey rId =
+ackSMPMessage c@SMPClient {smpServer, sessionId, msgQ} rpKey rId =
  sendSMPCommand c (Just rpKey) rId ACK >>= \case
    OK -> return ()
    cmd@MSG {} ->
-      lift . atomically $ writeTBQueue msgQ (smpServer, rId, cmd)
+      lift . atomically $ writeTBQueue msgQ (smpServer, sessionId, rId, cmd)
    _ -> throwE SMPUnexpectedResponse

 -- | Irreversibly suspend SMP queue.
@@ -364,6 +367,6 @@ sendSMPCommand SMPClient {sndQ, sentCommands, clientCorrId, sessionId, tcpTimeou
    send :: CorrId -> SentRawTransmission -> STM (TMVar Response)
    send corrId t = do
      r <- newEmptyTMVar
-      modifyTVar sentCommands . M.insert corrId $ Request qId r
+      TM.insert corrId (Request qId r) sentCommands
      writeTBQueue sndQ t
      return r
@@ -1,3 +1,4 @@
+{-# LANGUAGE CPP #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE OverloadedStrings #-}

@@ -97,13 +98,18 @@ enumJSON tagModifier =
    }

 sumTypeJSON :: (String -> String) -> J.Options
+#if defined(darwin_HOST_OS) && defined(swiftJSON)
 sumTypeJSON = singleFieldJSON
+#else
+sumTypeJSON = taggedObjectJSON
+#endif

 taggedObjectJSON :: (String -> String) -> J.Options
 taggedObjectJSON tagModifier =
  J.defaultOptions
    { J.sumEncoding = J.TaggedObject "type" "data",
      J.constructorTagModifier = tagModifier,
+      J.allNullaryToStringTag = False,
      J.nullaryToObject = True,
      J.omitNothingFields = True
    }
@@ -113,6 +119,7 @@ singleFieldJSON tagModifier =
  J.defaultOptions
    { J.sumEncoding = J.ObjectWithSingleField,
      J.constructorTagModifier = tagModifier,
+      J.allNullaryToStringTag = False,
      J.nullaryToObject = True,
      J.omitNothingFields = True
    }
@@ -108,7 +108,7 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers
-import Simplex.Messaging.Transport (THandle (..), Transport, TransportError (..), tGetBlock, tPutBlock)
+import Simplex.Messaging.Transport (SessionId, THandle (..), Transport, TransportError (..), tGetBlock, tPutBlock)
 import Simplex.Messaging.Util (bshow, (<$?>))
 import Simplex.Messaging.Version
 import Test.QuickCheck (Arbitrary (..))
@@ -172,14 +172,14 @@ type Signed = ByteString
 data RawTransmission = RawTransmission
  { signature :: ByteString,
    signed :: ByteString,
-    sessId :: ByteString,
+    sessId :: SessionId,
    corrId :: ByteString,
    queueId :: ByteString,
    command :: ByteString
  }

 -- | unparsed sent SMP transmission with signature, without session ID.
-type SignedRawTransmission = (Maybe C.ASignature, ByteString, ByteString, ByteString)
+type SignedRawTransmission = (Maybe C.ASignature, SessionId, ByteString, ByteString)

 -- | unparsed sent SMP transmission with signature.
 type SentRawTransmission = (Maybe C.ASignature, ByteString)
@@ -399,6 +399,10 @@ instance StrEncoding SMPServer where
    SrvLoc host port <- strP
    pure SMPServer {host, port, keyHash}

+instance ToJSON SMPServer where
+  toJSON = strToJSON
+  toEncoding = strToJEncoding
+
 data SrvLoc = SrvLoc HostName ServiceName
  deriving (Eq, Ord, Show)

@@ -25,7 +25,6 @@
 -- See https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md
 module Simplex.Messaging.Server (runSMPServer, runSMPServerBlocking) where

-import Control.Concurrent.STM (stateTVar)
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
@@ -34,9 +33,10 @@ import Crypto.Random
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Functor (($>))
+import Data.Int (Int64)
 import qualified Data.Map.Strict as M
 import Data.Maybe (isNothing)
-import Data.Time.Clock.System (getSystemTime)
+import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Type.Equality
 import Network.Socket (ServiceName)
 import qualified Simplex.Messaging.Crypto as C
@@ -47,6 +47,8 @@ import Simplex.Messaging.Server.MsgStore.STM (MsgQueue)
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.QueueStore.STM (QueueStore)
 import Simplex.Messaging.Server.StoreLog
+import Simplex.Messaging.TMap (TMap)
+import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Server
 import Simplex.Messaging.Util
@@ -68,34 +70,32 @@ runSMPServer cfg = do
 -- This function uses passed TMVar to signal when the server is ready to accept TCP requests (True)
 -- and when it is disconnected from the TCP socket once the server thread is killed (False).
 runSMPServerBlocking :: (MonadRandom m, MonadUnliftIO m) => TMVar Bool -> ServerConfig -> m ()
-runSMPServerBlocking started cfg@ServerConfig {transports} = do
-  env <- newEnv cfg
-  runReaderT smpServer env
-  where
-    smpServer :: (MonadUnliftIO m', MonadReader Env m') => m' ()
-    smpServer = do
-      s <- asks server
-      raceAny_
-        ( serverThread s subscribedQ subscribers subscriptions cancelSub :
-          serverThread s ntfSubscribedQ notifiers ntfSubscriptions (\_ -> pure ()) :
-          map runServer transports
-        )
-        `finally` withLog closeStoreLog
+runSMPServerBlocking started cfg = newEnv cfg >>= runReaderT (smpServer started)

-    runServer :: (MonadUnliftIO m', MonadReader Env m') => (ServiceName, ATransport) -> m' ()
+smpServer :: forall m. (MonadUnliftIO m, MonadReader Env m) => TMVar Bool -> m ()
+smpServer started = do
+  s <- asks server
+  cfg@ServerConfig {transports} <- asks config
+  raceAny_
+    ( serverThread s subscribedQ subscribers subscriptions cancelSub :
+      serverThread s ntfSubscribedQ notifiers ntfSubscriptions (\_ -> pure ()) :
+      map runServer transports <> expireMessagesThread_ cfg
+    )
+    `finally` withLog closeStoreLog
+  where
+    runServer :: (ServiceName, ATransport) -> m ()
    runServer (tcpPort, ATransport t) = do
      serverParams <- asks tlsServerParams
      runTransportServer started tcpPort serverParams (runClient t)

    serverThread ::
-      forall m' s.
-      MonadUnliftIO m' =>
+      forall s.
      Server ->
      (Server -> TBQueue (QueueId, Client)) ->
-      (Server -> TVar (M.Map QueueId Client)) ->
-      (Client -> TVar (M.Map QueueId s)) ->
-      (s -> m' ()) ->
-      m' ()
+      (Server -> TMap QueueId Client) ->
+      (Client -> TMap QueueId s) ->
+      (s -> m ()) ->
+      m ()
    serverThread s subQ subs clientSubs unsub = forever $ do
      atomically updateSubscribers
        >>= fmap join . mapM endPreviousSubscriptions
@@ -110,15 +110,33 @@ runSMPServerBlocking started cfg@ServerConfig {transports} = do
                  else do
                    yes <- readTVar $ connected c'
                    pure $ if yes then Just (qId, c') else Nothing
-          stateTVar (subs s) (\cs -> (M.lookup qId cs, M.insert qId clnt cs))
+          TM.lookupInsert qId clnt (subs s)
            >>= fmap join . mapM clientToBeNotified
-        endPreviousSubscriptions :: (QueueId, Client) -> m' (Maybe s)
+        endPreviousSubscriptions :: (QueueId, Client) -> m (Maybe s)
        endPreviousSubscriptions (qId, c) = do
          void . forkIO . atomically $
            writeTBQueue (sndQ c) (CorrId "", qId, END)
-          atomically . stateTVar (clientSubs c) $ \ss -> (M.lookup qId ss, M.delete qId ss)
+          atomically $ TM.lookupDelete qId (clientSubs c)

-    runClient :: (Transport c, MonadUnliftIO m, MonadReader Env m) => TProxy c -> c -> m ()
+    expireMessagesThread_ :: ServerConfig -> [m ()]
+    expireMessagesThread_ ServerConfig {messageTTL, expireMessagesInterval} =
+      case (messageTTL, expireMessagesInterval) of
+        (Just ttl, Just int) -> [expireMessages ttl int]
+        _ -> []
+
+    expireMessages :: Int64 -> Int -> m ()
+    expireMessages ttl interval = do
+      ms <- asks msgStore
+      quota <- asks $ msgQueueQuota . config
+      forever $ do
+        threadDelay interval
+        old <- subtract ttl . systemSeconds <$> liftIO getSystemTime
+        rIds <- M.keysSet <$> readTVarIO ms
+        forM_ rIds $ \rId ->
+          atomically (getMsgQueue ms rId quota)
+            >>= atomically . (`deleteExpiredMsgs` old)
+
+    runClient :: Transport c => TProxy c -> c -> m ()
    runClient _ h = do
      kh <- asks serverIdentity
      liftIO (runExceptT $ serverHandshake h kh) >>= \case
@@ -139,7 +157,7 @@ clientDisconnected c@Client {subscriptions, connected} = do
  subs <- readTVarIO subscriptions
  mapM_ cancelSub subs
  cs <- asks $ subscribers . server
-  atomically . mapM_ (modifyTVar cs . M.update deleteCurrentClient) $ M.keys subs
+  atomically . mapM_ (\rId -> TM.update deleteCurrentClient rId cs) $ M.keys subs
  where
    deleteCurrentClient :: Client -> Maybe Client
    deleteCurrentClient c'
@@ -230,7 +248,11 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ} Server {subscri
        Cmd SNotifier NSUB -> subscribeNotifications
        Cmd SRecipient command ->
          case command of
-            NEW rKey dhKey -> createQueue st rKey dhKey
+            NEW rKey dhKey ->
+              ifM
+                (asks $ allowNewQueues . config)
+                (createQueue st rKey dhKey)
+                (pure (corrId, queueId, ERR AUTH))
            SUB -> subscribeQueue queueId
            ACK -> acknowledgeMsg
            KEY sKey -> secureQueue_ st sKey
@@ -309,21 +331,19 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ} Server {subscri

        getSubscription :: RecipientId -> STM Sub
        getSubscription rId = do
-          subs <- readTVar subscriptions
-          case M.lookup rId subs of
+          TM.lookup rId subscriptions >>= \case
            Just s -> tryTakeTMVar (delivered s) $> s
            Nothing -> do
              writeTBQueue subscribedQ (rId, clnt)
              s <- newSubscription
-              writeTVar subscriptions $ M.insert rId s subs
+              TM.insert rId s subscriptions
              return s

        subscribeNotifications :: m (Transmission BrokerMsg)
        subscribeNotifications = atomically $ do
-          subs <- readTVar ntfSubscriptions
-          when (isNothing $ M.lookup queueId subs) $ do
+          whenM (isNothing <$> TM.lookup queueId ntfSubscriptions) $ do
            writeTBQueue ntfSubscribedQ (queueId, clnt)
-            writeTVar ntfSubscriptions $ M.insert queueId () subs
+            TM.insert queueId () ntfSubscriptions
          pure ok

        acknowledgeMsg :: m (Transmission BrokerMsg)
@@ -334,7 +354,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ} Server {subscri
              _ -> return $ err NO_MSG

        withSub :: RecipientId -> (Sub -> STM a) -> STM (Maybe a)
-        withSub rId f = readTVar subscriptions >>= mapM f . M.lookup rId
+        withSub rId f = mapM f =<< TM.lookup rId subscriptions

        sendMessage :: QueueStore -> MsgBody -> m (Transmission BrokerMsg)
        sendMessage st msgBody
@@ -351,9 +371,11 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ} Server {subscri
                  Left _ -> pure $ err LARGE_MSG
                  Right msg -> do
                    ms <- asks msgStore
-                    quota <- asks $ msgQueueQuota . config
+                    ServerConfig {messageTTL, msgQueueQuota} <- asks config
+                    old <- forM messageTTL $ \ttl -> subtract ttl . systemSeconds <$> liftIO getSystemTime
                    atomically $ do
-                      q <- getMsgQueue ms (recipientId qr) quota
+                      q <- getMsgQueue ms (recipientId qr) msgQueueQuota
+                      mapM_ (deleteExpiredMsgs q) old
                      ifM (isFull q) (pure $ err QUOTA) $ do
                        trySendNotification
                        writeMsg q msg
@@ -369,7 +391,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ} Server {subscri
                trySendNotification :: STM ()
                trySendNotification =
                  forM_ (notifier qr) $ \(nId, _) ->
-                    mapM_ (writeNtf nId) . M.lookup nId =<< readTVar notifiers
+                    mapM_ (writeNtf nId) =<< TM.lookup nId notifiers

                writeNtf :: NotifierId -> Client -> STM ()
                writeNtf nId Client {sndQ = q} =
@@ -403,7 +425,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ} Server {subscri
              void setDelivered

            setSub :: (Sub -> Sub) -> STM ()
-            setSub f = modifyTVar subscriptions $ M.adjust f rId
+            setSub f = TM.adjust f rId subscriptions

            setDelivered :: STM (Maybe Bool)
            setDelivered = withSub rId $ \s -> tryPutTMVar (delivered s) ()
@@ -9,6 +9,7 @@ import Control.Concurrent (ThreadId)
 import Control.Monad.IO.Unlift
 import Crypto.Random
 import Data.ByteString.Char8 (ByteString)
+import Data.Int (Int64)
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Data.X509.Validation (Fingerprint (..))
@@ -21,6 +22,8 @@ import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.QueueStore (QueueRec (..))
 import Simplex.Messaging.Server.QueueStore.STM
 import Simplex.Messaging.Server.StoreLog
+import Simplex.Messaging.TMap (TMap)
+import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (ATransport)
 import Simplex.Messaging.Transport.Server (loadFingerprint, loadTLSServerParams)
 import System.IO (IOMode (..))
@@ -34,6 +37,12 @@ data ServerConfig = ServerConfig
    queueIdBytes :: Int,
    msgIdBytes :: Int,
    storeLog :: Maybe (StoreLog 'ReadMode),
+    -- | set to False to prohibit creating new queues
+    allowNewQueues :: Bool,
+    -- | time after which the messages can be removed from the queues, seconds
+    messageTTL :: Maybe Int64,
+    -- | interval to periodically remove expired messages (when no messages are sent to the queue), microseconds
+    expireMessagesInterval :: Maybe Int,
    -- CA certificate private key is not needed for initialization
    caCertificateFile :: FilePath,
    privateKeyFile :: FilePath,
@@ -53,14 +62,14 @@ data Env = Env

 data Server = Server
  { subscribedQ :: TBQueue (RecipientId, Client),
-    subscribers :: TVar (Map RecipientId Client),
+    subscribers :: TMap RecipientId Client,
    ntfSubscribedQ :: TBQueue (NotifierId, Client),
-    notifiers :: TVar (Map NotifierId Client)
+    notifiers :: TMap NotifierId Client
  }

 data Client = Client
-  { subscriptions :: TVar (Map RecipientId Sub),
-    ntfSubscriptions :: TVar (Map NotifierId ()),
+  { subscriptions :: TMap RecipientId Sub,
+    ntfSubscriptions :: TMap NotifierId (),
    rcvQ :: TBQueue (Transmission Cmd),
    sndQ :: TBQueue (Transmission BrokerMsg),
    sessionId :: ByteString,
@@ -77,15 +86,15 @@ data Sub = Sub
 newServer :: Natural -> STM Server
 newServer qSize = do
  subscribedQ <- newTBQueue qSize
-  subscribers <- newTVar M.empty
+  subscribers <- TM.empty
  ntfSubscribedQ <- newTBQueue qSize
-  notifiers <- newTVar M.empty
+  notifiers <- TM.empty
  return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers}

 newClient :: Natural -> ByteString -> STM Client
 newClient qSize sessionId = do
-  subscriptions <- newTVar M.empty
-  ntfSubscriptions <- newTVar M.empty
+  subscriptions <- TM.empty
+  ntfSubscriptions <- TM.empty
  rcvQ <- newTBQueue qSize
  sndQ <- newTBQueue qSize
  connected <- newTVar True
@@ -109,15 +118,12 @@ newEnv config@ServerConfig {caCertificateFile, certificateFile, privateKeyFile}
  return Env {config, server, serverIdentity, queueStore, msgStore, idsDrg, storeLog = s', tlsServerParams}
  where
    restoreQueues :: QueueStore -> StoreLog 'ReadMode -> m (StoreLog 'WriteMode)
-    restoreQueues queueStore s = do
-      (queues, s') <- liftIO $ readWriteStoreLog s
-      atomically $
-        modifyTVar queueStore $ \d ->
-          d
-            { queues,
-              senders = M.foldr' addSender M.empty queues,
-              notifiers = M.foldr' addNotifier M.empty queues
-            }
+    restoreQueues QueueStore {queues, senders, notifiers} s = do
+      (qs, s') <- liftIO $ readWriteStoreLog s
+      atomically $ do
+        writeTVar queues =<< mapM newTVar qs
+        writeTVar senders $ M.foldr' addSender M.empty qs
+        writeTVar notifiers $ M.foldr' addNotifier M.empty qs
      pure s'
    addSender :: QueueRec -> Map SenderId RecipientId -> Map SenderId RecipientId
    addSender q = M.insert (senderId q) (recipientId q)
@@ -2,6 +2,7 @@

 module Simplex.Messaging.Server.MsgStore where

+import Data.Int (Int64)
 import Data.Time.Clock.System (SystemTime)
 import Numeric.Natural
 import Simplex.Messaging.Protocol (MsgBody, MsgId, RecipientId)
@@ -22,3 +23,4 @@ class MonadMsgQueue q m where
  tryPeekMsg :: q -> m (Maybe Message) -- non blocking
  peekMsg :: q -> m Message -- blocking
  tryDelPeekMsg :: q -> m (Maybe Message) -- atomic delete (== read) last and peek next message, if available
+  deleteExpiredMsgs :: q -> Int64 -> m ()
@@ -3,39 +3,38 @@
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE InstanceSigs #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
+{-# LANGUAGE NamedFieldPuns #-}

 module Simplex.Messaging.Server.MsgStore.STM where

-import Data.Map.Strict (Map)
-import qualified Data.Map.Strict as M
+import Control.Monad (when)
+import Data.Int (Int64)
+import Data.Time.Clock.System (SystemTime (systemSeconds))
 import Numeric.Natural
 import Simplex.Messaging.Protocol (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}

-newtype MsgStoreData = MsgStoreData {messages :: Map RecipientId MsgQueue}
-
-type STMMsgStore = TVar MsgStoreData
+type STMMsgStore = TMap RecipientId MsgQueue

 newMsgStore :: STM STMMsgStore
-newMsgStore = newTVar $ MsgStoreData M.empty
+newMsgStore = TM.empty

 instance MonadMsgStore STMMsgStore MsgQueue STM where
  getMsgQueue :: STMMsgStore -> RecipientId -> Natural -> STM MsgQueue
-  getMsgQueue store rId quota = do
-    m <- messages <$> readTVar store
-    maybe (newQ m) return $ M.lookup rId m
+  getMsgQueue st rId quota = maybe newQ pure =<< TM.lookup rId st
    where
-      newQ m' = do
+      newQ = do
        q <- MsgQueue <$> newTBQueue quota
-        writeTVar store . MsgStoreData $ M.insert rId q m'
+        TM.insert rId q st
        return q

  delMsgQueue :: STMMsgStore -> RecipientId -> STM ()
-  delMsgQueue store rId =
-    modifyTVar store $ MsgStoreData . M.delete rId . messages
+  delMsgQueue st rId = TM.delete rId st

 instance MonadMsgQueue MsgQueue STM where
  isFull :: MsgQueue -> STM Bool
@@ -53,3 +52,11 @@ instance MonadMsgQueue MsgQueue STM where
  -- atomic delete (== read) last and peek next message if available
  tryDelPeekMsg :: MsgQueue -> STM (Maybe Message)
  tryDelPeekMsg (MsgQueue q) = tryReadTBQueue q >> tryPeekTBQueue q
+
+  deleteExpiredMsgs :: MsgQueue -> Int64 -> STM ()
+  deleteExpiredMsgs (MsgQueue q) old = loop
+    where
+      loop = tryPeekTBQueue q >>= mapM_ delOldMsg
+      delOldMsg Message {ts} =
+        when (systemSeconds ts < old) $
+          tryReadTBQueue q >> loop
@@ -15,6 +15,7 @@ data QueueRec = QueueRec
    notifier :: Maybe (NotifierId, NtfPublicVerifyKey),
    status :: QueueStatus
  }
+  deriving (Eq, Show)

 data QueueStatus = QueueActive | QueueOff deriving (Eq, Show)

@@ -3,6 +3,7 @@
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE InstanceSigs #-}
 {-# LANGUAGE KindSignatures #-}
+{-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE RankNTypes #-}
@@ -11,107 +12,83 @@

 module Simplex.Messaging.Server.QueueStore.STM where

-import Data.Map.Strict (Map)
-import qualified Data.Map.Strict as M
+import Control.Monad
+import Data.Functor (($>))
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.QueueStore
+import Simplex.Messaging.TMap (TMap)
+import qualified Simplex.Messaging.TMap as TM
+import Simplex.Messaging.Util (ifM)
 import UnliftIO.STM

-data QueueStoreData = QueueStoreData
-  { queues :: Map RecipientId QueueRec,
-    senders :: Map SenderId RecipientId,
-    notifiers :: Map NotifierId RecipientId
+data QueueStore = QueueStore
+  { queues :: TMap RecipientId (TVar QueueRec),
+    senders :: TMap SenderId RecipientId,
+    notifiers :: TMap NotifierId RecipientId
  }

-type QueueStore = TVar QueueStoreData
-
 newQueueStore :: STM QueueStore
-newQueueStore = newTVar QueueStoreData {queues = M.empty, senders = M.empty, notifiers = M.empty}
+newQueueStore = do
+  queues <- TM.empty
+  senders <- TM.empty
+  notifiers <- TM.empty
+  pure QueueStore {queues, senders, notifiers}

 instance MonadQueueStore QueueStore STM where
  addQueue :: QueueStore -> QueueRec -> STM (Either ErrorType ())
-  addQueue store qRec@QueueRec {recipientId = rId, senderId = sId} = do
-    cs@QueueStoreData {queues, senders} <- readTVar store
-    if M.member rId queues || M.member sId senders
-      then return $ Left DUPLICATE_
-      else do
-        writeTVar store $
-          cs
-            { queues = M.insert rId qRec queues,
-              senders = M.insert sId rId senders
-            }
-        return $ Right ()
+  addQueue QueueStore {queues, senders} q@QueueRec {recipientId = rId, senderId = sId} = do
+    ifM hasId (pure $ Left DUPLICATE_) $ do
+      qVar <- newTVar q
+      TM.insert rId qVar queues
+      TM.insert sId rId senders
+      pure $ Right ()
+    where
+      hasId = (||) <$> TM.member rId queues <*> TM.member sId senders

  getQueue :: QueueStore -> SParty p -> QueueId -> STM (Either ErrorType QueueRec)
-  getQueue st party qId = do
-    cs <- readTVar st
-    pure $ case party of
-      SRecipient -> getRcpQueue cs qId
-      SSender -> getPartyQueue cs senders
-      SNotifier -> getPartyQueue cs notifiers
+  getQueue QueueStore {queues, senders, notifiers} party qId =
+    toResult <$> (mapM readTVar =<< getVar)
    where
-      getPartyQueue ::
-        QueueStoreData ->
-        (QueueStoreData -> Map QueueId RecipientId) ->
-        Either ErrorType QueueRec
-      getPartyQueue cs recipientIds =
-        case M.lookup qId $ recipientIds cs of
-          Just rId -> getRcpQueue cs rId
-          Nothing -> Left AUTH
+      getVar = case party of
+        SRecipient -> TM.lookup qId queues
+        SSender -> TM.lookup qId senders >>= get
+        SNotifier -> TM.lookup qId notifiers >>= get
+      get = fmap join . mapM (`TM.lookup` queues)

  secureQueue :: QueueStore -> RecipientId -> SndPublicVerifyKey -> STM (Either ErrorType QueueRec)
-  secureQueue store rId sKey =
-    updateQueues store rId $ \cs c ->
-      case senderKey c of
-        Just _ -> (Left AUTH, cs)
-        _ -> (Right c, cs {queues = M.insert rId c {senderKey = Just sKey} (queues cs)})
+  secureQueue QueueStore {queues} rId sKey =
+    withQueue rId queues $ \qVar ->
+      readTVar qVar >>= \q -> case senderKey q of
+        Just _ -> pure Nothing
+        _ -> writeTVar qVar q {senderKey = Just sKey} $> Just q

  addQueueNotifier :: QueueStore -> RecipientId -> NotifierId -> NtfPublicVerifyKey -> STM (Either ErrorType QueueRec)
-  addQueueNotifier store rId nId nKey = do
-    cs@QueueStoreData {queues, notifiers} <- readTVar store
-    if M.member nId notifiers
-      then pure $ Left DUPLICATE_
-      else case M.lookup rId queues of
-        Nothing -> pure $ Left AUTH
-        Just q -> case notifier q of
-          Just _ -> pure $ Left AUTH
+  addQueueNotifier QueueStore {queues, notifiers} rId nId nKey = do
+    ifM (TM.member nId notifiers) (pure $ Left DUPLICATE_) $
+      withQueue rId queues $ \qVar ->
+        readTVar qVar >>= \q -> case notifier q of
+          Just _ -> pure Nothing
          _ -> do
-            writeTVar store $
-              cs
-                { queues = M.insert rId q {notifier = Just (nId, nKey)} queues,
-                  notifiers = M.insert nId rId notifiers
-                }
-            pure $ Right q
+            writeTVar qVar q {notifier = Just (nId, nKey)}
+            TM.insert nId rId notifiers
+            pure $ Just q

  suspendQueue :: QueueStore -> RecipientId -> STM (Either ErrorType ())
-  suspendQueue store rId =
-    updateQueues store rId $ \cs c ->
-      (Right (), cs {queues = M.insert rId c {status = QueueOff} (queues cs)})
+  suspendQueue QueueStore {queues} rId =
+    withQueue rId queues $ \qVar -> modifyTVar' qVar (\q -> q {status = QueueOff}) $> Just ()

  deleteQueue :: QueueStore -> RecipientId -> STM (Either ErrorType ())
-  deleteQueue store rId =
-    updateQueues store rId $ \cs c ->
-      ( Right (),
-        cs
-          { queues = M.delete rId (queues cs),
-            senders = M.delete (senderId c) (senders cs)
-          }
-      )
+  deleteQueue QueueStore {queues, senders, notifiers} rId = do
+    TM.lookupDelete rId queues >>= \case
+      Just qVar ->
+        readTVar qVar >>= \q -> do
+          TM.delete (senderId q) senders
+          forM_ (notifier q) $ \(nId, _) -> TM.delete nId notifiers
+          pure $ Right ()
+      _ -> pure $ Left AUTH

-updateQueues ::
-  QueueStore ->
-  RecipientId ->
-  (QueueStoreData -> QueueRec -> (Either ErrorType a, QueueStoreData)) ->
-  STM (Either ErrorType a)
-updateQueues store rId update = do
-  cs <- readTVar store
-  let conn = getRcpQueue cs rId
-  either (return . Left) (_update cs) conn
-  where
-    _update cs c = do
-      let (res, cs') = update cs c
-      writeTVar store cs'
-      return res
+toResult :: Maybe a -> Either ErrorType a
+toResult = maybe (Left AUTH) Right

-getRcpQueue :: QueueStoreData -> RecipientId -> Either ErrorType QueueRec
-getRcpQueue cs rId = maybe (Left AUTH) Right . M.lookup rId $ queues cs
+withQueue :: RecipientId -> TMap RecipientId (TVar QueueRec) -> (TVar QueueRec -> STM (Maybe a)) -> STM (Either ErrorType a)
+withQueue rId queues f = toResult <$> (TM.lookup rId queues >>= fmap join . mapM f)
@@ -0,0 +1,70 @@
+module Simplex.Messaging.TMap
+  ( TMap,
+    empty,
+    singleton,
+    Simplex.Messaging.TMap.lookup,
+    member,
+    insert,
+    delete,
+    lookupInsert,
+    lookupDelete,
+    adjust,
+    update,
+    alter,
+    union,
+  )
+where
+
+import Control.Concurrent.STM
+import Data.Map.Strict (Map)
+import qualified Data.Map.Strict as M
+
+type TMap k a = TVar (Map k a)
+
+empty :: STM (TMap k a)
+empty = newTVar M.empty
+{-# INLINE empty #-}
+
+singleton :: k -> a -> STM (TMap k a)
+singleton k v = newTVar $ M.singleton k v
+{-# INLINE singleton #-}
+
+lookup :: Ord k => k -> TMap k a -> STM (Maybe a)
+lookup k m = M.lookup k <$> readTVar m
+{-# INLINE lookup #-}
+
+member :: Ord k => k -> TMap k a -> STM Bool
+member k m = M.member k <$> readTVar m
+{-# INLINE member #-}
+
+insert :: Ord k => k -> a -> TMap k a -> STM ()
+insert k v m = modifyTVar' m $ M.insert k v
+{-# INLINE insert #-}
+
+delete :: Ord k => k -> TMap k a -> STM ()
+delete k m = modifyTVar' m $ M.delete k
+{-# INLINE delete #-}
+
+lookupInsert :: Ord k => k -> a -> TMap k a -> STM (Maybe a)
+lookupInsert k v m = stateTVar m $ \mv -> (M.lookup k mv, M.insert k v mv)
+{-# INLINE lookupInsert #-}
+
+lookupDelete :: Ord k => k -> TMap k a -> STM (Maybe a)
+lookupDelete k m = stateTVar m $ \mv -> (M.lookup k mv, M.delete k mv)
+{-# INLINE lookupDelete #-}
+
+adjust :: Ord k => (a -> a) -> k -> TMap k a -> STM ()
+adjust f k m = modifyTVar' m $ M.adjust f k
+{-# INLINE adjust #-}
+
+update :: Ord k => (a -> Maybe a) -> k -> TMap k a -> STM ()
+update f k m = modifyTVar' m $ M.update f k
+{-# INLINE update #-}
+
+alter :: Ord k => (Maybe a -> Maybe a) -> k -> TMap k a -> STM ()
+alter f k m = modifyTVar' m $ M.alter f k
+{-# INLINE alter #-}
+
+union :: Ord k => Map k a -> TMap k a -> STM ()
+union m' m = modifyTVar' m $ M.union m'
+{-# INLINE union #-}
@@ -38,6 +38,7 @@ module Simplex.Messaging.Transport

    -- * TLS Transport
    TLS (..),
+    SessionId,
    connectTLS,
    closeTLS,
    supportedParameters,
@@ -96,7 +97,7 @@ supportedSMPVersions :: VersionRange
 supportedSMPVersions = mkVersionRange 1 1

 simplexMQVersion :: String
-simplexMQVersion = "1.0.2"
+simplexMQVersion = "1.1.0"

 -- * Transport connection class

@@ -115,7 +116,7 @@ class Transport c where
  getClientConnection :: T.Context -> IO c

  -- | tls-unique channel binding per RFC5929
-  tlsUnique :: c -> ByteString
+  tlsUnique :: c -> SessionId

  -- | Close connection
  closeConnection :: c -> IO ()
@@ -250,14 +251,17 @@ trimCR s = if B.last s == '\r' then B.init s else s
 -- | The handle for SMP encrypted transport connection over Transport .
 data THandle c = THandle
  { connection :: c,
-    sessionId :: ByteString,
+    sessionId :: SessionId,
    -- | agreed SMP server protocol version
    smpVersion :: Version
  }

+-- | TLS-unique channel binding
+type SessionId = ByteString
+
 data ServerHandshake = ServerHandshake
  { smpVersionRange :: VersionRange,
-    sessionId :: ByteString
+    sessionId :: SessionId
  }

 data ClientHandshake = ClientHandshake
@@ -20,19 +20,20 @@ import Network.Socket
 import qualified Network.TLS as T
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Transport
+import Simplex.Messaging.Transport.KeepAlive
 import System.IO.Error
 import UnliftIO.Exception (IOException)
 import qualified UnliftIO.Exception as E

 -- | Connect to passed TCP host:port and pass handle to the client.
-runTransportClient :: Transport c => MonadUnliftIO m => HostName -> ServiceName -> C.KeyHash -> (c -> m a) -> m a
-runTransportClient host port keyHash client = do
+runTransportClient :: Transport c => MonadUnliftIO m => HostName -> ServiceName -> C.KeyHash -> Maybe KeepAliveOpts -> (c -> m a) -> m a
+runTransportClient host port keyHash keepAliveOpts client = do
  let clientParams = mkTLSClientParams host port keyHash
-  c <- liftIO $ startTCPClient host port clientParams
+  c <- liftIO $ startTCPClient host port clientParams keepAliveOpts
  client c `E.finally` liftIO (closeConnection c)

-startTCPClient :: forall c. Transport c => HostName -> ServiceName -> T.ClientParams -> IO c
-startTCPClient host port clientParams = withSocketsDo $ resolve >>= tryOpen err
+startTCPClient :: forall c. Transport c => HostName -> ServiceName -> T.ClientParams -> Maybe KeepAliveOpts -> IO c
+startTCPClient host port clientParams keepAliveOpts = withSocketsDo $ resolve >>= tryOpen err
  where
    err :: IOException
    err = mkIOError NoSuchThing "no address" Nothing Nothing
@@ -51,6 +52,7 @@ startTCPClient host port clientParams = withSocketsDo $ resolve >>= tryOpen err
    open addr = do
      sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)
      connect sock $ addrAddress addr
+      mapM_ (setSocketKeepAlive sock) keepAliveOpts
      ctx <- connectTLS clientParams sock
      getClientConnection ctx

@@ -0,0 +1,62 @@
+{-# LANGUAGE CApiFFI #-}
+{-# LANGUAGE CPP #-}
+{-# LANGUAGE NamedFieldPuns #-}
+
+module Simplex.Messaging.Transport.KeepAlive where
+
+import Foreign.C (CInt (..))
+import Network.Socket
+
+data KeepAliveOpts = KeepAliveOpts
+  { keepIdle :: Int,
+    keepIntvl :: Int,
+    keepCnt :: Int
+  }
+
+defaultKeepAliveOpts :: KeepAliveOpts
+defaultKeepAliveOpts =
+  KeepAliveOpts
+    { keepIdle = 30,
+      keepIntvl = 15,
+      keepCnt = 4
+    }
+
+_SOL_TCP :: CInt
+_SOL_TCP = 6
+
+#if defined(mingw32_HOST_OS)
+-- Windows
+
+-- The values are copied from windows::Win32::Networking::WinSock
+-- https://microsoft.github.io/windows-docs-rs/doc/windows/Win32/Networking/WinSock/index.html
+
+_TCP_KEEPIDLE :: CInt
+_TCP_KEEPIDLE = 3
+
+_TCP_KEEPINTVL :: CInt
+_TCP_KEEPINTVL = 17
+
+_TCP_KEEPCNT :: CInt
+_TCP_KEEPCNT = 16
+
+#else
+-- Mac/Linux
+
+#if defined(darwin_HOST_OS)
+foreign import capi "netinet/tcp.h value TCP_KEEPALIVE" _TCP_KEEPIDLE :: CInt
+#else
+foreign import capi "netinet/tcp.h value TCP_KEEPIDLE" _TCP_KEEPIDLE :: CInt
+#endif
+
+foreign import capi "netinet/tcp.h value TCP_KEEPINTVL" _TCP_KEEPINTVL :: CInt
+
+foreign import capi "netinet/tcp.h value TCP_KEEPCNT" _TCP_KEEPCNT :: CInt
+
+#endif
+
+setSocketKeepAlive :: Socket -> KeepAliveOpts -> IO ()
+setSocketKeepAlive sock KeepAliveOpts {keepCnt, keepIdle, keepIntvl} = do
+  setSocketOption sock KeepAlive 1
+  setSocketOption sock (SockOpt _SOL_TCP _TCP_KEEPIDLE) keepIdle
+  setSocketOption sock (SockOpt _SOL_TCP _TCP_KEEPINTVL) keepIntvl
+  setSocketOption sock (SockOpt _SOL_TCP _TCP_KEEPCNT) keepCnt
@@ -4,6 +4,7 @@

 module Simplex.Messaging.Transport.Server
  ( runTransportServer,
+    runTCPServer,
    loadTLSServerParams,
    loadFingerprint,
    serverHandshake,
@@ -37,11 +38,11 @@ runTransportServer started port serverParams server = do
    clients <- newTVarIO S.empty
    E.bracket
      (startTCPServer started port)
-      (closeServer clients)
+      (closeServer started clients)
      $ \sock -> forever $ do
        (connSock, _) <- accept sock
        tid <- forkIO $ connectClient u connSock `E.catch` \(_ :: E.SomeException) -> pure ()
-        atomically . modifyTVar clients $ S.insert tid
+        atomically . modifyTVar' clients $ S.insert tid
  where
    connectClient :: UnliftIO m -> Socket -> IO ()
    connectClient u connSock =
@@ -49,11 +50,23 @@ runTransportServer started port serverParams server = do
        (connectTLS serverParams connSock >>= getServerConnection)
        closeConnection
        (unliftIO u . server)
-    closeServer :: TVar (Set ThreadId) -> Socket -> IO ()
-    closeServer clients sock = do
-      readTVarIO clients >>= mapM_ killThread
-      close sock
-      void . atomically $ tryPutTMVar started False
+
+runTCPServer :: TMVar Bool -> ServiceName -> (Socket -> IO ()) -> IO ()
+runTCPServer started port server = do
+  clients <- newTVarIO S.empty
+  E.bracket
+    (startTCPServer started port)
+    (closeServer started clients)
+    $ \sock -> forever $ do
+      (connSock, _) <- accept sock
+      tid <- forkIO $ server connSock `E.catch` \(_ :: E.SomeException) -> pure ()
+      atomically . modifyTVar' clients $ S.insert tid
+
+closeServer :: TMVar Bool -> TVar (Set ThreadId) -> Socket -> IO ()
+closeServer started clients sock = do
+  readTVarIO clients >>= mapM_ killThread
+  close sock
+  void . atomically $ tryPutTMVar started False

 startTCPServer :: TMVar Bool -> ServiceName -> IO Socket
 startTCPServer started port = withSocketsDo $ resolve >>= open >>= setStarted
@@ -58,6 +58,10 @@ 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 #-}

+whenM :: Monad m => m Bool -> m () -> m ()
+whenM b a = ifM b a $ pure ()
+{-# INLINE whenM #-}
+
 unlessM :: Monad m => m Bool -> m () -> m ()
 unlessM b = ifM b $ pure ()
 {-# INLINE unlessM #-}
@@ -36,6 +36,7 @@ packages:
 #
 extra-deps:
  - cryptostore-0.2.1.0@sha256:9896e2984f36a1c8790f057fd5ce3da4cbcaf8aa73eb2d9277916886978c5b19,3881
+  - network-3.1.2.7@sha256:e3d78b13db9512aeb106e44a334ab42b7aa48d26c097299084084cb8be5c5568,4888
  - simple-logger-0.1.0@sha256:be8ede4bd251a9cac776533bae7fb643369ebd826eb948a9a18df1a8dd252ff8,1079
  - tls-1.5.7@sha256:1cc30253a9696b65a9cafc0317fbf09f7dcea15e3a145ed6c9c0e28c632fa23a,6991
 # below dependancies are to update Aeson to 2.0.3
@@ -62,9 +62,11 @@ agentTests (ATransport t) = do
      smpAgentTest3_1_1 $ testSubscription t
    it "should send notifications to client when server disconnects" $
      smpAgentServerTest $ testSubscrNotification t
-  describe "Message delivery" $ do
+  describe "Message delivery and server reconnection" $ do
    it "should deliver messages after losing server connection and re-connecting" $
      smpAgentTest2_2_2_needs_server $ testMsgDeliveryServerRestart t
+    it "should connect to the server when server goes up if it initially was down" $
+      smpAgentTestN [] $ testServerConnectionAfterError t
    it "should deliver pending messages after agent restarting" $
      smpAgentTest1_1_1 $ testMsgDeliveryAgentRestart t
    it "should concurrently deliver messages to connections without blocking" $
@@ -126,25 +128,25 @@ testDuplexConnection _ alice bob = do
  bob <# ("", "alice", CON)
  alice <# ("", "bob", CON)
  -- message IDs 1 to 3 get assigned to control messages, so first MSG is assigned ID 4
-  alice #: ("3", "bob", "SEND :hello") #> ("3", "bob", MID 4)
-  alice <# ("", "bob", SENT 4)
-  bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False
-  bob #: ("12", "alice", "ACK 4") #> ("12", "alice", OK)
-  alice #: ("4", "bob", "SEND :how are you?") #> ("4", "bob", MID 5)
+  alice #: ("3", "bob", "SEND :hello") #> ("3", "bob", MID 5)
  alice <# ("", "bob", SENT 5)
+  bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False
+  bob #: ("12", "alice", "ACK 5") #> ("12", "alice", OK)
+  alice #: ("4", "bob", "SEND :how are you?") #> ("4", "bob", MID 6)
+  alice <# ("", "bob", SENT 6)
  bob <#= \case ("", "alice", Msg "how are you?") -> True; _ -> False
-  bob #: ("13", "alice", "ACK 5") #> ("13", "alice", OK)
-  bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", MID 6)
-  bob <# ("", "alice", SENT 6)
-  alice <#= \case ("", "bob", Msg "hello too") -> True; _ -> False
-  alice #: ("3a", "bob", "ACK 6") #> ("3a", "bob", OK)
-  bob #: ("15", "alice", "SEND 9\nmessage 1") #> ("15", "alice", MID 7)
+  bob #: ("13", "alice", "ACK 6") #> ("13", "alice", OK)
+  bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", MID 7)
  bob <# ("", "alice", SENT 7)
+  alice <#= \case ("", "bob", Msg "hello too") -> True; _ -> False
+  alice #: ("3a", "bob", "ACK 7") #> ("3a", "bob", OK)
+  bob #: ("15", "alice", "SEND 9\nmessage 1") #> ("15", "alice", MID 8)
+  bob <# ("", "alice", SENT 8)
  alice <#= \case ("", "bob", Msg "message 1") -> True; _ -> False
-  alice #: ("4a", "bob", "ACK 7") #> ("4a", "bob", OK)
+  alice #: ("4a", "bob", "ACK 8") #> ("4a", "bob", OK)
  alice #: ("5", "bob", "OFF") #> ("5", "bob", OK)
-  bob #: ("17", "alice", "SEND 9\nmessage 3") #> ("17", "alice", MID 8)
-  bob <# ("", "alice", MERR 8 (SMP AUTH))
+  bob #: ("17", "alice", "SEND 9\nmessage 3") #> ("17", "alice", MID 9)
+  bob <# ("", "alice", MERR 9 (SMP AUTH))
  alice #: ("6", "bob", "DEL") #> ("6", "bob", OK)
  alice #:# "nothing else should be delivered to alice"

@@ -159,25 +161,25 @@ testDuplexConnRandomIds _ alice bob = do
  bob <# ("", aliceConn, INFO "alice's connInfo")
  bob <# ("", aliceConn, CON)
  alice <# ("", bobConn, CON)
-  alice #: ("2", bobConn, "SEND :hello") #> ("2", bobConn, MID 4)
-  alice <# ("", bobConn, SENT 4)
-  bob <#= \case ("", c, Msg "hello") -> c == aliceConn; _ -> False
-  bob #: ("12", aliceConn, "ACK 4") #> ("12", aliceConn, OK)
-  alice #: ("3", bobConn, "SEND :how are you?") #> ("3", bobConn, MID 5)
+  alice #: ("2", bobConn, "SEND :hello") #> ("2", bobConn, MID 5)
  alice <# ("", bobConn, SENT 5)
+  bob <#= \case ("", c, Msg "hello") -> c == aliceConn; _ -> False
+  bob #: ("12", aliceConn, "ACK 5") #> ("12", aliceConn, OK)
+  alice #: ("3", bobConn, "SEND :how are you?") #> ("3", bobConn, MID 6)
+  alice <# ("", bobConn, SENT 6)
  bob <#= \case ("", c, Msg "how are you?") -> c == aliceConn; _ -> False
-  bob #: ("13", aliceConn, "ACK 5") #> ("13", aliceConn, OK)
-  bob #: ("14", aliceConn, "SEND 9\nhello too") #> ("14", aliceConn, MID 6)
-  bob <# ("", aliceConn, SENT 6)
-  alice <#= \case ("", c, Msg "hello too") -> c == bobConn; _ -> False
-  alice #: ("3a", bobConn, "ACK 6") #> ("3a", bobConn, OK)
-  bob #: ("15", aliceConn, "SEND 9\nmessage 1") #> ("15", aliceConn, MID 7)
+  bob #: ("13", aliceConn, "ACK 6") #> ("13", aliceConn, OK)
+  bob #: ("14", aliceConn, "SEND 9\nhello too") #> ("14", aliceConn, MID 7)
  bob <# ("", aliceConn, SENT 7)
+  alice <#= \case ("", c, Msg "hello too") -> c == bobConn; _ -> False
+  alice #: ("3a", bobConn, "ACK 7") #> ("3a", bobConn, OK)
+  bob #: ("15", aliceConn, "SEND 9\nmessage 1") #> ("15", aliceConn, MID 8)
+  bob <# ("", aliceConn, SENT 8)
  alice <#= \case ("", c, Msg "message 1") -> c == bobConn; _ -> False
-  alice #: ("4a", bobConn, "ACK 7") #> ("4a", bobConn, OK)
+  alice #: ("4a", bobConn, "ACK 8") #> ("4a", bobConn, OK)
  alice #: ("5", bobConn, "OFF") #> ("5", bobConn, OK)
-  bob #: ("17", aliceConn, "SEND 9\nmessage 3") #> ("17", aliceConn, MID 8)
-  bob <# ("", aliceConn, MERR 8 (SMP AUTH))
+  bob #: ("17", aliceConn, "SEND 9\nmessage 3") #> ("17", aliceConn, MID 9)
+  bob <# ("", aliceConn, MERR 9 (SMP AUTH))
  alice #: ("6", bobConn, "DEL") #> ("6", bobConn, OK)
  alice #:# "nothing else should be delivered to alice"

@@ -194,10 +196,10 @@ testContactConnection _ alice bob tom = do
  alice <# ("", "bob", INFO "bob's connInfo 2")
  alice <# ("", "bob", CON)
  bob <# ("", "alice", CON)
-  alice #: ("3", "bob", "SEND :hi") #> ("3", "bob", MID 4)
-  alice <# ("", "bob", SENT 4)
+  alice #: ("3", "bob", "SEND :hi") #> ("3", "bob", MID 5)
+  alice <# ("", "bob", SENT 5)
  bob <#= \case ("", "alice", Msg "hi") -> True; _ -> False
-  bob #: ("13", "alice", "ACK 4") #> ("13", "alice", OK)
+  bob #: ("13", "alice", "ACK 5") #> ("13", "alice", OK)

  tom #: ("21", "alice", "JOIN " <> cReq' <> " 14\ntom's connInfo") #> ("21", "alice", OK)
  ("", "alice_contact", Right (REQ aInvId' "tom's connInfo")) <- (alice <#:)
@@ -207,10 +209,10 @@ testContactConnection _ alice bob tom = do
  alice <# ("", "tom", INFO "tom's connInfo 2")
  alice <# ("", "tom", CON)
  tom <# ("", "alice", CON)
-  alice #: ("5", "tom", "SEND :hi there") #> ("5", "tom", MID 4)
-  alice <# ("", "tom", SENT 4)
+  alice #: ("5", "tom", "SEND :hi there") #> ("5", "tom", MID 5)
+  alice <# ("", "tom", SENT 5)
  tom <#= \case ("", "alice", Msg "hi there") -> True; _ -> False
-  tom #: ("23", "alice", "ACK 4") #> ("23", "alice", OK)
+  tom #: ("23", "alice", "ACK 5") #> ("23", "alice", OK)

 testContactConnRandomIds :: Transport c => TProxy c -> c -> c -> IO ()
 testContactConnRandomIds _ alice bob = do
@@ -230,10 +232,10 @@ testContactConnRandomIds _ alice bob = do
  alice <# ("", bobConn, CON)
  bob <# ("", aliceConn, CON)

-  alice #: ("3", bobConn, "SEND :hi") #> ("3", bobConn, MID 4)
-  alice <# ("", bobConn, SENT 4)
+  alice #: ("3", bobConn, "SEND :hi") #> ("3", bobConn, MID 5)
+  alice <# ("", bobConn, SENT 5)
  bob <#= \case ("", c, Msg "hi") -> c == aliceConn; _ -> False
-  bob #: ("13", aliceConn, "ACK 4") #> ("13", aliceConn, OK)
+  bob #: ("13", aliceConn, "ACK 5") #> ("13", aliceConn, OK)

 testRejectContactRequest :: Transport c => TProxy c -> c -> c -> IO ()
 testRejectContactRequest _ alice bob = do
@@ -250,20 +252,20 @@ testRejectContactRequest _ alice bob = do
 testSubscription :: Transport c => TProxy c -> c -> c -> c -> IO ()
 testSubscription _ alice1 alice2 bob = do
  (alice1, "alice") `connect` (bob, "bob")
-  bob #: ("12", "alice", "SEND 5\nhello") #> ("12", "alice", MID 4)
-  bob <# ("", "alice", SENT 4)
-  alice1 <#= \case ("", "bob", Msg "hello") -> True; _ -> False
-  alice1 #: ("1", "bob", "ACK 4") #> ("1", "bob", OK)
-  bob #: ("13", "alice", "SEND 11\nhello again") #> ("13", "alice", MID 5)
+  bob #: ("12", "alice", "SEND 5\nhello") #> ("12", "alice", MID 5)
  bob <# ("", "alice", SENT 5)
+  alice1 <#= \case ("", "bob", Msg "hello") -> True; _ -> False
+  alice1 #: ("1", "bob", "ACK 5") #> ("1", "bob", OK)
+  bob #: ("13", "alice", "SEND 11\nhello again") #> ("13", "alice", MID 6)
+  bob <# ("", "alice", SENT 6)
  alice1 <#= \case ("", "bob", Msg "hello again") -> True; _ -> False
-  alice1 #: ("2", "bob", "ACK 5") #> ("2", "bob", OK)
+  alice1 #: ("2", "bob", "ACK 6") #> ("2", "bob", OK)
  alice2 #: ("21", "bob", "SUB") #> ("21", "bob", OK)
  alice1 <# ("", "bob", END)
-  bob #: ("14", "alice", "SEND 2\nhi") #> ("14", "alice", MID 6)
-  bob <# ("", "alice", SENT 6)
+  bob #: ("14", "alice", "SEND 2\nhi") #> ("14", "alice", MID 7)
+  bob <# ("", "alice", SENT 7)
  alice2 <#= \case ("", "bob", Msg "hi") -> True; _ -> False
-  alice2 #: ("22", "bob", "ACK 6") #> ("22", "bob", OK)
+  alice2 #: ("22", "bob", "ACK 7") #> ("22", "bob", OK)
  alice1 #:# "nothing else should be delivered to alice1"

 testSubscrNotification :: Transport c => TProxy c -> (ThreadId, ThreadId) -> c -> IO ()
@@ -279,40 +281,77 @@ testMsgDeliveryServerRestart :: Transport c => TProxy c -> c -> c -> IO ()
 testMsgDeliveryServerRestart t alice bob = do
  withServer $ do
    connect (alice, "alice") (bob, "bob")
-    bob #: ("1", "alice", "SEND 2\nhi") #> ("1", "alice", MID 4)
-    bob <# ("", "alice", SENT 4)
+    bob #: ("1", "alice", "SEND 2\nhi") #> ("1", "alice", MID 5)
+    bob <# ("", "alice", SENT 5)
    alice <#= \case ("", "bob", Msg "hi") -> True; _ -> False
-    alice #: ("11", "bob", "ACK 4") #> ("11", "bob", OK)
+    alice #: ("11", "bob", "ACK 5") #> ("11", "bob", OK)
    alice #:# "nothing else delivered before the server is killed"

  alice <# ("", "bob", DOWN)
-  bob #: ("2", "alice", "SEND 11\nhello again") #> ("2", "alice", MID 5)
+  bob #: ("2", "alice", "SEND 11\nhello again") #> ("2", "alice", MID 6)
  bob #:# "nothing else delivered before the server is restarted"
  alice #:# "nothing else delivered before the server is restarted"

  withServer $ do
-    bob <# ("", "alice", SENT 5)
+    bob <# ("", "alice", SENT 6)
    alice <# ("", "bob", UP)
    alice <#= \case ("", "bob", Msg "hello again") -> True; _ -> False
-    alice #: ("12", "bob", "ACK 5") #> ("12", "bob", OK)
+    alice #: ("12", "bob", "ACK 6") #> ("12", "bob", OK)

  removeFile testStoreLogFile
  where
    withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()

+testServerConnectionAfterError :: forall c. Transport c => TProxy c -> [c] -> IO ()
+testServerConnectionAfterError t _ = do
+  withAgent1 $ \bob -> do
+    withAgent2 $ \alice -> do
+      withServer $ do
+        connect (bob, "bob") (alice, "alice")
+
+      bob <# ("", "alice", DOWN)
+      alice <# ("", "bob", DOWN)
+      alice #: ("1", "bob", "SEND 5\nhello") #> ("1", "bob", MID 5)
+      alice #:# "nothing else delivered before the server is restarted"
+      bob #:# "nothing else delivered before the server is restarted"
+
+  withAgent1 $ \bob -> do
+    withAgent2 $ \alice -> do
+      bob #: ("1", "alice", "SUB") #> ("1", "alice", ERR (BROKER NETWORK))
+      alice #: ("1", "bob", "SUB") #> ("1", "bob", ERR (BROKER NETWORK))
+      withServer $ do
+        alice <#= \case ("", "bob", cmd) -> cmd == UP || cmd == SENT 5; _ -> False
+        alice <#= \case ("", "bob", cmd) -> cmd == UP || cmd == SENT 5; _ -> False
+        bob <# ("", "alice", UP)
+        bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False
+        bob #: ("2", "alice", "ACK 5") #> ("2", "alice", OK)
+        alice #: ("1", "bob", "SEND 11\nhello again") #> ("1", "bob", MID 6)
+        alice <# ("", "bob", SENT 6)
+        bob <#= \case ("", "alice", Msg "hello again") -> True; _ -> False
+
+  removeFile testStoreLogFile
+  removeFile testDB
+  removeFile testDB2
+  where
+    withServer test' = withSmpServerStoreLogOn (ATransport t) testPort2 (const test') `shouldReturn` ()
+    withAgent1 = withAgent agentTestPort testDB
+    withAgent2 = withAgent agentTestPort2 testDB2
+    withAgent :: String -> String -> (c -> IO a) -> IO a
+    withAgent agentPort agentDB = withSmpAgentThreadOn_ (ATransport t) (agentPort, testPort2, agentDB) (pure ()) . const . testSMPAgentClientOn agentPort
+
 testMsgDeliveryAgentRestart :: Transport c => TProxy c -> c -> IO ()
 testMsgDeliveryAgentRestart t bob = do
  withAgent $ \alice -> do
    withServer $ do
      connect (bob, "bob") (alice, "alice")
-      alice #: ("1", "bob", "SEND 5\nhello") #> ("1", "bob", MID 4)
-      alice <# ("", "bob", SENT 4)
+      alice #: ("1", "bob", "SEND 5\nhello") #> ("1", "bob", MID 5)
+      alice <# ("", "bob", SENT 5)
      bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False
-      bob #: ("11", "alice", "ACK 4") #> ("11", "alice", OK)
+      bob #: ("11", "alice", "ACK 5") #> ("11", "alice", OK)
      bob #:# "nothing else delivered before the server is down"

    bob <# ("", "alice", DOWN)
-    alice #: ("2", "bob", "SEND 11\nhello again") #> ("2", "bob", MID 5)
+    alice #: ("2", "bob", "SEND 11\nhello again") #> ("2", "bob", MID 6)
    alice #:# "nothing else delivered before the server is restarted"
    bob #:# "nothing else delivered before the server is restarted"

@@ -322,11 +361,11 @@ testMsgDeliveryAgentRestart t bob = do
      alice <#= \case
        (corrId, "bob", cmd) ->
          (corrId == "3" && cmd == OK)
-            || (corrId == "" && cmd == SENT 5)
+            || (corrId == "" && cmd == SENT 6)
        _ -> False
      bob <# ("", "alice", UP)
      bob <#= \case ("", "alice", Msg "hello again") -> True; _ -> False
-      bob #: ("12", "alice", "ACK 5") #> ("12", "alice", OK)
+      bob #: ("12", "alice", "ACK 6") #> ("12", "alice", OK)

  removeFile testStoreLogFile
  removeFile testDB
@@ -354,11 +393,11 @@ testConcurrentMsgDelivery _ alice bob = do
  -- 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 5)
-  bob <# ("", "alice", SENT 5)
+  bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", MID 6)
+  bob <# ("", "alice", SENT 6)
  -- if delivery is blocked it won't go further
  alice <#= \case ("", "bob", Msg "hello too") -> True; _ -> False
-  alice #: ("3", "bob", "ACK 5") #> ("3", "bob", OK)
+  alice #: ("3", "bob", "ACK 6") #> ("3", "bob", OK)

 testMsgDeliveryQuotaExceeded :: Transport c => TProxy c -> c -> c -> IO ()
 testMsgDeliveryQuotaExceeded _ alice bob = do
@@ -371,9 +410,9 @@ testMsgDeliveryQuotaExceeded _ alice bob = do
    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 4)
+  alice #: ("1", "bob2", "SEND :hello") #> ("1", "bob2", MID 5)
  -- if delivery is blocked it won't go further
-  alice <# ("", "bob2", SENT 4)
+  alice <# ("", "bob2", SENT 5)

 connect :: forall c. Transport c => (c, ByteString) -> (c, ByteString) -> IO ()
 connect (h1, name1) (h2, name2) = do
@@ -10,7 +10,7 @@ module AgentTests.FunctionalAPITests (functionalAPITests) where
 import Control.Monad.Except (ExceptT, runExceptT)
 import Control.Monad.IO.Unlift
 import SMPAgentClient
-import SMPClient (withSmpServer)
+import SMPClient (testPort, withSmpServer, withSmpServerStoreLogOn)
 import Simplex.Messaging.Agent
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..))
 import Simplex.Messaging.Agent.Protocol
@@ -44,6 +44,8 @@ functionalAPITests t = do
      withSmpServer t testAsyncJoiningOfflineBeforeActivation
    it "should connect with both clients going offline" $
      withSmpServer t testAsyncBothOffline
+    it "should connect on the second attempt if server was offline" $
+      testAsyncServerOffline t
    it "should notify after HELLO timeout" $
      withSmpServer t testAsyncHelloTimeout

@@ -59,26 +61,26 @@ testAgentClient = do
    get alice ##> ("", bobId, CON)
    get bob ##> ("", aliceId, INFO "alice's connInfo")
    get bob ##> ("", aliceId, CON)
-    -- message IDs 1 to 3 get assigned to control messages, so first MSG is assigned ID 4
-    4 <- sendMessage alice bobId "hello"
-    get alice ##> ("", bobId, SENT 4)
-    5 <- sendMessage alice bobId "how are you?"
+    -- message IDs 1 to 4 get assigned to control messages, so first MSG is assigned ID 5
+    5 <- sendMessage alice bobId "hello"
    get alice ##> ("", bobId, SENT 5)
+    6 <- sendMessage alice bobId "how are you?"
+    get alice ##> ("", bobId, SENT 6)
    get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
-    ackMessage bob aliceId 4
-    get bob =##> \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False
    ackMessage bob aliceId 5
-    6 <- sendMessage bob aliceId "hello too"
-    get bob ##> ("", aliceId, SENT 6)
-    7 <- sendMessage bob aliceId "message 1"
+    get bob =##> \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False
+    ackMessage bob aliceId 6
+    7 <- sendMessage bob aliceId "hello too"
    get bob ##> ("", aliceId, SENT 7)
+    8 <- sendMessage bob aliceId "message 1"
+    get bob ##> ("", aliceId, SENT 8)
    get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False
-    ackMessage alice bobId 6
-    get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False
    ackMessage alice bobId 7
+    get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False
+    ackMessage alice bobId 8
    suspendConnection alice bobId
-    8 <- sendMessage bob aliceId "message 2"
-    get bob ##> ("", aliceId, MERR 8 (SMP AUTH))
+    9 <- sendMessage bob aliceId "message 2"
+    get bob ##> ("", aliceId, MERR 9 (SMP AUTH))
    deleteConnection alice bobId
    liftIO $ noMessages alice "nothing else should be delivered to alice"
  pure ()
@@ -148,6 +150,30 @@ testAsyncBothOffline = do
    exchangeGreetings alice' bobId bob' aliceId
  pure ()

+testAsyncServerOffline :: ATransport -> IO ()
+testAsyncServerOffline t = do
+  alice <- getSMPAgentClient cfg
+  bob <- getSMPAgentClient cfg {dbFile = testDB2}
+  -- create connection and shutdown the server
+  Right (bobId, cReq) <- withSmpServerStoreLogOn t testPort $ \_ ->
+    runExceptT $ createConnection alice SCMInvitation
+  -- connection fails
+  Left (BROKER NETWORK) <- runExceptT $ joinConnection bob cReq "bob's connInfo"
+  ("", bobId1, DOWN) <- get alice
+  bobId1 `shouldBe` bobId
+  -- connection succeeds after server start
+  Right () <- withSmpServerStoreLogOn t testPort $ \_ -> runExceptT $ do
+    ("", bobId2, UP) <- get alice
+    liftIO $ bobId2 `shouldBe` bobId
+    aliceId <- joinConnection bob cReq "bob's connInfo"
+    ("", _, CONF confId "bob's connInfo") <- get alice
+    allowConnection alice bobId confId "alice's connInfo"
+    get alice ##> ("", bobId, CON)
+    get bob ##> ("", aliceId, INFO "alice's connInfo")
+    get bob ##> ("", aliceId, CON)
+    exchangeGreetings alice bobId bob aliceId
+  pure ()
+
 testAsyncHelloTimeout :: IO ()
 testAsyncHelloTimeout = do
  alice <- getSMPAgentClient cfg
@@ -161,11 +187,11 @@ testAsyncHelloTimeout = do

 exchangeGreetings :: AgentClient -> ConnId -> AgentClient -> ConnId -> ExceptT AgentErrorType IO ()
 exchangeGreetings alice bobId bob aliceId = do
-  4 <- sendMessage alice bobId "hello"
-  get alice ##> ("", bobId, SENT 4)
+  5 <- sendMessage alice bobId "hello"
+  get alice ##> ("", bobId, SENT 5)
  get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
-  ackMessage bob aliceId 4
-  5 <- sendMessage bob aliceId "hello too"
-  get bob ##> ("", aliceId, SENT 5)
+  ackMessage bob aliceId 5
+  6 <- sendMessage bob aliceId "hello too"
+  get bob ##> ("", aliceId, SENT 6)
  get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False
-  ackMessage alice bobId 5
+  ackMessage alice bobId 6
@@ -51,7 +51,7 @@ createStore = do
  -- Randomize DB file name to avoid SQLite IO errors supposedly caused by asynchronous
  -- IO operations on multiple similarly named files; error seems to be environment specific
  r <- randomIO :: IO Word32
-  createSQLiteStore (testDB <> show r) 4 Migrations.app
+  createSQLiteStore (testDB <> show r) 4 Migrations.app True

 removeStore :: SQLiteStore -> IO ()
 removeStore store = do
@@ -173,7 +173,9 @@ sndQueue1 =
  SndQueue
    { server = SMPServer "smp.simplex.im" "5223" testKeyHash,
      sndId = "3456",
+      sndPublicKey = Nothing,
      sndPrivateKey = testPrivateSignKey,
+      e2ePubKey = Nothing,
      e2eDhSecret = testDhSecret,
      status = New
    }
@@ -303,7 +305,9 @@ testUpgradeRcvConnToDuplex =
          SndQueue
            { server = SMPServer "smp.simplex.im" "5223" testKeyHash,
              sndId = "2345",
+              sndPublicKey = Nothing,
              sndPrivateKey = testPrivateSignKey,
+              e2ePubKey = Nothing,
              e2eDhSecret = testDhSecret,
              status = New
            }
@@ -393,7 +397,7 @@ mkRcvMsgData internalId internalRcvId externalSndId brokerId internalHash =
            sndMsgId = externalSndId,
            broker = (brokerId, ts)
          },
-      msgType = A_MSG_,
+      msgType = AM_A_MSG_,
      msgBody = hw,
      internalHash,
      externalPrevSndHash = "hash_from_sender"
@@ -422,7 +426,7 @@ mkSndMsgData internalId internalSndId internalHash =
    { internalId,
      internalSndId,
      internalTs = ts,
-      msgType = A_MSG_,
+      msgType = AM_A_MSG_,
      msgBody = hw,
      internalHash,
      prevMsgHash = internalHash
@@ -27,6 +27,7 @@ import Simplex.Messaging.Agent.Server (runSMPAgentBlocking)
 import Simplex.Messaging.Client (SMPClientConfig (..), smpDefaultConfig)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client
+import Simplex.Messaging.Transport.KeepAlive
 import Test.Hspec
 import UnliftIO.Concurrent
 import UnliftIO.Directory
@@ -157,7 +158,7 @@ cfg :: AgentConfig
 cfg =
  defaultAgentConfig
    { tcpPort = agentTestPort,
-      smpServers = L.fromList ["smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5001"],
+      initialSMPServers = L.fromList ["smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5001"],
      tbqSize = 1,
      dbFile = testDB,
      smpCfg =
@@ -174,7 +175,7 @@ cfg =

 withSmpAgentThreadOn_ :: (MonadUnliftIO m, MonadRandom m) => ATransport -> (ServiceName, ServiceName, String) -> m () -> (ThreadId -> m a) -> m a
 withSmpAgentThreadOn_ t (port', smpPort', db') afterProcess =
-  let cfg' = cfg {tcpPort = port', dbFile = db', smpServers = L.fromList [SMPServer "localhost" smpPort' testKeyHash]}
+  let cfg' = cfg {tcpPort = port', dbFile = db', initialSMPServers = L.fromList [SMPServer "localhost" smpPort' testKeyHash]}
   in serverBracket
        (\started -> runSMPAgentBlocking t started cfg')
        afterProcess
@@ -190,7 +191,7 @@ withSmpAgent t = withSmpAgentOn t (agentTestPort, testPort, testDB)

 testSMPAgentClientOn :: (Transport c, MonadUnliftIO m) => ServiceName -> (c -> m a) -> m a
 testSMPAgentClientOn port' client = do
-  runTransportClient agentTestHost port' testKeyHash $ \h -> do
+  runTransportClient agentTestHost port' testKeyHash (Just defaultKeepAliveOpts) $ \h -> do
    line <- liftIO $ getLn h
    if line == "Welcome to SMP agent v" <> B.pack simplexMQVersion
      then client h
@@ -22,6 +22,7 @@ import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Server.StoreLog (openReadStoreLog)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client
+import Simplex.Messaging.Transport.KeepAlive
 import Test.Hspec
 import UnliftIO.Concurrent
 import qualified UnliftIO.Exception as E
@@ -45,7 +46,7 @@ testStoreLogFile = "tests/tmp/smp-server-store.log"

 testSMPClient :: (Transport c, MonadUnliftIO m) => (THandle c -> m a) -> m a
 testSMPClient client =
-  runTransportClient testHost testPort testKeyHash $ \h ->
+  runTransportClient testHost testPort testKeyHash (Just defaultKeepAliveOpts) $ \h ->
    liftIO (runExceptT $ clientHandshake h testKeyHash) >>= \case
      Right th -> client th
      Left e -> error $ show e
@@ -60,6 +61,9 @@ cfg =
      queueIdBytes = 24,
      msgIdBytes = 24,
      storeLog = Nothing,
+      allowNewQueues = True,
+      messageTTL = Just $ 7 * 86400, -- seconds, 7 days
+      expireMessagesInterval = Just 21600_000000, -- microseconds, 6 hours
      caCertificateFile = "tests/fixtures/ca.crt",
      privateKeyFile = "tests/fixtures/server.key",
      certificateFile = "tests/fixtures/server.crt"
@@ -68,16 +72,16 @@ cfg =
 withSmpServerStoreLogOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> ServiceName -> (ThreadId -> m a) -> m a
 withSmpServerStoreLogOn t port' client = do
  s <- liftIO $ openReadStoreLog testStoreLogFile
+  withSmpServerConfigOn t cfg {storeLog = Just s} port' client
+
+withSmpServerConfigOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> ServerConfig -> ServiceName -> (ThreadId -> m a) -> m a
+withSmpServerConfigOn t cfg' port' =
  serverBracket
-    (\started -> runSMPServerBlocking started cfg {transports = [(port', t)], storeLog = Just s})
+    (\started -> runSMPServerBlocking started cfg' {transports = [(port', t)]})
    (pure ())
-    client

 withSmpServerThreadOn :: (MonadUnliftIO m, MonadRandom m) => ATransport -> ServiceName -> (ThreadId -> m a) -> m a
-withSmpServerThreadOn t port' =
-  serverBracket
-    (\started -> runSMPServerBlocking started cfg {transports = [(port', t)]})
-    (pure ())
+withSmpServerThreadOn t = withSmpServerConfigOn t cfg

 serverBracket :: MonadUnliftIO m => (TMVar Bool -> m ()) -> m () -> (ThreadId -> m a) -> m a
 serverBracket process afterProcess f = do
@@ -9,7 +9,7 @@

 module ServerTests where

-import Control.Concurrent (ThreadId, killThread)
+import Control.Concurrent (ThreadId, killThread, threadDelay)
 import Control.Concurrent.STM
 import Control.Exception (SomeException, try)
 import Control.Monad.Except (forM, forM_, runExceptT)
@@ -21,6 +21,7 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
+import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Transport
 import System.Directory (removeFile)
 import System.TimeIt (timeItT)
@@ -29,18 +30,23 @@ import Test.HUnit
 import Test.Hspec

 serverTests :: ATransport -> Spec
-serverTests t = do
+serverTests t@(ATransport t') = do
  describe "SMP syntax" $ syntaxTests t
  describe "SMP queues" $ do
    describe "NEW and KEY commands, SEND messages" $ testCreateSecure t
    describe "NEW, OFF and DEL commands, SEND messages" $ testCreateDelete t
    describe "Stress test" $ stressTest t
+    describe "allowNewQueues setting" $ testAllowNewQueues t'
  describe "SMP messages" $ do
    describe "duplex communication over 2 SMP connections" $ testDuplex t
    describe "switch subscription to another TCP connection" $ testSwitchSub t
  describe "Store log" $ testWithStoreLog t
  describe "Timing of AUTH error" $ testTiming t
  describe "Message notifications" $ testMessageNotifications t
+  describe "Message expiration" $ do
+    testMsgExpireOnSend t'
+    testMsgExpireOnInterval t'
+    testMsgNOTExpireOnInterval t'

 pattern Resp :: CorrId -> QueueId -> BrokerMsg -> SignedTransmission BrokerMsg
 pattern Resp corrId queueId command <- (_, _, (corrId, queueId, Right command))
@@ -204,6 +210,16 @@ stressTest (ATransport t) =
      closeConnection $ connection h2
      subscribeQueues h3

+testAllowNewQueues :: forall c. Transport c => TProxy c -> Spec
+testAllowNewQueues t =
+  it "should prohibit creating new queues with allowNewQueues = False" $ do
+    withSmpServerConfigOn (ATransport t) cfg {allowNewQueues = False} testPort $ \_ ->
+      testSMPClient @c $ \h -> do
+        (rPub, rKey) <- C.generateSignatureKeyPair C.SEd448
+        (dhPub, _ :: C.PrivateKeyX25519) <- C.generateKeyPair'
+        Resp "abcd" "" (ERR AUTH) <- signSendRecv h rKey ("abcd", "", NEW rPub dhPub)
+        pure ()
+
 testDuplex :: ATransport -> Spec
 testDuplex (ATransport t) =
  it "should create 2 simplex connections and exchange messages" $
@@ -466,6 +482,56 @@ testMessageNotifications (ATransport t) =
        Nothing -> return ()
        Just _ -> error "nothing else should be delivered to the 1st notifier's TCP connection"

+testMsgExpireOnSend :: forall c. Transport c => TProxy c -> Spec
+testMsgExpireOnSend t =
+  it "should expire messages that are not received before messageTTL on SEND" $ do
+    (sPub, sKey) <- C.generateSignatureKeyPair C.SEd25519
+    withSmpServerConfigOn (ATransport t) cfg {messageTTL = Just 1} testPort $ \_ ->
+      testSMPClient @c $ \sh -> do
+        (sId, rId, rKey, dhShared) <- testSMPClient @c $ \rh -> createAndSecureQueue rh sPub
+        let dec nonce = C.cbDecrypt dhShared (C.cbNonce nonce)
+        Resp "1" _ OK <- signSendRecv sh sKey ("1", sId, SEND "hello (should expire)")
+        threadDelay 2500000
+        Resp "2" _ OK <- signSendRecv sh sKey ("2", sId, SEND "hello (should NOT expire)")
+        testSMPClient @c $ \rh -> do
+          Resp "3" _ (MSG mId _ msg) <- signSendRecv rh rKey ("3", rId, SUB)
+          (dec mId msg, Right "hello (should NOT expire)") #== "delivered"
+          1000 `timeout` tGet @BrokerMsg rh >>= \case
+            Nothing -> return ()
+            Just _ -> error "nothing else should be delivered"
+
+testMsgExpireOnInterval :: forall c. Transport c => TProxy c -> Spec
+testMsgExpireOnInterval t =
+  it "should expire messages that are not received before messageTTL after expiry interval" $ do
+    (sPub, sKey) <- C.generateSignatureKeyPair C.SEd25519
+    withSmpServerConfigOn (ATransport t) cfg {messageTTL = Just 1, expireMessagesInterval = Just 1000000} testPort $ \_ ->
+      testSMPClient @c $ \sh -> do
+        (sId, rId, rKey, _) <- testSMPClient @c $ \rh -> createAndSecureQueue rh sPub
+        Resp "1" _ OK <- signSendRecv sh sKey ("1", sId, SEND "hello (should expire)")
+        threadDelay 2500000
+        testSMPClient @c $ \rh -> do
+          Resp "2" _ OK <- signSendRecv rh rKey ("2", rId, SUB)
+          1000 `timeout` tGet @BrokerMsg rh >>= \case
+            Nothing -> return ()
+            Just _ -> error "nothing should be delivered"
+
+testMsgNOTExpireOnInterval :: forall c. Transport c => TProxy c -> Spec
+testMsgNOTExpireOnInterval t =
+  it "should NOT expire messages that are not received before messageTTL if expiry interval is not set" $ do
+    (sPub, sKey) <- C.generateSignatureKeyPair C.SEd25519
+    withSmpServerConfigOn (ATransport t) cfg {messageTTL = Just 1, expireMessagesInterval = Nothing} testPort $ \_ ->
+      testSMPClient @c $ \sh -> do
+        (sId, rId, rKey, dhShared) <- testSMPClient @c $ \rh -> createAndSecureQueue rh sPub
+        let dec nonce = C.cbDecrypt dhShared (C.cbNonce nonce)
+        Resp "1" _ OK <- signSendRecv sh sKey ("1", sId, SEND "hello (should NOT expire)")
+        threadDelay 2500000
+        testSMPClient @c $ \rh -> do
+          Resp "2" _ (MSG mId _ msg) <- signSendRecv rh rKey ("2", rId, SUB)
+          (dec mId msg, Right "hello (should NOT expire)") #== "delivered"
+          1000 `timeout` tGet @BrokerMsg rh >>= \case
+            Nothing -> return ()
+            Just _ -> error "nothing else should be delivered"
+
 samplePubKey :: C.APublicVerifyKey
 samplePubKey = C.APublicVerifyKey C.SEd25519 "MCowBQYDK2VwAyEAfAOflyvbJv1fszgzkQ6buiZJVgSpQWsucXq7U6zjMgY="
Author	SHA1	Message	Date
Evgeny Poberezkin	0a2cad2745	version 1.1.0 (#358 ) * version 1.1.0 * increase test delays to stabilize	2022-04-21 15:00:30 +01:00
JRoberts	12caa328a8	SMP confirmation timeout recovery rfc (#357 )	2022-04-20 22:17:51 +01:00
Evgeny Poberezkin	68f52c47dd	message TTL and periodic expiration; setting to prevent creation of the new queues (#355 ) * message TTL and periodic expiration; setting to prevent creation of the new queues * fix expiration interval	2022-04-20 08:14:26 +01:00
Evgeny Poberezkin	f84439f79d	function to run TCP server without TLS (#354 )	2022-04-18 08:56:08 +01:00
Evgeny Poberezkin	d38303d5f1	fix END from disconnected clients incorrectly removing the subscriptions, remove previous PING changes that attempted to solve the problem of lost subscriptions (#351 ) * Revert "increase PING timeout, add PING failure count" This reverts commit `a89e019bb0`. * Revert "PING error now throws error to restart SMPClient for more reliable re-connection (#342)" This reverts commit `62acbc4ad4`. * only process END from the currently active client * log ignored END * make PING sent every 15 sec and destroying connection on failure (for testing - to be reverted) * make removing subscription atomic too * Revert "make PING sent every 15 sec and destroying connection on failure (for testing - to be reverted)" This reverts commit `5520b318a2`. * refactor, aggressive PING settings for testing * revert PING breaking connection	2022-04-12 11:35:28 +01:00
Evgeny Poberezkin	6382367ec0	typo	2022-04-11 10:30:34 +01:00
Evgeny Poberezkin	a89e019bb0	increase PING timeout, add PING failure count	2022-04-11 10:24:02 +01:00
Evgeny Poberezkin	a638486bb3	add swift flag to package.yaml	2022-04-08 15:42:22 +01:00
Evgeny Poberezkin	390e99ca95	rename cabal flag	2022-04-08 08:51:07 +01:00
Moritz Angermann	ac7ffc413e	Add Flag (#349 ) * Update Parsers.hs * Update simplexmq.cabal	2022-04-07 08:44:37 +01:00
JRoberts	3ba1926b1e	remove manual vacuum (#346 )	2022-04-01 17:16:18 +04:00
JRoberts	c62730fe4c	Revert "catch db connection error" This reverts commit `1815e81c2e`.	2022-04-01 16:39:32 +04:00
JRoberts	bb99fdaaa2	catch db connection error (#345 )	2022-04-01 09:52:13 +01:00
JRoberts	a6ec93c38e	vacuum in each connection to enable auto-vacuum (#344 )	2022-03-31 15:26:13 +04:00
JRoberts	33f822d72c	add pragmas and vacuum db (#343 )	2022-03-31 14:51:59 +04:00
Evgeny Poberezkin	b6e87e4a3e	increase TCP timeout to 5 sec	2022-03-29 11:18:49 +01:00
Evgeny Poberezkin	62acbc4ad4	PING error now throws error to restart SMPClient for more reliable re-connection (#342 )	2022-03-29 13:21:21 +04:00
Evgeny Poberezkin	cd22e06b3a	use TMap for subscription maps (#341 ) * use TMap for subscription maps * refactor * correction	2022-03-28 18:49:17 +01:00
Evgeny Poberezkin	6ef6bedc03	refactor/optimize server queue/message store (#340 ) * refactor/optimize server queue/message store * change fst to pattern match * server store - wrap QueueRec into TVar	2022-03-28 10:29:21 +01:00
Evgeny Poberezkin	c380431b94	resubscribe concurrently when subscription is resumed (#339 ) * resubscribe concurrently when subscription is resumed * use strict modifyTVar, refactor with TMap * add inline * refactor	2022-03-28 07:30:29 +01:00
JRoberts	337d5bdfe7	add "SMP server on Linux" section to README (#337 )	2022-03-26 19:57:18 +04:00
Evgeny Poberezkin	800581b2bf	core: tcp keep-alive parameters for windows (#336 )	2022-03-26 13:11:29 +00:00
Evgeny Poberezkin	14d76a1582	fix keep-alive options to work on mobile (#331 )	2022-03-22 08:02:43 +00:00
Evgeny Poberezkin	a37b24a8c2	use TCP keep-alive instead of SMP protocol PING (#330 ) * use TCP keep-alive instead of SMP protocol PING * update header files * use CInt * use Int again * use network-3.1.2.7 * use https in cabal.project * confitional keep-alive parameters to work on mac * pass keep-alive opts via client/agent options * remove space	2022-03-21 16:13:34 +00:00
John Roberts	5c6ec96d64	make smp servers configurable for running agent (#326 )	2022-03-10 10:49:22 +04:00
Evgeny Poberezkin	7a19ab224b	make sending confirmation asynchronous (#327 ) * make sending confirmation asynchronous * confirm first queue synchronously (on JOIN, and delete connection on failure), second queue asynchronously (from REPLY msg) * test to retry JOIN when the first attempt fails * process confirmation delivery errors	2022-03-02 15:52:45 +00:00
Efim Poberezkin	7a611bed5a	configure systemd service restart and increase file descriptors limit for Linode script (#320 )	2022-02-26 18:42:16 +04:00
Efim Poberezkin	a406159ed4	1.0.3	2022-02-26 18:11:07 +04:00
Efim Poberezkin	14780a47d6	prepare v1.0.3 (#324 )	2022-02-26 18:07:11 +04:00
Evgeny Poberezkin	4a66f68c55	reduce server message queue quota to 128 messages (#323 )	2022-02-26 17:49:21 +04:00
Evgeny Poberezkin	d1e6147adf	encode all nullary constructors as objects for sumTypeJSON options	2022-02-25 18:50:04 +00:00
Evgeny Poberezkin	7d1fdadef0	reduce connection handshake expiration to 2 days	2022-02-25 10:57:23 +00:00
Evgeny Poberezkin	09e2e75c42	increase queue sizes (#322 ) * increase queue sizes * remove TODO	2022-02-25 09:02:52 +00:00
Evgeny Poberezkin	dff5cad1be	make CPP condition depend on HOST OS (#321 )	2022-02-18 08:12:08 +00:00
Evgeny Poberezkin	c1f5f9d846	flip condition for JSON encoding to make android (tagged objects) default (#319 )	2022-02-16 22:38:15 +00:00
Evgeny Poberezkin	9c4778b129	update condition in cabal	2022-02-16 17:28:14 +00:00
Evgeny Poberezkin	137afb68fe	conditional JSON options for iOS/Android (#318 ) * conditional JSON options for iOS/Android * update cabal file	2022-02-16 08:58:35 +00:00
Efim Poberezkin	229e2607d7	remove redundant asynchronous connection code (#317 )	2022-02-14 21:00:14 +04:00
Efim PoberezkinandEvgeny Poberezkin	c380c79560	make new smp client attempt to reconnect on network error (#315 ) * make new smp client attempt to reconnect on network error * reconnect async wip * remove comments * trigger tryReconnectClient * add subscription before trying to take smp client * move atomically modify asyncClients * remove logs * corrections * don't put error into new smp client TMVar * UnliftIO async * re-work * test for connection to the server that was initially down * fix tests * additionally test that bob receives the message after re-connect * make test independent of event order Co-authored-by: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com>	2022-02-10 14:48:11 +00:00
Efim Poberezkin	c9994c3a2c	add "yes to migrations" option (#316 )	2022-02-07 11:51:39 +04:00