5.4.0.1

* server: add control port commands for clients and ghc threads (#835)

* Add stats-rts control query

With supporting ghc-options that would provide the data.

* Add CPSkip command

Allows spamming empty lines a few times to clean up the view.

* server: Add CP commands to enumerate clients and threads

* style

---------

Co-authored-by: Evgeny Poberezkin <[email protected]>

* use base64 encoding for session ID

* fromMaybe

* whitespace

---------

Co-authored-by: Alexander Bondarenko <[email protected]>

.github/workflows/build.yml

@@ -23,13 +23,13 @@ jobs:
             platform_name: 22_04-x86-64
     steps:
       - name: Clone project
-        uses: actions/checkout@v2
+        uses: actions/checkout@v3
 
       - name: Setup Haskell
-        uses: haskell/actions/setup@v1
+        uses: haskell-actions/setup@v2
         with:
-          ghc-version: "8.10.7"
-          cabal-version: "latest"
+          ghc-version: "9.6.2"
+          cabal-version: "3.10.1.0"
 
       - name: Cache dependencies
         uses: actions/cache@v2

CHANGELOG.md

@@ -1,3 +1,12 @@
+# 5.3.0
+
+Agent:
+- improve performance, track slow database queries.
+- support delievery receipts.
+
+SMP server:
+- control port
+
 # 5.2.0 (NTF server 1.5.0)
 
 Agent:

apps/smp-server/Main.hs

@@ -1,13 +1,17 @@
+{-# LANGUAGE LambdaCase #-}
+
 module Main where
 
 import Control.Logger.Simple
+import Data.Maybe
 import Simplex.Messaging.Server.Main
+import System.Environment
 
-cfgPath :: FilePath
-cfgPath = "/etc/opt/simplex"
+defaultCfgPath :: FilePath
+defaultCfgPath = "/etc/opt/simplex"
 
-logPath :: FilePath
-logPath = "/var/opt/simplex"
+defaultLogPath :: FilePath
+defaultLogPath = "/var/opt/simplex"
 
 logCfg :: LogConfig
 logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
@@ -15,4 +19,9 @@ logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
 main :: IO ()
 main = do
   setLogLevel LogDebug
+  cfgPath <- getEnvPath "SMP_SERVER_CFG_PATH" defaultCfgPath
+  logPath <- getEnvPath "SMP_SERVER_LOG_PATH" defaultLogPath
   withGlobalLogging logCfg $ smpServerCLI cfgPath logPath
+    
+getEnvPath :: String -> FilePath -> IO FilePath
+getEnvPath name def = maybe def (\case "" -> def; f -> f) <$> lookupEnv name

cabal.project

@@ -2,11 +2,14 @@ packages: .
 -- packages: . ../direct-sqlcipher ../sqlcipher-simple
 -- packages: . ../hs-socks
 -- packages: . ../http2
+-- packages: . ../network-transport
+
+with-compiler: ghc-9.6.2
 
 source-repository-package
     type: git
     location: https://github.com/simplex-chat/aeson.git
-    tag: 3eb66f9a68f103b5f1489382aad89f5712a64db7
+    tag: 68330dce8208173c6acf5f62b23acb500ab5d873
 
 source-repository-package
     type: git
@@ -21,9 +24,14 @@ source-repository-package
 source-repository-package
     type: git
     location: https://github.com/simplex-chat/direct-sqlcipher.git
-    tag: 34309410eb2069b029b8fc1872deb1e0db123294
+    tag: f814ee68b16a9447fbb467ccc8f29bdd3546bfd9
 
 source-repository-package
     type: git
     location: https://github.com/simplex-chat/sqlcipher-simple.git
-    tag: 5e154a2aeccc33ead6c243ec07195ab673137221
+    tag: a46bd361a19376c5211f1058908fc0ae6bf42446
+
+source-repository-package
+    type: git
+    location: https://github.com/simplex-chat/network-transport.git
+    tag: 0013798272a683e35ca38d2fdaf480942311fba8

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.3.0.0
+version: 5.4.0.1
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -22,7 +22,7 @@ extra-source-files:
   - CHANGELOG.md
 
 dependencies:
-  - aeson == 2.0.*
+  - aeson == 2.2.*
   - ansi-terminal >= 0.10 && < 0.12
   - asn1-encoding == 0.9.*
   - asn1-types == 0.3.*
@@ -30,12 +30,12 @@ dependencies:
   - attoparsec == 0.14.*
   - base >= 4.14 && < 5
   - base64-bytestring >= 1.0 && < 1.3
-  - bytestring == 0.10.*
+  - bytestring == 0.11.*
   - case-insensitive == 1.2.*
   - composition == 1.0.*
   - constraints >= 0.12 && < 0.14
   - containers == 0.6.*
-  - cryptonite >= 0.27 && < 0.30
+  - cryptonite == 0.30.*
   - cryptostore == 0.2.*
   - data-default == 0.7.*
   - direct-sqlcipher == 2.3.*
@@ -43,14 +43,14 @@ dependencies:
   - filepath == 1.4.*
   - http-types == 0.12.*
   - http2 == 4.1.*
-  - generic-random >= 1.3 && < 1.5
+  - generic-random == 1.5.*
   - ini == 0.4.1
   - iproute == 1.7.*
   - iso8601-time == 0.1.*
-  - memory == 0.15.*
-  - mtl == 2.2.*
+  - memory == 0.18.*
+  - mtl == 2.3.*
   - network >= 3.1.2.7 && < 3.2
-  - network-transport == 0.5.4
+  - network-transport == 0.5.6
   - optparse-applicative >= 0.15 && < 0.17
   - QuickCheck == 2.14.*
   - process == 1.6.*
@@ -59,14 +59,14 @@ dependencies:
   - socks == 0.6.*
   - sqlcipher-simple == 0.4.*
   - stm == 2.5.*
-  - template-haskell == 2.16.*
+  - template-haskell == 2.20.*
   - temporary == 1.3.*
-  - text == 1.2.*
+  - text == 2.0.*
   - time == 1.9.*
   - time-compat == 1.9.*
   - time-manager == 0.0.*
   - tls >= 1.6.0 && < 1.7
-  - transformers == 0.5.*
+  - transformers == 0.6.*
   - unliftio == 0.2.*
   - unliftio-core == 0.2.*
   - websockets == 0.12.*
@@ -97,6 +97,7 @@ executables:
       - simplexmq
     ghc-options:
       - -threaded
+      - -rtsopts
 
   ntf-server:
     source-dirs: apps/ntf-server
@@ -105,6 +106,7 @@ executables:
       - simplexmq
     ghc-options:
       - -threaded
+      - -rtsopts
 
   xftp-server:
     source-dirs: apps/xftp-server
@@ -113,6 +115,7 @@ executables:
       - simplexmq
     ghc-options:
       - -threaded
+      - -rtsopts
 
   smp-agent:
     source-dirs: apps/smp-agent
@@ -121,6 +124,7 @@ executables:
       - simplexmq
     ghc-options:
       - -threaded
+      - -rtsopts
 
   xftp:
     source-dirs: apps/xftp
@@ -129,6 +133,7 @@ executables:
       - simplexmq
     ghc-options:
       - -threaded
+      - -rtsopts
 
 tests:
   simplexmq-test:
@@ -137,8 +142,8 @@ tests:
     dependencies:
       - simplexmq
       - deepseq == 1.4.*
-      - hspec == 2.7.*
-      - hspec-core == 2.7.*
+      - hspec == 2.11.*
+      - hspec-core == 2.11.*
       - HUnit == 1.6.*
       - QuickCheck == 2.14.*
       - silently == 1.2.*

protocol/simplex-messaging.md

@@ -347,7 +347,7 @@ Simplex messaging server implementations MUST NOT create, store or send to any o
 
 - Snapshots of the database they use to store queues and messages (instead simplex messaging clients must manage redundancy by using more than one simplex messaging server). In-memory persistence is recommended.
 
-- Any other information that may compromise privacy or [forward secrecy][4] of communication between clients using simplex messaging servers.
+- Any other information that may compromise privacy or [forward secrecy][4] of communication between clients using simplex messaging servers (the servers cannot compromise forward secrecy of any application layer protocol, such as double ratchet).
 
 ## Message delivery notifications
 

simplexmq.cabal

@@ -1,11 +1,11 @@
 cabal-version: 1.12
 
--- This file has been generated from package.yaml by hpack version 0.35.1.
+-- This file has been generated from package.yaml by hpack version 0.35.2.
 --
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        5.3.0.0
+version:        5.4.0.1
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -63,6 +63,7 @@ library
       Simplex.Messaging.Agent.Store
       Simplex.Messaging.Agent.Store.SQLite
       Simplex.Messaging.Agent.Store.SQLite.Common
+      Simplex.Messaging.Agent.Store.SQLite.DB
       Simplex.Messaging.Agent.Store.SQLite.Migrations
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys
@@ -85,6 +86,9 @@ library
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230531_switch_status
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230615_ratchet_sync
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230701_delivery_receipts
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230720_delete_expired_messages
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230722_indexes
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230814_indexes
       Simplex.Messaging.Agent.TAsyncs
       Simplex.Messaging.Agent.TRcvQueues
       Simplex.Messaging.Client
@@ -138,7 +142,7 @@ library
   ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns
   build-depends:
       QuickCheck ==2.14.*
-    , aeson ==2.0.*
+    , aeson ==2.2.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
@@ -146,27 +150,27 @@ library
     , attoparsec ==0.14.*
     , base >=4.14 && <5
     , base64-bytestring >=1.0 && <1.3
-    , bytestring ==0.10.*
+    , bytestring ==0.11.*
     , case-insensitive ==1.2.*
     , composition ==1.0.*
     , constraints >=0.12 && <0.14
     , containers ==0.6.*
-    , cryptonite >=0.27 && <0.30
+    , cryptonite ==0.30.*
     , cryptostore ==0.2.*
     , data-default ==0.7.*
     , direct-sqlcipher ==2.3.*
     , directory ==1.3.*
     , filepath ==1.4.*
-    , generic-random >=1.3 && <1.5
+    , generic-random ==1.5.*
     , http-types ==0.12.*
     , http2 ==4.1.*
     , ini ==0.4.1
     , iproute ==1.7.*
     , iso8601-time ==0.1.*
-    , memory ==0.15.*
-    , mtl ==2.2.*
+    , memory ==0.18.*
+    , mtl ==2.3.*
     , network >=3.1.2.7 && <3.2
-    , network-transport ==0.5.4
+    , network-transport ==0.5.6
     , optparse-applicative >=0.15 && <0.17
     , process ==1.6.*
     , random >=1.1 && <1.3
@@ -174,14 +178,14 @@ library
     , socks ==0.6.*
     , sqlcipher-simple ==0.4.*
     , stm ==2.5.*
-    , template-haskell ==2.16.*
+    , template-haskell ==2.20.*
     , temporary ==1.3.*
-    , text ==1.2.*
+    , text ==2.0.*
     , time ==1.9.*
     , time-compat ==1.9.*
     , time-manager ==0.0.*
     , tls >=1.6.0 && <1.7
-    , transformers ==0.5.*
+    , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
     , websockets ==0.12.*
@@ -199,10 +203,10 @@ executable ntf-server
       Paths_simplexmq
   hs-source-dirs:
       apps/ntf-server
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded -rtsopts
   build-depends:
       QuickCheck ==2.14.*
-    , aeson ==2.0.*
+    , aeson ==2.2.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
@@ -210,27 +214,27 @@ executable ntf-server
     , attoparsec ==0.14.*
     , base >=4.14 && <5
     , base64-bytestring >=1.0 && <1.3
-    , bytestring ==0.10.*
+    , bytestring ==0.11.*
     , case-insensitive ==1.2.*
     , composition ==1.0.*
     , constraints >=0.12 && <0.14
     , containers ==0.6.*
-    , cryptonite >=0.27 && <0.30
+    , cryptonite ==0.30.*
     , cryptostore ==0.2.*
     , data-default ==0.7.*
     , direct-sqlcipher ==2.3.*
     , directory ==1.3.*
     , filepath ==1.4.*
-    , generic-random >=1.3 && <1.5
+    , generic-random ==1.5.*
     , http-types ==0.12.*
     , http2 ==4.1.*
     , ini ==0.4.1
     , iproute ==1.7.*
     , iso8601-time ==0.1.*
-    , memory ==0.15.*
-    , mtl ==2.2.*
+    , memory ==0.18.*
+    , mtl ==2.3.*
     , network >=3.1.2.7 && <3.2
-    , network-transport ==0.5.4
+    , network-transport ==0.5.6
     , optparse-applicative >=0.15 && <0.17
     , process ==1.6.*
     , random >=1.1 && <1.3
@@ -239,14 +243,14 @@ executable ntf-server
     , socks ==0.6.*
     , sqlcipher-simple ==0.4.*
     , stm ==2.5.*
-    , template-haskell ==2.16.*
+    , template-haskell ==2.20.*
     , temporary ==1.3.*
-    , text ==1.2.*
+    , text ==2.0.*
     , time ==1.9.*
     , time-compat ==1.9.*
     , time-manager ==0.0.*
     , tls >=1.6.0 && <1.7
-    , transformers ==0.5.*
+    , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
     , websockets ==0.12.*
@@ -264,10 +268,10 @@ executable smp-agent
       Paths_simplexmq
   hs-source-dirs:
       apps/smp-agent
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded -rtsopts
   build-depends:
       QuickCheck ==2.14.*
-    , aeson ==2.0.*
+    , aeson ==2.2.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
@@ -275,27 +279,27 @@ executable smp-agent
     , attoparsec ==0.14.*
     , base >=4.14 && <5
     , base64-bytestring >=1.0 && <1.3
-    , bytestring ==0.10.*
+    , bytestring ==0.11.*
     , case-insensitive ==1.2.*
     , composition ==1.0.*
     , constraints >=0.12 && <0.14
     , containers ==0.6.*
-    , cryptonite >=0.27 && <0.30
+    , cryptonite ==0.30.*
     , cryptostore ==0.2.*
     , data-default ==0.7.*
     , direct-sqlcipher ==2.3.*
     , directory ==1.3.*
     , filepath ==1.4.*
-    , generic-random >=1.3 && <1.5
+    , generic-random ==1.5.*
     , http-types ==0.12.*
     , http2 ==4.1.*
     , ini ==0.4.1
     , iproute ==1.7.*
     , iso8601-time ==0.1.*
-    , memory ==0.15.*
-    , mtl ==2.2.*
+    , memory ==0.18.*
+    , mtl ==2.3.*
     , network >=3.1.2.7 && <3.2
-    , network-transport ==0.5.4
+    , network-transport ==0.5.6
     , optparse-applicative >=0.15 && <0.17
     , process ==1.6.*
     , random >=1.1 && <1.3
@@ -304,14 +308,14 @@ executable smp-agent
     , socks ==0.6.*
     , sqlcipher-simple ==0.4.*
     , stm ==2.5.*
-    , template-haskell ==2.16.*
+    , template-haskell ==2.20.*
     , temporary ==1.3.*
-    , text ==1.2.*
+    , text ==2.0.*
     , time ==1.9.*
     , time-compat ==1.9.*
     , time-manager ==0.0.*
     , tls >=1.6.0 && <1.7
-    , transformers ==0.5.*
+    , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
     , websockets ==0.12.*
@@ -329,10 +333,10 @@ executable smp-server
       Paths_simplexmq
   hs-source-dirs:
       apps/smp-server
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded -rtsopts
   build-depends:
       QuickCheck ==2.14.*
-    , aeson ==2.0.*
+    , aeson ==2.2.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
@@ -340,27 +344,27 @@ executable smp-server
     , attoparsec ==0.14.*
     , base >=4.14 && <5
     , base64-bytestring >=1.0 && <1.3
-    , bytestring ==0.10.*
+    , bytestring ==0.11.*
     , case-insensitive ==1.2.*
     , composition ==1.0.*
     , constraints >=0.12 && <0.14
     , containers ==0.6.*
-    , cryptonite >=0.27 && <0.30
+    , cryptonite ==0.30.*
     , cryptostore ==0.2.*
     , data-default ==0.7.*
     , direct-sqlcipher ==2.3.*
     , directory ==1.3.*
     , filepath ==1.4.*
-    , generic-random >=1.3 && <1.5
+    , generic-random ==1.5.*
     , http-types ==0.12.*
     , http2 ==4.1.*
     , ini ==0.4.1
     , iproute ==1.7.*
     , iso8601-time ==0.1.*
-    , memory ==0.15.*
-    , mtl ==2.2.*
+    , memory ==0.18.*
+    , mtl ==2.3.*
     , network >=3.1.2.7 && <3.2
-    , network-transport ==0.5.4
+    , network-transport ==0.5.6
     , optparse-applicative >=0.15 && <0.17
     , process ==1.6.*
     , random >=1.1 && <1.3
@@ -369,14 +373,14 @@ executable smp-server
     , socks ==0.6.*
     , sqlcipher-simple ==0.4.*
     , stm ==2.5.*
-    , template-haskell ==2.16.*
+    , template-haskell ==2.20.*
     , temporary ==1.3.*
-    , text ==1.2.*
+    , text ==2.0.*
     , time ==1.9.*
     , time-compat ==1.9.*
     , time-manager ==0.0.*
     , tls >=1.6.0 && <1.7
-    , transformers ==0.5.*
+    , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
     , websockets ==0.12.*
@@ -394,10 +398,10 @@ executable xftp
       Paths_simplexmq
   hs-source-dirs:
       apps/xftp
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded -rtsopts
   build-depends:
       QuickCheck ==2.14.*
-    , aeson ==2.0.*
+    , aeson ==2.2.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
@@ -405,27 +409,27 @@ executable xftp
     , attoparsec ==0.14.*
     , base >=4.14 && <5
     , base64-bytestring >=1.0 && <1.3
-    , bytestring ==0.10.*
+    , bytestring ==0.11.*
     , case-insensitive ==1.2.*
     , composition ==1.0.*
     , constraints >=0.12 && <0.14
     , containers ==0.6.*
-    , cryptonite >=0.27 && <0.30
+    , cryptonite ==0.30.*
     , cryptostore ==0.2.*
     , data-default ==0.7.*
     , direct-sqlcipher ==2.3.*
     , directory ==1.3.*
     , filepath ==1.4.*
-    , generic-random >=1.3 && <1.5
+    , generic-random ==1.5.*
     , http-types ==0.12.*
     , http2 ==4.1.*
     , ini ==0.4.1
     , iproute ==1.7.*
     , iso8601-time ==0.1.*
-    , memory ==0.15.*
-    , mtl ==2.2.*
+    , memory ==0.18.*
+    , mtl ==2.3.*
     , network >=3.1.2.7 && <3.2
-    , network-transport ==0.5.4
+    , network-transport ==0.5.6
     , optparse-applicative >=0.15 && <0.17
     , process ==1.6.*
     , random >=1.1 && <1.3
@@ -434,14 +438,14 @@ executable xftp
     , socks ==0.6.*
     , sqlcipher-simple ==0.4.*
     , stm ==2.5.*
-    , template-haskell ==2.16.*
+    , template-haskell ==2.20.*
     , temporary ==1.3.*
-    , text ==1.2.*
+    , text ==2.0.*
     , time ==1.9.*
     , time-compat ==1.9.*
     , time-manager ==0.0.*
     , tls >=1.6.0 && <1.7
-    , transformers ==0.5.*
+    , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
     , websockets ==0.12.*
@@ -459,10 +463,10 @@ executable xftp-server
       Paths_simplexmq
   hs-source-dirs:
       apps/xftp-server
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -threaded -rtsopts
   build-depends:
       QuickCheck ==2.14.*
-    , aeson ==2.0.*
+    , aeson ==2.2.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
@@ -470,27 +474,27 @@ executable xftp-server
     , attoparsec ==0.14.*
     , base >=4.14 && <5
     , base64-bytestring >=1.0 && <1.3
-    , bytestring ==0.10.*
+    , bytestring ==0.11.*
     , case-insensitive ==1.2.*
     , composition ==1.0.*
     , constraints >=0.12 && <0.14
     , containers ==0.6.*
-    , cryptonite >=0.27 && <0.30
+    , cryptonite ==0.30.*
     , cryptostore ==0.2.*
     , data-default ==0.7.*
     , direct-sqlcipher ==2.3.*
     , directory ==1.3.*
     , filepath ==1.4.*
-    , generic-random >=1.3 && <1.5
+    , generic-random ==1.5.*
     , http-types ==0.12.*
     , http2 ==4.1.*
     , ini ==0.4.1
     , iproute ==1.7.*
     , iso8601-time ==0.1.*
-    , memory ==0.15.*
-    , mtl ==2.2.*
+    , memory ==0.18.*
+    , mtl ==2.3.*
     , network >=3.1.2.7 && <3.2
-    , network-transport ==0.5.4
+    , network-transport ==0.5.6
     , optparse-applicative >=0.15 && <0.17
     , process ==1.6.*
     , random >=1.1 && <1.3
@@ -499,14 +503,14 @@ executable xftp-server
     , socks ==0.6.*
     , sqlcipher-simple ==0.4.*
     , stm ==2.5.*
-    , template-haskell ==2.16.*
+    , template-haskell ==2.20.*
     , temporary ==1.3.*
-    , text ==1.2.*
+    , text ==2.0.*
     , time ==1.9.*
     , time-compat ==1.9.*
     , time-manager ==0.0.*
     , tls >=1.6.0 && <1.7
-    , transformers ==0.5.*
+    , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
     , websockets ==0.12.*
@@ -531,6 +535,7 @@ test-suite simplexmq-test
       AgentTests.SchemaDump
       AgentTests.SQLiteTests
       CLITests
+      CoreTests.BatchingTests
       CoreTests.CryptoTests
       CoreTests.EncodingTests
       CoreTests.ProtocolErrorTests
@@ -554,7 +559,7 @@ test-suite simplexmq-test
   build-depends:
       HUnit ==1.6.*
     , QuickCheck ==2.14.*
-    , aeson ==2.0.*
+    , aeson ==2.2.*
     , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
@@ -562,31 +567,31 @@ test-suite simplexmq-test
     , attoparsec ==0.14.*
     , base >=4.14 && <5
     , base64-bytestring >=1.0 && <1.3
-    , bytestring ==0.10.*
+    , bytestring ==0.11.*
     , case-insensitive ==1.2.*
     , composition ==1.0.*
     , constraints >=0.12 && <0.14
     , containers ==0.6.*
-    , cryptonite >=0.27 && <0.30
+    , cryptonite ==0.30.*
     , cryptostore ==0.2.*
     , data-default ==0.7.*
     , deepseq ==1.4.*
     , direct-sqlcipher ==2.3.*
     , directory ==1.3.*
     , filepath ==1.4.*
-    , generic-random >=1.3 && <1.5
-    , hspec ==2.7.*
-    , hspec-core ==2.7.*
+    , generic-random ==1.5.*
+    , hspec ==2.11.*
+    , hspec-core ==2.11.*
     , http-types ==0.12.*
     , http2 ==4.1.*
     , ini ==0.4.1
     , iproute ==1.7.*
     , iso8601-time ==0.1.*
     , main-tester ==0.2.*
-    , memory ==0.15.*
-    , mtl ==2.2.*
+    , memory ==0.18.*
+    , mtl ==2.3.*
     , network >=3.1.2.7 && <3.2
-    , network-transport ==0.5.4
+    , network-transport ==0.5.6
     , optparse-applicative >=0.15 && <0.17
     , process ==1.6.*
     , random >=1.1 && <1.3
@@ -596,15 +601,15 @@ test-suite simplexmq-test
     , socks ==0.6.*
     , sqlcipher-simple ==0.4.*
     , stm ==2.5.*
-    , template-haskell ==2.16.*
+    , template-haskell ==2.20.*
     , temporary ==1.3.*
-    , text ==1.2.*
+    , text ==2.0.*
     , time ==1.9.*
     , time-compat ==1.9.*
     , time-manager ==0.0.*
     , timeit ==2.0.*
     , tls >=1.6.0 && <1.7
-    , transformers ==0.5.*
+    , transformers ==0.6.*
     , unliftio ==0.2.*
     , unliftio-core ==0.2.*
     , websockets ==0.12.*

src/Simplex/FileTransfer/Agent.hs

@@ -4,42 +4,37 @@
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedRecordDot #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}
 
+{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
+
 module Simplex.FileTransfer.Agent
-  ( startWorkers,
+  ( startXFTPWorkers,
     closeXFTPAgent,
     toFSFilePath,
     -- Receiving files
-    receiveFile,
-    deleteRcvFile,
+    xftpReceiveFile',
+    xftpDeleteRcvFile',
     -- Sending files
-    sendFileExperimental,
-    sendFile,
+    xftpSendFile',
     deleteSndFileInternal,
     deleteSndFileRemote,
   )
 where
 
-import Control.Concurrent.STM (stateTVar)
 import Control.Logger.Simple (logError)
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.Reader
-import Crypto.Random (ChaChaDRG, randomBytesGenerate)
-import Data.Bifunctor (first)
-import Data.ByteString (ByteString)
-import qualified Data.ByteString.Base64.URL as U
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
-import Data.Composition ((.:))
 import Data.Int (Int64)
-import Data.List (foldl', isSuffixOf, partition, sortOn)
-import Data.List.NonEmpty (nonEmpty)
+import Data.List (foldl', sortOn)
 import qualified Data.List.NonEmpty as L
 import Data.Map (Map)
 import qualified Data.Map.Strict as M
@@ -50,7 +45,7 @@ import Simplex.FileTransfer.Client (XFTPChunkSpec (..))
 import Simplex.FileTransfer.Client.Main
 import Simplex.FileTransfer.Crypto
 import Simplex.FileTransfer.Description
-import Simplex.FileTransfer.Protocol (FileParty (..), FilePartyI, SFileParty (..))
+import Simplex.FileTransfer.Protocol (FileParty (..), SFileParty (..))
 import Simplex.FileTransfer.Transport (XFTPRcvChunkSpec (..))
 import Simplex.FileTransfer.Types
 import Simplex.FileTransfer.Util (removePath, uniqueCombine)
@@ -62,18 +57,16 @@ import Simplex.Messaging.Agent.Store.SQLite
 import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Crypto.Lazy as LC
 import Simplex.Messaging.Encoding
-import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Protocol (EntityId, XFTPServer, XFTPServerWithAuth)
+import Simplex.Messaging.Protocol (EntityId, XFTPServer)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (liftError, liftIOEither, tshow, unlessM, whenM)
+import Simplex.Messaging.Util (liftError, tshow, unlessM, whenM)
 import System.FilePath (takeFileName, (</>))
 import UnliftIO
-import UnliftIO.Concurrent
 import UnliftIO.Directory
 
-startWorkers :: AgentMonad m => AgentClient -> Maybe FilePath -> m ()
-startWorkers c workDir = do
+startXFTPWorkers :: AgentMonad m => AgentClient -> Maybe FilePath -> m ()
+startXFTPWorkers c workDir = do
   wd <- asks $ xftpWorkDir . xftpAgent
   atomically $ writeTVar wd workDir
   startRcvFiles
@@ -108,8 +101,8 @@ closeXFTPAgent XFTPAgent {xftpRcvWorkers, xftpSndWorkers} = do
       ws <- atomically $ stateTVar wsSel (,M.empty)
       mapM_ (uninterruptibleCancel . snd) ws
 
-receiveFile :: AgentMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> m RcvFileId
-receiveFile c userId (ValidFileDescription fd@FileDescription {chunks}) = do
+xftpReceiveFile' :: AgentMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> m RcvFileId
+xftpReceiveFile' c userId (ValidFileDescription fd@FileDescription {chunks}) = do
   g <- asks idsDrg
   prefixPath <- getPrefixPath "rcv.xftp"
   createDirectory prefixPath
@@ -204,6 +197,7 @@ runXFTPRcvWorker c srv doWork = do
       let chunkSpec = XFTPRcvChunkSpec chunkPath (unFileSize chunkSize) (unFileDigest digest)
           relChunkPath = fileTmpPath </> takeFileName chunkPath
       agentXFTPDownloadChunk c userId digest replica chunkSpec
+      atomically $ waitUntilForeground c
       (complete, progress) <- withStore c $ \db -> runExceptT $ do
         liftIO $ updateRcvFileChunkReceived db (rcvChunkReplicaId replica) rcvChunkId relChunkPath
         RcvFile {size = FileSize total, chunks} <- ExceptT $ getRcvFile db rcvFileId
@@ -261,6 +255,7 @@ runXFTPRcvLocalWorker c doWork = do
       void $ liftError (INTERNAL . show) $ decryptChunks encSize chunkPaths key nonce $ \_ -> pure fsSavePath
       notify c rcvFileEntityId $ RFDONE fsSavePath
       forM_ tmpPath (removePath <=< toFSFilePath)
+      atomically $ waitUntilForeground c
       withStore' c (`updateRcvFileComplete` rcvFileId)
       where
         getChunkPaths :: [RcvFileChunk] -> m [FilePath]
@@ -272,8 +267,8 @@ runXFTPRcvLocalWorker c doWork = do
         getChunkPaths (RcvFileChunk {chunkTmpPath = Nothing} : _cs) =
           throwError $ INTERNAL "no chunk path"
 
-deleteRcvFile :: AgentMonad m => AgentClient -> RcvFileId -> m ()
-deleteRcvFile c rcvFileEntityId = do
+xftpDeleteRcvFile' :: AgentMonad m => AgentClient -> RcvFileId -> m ()
+xftpDeleteRcvFile' c rcvFileEntityId = do
   RcvFile {rcvFileId, prefixPath, status} <- withStore c $ \db -> getRcvFileByEntityId db rcvFileEntityId
   if status == RFSComplete || status == RFSError
     then do
@@ -281,66 +276,11 @@ deleteRcvFile c rcvFileEntityId = do
       withStore' c (`deleteRcvFile'` rcvFileId)
     else withStore' c (`updateRcvFileDeleted` rcvFileId)
 
-sendFileExperimental :: forall m. AgentMonad m => AgentClient -> UserId -> FilePath -> Int -> m SndFileId
-sendFileExperimental c@AgentClient {xftpServers} userId filePath numRecipients = do
-  g <- asks idsDrg
-  sndFileId <- liftIO $ randomId g 12
-  xftpSrvs <- atomically $ TM.lookup userId xftpServers
-  void $ forkIO $ sendCLI sndFileId $ maybe [] L.toList xftpSrvs
-  pure sndFileId
-  where
-    randomId :: TVar ChaChaDRG -> Int -> IO ByteString
-    randomId gVar n = U.encode <$> (atomically . stateTVar gVar $ randomBytesGenerate n)
-    sendCLI :: SndFileId -> [XFTPServerWithAuth] -> m ()
-    sendCLI sndFileId xftpSrvs = do
-      let fileName = takeFileName filePath
-      workPath <- getXFTPWorkPath
-      outputDir <- uniqueCombine workPath $ fileName <> ".descr"
-      createDirectory outputDir
-      let tempPath = workPath </> "snd"
-      createDirectoryIfMissing False tempPath
-      runSend fileName outputDir tempPath `catchAgentError` \e -> do
-        cleanup outputDir tempPath
-        notify c sndFileId $ SFERR e
-      where
-        runSend :: String -> FilePath -> FilePath -> m ()
-        runSend fileName outputDir tempPath = do
-          let sendOptions =
-                SendOptions
-                  { filePath,
-                    outputDir = Just outputDir,
-                    numRecipients,
-                    xftpServers = xftpSrvs,
-                    retryCount = 3,
-                    tempPath = Just tempPath,
-                    verbose = False
-                  }
-          liftCLI $ cliSendFileOpts sendOptions False $ notify c sndFileId .: SFPROG
-          (sndDescr, rcvDescrs) <- readDescrs outputDir fileName
-          cleanup outputDir tempPath
-          notify c sndFileId $ SFDONE sndDescr rcvDescrs
-        cleanup :: FilePath -> FilePath -> m ()
-        cleanup outputDir tempPath = do
-          removePath tempPath
-          removePath outputDir
-    liftCLI :: ExceptT CLIError IO () -> m ()
-    liftCLI = either (throwError . INTERNAL . show) pure <=< liftIO . runExceptT
-    readDescrs :: FilePath -> FilePath -> m (ValidFileDescription 'FSender, [ValidFileDescription 'FRecipient])
-    readDescrs outDir fileName = do
-      let descrDir = outDir </> (fileName <> ".xftp")
-      files <- listDirectory descrDir
-      let (sdFiles, rdFiles) = partition ("snd.xftp.private" `isSuffixOf`) files
-          sdFile = maybe "" (\l -> descrDir </> L.head l) (nonEmpty sdFiles)
-          rdFiles' = map (descrDir </>) rdFiles
-      (,) <$> readDescr sdFile <*> mapM readDescr rdFiles'
-    readDescr :: FilePartyI p => FilePath -> m (ValidFileDescription p)
-    readDescr f = liftIOEither $ first INTERNAL . strDecode <$> B.readFile f
-
 notify :: forall m e. (MonadUnliftIO m, AEntityI e) => AgentClient -> EntityId -> ACommand 'Agent e -> m ()
 notify c entId cmd = atomically $ writeTBQueue (subQ c) ("", entId, APC (sAEntity @e) cmd)
 
-sendFile :: AgentMonad m => AgentClient -> UserId -> FilePath -> Int -> m SndFileId
-sendFile c userId filePath numRecipients = do
+xftpSendFile' :: AgentMonad m => AgentClient -> UserId -> FilePath -> Int -> m SndFileId
+xftpSendFile' c userId filePath numRecipients = do
   g <- asks idsDrg
   prefixPath <- getPrefixPath "snd.xftp"
   createDirectory prefixPath
@@ -478,6 +418,7 @@ runXFTPSndWorker c srv doWork = do
       let chunkSpec' = chunkSpec {filePath = fsFilePath} :: XFTPChunkSpec
       atomically $ assertAgentForeground c
       agentXFTPUploadChunk c userId chunkDigest replica' chunkSpec'
+      atomically $ waitUntilForeground c
       sf@SndFile {sndFileEntityId, prefixPath, chunks} <- withStore c $ \db -> do
         updateSndChunkReplicaStatus db sndChunkReplicaId SFRSUploaded
         getSndFile db sndFileId
@@ -541,7 +482,7 @@ runXFTPSndWorker c srv doWork = do
             rcvChunks :: [[FileChunk]]
             rcvChunks = map (sortChunks . M.elems) $ M.elems $ foldl' addRcvChunk M.empty rcvReplicas
             sortChunks :: [FileChunk] -> [FileChunk]
-            sortChunks = map reverseReplicas . sortOn (chunkNo :: FileChunk -> Int)
+            sortChunks = map reverseReplicas . sortOn (\fc -> fc.chunkNo)
             reverseReplicas ch@FileChunk {replicas} = (ch :: FileChunk) {replicas = reverse replicas}
             addRcvChunk :: Map Int (Map Int FileChunk) -> SentRecipientReplica -> Map Int (Map Int FileChunk)
             addRcvChunk m SentRecipientReplica {chunkNo, server, rcvNo, replicaId, replicaKey, digest, chunkSize} =

src/Simplex/FileTransfer/Client.hs

@@ -3,12 +3,14 @@
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedRecordDot #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications #-}
 
 module Simplex.FileTransfer.Client where
 
+import Control.Monad
 import Control.Monad.Except
 import Data.Bifunctor (first)
 import Data.ByteString.Builder (Builder, byteString)
@@ -108,7 +110,7 @@ xftpClientServer = B.unpack . strEncode . snd3 . transportSession
     snd3 (_, s, _) = s
 
 xftpTransportHost :: XFTPClient -> TransportHost
-xftpTransportHost = (host :: HClient -> TransportHost) . client_ . http2Client
+xftpTransportHost c = c.http2Client.client_.host
 
 xftpSessionTs :: XFTPClient -> UTCTime
 xftpSessionTs = sessionTs . http2Client

src/Simplex/FileTransfer/Client/Agent.hs

@@ -8,6 +8,7 @@
 module Simplex.FileTransfer.Client.Agent where
 
 import Control.Logger.Simple (logInfo)
+import Control.Monad
 import Control.Monad.Except
 import Data.Bifunctor (first)
 import qualified Data.ByteString.Char8 as B
@@ -21,7 +22,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (ProtocolServer (..), XFTPServer)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (catchAll_, tryError)
+import Simplex.Messaging.Util (catchAll_)
 import UnliftIO
 
 type XFTPClientVar = TMVar (Either XFTPClientAgentError XFTPClient)

src/Simplex/FileTransfer/Client/Main.hs

@@ -5,10 +5,13 @@
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedRecordDot #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications #-}
 
+{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
+
 module Simplex.FileTransfer.Client.Main
   ( SendOptions (..),
     CLIError (..),
@@ -27,7 +30,6 @@ module Simplex.FileTransfer.Client.Main
   )
 where
 
-import Control.Concurrent.STM (stateTVar)
 import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Except
@@ -367,7 +369,7 @@ cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, re
         rcvChunks :: [[FileChunk]]
         rcvChunks = map (sortChunks . M.elems) $ M.elems $ foldl' addRcvChunk M.empty rcvReplicas
         sortChunks :: [FileChunk] -> [FileChunk]
-        sortChunks = map reverseReplicas . sortOn (chunkNo :: FileChunk -> Int)
+        sortChunks = map reverseReplicas . sortOn (\c -> c.chunkNo)
         reverseReplicas ch@FileChunk {replicas} = (ch :: FileChunk) {replicas = reverse replicas}
         addRcvChunk :: Map Int (Map Int FileChunk) -> SentRecipientReplica -> Map Int (Map Int FileChunk)
         addRcvChunk m SentRecipientReplica {chunkNo, server, rcvNo, replicaId, replicaKey, digest, chunkSize} =
@@ -417,16 +419,16 @@ getChunkDigest XFTPChunkSpec {filePath = chunkPath, chunkOffset, chunkSize} =
 
 cliReceiveFile :: ReceiveOptions -> ExceptT CLIError IO ()
 cliReceiveFile ReceiveOptions {fileDescription, filePath, retryCount, tempPath, verbose, yes} =
-  getFileDescription' fileDescription >>= receiveFile
+  getFileDescription' fileDescription >>= receive
   where
-    receiveFile :: ValidFileDescription 'FRecipient -> ExceptT CLIError IO ()
-    receiveFile (ValidFileDescription FileDescription {size, digest, key, nonce, chunks}) = do
+    receive :: ValidFileDescription 'FRecipient -> ExceptT CLIError IO ()
+    receive (ValidFileDescription FileDescription {size, digest, key, nonce, chunks}) = do
       encPath <- getEncPath tempPath "xftp"
       createDirectory encPath
       a <- atomically $ newXFTPAgent defaultXFTPClientAgentConfig
       liftIO $ printNoNewLine "Downloading file..."
       downloadedChunks <- newTVarIO []
-      let srv FileChunk {replicas} = server (head replicas :: FileChunkReplica)
+      let srv FileChunk {replicas} = (head replicas).server
           srvChunks = groupAllOn srv chunks
       chunkPaths <- map snd . sortOn fst . concat <$> pooledForConcurrentlyN 16 srvChunks (mapM $ downloadFileChunk a encPath size downloadedChunks)
       encDigest <- liftIO $ LC.sha512Hash <$> readChunks chunkPaths
@@ -505,7 +507,7 @@ cliFileDescrInfo InfoOptions {fileDescription} = do
         putStrLn $ "File download size: " <> strEnc size
         putStrLn "File server(s):"
         forM_ replicas $ \srvReplicas -> do
-          let srv = replicaServer $ head srvReplicas
+          let srv = (head srvReplicas).server
               chSizes = map (\FileServerReplica {chunkSize = chSize_} -> unFileSize $ fromMaybe chunkSize chSize_) srvReplicas
           putStrLn $ strEnc srv <> ": " <> strEnc (FileSize $ sum chSizes)
       where

src/Simplex/FileTransfer/Crypto.hs

@@ -6,6 +6,7 @@
 
 module Simplex.FileTransfer.Crypto where
 
+import Control.Monad
 import Control.Monad.Except
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Bifunctor (first)

src/Simplex/FileTransfer/Description.hs

@@ -5,10 +5,13 @@
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE KindSignatures #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedRecordDot #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 
+{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
+
 module Simplex.FileTransfer.Description
   ( FileDescription (..),
     AFileDescription (..),
@@ -25,7 +28,6 @@ module Simplex.FileTransfer.Description
     YAMLServerReplicas (..), -- for tests
     validateFileDescription,
     groupReplicasByServer,
-    replicaServer,
     fdSeparator,
     kb,
     mb,
@@ -199,7 +201,7 @@ validateFileDescription fd@FileDescription {size, chunks}
   | chunksSize chunks /= unFileSize size = Left "chunks total size is different than file size"
   | otherwise = Right $ ValidFD fd
   where
-    chunkNos = map (chunkNo :: FileChunk -> Int) chunks
+    chunkNos = map (\c -> c.chunkNo) chunks
     chunksSize = fromIntegral . foldl' (\s FileChunk {chunkSize} -> s + unFileSize chunkSize) 0
 
 encodeFileDescription :: FileDescription p -> YAMLFileDescription
@@ -257,7 +259,7 @@ instance (ToField a) => ToField (FileSize a) where toField (FileSize s) = toFiel
 
 groupReplicasByServer :: FileSize Word32 -> [FileChunk] -> [[FileServerReplica]]
 groupReplicasByServer defChunkSize =
-  groupAllOn replicaServer . unfoldChunksToReplicas defChunkSize
+  groupAllOn (\r -> r.server) . unfoldChunksToReplicas defChunkSize
 
 encodeFileReplicas :: FileSize Word32 -> [FileChunk] -> [YAMLServerReplicas]
 encodeFileReplicas defChunkSize =
@@ -265,13 +267,10 @@ encodeFileReplicas defChunkSize =
   where
     encodeServerReplicas fs =
       YAMLServerReplicas
-        { server = replicaServer $ head fs, -- groupAllOn guarantees that fs is not empty
+        { server = (head fs).server, -- groupAllOn guarantees that fs is not empty
           chunks = map (B.unpack . encodeServerReplica) fs
         }
 
-replicaServer :: FileServerReplica -> XFTPServer
-replicaServer = server
-
 encodeServerReplica :: FileServerReplica -> ByteString
 encodeServerReplica FileServerReplica {chunkNo, replicaId, replicaKey, digest, chunkSize} =
   bshow chunkNo
@@ -323,7 +322,7 @@ foldReplicasToChunks :: FileSize Word32 -> [FileServerReplica] -> Either String
 foldReplicasToChunks defChunkSize fs = do
   sd <- foldSizesDigests fs
   -- TODO validate (check that chunks match) or in separate function
-  sortOn (chunkNo :: FileChunk -> Int) . map reverseReplicas . M.elems <$> foldChunks sd fs
+  sortOn (\c -> c.chunkNo) . map reverseReplicas . M.elems <$> foldChunks sd fs
   where
     foldSizesDigests :: [FileServerReplica] -> Either String (Map Int (FileSize Word32), Map Int FileDigest)
     foldSizesDigests = foldl' addSizeDigest $ Right (M.empty, M.empty)

src/Simplex/FileTransfer/Server.hs

@@ -12,6 +12,7 @@
 module Simplex.FileTransfer.Server where
 
 import Control.Logger.Simple
+import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Unlift (MonadUnliftIO)
 import Control.Monad.Reader

src/Simplex/FileTransfer/Server/Main.hs

@@ -32,7 +32,7 @@ import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
 import Text.Read (readMaybe)
 
 xftpServerVersion :: String
-xftpServerVersion = "1.0.1"
+xftpServerVersion = "1.1.0"
 
 xftpServerCLI :: FilePath -> FilePath -> IO ()
 xftpServerCLI cfgPath logPath = do

src/Simplex/FileTransfer/Transport.hs

@@ -16,7 +16,9 @@ module Simplex.FileTransfer.Transport
 where
 
 import qualified Control.Exception as E
+import Control.Monad
 import Control.Monad.Except
+import Control.Monad.IO.Class
 import qualified Data.ByteArray as BA
 import Data.ByteString.Builder (Builder, byteString)
 import Data.ByteString.Char8 (ByteString)

src/Simplex/Messaging/Agent.hs

@@ -8,12 +8,15 @@
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedLists #-}
+{-# LANGUAGE OverloadedRecordDot #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}
 
+{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
+
 -- |
 -- Module      : Simplex.Messaging.Agent
 -- Copyright   : (c) simplex.chat
@@ -35,6 +38,7 @@ module Simplex.Messaging.Agent
     AgentClient (..),
     AgentMonad,
     AgentErrorMonad,
+    SubscriptionsInfo (..),
     getSMPAgentClient,
     disconnectAgentClient,
     resumeAgentClient,
@@ -96,12 +100,13 @@ module Simplex.Messaging.Agent
     debugAgentLocks,
     getAgentStats,
     resetAgentStats,
+    getAgentSubscriptions,
     logConnection,
   )
 where
 
-import Control.Concurrent.STM (stateTVar)
 import Control.Logger.Simple (logError, logInfo, showText)
+import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Unlift (MonadUnliftIO)
 import Control.Monad.Reader
@@ -122,8 +127,7 @@ import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Time.Clock
 import Data.Time.Clock.System (systemToUTCTime)
-import qualified Database.SQLite.Simple as DB
-import Simplex.FileTransfer.Agent (closeXFTPAgent, deleteRcvFile, deleteSndFileInternal, deleteSndFileRemote, receiveFile, sendFile, startWorkers, toFSFilePath)
+import Simplex.FileTransfer.Agent (closeXFTPAgent, xftpDeleteRcvFile', deleteSndFileInternal, deleteSndFileRemote, xftpReceiveFile', xftpSendFile', startXFTPWorkers, toFSFilePath)
 import Simplex.FileTransfer.Description (ValidFileDescription)
 import Simplex.FileTransfer.Protocol (FileParty (..))
 import Simplex.FileTransfer.Util (removePath)
@@ -135,6 +139,7 @@ import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite
+import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
 import Simplex.Messaging.Client (ProtocolClient (..), ServerTransmission)
 import qualified Simplex.Messaging.Crypto as C
@@ -223,7 +228,7 @@ createConnection c userId enableNtfs cMode clientData = withAgentEnv c $ newConn
 
 -- | Join SMP agent connection (JOIN command)
 joinConnection :: AgentErrorMonad m => AgentClient -> UserId -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
-joinConnection c userId enableNtfs = withAgentEnv c .: joinConn c userId "" False enableNtfs
+joinConnection c userId enableNtfs = withAgentEnv c .: joinConn c userId "" enableNtfs
 
 -- | Allow connection to continue after CONF notification (LET command)
 allowConnection :: AgentErrorMonad m => AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> m ()
@@ -351,19 +356,19 @@ toggleConnectionNtfs :: AgentErrorMonad m => AgentClient -> ConnId -> Bool -> m
 toggleConnectionNtfs c = withAgentEnv c .: toggleConnectionNtfs' c
 
 xftpStartWorkers :: AgentErrorMonad m => AgentClient -> Maybe FilePath -> m ()
-xftpStartWorkers c = withAgentEnv c . startWorkers c
+xftpStartWorkers c = withAgentEnv c . startXFTPWorkers c
 
 -- | Receive XFTP file
 xftpReceiveFile :: AgentErrorMonad m => AgentClient -> UserId -> ValidFileDescription 'FRecipient -> m RcvFileId
-xftpReceiveFile c = withAgentEnv c .: receiveFile c
+xftpReceiveFile c = withAgentEnv c .: xftpReceiveFile' c
 
 -- | Delete XFTP rcv file (deletes work files from file system and db records)
 xftpDeleteRcvFile :: AgentErrorMonad m => AgentClient -> RcvFileId -> m ()
-xftpDeleteRcvFile c = withAgentEnv c . deleteRcvFile c
+xftpDeleteRcvFile c = withAgentEnv c . xftpDeleteRcvFile' c
 
 -- | Send XFTP file
 xftpSendFile :: AgentErrorMonad m => AgentClient -> UserId -> FilePath -> Int -> m SndFileId
-xftpSendFile c = withAgentEnv c .:. sendFile c
+xftpSendFile c = withAgentEnv c .:. xftpSendFile' c
 
 -- | Delete XFTP snd file internally (deletes work files from file system and db records)
 xftpDeleteSndFileInternal :: AgentErrorMonad m => AgentClient -> SndFileId -> m ()
@@ -425,7 +430,7 @@ processCommand :: forall m. AgentMonad m => AgentClient -> (EntityId, APartyCmd
 processCommand c (connId, APC e cmd) =
   second (APC e) <$> case cmd of
     NEW enableNtfs (ACM cMode) -> second (INV . ACR cMode) <$> newConn c userId connId enableNtfs cMode Nothing
-    JOIN enableNtfs (ACR _ cReq) connInfo -> (,OK) <$> joinConn c userId connId False enableNtfs cReq connInfo
+    JOIN enableNtfs (ACR _ cReq) connInfo -> (,OK) <$> joinConn c userId connId enableNtfs cReq connInfo
     LET confId ownCInfo -> allowConnection' c connId confId ownCInfo $> (connId, OK)
     ACPT invId ownCInfo -> (,OK) <$> acceptContact' c connId True invId ownCInfo
     RJCT invId -> rejectContact' c connId invId $> (connId, OK)
@@ -519,9 +524,9 @@ ackMessageAsync' c corrId connId msgId rcptInfo_ = do
     enqueueAck :: m ()
     enqueueAck = do
       let mId = InternalId msgId
-      RcvMsg {msgType} <- withStore c $ \db -> getRcvMsg db connId mId
+      RcvMsg {msgType} <- withStoreCtx "ackMessageAsync': getRcvMsg" c $ \db -> getRcvMsg db connId mId
       when (isJust rcptInfo_ && msgType /= AM_A_MSG_) $ throwError $ CMD PROHIBITED
-      (RcvQueue {server}, _) <- withStore c $ \db -> setMsgUserAck db connId mId
+      (RcvQueue {server}, _) <- withStoreCtx "ackMessageAsync': setMsgUserAck" c $ \db -> setMsgUserAck db connId mId
       enqueueCommand c corrId connId (Just server) . AClientCommand $ APC SAEConn $ ACK msgId rcptInfo_
 
 deleteConnectionAsync' :: forall m. AgentMonad m => AgentClient -> ConnId -> m ()
@@ -581,16 +586,16 @@ newRcvConnSrv c userId connId enableNtfs cMode clientData srv = do
       withStore' c $ \db -> createRatchetX3dhKeys db connId pk1 pk2
       pure (connId, CRInvitationUri crData $ toVersionRangeT e2eRcvParams e2eEncryptVRange)
 
-joinConn :: AgentMonad m => AgentClient -> UserId -> ConnId -> Bool -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
-joinConn c userId connId asyncMode enableNtfs cReq cInfo = do
+joinConn :: AgentMonad m => AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId
+joinConn c userId connId enableNtfs cReq cInfo = do
   srv <- case cReq of
     CRInvitationUri ConnReqUriData {crSmpQueues = q :| _} _ ->
       getNextServer c userId [qServer q]
     _ -> getSMPServer c userId
-  joinConnSrv c userId connId asyncMode enableNtfs cReq cInfo srv
+  joinConnSrv c userId connId enableNtfs cReq cInfo srv
 
-joinConnSrv :: AgentMonad m => AgentClient -> UserId -> ConnId -> Bool -> Bool -> ConnectionRequestUri c -> ConnInfo -> SMPServerWithAuth -> m ConnId
-joinConnSrv c userId connId asyncMode enableNtfs (CRInvitationUri ConnReqUriData {crAgentVRange, crSmpQueues = (qUri :| _)} e2eRcvParamsUri) cInfo srv = do
+startJoinInvitation :: AgentMonad m => UserId -> ConnId -> Bool -> ConnectionRequestUri 'CMInvitation -> m (Compatible Version, ConnData, SndQueue, CR.Ratchet 'C.X448, CR.E2ERatchetParams 'C.X448)
+startJoinInvitation userId connId enableNtfs (CRInvitationUri ConnReqUriData {crAgentVRange, crSmpQueues = (qUri :| _)} e2eRcvParamsUri) = do
   AgentConfig {smpClientVRange, smpAgentVRange, e2eEncryptVRange} <- asks config
   case ( qUri `compatibleVersion` smpClientVRange,
          e2eRcvParamsUri `compatibleVersion` e2eEncryptVRange,
@@ -603,31 +608,29 @@ joinConnSrv c userId connId asyncMode enableNtfs (CRInvitationUri ConnReqUriData
       q <- newSndQueue userId "" qInfo
       let duplexHS = connAgentVersion /= 1
           cData = ConnData {userId, connId, connAgentVersion, enableNtfs, duplexHandshake = Just duplexHS, lastExternalSndId = 0, deleted = False, ratchetSyncState = RSOk}
-      connId' <- setUpConn asyncMode cData q rc
-      let sq = (q :: SndQueue) {connId = connId'}
-          cData' = (cData :: ConnData) {connId = connId'}
-      tryError (confirmQueue aVersion c cData' sq srv cInfo $ Just e2eSndParams) >>= \case
-        Right _ -> do
-          unless duplexHS . void $ enqueueMessage c cData' sq SMP.noMsgFlags HELLO
-          pure connId'
-        Left e -> do
-          -- possible improvement: recovery for failure on network timeout, see rfcs/2022-04-20-smp-conf-timeout-recovery.md
-          unless asyncMode $ withStore' c (`deleteConn` connId')
-          throwError e
-      where
-        setUpConn True _ sq rc =
-          withStore c $ \db -> runExceptT $ do
-            void . ExceptT $ updateNewConnSnd db connId sq
-            liftIO $ createRatchet db connId rc
-            pure connId
-        setUpConn False cData sq rc = do
-          g <- asks idsDrg
-          withStore c $ \db -> runExceptT $ do
-            connId' <- ExceptT $ createSndConn db g cData sq
-            liftIO $ createRatchet db connId' rc
-            pure connId'
+      pure (aVersion, cData, q, rc, e2eSndParams)
     _ -> throwError $ AGENT A_VERSION
-joinConnSrv c userId connId False enableNtfs (CRContactUri ConnReqUriData {crAgentVRange, crSmpQueues = (qUri :| _)}) cInfo srv = do
+
+joinConnSrv :: AgentMonad m => AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> SMPServerWithAuth -> m ConnId
+joinConnSrv c userId connId enableNtfs inv@CRInvitationUri {} cInfo srv = do
+  (aVersion, cData@ConnData {connAgentVersion}, q, rc, e2eSndParams) <- startJoinInvitation userId connId enableNtfs inv
+  g <- asks idsDrg
+  connId' <- withStore c $ \db -> runExceptT $ do
+    connId' <- ExceptT $ createSndConn db g cData q
+    liftIO $ createRatchet db connId' rc
+    pure connId'
+  let sq = (q :: SndQueue) {connId = connId'}
+      cData' = (cData :: ConnData) {connId = connId'}
+      duplexHS = connAgentVersion /= 1
+  tryError (confirmQueue aVersion c cData' sq srv cInfo $ Just e2eSndParams) >>= \case
+    Right _ -> do
+      unless duplexHS . void $ enqueueMessage c cData' sq SMP.noMsgFlags HELLO
+      pure connId'
+    Left e -> do
+      -- possible improvement: recovery for failure on network timeout, see rfcs/2022-04-20-smp-conf-timeout-recovery.md
+      withStore' c (`deleteConn` connId')
+      throwError e
+joinConnSrv c userId connId enableNtfs (CRContactUri ConnReqUriData {crAgentVRange, crSmpQueues = (qUri :| _)}) cInfo srv = do
   aVRange <- asks $ smpAgentVRange . config
   clientVRange <- asks $ smpClientVRange . config
   case ( qUri `compatibleVersion` clientVRange,
@@ -638,7 +641,16 @@ joinConnSrv c userId connId False enableNtfs (CRContactUri ConnReqUriData {crAge
       sendInvitation c userId qInfo vrsn cReq cInfo
       pure connId'
     _ -> throwError $ AGENT A_VERSION
-joinConnSrv _c _userId _connId True _enableNtfs (CRContactUri _) _cInfo _srv = do
+
+joinConnSrvAsync :: AgentMonad m => AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> SMPServerWithAuth -> m ()
+joinConnSrvAsync c userId connId enableNtfs inv@CRInvitationUri {} cInfo srv = do
+  (aVersion, cData, q, rc, e2eSndParams) <- startJoinInvitation userId connId enableNtfs inv
+  dbQueueId <- withStore c $ \db -> runExceptT $ do
+    liftIO $ createRatchet db connId rc
+    ExceptT $ updateNewConnSnd db connId q
+  let q' = (q :: SndQueue) {dbQueueId}
+  confirmQueueAsync aVersion c cData q' srv cInfo $ Just e2eSndParams
+joinConnSrvAsync _c _userId _connId _enableNtfs (CRContactUri _) _cInfo _srv = do
   throwError $ CMD PROHIBITED
 
 createReplyQueue :: AgentMonad m => AgentClient -> ConnData -> SndQueue -> SMPServerWithAuth -> m SMPQueueInfo
@@ -673,7 +685,7 @@ acceptContact' c connId enableNtfs invId ownConnInfo = withConnLock c connId "ac
   withStore c (`getConn` contactConnId) >>= \case
     SomeConn _ (ContactConnection ConnData {userId} _) -> do
       withStore' c $ \db -> acceptInvitation db invId ownConnInfo
-      joinConn c userId connId False enableNtfs connReq ownConnInfo `catchAgentError` \err -> do
+      joinConn c userId connId enableNtfs connReq ownConnInfo `catchAgentError` \err -> do
         withStore' c (`unacceptInvitation` invId)
         throwError err
     _ -> throwError $ CMD PROHIBITED
@@ -720,7 +732,7 @@ subscribeConnections' c connIds = do
       ContactConnection _ rq -> Right [rq]
       NewConnection _ -> Left (Right ())
     sndSubResult :: SndQueue -> Either AgentErrorType ()
-    sndSubResult sq = case status (sq :: SndQueue) of
+    sndSubResult SndQueue {status} = case status of
       Confirmed -> Right ()
       Active -> Left $ CONN SIMPLEX
       _ -> Left $ INTERNAL "unexpected queue status"
@@ -890,7 +902,7 @@ runCommandProcessing c@AgentClient {subQ} server_ = do
           let initUsed = [qServer q]
           usedSrvs <- newTVarIO initUsed
           tryCommand . withNextSrv c userId usedSrvs initUsed $ \srv -> do
-            void $ joinConnSrv c userId connId True enableNtfs cReq connInfo srv
+            joinConnSrvAsync c userId connId enableNtfs cReq connInfo srv
             notify OK
         LET confId ownCInfo -> withServer' . tryCommand $ allowConnection' c connId confId ownCInfo >> notify OK
         ACK msgId rcptInfo_ -> withServer' . tryCommand $ ackMessage' c connId msgId rcptInfo_ >> notify OK
@@ -927,7 +939,7 @@ runCommandProcessing c@AgentClient {subQ} server_ = do
           withServer $ \srv -> tryWithLock "ICQSecure" . withDuplexConn $ \(DuplexConnection cData rqs sqs) ->
             case find (sameQueue (srv, rId)) rqs of
               Just rq'@RcvQueue {server, sndId, status, dbReplaceQueueId = Just replaceQId} ->
-                case find ((replaceQId ==) . dbQId) rqs of
+                case find (\q -> replaceQId == q.dbQueueId) rqs of
                   Just rq1 -> when (status == Confirmed) $ do
                     secureQueue c rq' senderKey
                     withStore' c $ \db -> setRcvQueueStatus db rq' Secured
@@ -1087,6 +1099,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
         withRetryLock2 ri' qLock $ \riState loop -> do
           resp <- tryError $ case msgType of
             AM_CONN_INFO -> sendConfirmation c sq msgBody
+            AM_CONN_INFO_REPLY -> sendConfirmation c sq msgBody
             _ -> sendAgentMessage c sq msgFlags msgBody
           case resp of
             Left e -> do
@@ -1095,7 +1108,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                 SMP SMP.QUOTA -> case msgType of
                   AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                   AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
-                  _ -> retryIfNotExpired RISlow err
+                  _ -> retrySndMsg RISlow
                 SMP SMP.AUTH -> case msgType of
                   AM_CONN_INFO -> connError msgId NOT_AVAILABLE
                   AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
@@ -1124,12 +1137,11 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                 _
                   -- for other operations BROKER HOST is treated as a permanent error (e.g., when connecting to the server),
                   -- the message sending would be retried
-                  | temporaryOrHostError e -> retryIfNotExpired RIFast err
+                  | temporaryOrHostError e -> do
+                    let timeoutSel = if msgType == AM_HELLO_ then helloTimeout else messageTimeout
+                    ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySndMsg RIFast)
                   | otherwise -> notifyDel msgId err
               where
-                retryIfNotExpired riMode err = do
-                  let timeoutSel = if msgType == AM_HELLO_ then helloTimeout else messageTimeout
-                  ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySndMsg riMode)
                 msgExpired timeoutSel = do
                   msgTimeout <- asks $ timeoutSel . config
                   currentTime <- liftIO getCurrentTime
@@ -1139,12 +1151,8 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                   retrySndOp c $ loop riMode
             Right () -> do
               case msgType of
-                AM_CONN_INFO -> do
-                  withStore' c $ \db -> do
-                    setSndQueueStatus db sq Confirmed
-                    when (isJust rq_) $ removeConfirmations db connId
-                  unless (duplexHandshake == Just True) . void $ enqueueMessage c cData sq SMP.noMsgFlags HELLO
-                AM_CONN_INFO_REPLY -> pure ()
+                AM_CONN_INFO -> setConfirmed
+                AM_CONN_INFO_REPLY -> setConfirmed
                 AM_RATCHET_INFO -> pure ()
                 AM_REPLY_ -> pure ()
                 AM_HELLO_ -> do
@@ -1188,7 +1196,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                         -- this is the same queue where this loop delivers messages to but with updated state
                         Just SndQueue {dbReplaceQueueId = Just replacedId, primary} ->
                           -- second part of this condition is a sanity check because dbReplaceQueueId cannot point to the same queue, see switchConnection'
-                          case removeQP (\sq' -> dbQId sq' == replacedId && not (sameQueue addr sq')) sqs of
+                          case removeQP (\sq' -> sq'.dbQueueId == replacedId && not (sameQueue addr sq')) sqs of
                             Nothing -> internalErr msgId "sent QTEST: queue not found in connection"
                             Just (sq', sq'' : sqs') -> do
                               checkSQSwchStatus sq' SSSendingQTEST
@@ -1206,6 +1214,12 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl
                     _ -> internalErr msgId "QTEST sent not in duplex connection"
                 AM_EREADY_ -> pure ()
               delMsgKeep (msgType == AM_A_MSG_) msgId
+              where
+                setConfirmed = do
+                  withStore' c $ \db -> do
+                    setSndQueueStatus db sq Confirmed
+                    when (isJust rq_) $ removeConfirmations db connId
+                  unless (duplexHandshake == Just True) . void $ enqueueMessage c cData sq SMP.noMsgFlags HELLO
   where
     delMsg :: InternalId -> m ()
     delMsg = delMsgKeep False
@@ -1240,13 +1254,13 @@ ackMessage' c connId msgId rcptInfo_ = withConnLock c connId "ackMessage" $ do
     ack :: m ()
     ack = do
       -- the stored message was delivered via a specific queue, the rest failed to decrypt and were already acknowledged
-      (rq, srvMsgId) <- withStore c $ \db -> setMsgUserAck db connId $ InternalId msgId
+      (rq, srvMsgId) <- withStoreCtx "ackMessage': setMsgUserAck" c $ \db -> setMsgUserAck db connId $ InternalId msgId
       ackQueueMessage c rq srvMsgId
     del :: m ()
-    del = withStore' c $ \db -> deleteMsg db connId $ InternalId msgId
+    del = withStoreCtx' "ackMessage': deleteMsg" c $ \db -> deleteMsg db connId $ InternalId msgId
     sendRcpt :: Connection 'CDuplex -> m ()
     sendRcpt (DuplexConnection cData _ sqs) = do
-      msg@RcvMsg {msgType, msgReceipt} <- withStore c $ \db -> getRcvMsg db connId $ InternalId msgId
+      msg@RcvMsg {msgType, msgReceipt} <- withStoreCtx "ackMessage': getRcvMsg" c $ \db -> getRcvMsg db connId $ InternalId msgId
       case rcptInfo_ of
         Just rcptInfo -> do
           unless (msgType == AM_A_MSG_) $ throwError (CMD PROHIBITED)
@@ -1257,7 +1271,7 @@ ackMessage' c connId msgId rcptInfo_ = withConnLock c connId "ackMessage" $ do
         Nothing -> case (msgType, msgReceipt) of
           -- only remove sent message if receipt hash was Ok, both to debug and for future redundancy
           (AM_A_RCVD_, Just MsgReceipt {agentMsgId = sndMsgId, msgRcptStatus = MROk}) ->
-            withStore' c $ \db -> deleteDeliveredSndMsg db connId $ InternalId sndMsgId
+            withStoreCtx' "ackMessage': deleteDeliveredSndMsg" c $ \db -> deleteDeliveredSndMsg db connId $ InternalId sndMsgId
           _ -> pure ()
 
 switchConnection' :: AgentMonad m => AgentClient -> ConnId -> m ConnectionStats
@@ -1297,7 +1311,7 @@ abortConnectionSwitch' c connId =
           | canAbortRcvSwitch rq -> do
             when (ratchetSyncSendProhibited cData) $ throwError $ CMD PROHIBITED
             -- multiple queues to which the connections switches were possible when repeating switch was allowed
-            let (delRqs, keepRqs) = L.partition ((Just (dbQId rq) ==) . dbReplaceQId) rqs
+            let (delRqs, keepRqs) = L.partition (\q -> Just rq.dbQueueId == q.dbReplaceQueueId) rqs
             case L.nonEmpty keepRqs of
               Just rqs' -> do
                 rq' <- withStore' c $ \db -> do
@@ -1415,7 +1429,7 @@ deleteConnQueues c ntf rqs = do
             | temporaryOrHostError e && deleteErrors rq + 1 < maxErrs -> withStore' c (`incRcvDeleteErrors` rq) $> r
             | otherwise -> withStore' c (`deleteConnRcvQueue` rq) >> notifyRQ rq (Just e) $> Right ()
         pure (rq, r')
-    notifyRQ rq e_ = notify ("", qConnId rq, APC SAEConn $ DEL_RCVQ (qServer rq) (queueId rq) e_)
+    notifyRQ rq e_ = notify ("", rq.connId, APC SAEConn $ DEL_RCVQ (qServer rq) (queueId rq) e_)
     notify = when ntf . atomically . writeTBQueue (subQ c)
     connResults :: [(RcvQueue, Either AgentErrorType ())] -> Map ConnId (Either AgentErrorType ())
     connResults = M.map snd . foldl' addResult M.empty
@@ -1721,7 +1735,7 @@ execAgentStoreSQL' :: AgentMonad m => AgentClient -> Text -> m [Text]
 execAgentStoreSQL' c sql = withStore' c (`execSQL` sql)
 
 getAgentMigrations' :: AgentMonad m => AgentClient -> m [UpMigration]
-getAgentMigrations' c = map upMigration <$> withStore' c Migrations.getCurrent
+getAgentMigrations' c = map upMigration <$> withStore' c (Migrations.getCurrent . DB.conn)
 
 debugAgentLocks' :: AgentMonad' m => AgentClient -> m AgentLocks
 debugAgentLocks' AgentClient {connLocks = cs, reconnectLocks = rs, deleteLock = d} = do
@@ -1860,7 +1874,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
                         case (conn', dbReplaceQueueId) of
                           (DuplexConnection _ rqs _, Just replacedId) -> do
                             when primary . withStore' c $ \db -> setRcvQueuePrimary db connId rq
-                            case find ((replacedId ==) . dbQId) rqs of
+                            case find (\q -> replacedId == q.dbQueueId) rqs of
                               Just rq'@RcvQueue {server, rcvId} -> do
                                 checkRQSwchStatus rq' RSSendingQUSE
                                 void $ withStore' c $ \db -> setRcvSwitchStatus db rq' $ Just RSReceivedMessage
@@ -1901,7 +1915,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
                                 | otherwise = pure conn'
                           Right _ -> prohibited >> ack
                           Left e@(AGENT A_DUPLICATE) -> do
-                            withStore' c (\db -> getLastMsg db connId srvMsgId) >>= \case
+                            withStoreCtx' "processSMP: getLastMsg" c (\db -> getLastMsg db connId srvMsgId) >>= \case
                               Just RcvMsg {internalId, msgMeta, msgBody = agentMsgBody, userAck}
                                 | userAck -> ackDel internalId
                                 | otherwise -> do
@@ -2113,7 +2127,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
               clientReceipt :: AMessageReceipt -> m (Maybe MsgReceipt)
               clientReceipt AMessageReceipt {agentMsgId, msgHash} = do
                 let sndMsgId = InternalSndId agentMsgId
-                SndMsg {internalId = InternalId msgId, msgType, internalHash, msgReceipt} <- withStore c $ \db -> getSndMsgViaRcpt db connId sndMsgId
+                SndMsg {internalId = InternalId msgId, msgType, internalHash, msgReceipt} <- withStoreCtx "messagesRcvd: getSndMsgViaRcpt" c $ \db -> getSndMsgViaRcpt db connId sndMsgId
                 if msgType /= AM_A_MSG_
                   then notify (ERR $ AGENT A_PROHIBITED) $> Nothing -- unexpected message type for receipt
                   else case msgReceipt of
@@ -2135,7 +2149,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
                 case (findQ (qAddress sqInfo) sqs, findQ addr sqs) of
                   (Just _, _) -> qError "QADD: queue address is already used in connection"
                   (_, Just sq@SndQueue {dbQueueId}) -> do
-                    let (delSqs, keepSqs) = L.partition ((Just dbQueueId ==) . dbReplaceQId) sqs
+                    let (delSqs, keepSqs) = L.partition (\q -> Just dbQueueId == q.dbReplaceQueueId) sqs
                     case L.nonEmpty keepSqs of
                       Just sqs' -> do
                         -- move inside case?
@@ -2186,7 +2200,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s
             when (ratchetSyncSendProhibited cData') $ throwError $ AGENT (A_QUEUE "ratchet is not synchronized")
             case findQ addr sqs of
               Just sq'@SndQueue {dbReplaceQueueId = Just replaceQId} -> do
-                case find ((replaceQId ==) . dbQId) sqs of
+                case find (\q -> replaceQId == q.dbQueueId) sqs of
                   Just sq1 -> do
                     checkSQSwchStatus sq1 SSSendingQKEY
                     logServer "<--" c srv rId $ "MSG <QUSE> " <> logSecret (snd addr)
@@ -2306,10 +2320,15 @@ connectReplyQueues c cData@ConnData {userId, connId} ownConnInfo (qInfo :| _) =
       dbQueueId <- withStore c $ \db -> upgradeRcvConnToDuplex db connId sq
       enqueueConfirmation c cData sq {dbQueueId} ownConnInfo Nothing
 
+confirmQueueAsync :: forall m. AgentMonad m => Compatible Version -> AgentClient -> ConnData -> SndQueue -> SMPServerWithAuth -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()
+confirmQueueAsync v c cData sq srv connInfo e2eEncryption_ = do
+  resumeMsgDelivery c cData sq
+  msgId <- storeConfirmation c cData sq e2eEncryption_ =<< mkAgentConfirmation v c cData sq srv connInfo
+  queuePendingMsgs c sq [msgId]
+
 confirmQueue :: forall m. AgentMonad m => Compatible Version -> AgentClient -> ConnData -> SndQueue -> SMPServerWithAuth -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()
-confirmQueue (Compatible agentVersion) c cData@ConnData {connId} sq srv connInfo e2eEncryption_ = do
-  aMessage <- mkAgentMessage agentVersion
-  msg <- mkConfirmation aMessage
+confirmQueue v@(Compatible agentVersion) c cData@ConnData {connId} sq srv connInfo e2eEncryption_ = do
+  msg <- mkConfirmation =<< mkAgentConfirmation v c cData sq srv connInfo
   sendConfirmation c sq msg
   withStore' c $ \db -> setSndQueueStatus db sq Confirmed
   where
@@ -2318,32 +2337,33 @@ confirmQueue (Compatible agentVersion) c cData@ConnData {connId} sq srv connInfo
       void . liftIO $ updateSndIds db connId
       encConnInfo <- agentRatchetEncrypt db connId (smpEncode aMessage) e2eEncConnInfoLength
       pure . smpEncode $ AgentConfirmation {agentVersion, e2eEncryption_, encConnInfo}
-    mkAgentMessage :: Version -> m AgentMessage
-    mkAgentMessage 1 = pure $ AgentConnInfo connInfo
-    mkAgentMessage _ = do
+
+mkAgentConfirmation :: AgentMonad m => Compatible Version -> AgentClient -> ConnData -> SndQueue -> SMPServerWithAuth -> ConnInfo -> m AgentMessage
+mkAgentConfirmation (Compatible agentVersion) c cData sq srv connInfo
+  | agentVersion == 1 = pure $ AgentConnInfo connInfo
+  | otherwise = do
       qInfo <- createReplyQueue c cData sq srv
       pure $ AgentConnInfoReply (qInfo :| []) connInfo
 
-enqueueConfirmation :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()
-enqueueConfirmation c cData@ConnData {connId, connAgentVersion} sq connInfo e2eEncryption_ = do
+enqueueConfirmation :: AgentMonad m => AgentClient -> ConnData -> SndQueue -> ConnInfo -> Maybe (CR.E2ERatchetParams 'C.X448) -> m ()
+enqueueConfirmation c cData sq connInfo e2eEncryption_ = do
   resumeMsgDelivery c cData sq
-  msgId <- storeConfirmation
+  msgId <- storeConfirmation c cData sq e2eEncryption_ $ AgentConnInfo connInfo
   queuePendingMsgs c sq [msgId]
-  where
-    storeConfirmation :: m InternalId
-    storeConfirmation = withStore c $ \db -> runExceptT $ do
-      internalTs <- liftIO getCurrentTime
-      (internalId, internalSndId, prevMsgHash) <- liftIO $ updateSndIds db connId
-      let agentMsg = AgentConnInfo connInfo
-          agentMsgStr = smpEncode agentMsg
-          internalHash = C.sha256Hash agentMsgStr
-      encConnInfo <- agentRatchetEncrypt db connId agentMsgStr e2eEncConnInfoLength
-      let msgBody = smpEncode $ AgentConfirmation {agentVersion = connAgentVersion, e2eEncryption_, encConnInfo}
-          msgType = agentMessageType agentMsg
-          msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash}
-      liftIO $ createSndMsg db connId msgData
-      liftIO $ createSndMsgDelivery db connId sq internalId
-      pure internalId
+
+storeConfirmation :: AgentMonad m => AgentClient -> ConnData -> SndQueue -> Maybe (CR.E2ERatchetParams 'C.X448) -> AgentMessage -> m InternalId
+storeConfirmation c ConnData {connId, connAgentVersion} sq e2eEncryption_ agentMsg = withStore c $ \db -> runExceptT $ do
+  internalTs <- liftIO getCurrentTime
+  (internalId, internalSndId, prevMsgHash) <- liftIO $ updateSndIds db connId
+  let agentMsgStr = smpEncode agentMsg
+      internalHash = C.sha256Hash agentMsgStr
+  encConnInfo <- agentRatchetEncrypt db connId agentMsgStr e2eEncConnInfoLength
+  let msgBody = smpEncode $ AgentConfirmation {agentVersion = connAgentVersion, e2eEncryption_, encConnInfo}
+      msgType = agentMessageType agentMsg
+      msgData = SndMsgData {internalId, internalSndId, internalTs, msgType, msgBody, msgFlags = SMP.MsgFlags {notification = True}, internalHash, prevMsgHash}
+  liftIO $ createSndMsg db connId msgData
+  liftIO $ createSndMsgDelivery db connId sq internalId
+  pure internalId
 
 enqueueRatchetKeyMsgs :: forall m. AgentMonad m => AgentClient -> ConnData -> NonEmpty SndQueue -> CR.E2ERatchetParams 'C.X448 -> m AgentMsgId
 enqueueRatchetKeyMsgs c cData (sq :| sqs) e2eEncryption = do

src/Simplex/Messaging/Agent/Client.hs

@@ -10,6 +10,7 @@
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedLists #-}
+{-# LANGUAGE OverloadedRecordDot #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
@@ -71,6 +72,9 @@ module Simplex.Messaging.Agent.Client
     removeSubscription,
     hasActiveSubscription,
     agentClientStore,
+    getAgentSubscriptions,
+    SubscriptionsInfo (..),
+    SubInfo (..),
     AgentOperation (..),
     AgentOpState (..),
     AgentState (..),
@@ -90,6 +94,8 @@ module Simplex.Messaging.Agent.Client
     whenSuspending,
     withStore,
     withStore',
+    withStoreCtx,
+    withStoreCtx',
     storeError,
     userServers,
     pickServer,
@@ -102,9 +108,10 @@ where
 import Control.Applicative ((<|>))
 import Control.Concurrent (forkIO, threadDelay)
 import Control.Concurrent.Async (Async, uninterruptibleCancel)
-import Control.Concurrent.STM (retry, stateTVar, throwSTM)
+import Control.Concurrent.STM (retry, throwSTM)
 import Control.Exception (AsyncException (..))
 import Control.Logger.Simple
+import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
 import Control.Monad.Reader
@@ -125,10 +132,10 @@ import qualified Data.Map.Strict as M
 import Data.Maybe (isJust, listToMaybe)
 import Data.Set (Set)
 import qualified Data.Set as S
+import Data.Text (Text)
 import Data.Text.Encoding
 import Data.Time (UTCTime, defaultTimeLocale, formatTime, getCurrentTime)
 import Data.Word (Word16)
-import qualified Database.SQLite.Simple as DB
 import GHC.Generics (Generic)
 import Network.Socket (HostName)
 import Simplex.FileTransfer.Client (XFTPChunkSpec (..), XFTPClient, XFTPClientConfig (..), XFTPClientError)
@@ -144,8 +151,9 @@ import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore (..), withTransaction)
+import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
 import Simplex.Messaging.Agent.TAsyncs
-import Simplex.Messaging.Agent.TRcvQueues (TRcvQueues)
+import Simplex.Messaging.Agent.TRcvQueues (TRcvQueues (getRcvQueues))
 import qualified Simplex.Messaging.Agent.TRcvQueues as RQ
 import Simplex.Messaging.Client
 import Simplex.Messaging.Client.Agent ()
@@ -421,7 +429,7 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv,
           TM.delete tSess smpClients
           qs <- RQ.getDelSessQueues tSess $ activeSubs c
           mapM_ (`RQ.addQueue` pendingSubs c) qs
-          let cs = S.fromList $ map qConnId qs
+          let cs = S.fromList $ map (\q -> q.connId) qs
           cs' <- RQ.getConns $ activeSubs c
           pure (qs, S.toList $ cs `S.difference` cs')
 
@@ -809,7 +817,7 @@ mkSMPTransportSession :: (AgentMonad' m, SMPQueueRec q) => AgentClient -> q -> m
 mkSMPTransportSession c q = mkSMPTSession q <$> getSessionMode c
 
 mkSMPTSession :: SMPQueueRec q => q -> TransportSessionMode -> SMPTransportSession
-mkSMPTSession q = mkTSession (qUserId q) (qServer q) (qConnId q)
+mkSMPTSession q = mkTSession q.userId (qServer q) q.connId
 
 getSessionMode :: AgentMonad' m => AgentClient -> m TransportSessionMode
 getSessionMode = fmap sessionMode . readTVarIO . useNetworkConfig
@@ -1236,16 +1244,37 @@ waitUntilForeground :: AgentClient -> STM ()
 waitUntilForeground c = unlessM ((ASForeground ==) <$> readTVar (agentState c)) retry
 
 withStore' :: AgentMonad m => AgentClient -> (DB.Connection -> IO a) -> m a
-withStore' c action = withStore c $ fmap Right . action
+withStore' = withStoreCtx_' Nothing
 
 withStore :: AgentMonad m => AgentClient -> (DB.Connection -> IO (Either StoreError a)) -> m a
-withStore c action = do
+withStore = withStoreCtx_ Nothing
+
+withStoreCtx' :: AgentMonad m => String -> AgentClient -> (DB.Connection -> IO a) -> m a
+withStoreCtx' = withStoreCtx_' . Just
+
+withStoreCtx :: AgentMonad m => String -> AgentClient -> (DB.Connection -> IO (Either StoreError a)) -> m a
+withStoreCtx = withStoreCtx_ . Just
+
+withStoreCtx_' :: AgentMonad m => Maybe String -> AgentClient -> (DB.Connection -> IO a) -> m a
+withStoreCtx_' ctx_ c action = withStoreCtx_ ctx_ c $ fmap Right . action
+
+withStoreCtx_ :: AgentMonad m => Maybe String -> AgentClient -> (DB.Connection -> IO (Either StoreError a)) -> m a
+withStoreCtx_ ctx_ c action = do
   st <- asks store
-  liftEitherError storeError . agentOperationBracket c AODatabase (\_ -> pure ()) $
-    withTransaction st action `E.catch` handleInternal
+  liftEitherError storeError . agentOperationBracket c AODatabase (\_ -> pure ()) $ case ctx_ of
+    Nothing -> withTransaction st action `E.catch` handleInternal ""
+    -- uncomment to debug store performance
+    -- Just ctx -> do
+    --   t1 <- liftIO getCurrentTime
+    --   putStrLn $ "agent withStoreCtx start       :: " <> show t1 <> " :: " <> ctx
+    --   r <- withTransaction st action `E.catch` handleInternal (" (" <> ctx <> ")")
+    --   t2 <- liftIO getCurrentTime
+    --   putStrLn $ "agent withStoreCtx end         :: " <> show t2 <> " :: " <> ctx <> " :: duration=" <> show (diffToMilliseconds $ diffUTCTime t2 t1)
+    --   pure r
+    Just _ -> withTransaction st action `E.catch` handleInternal ""
   where
-    handleInternal :: E.SomeException -> IO (Either StoreError a)
-    handleInternal = pure . Left . SEInternal . bshow
+    handleInternal :: String -> E.SomeException -> IO (Either StoreError a)
+    handleInternal ctxStr e = pure . Left . SEInternal . B.pack $ show e <> ctxStr
 
 storeError :: StoreError -> AgentErrorType
 storeError = \case
@@ -1317,3 +1346,27 @@ withNextSrv c userId usedSrvs initUsed action = do
         used' = if null unused then initUsed else srv : used
     writeTVar usedSrvs $! used'
   action srvAuth
+
+data SubInfo = SubInfo {userId :: UserId, server :: Text, rcvId :: Text}
+  deriving (Show, Generic)
+
+instance ToJSON SubInfo where toEncoding = J.genericToEncoding J.defaultOptions
+
+data SubscriptionsInfo = SubscriptionsInfo
+  { activeSubscriptions :: [SubInfo],
+    pendingSubscriptions :: [SubInfo]
+  }
+  deriving (Show, Generic)
+
+instance ToJSON SubscriptionsInfo where toEncoding = J.genericToEncoding J.defaultOptions
+
+getAgentSubscriptions :: MonadIO m => AgentClient -> m SubscriptionsInfo
+getAgentSubscriptions c = do
+  activeSubscriptions <- getSubs activeSubs
+  pendingSubscriptions <- getSubs pendingSubs
+  pure $ SubscriptionsInfo {activeSubscriptions, pendingSubscriptions}
+  where
+    getSubs sel = map subInfo . M.keys <$> readTVarIO (getRcvQueues $ sel c)
+    subInfo (uId, srv, rId) = SubInfo {userId = uId, server = enc srv, rcvId = enc rId}
+    enc :: StrEncoding a => a -> Text
+    enc = decodeLatin1 . strEncode

src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

@@ -6,6 +6,8 @@
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TupleSections #-}
 
+{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
+
 module Simplex.Messaging.Agent.NtfSubSupervisor
   ( runNtfSupervisor,
     nsUpdateToken,
@@ -17,7 +19,6 @@ module Simplex.Messaging.Agent.NtfSubSupervisor
   )
 where
 
-import Control.Concurrent.STM (stateTVar)
 import Control.Logger.Simple (logError, logInfo)
 import Control.Monad
 import Control.Monad.Except
@@ -51,13 +52,13 @@ runNtfSupervisor c = do
   ns <- asks ntfSupervisor
   forever $ do
     cmd@(connId, _) <- atomically . readTBQueue $ ntfSubQ ns
-    handleError connId . agentOperationBracket c AONtfNetwork waitUntilActive $
+    handleErr connId . agentOperationBracket c AONtfNetwork waitUntilActive $
       runExceptT (processNtfSub c cmd) >>= \case
         Left e -> notifyErr connId e
         Right _ -> return ()
   where
-    handleError :: ConnId -> m () -> m ()
-    handleError connId = E.handle $ \(e :: E.SomeException) -> do
+    handleErr :: ConnId -> m () -> m ()
+    handleErr connId = E.handle $ \(e :: E.SomeException) -> do
       logError $ "runNtfSupervisor error " <> tshow e
       notifyErr connId e
     notifyErr connId e = notifyInternalError c connId $ "runNtfSupervisor error " <> show e

src/Simplex/Messaging/Agent/Protocol.hs

@@ -80,6 +80,8 @@ module Simplex.Messaging.Agent.Protocol
     SMPServerWithAuth,
     SrvLoc (..),
     SMPQueue (..),
+    qAddress,
+    sameQueue,
     sameQAddress,
     noAuthSrv,
     SMPQueueUri (..),
@@ -1195,8 +1197,15 @@ updateSMPServerHosts srv@ProtocolServer {host} = case host of
 
 class SMPQueue q where
   qServer :: q -> SMPServer
-  qAddress :: q -> (SMPServer, SMP.QueueId)
-  sameQueue :: (SMPServer, SMP.QueueId) -> q -> Bool
+  queueId :: q -> SMP.QueueId
+
+qAddress :: SMPQueue q => q -> (SMPServer, SMP.QueueId)
+qAddress q = (qServer q, queueId q)
+{-# INLINE qAddress #-}
+
+sameQueue :: SMPQueue q => (SMPServer, SMP.QueueId) -> q -> Bool
+sameQueue addr q = sameQAddress addr (qAddress q)
+{-# INLINE sameQueue #-}
 
 data SMPQueueInfo = SMPQueueInfo {clientVersion :: Version, queueAddress :: SMPQueueAddress}
   deriving (Eq, Show)
@@ -1241,26 +1250,20 @@ data SMPQueueAddress = SMPQueueAddress
 instance SMPQueue SMPQueueUri where
   qServer SMPQueueUri {queueAddress} = qServer queueAddress
   {-# INLINE qServer #-}
-  qAddress SMPQueueUri {queueAddress} = qAddress queueAddress
-  {-# INLINE qAddress #-}
-  sameQueue addr q = sameQAddress addr (qAddress q)
-  {-# INLINE sameQueue #-}
+  queueId SMPQueueUri {queueAddress} = queueId queueAddress
+  {-# INLINE queueId #-}
 
 instance SMPQueue SMPQueueInfo where
   qServer SMPQueueInfo {queueAddress} = qServer queueAddress
   {-# INLINE qServer #-}
-  qAddress SMPQueueInfo {queueAddress} = qAddress queueAddress
-  {-# INLINE qAddress #-}
-  sameQueue addr q = sameQAddress addr (qAddress q)
-  {-# INLINE sameQueue #-}
+  queueId SMPQueueInfo {queueAddress} = queueId queueAddress
+  {-# INLINE queueId #-}
 
 instance SMPQueue SMPQueueAddress where
   qServer SMPQueueAddress {smpServer} = smpServer
   {-# INLINE qServer #-}
-  qAddress SMPQueueAddress {smpServer, senderId} = (smpServer, senderId)
-  {-# INLINE qAddress #-}
-  sameQueue addr q = sameQAddress addr (qAddress q)
-  {-# INLINE sameQueue #-}
+  queueId SMPQueueAddress {senderId} = senderId
+  {-# INLINE queueId #-}
 
 sameQAddress :: (SMPServer, SMP.QueueId) -> (SMPServer, SMP.QueueId) -> Bool
 sameQAddress (srv, qId) (srv', qId') = sameSrvAddr srv srv' && qId == qId'

src/Simplex/Messaging/Agent/Server.hs

@@ -11,7 +11,7 @@ module Simplex.Messaging.Agent.Server
 where
 
 import Control.Logger.Simple (logInfo)
-import Control.Monad.Except
+import Control.Monad
 import Control.Monad.IO.Unlift (MonadUnliftIO)
 import Control.Monad.Reader
 import Crypto.Random (MonadRandom)

src/Simplex/Messaging/Agent/Store.hs

@@ -1,11 +1,14 @@
+{-# LANGUAGE ConstraintKinds #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DeriveAnyClass #-}
 {-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE KindSignatures #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedRecordDot #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StandaloneDeriving #-}
@@ -24,6 +27,7 @@ import qualified Data.List.NonEmpty as L
 import Data.Maybe (isJust)
 import Data.Time (UTCTime)
 import Data.Type.Equality
+import GHC.Records (HasField)
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval (RI2State)
 import qualified Simplex.Messaging.Crypto as C
@@ -36,7 +40,6 @@ import Simplex.Messaging.Protocol
     NotifierId,
     NtfPrivateSignKey,
     NtfPublicVerifyKey,
-    QueueId,
     RcvDhSecret,
     RcvNtfDhSecret,
     RcvPrivateSignKey,
@@ -144,18 +147,14 @@ sndQueueInfo SndQueue {server, sndSwchStatus} =
 instance SMPQueue RcvQueue where
   qServer RcvQueue {server} = server
   {-# INLINE qServer #-}
-  qAddress RcvQueue {server, rcvId} = (server, rcvId)
-  {-# INLINE qAddress #-}
-  sameQueue addr q = sameQAddress addr (qAddress q)
-  {-# INLINE sameQueue #-}
+  queueId RcvQueue {rcvId} = rcvId
+  {-# INLINE queueId #-}
 
 instance SMPQueue SndQueue where
   qServer SndQueue {server} = server
   {-# INLINE qServer #-}
-  qAddress SndQueue {server, sndId} = (server, sndId)
-  {-# INLINE qAddress #-}
-  sameQueue addr q = sameQAddress addr (qAddress q)
-  {-# INLINE sameQueue #-}
+  queueId SndQueue {sndId} = sndId
+  {-# INLINE queueId #-}
 
 findQ :: SMPQueue q => (SMPServer, SMP.QueueId) -> NonEmpty q -> Maybe q
 findQ = find . sameQueue
@@ -183,39 +182,16 @@ switchingRQ = find $ isJust . rcvSwchStatus
 {-# INLINE switchingRQ #-}
 
 updatedQs :: SMPQueueRec q => q -> NonEmpty q -> NonEmpty q
-updatedQs q = L.map $ \q' -> if dbQId q == dbQId q' then q else q'
+updatedQs q = L.map $ \q' -> if q.dbQueueId == q'.dbQueueId then q else q'
 {-# INLINE updatedQs #-}
 
-class SMPQueue q => SMPQueueRec q where
-  qUserId :: q -> UserId
-  qConnId :: q -> ConnId
-  queueId :: q -> QueueId
-  dbQId :: q -> Int64
-  dbReplaceQId :: q -> Maybe Int64
-
-instance SMPQueueRec RcvQueue where
-  qUserId = userId
-  {-# INLINE qUserId #-}
-  qConnId = connId
-  {-# INLINE qConnId #-}
-  queueId = rcvId
-  {-# INLINE queueId #-}
-  dbQId = dbQueueId
-  {-# INLINE dbQId #-}
-  dbReplaceQId = dbReplaceQueueId
-  {-# INLINE dbReplaceQId #-}
-
-instance SMPQueueRec SndQueue where
-  qUserId = userId
-  {-# INLINE qUserId #-}
-  qConnId = connId
-  {-# INLINE qConnId #-}
-  queueId = sndId
-  {-# INLINE queueId #-}
-  dbQId = dbQueueId
-  {-# INLINE dbQId #-}
-  dbReplaceQId = dbReplaceQueueId
-  {-# INLINE dbReplaceQId #-}
+type SMPQueueRec q =
+  ( SMPQueue q,
+    HasField "userId" q UserId,
+    HasField "connId" q ConnId,
+    HasField "dbQueueId" q Int64,
+    HasField "dbReplaceQueueId" q (Maybe Int64)
+  )
 
 -- * Connection types
 

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

@@ -10,6 +10,7 @@
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE NumericUnderscores #-}
+{-# LANGUAGE OverloadedRecordDot #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE QuasiQuotes #-}
@@ -18,7 +19,9 @@
 {-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeOperators #-}
 {-# LANGUAGE UndecidableInstances #-}
+
 {-# OPTIONS_GHC -fno-warn-orphans #-}
+{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
 
 module Simplex.Messaging.Agent.Store.SQLite
   ( SQLiteStore (..),
@@ -211,8 +214,9 @@ module Simplex.Messaging.Agent.Store.SQLite
   )
 where
 
-import Control.Concurrent.STM (stateTVar)
+import Control.Monad
 import Control.Monad.Except
+import Control.Monad.IO.Class
 import Crypto.Random (ChaChaDRG, randomBytesGenerate)
 import Data.Aeson (ToJSON)
 import qualified Data.Aeson as J
@@ -235,8 +239,8 @@ import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1, encodeUtf8)
 import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, getCurrentTime)
 import Data.Word (Word32)
-import Database.SQLite.Simple (FromRow, NamedParam (..), Only (..), Query (..), SQLError, ToRow, field, (:.) (..))
-import qualified Database.SQLite.Simple as DB
+import Database.SQLite.Simple (FromRow (..), NamedParam (..), Only (..), Query (..), SQLError, ToRow (..), field, (:.) (..))
+import qualified Database.SQLite.Simple as SQL
 import Database.SQLite.Simple.FromField
 import Database.SQLite.Simple.QQ (sql)
 import Database.SQLite.Simple.ToField (ToField (..))
@@ -251,6 +255,7 @@ import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval (RI2State (..))
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite.Common
+import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
 import Simplex.Messaging.Agent.Store.SQLite.Migrations (DownMigration (..), MTRError, Migration (..), MigrationsToRun (..), mtrErrorDescription)
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
 import qualified Simplex.Messaging.Crypto as C
@@ -386,13 +391,16 @@ connectDB path key = do
   pure db
   where
     prepare db = do
-      let exec = SQLite3.exec $ DB.connectionHandle db
+      let exec = SQLite3.exec $ SQL.connectionHandle $ DB.conn db
       unless (null key) . exec $ "PRAGMA key = " <> sqlString key <> ";"
-      exec "PRAGMA busy_timeout = 100;"
-      exec "PRAGMA foreign_keys = ON;"
-      -- exec "PRAGMA trusted_schema = OFF;"
-      exec "PRAGMA secure_delete = ON;"
-      exec "PRAGMA auto_vacuum = FULL;"
+      exec . fromQuery $
+        [sql|
+          PRAGMA busy_timeout = 100;
+          PRAGMA foreign_keys = ON;
+          -- PRAGMA trusted_schema = OFF;
+          PRAGMA secure_delete = ON;
+          PRAGMA auto_vacuum = FULL;
+        |]
 
 closeSQLiteStore :: SQLiteStore -> IO ()
 closeSQLiteStore st = atomically (takeTMVar $ dbConnection st) >>= DB.close
@@ -417,7 +425,7 @@ sqlString s = quote <> T.replace quote "''" (T.pack s) <> quote
 execSQL :: DB.Connection -> Text -> IO [Text]
 execSQL db query = do
   rs <- newIORef []
-  SQLite3.execWithCallback (DB.connectionHandle db) query (addSQLResultRow rs)
+  SQLite3.execWithCallback (SQL.connectionHandle $ DB.conn db) query (addSQLResultRow rs)
   reverse <$> readIORef rs
 
 addSQLResultRow :: IORef [Text] -> SQLite3.ColumnIndex -> [Text] -> [Maybe Text] -> IO ()
@@ -432,7 +440,7 @@ checkConstraint err action = action `E.catch` (pure . Left . handleSQLError err)
 
 handleSQLError :: StoreError -> SQLError -> StoreError
 handleSQLError err e
-  | DB.sqlError e == DB.ErrorConstraint = err
+  | SQL.sqlError e == SQL.ErrorConstraint = err
   | otherwise = SEInternal $ bshow e
 
 createUserRecord :: DB.Connection -> IO UserId
@@ -906,7 +914,7 @@ getSndMsgViaRcpt db connId sndMsgId =
       [sql|
         SELECT s.internal_id, m.msg_type, s.internal_hash, s.rcpt_internal_id, s.rcpt_status
         FROM snd_messages s
-        JOIN messages m ON s.internal_id = m.internal_id
+        JOIN messages m ON s.conn_id = m.conn_id AND s.internal_id = m.internal_id
         WHERE s.conn_id = ? AND s.internal_snd_id = ?
       |]
       (connId, sndMsgId)
@@ -976,8 +984,8 @@ getRcvMsg db connId agentMsgId =
           r.internal_id, m.internal_ts, r.broker_id, r.broker_ts, r.external_snd_id, r.integrity, r.internal_hash,
           m.msg_type, m.msg_body, s.internal_id, s.rcpt_status, r.user_ack
         FROM rcv_messages r
-        JOIN messages m ON r.internal_id = m.internal_id
-        LEFT JOIN snd_messages s ON s.rcpt_internal_id = r.internal_id
+        JOIN messages m ON r.conn_id = m.conn_id AND r.internal_id = m.internal_id
+        LEFT JOIN snd_messages s ON s.conn_id = r.conn_id AND s.rcpt_internal_id = r.internal_id
         WHERE r.conn_id = ? AND r.internal_id = ?
       |]
       (connId, agentMsgId)
@@ -992,9 +1000,9 @@ getLastMsg db connId msgId =
           r.internal_id, m.internal_ts, r.broker_id, r.broker_ts, r.external_snd_id, r.integrity, r.internal_hash,
           m.msg_type, m.msg_body, s.internal_id, s.rcpt_status, r.user_ack
         FROM rcv_messages r
-        JOIN messages m ON r.internal_id = m.internal_id
+        JOIN messages m ON r.conn_id = m.conn_id AND r.internal_id = m.internal_id
         JOIN connections c ON r.conn_id = c.conn_id AND c.last_internal_msg_id = r.internal_id
-        LEFT JOIN snd_messages s ON s.rcpt_internal_id = r.internal_id
+        LEFT JOIN snd_messages s ON s.conn_id = r.conn_id AND s.rcpt_internal_id = r.internal_id
         WHERE r.conn_id = ? AND r.broker_id = ?
       |]
       (connId, msgId)
@@ -1058,15 +1066,7 @@ deleteSndMsgsExpired db ttl = do
   cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime
   DB.execute
     db
-    [sql|
-      DELETE FROM messages
-      WHERE internal_id IN (
-        SELECT s.internal_id
-        FROM snd_messages s
-        JOIN messages m USING (internal_id)
-        WHERE m.internal_ts < ?
-      )
-    |]
+    "DELETE FROM messages WHERE internal_ts < ? AND internal_snd_id IS NOT NULL"
     (Only cutoffTs)
 
 createRatchetX3dhKeys :: DB.Connection -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> IO ()
@@ -1688,7 +1688,7 @@ insertSndQueue_ db connId' SndQueue {..} serverKeyHash_ = do
   DB.execute
     db
     [sql|
-      INSERT INTO snd_queues
+      INSERT OR REPLACE INTO snd_queues
         (host, port, snd_id, conn_id, snd_public_key, snd_private_key, e2e_pub_key, e2e_dh_secret, status, snd_queue_id, snd_primary, replace_snd_queue_id, smp_client_version, server_key_hash) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?);
     |]
     ((host server, port server, sndId, connId', sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret) :. (status, qId, primary, dbReplaceQueueId, smpClientVersion, serverKeyHash_))
@@ -1958,7 +1958,7 @@ updateHashRcv_ dbConn connId RcvMsgData {msgMeta, internalHash, internalRcvId} =
       WHERE conn_id = :conn_id
         AND last_internal_rcv_msg_id = :last_internal_rcv_msg_id;
     |]
-    [ ":last_external_snd_msg_id" := sndMsgId (msgMeta :: MsgMeta),
+    [ ":last_external_snd_msg_id" := msgMeta.sndMsgId,
       ":last_rcv_msg_hash" := internalHash,
       ":conn_id" := connId,
       ":last_internal_rcv_msg_id" := internalRcvId
@@ -2059,7 +2059,7 @@ createWithRandomId gVar create = tryCreate 3
       E.try (create id') >>= \case
         Right _ -> pure $ Right id'
         Left e
-          | DB.sqlError e == DB.ErrorConstraint -> tryCreate (n - 1)
+          | SQL.sqlError e == SQL.ErrorConstraint -> tryCreate (n - 1)
           | otherwise -> pure . Left . SEInternal $ bshow e
 
 randomId :: TVar ChaChaDRG -> Int -> IO ByteString

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

@@ -1,10 +1,13 @@
+{-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 
 module Simplex.Messaging.Agent.Store.SQLite.Common
   ( SQLiteStore (..),
     withConnection,
+    withConnection',
     withTransaction,
+    withTransaction',
     withTransactionCtx,
     dbBusyLoop,
   )
@@ -13,7 +16,8 @@ where
 import Control.Concurrent (threadDelay)
 import Data.Time.Clock (diffUTCTime, getCurrentTime)
 import Database.SQLite.Simple (SQLError)
-import qualified Database.SQLite.Simple as DB
+import qualified Database.SQLite.Simple as SQL
+import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
 import Simplex.Messaging.Util (diffToMilliseconds)
 import UnliftIO.Exception (bracket)
 import qualified UnliftIO.Exception as E
@@ -32,17 +36,23 @@ withConnection SQLiteStore {dbConnection} =
     (atomically $ takeTMVar dbConnection)
     (atomically . putTMVar dbConnection)
 
-withTransaction :: forall a. SQLiteStore -> (DB.Connection -> IO a) -> IO a
+withConnection' :: SQLiteStore -> (SQL.Connection -> IO a) -> IO a
+withConnection' st action = withConnection st $ action . DB.conn
+
+withTransaction :: SQLiteStore -> (DB.Connection -> IO a) -> IO a
 withTransaction = withTransactionCtx Nothing
 
-withTransactionCtx :: forall a. Maybe String -> SQLiteStore -> (DB.Connection -> IO a) -> IO a
-withTransactionCtx ctx_ st action = withConnection st $ \db -> dbBusyLoop (transactionWithCtx db)
+withTransaction' :: SQLiteStore -> (SQL.Connection -> IO a) -> IO a
+withTransaction' st action = withTransaction st $ action . DB.conn
+
+withTransactionCtx :: Maybe String -> SQLiteStore -> (DB.Connection -> IO a) -> IO a
+withTransactionCtx ctx_ st action = withConnection st $ dbBusyLoop . transactionWithCtx
   where
-    transactionWithCtx db = case ctx_ of
-      Nothing -> DB.withImmediateTransaction db (action db)
+    transactionWithCtx db@DB.Connection {conn} = case ctx_ of
+      Nothing -> SQL.withImmediateTransaction conn $ action db
       Just ctx -> do
         t1 <- getCurrentTime
-        r <- DB.withImmediateTransaction db (action db)
+        r <- SQL.withImmediateTransaction conn $ action db
         t2 <- getCurrentTime
         putStrLn $ "withTransactionCtx start :: " <> show t1 <> " :: " <> ctx
         putStrLn $ "withTransactionCtx end   :: " <> show t2 <> " :: " <> ctx <> " :: duration=" <> show (diffToMilliseconds $ diffUTCTime t2 t1)
@@ -54,7 +64,7 @@ dbBusyLoop action = loop 500 3000000
     loop :: Int -> Int -> IO a
     loop t tLim =
       action `E.catch` \(e :: SQLError) ->
-        if tLim > t && DB.sqlError e == DB.ErrorBusy
+        if tLim > t && SQL.sqlError e == SQL.ErrorBusy
           then do
             threadDelay t
             loop (t * 9 `div` 8) (tLim - t)

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

@@ -0,0 +1,101 @@
+{-# LANGUAGE DeriveGeneric #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE StrictData #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.DB
+  ( Connection (..),
+    SlowQueryStats (..),
+    open,
+    close,
+    execute,
+    execute_,
+    executeNamed,
+    executeMany,
+    query,
+    query_,
+    queryNamed,
+  )
+where
+
+import Control.Concurrent.STM
+import Control.Monad (when)
+import Data.Aeson (ToJSON (..))
+import qualified Data.Aeson as J
+import Data.Int (Int64)
+import Data.Time (diffUTCTime, getCurrentTime)
+import Database.SQLite.Simple (FromRow, NamedParam, Query, ToRow)
+import qualified Database.SQLite.Simple as SQL
+import GHC.Generics (Generic)
+import Simplex.Messaging.TMap (TMap)
+import qualified Simplex.Messaging.TMap as TM
+import Simplex.Messaging.Util (diffToMilliseconds)
+
+data Connection = Connection
+  { conn :: SQL.Connection,
+    slow :: TMap Query SlowQueryStats
+  }
+
+data SlowQueryStats = SlowQueryStats
+  { count :: Int64,
+    timeMax :: Int64,
+    timeAvg :: Int64
+  }
+  deriving (Show, Generic)
+
+instance ToJSON SlowQueryStats where toEncoding = J.genericToEncoding J.defaultOptions
+
+timeIt :: TMap Query SlowQueryStats -> Query -> IO a -> IO a
+timeIt slow sql a = do
+  t <- getCurrentTime
+  r <- a
+  t' <- getCurrentTime
+  let diff = diffToMilliseconds $ diffUTCTime t' t
+  atomically $ when (diff > 50) $ TM.alter (updateQueryStats diff) sql slow
+  pure r
+  where
+    updateQueryStats :: Int64 -> Maybe SlowQueryStats -> Maybe SlowQueryStats
+    updateQueryStats diff Nothing = Just $ SlowQueryStats 1 diff diff
+    updateQueryStats diff (Just SlowQueryStats {count, timeMax, timeAvg}) =
+      Just $
+        SlowQueryStats
+          { count = count + 1,
+            timeMax = max timeMax diff,
+            timeAvg = (timeAvg * count + diff) `div` (count + 1)
+          }
+
+open :: String -> IO Connection
+open f = do
+  conn <- SQL.open f
+  slow <- atomically $ TM.empty
+  pure Connection {conn, slow}
+
+close :: Connection -> IO ()
+close = SQL.close . conn
+
+execute :: ToRow q => Connection -> Query -> q -> IO ()
+execute Connection {conn, slow} sql = timeIt slow sql . SQL.execute conn sql
+{-# INLINE execute #-}
+
+execute_ :: Connection -> Query -> IO ()
+execute_ Connection {conn, slow} sql = timeIt slow sql $ SQL.execute_ conn sql
+{-# INLINE execute_ #-}
+
+executeNamed :: Connection -> Query -> [NamedParam] -> IO ()
+executeNamed Connection {conn, slow} sql = timeIt slow sql . SQL.executeNamed conn sql
+{-# INLINE executeNamed #-}
+
+executeMany :: ToRow q => Connection -> Query -> [q] -> IO ()
+executeMany Connection {conn, slow} sql = timeIt slow sql . SQL.executeMany conn sql
+{-# INLINE executeMany #-}
+
+query :: (ToRow q, FromRow r) => Connection -> Query -> q -> IO [r]
+query Connection {conn, slow} sql = timeIt slow sql . SQL.query conn sql
+{-# INLINE query #-}
+
+query_ :: FromRow r => Connection -> Query -> IO [r]
+query_ Connection {conn, slow} sql = timeIt slow sql $ SQL.query_ conn sql
+{-# INLINE query_ #-}
+
+queryNamed :: FromRow r => Connection -> Query -> [NamedParam] -> IO [r]
+queryNamed Connection {conn, slow} sql = timeIt slow sql . SQL.queryNamed conn sql
+{-# INLINE queryNamed #-}

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

@@ -64,6 +64,9 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230516_encrypted_rcv_m
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230531_switch_status
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230615_ratchet_sync
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230701_delivery_receipts
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230720_delete_expired_messages
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230722_indexes
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230814_indexes
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (dropPrefix, sumTypeJSON)
 import Simplex.Messaging.Transport.Client (TransportHost)
@@ -93,7 +96,10 @@ schemaMigrations =
     ("m20230516_encrypted_rcv_message_hashes", m20230516_encrypted_rcv_message_hashes, Just down_m20230516_encrypted_rcv_message_hashes),
     ("m20230531_switch_status", m20230531_switch_status, Just down_m20230531_switch_status),
     ("m20230615_ratchet_sync", m20230615_ratchet_sync, Just down_m20230615_ratchet_sync),
-    ("m20230701_delivery_receipts", m20230701_delivery_receipts, Just down_m20230701_delivery_receipts)
+    ("m20230701_delivery_receipts", m20230701_delivery_receipts, Just down_m20230701_delivery_receipts),
+    ("m20230720_delete_expired_messages", m20230720_delete_expired_messages, Just down_m20230720_delete_expired_messages),
+    ("m20230722_indexes", m20230722_indexes, Just down_m20230722_indexes),
+    ("m20230814_indexes", m20230814_indexes, Just down_m20230814_indexes)
   ]
 
 -- | The list of migrations in ascending order by date
@@ -103,7 +109,7 @@ app = sortOn name $ map migration schemaMigrations
     migration (name, up, down) = Migration {name, up = fromQuery up, down = fromQuery <$> down}
 
 get :: SQLiteStore -> [Migration] -> IO (Either MTRError MigrationsToRun)
-get st migrations = migrationsToRun migrations <$> withTransaction st getCurrent
+get st migrations = migrationsToRun migrations <$> withTransaction' st getCurrent
 
 getCurrent :: Connection -> IO [Migration]
 getCurrent db = map toMigration <$> DB.query_ db "SELECT name, down FROM migrations ORDER BY name ASC;"
@@ -113,11 +119,11 @@ getCurrent db = map toMigration <$> DB.query_ db "SELECT name, down FROM migrati
 run :: SQLiteStore -> MigrationsToRun -> IO ()
 run st = \case
   MTRUp [] -> pure ()
-  MTRUp ms -> mapM_ runUp ms >> withConnection st (`execSQL` "VACUUM;")
+  MTRUp ms -> mapM_ runUp ms >> withConnection' st (`execSQL` "VACUUM;")
   MTRDown ms -> mapM_ runDown $ reverse ms
   MTRNone -> pure ()
   where
-    runUp Migration {name, up, down} = withTransaction st $ \db -> do
+    runUp Migration {name, up, down} = withTransaction' st $ \db -> do
       when (name == "m20220811_onion_hosts") $ updateServers db
       insert db >> execSQL db up
       where
@@ -125,13 +131,13 @@ run st = \case
         updateServers db = forM_ (M.assocs extraSMPServerHosts) $ \(h, h') ->
           let hs = decodeLatin1 . strEncode $ ([h, h'] :: NonEmpty TransportHost)
            in DB.execute db "UPDATE servers SET host = ? WHERE host = ?" (hs, decodeLatin1 $ strEncode h)
-    runDown DownMigration {downName, downQuery} = withTransaction st $ \db -> do
+    runDown DownMigration {downName, downQuery} = withTransaction' st $ \db -> do
       execSQL db downQuery
       DB.execute db "DELETE FROM migrations WHERE name = ?" (Only downName)
     execSQL db = SQLite3.exec $ DB.connectionHandle db
 
 initialize :: SQLiteStore -> IO ()
-initialize st = withTransaction st $ \db -> do
+initialize st = withTransaction' st $ \db -> do
   cs :: [Text] <- map fromOnly <$> DB.query_ db "SELECT name FROM pragma_table_info('migrations')"
   case cs of
     [] -> createMigrations db

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

@@ -0,0 +1,20 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230720_delete_expired_messages where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20230720_delete_expired_messages :: Query
+m20230720_delete_expired_messages =
+  [sql|
+CREATE INDEX idx_messages_internal_snd_id_ts ON messages(internal_snd_id, internal_ts);
+
+DELETE FROM messages WHERE internal_snd_id IS NOT NULL AND internal_ts < datetime('now', '-3 days');
+|]
+
+down_m20230720_delete_expired_messages :: Query
+down_m20230720_delete_expired_messages =
+  [sql|
+DROP INDEX idx_messages_internal_snd_id_ts;
+|]

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

@@ -0,0 +1,20 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230722_indexes where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20230722_indexes :: Query
+m20230722_indexes =
+  [sql|
+CREATE INDEX idx_processed_ratchet_key_hashes_created_at ON processed_ratchet_key_hashes(created_at);
+CREATE INDEX idx_encrypted_rcv_message_hashes_created_at ON encrypted_rcv_message_hashes(created_at);
+|]
+
+down_m20230722_indexes :: Query
+down_m20230722_indexes =
+  [sql|
+DROP INDEX idx_encrypted_rcv_message_hashes_created_at;
+DROP INDEX idx_processed_ratchet_key_hashes_created_at;
+|]

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

@@ -0,0 +1,22 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230814_indexes where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20230814_indexes :: Query
+m20230814_indexes =
+  [sql|
+DROP INDEX idx_messages_internal_snd_id_ts;
+
+CREATE INDEX idx_messages_internal_ts ON messages(internal_ts);
+|]
+
+down_m20230814_indexes :: Query
+down_m20230814_indexes =
+  [sql|
+DROP INDEX idx_messages_internal_ts;
+
+CREATE INDEX idx_messages_internal_snd_id_ts ON messages(internal_snd_id, internal_ts);
+|]

src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql

@@ -467,3 +467,10 @@ CREATE INDEX idx_snd_messages_rcpt_internal_id ON snd_messages(
   conn_id,
   rcpt_internal_id
 );
+CREATE INDEX idx_processed_ratchet_key_hashes_created_at ON processed_ratchet_key_hashes(
+  created_at
+);
+CREATE INDEX idx_encrypted_rcv_message_hashes_created_at ON encrypted_rcv_message_hashes(
+  created_at
+);
+CREATE INDEX idx_messages_internal_ts ON messages(internal_ts);

src/Simplex/Messaging/Agent/TAsyncs.hs

@@ -1,6 +1,5 @@
 module Simplex.Messaging.Agent.TAsyncs where
 
-import Control.Concurrent.STM (stateTVar)
 import Control.Monad.IO.Unlift (MonadUnliftIO)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM

src/Simplex/Messaging/Agent/TRcvQueues.hs

@@ -1,5 +1,6 @@
+{-# LANGUAGE LambdaCase #-}
 module Simplex.Messaging.Agent.TRcvQueues
-  ( TRcvQueues,
+  ( TRcvQueues (getRcvQueues),
     empty,
     clear,
     deleteConn,
@@ -22,7 +23,7 @@ import Simplex.Messaging.Protocol (RecipientId, SMPServer)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 
-newtype TRcvQueues = TRcvQueues (TMap (UserId, SMPServer, RecipientId) RcvQueue)
+newtype TRcvQueues = TRcvQueues {getRcvQueues :: TMap (UserId, SMPServer, RecipientId) RcvQueue}
 
 empty :: STM TRcvQueues
 empty = TRcvQueues <$> TM.empty

src/Simplex/Messaging/Client.hs

@@ -10,8 +10,8 @@
 {-# LANGUAGE OverloadedLists #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
-{-# LANGUAGE TypeApplications #-}
 {-# LANGUAGE TupleSections #-}
+{-# LANGUAGE TypeApplications #-}
 
 -- |
 -- Module      : Simplex.Messaging.Client
@@ -40,6 +40,7 @@ module Simplex.Messaging.Client
     createSMPQueue,
     subscribeSMPQueue,
     subscribeSMPQueues,
+    streamSubscribeSMPQueues,
     getSMPMessage,
     subscribeSMPQueueNotifications,
     subscribeSMPQueuesNtfs,
@@ -69,6 +70,13 @@ module Simplex.Messaging.Client
     temporaryClientError,
     ServerTransmission,
     ClientCommand,
+
+    -- * For testing
+    ClientBatch (..),
+    PCTransmission,
+    batchClientTransmissions,
+    mkTransmission,
+    clientStub,
   )
 where
 
@@ -82,12 +90,10 @@ import Data.Aeson (FromJSON (..), ToJSON (..))
 import qualified Data.Aeson as J
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import Data.Either (rights)
-import Data.Foldable (foldl')
 import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.List (find)
-import Data.List.NonEmpty (NonEmpty (..), (<|))
+import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
 import Data.Maybe (fromMaybe)
 import Data.Time.Clock (UTCTime, getCurrentTime)
@@ -95,6 +101,7 @@ import GHC.Generics (Generic)
 import Network.Socket (ServiceName)
 import Numeric.Natural
 import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (dropPrefix, enumJSON)
 import Simplex.Messaging.Protocol as SMP
@@ -131,18 +138,50 @@ data PClient err msg = PClient
     pingErrorCount :: TVar Int,
     clientCorrId :: TVar Natural,
     sentCommands :: TMap CorrId (Request err msg),
-    sndQ :: TBQueue (NonEmpty SentRawTransmission),
+    sndQ :: TBQueue ByteString,
     rcvQ :: TBQueue (NonEmpty (SignedTransmission err msg)),
     msgQ :: Maybe (TBQueue (ServerTransmission msg))
   }
 
+clientStub :: ByteString -> STM (ProtocolClient err msg)
+clientStub sessionId = do
+  connected <- newTVar False
+  clientCorrId <- newTVar 0
+  sentCommands <- TM.empty
+  sndQ <- newTBQueue 100
+  rcvQ <- newTBQueue 100
+  return
+    ProtocolClient
+      { action = Nothing,
+        sessionId,
+        sessionTs = undefined,
+        thVersion = 5,
+        timeoutPerBlock = undefined,
+        blockSize = smpBlockSize,
+        batch = undefined,
+        client_ =
+          PClient
+            { connected,
+              transportSession = undefined,
+              transportHost = undefined,
+              tcpTimeout = undefined,
+              batchDelay = Nothing,
+              pingErrorCount = undefined,
+              clientCorrId,
+              sentCommands,
+              sndQ,
+              rcvQ,
+              msgQ = Nothing
+            }
+      }
+
 type SMPClient = ProtocolClient ErrorType SMP.BrokerMsg
 
 -- | Type for client command data
-type ClientCommand msg = (Maybe C.APrivateSignKey, QueueId, ProtoCommand msg)
+type ClientCommand msg = (Maybe C.APrivateSignKey, EntityId, ProtoCommand msg)
 
 -- | Type synonym for transmission from some SPM server queue.
-type ServerTransmission msg = (TransportSession msg, Version, SessionId, QueueId, msg)
+type ServerTransmission msg = (TransportSession msg, Version, SessionId, EntityId, msg)
 
 data HostMode
   = -- | prefer (or require) onion hosts when connecting via SOCKS proxy
@@ -246,11 +285,14 @@ defaultClientConfig =
     }
 
 data Request err msg = Request
-  { queueId :: QueueId,
-    responseVar :: TMVar (Response err msg)
+  { entityId :: EntityId,
+    responseVar :: TMVar (Either (ProtocolClientError err) msg)
   }
 
-type Response err msg = Either (ProtocolClientError err) msg
+data Response err msg = Response
+  { entityId :: EntityId,
+    response :: Either (ProtocolClientError err) msg
+  }
 
 chooseTransportHost :: NetworkConfig -> NonEmpty TransportHost -> Either (ProtocolClientError err) TransportHost
 chooseTransportHost NetworkConfig {socksProxy, hostMode, requiredHostMode} hosts =
@@ -355,7 +397,7 @@ getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize,
             `finally` disconnected c'
 
     send :: Transport c => ProtocolClient err msg -> THandle c -> IO ()
-    send ProtocolClient {client_ = PClient {sndQ}} h = forever $ atomically (readTBQueue sndQ) >>= tPut h batchDelay
+    send ProtocolClient {client_ = PClient {sndQ}} h = forever $ atomically (readTBQueue sndQ) >>= tPutLog h
 
     receive :: Transport c => ProtocolClient err msg -> THandle c -> IO ()
     receive ProtocolClient {client_ = PClient {rcvQ}} h = forever $ tGet h >>= atomically . writeTBQueue rcvQ
@@ -375,26 +417,27 @@ getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize,
     process c = forever $ atomically (readTBQueue $ rcvQ $ client_ c) >>= mapM_ (processMsg c)
 
     processMsg :: ProtocolClient err msg -> SignedTransmission err msg -> IO ()
-    processMsg c@ProtocolClient {client_ = PClient {sentCommands}} (_, _, (corrId, qId, respOrErr)) =
+    processMsg c@ProtocolClient {client_ = PClient {sentCommands}} (_, _, (corrId, entId, respOrErr)) =
       if B.null $ bs corrId
         then sendMsg respOrErr
         else do
           atomically (TM.lookup corrId sentCommands) >>= \case
             Nothing -> sendMsg respOrErr
-            Just Request {queueId, responseVar} -> atomically $ do
+            Just Request {entityId, responseVar} -> atomically $ do
               TM.delete corrId sentCommands
-              putTMVar responseVar $
-                if queueId == qId
-                  then case respOrErr of
-                    Left e -> Left $ PCEResponseError e
-                    Right r -> case protocolError r of
-                      Just e -> Left $ PCEProtocolError e
-                      _ -> Right r
-                  else Left . PCEUnexpectedResponse $ bshow respOrErr
+              putTMVar responseVar $ response entityId
       where
+        response entityId
+          | entityId == entId =
+            case respOrErr of
+              Left e -> Left $ PCEResponseError e
+              Right r -> case protocolError r of
+                Just e -> Left $ PCEProtocolError e
+                _ -> Right r
+          | otherwise = Left . PCEUnexpectedResponse $ bshow respOrErr
         sendMsg :: Either err msg -> IO ()
         sendMsg = \case
-          Right msg -> atomically $ mapM_ (`writeTBQueue` serverTransmission c qId msg) msgQ
+          Right msg -> atomically $ mapM_ (`writeTBQueue` serverTransmission c entId msg) msgQ
           Left e -> putStrLn $ "SMP client error: " <> show e
 
 proxyUsername :: TransportSession msg -> ByteString
@@ -470,14 +513,22 @@ subscribeSMPQueue c rpKey rId =
 
 -- | Subscribe to multiple SMP queues batching commands if supported.
 subscribeSMPQueues :: SMPClient -> NonEmpty (RcvPrivateSignKey, RecipientId) -> IO (NonEmpty (Either SMPClientError ()))
-subscribeSMPQueues c qs = sendProtocolCommands c cs >>= mapM response . L.zip qs
+subscribeSMPQueues c qs = sendProtocolCommands c cs >>= mapM (processSUBResponse c)
   where
     cs = L.map (\(rpKey, rId) -> (Just rpKey, rId, Cmd SRecipient SUB)) qs
-    response ((_, rId), r) = case r of
-      Right OK -> pure $ Right ()
-      Right cmd@MSG {} -> writeSMPMessage c rId cmd $> Right ()
-      Right r' -> pure . Left . PCEUnexpectedResponse $ bshow r'
-      Left e -> pure $ Left e
+
+streamSubscribeSMPQueues :: SMPClient -> NonEmpty (RcvPrivateSignKey, RecipientId) -> ([(RecipientId, Either SMPClientError ())] -> IO ()) -> IO ()
+streamSubscribeSMPQueues c qs cb = streamProtocolCommands c cs $ mapM process >=> cb
+  where
+    cs = L.map (\(rpKey, rId) -> (Just rpKey, rId, Cmd SRecipient SUB)) qs
+    process r@(Response rId _) = (rId,) <$> processSUBResponse c r
+
+processSUBResponse :: SMPClient -> Response ErrorType BrokerMsg -> IO (Either SMPClientError ())
+processSUBResponse c (Response rId r) = case r of
+  Right OK -> pure $ Right ()
+  Right cmd@MSG {} -> writeSMPMessage c rId cmd $> Right ()
+  Right r' -> pure . Left . PCEUnexpectedResponse $ bshow r'
+  Left e -> pure $ Left e
 
 writeSMPMessage :: SMPClient -> RecipientId -> BrokerMsg -> IO ()
 writeSMPMessage c rId msg = atomically $ mapM_ (`writeTBQueue` serverTransmission c rId msg) (msgQ $ client_ c)
@@ -523,12 +574,12 @@ enableSMPQueueNotifications c rpKey rId notifierKey rcvNtfPublicDhKey =
 
 -- | Enable notifications for the multiple queues for push notifications server.
 enableSMPQueuesNtfs :: SMPClient -> NonEmpty (RcvPrivateSignKey, RecipientId, NtfPublicVerifyKey, RcvNtfPublicDhKey) -> IO (NonEmpty (Either SMPClientError (NotifierId, RcvNtfPublicDhKey)))
-enableSMPQueuesNtfs c qs = L.map response <$> sendProtocolCommands c cs
+enableSMPQueuesNtfs c qs = L.map process <$> sendProtocolCommands c cs
   where
     cs = L.map (\(rpKey, rId, notifierKey, rcvNtfPublicDhKey) -> (Just rpKey, rId, Cmd SRecipient $ NKEY notifierKey rcvNtfPublicDhKey)) qs
-    response = \case
+    process (Response _ r) = case r of
       Right (NID nId rcvNtfSrvPublicDhKey) -> Right (nId, rcvNtfSrvPublicDhKey)
-      Right r -> Left . PCEUnexpectedResponse $ bshow r
+      Right r' -> Left . PCEUnexpectedResponse $ bshow r'
       Left e -> Left e
 
 -- | Disable notifications for the queue for push notifications server.
@@ -584,78 +635,122 @@ okSMPCommand cmd c pKey qId =
     r -> throwE . PCEUnexpectedResponse $ bshow r
 
 okSMPCommands :: PartyI p => Command p -> SMPClient -> NonEmpty (C.APrivateSignKey, QueueId) -> IO (NonEmpty (Either SMPClientError ()))
-okSMPCommands cmd c qs = L.map response <$> sendProtocolCommands c cs
+okSMPCommands cmd c qs = L.map process <$> sendProtocolCommands c cs
   where
     aCmd = Cmd sParty cmd
     cs = L.map (\(pKey, qId) -> (Just pKey, qId, aCmd)) qs
-    response = \case
+    process (Response _ r) = case r of
       Right OK -> Right ()
-      Right r -> Left . PCEUnexpectedResponse $ bshow r
+      Right r' -> Left . PCEUnexpectedResponse $ bshow r'
       Left e -> Left e
 
 -- | Send SMP command
 sendSMPCommand :: PartyI p => SMPClient -> Maybe C.APrivateSignKey -> QueueId -> Command p -> ExceptT SMPClientError IO BrokerMsg
 sendSMPCommand c pKey qId cmd = sendProtocolCommand c pKey qId (Cmd sParty cmd)
 
-type PCTransmission err msg = (SentRawTransmission, TMVar (Response err msg))
+type PCTransmission err msg = (SentRawTransmission, Request err msg)
 
 -- | Send multiple commands with batching and collect responses
--- It will result in Int overflow on 32 bit platform for a large number of blocks (~13.4k blocks / ~1.2m subscriptions)
--- TODO switch to timeout or TimeManager that supports Int64
-sendProtocolCommands :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Either (ProtocolClientError err) msg))
-sendProtocolCommands c@ProtocolClient {client_ = PClient {sndQ, tcpTimeout, batchDelay}, batch, blockSize, timeoutPerBlock} cs = do
-  (h :| ts) <- mapM (runExceptT . mkTransmission c) cs
-  let h' :: Either (ProtocolClientError err) (PCTransmission err msg, Int) = (,timeoutPerBlock) <$> h
-      batchSz = if batch then either (const 0) tSize h else 0
-      ts' :: NonEmpty (Either (ProtocolClientError err) (PCTransmission err msg, Int)) =
-        L.reverse . fst3 $ foldl' batchTimeouts ([h'], timeoutPerBlock, batchSz) ts
-      ts_ :: (Maybe (NonEmpty SentRawTransmission)) =
-        L.nonEmpty . map (fst . fst) . rights $ L.toList ts'
-  mapM_ (atomically . writeTBQueue sndQ) ts_
-  forConcurrently ts' $ \case
-    Right ((_t, r), bt) -> withTimeout c (tcpTimeout + bt) (atomically $ takeTMVar r)
-    Left e -> pure $ Left e
+sendProtocolCommands :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Response err msg))
+sendProtocolCommands c@ProtocolClient {batch, blockSize} cs = do
+  bs <- batchClientTransmissions batch blockSize <$> mapM (mkTransmission c) cs
+  validate . concat =<< mapM (sendBatch c) bs
   where
-    fst3 (x, _, _) = x
-    -- tSize calculation matches the batching logic in tPut that does actual breaking of transmissions into blocks
-    tSize :: PCTransmission err msg -> Int
-    tSize ((sig, t), _) = maybe 0 C.signatureSize sig + B.length t + 3 -- 1 byte for signature size + 2 bytes for transmission size
-    batchTimeouts :: (NonEmpty (Either (ProtocolClientError err) (PCTransmission err msg, Int)), Int, Int) -> Either (ProtocolClientError err) (PCTransmission err msg) -> (NonEmpty (Either (ProtocolClientError err) (PCTransmission err msg, Int)), Int, Int)
-    batchTimeouts (ts, bt, batchSz) = \case
-      Left e -> (Left e <| ts, bt, batchSz)
-      Right t
-        | not batch ->
-          (Right (t, bt') <| ts, bt', 0)
-        | batchSz' + 1 > blockSize ->
-          (Right (t, bt') <| ts, bt', tSz)
-        | otherwise -> -- same block in the batch
-          (Right (t, bt) <| ts, bt, batchSz') -- 1 byte for the number of transmissions in the batch
+    validate :: [Response err msg] -> IO (NonEmpty (Response err msg))
+    validate rs
+      | diff == 0 = pure $ L.fromList rs
+      | diff > 0 = do
+        putStrLn "send error: fewer responses than expected"
+        pure $ L.fromList $ rs <> replicate diff (Response "" $ Left $ PCETransportError TEBadBlock)
+      | otherwise = do
+        putStrLn "send error: more responses than expected"
+        pure $ L.fromList $ take (L.length cs) rs
         where
-          batchSz' = batchSz + tSz
-          bt' = bt + timeoutPerBlock + fromMaybe 0 batchDelay
-          tSz = tSize t
+          diff = L.length cs - length rs
+
+streamProtocolCommands :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> NonEmpty (ClientCommand msg) -> ([Response err msg] -> IO ()) -> IO ()
+streamProtocolCommands c@ProtocolClient {batch, blockSize} cs cb = do
+  bs <- batchClientTransmissions batch blockSize <$> mapM (mkTransmission c) cs
+  mapM_ (cb <=< sendBatch c) bs
+
+sendBatch :: ProtocolClient err msg -> ClientBatch err msg -> IO [Response err msg]
+sendBatch c@ProtocolClient {client_ = PClient {sndQ}} b = do
+  case b of
+    CBLargeTransmission Request {entityId} -> do
+      putStrLn "send error: large message"
+      pure [Response entityId $ Left $ PCETransportError TELargeMsg]
+    CBTransmissions s n rs -> do
+      when (n > 0) $ atomically $ writeTBQueue sndQ $ tEncodeBatch n s
+      mapConcurrently (getResponse c) rs
+    CBTransmission s r -> do
+      atomically $ writeTBQueue sndQ s
+      (: []) <$> getResponse c r
+
+data ClientBatch err msg
+  -- ByteString in CBTransmissions does not include count byte, it is added by tEncodeBatch
+  = CBTransmissions ByteString Int [Request err msg]
+  | CBTransmission ByteString (Request err msg)
+  | CBLargeTransmission (Request err msg)
+
+-- | encodes and batches transmissions into blocks
+batchClientTransmissions :: forall err msg. Bool -> Int -> NonEmpty (PCTransmission err msg) -> [ClientBatch err msg]
+batchClientTransmissions batch blkSize
+  | batch = reverse . mkBatch []
+  | otherwise = map mkBatch1 . L.toList
+  where
+    mkBatch :: [ClientBatch err msg] -> NonEmpty (PCTransmission err msg) -> [ClientBatch err msg]
+    mkBatch bs ts =
+      let (b, ts_) = encodeBatch "" 0 [] ts
+          bs' = b : bs
+       in maybe bs' (mkBatch bs') ts_
+    mkBatch1 :: PCTransmission err msg -> ClientBatch err msg
+    mkBatch1 (t, r)
+      | B.length s <= blkSize - 2 = CBTransmission s r
+      | otherwise = CBLargeTransmission r
+      where
+        s = tEncode t
+    encodeBatch :: ByteString -> Int -> [Request err msg] -> NonEmpty (PCTransmission err msg) -> (ClientBatch err msg, Maybe (NonEmpty (PCTransmission err msg)))
+    encodeBatch s n rs ts@((t, r) :| ts_)
+      | B.length s' <= blkSize - 3 && n < 255 =
+        case L.nonEmpty ts_ of
+          Just ts' -> encodeBatch s' n' rs' ts'
+          Nothing -> (CBTransmissions s' n' (reverse rs'), Nothing)
+      | n == 0 = (CBLargeTransmission r, L.nonEmpty ts_)
+      | otherwise = (CBTransmissions s n (reverse rs), Just ts)
+      where
+        s' = s <> smpEncode (Large $ tEncode t)
+        n' = n + 1
+        rs' = r : rs
 
 -- | Send Protocol command
-sendProtocolCommand :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> Maybe C.APrivateSignKey -> QueueId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg
-sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ, tcpTimeout}} pKey qId cmd = do
-  (t, r) <- mkTransmission c (pKey, qId, cmd)
-  ExceptT $ sendRecv t r
+sendProtocolCommand :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> Maybe C.APrivateSignKey -> EntityId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg
+sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, batch, blockSize} pKey entId cmd =
+  ExceptT $ uncurry sendRecv =<< mkTransmission c (pKey, entId, cmd)
   where
     -- two separate "atomically" needed to avoid blocking
-    sendRecv :: SentRawTransmission -> TMVar (Response err msg) -> IO (Response err msg)
-    sendRecv t r = atomically (writeTBQueue sndQ [t]) >> withTimeout c tcpTimeout (atomically $ takeTMVar r)
+    sendRecv :: SentRawTransmission -> Request err msg -> IO (Either (ProtocolClientError err) msg)
+    sendRecv t r
+      | B.length s > blockSize - 2 = pure $ Left $ PCETransportError TELargeMsg
+      | otherwise = atomically (writeTBQueue sndQ s) >> response <$> getResponse c r
+      where
+        s
+          | batch = tEncodeBatch 1 . smpEncode . Large $ tEncode t
+          | otherwise = tEncode t
 
-withTimeout :: ProtocolClient err msg -> Int -> IO (Either (ProtocolClientError err) msg) -> IO (Either (ProtocolClientError err) msg)
-withTimeout ProtocolClient {client_ = PClient {pingErrorCount}} t a = do
-  timeout t a >>= \case
-    Just r -> atomically (writeTVar pingErrorCount 0) >> pure r
-    _ -> pure $ Left PCEResponseTimeout
+-- TODO switch to timeout or TimeManager that supports Int64
+getResponse :: ProtocolClient err msg -> Request err msg -> IO (Response err msg)
+getResponse ProtocolClient {client_ = PClient {tcpTimeout, pingErrorCount}} Request {entityId, responseVar} = do
+  response <-
+    timeout tcpTimeout (atomically (takeTMVar responseVar)) >>= \case
+      Just r -> atomically (writeTVar pingErrorCount 0) $> r
+      Nothing -> pure $ Left PCEResponseTimeout
+  pure Response {entityId, response}
 
-mkTransmission :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> ClientCommand msg -> ExceptT (ProtocolClientError err) IO (PCTransmission err msg)
-mkTransmission ProtocolClient {sessionId, thVersion, client_ = PClient {clientCorrId, sentCommands}} (pKey, qId, cmd) = do
-  corrId <- liftIO $ atomically getNextCorrId
-  let t = signTransmission $ encodeTransmission thVersion sessionId (corrId, qId, cmd)
-  r <- liftIO . atomically $ mkRequest corrId
+mkTransmission :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> ClientCommand msg -> IO (PCTransmission err msg)
+mkTransmission ProtocolClient {sessionId, thVersion, client_ = PClient {clientCorrId, sentCommands}} (pKey, entId, cmd) = do
+  corrId <- atomically getNextCorrId
+  let t = signTransmission $ encodeTransmission thVersion sessionId (corrId, entId, cmd)
+  r <- atomically $ mkRequest corrId
   pure (t, r)
   where
     getNextCorrId :: STM CorrId
@@ -664,8 +759,8 @@ mkTransmission ProtocolClient {sessionId, thVersion, client_ = PClient {clientCo
       pure . CorrId $ bshow i
     signTransmission :: ByteString -> SentRawTransmission
     signTransmission t = ((`C.sign` t) <$> pKey, t)
-    mkRequest :: CorrId -> STM (TMVar (Response err msg))
+    mkRequest :: CorrId -> STM (Request err msg)
     mkRequest corrId = do
-      r <- newEmptyTMVar
-      TM.insert corrId (Request qId r) sentCommands
+      r <- Request entId <$> newEmptyTMVar
+      TM.insert corrId r sentCommands
       pure r

src/Simplex/Messaging/Client/Agent.hs

@@ -14,6 +14,7 @@ module Simplex.Messaging.Client.Agent where
 import Control.Concurrent (forkIO)
 import Control.Concurrent.Async (Async, uninterruptibleCancel)
 import Control.Logger.Simple
+import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
 import Control.Monad.Trans.Except
@@ -39,7 +40,7 @@ import Simplex.Messaging.Protocol (BrokerMsg, ProtocolServer (..), QueueId, SMPS
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
-import Simplex.Messaging.Util (catchAll_, tryE, ($>>=), toChunks)
+import Simplex.Messaging.Util (catchAll_, ($>>=), toChunks)
 import System.Timeout (timeout)
 import UnliftIO (async)
 import UnliftIO.Exception (Exception)
@@ -272,13 +273,13 @@ withSMP ca srv action = (getSMPServerClient' ca srv >>= action) `catchE` logSMPE
 subscribeQueue :: SMPClientAgent -> SMPServer -> (SMPSub, C.APrivateSignKey) -> ExceptT SMPClientError IO ()
 subscribeQueue ca srv sub = do
   atomically $ addPendingSubscription ca srv sub
-  withSMP ca srv $ \smp -> subscribe_ smp `catchE` handleError
+  withSMP ca srv $ \smp -> subscribe_ smp `catchE` handleErr
   where
     subscribe_ smp = do
       smpSubscribe smp sub
       atomically $ addSubscription ca srv sub
 
-    handleError e = do
+    handleErr e = do
       atomically . when (e /= PCENetworkError && e /= PCEResponseTimeout) $
         removePendingSubscription ca srv $ fst sub
       throwE e

src/Simplex/Messaging/Crypto.hs

@@ -149,6 +149,7 @@ where
 
 import Control.Concurrent.STM
 import Control.Exception (Exception)
+import Control.Monad
 import Control.Monad.Except
 import Control.Monad.Trans.Except
 import Crypto.Cipher.AES (AES256)

src/Simplex/Messaging/Crypto/Ratchet.hs

@@ -15,6 +15,7 @@
 module Simplex.Messaging.Crypto.Ratchet where
 
 import Control.Monad.Except
+import Control.Monad.IO.Class
 import Control.Monad.Trans.Except
 import Crypto.Cipher.AES (AES256)
 import Crypto.Hash (SHA512)
@@ -38,7 +39,6 @@ import Simplex.Messaging.Crypto
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (blobFieldDecoder, parseE, parseE')
-import Simplex.Messaging.Util (tryE)
 import Simplex.Messaging.Version
 
 currentE2EEncryptVersion :: Version

src/Simplex/Messaging/Notifications/Server.hs

@@ -12,8 +12,8 @@
 
 module Simplex.Messaging.Notifications.Server where
 
-import Control.Concurrent.STM (stateTVar)
 import Control.Logger.Simple
+import Control.Monad
 import Control.Monad.Except
 import Control.Monad.Reader
 import Data.Bifunctor (second)

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

@@ -30,7 +30,7 @@ import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
 import Text.Read (readMaybe)
 
 ntfServerVersion :: String
-ntfServerVersion = "1.5.1"
+ntfServerVersion = "1.6.0"
 
 defaultSMPBatchDelay :: Int
 defaultSMPBatchDelay = 10000

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

@@ -12,7 +12,9 @@ module Simplex.Messaging.Notifications.Server.Push.APNS where
 
 import Control.Exception (Exception)
 import Control.Logger.Simple
+import Control.Monad
 import Control.Monad.Except
+import Control.Monad.IO.Class
 import Crypto.Hash.Algorithms (SHA256 (..))
 import qualified Crypto.PubKey.ECC.ECDSA as EC
 import qualified Crypto.PubKey.ECC.Types as ECT

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

@@ -1,5 +1,6 @@
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedRecordDot #-}
 {-# LANGUAGE OverloadedStrings #-}
 
 module Simplex.Messaging.Notifications.Server.Stats where
@@ -56,30 +57,30 @@ newNtfServerStats ts = do
 
 getNtfServerStatsData :: NtfServerStats -> STM NtfServerStatsData
 getNtfServerStatsData s = do
-  _fromTime <- readTVar $ fromTime (s :: NtfServerStats)
-  _tknCreated <- readTVar $ tknCreated s
-  _tknVerified <- readTVar $ tknVerified s
-  _tknDeleted <- readTVar $ tknDeleted s
-  _subCreated <- readTVar $ subCreated s
-  _subDeleted <- readTVar $ subDeleted s
-  _ntfReceived <- readTVar $ ntfReceived s
-  _ntfDelivered <- readTVar $ ntfDelivered s
-  _activeTokens <- getPeriodStatsData $ activeTokens s
-  _activeSubs <- getPeriodStatsData $ activeSubs s
+  _fromTime <- readTVar s.fromTime
+  _tknCreated <- readTVar s.tknCreated
+  _tknVerified <- readTVar s.tknVerified
+  _tknDeleted <- readTVar s.tknDeleted
+  _subCreated <- readTVar s.subCreated
+  _subDeleted <- readTVar s.subDeleted
+  _ntfReceived <- readTVar s.ntfReceived
+  _ntfDelivered <- readTVar s.ntfDelivered
+  _activeTokens <- getPeriodStatsData s.activeTokens
+  _activeSubs <- getPeriodStatsData s.activeSubs
   pure NtfServerStatsData {_fromTime, _tknCreated, _tknVerified, _tknDeleted, _subCreated, _subDeleted, _ntfReceived, _ntfDelivered, _activeTokens, _activeSubs}
 
 setNtfServerStats :: NtfServerStats -> NtfServerStatsData -> STM ()
 setNtfServerStats s d = do
-  writeTVar (fromTime (s :: NtfServerStats)) $! _fromTime (d :: NtfServerStatsData)
-  writeTVar (tknCreated s) $! _tknCreated d
-  writeTVar (tknVerified s) $! _tknVerified d
-  writeTVar (tknDeleted s) $! _tknDeleted d
-  writeTVar (subCreated s) $! _subCreated d
-  writeTVar (subDeleted s) $! _subDeleted d
-  writeTVar (ntfReceived s) $! _ntfReceived d
-  writeTVar (ntfDelivered s) $! _ntfDelivered d
-  setPeriodStats (activeTokens s) (_activeTokens d)
-  setPeriodStats (activeSubs s) (_activeSubs d)
+  writeTVar s.fromTime $! d._fromTime
+  writeTVar s.tknCreated $! _tknCreated d
+  writeTVar s.tknVerified $! _tknVerified d
+  writeTVar s.tknDeleted $! _tknDeleted d
+  writeTVar s.subCreated $! _subCreated d
+  writeTVar s.subDeleted $! _subDeleted d
+  writeTVar s.ntfReceived $! _ntfReceived d
+  writeTVar s.ntfDelivered $! _ntfDelivered d
+  setPeriodStats s.activeTokens (_activeTokens d)
+  setPeriodStats s.activeSubs (_activeSubs d)
 
 instance StrEncoding NtfServerStatsData where
   strEncode NtfServerStatsData {_fromTime, _tknCreated, _tknVerified, _tknDeleted, _subCreated, _subDeleted, _ntfReceived, _ntfDelivered, _activeTokens, _activeSubs} =

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

@@ -5,6 +5,8 @@
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 
+{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
+
 module Simplex.Messaging.Notifications.Server.StoreLog
   ( StoreLog,
     NtfStoreLogRecord (..),

src/Simplex/Messaging/Protocol.hs

@@ -15,6 +15,7 @@
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StandaloneDeriving #-}
+{-# LANGUAGE StrictData #-}
 {-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}
 {-# LANGUAGE TypeFamilyDependencies #-}
@@ -132,12 +133,15 @@ module Simplex.Messaging.Protocol
     noAuthSrv,
 
     -- * TCP transport functions
+    TransportBatch(..),
     tPut,
+    tPutLog,
     tGet,
     tParse,
     tDecodeParseValidate,
     tEncode,
     tEncodeBatch,
+    batchTransmissions,
 
     -- * exports for tests
     CommandTag (..),
@@ -1246,50 +1250,70 @@ instance Encoding CommandError where
 
 -- | Send signed SMP transmission to TCP transport.
 tPut :: Transport c => THandle c -> Maybe Int -> NonEmpty SentRawTransmission -> IO [Either TransportError ()]
-tPut th delay_ trs 
-  | batch th = tPutBatch [] $ L.map tEncode trs
-  | otherwise = forM (L.toList trs) $ tPutLog . tEncode
+tPut th delay_ = fmap concat . mapM tPutBatch . batchTransmissions (batch th) (blockSize th)
   where
-    tPutBatch :: [Either TransportError ()] -> NonEmpty ByteString -> IO [Either TransportError ()]
-    tPutBatch rs ts = do
+    tPutBatch :: TransportBatch -> IO [Either TransportError ()]
+    tPutBatch = \case
+      TBLargeTransmission -> [Left TELargeMsg] <$ putStrLn "tPut error: large message"
+      TBTransmissions n s -> replicate n <$> (tPutLog th (tEncodeBatch n s) <* mapM_ threadDelay delay_)
+      TBTransmission s -> (: []) <$> tPutLog th s
+
+tPutLog :: Transport c => THandle c -> ByteString -> IO (Either TransportError ())
+tPutLog th s = do
+  r <- tPutBlock th s
+  case r of
+    Left e -> putStrLn ("tPut error: " <> show e)
+    _ -> pure ()
+  pure r
+
+-- ByteString does not include length byte, it is added by tEncodeBatch
+data TransportBatch = TBTransmissions Int ByteString | TBTransmission ByteString |  TBLargeTransmission
+
+-- | encodes and batches transmissions into blocks, 
+batchTransmissions :: Bool -> Int -> NonEmpty SentRawTransmission -> [TransportBatch]
+batchTransmissions batch bSize
+  | batch = reverse . mkBatch [] . L.map tEncode
+  | otherwise = map (mkBatch1 . tEncode) . L.toList
+  where
+    mkBatch :: [TransportBatch] -> NonEmpty ByteString -> [TransportBatch]
+    mkBatch rs ts =
       let (n, s, ts_) = encodeBatch 0 "" ts
-      r <- if n == 0 then largeMsg else replicate n <$> tPutLog (tEncodeBatch n s)
-      let rs' = rs <> r
-      case ts_ of
-        Just ts' -> mapM_ threadDelay delay_ >> tPutBatch rs' ts'
-        _ -> pure rs'
-    largeMsg = putStrLn "tPut error: large message" >> pure [Left TELargeMsg]
-    tPutLog s = do
-      r <- tPutBlock th s
-      case r of
-        Left e -> putStrLn ("tPut error: " <> show e)
-        _ -> pure ()
-      pure r
+          r = if n == 0 then TBLargeTransmission else TBTransmissions n s
+          rs' = r : rs
+      in case ts_ of
+          Just ts' -> mkBatch rs' ts'
+          _ -> rs'
+    mkBatch1 :: ByteString -> TransportBatch
+    mkBatch1 s = if B.length s > bSize - 2 then TBLargeTransmission else TBTransmission s
     encodeBatch :: Int -> ByteString -> NonEmpty ByteString -> (Int, ByteString, Maybe (NonEmpty ByteString))
     encodeBatch n s ts@(t :| ts_)
       | n == 255 = (n, s, Just ts)
       | otherwise =
         let s' = s <> smpEncode (Large t)
             n' = n + 1
-         in if B.length s' > blockSize th - 1 -- one byte is reserved for the number of messages in the batch
+         in if B.length s' > bSize - 3 -- one byte is reserved for the number of messages in the batch
               then (n,s,) $ if n == 0 then L.nonEmpty ts_ else Just ts
               else case L.nonEmpty ts_ of
                 Just ts' -> encodeBatch n' s' ts'
                 _ -> (n', s', Nothing)
 
-tEncode :: (Maybe C.ASignature, ByteString) -> ByteString
+tEncode :: SentRawTransmission -> ByteString
 tEncode (sig, t) = smpEncode (C.signatureBytes sig) <> t
+{-# INLINE tEncode #-}
 
 tEncodeBatch :: Int -> ByteString -> ByteString
 tEncodeBatch n s = lenEncode n `B.cons` s
+{-# INLINE tEncodeBatch #-}
 
 encodeTransmission :: ProtocolEncoding e c => Version -> ByteString -> Transmission c -> ByteString
 encodeTransmission v sessionId (CorrId corrId, queueId, command) =
   smpEncode (sessionId, corrId, queueId) <> encodeProtocol v command
+{-# INLINE encodeTransmission #-}
 
 -- | Receive and parse transmission from the TCP transport (ignoring any trailing padding).
 tGetParse :: Transport c => THandle c -> IO (NonEmpty (Either TransportError RawTransmission))
 tGetParse th = eitherList (tParse $ batch th) <$> tGetBlock th
+{-# INLINE tGetParse #-}
 
 tParse :: Bool -> ByteString -> NonEmpty (Either TransportError RawTransmission)
 tParse batch s

src/Simplex/Messaging/Server.hs

@@ -1,5 +1,6 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE OverloadedRecordDot #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE KindSignatures #-}
@@ -31,6 +32,7 @@ module Simplex.Messaging.Server
     disconnectTransport,
     verifyCmdSignature,
     dummyVerifyCmd,
+    randomId,
   )
 where
 
@@ -47,16 +49,19 @@ import qualified Data.ByteString.Char8 as B
 import Data.Either (fromRight, partitionEithers)
 import Data.Functor (($>))
 import Data.Int (Int64)
-import Data.List (intercalate)
+import Data.List (intercalate, sort)
 import qualified Data.List.NonEmpty as L
 import qualified Data.Map.Strict as M
-import Data.Maybe (isNothing)
+import Data.Maybe (fromMaybe, isNothing)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1)
 import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime)
 import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Time.Format.ISO8601 (iso8601Show)
 import Data.Type.Equality
+import GHC.Conc (listThreads, threadStatus)
+import GHC.Conc.Sync (threadLabel)
+import GHC.Stats (getRTSStats)
 import GHC.TypeLits (KnownNat)
 import Network.Socket (ServiceName, Socket, socketToHandle)
 import Simplex.Messaging.Agent.Lock
@@ -65,12 +70,12 @@ import Simplex.Messaging.Encoding (Encoding (smpEncode))
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.Control
-import Simplex.Messaging.Server.Env.STM
+import Simplex.Messaging.Server.Env.STM as Env
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.MsgStore
 import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.QueueStore
-import Simplex.Messaging.Server.QueueStore.STM
+import Simplex.Messaging.Server.QueueStore.STM as QS
 import Simplex.Messaging.Server.Stats
 import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.TMap (TMap)
@@ -111,8 +116,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
   restoreServerMessages
   restoreServerStats
   raceAny_
-    ( serverThread s subscribedQ subscribers subscriptions cancelSub :
-      serverThread s ntfSubscribedQ notifiers ntfSubscriptions (\_ -> pure ()) :
+    ( serverThread s "server subscribedQ" subscribedQ subscribers subscriptions cancelSub :
+      serverThread s "server ntfSubscribedQ" ntfSubscribedQ Env.notifiers ntfSubscriptions (\_ -> pure ()) :
       map runServer transports <> expireMessagesThread_ cfg <> serverStatsThread_ cfg <> controlPortThread_ cfg
     )
     `finally` withLock (savingLock s) "final" (saveServer False)
@@ -128,12 +133,14 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
     serverThread ::
       forall s.
       Server ->
+      String ->
       (Server -> TQueue (QueueId, Client)) ->
       (Server -> TMap QueueId Client) ->
       (Client -> TMap QueueId s) ->
       (s -> IO ()) ->
       M ()
-    serverThread s subQ subs clientSubs unsub = forever $ do
+    serverThread s label subQ subs clientSubs unsub = forever $ do
+      labelMyThread label
       atomically updateSubscribers
         $>>= endPreviousSubscriptions
         >>= liftIO . mapM_ unsub
@@ -150,6 +157,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
           TM.lookupInsert qId clnt (subs s) $>>= clientToBeNotified
         endPreviousSubscriptions :: (QueueId, Client) -> M (Maybe s)
         endPreviousSubscriptions (qId, c) = do
+          labelMyThread $ label <> ".endPreviousSubscriptions"
           void . forkIO . atomically $
             writeTBQueue (sndQ c) [(CorrId "", qId, END)]
           atomically $ TM.lookupDelete qId (clientSubs c)
@@ -163,6 +171,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
       ms <- asks msgStore
       quota <- asks $ msgQueueQuota . config
       let interval = checkInterval expCfg * 1000000
+      labelMyThread "expireMessages"
       forever $ do
         liftIO $ threadDelay' interval
         old <- liftIO $ expireBeforeEpoch expCfg
@@ -178,6 +187,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
 
     logServerStats :: Int64 -> Int64 -> FilePath -> M ()
     logServerStats startAt logInterval statsFilePath = do
+      labelMyThread "logServerStats"
       initialDelay <- (startAt -) . fromIntegral . (`div` 1000000_000000) . diffTimeToPicoseconds . utctDayTime <$> liftIO getCurrentTime
       liftIO $ putStrLn $ "server stats log enabled: " <> statsFilePath
       liftIO $ threadDelay' $ 1000000 * (initialDelay + if initialDelay < 0 then 86400 else 0)
@@ -222,9 +232,10 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
           threadDelay' interval
 
     runClient :: Transport c => TProxy c -> c -> M ()
-    runClient _ h = do
+    runClient tp h = do
       kh <- asks serverIdentity
       smpVRange <- asks $ smpServerVRange . config
+      labelMyThread $ "smp handshake for " <> transportName tp
       liftIO (runExceptT $ smpServerHandshake h kh smpVRange) >>= \case
         Right th -> runClientTransport th
         Left _ -> pure ()
@@ -238,10 +249,13 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
       srv <- asks server
       cpStarted <- newEmptyTMVarIO
       u <- askUnliftIO
-      liftIO $ runTCPServer cpStarted port $ runCPClient u srv
+      liftIO $ do
+        labelMyThread "control port server"
+        runTCPServer cpStarted port $ runCPClient u srv
       where
         runCPClient :: UnliftIO (ReaderT Env IO) -> Server -> Socket -> IO ()
         runCPClient u srv sock = do
+          labelMyThread "control port client"
           h <- socketToHandle sock ReadWriteMode
           hSetBuffering h LineBuffering
           hSetNewlineMode h universalNewlineMode
@@ -257,7 +271,15 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
             processCP h = \case
               CPSuspend -> hPutStrLn h "suspend not implemented"
               CPResume -> hPutStrLn h "resume not implemented"
-              CPClients -> hPutStrLn h "clients not implemented"
+              CPClients -> do
+                Server {subscribers} <- unliftIO u $ asks server
+                clients <- readTVarIO subscribers
+                hPutStrLn h $ "Clients: " <> show (length clients)
+                forM_ (M.toList clients) $ \(cid, Client {sessionId, connected, activeAt, subscriptions}) -> do
+                  hPutStrLn h . B.unpack $ "Client " <> encode cid <> " $" <> encode sessionId
+                  readTVarIO connected >>= hPutStrLn h . ("  connected: " <>)  . show
+                  readTVarIO activeAt >>= hPutStrLn h . ("  activeAt: " <>)  . B.unpack . strEncode
+                  readTVarIO subscriptions >>= hPutStrLn h . ("  subscriptions: " <>)  . show . M.size
               CPStats -> do
                 ServerStats {fromTime, qCreated, qSecured, qDeleted, msgSent, msgRecv, msgSentNtf, msgRecvNtf, qCount, msgCount} <- unliftIO u $ asks serverStats
                 putStat "fromTime" fromTime
@@ -273,12 +295,21 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
                 where
                   putStat :: Show a => String -> TVar a -> IO ()
                   putStat label var = readTVarIO var >>= \v -> hPutStrLn h $ label <> ": " <> show v
+              CPStatsRTS -> getRTSStats >>= hPutStrLn h . show
+              CPThreads -> do
+                threads <- liftIO listThreads
+                hPutStrLn h $ "Threads: " <> show (length threads)
+                forM_ (sort threads) $ \tid -> do
+                  label <- threadLabel tid
+                  status <- threadStatus tid
+                  hPutStrLn h $ show tid <> " (" <> show status <> ") " <> fromMaybe "" label
               CPSave -> withLock (savingLock srv) "control" $ do
                 hPutStrLn h "saving server state..."
                 unliftIO u $ saveServer True
                 hPutStrLn h "server state saved!"
-              CPHelp -> hPutStrLn h "commands: stats, save, help, quit"
+              CPHelp -> hPutStrLn h "commands: stats, stats-rts, clients, threads, save, help, quit"
               CPQuit -> pure ()
+              CPSkip -> pure ()
 
 runClientTransport :: Transport c => THandle c -> M ()
 runClientTransport th@THandle {thVersion, sessionId} = do
@@ -287,6 +318,7 @@ runClientTransport th@THandle {thVersion, sessionId} = do
   c <- atomically $ newClient q thVersion sessionId ts
   s <- asks server
   expCfg <- asks $ inactiveClientExpiration . config
+  labelMyThread . B.unpack $ "client $" <> encode c.sessionId
   raceAny_ ([liftIO $ send th c, client c s, receive th c] <> disconnectThread_ c expCfg)
     `finally` clientDisconnected c
   where
@@ -317,12 +349,14 @@ cancelSub sub =
     _ -> return ()
 
 receive :: Transport c => THandle c -> Client -> M ()
-receive th Client {rcvQ, sndQ, activeAt} = forever $ do
-  ts <- L.toList <$> liftIO (tGet th)
-  atomically . writeTVar activeAt =<< liftIO getSystemTime
-  as <- partitionEithers <$> mapM cmdAction ts
-  write sndQ $ fst as
-  write rcvQ $ snd as
+receive th Client {rcvQ, sndQ, activeAt, sessionId} = do
+  labelMyThread . B.unpack $ "client $" <> encode sessionId <> " receive"
+  forever $ do
+    ts <- L.toList <$> liftIO (tGet th)
+    atomically . writeTVar activeAt =<< liftIO getSystemTime
+    as <- partitionEithers <$> mapM cmdAction ts
+    write sndQ $ fst as
+    write rcvQ $ snd as
   where
     cmdAction :: SignedTransmission ErrorType Cmd -> M (Either (Transmission BrokerMsg) (Maybe QueueRec, Transmission Cmd))
     cmdAction (sig, signed, (corrId, queueId, cmdOrError)) =
@@ -336,10 +370,12 @@ receive th Client {rcvQ, sndQ, activeAt} = forever $ do
     write q = mapM_ (atomically . writeTBQueue q) . L.nonEmpty
 
 send :: Transport c => THandle c -> Client -> IO ()
-send h@THandle {thVersion = v} Client {sndQ, sessionId, activeAt} = forever $ do
-  ts <- atomically $ L.sortWith tOrder <$> readTBQueue sndQ
-  void . liftIO . tPut h Nothing $ L.map ((Nothing,) . encodeTransmission v sessionId) ts
-  atomically . writeTVar activeAt =<< liftIO getSystemTime
+send h@THandle {thVersion = v} Client {sndQ, sessionId, activeAt} = do
+  labelMyThread . B.unpack $ "client $" <> encode sessionId <> " send"
+  forever $ do
+    ts <- atomically $ L.sortWith tOrder <$> readTBQueue sndQ
+    void . liftIO . tPut h Nothing $ L.map ((Nothing,) . encodeTransmission v sessionId) ts
+    atomically . writeTVar activeAt =<< liftIO getSystemTime
   where
     tOrder :: Transmission BrokerMsg -> Int
     tOrder (_, _, cmd) = case cmd of
@@ -348,7 +384,8 @@ send h@THandle {thVersion = v} Client {sndQ, sessionId, activeAt} = forever $ do
       _ -> 1
 
 disconnectTransport :: Transport c => THandle c -> client -> (client -> TVar SystemTime) -> ExpirationConfig -> IO ()
-disconnectTransport THandle {connection} c activeAt expCfg = do
+disconnectTransport THandle {connection, sessionId} c activeAt expCfg = do
+  labelMyThread . B.unpack $ "client $" <> encode sessionId <> " disconnectTransport"
   let interval = checkInterval expCfg * 1000000
   forever . liftIO $ do
     threadDelay' interval
@@ -404,7 +441,8 @@ dummyKeyEd448 :: C.PublicKey 'C.Ed448
 dummyKeyEd448 = "MEMwBQYDK2VxAzoA6ibQc9XpkSLtwrf7PLvp81qW/etiumckVFImCMRdftcG/XopbOSaq9qyLhrgJWKOLyNrQPNVvpMA"
 
 client :: forall m. (MonadUnliftIO m, MonadReader Env m) => Client -> Server -> m ()
-client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Server {subscribedQ, ntfSubscribedQ, notifiers} =
+client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Server {subscribedQ, ntfSubscribedQ, notifiers} = do
+  labelMyThread . B.unpack $ "client $" <> encode clnt.sessionId <> " commands"
   forever $
     atomically (readTBQueue rcvQ)
       >>= mapM processCommand
@@ -722,11 +760,11 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
             encrypt msgFlags body =
               let encBody = EncRcvMsgBody $ C.cbEncryptMaxLenBS (rcvDhSecret qr) (C.cbNonce msgId') body
                in RcvMessage msgId' msgTs' msgFlags encBody
-            msgId' = msgId (msg :: Message)
-            msgTs' = msgTs (msg :: Message)
+            msgId' = msg.msgId
+            msgTs' = msg.msgTs
 
         setDelivered :: Sub -> Message -> STM Bool
-        setDelivered s msg = tryPutTMVar (delivered s) $ msgId (msg :: Message)
+        setDelivered s msg = tryPutTMVar (delivered s) $ msg.msgId
 
         getStoreMsgQueue :: T.Text -> RecipientId -> m MsgQueue
         getStoreMsgQueue name rId = time (name <> " getMsgQueue") $ do
@@ -826,7 +864,7 @@ restoreServerMessages = asks (storeMsgsFile . config) >>= mapM_ restoreMessages
                     | maybe True (systemSeconds msgTs >=) old_ -> isNothing <$> writeMsg q msg
                     | otherwise -> pure False
                   MessageQuota {} -> writeMsg q msg $> False
-              when logFull . logError . decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode (msgId (msg :: Message))
+              when logFull . logError . decodeLatin1 $ "message queue " <> strEncode rId <> " is full, message not restored: " <> strEncode msg.msgId
             updateMsgV1toV3 QueueRec {rcvDhSecret} RcvMessage {msgId, msgTs, msgFlags, msgBody = EncRcvMsgBody body} = do
               let nonce = C.cbNonce msgId
               msgBody <- liftEither . first (msgErr "v1 message decryption") $ C.maxLenBS =<< C.cbDecrypt rcvDhSecret nonce body

src/Simplex/Messaging/Server/Control.hs

@@ -11,9 +11,12 @@ data ControlProtocol
   | CPResume
   | CPClients
   | CPStats
+  | CPStatsRTS
+  | CPThreads
   | CPSave
   | CPHelp
   | CPQuit
+  | CPSkip
 
 instance StrEncoding ControlProtocol where
   strEncode = \case
@@ -21,16 +24,22 @@ instance StrEncoding ControlProtocol where
     CPResume -> "resume"
     CPClients -> "clients"
     CPStats -> "stats"
+    CPStatsRTS -> "stats-rts"
+    CPThreads -> "threads"
     CPSave -> "save"
     CPHelp -> "help"
     CPQuit -> "quit"
+    CPSkip -> ""
   strP =
     A.takeTill (== ' ') >>= \case
       "suspend" -> pure CPSuspend
       "resume" -> pure CPResume
       "clients" -> pure CPClients
       "stats" -> pure CPStats
+      "stats-rts" -> pure CPStatsRTS
+      "threads" -> pure CPThreads
       "save" -> pure CPSave
       "help" -> pure CPHelp
       "quit" -> pure CPQuit
+      "" -> pure CPSkip
       _ -> fail "bad ControlProtocol command"

src/Simplex/Messaging/Transport/Client.hs

@@ -21,7 +21,6 @@ module Simplex.Messaging.Transport.Client
 where
 
 import Control.Applicative (optional)
-import Control.Monad.Except
 import Control.Monad.IO.Unlift
 import Data.Aeson (FromJSON (..), ToJSON (..))
 import qualified Data.Attoparsec.ByteString.Char8 as A

src/Simplex/Messaging/Transport/HTTP2/Client.hs

@@ -9,7 +9,7 @@ module Simplex.Messaging.Transport.HTTP2.Client where
 import Control.Concurrent.Async
 import Control.Exception (IOException, try)
 import qualified Control.Exception as E
-import Control.Monad.Except
+import Control.Monad
 import Data.ByteString.Char8 (ByteString)
 import Data.Functor (($>))
 import Data.Time (UTCTime, getCurrentTime)

src/Simplex/Messaging/Transport/Server.hs

@@ -17,9 +17,8 @@ module Simplex.Messaging.Transport.Server
 where
 
 import Control.Applicative ((<|>))
-import Control.Concurrent.STM (stateTVar)
 import Control.Logger.Simple
-import Control.Monad.Except
+import Control.Monad
 import Control.Monad.IO.Unlift
 import qualified Crypto.Store.X509 as SX
 import Data.Default (def)
@@ -33,7 +32,7 @@ import qualified Network.TLS as T
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
-import Simplex.Messaging.Util (catchAll_, tshow)
+import Simplex.Messaging.Util (catchAll_, labelMyThread, tshow)
 import System.Exit (exitFailure)
 import System.Mem.Weak (Weak, deRefWeak)
 import UnliftIO.Concurrent
@@ -64,6 +63,7 @@ runTransportServer :: forall c m. (Transport c, MonadUnliftIO m) => TMVar Bool -
 runTransportServer started port serverParams cfg server = do
   u <- askUnliftIO
   let tCfg = serverTransportConfig cfg
+  labelMyThread $ "transport server for " <> transportName (TProxy :: TProxy c)
   liftIO . runTCPServer started port $ \conn ->
     E.bracket
       (connectTLS Nothing tCfg serverParams conn >>= getServerConnection tCfg)

src/Simplex/Messaging/Util.hs

@@ -5,10 +5,11 @@ module Simplex.Messaging.Util where
 
 import Control.Concurrent (threadDelay)
 import qualified Control.Exception as E
+import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
-import Control.Monad.Trans.Except
 import Data.Bifunctor (first)
+import qualified Data.ByteString as BW
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Int (Int64)
@@ -19,6 +20,8 @@ import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeUtf8With)
 import Data.Time (NominalDiffTime)
+import GHC.Conc
+import Numeric (showHex)
 import UnliftIO.Async
 import qualified UnliftIO.Exception as UE
 
@@ -66,14 +69,6 @@ liftEitherWith :: (MonadError e' m) => (e -> e') -> Either e a -> m a
 liftEitherWith f = liftEither . first f
 {-# INLINE liftEitherWith #-}
 
-tryError :: MonadError e m => m a -> m (Either e a)
-tryError action = (Right <$> action) `catchError` (pure . Left)
-{-# INLINE tryError #-}
-
-tryE :: Monad m => ExceptT e m a -> ExceptT e m (Either e a)
-tryE m = (Right <$> m) `catchE` (pure . Left)
-{-# INLINE tryE #-}
-
 liftE :: (e -> e') -> ExceptT e IO a -> ExceptT e' IO a
 liftE f a = ExceptT $ first f <$> runExceptT a
 {-# INLINE liftE #-}
@@ -157,3 +152,6 @@ diffToMicroseconds diff = fromIntegral ((truncate $ diff * 1000000) :: Integer)
 
 diffToMilliseconds :: NominalDiffTime -> Int64
 diffToMilliseconds diff = fromIntegral ((truncate $ diff * 1000) :: Integer)
+
+labelMyThread :: MonadIO m => String -> m ()
+labelMyThread label = liftIO $ myThreadId >>= (`labelThread` label)

stack.yaml

@@ -17,7 +17,7 @@
 #
 # resolver: ./custom-snapshot.yaml
 # resolver: https://example.com/snapshots/2018-01-01.yaml
-resolver: lts-18.28
+resolver: nightly-2023-08-22
 
 # User packages to be built.
 # Various formats can be used as shown in the example below.
@@ -47,15 +47,17 @@ extra-deps:
   - text-short-0.1.5@sha256:962c6228555debdc46f758d0317dea16e5240d01419b42966674b08a5c3d8fa6,3498
   - time-compat-1.9.6.1@sha256:42d8f2e08e965e1718917d54ad69e1d06bd4b87d66c41dc7410f59313dba4ed1,5033
   - github: simplex-chat/aeson
-    commit: 3eb66f9a68f103b5f1489382aad89f5712a64db7
+    commit: 68330dce8208173c6acf5f62b23acb500ab5d873
   - github: kazu-yamamoto/http2
     commit: b5a1b7200cf5bc7044af34ba325284271f6dff25
   # - ../direct-sqlcipher
   - github: simplex-chat/direct-sqlcipher
-    commit: 34309410eb2069b029b8fc1872deb1e0db123294
+    commit: f814ee68b16a9447fbb467ccc8f29bdd3546bfd9
   # - ../sqlcipher-simple
   - github: simplex-chat/sqlcipher-simple
-    commit: 5e154a2aeccc33ead6c243ec07195ab673137221
+    commit: a46bd361a19376c5211f1058908fc0ae6bf42446
+  - github: simplex-chat/network-transport
+    commit: 0013798272a683e35ca38d2fdaf480942311fba8
   # - ../hs-tls/core
   # - github: simplex-chat/hs-tls
   #   commit: f6cc753611f80af300401cfae63846e9d7c40d9e

tests/AgentTests.hs

@@ -16,7 +16,6 @@ import AgentTests.FunctionalAPITests (functionalAPITests)
 import AgentTests.MigrationTests (migrationTests)
 import AgentTests.NotificationTests (notificationTests)
 import AgentTests.SQLiteTests (storeTests)
-import AgentTests.SchemaDump (schemaDumpTest)
 import Control.Concurrent
 import Control.Monad (forM_)
 import Data.ByteString.Char8 (ByteString)
@@ -42,7 +41,6 @@ agentTests (ATransport t) = do
   describe "Functional API" $ functionalAPITests (ATransport t)
   describe "Notification tests" $ notificationTests (ATransport t)
   describe "SQLite store" storeTests
-  describe "SQLite schema dump" schemaDumpTest
   describe "Migration tests" migrationTests
   describe "SMP agent protocol syntax" $ syntaxTests t
   describe "Establishing duplex connection" $ do

tests/AgentTests/ConnectionRequestTests.hs

@@ -3,6 +3,8 @@
 {-# LANGUAGE OverloadedLists #-}
 {-# LANGUAGE OverloadedStrings #-}
 
+{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
+
 module AgentTests.ConnectionRequestTests where
 
 import Data.ByteString (ByteString)

tests/AgentTests/FunctionalAPITests.hs

@@ -59,7 +59,6 @@ import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Transport (ATransport (..))
-import Simplex.Messaging.Util (tryError)
 import Simplex.Messaging.Version
 import System.Directory (copyFile, renameFile)
 import Test.Hspec
@@ -231,6 +230,8 @@ functionalAPITests t = do
       withSmpServer t testAcceptContactAsync
     it "should delete connections using async command when server connection fails" $
       testDeleteConnectionAsync t
+    it "join connection when reply queue creation fails" $
+      testJoinConnectionAsyncReplyError t
   describe "Users" $ do
     it "should create and delete user with connections" $
       withSmpServer t testUsers
@@ -337,10 +338,19 @@ testServerMatrix2 t runTest = do
   it "2 servers" $ withSmpServer t . withSmpServerOn t testPort2 $ runTest initAgentServers2
 
 runTestCfg2 :: AgentConfig -> AgentConfig -> AgentMsgId -> (AgentClient -> AgentClient -> AgentMsgId -> IO ()) -> IO ()
-runTestCfg2 aliceCfg bobCfg baseMsgId runTest = do
-  alice <- getSMPAgentClient' aliceCfg initAgentServers testDB
-  bob <- getSMPAgentClient' bobCfg initAgentServers testDB2
-  runTest alice bob baseMsgId
+runTestCfg2 aCfg bCfg baseMsgId runTest =
+  withAgentClientsCfg2 aCfg bCfg $ \a b -> runTest a b baseMsgId
+
+withAgentClientsCfg2 :: AgentConfig -> AgentConfig -> (AgentClient -> AgentClient -> IO ()) -> IO ()
+withAgentClientsCfg2 aCfg bCfg runTest = do
+  a <- getSMPAgentClient' aCfg initAgentServers testDB
+  b <- getSMPAgentClient' bCfg initAgentServers testDB2
+  runTest a b
+  disconnectAgentClient a
+  disconnectAgentClient b
+
+withAgentClients2 :: (AgentClient -> AgentClient -> IO ()) -> IO ()
+withAgentClients2 = withAgentClientsCfg2 agentCfg agentCfg
 
 runAgentClientTest :: HasCallStack => AgentClient -> AgentClient -> AgentMsgId -> IO ()
 runAgentClientTest alice bob baseId = do
@@ -423,10 +433,8 @@ noMessages c err = tryGet `shouldReturn` ()
         _ -> return ()
 
 testAsyncInitiatingOffline :: HasCallStack => IO ()
-testAsyncInitiatingOffline = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
-  runRight_ $ do
+testAsyncInitiatingOffline =
+  withAgentClients2 $ \alice bob -> runRight_ $ do
     (bobId, cReq) <- createConnection alice 1 True SCMInvitation Nothing
     disconnectAgentClient alice
     aliceId <- joinConnection bob 1 True cReq "bob's connInfo"
@@ -440,10 +448,8 @@ testAsyncInitiatingOffline = do
     exchangeGreetings alice' bobId bob aliceId
 
 testAsyncJoiningOfflineBeforeActivation :: HasCallStack => IO ()
-testAsyncJoiningOfflineBeforeActivation = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
-  runRight_ $ do
+testAsyncJoiningOfflineBeforeActivation =
+  withAgentClients2 $ \alice bob -> runRight_ $ do
     (bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing
     aliceId <- joinConnection bob 1 True qInfo "bob's connInfo"
     disconnectAgentClient bob
@@ -457,10 +463,8 @@ testAsyncJoiningOfflineBeforeActivation = do
     exchangeGreetings alice bobId bob' aliceId
 
 testAsyncBothOffline :: HasCallStack => IO ()
-testAsyncBothOffline = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
-  runRight_ $ do
+testAsyncBothOffline =
+  withAgentClients2 $ \alice bob -> runRight_ $ do
     (bobId, cReq) <- createConnection alice 1 True SCMInvitation Nothing
     disconnectAgentClient alice
     aliceId <- joinConnection bob 1 True cReq "bob's connInfo"
@@ -477,9 +481,7 @@ testAsyncBothOffline = do
     exchangeGreetings alice' bobId bob' aliceId
 
 testAsyncServerOffline :: HasCallStack => ATransport -> IO ()
-testAsyncServerOffline t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+testAsyncServerOffline t = withAgentClients2 $ \alice bob -> do
   -- create connection and shutdown the server
   (bobId, cReq) <- withSmpServerStoreLogOn t testPort $ \_ ->
     runRight $ createConnection alice 1 True SCMInvitation Nothing
@@ -508,9 +510,7 @@ testAsyncHelloTimeout = do
   let vr11 = mkVersionRange 1 1
       smpCfgV1 = (smpCfg agentCfg) {serverVRange = vr11}
       agentCfgV1 = agentCfg {smpAgentVRange = vr11, smpClientVRange = vr11, e2eEncryptVRange = vr11, smpCfg = smpCfgV1}
-  alice <- getSMPAgentClient' agentCfgV1 initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg {helloTimeout = 1} initAgentServers testDB2
-  runRight_ $ do
+  withAgentClientsCfg2 agentCfgV1 agentCfg {helloTimeout = 1} $ \alice bob -> runRight_ $ do
     (_, cReq) <- createConnection alice 1 True SCMInvitation Nothing
     disconnectAgentClient alice
     aliceId <- joinConnection bob 1 True cReq "bob's connInfo"
@@ -553,6 +553,8 @@ testAllowConnectionClientRestart t = do
         get bob ##> ("", aliceId, CON)
 
         exchangeGreetingsMsgId 4 alice2 bobId bob aliceId
+    disconnectAgentClient alice2
+    disconnectAgentClient bob
 
 testIncreaseConnAgentVersion :: HasCallStack => ATransport -> IO ()
 testIncreaseConnAgentVersion t = do
@@ -607,6 +609,8 @@ testIncreaseConnAgentVersion t = do
       exchangeGreetingsMsgId 12 alice3 bobId bob3 aliceId
       checkVersion alice3 bobId 3
       checkVersion bob3 aliceId 3
+    disconnectAgentClient alice3
+    disconnectAgentClient bob3
 
 checkVersion :: AgentClient -> ConnId -> Version -> ExceptT AgentErrorType IO ()
 checkVersion c connId v = do
@@ -638,6 +642,8 @@ testIncreaseConnAgentVersionMaxCompatible t = do
       exchangeGreetingsMsgId 6 alice2 bobId bob2 aliceId
       checkVersion alice2 bobId 3
       checkVersion bob2 aliceId 3
+    disconnectAgentClient alice2
+    disconnectAgentClient bob2
 
 testIncreaseConnAgentVersionStartDifferentVersion :: HasCallStack => ATransport -> IO ()
 testIncreaseConnAgentVersionStartDifferentVersion t = do
@@ -661,6 +667,8 @@ testIncreaseConnAgentVersionStartDifferentVersion t = do
       exchangeGreetingsMsgId 6 alice2 bobId bob aliceId
       checkVersion alice2 bobId 3
       checkVersion bob aliceId 3
+    disconnectAgentClient alice2
+    disconnectAgentClient bob
 
 testDeliverClientRestart :: HasCallStack => ATransport -> IO ()
 testDeliverClientRestart t = do
@@ -690,6 +698,8 @@ testDeliverClientRestart t = do
 
       get bob2 ##> ("", aliceId, SENT 6)
       get alice =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False
+  disconnectAgentClient alice
+  disconnectAgentClient bob2
 
 testDuplicateMessage :: HasCallStack => ATransport -> IO ()
 testDuplicateMessage t = do
@@ -738,6 +748,8 @@ testDuplicateMessage t = do
       6 <- sendMessage alice2 bobId SMP.noMsgFlags "hello 3"
       get alice2 ##> ("", bobId, SENT 6)
       get bob2 =##> \case ("", c, Msg "hello 3") -> c == aliceId; _ -> False
+  disconnectAgentClient alice2
+  disconnectAgentClient bob2
 
 testSkippedMessages :: HasCallStack => ATransport -> IO ()
 testSkippedMessages t = do
@@ -785,25 +797,20 @@ testSkippedMessages t = do
       get alice2 ##> ("", bobId, SENT 9)
       get bob2 =##> \case ("", c, Msg "hello 6") -> c == aliceId; _ -> False
       ackMessage bob2 aliceId 6 Nothing
+  disconnectAgentClient alice2
+  disconnectAgentClient bob2
 
 testRatchetSync :: HasCallStack => ATransport -> IO ()
-testRatchetSync t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+testRatchetSync t = withAgentClients2 $ \alice bob ->
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     (aliceId, bobId, bob2) <- setupDesynchronizedRatchet alice bob
     runRight $ do
       ConnectionStats {ratchetSyncState} <- synchronizeRatchet bob2 aliceId False
       liftIO $ ratchetSyncState `shouldBe` RSStarted
-
       get alice =##> ratchetSyncP bobId RSAgreed
-
       get bob2 =##> ratchetSyncP aliceId RSAgreed
-
       get alice =##> ratchetSyncP bobId RSOk
-
       get bob2 =##> ratchetSyncP aliceId RSOk
-
       exchangeGreetingsMsgIds alice bobId 12 bob2 aliceId 9
 
 setupDesynchronizedRatchet :: HasCallStack => AgentClient -> AgentClient -> IO (ConnId, ConnId, AgentClient)
@@ -866,9 +873,7 @@ ratchetSyncP' cId rss = \case
   _ -> False
 
 testRatchetSyncServerOffline :: HasCallStack => ATransport -> IO ()
-testRatchetSyncServerOffline t = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
+testRatchetSyncServerOffline t = withAgentClients2 $ \alice bob -> do
   (aliceId, bobId, bob2) <- withSmpServerStoreMsgLogOn t testPort $ \_ ->
     setupDesynchronizedRatchet alice bob
 
@@ -884,16 +889,12 @@ testRatchetSyncServerOffline t = do
         [ ratchetSyncP' bobId RSAgreed,
           serverUpP
         ]
-
       liftIO . getInAnyOrder bob2 $
         [ ratchetSyncP' aliceId RSAgreed,
           serverUpP
         ]
-
       get alice =##> ratchetSyncP bobId RSOk
-
       get bob2 =##> ratchetSyncP aliceId RSOk
-
       exchangeGreetingsMsgIds alice bobId 12 bob2 aliceId 9
 
 serverUpP :: ATransmission 'Agent -> Bool
@@ -907,32 +908,24 @@ testRatchetSyncClientRestart t = do
   bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
   (aliceId, bobId, bob2) <- withSmpServerStoreMsgLogOn t testPort $ \_ ->
     setupDesynchronizedRatchet alice bob
-
   ("", "", DOWN _ _) <- nGet alice
   ("", "", DOWN _ _) <- nGet bob2
-
   ConnectionStats {ratchetSyncState} <- runRight $ synchronizeRatchet bob2 aliceId False
   liftIO $ ratchetSyncState `shouldBe` RSStarted
-
   disconnectAgentClient bob2
-
   bob3 <- getSMPAgentClient' agentCfg initAgentServers testDB2
-
   withSmpServerStoreMsgLogOn t testPort $ \_ -> do
     runRight_ $ do
       ("", "", UP _ _) <- nGet alice
-
       subscribeConnection bob3 aliceId
-
       get alice =##> ratchetSyncP bobId RSAgreed
-
       get bob3 =##> ratchetSyncP aliceId RSAgreed
-
       get alice =##> ratchetSyncP bobId RSOk
-
       get bob3 =##> ratchetSyncP aliceId RSOk
-
       exchangeGreetingsMsgIds alice bobId 12 bob3 aliceId 9
+  disconnectAgentClient alice
+  disconnectAgentClient bob
+  disconnectAgentClient bob3
 
 testRatchetSyncSuspendForeground :: HasCallStack => ATransport -> IO ()
 testRatchetSyncSuspendForeground t = do
@@ -957,17 +950,16 @@ testRatchetSyncSuspendForeground t = do
         [ ratchetSyncP' bobId RSAgreed,
           serverUpP
         ]
-
       liftIO . getInAnyOrder bob2 $
         [ ratchetSyncP' aliceId RSAgreed,
           serverUpP
         ]
-
       get alice =##> ratchetSyncP bobId RSOk
-
       get bob2 =##> ratchetSyncP aliceId RSOk
-
       exchangeGreetingsMsgIds alice bobId 12 bob2 aliceId 9
+  disconnectAgentClient alice
+  disconnectAgentClient bob
+  disconnectAgentClient bob2
 
 testRatchetSyncSimultaneous :: HasCallStack => ATransport -> IO ()
 testRatchetSyncSimultaneous t = do
@@ -991,17 +983,16 @@ testRatchetSyncSimultaneous t = do
         [ ratchetSyncP' bobId RSAgreed,
           serverUpP
         ]
-
       liftIO . getInAnyOrder bob2 $
         [ ratchetSyncP' aliceId RSAgreed,
           serverUpP
         ]
-
       get alice =##> ratchetSyncP bobId RSOk
-
       get bob2 =##> ratchetSyncP aliceId RSOk
-
       exchangeGreetingsMsgIds alice bobId 12 bob2 aliceId 9
+  disconnectAgentClient alice
+  disconnectAgentClient bob
+  disconnectAgentClient bob2
 
 makeConnection :: AgentClient -> AgentClient -> ExceptT AgentErrorType IO (ConnId, ConnId)
 makeConnection alice bob = makeConnectionForUsers alice 1 bob 1
@@ -1025,6 +1016,7 @@ testInactiveClientDisconnected t = do
     runRight_ $ do
       (connId, _cReq) <- createConnection alice 1 True SCMInvitation Nothing
       nGet alice ##> ("", "", DOWN testSMPServer [connId])
+    disconnectAgentClient alice
 
 testActiveClientNotDisconnected :: ATransport -> IO ()
 testActiveClientNotDisconnected t = do
@@ -1035,6 +1027,7 @@ testActiveClientNotDisconnected t = do
     runRight_ $ do
       (connId, _cReq) <- createConnection alice 1 True SCMInvitation Nothing
       keepSubscribing alice connId ts
+    disconnectAgentClient alice
   where
     keepSubscribing :: AgentClient -> ConnId -> SystemTime -> ExceptT AgentErrorType IO ()
     keepSubscribing alice connId ts = do
@@ -1054,10 +1047,8 @@ testActiveClientNotDisconnected t = do
     milliseconds ts = systemSeconds ts * 1000 + fromIntegral (systemNanoseconds ts `div` 1000000)
 
 testSuspendingAgent :: IO ()
-testSuspendingAgent = do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
-  b <- getSMPAgentClient' agentCfg initAgentServers testDB2
-  runRight_ $ do
+testSuspendingAgent =
+  withAgentClients2 $ \a b -> runRight_ $ do
     (aId, bId) <- makeConnection a b
     4 <- sendMessage a bId SMP.noMsgFlags "hello"
     get a ##> ("", bId, SENT 4)
@@ -1072,9 +1063,7 @@ testSuspendingAgent = do
     get b =##> \case ("", c, Msg "hello 2") -> c == aId; _ -> False
 
 testSuspendingAgentCompleteSending :: ATransport -> IO ()
-testSuspendingAgentCompleteSending t = do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
-  b <- getSMPAgentClient' agentCfg initAgentServers testDB2
+testSuspendingAgentCompleteSending t = withAgentClients2 $ \a b -> do
   (aId, bId) <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ do
     (aId, bId) <- makeConnection a b
     4 <- sendMessage a bId SMP.noMsgFlags "hello"
@@ -1104,9 +1093,7 @@ testSuspendingAgentCompleteSending t = do
     ackMessage a bId 6 Nothing
 
 testSuspendingAgentTimeout :: ATransport -> IO ()
-testSuspendingAgentTimeout t = do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
-  b <- getSMPAgentClient' agentCfg initAgentServers testDB2
+testSuspendingAgentTimeout t = withAgentClients2 $ \a b -> do
   (aId, _) <- withSmpServer t . runRight $ do
     (aId, bId) <- makeConnection a b
     4 <- sendMessage a bId SMP.noMsgFlags "hello"
@@ -1159,6 +1146,8 @@ testBatchedSubscriptions nCreate nDel t = do
     delete b aIds'
     deleteFail a bIds'
     deleteFail b aIds'
+  disconnectAgentClient a
+  disconnectAgentClient b
   where
     subscribe :: AgentClient -> [ConnId] -> ExceptT AgentErrorType IO ()
     subscribe c cs = do
@@ -1189,10 +1178,8 @@ testBatchedSubscriptions nCreate nDel t = do
         pure res
 
 testAsyncCommands :: IO ()
-testAsyncCommands = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
-  runRight_ $ do
+testAsyncCommands =
+  withAgentClients2 $ \alice bob -> runRight_ $ do
     bobId <- createConnectionAsync alice 1 "1" True SCMInvitation
     ("1", bobId', INV (ACR _ qInfo)) <- get alice
     liftIO $ bobId' `shouldBe` bobId
@@ -1246,12 +1233,11 @@ testAsyncCommandsRestore t = do
       subscribeConnection alice' bobId
       ("1", _, INV _) <- get alice'
       pure ()
+  disconnectAgentClient alice'
 
 testAcceptContactAsync :: IO ()
-testAcceptContactAsync = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
-  runRight_ $ do
+testAcceptContactAsync =
+  withAgentClients2 $ \alice bob -> runRight_ $ do
     (_, qInfo) <- createConnection alice 1 True SCMContact Nothing
     aliceId <- joinConnection bob 1 True qInfo "bob's connInfo"
     ("", _, REQ invId _ "bob's connInfo") <- get alice
@@ -1306,12 +1292,53 @@ testDeleteConnectionAsync t = do
     get a =##> \case ("", c, DEL_CONN) -> c `elem` connIds; _ -> False
     get a =##> \case ("", c, DEL_CONN) -> c `elem` connIds; _ -> False
     liftIO $ noMessages a "nothing else should be delivered to alice"
+  disconnectAgentClient a
+
+testJoinConnectionAsyncReplyError :: HasCallStack => ATransport -> IO ()
+testJoinConnectionAsyncReplyError t = do
+  let initAgentServersSrv2 = initAgentServers {smp = userServers [noAuthSrv testSMPServer2]}
+  a <- getSMPAgentClient' agentCfg initAgentServers testDB
+  b <- getSMPAgentClient' agentCfg initAgentServersSrv2 testDB2
+  (aId, bId) <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ do
+    bId <- createConnectionAsync a 1 "1" True SCMInvitation
+    ("1", bId', INV (ACR _ qInfo)) <- get a
+    liftIO $ bId' `shouldBe` bId
+    aId <- joinConnectionAsync b 1 "2" True qInfo "bob's connInfo"
+    liftIO $ threadDelay 500000
+    ConnectionStats {rcvQueuesInfo = [], sndQueuesInfo = [SndQueueInfo {}]} <- getConnectionServers b aId
+    pure (aId, bId)
+  nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
+  withSmpServerOn t testPort2 $ do
+    ("2", aId', OK) <- get b
+    liftIO $ aId' `shouldBe` aId
+    confId <- withSmpServerStoreLogOn t testPort $ \_ -> do
+      pGet a >>= \case
+        ("", "", APC _ (UP _ [_])) -> do
+          ("", _, CONF confId _ "bob's connInfo") <- get a
+          pure confId
+        ("", _, APC _ (CONF confId _ "bob's connInfo")) -> do
+          ("", "", UP _ [_]) <- nGet a
+          pure confId
+        r -> error $ "unexpected response " <> show r
+    nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
+    runRight_ $ do
+      allowConnectionAsync a "3" bId confId "alice's connInfo"
+      liftIO $ threadDelay 500000
+      ConnectionStats {rcvQueuesInfo = [RcvQueueInfo {}], sndQueuesInfo = [SndQueueInfo {}]} <- getConnectionServers b aId
+      pure ()
+    withSmpServerStoreLogOn t testPort $ \_ -> runRight_ $ do
+      pGet a =##> \case ("3", c, APC _ OK) -> c == bId; ("", "", APC _ (UP _ [c])) -> c == bId; _ -> False
+      pGet a =##> \case ("3", c, APC _ OK) -> c == bId; ("", "", APC _ (UP _ [c])) -> c == bId; _ -> False
+      get a ##> ("", bId, CON)
+      get b ##> ("", aId, INFO "alice's connInfo")
+      get b ##> ("", aId, CON)
+      exchangeGreetings a bId b aId
+  disconnectAgentClient a
+  disconnectAgentClient b
 
 testUsers :: IO ()
-testUsers = do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
-  b <- getSMPAgentClient' agentCfg initAgentServers testDB2
-  runRight_ $ do
+testUsers =
+  withAgentClients2 $ \a b -> runRight_ $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetingsMsgId 4 a bId b aId
     auId <- createUser a [noAuthSrv testSMPServer] [noAuthSrv testXFTPServer]
@@ -1325,10 +1352,8 @@ testUsers = do
     liftIO $ noMessages a "nothing else should be delivered to alice"
 
 testDeleteUserQuietly :: IO ()
-testDeleteUserQuietly = do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
-  b <- getSMPAgentClient' agentCfg initAgentServers testDB2
-  runRight_ $ do
+testDeleteUserQuietly =
+  withAgentClients2 $ \a b -> runRight_ $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetingsMsgId 4 a bId b aId
     auId <- createUser a [noAuthSrv testSMPServer] [noAuthSrv testXFTPServer]
@@ -1339,9 +1364,7 @@ testDeleteUserQuietly = do
     liftIO $ noMessages a "nothing else should be delivered to alice"
 
 testUsersNoServer :: HasCallStack => ATransport -> IO ()
-testUsersNoServer t = do
-  a <- getSMPAgentClient' agentCfg {initialCleanupDelay = 10000, cleanupInterval = 10000, deleteErrorCount = 3} initAgentServers testDB
-  b <- getSMPAgentClient' agentCfg initAgentServers testDB2
+testUsersNoServer t = withAgentClientsCfg2 aCfg agentCfg $ \a b -> do
   (aId, bId, auId, _aId', bId') <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ do
     (aId, bId) <- makeConnection a b
     exchangeGreetingsMsgId 4 a bId b aId
@@ -1362,6 +1385,8 @@ testUsersNoServer t = do
     nGet a =##> \case ("", "", UP _ [c]) -> c == bId; _ -> False
     nGet b =##> \case ("", "", UP _ cs) -> length cs == 2; _ -> False
     exchangeGreetingsMsgId 6 a bId b aId
+  where
+    aCfg = agentCfg {initialCleanupDelay = 10000, cleanupInterval = 10000, deleteErrorCount = 3}
 
 testSwitchConnection :: InitialAgentServers -> IO ()
 testSwitchConnection servers = do
@@ -1372,6 +1397,8 @@ testSwitchConnection servers = do
     exchangeGreetingsMsgId 4 a bId b aId
     testFullSwitch a bId b aId 10
     testFullSwitch a bId b aId 16
+  disconnectAgentClient a
+  disconnectAgentClient b
 
 testFullSwitch :: AgentClient -> ByteString -> AgentClient -> ByteString -> Int64 -> ExceptT AgentErrorType IO ()
 testFullSwitch a bId b aId msgId = do
@@ -1479,6 +1506,8 @@ testSwitchDelete servers = do
     get a =##> \case ("", c, DEL_RCVQ _ _ Nothing) -> c == bId; _ -> False
     get a =##> \case ("", c, DEL_CONN) -> c == bId; _ -> False
     liftIO $ noMessages a "nothing else should be delivered to alice"
+  disconnectAgentClient a
+  disconnectAgentClient b
 
 testAbortSwitchStarted :: HasCallStack => InitialAgentServers -> IO ()
 testAbortSwitchStarted servers = do
@@ -1750,7 +1779,7 @@ testCreateQueueAuth :: HasCallStack => (Maybe BasicAuth, Version) -> (Maybe Basi
 testCreateQueueAuth clnt1 clnt2 = do
   a <- getClient clnt1
   b <- getClient clnt2
-  runRight $ do
+  r <- runRight $ do
     tryError (createConnection a 1 True SCMInvitation Nothing) >>= \case
       Left (SMP AUTH) -> pure 0
       Left e -> throwError e
@@ -1766,6 +1795,9 @@ testCreateQueueAuth clnt1 clnt2 = do
             get b ##> ("", aId, CON)
             exchangeGreetings a bId b aId
             pure 2
+  disconnectAgentClient a
+  disconnectAgentClient b
+  pure r
   where
     getClient (clntAuth, clntVersion) =
       let servers = initAgentServers {smp = userServers [ProtoServerWithAuth testSMPServer clntAuth]}
@@ -1779,20 +1811,16 @@ testSMPServerConnectionTest t newQueueBasicAuth srv =
     runRight $ testProtocolServer a 1 srv
 
 testRatchetAdHash :: HasCallStack => IO ()
-testRatchetAdHash = do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
-  b <- getSMPAgentClient' agentCfg initAgentServers testDB2
-  runRight_ $ do
+testRatchetAdHash =
+  withAgentClients2 $ \a b -> runRight_ $ do
     (aId, bId) <- makeConnection a b
     ad1 <- getConnectionRatchetAdHash a bId
     ad2 <- getConnectionRatchetAdHash b aId
     liftIO $ ad1 `shouldBe` ad2
 
 testDeliveryReceipts :: HasCallStack => IO ()
-testDeliveryReceipts = do
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
-  b <- getSMPAgentClient' agentCfg initAgentServers testDB2
-  runRight_ $ do
+testDeliveryReceipts =
+  withAgentClients2 $ \a b -> runRight_ $ do
     (aId, bId) <- makeConnection a b
     -- a sends, b receives and sends delivery receipt
     4 <- sendMessage a bId SMP.noMsgFlags "hello"
@@ -1854,12 +1882,12 @@ testDeliveryReceiptsVersion t = do
       ackMessage a' bId 10 $ Just ""
       get b' =##> \case ("", c, Rcvd 10) -> c == aId; _ -> False
       ackMessage b' aId 11 Nothing
+    disconnectAgentClient a'
+    disconnectAgentClient b'
 
 testTwoUsers :: HasCallStack => IO ()
-testTwoUsers = do
+testTwoUsers = withAgentClients2 $ \a b -> do
   let nc = netCfg initAgentServers
-  a <- getSMPAgentClient' agentCfg initAgentServers testDB
-  b <- getSMPAgentClient' agentCfg initAgentServers testDB2
   sessionMode nc `shouldBe` TSMUser
   runRight_ $ do
     (aId1, bId1) <- makeConnectionForUsers a 1 b 1
@@ -1929,10 +1957,8 @@ getSMPAgentClient' cfg' initServers dbPath = do
   getSMPAgentClient cfg' initServers st
 
 testServerMultipleIdentities :: HasCallStack => IO ()
-testServerMultipleIdentities = do
-  alice <- getSMPAgentClient' agentCfg initAgentServers testDB
-  bob <- getSMPAgentClient' agentCfg initAgentServers testDB2
-  runRight_ $ do
+testServerMultipleIdentities =
+  withAgentClients2 $ \alice bob -> runRight_ $ do
     (bobId, cReq) <- createConnection alice 1 True SCMInvitation Nothing
     aliceId <- joinConnection bob 1 True cReq "bob's connInfo"
     ("", _, CONF confId _ "bob's connInfo") <- get alice

tests/AgentTests/MigrationTests.hs

@@ -6,8 +6,8 @@ import Control.Monad
 import Data.Maybe (fromJust)
 import Data.Word (Word32)
 import Database.SQLite.Simple (fromOnly)
-import qualified Database.SQLite.Simple as DB
 import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..), MigrationError (MEDowngrade, MEUpgrade, MigrationError), SQLiteStore, closeSQLiteStore, createSQLiteStore, upMigration, withTransaction)
+import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
 import Simplex.Messaging.Agent.Store.SQLite.Migrations
 import System.Directory (removeFile)
 import System.Random (randomIO)

tests/AgentTests/NotificationTests.hs

@@ -11,7 +11,9 @@ module AgentTests.NotificationTests where
 -- import Control.Logger.Simple (LogConfig (..), LogLevel (..), setLogLevel, withGlobalLogging)
 import AgentTests.FunctionalAPITests (exchangeGreetingsMsgId, get, getSMPAgentClient', makeConnection, nGet, runRight, runRight_, switchComplete, testServerMatrix2, (##>), (=##>), pattern Msg)
 import Control.Concurrent (killThread, threadDelay)
+import Control.Monad
 import Control.Monad.Except
+import Control.Monad.Trans.Except
 import qualified Data.Aeson as J
 import qualified Data.Aeson.Types as JT
 import Data.Bifunctor (bimap, first)
@@ -33,7 +35,6 @@ import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), SMPMsg
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Transport (ATransport)
-import Simplex.Messaging.Util (tryE)
 import System.Directory (doesFileExist, removeFile)
 import Test.Hspec
 import UnliftIO
@@ -111,7 +112,7 @@ testNotificationToken APNSMockServer {apnsQ} = do
     deleteNtfToken a tkn
     -- agent deleted this token
     Left (CMD PROHIBITED) <- tryE $ checkNtfToken a tkn
-    pure ()
+    disconnectAgentClient a
 
 (.->) :: J.Value -> J.Key -> ExceptT AgentErrorType IO ByteString
 v .-> key = do
@@ -143,7 +144,7 @@ testNtfTokenRepeatRegistration APNSMockServer {apnsQ} = do
     -- can still use the first verification code, it is the same after decryption
     verifyNtfToken a tkn nonce verification
     NTActive <- checkNtfToken a tkn
-    pure ()
+    disconnectAgentClient a
 
 testNtfTokenSecondRegistration :: APNSMockServer -> IO ()
 testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
@@ -179,7 +180,8 @@ testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
     Left (NTF AUTH) <- tryE $ checkNtfToken a tkn
     -- and the second is active
     NTActive <- checkNtfToken a' tkn
-    pure ()
+    disconnectAgentClient a
+    disconnectAgentClient a'
 
 testNtfTokenServerRestart :: ATransport -> APNSMockServer -> IO ()
 testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
@@ -208,7 +210,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
     liftIO $ sendApnsResponse' APNSRespOk
     verifyNtfToken a' tkn nonce' verification'
     NTActive <- checkNtfToken a' tkn
-    pure ()
+    disconnectAgentClient a'
 
 testNotificationSubscriptionExistingConnection :: APNSMockServer -> IO ()
 testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
@@ -262,6 +264,8 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
     get bob ##> ("", aliceId, SENT $ baseId + 2)
     -- no notifications should follow
     noNotification apnsQ
+  disconnectAgentClient alice
+  disconnectAgentClient bob
   where
     baseId = 3
     msgId = subtract baseId
@@ -305,6 +309,8 @@ testNotificationSubscriptionNewConnection APNSMockServer {apnsQ} = do
     ackMessage bob aliceId (baseId + 2) Nothing
     -- no unexpected notifications should follow
     noNotification apnsQ
+  disconnectAgentClient alice
+  disconnectAgentClient bob
   where
     baseId = 3
     msgId = subtract baseId
@@ -382,6 +388,8 @@ testChangeNotificationsMode APNSMockServer {apnsQ} = do
     ackMessage alice bobId (baseId + 5) Nothing
     -- no notifications should follow
     noNotification apnsQ
+  disconnectAgentClient alice
+  disconnectAgentClient bob
   where
     baseId = 3
     msgId = subtract baseId
@@ -425,6 +433,8 @@ testChangeToken APNSMockServer {apnsQ} = do
     ackMessage alice1 bobId (baseId + 2) Nothing
     -- no notifications should follow
     noNotification apnsQ
+  disconnectAgentClient alice1
+  disconnectAgentClient bob
   where
     baseId = 3
     msgId = subtract baseId
@@ -454,6 +464,8 @@ testNotificationsStoreLog t APNSMockServer {apnsQ} = do
     void $ messageNotification apnsQ
     get alice =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False
     liftIO $ killThread threadId
+  disconnectAgentClient alice
+  disconnectAgentClient bob
 
 testNotificationsSMPRestart :: ATransport -> APNSMockServer -> IO ()
 testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
@@ -484,6 +496,8 @@ testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
     _ <- messageNotificationData alice apnsQ
     get alice =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False
     liftIO $ killThread threadId
+  disconnectAgentClient alice
+  disconnectAgentClient bob
 
 testNotificationsSMPRestartBatch :: Int -> ATransport -> APNSMockServer -> IO ()
 testNotificationsSMPRestartBatch n t APNSMockServer {apnsQ} = do
@@ -522,6 +536,8 @@ testNotificationsSMPRestartBatch n t APNSMockServer {apnsQ} = do
       get b ##> ("", aliceId, SENT msgId)
       _ <- messageNotificationData a apnsQ
       get a =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False
+  disconnectAgentClient a
+  disconnectAgentClient b
   where
     runServers :: ExceptT AgentErrorType IO a -> IO a
     runServers a = do
@@ -551,6 +567,8 @@ testSwitchNotifications servers APNSMockServer {apnsQ} = do
     switchComplete a bId b aId
     liftIO $ threadDelay 500000
     testMessage "hello again"
+  disconnectAgentClient a
+  disconnectAgentClient b
 
 messageNotification :: TBQueue APNSMockRequest -> ExceptT AgentErrorType IO (C.CbNonce, ByteString)
 messageNotification apnsQ = do

tests/AgentTests/SQLiteTests.hs

@@ -7,6 +7,8 @@
 {-# LANGUAGE QuasiQuotes #-}
 {-# LANGUAGE RecordWildCards #-}
 
+{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
+
 module AgentTests.SQLiteTests (storeTests) where
 
 import Control.Concurrent.Async (concurrently_)
@@ -18,13 +20,14 @@ import qualified Data.Text as T
 import Data.Text.Encoding (encodeUtf8)
 import Data.Time
 import Data.Word (Word32)
-import qualified Database.SQLite.Simple as DB
+import qualified Database.SQLite.Simple as SQL
 import Database.SQLite.Simple.QQ (sql)
 import SMPClient (testKeyHash)
 import Simplex.Messaging.Agent.Client ()
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.Store
 import Simplex.Messaging.Agent.Store.SQLite
+import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
 import qualified Simplex.Messaging.Crypto as C
 import qualified Simplex.Messaging.Protocol as SMP
@@ -137,7 +140,7 @@ testForeignKeysEnabled =
               ('smp.simplex.im', '5223', '1234', '2345', x'', x'', 'new');
           |]
     DB.execute_ db inconsistentQuery
-      `shouldThrow` (\e -> DB.sqlError e == DB.ErrorConstraint)
+      `shouldThrow` (\e -> SQL.sqlError e == SQL.ErrorConstraint)
 
 cData1 :: ConnData
 cData1 = ConnData {userId = 1, connId = "conn1", connAgentVersion = 1, enableNtfs = True, duplexHandshake = Nothing, lastExternalSndId = 0, deleted = False, ratchetSyncState = RSOk}

tests/AgentTests/SchemaDump.hs

@@ -4,7 +4,7 @@
 module AgentTests.SchemaDump where
 
 import Control.DeepSeq
-import Control.Monad (void)
+import Control.Monad (unless, void)
 import Data.List (dropWhileEnd)
 import Data.Maybe (fromJust, isJust)
 import Simplex.Messaging.Agent.Store.SQLite
@@ -54,12 +54,19 @@ testSchemaMigrations = do
       schema' <- getSchema testDB testSchema
       schema' `shouldNotBe` schema
       Migrations.run st $ MTRDown [downMigr]
-      schema'' <- getSchema testDB testSchema
-      schema'' `shouldBe` schema
+      unless (name m `elem` skipComparisonForDownMigrations) $ do
+        schema'' <- getSchema testDB testSchema
+        schema'' `shouldBe` schema
       Migrations.run st $ MTRUp [m]
       schema''' <- getSchema testDB testSchema
       schema''' `shouldBe` schema'
 
+skipComparisonForDownMigrations :: [String]
+skipComparisonForDownMigrations =
+  [ -- on down migration idx_messages_internal_snd_id_ts index moves down to the end of the file
+    "m20230814_indexes"
+  ]
+
 getSchema :: FilePath -> FilePath -> IO String
 getSchema dpPath schemaPath = do
   void $ readCreateProcess (shell $ "sqlite3 " <> dpPath <> " '.schema --indent' > " <> schemaPath) ""

tests/CoreTests/BatchingTests.hs

@@ -0,0 +1,180 @@
+{-# LANGUAGE LambdaCase #-}
+
+module CoreTests.BatchingTests (batchingTests) where
+
+import Control.Concurrent.STM
+import Control.Monad
+import Crypto.Random (MonadRandom(..))
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
+import qualified Data.List.NonEmpty as L
+import Simplex.Messaging.Client
+import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Protocol
+import Simplex.Messaging.Transport
+import Simplex.Messaging.Version (VersionRange(..))
+import Test.Hspec
+
+batchingTests :: Spec
+batchingTests = do
+  describe "batchTransmissions" $ do
+    it "should batch with 90 subscriptions per batch" testBatchSubscriptions
+    it "should break on message that does not fit" testBatchWithMessage
+    it "should break on large message" testBatchWithLargeMessage
+  describe "batchClientTransmissions" $ do
+    it "should batch with 90 subscriptions per batch" testClientBatchSubscriptions
+    it "should break on message that does not fit" testClientBatchWithMessage
+    it "should break on large message" testClientBatchWithLargeMessage
+
+testBatchSubscriptions :: IO ()
+testBatchSubscriptions = do
+  sessId <- getRandomBytes 32
+  subs <- replicateM 200 $ randomSUB sessId
+  let batches1 = batchTransmissions False smpBlockSize $ L.fromList subs
+  all lenOk1 batches1 `shouldBe` True
+  length batches1 `shouldBe` 200
+  let batches = batchTransmissions True smpBlockSize $ L.fromList subs
+  length batches `shouldBe` 3
+  [TBTransmissions n1 s1, TBTransmissions n2 s2, TBTransmissions n3 s3] <- pure batches
+  (n1, n2, n3) `shouldBe` (90, 90, 20)
+  all lenOk [s1, s2, s3] `shouldBe` True
+
+testBatchWithMessage :: IO ()
+testBatchWithMessage = do
+  sessId <- getRandomBytes 32
+  subs1 <- replicateM 60 $ randomSUB sessId
+  send <- randomSEND sessId 8000
+  subs2 <- replicateM 40 $ randomSUB sessId
+  let cmds = subs1 <> [send] <> subs2
+      batches1 = batchTransmissions False smpBlockSize $ L.fromList cmds
+  all lenOk1 batches1 `shouldBe` True
+  length batches1 `shouldBe` 101
+  let batches = batchTransmissions True smpBlockSize $ L.fromList cmds
+  length batches `shouldBe` 2
+  [TBTransmissions n1 s1, TBTransmissions n2 s2] <- pure batches
+  (n1, n2) `shouldBe` (60, 41)
+  all lenOk [s1, s2] `shouldBe` True
+
+testBatchWithLargeMessage :: IO ()
+testBatchWithLargeMessage = do
+  sessId <- getRandomBytes 32
+  subs1 <- replicateM 60 $ randomSUB sessId
+  send <- randomSEND sessId 17000
+  subs2 <- replicateM 100 $ randomSUB sessId
+  let cmds = subs1 <> [send] <> subs2
+      batches1 = batchTransmissions False smpBlockSize $ L.fromList cmds
+  all lenOk1 batches1 `shouldBe` False
+  length batches1 `shouldBe` 161
+  let batches1' = take 60 batches1 <> drop 61 batches1
+  all lenOk1 batches1' `shouldBe` True
+  length batches1' `shouldBe` 160
+  let batches = batchTransmissions True smpBlockSize $ L.fromList cmds
+  length batches `shouldBe` 4
+  [TBTransmissions n1 s1, TBLargeTransmission, TBTransmissions n2 s2, TBTransmissions n3 s3] <- pure batches
+  (n1, n2, n3) `shouldBe` (60, 90, 10)
+  all lenOk [s1, s2, s3] `shouldBe` True
+
+testClientBatchSubscriptions :: IO ()
+testClientBatchSubscriptions = do
+  sessId <- getRandomBytes 32
+  client <- atomically $ clientStub sessId
+  subs <- replicateM 200 $ randomSUBCmd client
+  let batches1 = batchClientTransmissions False smpBlockSize $ L.fromList subs
+  all lenOk1' batches1 `shouldBe` True
+  let batches = batchClientTransmissions True smpBlockSize $ L.fromList subs
+  length batches `shouldBe` 3
+  [CBTransmissions s1 n1 rs1, CBTransmissions s2 n2 rs2, CBTransmissions s3 n3 rs3] <- pure batches
+  (n1, n2, n3) `shouldBe` (90, 90, 20)
+  (length rs1, length rs2, length rs3) `shouldBe` (90, 90, 20)
+  all lenOk [s1, s2, s3] `shouldBe` True
+
+testClientBatchWithMessage :: IO ()
+testClientBatchWithMessage = do
+  sessId <- getRandomBytes 32
+  client <- atomically $ clientStub sessId
+  subs1 <- replicateM 60 $ randomSUBCmd client
+  send <- randomSENDCmd client 8000
+  subs2 <- replicateM 40 $ randomSUBCmd client
+  let cmds = subs1 <> [send] <> subs2
+      batches1 = batchClientTransmissions False smpBlockSize $ L.fromList cmds
+  all lenOk1' batches1 `shouldBe` True
+  length batches1 `shouldBe` 101
+  let batches = batchClientTransmissions True smpBlockSize $ L.fromList cmds
+  length batches `shouldBe` 2
+  [CBTransmissions s1 n1 rs1, CBTransmissions s2 n2 rs2] <- pure batches
+  (n1, n2) `shouldBe` (60, 41)
+  (length rs1, length rs2) `shouldBe` (60, 41)
+  all lenOk [s1, s2] `shouldBe` True
+
+testClientBatchWithLargeMessage :: IO ()
+testClientBatchWithLargeMessage = do
+  sessId <- getRandomBytes 32
+  client <- atomically $ clientStub sessId
+  subs1 <- replicateM 60 $ randomSUBCmd client
+  send <- randomSENDCmd client 17000
+  subs2 <- replicateM 100 $ randomSUBCmd client
+  let cmds = subs1 <> [send] <> subs2
+      batches1 = batchClientTransmissions False smpBlockSize $ L.fromList cmds
+  all lenOk1' batches1 `shouldBe` False
+  length batches1 `shouldBe` 161
+  let batches1' = take 60 batches1 <> drop 61 batches1
+  all lenOk1' batches1' `shouldBe` True
+  length batches1' `shouldBe` 160
+  --
+  let batches = batchClientTransmissions True smpBlockSize $ L.fromList cmds
+  length batches `shouldBe` 4
+  [CBTransmissions s1 n1 rs1, CBLargeTransmission _, CBTransmissions s2 n2 rs2, CBTransmissions s3 n3 rs3] <- pure batches
+  (n1, n2, n3) `shouldBe` (60, 90, 10)
+  (length rs1, length rs2, length rs3) `shouldBe` (60, 90, 10)
+  all lenOk [s1, s2, s3] `shouldBe` True
+  --
+  let cmds' = [send] <> subs1 <> subs2
+  let batches' = batchClientTransmissions True smpBlockSize $ L.fromList cmds'
+  length batches' `shouldBe` 3
+  [CBLargeTransmission _, CBTransmissions s1' n1' rs1', CBTransmissions s2' n2' rs2'] <- pure batches'
+  (n1', n2') `shouldBe` (90, 70)
+  (length rs1', length rs2') `shouldBe` (90, 70)
+  all lenOk [s1', s2'] `shouldBe` True
+
+randomSUB :: ByteString -> IO (Maybe C.ASignature, ByteString)
+randomSUB sessId = do
+  rId <- getRandomBytes 24
+  corrId <- CorrId <$> getRandomBytes 3
+  (_, rpKey) <- C.generateSignatureKeyPair C.SEd448
+  let s = encodeTransmission (maxVersion supportedSMPServerVRange) sessId (corrId, rId, Cmd SRecipient SUB)
+  pure (Just $ C.sign rpKey s, s)
+
+randomSUBCmd :: ProtocolClient ErrorType BrokerMsg -> IO (PCTransmission ErrorType BrokerMsg)
+randomSUBCmd c = do
+  rId <- getRandomBytes 24
+  (_, rpKey) <- C.generateSignatureKeyPair C.SEd448
+  mkTransmission c (Just rpKey, rId, Cmd SRecipient SUB)
+
+randomSEND :: ByteString -> Int -> IO (Maybe C.ASignature, ByteString)
+randomSEND sessId len = do
+  sId <- getRandomBytes 24
+  corrId <- CorrId <$> getRandomBytes 3
+  (_, rpKey) <- C.generateSignatureKeyPair C.SEd448
+  msg <- getRandomBytes len
+  let s = encodeTransmission (maxVersion supportedSMPServerVRange) sessId (corrId, sId, Cmd SSender $ SEND noMsgFlags msg)
+  pure (Just $ C.sign rpKey s, s)
+
+randomSENDCmd :: ProtocolClient ErrorType BrokerMsg -> Int -> IO (PCTransmission ErrorType BrokerMsg)
+randomSENDCmd c len = do
+  sId <- getRandomBytes 24
+  (_, rpKey) <- C.generateSignatureKeyPair C.SEd448
+  msg <- getRandomBytes len
+  mkTransmission c (Just rpKey, sId, Cmd SSender $ SEND noMsgFlags msg)
+
+lenOk :: ByteString -> Bool
+lenOk s = 0 < B.length s && B.length s <= smpBlockSize - 2
+
+lenOk1 :: TransportBatch -> Bool
+lenOk1 = \case
+  TBTransmission s -> lenOk s
+  _ -> False
+
+lenOk1' :: ClientBatch err msg -> Bool
+lenOk1' = \case
+  CBTransmission s _ -> lenOk s
+  _ -> False

tests/CoreTests/UtilTests.hs

@@ -4,6 +4,7 @@ module CoreTests.UtilTests where
 
 import Control.Exception (Exception, SomeException, throwIO)
 import Control.Monad.Except
+import Control.Monad.IO.Class
 import Data.IORef
 import Simplex.Messaging.Util
 import Simplex.Messaging.Client.Agent ()

tests/NtfClient.hs

@@ -5,8 +5,8 @@
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
-{-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedLists #-}
+{-# LANGUAGE OverloadedRecordDot #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StandaloneDeriving #-}
@@ -133,8 +133,9 @@ ntfServerTest ::
   IO (Maybe C.ASignature, ByteString, ByteString, BrokerMsg)
 ntfServerTest _ t = runNtfTest $ \h -> tPut' h t >> tGet' h
   where
+    tPut' :: THandle c -> (Maybe C.ASignature, ByteString, ByteString, smp) -> IO ()
     tPut' h (sig, corrId, queueId, smp) = do
-      let t' = smpEncode (sessionId (h :: THandle c), corrId, queueId, smp)
+      let t' = smpEncode (h.sessionId, corrId, queueId, smp)
       [Right ()] <- tPut h Nothing [(sig, t')]
       pure ()
     tGet' h = do

tests/SMPClient.hs

@@ -4,6 +4,7 @@
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedLists #-}
+{-# LANGUAGE OverloadedRecordDot #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
@@ -159,8 +160,9 @@ smpServerTest ::
   IO (Maybe C.ASignature, ByteString, ByteString, BrokerMsg)
 smpServerTest _ t = runSmpTest $ \h -> tPut' h t >> tGet' h
   where
+    tPut' :: THandle c -> (Maybe C.ASignature, ByteString, ByteString, smp) -> IO ()
     tPut' h (sig, corrId, queueId, smp) = do
-      let t' = smpEncode (sessionId (h :: THandle c), corrId, queueId, smp)
+      let t' = smpEncode (h.sessionId, corrId, queueId, smp)
       [Right ()] <- tPut h Nothing [(sig, t')]
       pure ()
     tGet' h = do
@@ -173,20 +175,26 @@ smpTest _ test' = runSmpTest test' `shouldReturn` ()
 smpTestN :: (HasCallStack, Transport c) => Int -> (HasCallStack => [THandle c] -> IO ()) -> Expectation
 smpTestN n test' = runSmpTestN n test' `shouldReturn` ()
 
-smpTest2 :: (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandle c -> THandle c -> IO ()) -> Expectation
+smpTest2 :: forall c. (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandle c -> THandle c -> IO ()) -> Expectation
 smpTest2 _ test' = smpTestN 2 _test
   where
+    _test :: HasCallStack => [THandle c] -> IO ()
     _test [h1, h2] = test' h1 h2
     _test _ = error "expected 2 handles"
 
-smpTest3 :: (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandle c -> THandle c -> THandle c -> IO ()) -> Expectation
+smpTest3 :: forall c. (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandle c -> THandle c -> THandle c -> IO ()) -> Expectation
 smpTest3 _ test' = smpTestN 3 _test
   where
+    _test :: HasCallStack => [THandle c] -> IO ()
     _test [h1, h2, h3] = test' h1 h2 h3
     _test _ = error "expected 3 handles"
 
-smpTest4 :: (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandle c -> THandle c -> THandle c -> THandle c -> IO ()) -> Expectation
+smpTest4 :: forall c. (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandle c -> THandle c -> THandle c -> THandle c -> IO ()) -> Expectation
 smpTest4 _ test' = smpTestN 4 _test
   where
+    _test :: HasCallStack => [THandle c] -> IO ()
     _test [h1, h2, h3, h4] = test' h1 h2 h3 h4
     _test _ = error "expected 4 handles"
+
+unexpected :: (HasCallStack, Show a) => a -> Expectation
+unexpected r = expectationFailure $ "unexpected response " <> show r

tests/ServerTests.hs

@@ -15,7 +15,7 @@ import AgentTests.NotificationTests (removeFileIfExists)
 import Control.Concurrent (ThreadId, killThread, threadDelay)
 import Control.Concurrent.STM
 import Control.Exception (SomeException, try)
-import Control.Monad.Except (forM, forM_)
+import Control.Monad
 import Control.Monad.IO.Class
 import Data.Bifunctor (first)
 import Data.ByteString.Base64
@@ -182,65 +182,65 @@ testCreateSecureV2 _ =
 testCreateSecure :: ATransport -> Spec
 testCreateSecure (ATransport t) =
   it "should create (NEW) and secure (KEY) queue" $
-    smpTest t $ \h -> do
+    smpTest2 t $ \r s -> do
       (rPub, rKey) <- C.generateSignatureKeyPair C.SEd448
       (dhPub, dhPriv :: C.PrivateKeyX25519) <- C.generateKeyPair'
-      Resp "abcd" rId1 (Ids rId sId srvDh) <- signSendRecv h rKey ("abcd", "", NEW rPub dhPub Nothing)
+      Resp "abcd" rId1 (Ids rId sId srvDh) <- signSendRecv r rKey ("abcd", "", NEW rPub dhPub Nothing)
       let dec = decryptMsgV3 $ C.dh' srvDh dhPriv
       (rId1, "") #== "creates queue"
 
-      Resp "bcda" sId1 ok1 <- sendRecv h ("", "bcda", sId, _SEND "hello")
+      Resp "bcda" sId1 ok1 <- sendRecv s ("", "bcda", sId, _SEND "hello")
       (ok1, OK) #== "accepts unsigned SEND"
       (sId1, sId) #== "same queue ID in response 1"
 
-      Resp "" _ (Msg mId1 msg1) <- tGet1 h
+      Resp "" _ (Msg mId1 msg1) <- tGet1 r
       (dec mId1 msg1, Right "hello") #== "delivers message"
 
-      Resp "cdab" _ ok4 <- signSendRecv h rKey ("cdab", rId, ACK mId1)
+      Resp "cdab" _ ok4 <- signSendRecv r rKey ("cdab", rId, ACK mId1)
       (ok4, OK) #== "replies OK when message acknowledged if no more messages"
 
-      Resp "dabc" _ err6 <- signSendRecv h rKey ("dabc", rId, ACK mId1)
+      Resp "dabc" _ err6 <- signSendRecv r rKey ("dabc", rId, ACK mId1)
       (err6, ERR NO_MSG) #== "replies ERR when message acknowledged without messages"
 
       (sPub, sKey) <- C.generateSignatureKeyPair C.SEd448
-      Resp "abcd" sId2 err1 <- signSendRecv h sKey ("abcd", sId, _SEND "hello")
+      Resp "abcd" sId2 err1 <- signSendRecv s sKey ("abcd", sId, _SEND "hello")
       (err1, ERR AUTH) #== "rejects signed SEND"
       (sId2, sId) #== "same queue ID in response 2"
 
-      Resp "bcda" _ err2 <- sendRecv h (sampleSig, "bcda", rId, KEY sPub)
+      Resp "bcda" _ err2 <- sendRecv r (sampleSig, "bcda", rId, KEY sPub)
       (err2, ERR AUTH) #== "rejects KEY with wrong signature"
 
-      Resp "cdab" _ err3 <- signSendRecv h rKey ("cdab", sId, KEY sPub)
+      Resp "cdab" _ err3 <- signSendRecv r rKey ("cdab", sId, KEY sPub)
       (err3, ERR AUTH) #== "rejects KEY with sender's ID"
 
-      Resp "dabc" rId2 ok2 <- signSendRecv h rKey ("dabc", rId, KEY sPub)
+      Resp "dabc" rId2 ok2 <- signSendRecv r rKey ("dabc", rId, KEY sPub)
       (ok2, OK) #== "secures queue"
       (rId2, rId) #== "same queue ID in response 3"
 
-      Resp "abcd" _ OK <- signSendRecv h rKey ("abcd", rId, KEY sPub)
+      Resp "abcd" _ OK <- signSendRecv r rKey ("abcd", rId, KEY sPub)
       (sPub', _) <- C.generateSignatureKeyPair C.SEd448
-      Resp "abcd" _ err4 <- signSendRecv h rKey ("abcd", rId, KEY sPub')
+      Resp "abcd" _ err4 <- signSendRecv r rKey ("abcd", rId, KEY sPub')
       (err4, ERR AUTH) #== "rejects if secured with different key"
 
-      Resp "bcda" _ ok3 <- signSendRecv h sKey ("bcda", sId, _SEND "hello again")
+      Resp "bcda" _ ok3 <- signSendRecv s sKey ("bcda", sId, _SEND "hello again")
       (ok3, OK) #== "accepts signed SEND"
 
-      Resp "" _ (Msg mId2 msg2) <- tGet1 h
+      Resp "" _ (Msg mId2 msg2) <- tGet1 r
       (dec mId2 msg2, Right "hello again") #== "delivers message 2"
 
-      Resp "cdab" _ ok5 <- signSendRecv h rKey ("cdab", rId, ACK mId2)
+      Resp "cdab" _ ok5 <- signSendRecv r rKey ("cdab", rId, ACK mId2)
       (ok5, OK) #== "replies OK when message acknowledged 2"
 
-      Resp "dabc" _ err5 <- sendRecv h ("", "dabc", sId, _SEND "hello")
+      Resp "dabc" _ err5 <- sendRecv s ("", "dabc", sId, _SEND "hello")
       (err5, ERR AUTH) #== "rejects unsigned SEND"
 
       let maxAllowedMessage = B.replicate maxMessageLength '-'
-      Resp "bcda" _ OK <- signSendRecv h sKey ("bcda", sId, _SEND maxAllowedMessage)
-      Resp "" _ (Msg mId3 msg3) <- tGet1 h
+      Resp "bcda" _ OK <- signSendRecv s sKey ("bcda", sId, _SEND maxAllowedMessage)
+      Resp "" _ (Msg mId3 msg3) <- tGet1 r
       (dec mId3 msg3, Right maxAllowedMessage) #== "delivers message of max size"
 
       let biggerMessage = B.replicate (maxMessageLength + 1) '-'
-      Resp "bcda" _ (ERR LARGE_MSG) <- signSendRecv h sKey ("bcda", sId, _SEND biggerMessage)
+      Resp "bcda" _ (ERR LARGE_MSG) <- signSendRecv s sKey ("bcda", sId, _SEND biggerMessage)
       pure ()
 
 testCreateDelete :: ATransport -> Spec
@@ -551,14 +551,18 @@ testWithStoreLog at@(ATransport t) =
         writeTVar senderId1 sId1
         writeTVar notifierId nId
       Resp "dabc" _ OK <- signSendRecv h1 nKey ("dabc", nId, NSUB)
-      Resp "bcda" _ OK <- signSendRecv h sKey1 ("bcda", sId1, _SEND' "hello")
+      signSendRecv h sKey1 ("bcda", sId1, _SEND' "hello") >>= \case
+        Resp "bcda" _ OK -> pure ()
+        r -> unexpected r
       Resp "" _ (Msg mId1 msg1) <- tGet1 h
       (decryptMsgV3 dhShared mId1 msg1, Right "hello") #== "delivered from queue 1"
       Resp "" _ (NMSG _ _) <- tGet1 h1
 
       (sId2, rId2, rKey2, dhShared2) <- createAndSecureQueue h sPub2
       atomically $ writeTVar senderId2 sId2
-      Resp "cdab" _ OK <- signSendRecv h sKey2 ("cdab", sId2, _SEND "hello too")
+      signSendRecv h sKey2 ("cdab", sId2, _SEND "hello too") >>= \case
+        Resp "cdab" _ OK -> pure ()
+        r -> unexpected r
       Resp "" _ (Msg mId2 msg2) <- tGet1 h
       (decryptMsgV3 dhShared2 mId2 msg2, Right "hello too") #== "delivered from queue 2"
 
@@ -969,7 +973,9 @@ testMsgExpireOnInterval t =
         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)
+          signSendRecv rh rKey ("2", rId, SUB) >>= \case
+            Resp "2" _ OK -> pure ()
+            r -> unexpected r
           1000 `timeout` tGet @ErrorType @BrokerMsg rh >>= \case
             Nothing -> return ()
             Just _ -> error "nothing should be delivered"

tests/Test.hs

@@ -1,8 +1,10 @@
 {-# LANGUAGE TypeApplications #-}
 
 import AgentTests (agentTests)
+import AgentTests.SchemaDump (schemaDumpTest)
 import CLITests
 import Control.Logger.Simple
+import CoreTests.BatchingTests
 import CoreTests.CryptoTests
 import CoreTests.EncodingTests
 import CoreTests.ProtocolErrorTests
@@ -34,7 +36,9 @@ main = do
       . before_ (createDirectoryIfMissing False "tests/tmp")
       . after_ (removeDirectoryRecursive "tests/tmp")
       $ do
+        describe "Agent SQLite schema dump" schemaDumpTest
         describe "Core tests" $ do
+          describe "Batching tests" batchingTests
           describe "Encoding tests" encodingTests
           describe "Protocol error tests" protocolErrorTests
           describe "Version range" versionRangeTests

tests/XFTPAgent.hs

@@ -9,14 +9,15 @@ module XFTPAgent where
 import AgentTests.FunctionalAPITests (get, getSMPAgentClient', rfGet, runRight, runRight_, sfGet)
 import Control.Concurrent (threadDelay)
 import Control.Logger.Simple
+import Control.Monad
 import Control.Monad.Except
+import Control.Monad.IO.Class
 import Data.Bifunctor (first)
 import qualified Data.ByteString.Char8 as B
 import Data.Int (Int64)
 import Data.List (find, isSuffixOf)
 import Data.Maybe (fromJust)
-import SMPAgentClient (agentCfg, initAgentServers, testDB)
-import SMPClient (xit'')
+import SMPAgentClient (agentCfg, initAgentServers, testDB, testDB2, testDB3)
 import Simplex.FileTransfer.Description
 import Simplex.FileTransfer.Protocol (FileParty (..), XFTPErrorType (AUTH))
 import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..))
@@ -37,7 +38,7 @@ xftpAgentTests = around_ testBracket . describe "Functional API" $ do
   it "should send and receive file" testXFTPAgentSendReceive
   it "should resume receiving file after restart" testXFTPAgentReceiveRestore
   it "should cleanup rcv tmp path after permanent error" testXFTPAgentReceiveCleanup
-  xit'' "should resume sending file after restart" testXFTPAgentSendRestore
+  it "should resume sending file after restart" testXFTPAgentSendRestore
   it "should cleanup snd prefix path after permanent error" testXFTPAgentSendCleanup
   it "should delete sent file on server" testXFTPAgentDelete
   it "should resume deleting file after restart" testXFTPAgentDeleteRestore
@@ -94,10 +95,11 @@ testXFTPAgentSendReceive = withXFTPServer $ do
   testReceiveDelete rfd2 filePath
   where
     testReceiveDelete rfd originalFilePath = do
-      rcp <- getSMPAgentClient' agentCfg initAgentServers testDB
+      rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
       runRight_ $ do
         rfId <- testReceive rcp rfd originalFilePath
         xftpDeleteRcvFile rcp rfId
+      disconnectAgentClient rcp
 
 createRandomFile :: IO FilePath
 createRandomFile = do
@@ -146,7 +148,7 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
       pure rfd
 
   -- receive file - should not succeed with server down
-  rcp <- getSMPAgentClient' agentCfg initAgentServers testDB
+  rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
   rfId <- runRight $ do
     xftpStartWorkers rcp (Just recipientFiles)
     rfId <- xftpReceiveFile rcp 1 rfd
@@ -160,7 +162,7 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerStoreLogOn $ \_ -> do
     -- receive file - should start downloading with server up
-    rcp' <- getSMPAgentClient' agentCfg initAgentServers testDB
+    rcp' <- getSMPAgentClient' agentCfg initAgentServers testDB2
     runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
     ("", rfId', RFPROG _ _) <- rfGet rcp'
     liftIO $ rfId' `shouldBe` rfId
@@ -170,7 +172,7 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerStoreLogOn $ \_ -> do
     -- receive file - should continue downloading with server up
-    rcp' <- getSMPAgentClient' agentCfg initAgentServers testDB
+    rcp' <- getSMPAgentClient' agentCfg initAgentServers testDB2
     runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
     rfProgress rcp' $ mb 18
     ("", rfId', RFDONE path) <- rfGet rcp'
@@ -194,7 +196,7 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do
       pure rfd
 
   -- receive file - should not succeed with server down
-  rcp <- getSMPAgentClient' agentCfg initAgentServers testDB
+  rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
   rfId <- runRight $ do
     xftpStartWorkers rcp (Just recipientFiles)
     rfId <- xftpReceiveFile rcp 1 rfd
@@ -208,7 +210,7 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do
 
   withXFTPServerThreadOn $ \_ -> do
     -- receive file - should fail with AUTH error
-    rcp' <- getSMPAgentClient' agentCfg initAgentServers testDB
+    rcp' <- getSMPAgentClient' agentCfg initAgentServers testDB2
     runRight_ $ xftpStartWorkers rcp' (Just recipientFiles)
     ("", rfId', RFERR (INTERNAL "XFTP {xftpErr = AUTH}")) <- rfGet rcp'
     rfId' `shouldBe` rfId
@@ -255,11 +257,12 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
     liftIO $ sfId' `shouldBe` sfId
 
     -- prefix path should be removed after sending file
+    threadDelay 100000
     doesDirectoryExist prefixPath `shouldReturn` False
     doesFileExist encPath `shouldReturn` False
 
     -- receive file
-    rcp <- getSMPAgentClient' agentCfg initAgentServers testDB
+    rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
     runRight_ $
       void $ testReceive rcp rfd1 filePath
 
@@ -309,7 +312,7 @@ testXFTPAgentDelete = withGlobalLogging logCfgNoLogs $
     (sfId, sndDescr, rfd1, rfd2) <- runRight $ testSend sndr filePath
 
     -- receive file
-    rcp1 <- getSMPAgentClient' agentCfg initAgentServers testDB
+    rcp1 <- getSMPAgentClient' agentCfg initAgentServers testDB2
     runRight_ $
       void $ testReceive rcp1 rfd1 filePath
 
@@ -321,12 +324,13 @@ testXFTPAgentDelete = withGlobalLogging logCfgNoLogs $
       xftpDeleteSndFileRemote sndr 1 sfId sndDescr
       Nothing <- liftIO $ 100000 `timeout` sfGet sndr
       pure ()
+    disconnectAgentClient rcp1
 
     threadDelay 1000000
     length <$> listDirectory xftpServerFiles `shouldReturn` 0
 
     -- receive file - should fail with AUTH error
-    rcp2 <- getSMPAgentClient' agentCfg initAgentServers testDB
+    rcp2 <- getSMPAgentClient' agentCfg initAgentServers testDB2
     runRight $ do
       xftpStartWorkers rcp2 (Just recipientFiles)
       rfId <- xftpReceiveFile rcp2 1 rfd2
@@ -343,10 +347,11 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
     (sfId, sndDescr, rfd1, rfd2) <- runRight $ testSend sndr filePath
 
     -- receive file
-    rcp1 <- getSMPAgentClient' agentCfg initAgentServers testDB
+    rcp1 <- getSMPAgentClient' agentCfg initAgentServers testDB2
     runRight_ $
       void $ testReceive rcp1 rfd1 filePath
-
+    disconnectAgentClient rcp1
+    disconnectAgentClient sndr
     pure (sfId, sndDescr, rfd2)
 
   -- delete file - should not succeed with server down
@@ -369,7 +374,7 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do
     length <$> listDirectory xftpServerFiles `shouldReturn` 0
 
     -- receive file - should fail with AUTH error
-    rcp2 <- getSMPAgentClient' agentCfg initAgentServers testDB
+    rcp2 <- getSMPAgentClient' agentCfg initAgentServers testDB3
     runRight $ do
       xftpStartWorkers rcp2 (Just recipientFiles)
       rfId <- xftpReceiveFile rcp2 1 rfd2
@@ -394,15 +399,12 @@ testXFTPAgentRequestAdditionalRecipientIDs = withXFTPServer $ do
 
   -- receive file using different descriptions
   -- ! revise number of recipients and indexes if xftpMaxRecipientsPerRequest is changed
-  testReceive' (head rfds) filePath
-  testReceive' (rfds !! 99) filePath
-  testReceive' (rfds !! 299) filePath
-  testReceive' (rfds !! 499) filePath
-  where
-    testReceive' rfd originalFilePath = do
-      rcp <- getSMPAgentClient' agentCfg initAgentServers testDB
-      runRight_ $
-        void $ testReceive rcp rfd originalFilePath
+  rcp <- getSMPAgentClient' agentCfg initAgentServers testDB2
+  runRight_ $ do
+    void $ testReceive rcp (head rfds) filePath
+    void $ testReceive rcp (rfds !! 99) filePath
+    void $ testReceive rcp (rfds !! 299) filePath
+    void $ testReceive rcp (rfds !! 499) filePath
 
 testXFTPServerTest :: Maybe BasicAuth -> XFTPServerWithAuth -> IO (Maybe ProtocolTestFailure)
 testXFTPServerTest newFileBasicAuth srv =

tests/XFTPServerTests.hs

@@ -10,7 +10,9 @@ import AgentTests.FunctionalAPITests (runRight_)
 import Control.Concurrent (threadDelay)
 import Control.Concurrent.STM
 import Control.Exception (SomeException)
+import Control.Monad
 import Control.Monad.Except
+import Control.Monad.IO.Unlift
 import Crypto.Random (getRandomBytes)
 import qualified Data.ByteString.Base64.URL as B64
 import Data.ByteString.Char8 (ByteString)
@@ -140,7 +142,7 @@ runTestFileChunkDelete s r = do
   deleteXFTPChunk s spKey sId
   liftIO $
     readChunk sId
-      `shouldThrow` \(e :: SomeException) -> "openBinaryFile: does not exist" `isInfixOf` show e
+      `shouldThrow` \(e :: SomeException) -> "withBinaryFile: does not exist" `isInfixOf` show e
   downloadXFTPChunk r rpKey rId (XFTPRcvChunkSpec "tests/tmp/received_chunk2" chSize digest)
     `catchError` (liftIO . (`shouldBe` PCEProtocolError AUTH))
   deleteXFTPChunk s spKey sId