6.2.0.6

* rfc: moving queue records to journal storage and queue blobs

* rfc: expiring messages in journal storage

* add comment

CHANGELOG.md

@@ -1,3 +1,7 @@
+# 6.1.3
+
+SMP server: fix restoring notification credentials.
+
 # 6.1.2
 
 Servers: more reliable restoring of state.

cabal.project

@@ -4,6 +4,14 @@ packages: .
 -- packages: . ../http2
 -- packages: . ../network-transport
 
+-- package *
+   -- coverage: True
+   -- library-coverage: True
+
+-- package attoparsec
+   -- coverage: False
+   -- library-coverage: False
+
 index-state: 2023-12-12T00:00:00Z
 
 package cryptostore

package.yaml

@@ -1,223 +0,0 @@
-name: simplexmq
-version: 6.1.2.0
-synopsis: SimpleXMQ message broker
-description: |
-  This package includes <./docs/Simplex-Messaging-Server.html server>,
-  <./docs/Simplex-Messaging-Client.html client> and
-  <./docs/Simplex-Messaging-Agent.html agent> for SMP protocols:
-  .
-  * <https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md SMP protocol>
-  * <https://github.com/simplex-chat/simplexmq/blob/master/protocol/agent-protocol.md SMP agent protocol>
-  .
-  See <https://github.com/simplex-chat/simplex-chat terminal chat prototype> built with SimpleXMQ broker.
-
-homepage: https://github.com/simplex-chat/simplexmq#readme
-license: AGPL-3
-author: simplex.chat
-maintainer: chat@simplex.chat
-copyright: 2020-2022 simplex.chat
-category: Chat, Network, Web, System, Cryptography
-extra-source-files:
-  - README.md
-  - CHANGELOG.md
-  - cbits/sha512.h
-  - cbits/sntrup761.h
-  - apps/smp-server/static/*.html
-  - apps/smp-server/static/media/*
-
-dependencies:
-  - aeson == 2.2.*
-  - ansi-terminal >= 0.10 && < 0.12
-  - asn1-encoding == 0.9.*
-  - asn1-types == 0.3.*
-  - async == 2.2.*
-  - attoparsec == 0.14.*
-  - base >= 4.14 && < 5
-  - base64-bytestring >= 1.0 && < 1.3
-  - case-insensitive == 1.2.*
-  - composition == 1.0.*
-  - constraints >= 0.12 && < 0.14
-  - containers == 0.6.*
-  - crypton == 0.34.*
-  - crypton-x509 == 1.7.*
-  - crypton-x509-store == 1.6.*
-  - crypton-x509-validation == 1.6.*
-  - cryptostore == 0.3.*
-  - data-default == 0.7.*
-  - direct-sqlcipher == 2.3.*
-  - directory == 1.3.*
-  - filepath == 1.4.*
-  - hashable == 1.4.*
-  - hourglass == 0.2.*
-  - http-types == 0.12.*
-  - http2 >= 4.2.2 && < 4.3
-  - ini == 0.4.1
-  - iproute == 1.7.*
-  - iso8601-time == 0.1.*
-  - memory == 0.18.*
-  - mtl >= 2.3.1 && < 3.0
-  - network >= 3.1.2.7 && < 3.2
-  - network-info >= 0.2 && < 0.3
-  - network-transport == 0.5.6
-  - network-udp >= 0.0 && < 0.1
-  - optparse-applicative >= 0.15 && < 0.17
-  - process == 1.6.*
-  - random >= 1.1 && < 1.3
-  - simple-logger == 0.1.*
-  - socks == 0.6.*
-  - sqlcipher-simple == 0.4.*
-  - stm == 2.5.*
-  - temporary == 1.3.*
-  - time == 1.12.*
-  - time-manager == 0.0.*
-  - tls >= 1.9.0 && < 1.10
-  - transformers == 0.6.*
-  - unliftio == 0.2.*
-  - unliftio-core == 0.2.*
-  - websockets == 0.12.*
-  - yaml == 0.11.*
-  - zstd == 0.1.3.*
-
-flags:
-  swift:
-    description: Enable swift JSON format
-    manual: True
-    default: False
-  use_crypton:
-    description: Use crypton etc. in cryptostore
-    manual: True
-    default: True
-
-# cpp-options:
-#   - -Dslow_servers
-
-when:
-  - condition: flag(swift)
-    cpp-options:
-      - -DswiftJSON
-  - condition: impl(ghc >= 9.6.2)
-    dependencies:
-      - bytestring == 0.11.*
-      - template-haskell == 2.20.*
-      - text >= 2.0.1 && < 2.2
-  - condition: impl(ghc < 9.6.2)
-    dependencies:
-      - bytestring == 0.10.*
-      - template-haskell == 2.16.*
-      - text >= 1.2.3.0 && < 1.3
-
-library:
-  source-dirs: src
-  c-sources:
-    - cbits/sha512.c
-    - cbits/sntrup761.c
-  include-dirs: cbits
-  extra-libraries: crypto
-
-executables:
-  smp-server:
-    source-dirs:
-    - apps/smp-server
-    - apps/smp-server/web
-    main: Main.hs
-    dependencies:
-      - file-embed
-      - simplexmq
-      - wai
-      - wai-app-static
-      - warp ==3.3.30 # the last one before http2-5.0
-      - warp-tls ==3.4.7 # extra internals exposed
-    ghc-options:
-      - -threaded
-      - -rtsopts
-
-  ntf-server:
-    source-dirs: apps/ntf-server
-    main: Main.hs
-    dependencies:
-      - simplexmq
-    ghc-options:
-      - -threaded
-      - -rtsopts
-
-  xftp-server:
-    source-dirs: apps/xftp-server
-    main: Main.hs
-    dependencies:
-      - simplexmq
-    ghc-options:
-      - -threaded
-      - -rtsopts
-
-  xftp:
-    source-dirs: apps/xftp
-    main: Main.hs
-    dependencies:
-      - simplexmq
-    ghc-options:
-      - -threaded
-      - -rtsopts
-
-tests:
-  simplexmq-test:
-    source-dirs:
-    - tests
-    - apps/smp-server/web
-    main: Test.hs
-    dependencies:
-      - deepseq == 1.4.*
-      - file-embed
-      - generic-random == 1.5.*
-      - hspec == 2.11.*
-      - hspec-core == 2.11.*
-      - http-client
-      - http-client-tls
-      - HUnit == 1.6.*
-      - main-tester == 0.2.*
-      - QuickCheck == 2.14.*
-      - silently == 1.2.*
-      - simplexmq
-      - timeit == 2.0.*
-      - unordered-containers
-      - wai
-      - wai-app-static
-      - warp
-      - warp-tls
-    ghc-options:
-      - -threaded
-      - -rtsopts
-      - -with-rtsopts=-A64M
-      - -with-rtsopts=-N1
-
-ghc-options:
-  # - -haddock
-  - -Weverything
-  - -Wno-missing-exported-signatures
-  - -Wno-missing-import-lists
-  - -Wno-missed-specialisations
-  - -Wno-all-missed-specialisations
-  - -Wno-unsafe
-  - -Wno-safe
-  - -Wno-missing-local-signatures
-  - -Wno-missing-kind-signatures
-  - -Wno-missing-deriving-strategies
-  - -Wno-monomorphism-restriction
-  - -Wno-prepositive-qualified-module
-  - -Wno-unused-packages
-  - -Wno-implicit-prelude
-  - -Wno-missing-safe-haskell-mode
-  - -Wno-missing-export-lists
-  - -Wno-partial-fields
-  - -Wcompat
-  - -Werror=incomplete-record-updates
-  - -Werror=incomplete-patterns
-  - -Werror=incomplete-uni-patterns
-  - -Werror=missing-methods
-  - -Werror=tabs
-  - -Wredundant-constraints
-  - -Wincomplete-record-updates
-  - -Wunused-type-patterns
-  - -O2
-
-default-extensions:
-  - StrictData

rfcs/2024-09-01-smp-message-storage.md

@@ -96,7 +96,7 @@ else if write_msg = max_queue_messages:
     add quota_exceeded message to write_file
     update queue state: write_msg += 1
     append updated queue state to queue.log
-else ]
+else
     // It is required that `max_queue_messages < max_file_messages`,
     // so that we never need more than one additional write file.
     if write_msg >= max_file_messages: // queue file rotation

rfcs/2024-11-25-journal-expiration.md

@@ -0,0 +1,15 @@
+# Expiring messages in journal storage
+
+## Problem
+
+The journal storage servers recently migrated to do not delete delivered or expired messages, they only update pointers to journal file lines. The messages are actually deleted when the whole journal file is deleted (when fully deleted or fully expired).
+
+The problem is that in case the queue stops receiving the new messages then writing of messages won't switch to the new journal file, and the current journal file containing delivered or expired messages would never be deleted.
+
+## Solution
+
+Remove current journal file and update queue_state.log during message expiration of "idle" queue (that is, without any new messages received or delivered within 3 hours) in case when:
+- the queue is "empty" after the expiration
+- the queue contains only quota marker(s), in which case move them to a new journal file and update the queue_state accordingly. Quota markers can be kept indefinitely to prevent writing the new messages to the dormant queues that reached capacity, so it's important to handle this case.
+
+Also remove current journal file when the queue is opened in case it is empty (as it would not be ever expired in case it remains empty), and also update queue_state.log

rfcs/2024-11-25-queue-blobs-2.md

@@ -0,0 +1,58 @@
+# Blob extensions for SMP queues 2 and queue storage
+
+This document evolves the design proposed [here](./2024-09-09-smp-blobs.md).
+
+## Problems
+
+In addition to problems in the first doc, we have these issues with in-memory queue record storage:
+- many queues are idle or rarely used, but they are loaded to memory, and currently just loading all queues uses 20gb RAM on each server, and takes 10 min to process, increasing downtimes during restarts.
+- adding blobs to memory would make this problem much worse.
+
+## Proposed solution
+
+Move queues to the same journalling approach as [used for messages](./2024-09-01-smp-message-storage.md) now, with independent file names in the same folders.
+
+Each queue change would be logged to its own file, and every time the queue is opened the whole file will be read and compacted to a single line - replacing one store log for all queues, with individual log files for each queue.
+
+Queue deletion would not be making a record in the file, instead it would be deleting the entire folder - it would reduce retention period for any metadata of deleted queues.
+
+We could additionally record deletions to the central log, for debugging, and reset it on every start. But in this case we should not remove folders at the point of deletion, but rather mark them as deleted and delete on restart. TBC
+
+It would also allow simplifying blob storage by having only one blob per queue - for example, limied to 16kb (a bit smaller to fit in block) for contact address queues and 4-8kb for invitations (to fit PQ keys and conversation preferences).
+
+We would also need to be able to lookup recipient ID via sender/notifier/link IDs.
+
+One possible solution is to use and load to memory a central index file. But it is likely to also consume a lot of memory and result in slow starts.
+
+Another solution that is probably better is to use the same folder structure and put notifier/sender/link files with the ID of the recipient queue inside the files. So to locate recipient queue the sender would have to locate folder containing the reference file pointing to the recipient queue and then to locate the actual queue data.
+
+## Implementation details
+
+Each queue folder would these files:
+
+- queue_state.log (and timestamped backups) - to store pointers to message journals (already implemented)
+- messages.randomBase64.log - message journals (already implemented)
+- queue_rec.log (and timestamped backups) - to log complete queue record every time it is changed (so only the last line needs to be read following the same logic as with queue_state.log, to prevent file corruption).
+- blob.data, blob.data.bak, blob.timestamp.data - files for data blobs (to make sure some copy of this file is readable/correct in case of write corruption) - the same two step overwrite process will be used as currently with store log compacting:
+  - on write: 1. if file exists, move it to .bak, 2. store new blob to .data, 3. move .bak to .timestamp.data
+  - on read: 1. if .bak exists, move it to .data 2. use .data
+
+Additional suggestion to reduce probability of queue_state.log and queue_rec.log file corruption is to do one of the following:
+- log end of lines in the beginning of the output, not in the end, to prevent the last line from being corrupted in case the previous line was not fully stored. The downside is that the file will not be EOL terminated, and there will be no confirmation that the output was fully made.
+- log EOL both in the beginning and at the end of output, and ignore empty lines in between - this would both confirm that the last line is fully logged and prevent corruption of the next line in case it was not.
+- check the last byte of the file and log EOL if it is not EOL. Probably cleanest approach, but with a small performance cost.
+
+If queue folder is a reference to the queue, it may have one of these files:
+- notifier.id
+- sender.id
+- link.id
+
+These files would contain a one line with the recipient ID of the queue. These files would never change, they can only be deleted when queue is deleted or when notifier/link is deleted.
+
+There is logic in code preventing using the same ID in different contexts, and the ID size is large enough to make any collisions unlikely (192 bits), so with correctly working code the queue folder would either have one of reference files, and nothing else, or the queue and message files from the beginning of this section. But even if the same ID is re-used in different context, it should not cause any problems as file names don't overlap.
+
+While we could store different types of references in different types of folders, it would have additional costs of maintaining 4 folder hierarchies. Instead we could use the fact that it is one hierarchy to prevent using the same ID in different contexts.
+
+## Protocol
+
+The only change in protocol is that there will be only one blob per queue, without markers (see the previous doc). Otherwise the protocol and proposed privacy improvement seem reasonable.

simplexmq.cabal

@@ -1,11 +1,7 @@
 cabal-version: 1.12
 
--- This file has been generated from package.yaml by hpack version 0.35.0.
---
--- see: https://github.com/sol/hpack
-
 name:           simplexmq
-version:        6.1.2.0
+version:        6.2.0.6
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -66,6 +62,11 @@ flag use_crypton
   manual: True
   default: True
 
+flag client_library
+  description: Don't build server-related code.
+  manual: True
+  default: False
+
 library
   exposed-modules:
       Simplex.FileTransfer.Agent
@@ -77,13 +78,6 @@ library
       Simplex.FileTransfer.Crypto
       Simplex.FileTransfer.Description
       Simplex.FileTransfer.Protocol
-      Simplex.FileTransfer.Server
-      Simplex.FileTransfer.Server.Control
-      Simplex.FileTransfer.Server.Env
-      Simplex.FileTransfer.Server.Main
-      Simplex.FileTransfer.Server.Stats
-      Simplex.FileTransfer.Server.Store
-      Simplex.FileTransfer.Server.StoreLog
       Simplex.FileTransfer.Transport
       Simplex.FileTransfer.Types
       Simplex.FileTransfer.Util
@@ -154,34 +148,13 @@ library
       Simplex.Messaging.Encoding.String
       Simplex.Messaging.Notifications.Client
       Simplex.Messaging.Notifications.Protocol
-      Simplex.Messaging.Notifications.Server
-      Simplex.Messaging.Notifications.Server.Control
-      Simplex.Messaging.Notifications.Server.Env
-      Simplex.Messaging.Notifications.Server.Main
-      Simplex.Messaging.Notifications.Server.Push.APNS
-      Simplex.Messaging.Notifications.Server.Push.APNS.Internal
-      Simplex.Messaging.Notifications.Server.Stats
-      Simplex.Messaging.Notifications.Server.Store
-      Simplex.Messaging.Notifications.Server.StoreLog
       Simplex.Messaging.Notifications.Transport
       Simplex.Messaging.Notifications.Types
       Simplex.Messaging.Parsers
       Simplex.Messaging.Protocol
-      Simplex.Messaging.Server
       Simplex.Messaging.Server.CLI
-      Simplex.Messaging.Server.Control
-      Simplex.Messaging.Server.Env.STM
       Simplex.Messaging.Server.Expiration
-      Simplex.Messaging.Server.Information
-      Simplex.Messaging.Server.Main
-      Simplex.Messaging.Server.MsgStore
-      Simplex.Messaging.Server.MsgStore.STM
-      Simplex.Messaging.Server.NtfStore
-      Simplex.Messaging.Server.QueueStore
       Simplex.Messaging.Server.QueueStore.QueueInfo
-      Simplex.Messaging.Server.QueueStore.STM
-      Simplex.Messaging.Server.Stats
-      Simplex.Messaging.Server.StoreLog
       Simplex.Messaging.ServiceScheme
       Simplex.Messaging.Session
       Simplex.Messaging.TMap
@@ -204,13 +177,46 @@ library
       Simplex.RemoteControl.Discovery.Multicast
       Simplex.RemoteControl.Invitation
       Simplex.RemoteControl.Types
+  if !flag(client_library)
+      exposed-modules:
+          Simplex.FileTransfer.Server
+          Simplex.FileTransfer.Server.Control
+          Simplex.FileTransfer.Server.Env
+          Simplex.FileTransfer.Server.Main
+          Simplex.FileTransfer.Server.Stats
+          Simplex.FileTransfer.Server.Store
+          Simplex.FileTransfer.Server.StoreLog
+          Simplex.Messaging.Server
+          Simplex.Messaging.Server.Control
+          Simplex.Messaging.Server.Env.STM
+          Simplex.Messaging.Server.Information
+          Simplex.Messaging.Server.Main
+          Simplex.Messaging.Server.MsgStore
+          Simplex.Messaging.Server.MsgStore.Journal
+          Simplex.Messaging.Server.MsgStore.STM
+          Simplex.Messaging.Server.MsgStore.Types
+          Simplex.Messaging.Server.NtfStore
+          Simplex.Messaging.Server.QueueStore
+          Simplex.Messaging.Server.QueueStore.STM
+          Simplex.Messaging.Server.Stats
+          Simplex.Messaging.Server.StoreLog
+          Simplex.Messaging.Server.StoreLog.Types
+          Simplex.Messaging.Notifications.Server
+          Simplex.Messaging.Notifications.Server.Control
+          Simplex.Messaging.Notifications.Server.Env
+          Simplex.Messaging.Notifications.Server.Main
+          Simplex.Messaging.Notifications.Server.Push.APNS
+          Simplex.Messaging.Notifications.Server.Push.APNS.Internal
+          Simplex.Messaging.Notifications.Server.Stats
+          Simplex.Messaging.Notifications.Server.Store
+          Simplex.Messaging.Notifications.Server.StoreLog
   other-modules:
       Paths_simplexmq
   hs-source-dirs:
       src
   default-extensions:
       StrictData
-  ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-unused-packages -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2
+  ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-home-modules -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2
   include-dirs:
       cbits
   c-sources:
@@ -220,14 +226,12 @@ library
       crypto
   build-depends:
       aeson ==2.2.*
-    , ansi-terminal >=0.10 && <0.12
     , asn1-encoding ==0.9.*
     , asn1-types ==0.3.*
     , async ==2.2.*
     , attoparsec ==0.14.*
     , base >=4.14 && <5
     , base64-bytestring >=1.0 && <1.3
-    , case-insensitive ==1.2.*
     , composition ==1.0.*
     , constraints >=0.12 && <0.14
     , containers ==0.6.*
@@ -240,7 +244,6 @@ library
     , direct-sqlcipher ==2.3.*
     , directory ==1.3.*
     , filepath ==1.4.*
-    , hashable ==1.4.*
     , hourglass ==0.2.*
     , http-types ==0.12.*
     , http2 >=4.2.2 && <4.3
@@ -273,18 +276,22 @@ library
   default-language: Haskell2010
   if flag(swift)
     cpp-options: -DswiftJSON
+  if !flag(client_library)
+    build-depends:
+        case-insensitive ==1.2.*
+      , hashable ==1.4.*
   if impl(ghc >= 9.6.2)
     build-depends:
         bytestring ==0.11.*
-      , template-haskell ==2.20.*
       , text >=2.0.1 && <2.2
   if impl(ghc < 9.6.2)
     build-depends:
         bytestring ==0.10.*
-      , template-haskell ==2.16.*
       , text >=1.2.3.0 && <1.3
 
 executable ntf-server
+  if flag(client_library)
+      buildable: False
   main-is: Main.hs
   other-modules:
       Paths_simplexmq
@@ -292,75 +299,16 @@ executable ntf-server
       apps/ntf-server
   default-extensions:
       StrictData
-  ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-unused-packages -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts
+  ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts
   build-depends:
-      aeson ==2.2.*
-    , ansi-terminal >=0.10 && <0.12
-    , asn1-encoding ==0.9.*
-    , asn1-types ==0.3.*
-    , async ==2.2.*
-    , attoparsec ==0.14.*
-    , base >=4.14 && <5
-    , base64-bytestring >=1.0 && <1.3
-    , case-insensitive ==1.2.*
-    , composition ==1.0.*
-    , constraints >=0.12 && <0.14
-    , containers ==0.6.*
-    , crypton ==0.34.*
-    , crypton-x509 ==1.7.*
-    , crypton-x509-store ==1.6.*
-    , crypton-x509-validation ==1.6.*
-    , cryptostore ==0.3.*
-    , data-default ==0.7.*
-    , direct-sqlcipher ==2.3.*
-    , directory ==1.3.*
-    , filepath ==1.4.*
-    , hashable ==1.4.*
-    , hourglass ==0.2.*
-    , http-types ==0.12.*
-    , http2 >=4.2.2 && <4.3
-    , ini ==0.4.1
-    , iproute ==1.7.*
-    , iso8601-time ==0.1.*
-    , memory ==0.18.*
-    , mtl >=2.3.1 && <3.0
-    , network >=3.1.2.7 && <3.2
-    , network-info ==0.2.*
-    , network-transport ==0.5.6
-    , network-udp ==0.0.*
-    , optparse-applicative >=0.15 && <0.17
-    , process ==1.6.*
-    , random >=1.1 && <1.3
-    , simple-logger ==0.1.*
+      base
+    , simple-logger
     , simplexmq
-    , socks ==0.6.*
-    , sqlcipher-simple ==0.4.*
-    , stm ==2.5.*
-    , temporary ==1.3.*
-    , time ==1.12.*
-    , time-manager ==0.0.*
-    , tls >=1.9.0 && <1.10
-    , transformers ==0.6.*
-    , unliftio ==0.2.*
-    , unliftio-core ==0.2.*
-    , websockets ==0.12.*
-    , yaml ==0.11.*
-    , zstd ==0.1.3.*
   default-language: Haskell2010
-  if flag(swift)
-    cpp-options: -DswiftJSON
-  if impl(ghc >= 9.6.2)
-    build-depends:
-        bytestring ==0.11.*
-      , template-haskell ==2.20.*
-      , text >=2.0.1 && <2.2
-  if impl(ghc < 9.6.2)
-    build-depends:
-        bytestring ==0.10.*
-      , template-haskell ==2.16.*
-      , text >=1.2.3.0 && <1.3
 
 executable smp-server
+  if flag(client_library)
+      buildable: False
   main-is: Main.hs
   other-modules:
       Static
@@ -371,80 +319,27 @@ executable smp-server
       apps/smp-server/web
   default-extensions:
       StrictData
-  ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-unused-packages -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts
+  ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts
   build-depends:
-      aeson ==2.2.*
-    , ansi-terminal >=0.10 && <0.12
-    , asn1-encoding ==0.9.*
-    , asn1-types ==0.3.*
-    , async ==2.2.*
-    , attoparsec ==0.14.*
-    , base >=4.14 && <5
-    , base64-bytestring >=1.0 && <1.3
-    , case-insensitive ==1.2.*
-    , composition ==1.0.*
-    , constraints >=0.12 && <0.14
-    , containers ==0.6.*
-    , crypton ==0.34.*
-    , crypton-x509 ==1.7.*
-    , crypton-x509-store ==1.6.*
-    , crypton-x509-validation ==1.6.*
-    , cryptostore ==0.3.*
-    , data-default ==0.7.*
-    , direct-sqlcipher ==2.3.*
-    , directory ==1.3.*
+      base
+    , bytestring
+    , directory
     , file-embed
-    , filepath ==1.4.*
-    , hashable ==1.4.*
-    , hourglass ==0.2.*
-    , http-types ==0.12.*
-    , http2 >=4.2.2 && <4.3
-    , ini ==0.4.1
-    , iproute ==1.7.*
-    , iso8601-time ==0.1.*
-    , memory ==0.18.*
-    , mtl >=2.3.1 && <3.0
-    , network >=3.1.2.7 && <3.2
-    , network-info ==0.2.*
-    , network-transport ==0.5.6
-    , network-udp ==0.0.*
-    , optparse-applicative >=0.15 && <0.17
-    , process ==1.6.*
-    , random >=1.1 && <1.3
-    , simple-logger ==0.1.*
+    , filepath
+    , network
+    , simple-logger
     , simplexmq
-    , socks ==0.6.*
-    , sqlcipher-simple ==0.4.*
-    , stm ==2.5.*
-    , temporary ==1.3.*
-    , time ==1.12.*
-    , time-manager ==0.0.*
-    , tls >=1.9.0 && <1.10
-    , transformers ==0.6.*
-    , unliftio ==0.2.*
-    , unliftio-core ==0.2.*
+    , text
+    , unliftio
     , wai
     , wai-app-static
     , warp ==3.3.30
     , warp-tls ==3.4.7
-    , websockets ==0.12.*
-    , yaml ==0.11.*
-    , zstd ==0.1.3.*
   default-language: Haskell2010
-  if flag(swift)
-    cpp-options: -DswiftJSON
-  if impl(ghc >= 9.6.2)
-    build-depends:
-        bytestring ==0.11.*
-      , template-haskell ==2.20.*
-      , text >=2.0.1 && <2.2
-  if impl(ghc < 9.6.2)
-    build-depends:
-        bytestring ==0.10.*
-      , template-haskell ==2.16.*
-      , text >=1.2.3.0 && <1.3
 
 executable xftp
+  if flag(client_library)
+      buildable: False
   main-is: Main.hs
   other-modules:
       Paths_simplexmq
@@ -452,75 +347,15 @@ executable xftp
       apps/xftp
   default-extensions:
       StrictData
-  ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-unused-packages -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts
+  ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts
   build-depends:
-      aeson ==2.2.*
-    , ansi-terminal >=0.10 && <0.12
-    , asn1-encoding ==0.9.*
-    , asn1-types ==0.3.*
-    , async ==2.2.*
-    , attoparsec ==0.14.*
-    , base >=4.14 && <5
-    , base64-bytestring >=1.0 && <1.3
-    , case-insensitive ==1.2.*
-    , composition ==1.0.*
-    , constraints >=0.12 && <0.14
-    , containers ==0.6.*
-    , crypton ==0.34.*
-    , crypton-x509 ==1.7.*
-    , crypton-x509-store ==1.6.*
-    , crypton-x509-validation ==1.6.*
-    , cryptostore ==0.3.*
-    , data-default ==0.7.*
-    , direct-sqlcipher ==2.3.*
-    , directory ==1.3.*
-    , filepath ==1.4.*
-    , hashable ==1.4.*
-    , hourglass ==0.2.*
-    , http-types ==0.12.*
-    , http2 >=4.2.2 && <4.3
-    , ini ==0.4.1
-    , iproute ==1.7.*
-    , iso8601-time ==0.1.*
-    , memory ==0.18.*
-    , mtl >=2.3.1 && <3.0
-    , network >=3.1.2.7 && <3.2
-    , network-info ==0.2.*
-    , network-transport ==0.5.6
-    , network-udp ==0.0.*
-    , optparse-applicative >=0.15 && <0.17
-    , process ==1.6.*
-    , random >=1.1 && <1.3
-    , simple-logger ==0.1.*
+      base
     , simplexmq
-    , socks ==0.6.*
-    , sqlcipher-simple ==0.4.*
-    , stm ==2.5.*
-    , temporary ==1.3.*
-    , time ==1.12.*
-    , time-manager ==0.0.*
-    , tls >=1.9.0 && <1.10
-    , transformers ==0.6.*
-    , unliftio ==0.2.*
-    , unliftio-core ==0.2.*
-    , websockets ==0.12.*
-    , yaml ==0.11.*
-    , zstd ==0.1.3.*
   default-language: Haskell2010
-  if flag(swift)
-    cpp-options: -DswiftJSON
-  if impl(ghc >= 9.6.2)
-    build-depends:
-        bytestring ==0.11.*
-      , template-haskell ==2.20.*
-      , text >=2.0.1 && <2.2
-  if impl(ghc < 9.6.2)
-    build-depends:
-        bytestring ==0.10.*
-      , template-haskell ==2.16.*
-      , text >=1.2.3.0 && <1.3
 
 executable xftp-server
+  if flag(client_library)
+      buildable: False
   main-is: Main.hs
   other-modules:
       Paths_simplexmq
@@ -528,75 +363,16 @@ executable xftp-server
       apps/xftp-server
   default-extensions:
       StrictData
-  ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-unused-packages -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts
+  ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts
   build-depends:
-      aeson ==2.2.*
-    , ansi-terminal >=0.10 && <0.12
-    , asn1-encoding ==0.9.*
-    , asn1-types ==0.3.*
-    , async ==2.2.*
-    , attoparsec ==0.14.*
-    , base >=4.14 && <5
-    , base64-bytestring >=1.0 && <1.3
-    , case-insensitive ==1.2.*
-    , composition ==1.0.*
-    , constraints >=0.12 && <0.14
-    , containers ==0.6.*
-    , crypton ==0.34.*
-    , crypton-x509 ==1.7.*
-    , crypton-x509-store ==1.6.*
-    , crypton-x509-validation ==1.6.*
-    , cryptostore ==0.3.*
-    , data-default ==0.7.*
-    , direct-sqlcipher ==2.3.*
-    , directory ==1.3.*
-    , filepath ==1.4.*
-    , hashable ==1.4.*
-    , hourglass ==0.2.*
-    , http-types ==0.12.*
-    , http2 >=4.2.2 && <4.3
-    , ini ==0.4.1
-    , iproute ==1.7.*
-    , iso8601-time ==0.1.*
-    , memory ==0.18.*
-    , mtl >=2.3.1 && <3.0
-    , network >=3.1.2.7 && <3.2
-    , network-info ==0.2.*
-    , network-transport ==0.5.6
-    , network-udp ==0.0.*
-    , optparse-applicative >=0.15 && <0.17
-    , process ==1.6.*
-    , random >=1.1 && <1.3
-    , simple-logger ==0.1.*
+      base
+    , simple-logger
     , simplexmq
-    , socks ==0.6.*
-    , sqlcipher-simple ==0.4.*
-    , stm ==2.5.*
-    , temporary ==1.3.*
-    , time ==1.12.*
-    , time-manager ==0.0.*
-    , tls >=1.9.0 && <1.10
-    , transformers ==0.6.*
-    , unliftio ==0.2.*
-    , unliftio-core ==0.2.*
-    , websockets ==0.12.*
-    , yaml ==0.11.*
-    , zstd ==0.1.3.*
   default-language: Haskell2010
-  if flag(swift)
-    cpp-options: -DswiftJSON
-  if impl(ghc >= 9.6.2)
-    build-depends:
-        bytestring ==0.11.*
-      , template-haskell ==2.20.*
-      , text >=2.0.1 && <2.2
-  if impl(ghc < 9.6.2)
-    build-depends:
-        bytestring ==0.10.*
-      , template-haskell ==2.16.*
-      , text >=1.2.3.0 && <1.3
 
 test-suite simplexmq-test
+  if flag(client_library)
+      buildable: False
   type: exitcode-stdio-1.0
   main-is: Test.hs
   other-modules:
@@ -608,12 +384,14 @@ test-suite simplexmq-test
       AgentTests.MigrationTests
       AgentTests.NotificationTests
       AgentTests.SchemaDump
+      AgentTests.ServerChoice
       AgentTests.SQLiteTests
       CLITests
       CoreTests.BatchingTests
       CoreTests.CryptoFileTests
       CoreTests.CryptoTests
       CoreTests.EncodingTests
+      CoreTests.MsgStoreTests
       CoreTests.RetryIntervalTests
       CoreTests.SOCKSSettings
       CoreTests.StoreLogTests
@@ -641,87 +419,53 @@ test-suite simplexmq-test
       apps/smp-server/web
   default-extensions:
       StrictData
-  ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-unused-packages -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts -with-rtsopts=-A64M -with-rtsopts=-N1
+  ghc-options: -Weverything -Wno-missing-exported-signatures -Wno-missing-import-lists -Wno-missed-specialisations -Wno-all-missed-specialisations -Wno-unsafe -Wno-safe -Wno-missing-local-signatures -Wno-missing-kind-signatures -Wno-missing-deriving-strategies -Wno-monomorphism-restriction -Wno-prepositive-qualified-module -Wno-implicit-prelude -Wno-missing-safe-haskell-mode -Wno-missing-export-lists -Wno-partial-fields -Wcompat -Werror=incomplete-record-updates -Werror=incomplete-patterns -Werror=incomplete-uni-patterns -Werror=missing-methods -Werror=tabs -Wredundant-constraints -Wincomplete-record-updates -Wunused-type-patterns -O2 -threaded -rtsopts -with-rtsopts=-A64M -with-rtsopts=-N1
   build-depends:
-      HUnit ==1.6.*
-    , QuickCheck ==2.14.*
-    , aeson ==2.2.*
-    , ansi-terminal >=0.10 && <0.12
-    , asn1-encoding ==0.9.*
-    , asn1-types ==0.3.*
-    , async ==2.2.*
-    , attoparsec ==0.14.*
-    , base >=4.14 && <5
-    , base64-bytestring >=1.0 && <1.3
-    , case-insensitive ==1.2.*
-    , composition ==1.0.*
-    , constraints >=0.12 && <0.14
-    , containers ==0.6.*
-    , crypton ==0.34.*
-    , crypton-x509 ==1.7.*
-    , crypton-x509-store ==1.6.*
-    , crypton-x509-validation ==1.6.*
-    , cryptostore ==0.3.*
-    , data-default ==0.7.*
+      base
+    , aeson
+    , async
+    , base64-bytestring
+    , bytestring
+    , containers
+    , crypton
+    , crypton-x509
+    , crypton-x509-store
+    , crypton-x509-validation
     , deepseq ==1.4.*
-    , direct-sqlcipher ==2.3.*
-    , directory ==1.3.*
+    , directory
     , file-embed
-    , filepath ==1.4.*
+    , filepath
     , generic-random ==1.5.*
-    , hashable ==1.4.*
-    , hourglass ==0.2.*
+    , hashable
     , hspec ==2.11.*
-    , hspec-core ==2.11.*
     , http-client
-    , http-client-tls
-    , http-types ==0.12.*
-    , http2 >=4.2.2 && <4.3
-    , ini ==0.4.1
-    , iproute ==1.7.*
-    , iso8601-time ==0.1.*
+    , http-types
+    , http2
+    , HUnit ==1.6.*
+    , ini
+    , iso8601-time
     , main-tester ==0.2.*
-    , memory ==0.18.*
-    , mtl >=2.3.1 && <3.0
-    , network >=3.1.2.7 && <3.2
-    , network-info ==0.2.*
-    , network-transport ==0.5.6
-    , network-udp ==0.0.*
-    , optparse-applicative >=0.15 && <0.17
-    , process ==1.6.*
-    , random >=1.1 && <1.3
+    , memory
+    , mtl
+    , network
+    , process
+    , QuickCheck ==2.14.*
+    , random
     , silently ==1.2.*
-    , simple-logger ==0.1.*
+    , simple-logger
     , simplexmq
-    , socks ==0.6.*
-    , sqlcipher-simple ==0.4.*
-    , stm ==2.5.*
-    , temporary ==1.3.*
-    , time ==1.12.*
-    , time-manager ==0.0.*
+    , sqlcipher-simple
+    , stm
+    , text
+    , time
     , timeit ==2.0.*
-    , tls >=1.9.0 && <1.10
-    , transformers ==0.6.*
-    , unliftio ==0.2.*
-    , unliftio-core ==0.2.*
+    , transformers
+    , unliftio
+    , unliftio-core
     , unordered-containers
     , wai
     , wai-app-static
     , warp
     , warp-tls
-    , websockets ==0.12.*
-    , yaml ==0.11.*
-    , zstd ==0.1.3.*
+    , yaml
   default-language: Haskell2010
-  if flag(swift)
-    cpp-options: -DswiftJSON
-  if impl(ghc >= 9.6.2)
-    build-depends:
-        bytestring ==0.11.*
-      , template-haskell ==2.20.*
-      , text >=2.0.1 && <2.2
-  if impl(ghc < 9.6.2)
-    build-depends:
-        bytestring ==0.10.*
-      , template-haskell ==2.16.*
-      , text >=1.2.3.0 && <1.3

src/Simplex/FileTransfer/Agent.hs

@@ -461,14 +461,14 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
           pure srv
           where
             tryCreate = do
-              usedSrvs <- newTVarIO ([] :: [XFTPServer])
+              triedHosts <- newTVarIO S.empty
               let AgentClient {xftpServers} = c
               userSrvCount <- liftIO $ length <$> TM.lookupIO userId xftpServers
               withRetryIntervalCount (riFast ri) $ \n _ loop -> do
                 liftIO $ waitWhileSuspended c
                 liftIO $ waitForUserNetwork c
                 let triedAllSrvs = n > userSrvCount
-                createWithNextSrv usedSrvs
+                createWithNextSrv triedHosts
                   `catchAgentError` \e -> retryOnError "XFTP prepare worker" (retryLoop loop triedAllSrvs e) (throwE e) e
               where
                 -- we don't do closeXFTPServerClient here to not risk closing connection for concurrent chunk upload
@@ -477,10 +477,10 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
                     when (triedAllSrvs && serverHostError e) $ notify c sndFileEntityId $ SFWARN e
                   liftIO $ assertAgentForeground c
                   loop
-            createWithNextSrv usedSrvs = do
+            createWithNextSrv triedHosts = do
               deleted <- withStore' c $ \db -> getSndFileDeleted db sndFileId
               when deleted $ throwE $ FILE NO_FILE
-              withNextSrv c userId usedSrvs [] $ \srvAuth -> do
+              withNextSrv c userId storageSrvs triedHosts [] $ \srvAuth -> do
                 replica <- agentXFTPNewChunk c ch numRecipients' srvAuth
                 pure (replica, srvAuth)
 
@@ -546,8 +546,8 @@ runXFTPSndWorker c srv Worker {doWork} = do
         withStore' c $ \db -> updateSndFileComplete db sndFileId
       where
         addRecipients :: SndFileChunk -> SndFileChunkReplica -> AM SndFileChunkReplica
-        addRecipients ch@SndFileChunk {numRecipients} cr@SndFileChunkReplica {rcvIdsKeys}
-          | length rcvIdsKeys > numRecipients = throwE $ INTERNAL "too many recipients"
+        addRecipients ch@SndFileChunk {numRecipients} cr@SndFileChunkReplica {sndChunkReplicaId, rcvIdsKeys}
+          | length rcvIdsKeys > numRecipients = throwE $ INTERNAL ("too many recipients, sndChunkReplicaId = " <> show sndChunkReplicaId)
           | length rcvIdsKeys == numRecipients = pure cr
           | otherwise = do
               let numRecipients' = min (numRecipients - length rcvIdsKeys) maxRecipients

src/Simplex/FileTransfer/Client.hs

@@ -29,7 +29,6 @@ import qualified Data.X509.Validation as XV
 import qualified Network.HTTP.Types as N
 import qualified Network.HTTP2.Client as H
 import Simplex.FileTransfer.Protocol
-import Simplex.FileTransfer.Server.Env (supportedXFTPhandshakes)
 import Simplex.FileTransfer.Transport
 import Simplex.Messaging.Client
   ( NetworkConfig (..),

src/Simplex/FileTransfer/Server.hs

@@ -181,6 +181,7 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
     stopServer = do
       withFileLog closeStoreLog
       saveServerStats
+      logInfo "Server stopped"
 
     expireFilesThread_ :: XFTPServerConfig -> [M ()]
     expireFilesThread_ XFTPServerConfig {fileExpiration = Just fileExp} = [expireFiles fileExp]

src/Simplex/FileTransfer/Server/Env.hs

@@ -28,7 +28,6 @@ import Simplex.FileTransfer.Transport (VersionRangeXFTP)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol (BasicAuth, RcvPublicAuthKey)
 import Simplex.Messaging.Server.Expiration
-import Simplex.Messaging.Transport (ALPN)
 import Simplex.Messaging.Transport.Server (ServerCredentials (..), TransportServerConfig (..), loadFingerprint, loadServerCredential)
 import Simplex.Messaging.Util (tshow)
 import System.IO (IOMode (..))
@@ -96,9 +95,6 @@ defaultFileExpiration =
       checkInterval = 2 * 3600 -- seconds, 2 hours
     }
 
-supportedXFTPhandshakes :: [ALPN]
-supportedXFTPhandshakes = ["xftp/1"]
-
 newXFTPServerEnv :: XFTPServerConfig -> IO XFTPEnv
 newXFTPServerEnv config@XFTPServerConfig {storeLogFile, fileSizeQuota, xftpCredentials} = do
   random <- C.newRandom

src/Simplex/FileTransfer/Server/Main.hs

@@ -51,7 +51,9 @@ xftpServerCLI cfgPath logPath = do
         True -> readIniFile iniFile >>= either exitError runServer
         _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
     Delete -> do
-      confirmOrExit "WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
+      confirmOrExit
+        "WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
+        "Server NOT deleted"
       deleteDirIfExists cfgPath
       deleteDirIfExists logPath
       putStrLn "Deleted configuration and log files"

src/Simplex/FileTransfer/Transport.hs

@@ -12,6 +12,7 @@ module Simplex.FileTransfer.Transport
   ( supportedFileServerVRange,
     authCmdsXFTPVersion,
     xftpClientHandshakeStub,
+    supportedXFTPhandshakes,
     XFTPClientHandshake (..),
     -- xftpClientHandshake,
     XFTPServerHandshake (..),
@@ -57,7 +58,7 @@ import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers
 import Simplex.Messaging.Protocol (CommandError)
-import Simplex.Messaging.Transport (SessionId, THandle (..), THandleParams (..), TransportError (..), TransportPeer (..))
+import Simplex.Messaging.Transport (ALPN, SessionId, THandle (..), THandleParams (..), TransportError (..), TransportPeer (..))
 import Simplex.Messaging.Transport.HTTP2.File
 import Simplex.Messaging.Util (bshow, tshow)
 import Simplex.Messaging.Version
@@ -101,6 +102,9 @@ supportedFileServerVRange = mkVersionRange initialXFTPVersion currentXFTPVersion
 xftpClientHandshakeStub :: c -> Maybe C.KeyPairX25519 -> C.KeyHash -> VersionRangeXFTP -> ExceptT TransportError IO (THandle XFTPVersion c 'TClient)
 xftpClientHandshakeStub _c _ks _keyHash _xftpVRange = throwE TEVersion
 
+supportedXFTPhandshakes :: [ALPN]
+supportedXFTPhandshakes = ["xftp/1"]
+
 data XFTPServerHandshake = XFTPServerHandshake
   { xftpVersionRange :: VersionRangeXFTP,
     sessionId :: SessionId,

src/Simplex/Messaging/Agent.hs

@@ -120,6 +120,8 @@ module Simplex.Messaging.Agent
     debugAgentLocks,
     getAgentSubscriptions,
     logConnection,
+    -- for tests
+    withAgentEnv,
   )
 where
 
@@ -815,11 +817,13 @@ newConnToAccept c connId enableNtfs invId pqSup = do
 
 joinConn :: AgentClient -> UserId -> ConnId -> Bool -> ConnectionRequestUri c -> ConnInfo -> PQSupport -> SubscriptionMode -> AM SndQueueSecured
 joinConn c userId connId enableNtfs cReq cInfo pqSupport subMode = do
-  srv <- case cReq of
-    CRInvitationUri ConnReqUriData {crSmpQueues = q :| _} _ ->
-      getNextServer c userId [qServer q]
-    _ -> getSMPServer c userId
+  srv <- getNextSMPServer c userId [qServer cReqQueue]
   joinConnSrv c userId connId enableNtfs cReq cInfo pqSupport subMode srv
+  where
+    cReqQueue :: SMPQueueUri
+    cReqQueue = case cReq of
+      CRInvitationUri ConnReqUriData {crSmpQueues = q :| _} _ -> q
+      CRContactUri ConnReqUriData {crSmpQueues = q :| _} -> q
 
 startJoinInvitation :: AgentClient -> UserId -> ConnId -> Maybe SndQueue -> Bool -> ConnectionRequestUri 'CMInvitation -> PQSupport -> AM (ConnData, SndQueue, CR.SndE2ERatchetParams 'C.X448)
 startJoinInvitation c userId connId sq_ enableNtfs cReqUri pqSup =
@@ -1194,14 +1198,13 @@ runCommandProcessing c@AgentClient {subQ} connId server_ Worker {doWork} = do
     processCmd ri PendingCommand {cmdId, corrId, userId, command} pendingCmds = case command of
       AClientCommand cmd -> case cmd of
         NEW enableNtfs (ACM cMode) pqEnc subMode -> noServer $ do
-          usedSrvs <- newTVarIO ([] :: [SMPServer])
-          tryCommand . withNextSrv c userId usedSrvs [] $ \srv -> do
+          triedHosts <- newTVarIO S.empty
+          tryCommand . withNextSrv c userId storageSrvs triedHosts [] $ \srv -> do
             cReq <- newRcvConnSrv c userId connId enableNtfs cMode Nothing pqEnc subMode srv
             notify $ INV (ACR cMode cReq)
         JOIN enableNtfs (ACR _ cReq@(CRInvitationUri ConnReqUriData {crSmpQueues = q :| _} _)) pqEnc subMode connInfo -> noServer $ do
-          let initUsed = [qServer q]
-          usedSrvs <- newTVarIO initUsed
-          tryCommand . withNextSrv c userId usedSrvs initUsed $ \srv -> do
+          triedHosts <- newTVarIO S.empty
+          tryCommand . withNextSrv c userId storageSrvs triedHosts [qServer q] $ \srv -> do
             sqSecured <- joinConnSrvAsync c userId connId enableNtfs cReq connInfo pqEnc subMode srv
             notify $ JOINED sqSecured
         LET confId ownCInfo -> withServer' . tryCommand $ allowConnection' c connId confId ownCInfo >> notify OK
@@ -1649,8 +1652,8 @@ switchDuplexConnection c (DuplexConnection cData@ConnData {connId, userId} rqs s
   checkRQSwchStatus rq RSSwitchStarted
   clientVRange <- asks $ smpClientVRange . config
   -- try to get the server that is different from all queues, or at least from the primary rcv queue
-  srvAuth@(ProtoServerWithAuth srv _) <- getNextServer c userId $ map qServer (L.toList rqs) <> map qServer (L.toList sqs)
-  srv' <- if srv == server then getNextServer c userId [server] else pure srvAuth
+  srvAuth@(ProtoServerWithAuth srv _) <- getNextSMPServer c userId $ map qServer (L.toList rqs) <> map qServer (L.toList sqs)
+  srv' <- if srv == server then getNextSMPServer c userId [server] else pure srvAuth
   (q, qUri, tSess, sessId) <- newRcvQueue c userId connId srv' clientVRange SMSubscribe False
   let rq' = (q :: NewRcvQueue) {primary = True, dbReplaceQueueId = Just dbQueueId}
   rq'' <- withStore c $ \db -> addConnRcvQueue db connId rq'
@@ -2158,9 +2161,13 @@ debugAgentLocks AgentClient {connLocks = cs, invLocks = is, deleteLock = d} = do
     getLocks ls = atomically $ M.mapKeys (B.unpack . strEncode) . M.mapMaybe id <$> (mapM tryReadTMVar =<< readTVar ls)
 
 getSMPServer :: AgentClient -> UserId -> AM SMPServerWithAuth
-getSMPServer c userId = withUserServers c userId pickServer
+getSMPServer c userId = getNextSMPServer c userId []
 {-# INLINE getSMPServer #-}
 
+getNextSMPServer :: AgentClient -> UserId -> [SMPServer] -> AM SMPServerWithAuth
+getNextSMPServer c userId = getNextServer c userId storageSrvs
+{-# INLINE getNextSMPServer #-}
+
 subscriber :: AgentClient -> AM' ()
 subscriber c@AgentClient {msgQ} = forever $ do
   t <- atomically $ readTBQueue msgQ

src/Simplex/Messaging/Agent/Client.hs

@@ -30,6 +30,7 @@ module Simplex.Messaging.Agent.Client
     withConnLocks,
     withInvLock,
     withLockMap,
+    getMapLock,
     ipAddressProtected,
     closeAgentClient,
     closeProtocolServerClients,
@@ -148,7 +149,6 @@ module Simplex.Messaging.Agent.Client
     userServers,
     pickServer,
     getNextServer,
-    withUserServers,
     withNextSrv,
     incSMPServerStat,
     incSMPServerStat',
@@ -192,12 +192,12 @@ import qualified Data.ByteString.Char8 as B
 import Data.Either (isRight, partitionEithers)
 import Data.Functor (($>))
 import Data.Int (Int64)
-import Data.List (deleteFirstsBy, find, foldl', partition, (\\))
+import Data.List (find, foldl', partition)
 import Data.List.NonEmpty (NonEmpty (..), (<|))
 import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
-import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing, listToMaybe)
+import Data.Maybe (catMaybes, fromMaybe, isJust, isNothing, listToMaybe, mapMaybe)
 import Data.Set (Set)
 import qualified Data.Set as S
 import Data.Text (Text)
@@ -263,7 +263,6 @@ import Simplex.Messaging.Protocol
     VersionSMPC,
     XFTPServer,
     XFTPServerWithAuth,
-    sameSrvAddr',
     pattern NoEntity,
   )
 import qualified Simplex.Messaging.Protocol as SMP
@@ -500,7 +499,7 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} currentTs a
   getMsgLocks <- TM.emptyIO
   connLocks <- TM.emptyIO
   invLocks <- TM.emptyIO
-  deleteLock <- atomically createLock
+  deleteLock <- createLockIO
   smpSubWorkers <- TM.emptyIO
   smpServersStats <- TM.emptyIO
   xftpServersStats <- TM.emptyIO
@@ -618,7 +617,7 @@ getSMPServerClient c@AgentClient {active, smpClients, workerSeq} tSess = do
 getSMPProxyClient :: AgentClient -> Maybe SMPServerWithAuth -> SMPTransportSession -> AM (SMPConnectedClient, Either AgentErrorType ProxiedRelay)
 getSMPProxyClient c@AgentClient {active, smpClients, smpProxiedRelays, workerSeq} proxySrv_ destSess@(userId, destSrv, qId) = do
   unlessM (readTVarIO active) $ throwE INACTIVE
-  proxySrv <- maybe (getNextServer c userId [destSrv]) pure proxySrv_
+  proxySrv <- maybe (getNextServer c userId proxySrvs [destSrv]) pure proxySrv_
   ts <- liftIO getCurrentTime
   atomically (getClientVar proxySrv ts) >>= \(tSess, auth, v) ->
     either (newProxyClient tSess auth ts) (waitForProxyClient tSess auth) v
@@ -1073,7 +1072,7 @@ sendOrProxySMPCommand ::
   (SMPClient -> ProxiedRelay -> ExceptT SMPClientError IO (Either ProxyClientError ())) ->
   (SMPClient -> ExceptT SMPClientError IO ()) ->
   AM (Maybe SMPServer)
-sendOrProxySMPCommand c userId destSrv connId cmdStr senderId sendCmdViaProxy sendCmdDirectly = do
+sendOrProxySMPCommand c userId destSrv@ProtocolServer {host = destHosts} connId cmdStr senderId sendCmdViaProxy sendCmdDirectly = do
   tSess <- mkTransportSession c userId destSrv connId
   ifM shouldUseProxy (sendViaProxy Nothing tSess) (sendDirectly tSess $> Nothing)
   where
@@ -1092,7 +1091,7 @@ sendOrProxySMPCommand c userId destSrv connId cmdStr senderId sendCmdViaProxy se
         SPFAllow -> True
         SPFAllowProtected -> ipAddressProtected cfg destSrv
         SPFProhibit -> False
-    unknownServer = liftIO $ maybe True (notElem destSrv . knownSrvs) <$> TM.lookupIO userId (smpServers c)
+    unknownServer = liftIO $ maybe True (\srvs -> all (`S.notMember` knownHosts srvs) destHosts) <$> TM.lookupIO userId (smpServers c)
     sendViaProxy :: Maybe SMPServerWithAuth -> SMPTransportSession -> AM (Maybe SMPServer)
     sendViaProxy proxySrv_ destSess@(_, _, connId_) = do
       r <- tryAgentError . withProxySession c proxySrv_ destSess senderId ("PFWD " <> cmdStr) $ \(SMPConnectedClient smp _, proxySess@ProxiedRelay {prBasicAuth}) -> do
@@ -1387,6 +1386,7 @@ temporaryAgentError = \case
   PROXY _ _ (ProxyProtocolError (SMP.PROXY (SMP.BROKER e))) -> tempBrokerError e
   PROXY _ _ (ProxyProtocolError (SMP.PROXY SMP.NO_SESSION)) -> True
   INACTIVE -> True
+  CRITICAL True _ -> True -- critical errors that do not show restart button are likely to be permanent
   _ -> False
   where
     tempBrokerError = \case
@@ -2034,33 +2034,82 @@ userServers c = case protocolTypeI @p of
   SPXFTP -> xftpServers c
 {-# INLINE userServers #-}
 
-pickServer :: forall p. NonEmpty (ProtoServerWithAuth p) -> AM (ProtoServerWithAuth p)
+pickServer :: NonEmpty (Maybe OperatorId, ProtoServerWithAuth p) -> AM (ProtoServerWithAuth p)
 pickServer = \case
-  srv :| [] -> pure srv
+  (_, srv) :| [] -> pure srv
   servers -> do
     gen <- asks randomServer
-    atomically $ (servers L.!!) <$> stateTVar gen (randomR (0, L.length servers - 1))
+    atomically $ snd . (servers L.!!) <$> stateTVar gen (randomR (0, L.length servers - 1))
 
-getNextServer :: forall p. (ProtocolTypeI p, UserProtocol p) => AgentClient -> UserId -> [ProtocolServer p] -> AM (ProtoServerWithAuth p)
-getNextServer c userId usedSrvs = withUserServers c userId $ \srvs ->
-  case L.nonEmpty $ deleteFirstsBy sameSrvAddr' (L.toList srvs) (map noAuthSrv usedSrvs) of
-    Just srvs' -> pickServer srvs'
-    _ -> pickServer srvs
+getNextServer ::
+  (ProtocolTypeI p, UserProtocol p) =>
+  AgentClient ->
+  UserId -> 
+  (UserServers p -> NonEmpty (Maybe OperatorId, ProtoServerWithAuth p)) ->
+  [ProtocolServer p] ->
+  AM (ProtoServerWithAuth p)
+getNextServer c userId srvsSel usedSrvs = do
+  srvs <- getUserServers_ c userId srvsSel
+  snd <$> getNextServer_ srvs (usedOperatorsHosts srvs usedSrvs)
 
-withUserServers :: forall p a. (ProtocolTypeI p, UserProtocol p) => AgentClient -> UserId -> (NonEmpty (ProtoServerWithAuth p) -> AM a) -> AM a
-withUserServers c userId action =
+usedOperatorsHosts :: NonEmpty (Maybe OperatorId, ProtoServerWithAuth p) -> [ProtocolServer p] -> (Set (Maybe OperatorId), Set TransportHost)
+usedOperatorsHosts srvs usedSrvs = (usedOperators, usedHosts)
+  where
+    usedHosts = S.unions $ map serverHosts usedSrvs
+    usedOperators = S.fromList $ mapMaybe usedOp $ L.toList srvs
+    usedOp (op, srv) = if hasUsedHost srv then Just op else Nothing
+    hasUsedHost (ProtoServerWithAuth srv _) = any (`S.member` usedHosts) $ serverHosts srv
+
+getNextServer_ ::
+  (ProtocolTypeI p, UserProtocol p) =>
+  NonEmpty (Maybe OperatorId, ProtoServerWithAuth p) ->
+  (Set (Maybe OperatorId), Set TransportHost) ->
+  AM (NonEmpty (Maybe OperatorId, ProtoServerWithAuth p), ProtoServerWithAuth p)
+getNextServer_ servers (usedOperators, usedHosts) = do
+  -- choose from servers of unused operators, when possible
+  let otherOpsSrvs = filterOrAll ((`S.notMember` usedOperators) . fst) servers
+      -- choose from servers with unused hosts when possible
+      unusedSrvs = filterOrAll (isUnusedServer usedHosts) otherOpsSrvs
+  (otherOpsSrvs,) <$> pickServer unusedSrvs
+  where
+    filterOrAll p srvs = fromMaybe srvs $ L.nonEmpty $ L.filter p srvs
+
+isUnusedServer :: Set TransportHost -> (Maybe OperatorId, ProtoServerWithAuth p) -> Bool
+isUnusedServer usedHosts (_, ProtoServerWithAuth ProtocolServer {host} _) = all (`S.notMember` usedHosts) host
+
+getUserServers_ ::
+  (ProtocolTypeI p, UserProtocol p) =>
+  AgentClient ->
+  UserId ->
+  (UserServers p -> NonEmpty (Maybe OperatorId, ProtoServerWithAuth p)) ->
+  AM (NonEmpty (Maybe OperatorId, ProtoServerWithAuth p))
+getUserServers_ c userId srvsSel =
   liftIO (TM.lookupIO userId $ userServers c) >>= \case
-    Just srvs -> action $ enabledSrvs srvs
+    Just srvs -> pure $ srvsSel srvs
     _ -> throwE $ INTERNAL "unknown userId - no user servers"
 
-withNextSrv :: forall p a. (ProtocolTypeI p, UserProtocol p) => AgentClient -> UserId -> TVar [ProtocolServer p] -> [ProtocolServer p] -> (ProtoServerWithAuth p -> AM a) -> AM a
-withNextSrv c userId usedSrvs initUsed action = do
-  used <- readTVarIO usedSrvs
-  srvAuth@(ProtoServerWithAuth srv _) <- getNextServer c userId used
-  srvs_ <- liftIO $ TM.lookupIO userId $ userServers c
-  let unused = maybe [] ((\\ used) . map protoServer . L.toList . enabledSrvs) srvs_
-      used' = if null unused then initUsed else srv : used
-  atomically $ writeTVar usedSrvs $! used'
+-- This function checks used servers and operators every time to allow
+-- changing configuration while retry look is executing.
+-- This function is not thread safe.
+withNextSrv ::
+  (ProtocolTypeI p, UserProtocol p) =>
+  AgentClient ->
+  UserId ->
+  (UserServers p -> NonEmpty (Maybe OperatorId, ProtoServerWithAuth p)) ->
+  TVar (Set TransportHost) ->
+  [ProtocolServer p] -> 
+  (ProtoServerWithAuth p -> AM a) ->
+  AM a
+withNextSrv c userId srvsSel triedHosts usedSrvs action = do
+  srvs <- getUserServers_ c userId srvsSel
+  let (usedOperators, usedHosts) = usedOperatorsHosts srvs usedSrvs
+  tried <- readTVarIO triedHosts
+  let triedOrUsed = S.union tried usedHosts
+  (otherOpsSrvs, srvAuth@(ProtoServerWithAuth srv _)) <- getNextServer_ srvs (usedOperators, triedOrUsed)
+  let newHosts = serverHosts srv
+      unusedSrvs = L.filter (isUnusedServer $ S.union triedOrUsed newHosts) otherOpsSrvs
+      !tried' = if null unusedSrvs then S.empty else S.union tried newHosts
+  atomically $ writeTVar triedHosts tried'
   action srvAuth
 
 incSMPServerStat :: AgentClient -> UserId -> SMPServer -> (AgentSMPServerStats -> TVar Int) -> STM ()

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

@@ -17,11 +17,14 @@ module Simplex.Messaging.Agent.Env.SQLite
     AgentConfig (..),
     InitialAgentServers (..),
     ServerCfg (..),
+    ServerRoles (..),
+    OperatorId,
     UserServers (..),
     NetworkConfig (..),
     presetServerCfg,
-    enabledServerCfg,
+    allRoles,
     mkUserServers,
+    serverHosts,
     defaultAgentConfig,
     defaultReconnectInterval,
     tryAgentError,
@@ -42,6 +45,7 @@ module Simplex.Messaging.Agent.Env.SQLite
 where
 
 import Control.Concurrent (ThreadId)
+import Control.Exception (BlockedIndefinitelyOnSTM (..), SomeException, fromException)
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
 import Control.Monad.Reader
@@ -54,6 +58,8 @@ import Data.List.NonEmpty (NonEmpty)
 import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
 import Data.Maybe (fromMaybe)
+import Data.Set (Set)
+import qualified Data.Set as S
 import Data.Time.Clock (NominalDiffTime, nominalDay)
 import Data.Time.Clock.System (SystemTime (..))
 import Data.Word (Word16)
@@ -71,14 +77,14 @@ import Simplex.Messaging.Notifications.Client (defaultNTFClientConfig)
 import Simplex.Messaging.Notifications.Transport (NTFVersion)
 import Simplex.Messaging.Notifications.Types
 import Simplex.Messaging.Parsers (defaultJSON)
-import Simplex.Messaging.Protocol (NtfServer, ProtoServerWithAuth, ProtocolServer, ProtocolType (..), ProtocolTypeI, VersionRangeSMPC, XFTPServer, supportedSMPClientVRange)
+import Simplex.Messaging.Protocol (NtfServer, ProtoServerWithAuth (..), ProtocolServer (..), ProtocolType (..), ProtocolTypeI, VersionRangeSMPC, XFTPServer, supportedSMPClientVRange)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (SMPVersion)
+import Simplex.Messaging.Transport.Client (TransportHost)
 import Simplex.Messaging.Util (allFinally, catchAllErrors, catchAllErrors', tryAllErrors, tryAllErrors')
 import System.Mem.Weak (Weak)
 import System.Random (StdGen, newStdGen)
-import UnliftIO (SomeException)
 import UnliftIO.STM
 
 type AM' a = ReaderT Env IO a
@@ -94,29 +100,42 @@ data InitialAgentServers = InitialAgentServers
 
 data ServerCfg p = ServerCfg
   { server :: ProtoServerWithAuth p,
-    preset :: Bool,
-    tested :: Maybe Bool,
-    enabled :: Bool
+    operator :: Maybe OperatorId,
+    enabled :: Bool,
+    roles :: ServerRoles
   }
   deriving (Show)
 
-enabledServerCfg :: ProtoServerWithAuth p -> ServerCfg p
-enabledServerCfg server = ServerCfg {server, preset = False, tested = Nothing, enabled = True}
+data ServerRoles = ServerRoles
+  { storage :: Bool,
+    proxy :: Bool
+  }
+  deriving (Show)
 
-presetServerCfg :: Bool -> ProtoServerWithAuth p -> ServerCfg p
-presetServerCfg enabled server = ServerCfg {server, preset = True, tested = Nothing, enabled}
+allRoles :: ServerRoles
+allRoles = ServerRoles True True
+
+presetServerCfg :: Bool -> ServerRoles -> Maybe OperatorId -> ProtoServerWithAuth p -> ServerCfg p
+presetServerCfg enabled roles operator server =
+  ServerCfg {server, operator, enabled, roles}
 
 data UserServers p = UserServers
-  { enabledSrvs :: NonEmpty (ProtoServerWithAuth p),
-    knownSrvs :: NonEmpty (ProtocolServer p)
+  { storageSrvs :: NonEmpty (Maybe OperatorId, ProtoServerWithAuth p),
+    proxySrvs :: NonEmpty (Maybe OperatorId, ProtoServerWithAuth p),
+    knownHosts :: Set TransportHost
   }
 
+type OperatorId = Int64
+
 -- This function sets all servers as enabled in case all passed servers are disabled.
 mkUserServers :: NonEmpty (ServerCfg p) -> UserServers p
-mkUserServers srvs = UserServers {enabledSrvs, knownSrvs}
+mkUserServers srvs = UserServers {storageSrvs = filterSrvs storage, proxySrvs = filterSrvs proxy, knownHosts}
   where
-    enabledSrvs = L.map (\ServerCfg {server} -> server) $ fromMaybe srvs $ L.nonEmpty $ L.filter (\ServerCfg {enabled} -> enabled) srvs
-    knownSrvs = L.map (\ServerCfg {server = ProtoServerWithAuth srv _} -> srv) srvs
+    filterSrvs role = L.map (\ServerCfg {operator, server} -> (operator, server)) $ fromMaybe srvs $ L.nonEmpty $ L.filter (\ServerCfg {enabled, roles} -> enabled && role roles) srvs
+    knownHosts = S.unions $ L.map (\ServerCfg {server = ProtoServerWithAuth srv _} -> serverHosts srv) srvs
+
+serverHosts :: ProtocolServer p -> Set TransportHost
+serverHosts ProtocolServer {host} = S.fromList $ L.toList host
 
 data AgentConfig = AgentConfig
   { tcpPort :: Maybe ServiceName,
@@ -313,7 +332,9 @@ agentFinally = allFinally mkInternal
 {-# INLINE agentFinally #-}
 
 mkInternal :: SomeException -> AgentErrorType
-mkInternal = INTERNAL . show
+mkInternal e = case fromException e of
+  Just BlockedIndefinitelyOnSTM -> CRITICAL True "Thread blocked indefinitely in STM transaction"
+  _ -> INTERNAL $ show e
 {-# INLINE mkInternal #-}
 
 data Worker = Worker
@@ -335,6 +356,8 @@ updateRestartCount t (RestartCount minute count) = do
 
 $(pure [])
 
+$(JQ.deriveJSON defaultJSON ''ServerRoles)
+
 instance ProtocolTypeI p => ToJSON (ServerCfg p) where
   toEncoding = $(JQ.mkToEncoding defaultJSON ''ServerCfg)
   toJSON = $(JQ.mkToJSON defaultJSON ''ServerCfg)

src/Simplex/Messaging/Agent/Lock.hs

@@ -1,6 +1,7 @@
 module Simplex.Messaging.Agent.Lock
   ( Lock,
     createLock,
+    createLockIO,
     withLock,
     withLock',
     withGetLock,
@@ -24,6 +25,10 @@ createLock :: STM Lock
 createLock = newEmptyTMVar
 {-# INLINE createLock #-}
 
+createLockIO :: IO Lock
+createLockIO = newEmptyTMVarIO
+{-# INLINE createLockIO #-}
+
 withLock :: MonadUnliftIO m => Lock -> String -> ExceptT e m a -> ExceptT e m a
 withLock lock name = ExceptT . withLock' lock name . runExceptT
 {-# INLINE withLock #-}

src/Simplex/Messaging/Agent/Protocol.hs

@@ -108,6 +108,7 @@ module Simplex.Messaging.Agent.Protocol
     ConnReqUriData (..),
     CRClientData,
     ServiceScheme,
+    sameConnReqContact,
     simplexChat,
     connReqUriP',
     AgentErrorType (..),
@@ -1263,6 +1264,12 @@ instance Eq AConnectionRequestUri where
 
 deriving instance Show AConnectionRequestUri
 
+sameConnReqContact :: ConnectionRequestUri 'CMContact -> ConnectionRequestUri 'CMContact -> Bool
+sameConnReqContact (CRContactUri ConnReqUriData {crSmpQueues = qs}) (CRContactUri ConnReqUriData {crSmpQueues = qs'}) =
+  L.length qs == L.length qs' && all same (L.zip qs qs')
+  where
+    same (q, q') = sameQAddress (qAddress q) (qAddress q')
+
 data ConnReqUriData = ConnReqUriData
   { crScheme :: ServiceScheme,
     crAgentVRange :: VersionRangeSMPA,

src/Simplex/Messaging/Notifications/Server.hs

@@ -115,10 +115,12 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
 
     stopServer :: M ()
     stopServer = do
+      logInfo "Saving server state..."
       saveServer
       NtfSubscriber {smpSubscribers, smpAgent} <- asks subscriber
       liftIO $ readTVarIO smpSubscribers >>= mapM_ (\SMPSubscriber {subThreadId} -> readTVarIO subThreadId >>= mapM_ (deRefWeak >=> mapM_ killThread))
       liftIO $ closeSMPClientAgent smpAgent
+      logInfo "Server stopped"
 
     saveServer :: M ()
     saveServer = withNtfLog closeStoreLog >> saveServerLastNtfs >> saveServerStats

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

@@ -53,7 +53,9 @@ ntfServerCLI cfgPath logPath =
         True -> readIniFile iniFile >>= either exitError runServer
         _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
     Delete -> do
-      confirmOrExit "WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
+      confirmOrExit
+        "WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
+        "Server NOT deleted"
       deleteDirIfExists cfgPath
       deleteDirIfExists logPath
       putStrLn "Deleted configuration and log files"

src/Simplex/Messaging/Protocol.hs

@@ -399,8 +399,6 @@ data Command (p :: Party) where
   NKEY :: NtfPublicAuthKey -> RcvNtfPublicDhKey -> Command Recipient
   NDEL :: Command Recipient
   GET :: Command Recipient
-  -- ACK v1 has to be supported for encoding/decoding
-  -- ACK :: Command Recipient
   ACK :: MsgId -> Command Recipient
   OFF :: Command Recipient
   DEL :: Command Recipient
@@ -540,11 +538,13 @@ messageId :: Message -> MsgId
 messageId = \case
   Message {msgId} -> msgId
   MessageQuota {msgId} -> msgId
+{-# INLINE messageId #-}
 
 messageTs :: Message -> SystemTime
 messageTs = \case
   Message {msgTs} -> msgTs
   MessageQuota {msgTs} -> msgTs
+{-# INLINE messageTs #-}
 
 newtype EncRcvMsgBody = EncRcvMsgBody ByteString
   deriving (Eq, Show)

src/Simplex/Messaging/Server/CLI.hs

@@ -44,11 +44,11 @@ import Text.Read (readMaybe)
 exitError :: String -> IO a
 exitError msg = putStrLn msg >> exitFailure
 
-confirmOrExit :: String -> IO ()
-confirmOrExit s =
+confirmOrExit :: String -> String -> IO ()
+confirmOrExit s no =
   withPrompt (s <> "\nContinue (Y/n): ") $ do
     ok <- getLine
-    when (ok /= "Y") $ putStrLn "Server NOT deleted" >> exitFailure
+    when (ok /= "Y") $ putStrLn no >> exitFailure
 
 data SignAlgorithm = ED448 | ED25519
   deriving (Read, Show)

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

@@ -1,10 +1,14 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE KindSignatures #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StrictData #-}
+{-# LANGUAGE TypeFamilies #-}
 
 module Simplex.Messaging.Server.Env.STM where
 
@@ -35,9 +39,11 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.Information
+import Simplex.Messaging.Server.MsgStore.Journal
 import Simplex.Messaging.Server.MsgStore.STM
+import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.NtfStore
-import Simplex.Messaging.Server.QueueStore (QueueRec (..))
+import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.QueueStore.STM
 import Simplex.Messaging.Server.Stats
 import Simplex.Messaging.Server.StoreLog
@@ -55,7 +61,10 @@ data ServerConfig = ServerConfig
   { transports :: [(ServiceName, ATransport, AddHTTP)],
     smpHandshakeTimeout :: Int,
     tbqSize :: Natural,
+    msgStoreType :: AMSType,
     msgQueueQuota :: Int,
+    maxJournalMsgCount :: Int,
+    maxJournalStateLines :: Int,
     queueIdBytes :: Int,
     msgIdBytes :: Int,
     storeLogFile :: Maybe FilePath,
@@ -70,6 +79,9 @@ data ServerConfig = ServerConfig
     controlPortAdminAuth :: Maybe BasicAuth,
     -- | time after which the messages can be removed from the queues and check interval, seconds
     messageExpiration :: Maybe ExpirationConfig,
+    expireMessagesOnStart :: Bool,
+    -- | interval of inactivity after which journal queue is closed
+    idleQueueInterval :: Int64,
     -- | notification expiration interval (seconds)
     notificationExpiration :: ExpirationConfig,
     -- | time after which the socket with inactive client can be disconnected (without any messages or commands, incl. PING),
@@ -111,9 +123,12 @@ defaultMessageExpiration :: ExpirationConfig
 defaultMessageExpiration =
   ExpirationConfig
     { ttl = defMsgExpirationDays * 86400, -- seconds
-      checkInterval = 43200 -- seconds, 12 hours
+      checkInterval = 14400 -- seconds, 4 hours
     }
 
+defaultIdleQueueInterval :: Int64
+defaultIdleQueueInterval = 28800 -- seconds, 8 hours
+
 defNtfExpirationHours :: Int64
 defNtfExpirationHours = 24
 
@@ -134,35 +149,62 @@ defaultInactiveClientExpiration =
 defaultProxyClientConcurrency :: Int
 defaultProxyClientConcurrency = 32
 
+journalMsgStoreDepth :: Int
+journalMsgStoreDepth = 5
+
+defaultMaxJournalStateLines :: Int
+defaultMaxJournalStateLines = 16
+
+defaultMaxJournalMsgCount :: Int
+defaultMaxJournalMsgCount = 256
+
+defaultMsgQueueQuota :: Int
+defaultMsgQueueQuota = 128
+
+defaultStateTailSize :: Int
+defaultStateTailSize = 512
+
 data Env = Env
   { config :: ServerConfig,
     serverActive :: TVar Bool,
     serverInfo :: ServerInformation,
     server :: Server,
     serverIdentity :: KeyHash,
-    queueStore :: QueueStore,
-    msgStore :: STMMsgStore,
+    msgStore :: AMsgStore,
     ntfStore :: NtfStore,
     random :: TVar ChaChaDRG,
-    storeLog :: Maybe (StoreLog 'WriteMode),
     tlsServerCreds :: T.Credential,
     httpServerCreds :: Maybe T.Credential,
     serverStats :: ServerStats,
     sockets :: TVar [(ServiceName, SocketState)],
     clientSeq :: TVar ClientId,
-    clients :: TVar (IntMap (Maybe Client)),
+    clients :: TVar (IntMap (Maybe AClient)),
     proxyAgent :: ProxyAgent -- senders served on this proxy
   }
 
+type family MsgStore s where
+  MsgStore 'MSMemory = STMMsgStore
+  MsgStore 'MSJournal = JournalMsgStore
+
+data AMsgStore = forall s. (STMQueueStore (MsgStore s), MsgStoreClass (MsgStore s)) => AMS (SMSType s) (MsgStore s)
+
+data AStoreQueue = forall s. MsgStoreClass (MsgStore s) => ASQ (SMSType s) (StoreQueue (MsgStore s))
+
+data AMsgStoreCfg = forall s. MsgStoreClass (MsgStore s) => AMSC (SMSType s) (MsgStoreConfig (MsgStore s))
+
+msgPersistence :: AMsgStoreCfg -> Bool
+msgPersistence (AMSC SMSMemory (STMStoreConfig {storePath})) = isJust storePath
+msgPersistence (AMSC SMSJournal _) = True
+
 type Subscribed = Bool
 
 data Server = Server
   { subscribedQ :: TQueue (RecipientId, ClientId, Subscribed),
-    subscribers :: TMap RecipientId (TVar Client),
+    subscribers :: TMap RecipientId (TVar AClient),
     ntfSubscribedQ :: TQueue (NotifierId, ClientId, Subscribed),
-    notifiers :: TMap NotifierId (TVar Client),
-    subClients :: TVar (IntMap Client), -- clients with SMP subscriptions
-    ntfSubClients :: TVar (IntMap Client), -- clients with Ntf subscriptions
+    notifiers :: TMap NotifierId (TVar AClient),
+    subClients :: TVar (IntMap AClient), -- clients with SMP subscriptions
+    ntfSubClients :: TVar (IntMap AClient), -- clients with Ntf subscriptions
     pendingSubEvents :: TVar (IntMap (NonEmpty (RecipientId, Subscribed))),
     pendingNtfSubEvents :: TVar (IntMap (NonEmpty (NotifierId, Subscribed))),
     savingLock :: Lock
@@ -174,11 +216,16 @@ newtype ProxyAgent = ProxyAgent
 
 type ClientId = Int
 
-data Client = Client
+data AClient = forall s. MsgStoreClass (MsgStore s) => AClient (SMSType s) (Client (MsgStore s))
+
+clientId' :: AClient -> ClientId
+clientId' (AClient _ Client {clientId}) = clientId
+
+data Client s = Client
   { clientId :: ClientId,
     subscriptions :: TMap RecipientId Sub,
     ntfSubscriptions :: TMap NotifierId (),
-    rcvQ :: TBQueue (NonEmpty (Maybe QueueRec, Transmission Cmd)),
+    rcvQ :: TBQueue (NonEmpty (Maybe (StoreQueue s, QueueRec), Transmission Cmd)),
     sndQ :: TBQueue (NonEmpty (Transmission BrokerMsg)),
     msgQ :: TBQueue (NonEmpty (Transmission BrokerMsg)),
     procThreads :: TVar Int,
@@ -211,11 +258,11 @@ newServer = do
   ntfSubClients <- newTVarIO IM.empty
   pendingSubEvents <- newTVarIO IM.empty
   pendingNtfSubEvents <- newTVarIO IM.empty
-  savingLock <- atomically createLock
+  savingLock <- createLockIO
   return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers, subClients, ntfSubClients, pendingSubEvents, pendingNtfSubEvents, savingLock}
 
-newClient :: ClientId -> Natural -> VersionSMP -> ByteString -> SystemTime -> IO Client
-newClient clientId qSize thVersion sessionId createdAt = do
+newClient :: SMSType s -> ClientId -> Natural -> VersionSMP -> ByteString -> SystemTime -> IO (Client (MsgStore s))
+newClient _msType clientId qSize thVersion sessionId createdAt = do
   subscriptions <- TM.emptyIO
   ntfSubscriptions <- TM.emptyIO
   rcvQ <- newTBQueueIO qSize
@@ -241,17 +288,22 @@ newProhibitedSub = do
   return Sub {subThread = ProhibitSub, delivered}
 
 newEnv :: ServerConfig -> IO Env
-newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, smpAgentCfg, information, messageExpiration} = do
+newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, msgStoreType, storeMsgsFile, smpAgentCfg, information, messageExpiration, idleQueueInterval, msgQueueQuota, maxJournalMsgCount, maxJournalStateLines} = do
   serverActive <- newTVarIO True
   server <- newServer
-  queueStore <- newQueueStore
-  msgStore <- newMsgStore
+  msgStore@(AMS _ store) <- case msgStoreType of
+    AMSType SMSMemory -> AMS SMSMemory <$> newMsgStore STMStoreConfig {storePath = storeMsgsFile, quota = msgQueueQuota}
+    AMSType SMSJournal -> case storeMsgsFile of
+      Just storePath -> 
+        let cfg = JournalStoreConfig {storePath, quota = msgQueueQuota, pathParts = journalMsgStoreDepth, maxMsgCount = maxJournalMsgCount, maxStateLines = maxJournalStateLines, stateTailSize = defaultStateTailSize, idleInterval = idleQueueInterval}
+         in AMS SMSJournal <$> newMsgStore cfg
+      Nothing -> putStrLn "Error: journal msg store require path in [STORE_LOG], restore_messages" >> exitFailure
   ntfStore <- NtfStore <$> TM.emptyIO
   random <- C.newRandom
-  storeLog <-
-    forM storeLogFile $ \f -> do
-      logInfo $ "restoring queues from file " <> T.pack f
-      readWriteQueueStore f queueStore
+  forM_ storeLogFile $ \f -> do
+    logInfo $ "restoring queues from file " <> T.pack f
+    sl <- readWriteQueueStore f store
+    setStoreLog store sl
   tlsServerCreds <- getCredentials "SMP" smpCredentials
   httpServerCreds <- mapM (getCredentials "HTTPS") httpCredentials
   mapM_ checkHTTPSCredentials httpServerCreds
@@ -262,7 +314,7 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, smpAg
   clientSeq <- newTVarIO 0
   clients <- newTVarIO mempty
   proxyAgent <- newSMPProxyAgent smpAgentCfg random
-  pure Env {serverActive, config, serverInfo, server, serverIdentity, queueStore, msgStore, ntfStore, random, storeLog, tlsServerCreds, httpServerCreds, serverStats, sockets, clientSeq, clients, proxyAgent}
+  pure Env {serverActive, config, serverInfo, server, serverIdentity, msgStore, ntfStore, random, tlsServerCreds, httpServerCreds, serverStats, sockets, clientSeq, clients, proxyAgent}
   where
     getCredentials protocol creds = do
       files <- missingCreds
@@ -299,10 +351,13 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, smpAg
       where
         persistence
           | isNothing storeLogFile = SPMMemoryOnly
-          | isJust (storeMsgsFile config) = SPMMessages
+          | isJust storeMsgsFile = SPMMessages
           | otherwise = SPMQueues
 
 newSMPProxyAgent :: SMPClientAgentConfig -> TVar ChaChaDRG -> IO ProxyAgent
 newSMPProxyAgent smpAgentCfg random = do
   smpAgent <- newSMPClientAgent smpAgentCfg random
   pure ProxyAgent {smpAgent}
+
+readWriteQueueStore :: STMQueueStore s => FilePath -> s -> IO (StoreLog 'WriteMode)
+readWriteQueueStore = readWriteStoreLog readQueueStore writeQueueStore

src/Simplex/Messaging/Server/Main.hs

@@ -1,6 +1,6 @@
 {-# LANGUAGE ApplicativeDo #-}
-{-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedLists #-}
@@ -20,6 +20,7 @@ import Control.Monad
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Char (isAlpha, isAscii, toUpper)
+import Data.Either (fromRight)
 import Data.Functor (($>))
 import Data.Ini (Ini, lookupValue, readIniFile)
 import Data.List (find, isPrefixOf)
@@ -27,7 +28,7 @@ import qualified Data.List.NonEmpty as L
 import Data.Maybe (fromMaybe, isJust, isNothing)
 import Data.Text (Text)
 import qualified Data.Text as T
-import Data.Text.Encoding (encodeUtf8, decodeLatin1)
+import Data.Text.Encoding (decodeLatin1, encodeUtf8)
 import qualified Data.Text.IO as T
 import Network.Socket (HostName)
 import Options.Applicative
@@ -38,17 +39,20 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Protocol (BasicAuth (..), ProtoServerWithAuth (ProtoServerWithAuth), pattern SMPServer)
-import Simplex.Messaging.Server (AttachHTTP, runSMPServer)
+import Simplex.Messaging.Server (AttachHTTP, exportMessages, importMessages, printMessageStats, runSMPServer)
 import Simplex.Messaging.Server.CLI
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.Information
+import Simplex.Messaging.Server.MsgStore.Journal (JournalStoreConfig (..))
+import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..), newMsgStore)
+import Simplex.Messaging.Server.QueueStore.STM (readQueueStore)
 import Simplex.Messaging.Transport (batchCmdsSMPVersion, sendingProxySMPVersion, simplexMQVersion, supportedServerSMPRelayVRange)
 import Simplex.Messaging.Transport.Client (SocksProxy, TransportHost (..), defaultSocksProxy)
 import Simplex.Messaging.Transport.Server (ServerCredentials (..), TransportServerConfig (..), defaultTransportServerConfig)
-import Simplex.Messaging.Util (eitherToMaybe, safeDecodeUtf8, tshow)
+import Simplex.Messaging.Util (eitherToMaybe, ifM, safeDecodeUtf8, tshow)
 import Simplex.Messaging.Version (mkVersionRange)
-import System.Directory (createDirectoryIfMissing, doesFileExist)
+import System.Directory (createDirectoryIfMissing, doesDirectoryExist, doesFileExist)
 import System.Exit (exitFailure)
 import System.FilePath (combine)
 import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
@@ -70,25 +74,95 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
       doesFileExist iniFile >>= \case
         True -> exitError $ "Error: server is already initialized (" <> iniFile <> " exists).\nRun `" <> executableName <> " start`."
         _ -> initializeServer opts
-    OnlineCert certOpts ->
-      doesFileExist iniFile >>= \case
-        True -> genOnline cfgPath certOpts
-        _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
-    Start ->
-      doesFileExist iniFile >>= \case
-        True -> readIniFile iniFile >>= either exitError runServer
-        _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
+    OnlineCert certOpts -> withIniFile $ \_ -> genOnline cfgPath certOpts
+    Start -> withIniFile runServer
     Delete -> do
-      confirmOrExit "WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
+      confirmOrExit
+        "WARNING: deleting the server will make all queues inaccessible, because the server identity (certificate fingerprint) will change.\nTHIS CANNOT BE UNDONE!"
+        "Server NOT deleted"
       deleteDirIfExists cfgPath
       deleteDirIfExists logPath
       putStrLn "Deleted configuration and log files"
+    Journal cmd -> withIniFile $ \ini -> do
+      msgsDirExists <- doesDirectoryExist storeMsgsJournalDir
+      msgsFileExists <- doesFileExist storeMsgsFilePath
+      let enableStoreLog = settingIsOn "STORE_LOG" "enable" ini
+      storeLogFile <- case enableStoreLog $> storeLogFilePath of
+        Just storeLogFile -> do
+          ifM
+            (doesFileExist storeLogFile)
+            (pure storeLogFile)
+            (putStrLn ("Store log file " <> storeLogFile <> " not found") >> exitFailure)
+        Nothing -> putStrLn "Store log disabled, see `[STORE_LOG] enable`" >> exitFailure
+      case cmd of
+        JCImport
+          | msgsFileExists && msgsDirExists -> exitConfigureMsgStorage
+          | msgsDirExists -> do
+              putStrLn $ storeMsgsJournalDir <> " directory already exists."
+              exitFailure
+          | not msgsFileExists -> do
+              putStrLn $ storeMsgsFilePath <> " file does not exists."
+              exitFailure
+          | otherwise -> do
+              confirmOrExit
+                ("WARNING: message log file " <> storeMsgsFilePath <> " will be imported to journal directory " <> storeMsgsJournalDir)
+                "Messages not imported"
+              ms <- newJournalMsgStore
+              readQueueStore storeLogFile ms
+              msgStats <- importMessages True ms storeMsgsFilePath Nothing -- no expiration
+              putStrLn "Import completed"
+              printMessageStats "Messages" msgStats
+              putStrLn $ case readMsgStoreType ini of
+                Right (AMSType SMSMemory) -> "store_messages set to `memory`, update it to `journal` in INI file"
+                Right (AMSType SMSJournal) -> "store_messages set to `journal`"
+                Left e -> e <> ", update it to `journal` in INI file"
+        JCExport
+          | msgsFileExists && msgsDirExists -> exitConfigureMsgStorage
+          | msgsFileExists -> do
+              putStrLn $ storeMsgsFilePath <> " file already exists."
+              exitFailure
+          | otherwise -> do
+              confirmOrExit
+                ("WARNING: journal directory " <> storeMsgsJournalDir <> " will be exported to message log file " <> storeMsgsFilePath)
+                "Journal not exported"
+              ms <- newJournalMsgStore
+              readQueueStore storeLogFile ms
+              exportMessages True ms storeMsgsFilePath False
+              putStrLn "Export completed"
+              putStrLn $ case readMsgStoreType ini of
+                Right (AMSType SMSMemory) -> "store_messages set to `memory`"
+                Right (AMSType SMSJournal) -> "store_messages set to `journal`, update it to `memory` in INI file"
+                Left e -> e <> ", update it to `memory` in INI file"
+        JCDelete
+          | not msgsDirExists -> do
+              putStrLn $ storeMsgsJournalDir <> " directory does not exists."
+              exitFailure
+          | otherwise -> do
+              confirmOrExit
+                ("WARNING: journal directory " <> storeMsgsJournalDir <> " will be permanently deleted.\nTHIS CANNOT BE UNDONE!")
+                "Messages NOT deleted"
+              deleteDirIfExists storeMsgsJournalDir
+              putStrLn $ "Deleted all messages in journal " <> storeMsgsJournalDir
   where
+    withIniFile a =
+      doesFileExist iniFile >>= \case
+        True -> readIniFile iniFile >>= either exitError a
+        _ -> exitError $ "Error: server is not initialized (" <> iniFile <> " does not exist).\nRun `" <> executableName <> " init`."
+    newJournalMsgStore = newMsgStore JournalStoreConfig {storePath = storeMsgsJournalDir, pathParts = journalMsgStoreDepth, quota = defaultMsgQueueQuota, maxMsgCount = defaultMaxJournalMsgCount, maxStateLines = defaultMaxJournalStateLines, stateTailSize = defaultStateTailSize, idleInterval = checkInterval defaultMessageExpiration}
     iniFile = combine cfgPath "smp-server.ini"
     serverVersion = "SMP server v" <> simplexMQVersion
     defaultServerPorts = "5223,443"
     executableName = "smp-server"
     storeLogFilePath = combine logPath "smp-server-store.log"
+    storeMsgsFilePath = combine logPath "smp-server-messages.log"
+    storeMsgsJournalDir = combine logPath "messages"
+    storeNtfsFilePath = combine logPath "smp-server-ntfs.log"
+    readMsgStoreType :: Ini -> Either String AMSType
+    readMsgStoreType = textToMsgStoreType . fromRight "memory" . lookupValue "STORE_LOG" "store_messages"
+    textToMsgStoreType = \case
+      "memory" -> Right $ AMSType SMSMemory
+      "journal" -> Right $ AMSType SMSJournal
+      s -> Left $ "invalid store_messages: " <> T.unpack s
     httpsCertFile = combine cfgPath "web.crt"
     httpsKeyFile = combine cfgPath "web.key"
     defaultStaticPath = combine logPath "www"
@@ -128,7 +202,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                 _ -> putStrLn "Invalid password. Only latin letters, digits and symbols other than '@' and ':' are allowed" >> serverPassword
         checkInitOptions InitOptions {sourceCode, serverInfo, operatorCountry, hostingCountry} = do
           let err_
-                | isNothing sourceCode && hasServerInfo serverInfo = 
+                | isNothing sourceCode && hasServerInfo serverInfo =
                     Just "Error: passing any server information requires passing --source-code"
                 | isNothing (operator serverInfo) && isJust operatorCountry =
                     Just "Error: passing --operator-country requires passing --operator"
@@ -168,10 +242,14 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                    \# and restoring it when the server is started.\n\
                    \# Log is compacted on start (deleted objects are removed).\n"
                 <> ("enable: " <> onOff enableStoreLog <> "\n\n")
-                <> "# Undelivered messages are optionally saved and restored when the server restarts,\n\
-                   \# they are preserved in the .bak file until the next restart.\n"
-                <> ("restore_messages: " <> onOff enableStoreLog <> "\n")
+                <> "# Message storage mode: `memory` or `journal`.\n\
+                   \store_messages: memory\n\n\
+                   \# When store_messages is `memory`, undelivered messages are optionally saved and restored\n\
+                   \# when the server restarts, they are preserved in the .bak file until the next restart.\n"
+                <> ("restore_messages: " <> onOff enableStoreLog <> "\n\n")
+                <> "# Messages and notifications expiration periods.\n"
                 <> ("expire_messages_days: " <> tshow defMsgExpirationDays <> "\n")
+                <> "expire_messages_on_start: on\n"
                 <> ("expire_ntfs_hours: " <> tshow defNtfExpirationHours <> "\n\n")
                 <> "# Log daily server statistics to CSV file\n"
                 <> ("log_stats: " <> onOff logStats <> "\n\n")
@@ -248,12 +326,13 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
       fp <- checkSavedFingerprint cfgPath defaultX509Config
       let host = either (const "<hostnames>") T.unpack $ lookupValue "TRANSPORT" "host" ini
           port = T.unpack $ strictIni "TRANSPORT" "port" ini
-          cfg@ServerConfig {information, storeLogFile, newQueueBasicAuth, messageExpiration, inactiveClientExpiration} = serverConfig
+          cfg@ServerConfig {information, storeLogFile, msgStoreType, newQueueBasicAuth, messageExpiration, inactiveClientExpiration} = serverConfig
           sourceCode' = (\ServerPublicInfo {sourceCode} -> sourceCode) <$> information
           srv = ProtoServerWithAuth (SMPServer [THDomainName host] (if port == "5223" then "" else port) (C.KeyHash fp)) newQueueBasicAuth
       printServiceInfo serverVersion srv
       printSourceCode sourceCode'
       printServerConfig transports storeLogFile
+      checkMsgStoreMode msgStoreType
       putStrLn $ case messageExpiration of
         Just ExpirationConfig {ttl} -> "expiring messages after " <> showTTL ttl
         _ -> "not expiring messages"
@@ -302,12 +381,16 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
           _ -> enableStoreLog $> path
         transports = iniTransports ini
         sharedHTTP = any (\(_, _, addHTTP) -> addHTTP) transports
+        iniMsgStoreType = either error id $! readMsgStoreType ini
         serverConfig =
           ServerConfig
             { transports,
               smpHandshakeTimeout = 120000000,
               tbqSize = 128,
-              msgQueueQuota = 128,
+              msgStoreType = iniMsgStoreType,
+              msgQueueQuota = defaultMsgQueueQuota,
+              maxJournalMsgCount = defaultMaxJournalMsgCount,
+              maxJournalStateLines = defaultMaxJournalStateLines,
               queueIdBytes = 24,
               msgIdBytes = 24, -- must be at least 24 bytes, it is used as 192-bit nonce for XSalsa20
               smpCredentials =
@@ -318,8 +401,10 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                   },
               httpCredentials = (\WebHttpsParams {key, cert} -> ServerCredentials {caCertificateFile = Nothing, privateKeyFile = key, certificateFile = cert}) <$> webHttpsParams',
               storeLogFile = enableStoreLog $> storeLogFilePath,
-              storeMsgsFile = restoreMessagesFile $ combine logPath "smp-server-messages.log",
-              storeNtfsFile = restoreMessagesFile $ combine logPath "smp-server-ntfs.log",
+              storeMsgsFile = case iniMsgStoreType of
+                AMSType SMSMemory -> restoreMessagesFile storeMsgsFilePath
+                AMSType SMSJournal -> Just storeMsgsJournalDir,
+              storeNtfsFile = restoreMessagesFile storeNtfsFilePath,
               -- allow creating new queues by default
               allowNewQueues = fromMaybe True $ iniOnOff "AUTH" "new_queues" ini,
               newQueueBasicAuth = either error id <$!> strDecodeIni "AUTH" "create_password" ini,
@@ -330,6 +415,8 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                   defaultMessageExpiration
                     { ttl = 86400 * readIniDefault defMsgExpirationDays "STORE_LOG" "expire_messages_days" ini
                     },
+              expireMessagesOnStart = fromMaybe True $ iniOnOff "STORE_LOG" "expire_messages_on_start" ini,
+              idleQueueInterval = defaultIdleQueueInterval,
               notificationExpiration =
                 defaultNtfExpiration
                   { ttl = 3600 * readIniDefault defNtfExpirationHours "STORE_LOG" "expire_ntfs_hours" ini
@@ -393,6 +480,31 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
             pure WebHttpsParams {port, cert, key}
         webStaticPath' = eitherToMaybe $ T.unpack <$> lookupValue "WEB" "static_path" ini
 
+    checkMsgStoreMode :: AMSType -> IO ()
+    checkMsgStoreMode mode = do
+      msgsDirExists <- doesDirectoryExist storeMsgsJournalDir
+      msgsFileExists <- doesFileExist storeMsgsFilePath
+      case mode of
+        _ | msgsFileExists && msgsDirExists -> exitConfigureMsgStorage
+        AMSType SMSJournal
+          | msgsFileExists -> do
+              putStrLn $ "Error: store_messages is `journal` with " <> storeMsgsFilePath <> " file present."
+              putStrLn "Set store_messages to `memory` or use `smp-server journal export` to migrate."
+              exitFailure
+          | not msgsDirExists ->
+              putStrLn $ "store_messages is `journal`, " <> storeMsgsJournalDir <> " directory will be created."
+        AMSType SMSMemory
+          | msgsDirExists -> do
+              putStrLn $ "Error: store_messages is `memory` with " <> storeMsgsJournalDir <> " directory present."
+              putStrLn "Set store_messages to `journal` or use `smp-server journal import` to migrate."
+              exitFailure
+        _ -> pure ()
+
+    exitConfigureMsgStorage = do
+      putStrLn $ "Error: both " <> storeMsgsFilePath <> " file and " <> storeMsgsJournalDir <> " directory are present."
+      putStrLn "Configure memory storage."
+      exitFailure
+
 data EmbeddedWebParams = EmbeddedWebParams
   { webStaticPath :: FilePath,
     webHttpPort :: Maybe Int,
@@ -458,12 +570,13 @@ informationIniContent InitOptions {sourceCode, serverInfo} =
        \# Hosting type can be `virtual`, `dedicated`, `colocation`, `owned`\n"
     <> ("hosting_type: " <> maybe "virtual" (decodeLatin1 . strEncode) hostingType <> "\n\n")
   where
-    ServerPublicInfo {operator, website, hosting, hostingType, serverCountry}= serverInfo
+    ServerPublicInfo {operator, website, hosting, hostingType, serverCountry} = serverInfo
     countryStr optName country = optDisabled country <> optName <> "_country: " <> fromMaybe "ISO-3166 2-letter code" country <> "\n"
     enitiyStrs optName entity =
       optDisabled entity
-        <> optName <> ": "
-        <> maybe ("entity (organization or person name)") name entity
+        <> optName
+        <> ": "
+        <> maybe "entity (organization or person name)" name entity
         <> "\n"
         <> countryStr optName (country =<< entity)
 
@@ -519,6 +632,9 @@ data CliCommand
   | OnlineCert CertOptions
   | Start
   | Delete
+  | Journal JournalCmd
+
+data JournalCmd = JCImport | JCExport | JCDelete
 
 data InitOptions = InitOptions
   { enableStoreLog :: Bool,
@@ -550,6 +666,7 @@ cliCommandP cfgPath logPath iniFile =
         <> command "cert" (info (OnlineCert <$> certOptionsP) (progDesc $ "Generate new online TLS server credentials (configuration: " <> iniFile <> ")"))
         <> command "start" (info (pure Start) (progDesc $ "Start server (configuration: " <> iniFile <> ")"))
         <> command "delete" (info (pure Delete) (progDesc "Delete configuration and log files"))
+        <> command "journal" (info (Journal <$> journalCmdP) (progDesc "Import/export messages to/from journal storage"))
     )
   where
     initP :: Parser InitOptions
@@ -607,22 +724,21 @@ cliCommandP cfgPath logPath iniFile =
           <|> option strParse (long "control-port" <> help "Enable control port" <> metavar "PORT" <> value Nothing)
       socksProxy <-
         flag' (Just defaultSocksProxy) (long "socks-proxy" <> help "Outgoing SOCKS proxy on port 9050")
-          <|>
-            option
-              strParse
-              ( long "socks-proxy"
-                  <> metavar "PROXY"
-                  <> help "Outgoing SOCKS proxy to forward messages to onion-only servers"
-                  <> value Nothing
-              )
+          <|> option
+            strParse
+            ( long "socks-proxy"
+                <> metavar "PROXY"
+                <> help "Outgoing SOCKS proxy to forward messages to onion-only servers"
+                <> value Nothing
+            )
       ownDomains :: Maybe (L.NonEmpty TransportHost) <-
         option
           strParse
-            ( long "own-domains"
-                <> metavar "DOMAINS"
-                <> help "Own server domain names (comma-separated)"
-                <> value Nothing
-            )
+          ( long "own-domains"
+              <> metavar "DOMAINS"
+              <> help "Own server domain names (comma-separated)"
+              <> value Nothing
+          )
       sourceCode <-
         flag' (Just simplexmqSource) (long "source-code" <> help ("Server source code (default: " <> simplexmqSource <> ")"))
           <|> (optional . strOption) (long "source-code" <> metavar "URI" <> help "Server source code")
@@ -690,6 +806,13 @@ cliCommandP cfgPath logPath iniFile =
             disableWeb,
             scripted
           }
+    journalCmdP =
+      hsubparser
+        ( command "import" (info (pure JCImport) (progDesc "Import message log file into a new journal storage"))
+            <> command "export" (info (pure JCExport) (progDesc "Export journal storage to message log file"))
+            <> command "delete" (info (pure JCDelete) (progDesc "Delete journal storage"))
+        )
+
     parseBasicAuth :: ReadM ServerPassword
     parseBasicAuth = eitherReader $ fmap ServerPassword . strDecode . B.pack
     entityP :: String -> String -> String -> Parser (Maybe Entity, Maybe Text)

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

@@ -0,0 +1,782 @@
+{-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE DerivingStrategies #-}
+{-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE GeneralizedNewtypeDeriving #-}
+{-# LANGUAGE InstanceSigs #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE MultiWayIf #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE StandaloneDeriving #-}
+{-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE TupleSections #-}
+
+module Simplex.Messaging.Server.MsgStore.Journal
+  ( JournalMsgStore (queues, senders, notifiers, random),
+    JournalQueue,
+    JournalMsgQueue (queue, state),
+    JMQueue (queueDirectory, statePath),
+    JournalStoreConfig (..),
+    closeMsgQueue,
+    closeMsgQueueHandles,
+    -- below are exported for tests
+    MsgQueueState (..),
+    JournalState (..),
+    SJournalType (..),
+    msgQueueDirectory,
+    msgQueueStatePath,
+    readWriteQueueState,
+    newMsgQueueState,
+    newJournalId,
+    appendState,
+    queueLogFileName,
+    journalFilePath,
+    logFileExt,
+  )
+where
+
+import Control.Concurrent.STM
+import qualified Control.Exception as E
+import Control.Logger.Simple
+import Control.Monad
+import Control.Monad.Trans.Except
+import qualified Data.Attoparsec.ByteString.Char8 as A
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
+import Data.Functor (($>))
+import Data.Int (Int64)
+import Data.List (intercalate)
+import Data.Maybe (catMaybes, fromMaybe, isNothing)
+import qualified Data.Text as T
+import Data.Time.Clock (getCurrentTime)
+import Data.Time.Clock.System (SystemTime (..), getSystemTime)
+import Data.Time.Format.ISO8601 (iso8601Show)
+import GHC.IO (catchAny)
+import Simplex.Messaging.Agent.Client (getMapLock, withLockMap)
+import Simplex.Messaging.Agent.Lock
+import Simplex.Messaging.Encoding.String
+import Simplex.Messaging.Protocol
+import Simplex.Messaging.Server.MsgStore.Types
+import Simplex.Messaging.Server.QueueStore
+import Simplex.Messaging.Server.QueueStore.STM
+import Simplex.Messaging.TMap (TMap)
+import qualified Simplex.Messaging.TMap as TM
+import Simplex.Messaging.Server.StoreLog
+import Simplex.Messaging.Util (ifM, tshow, ($>>=), (<$$>))
+import System.Directory
+import System.Exit
+import System.FilePath ((</>))
+import System.IO (BufferMode (..), Handle, IOMode (..), SeekMode (..), stdout)
+import qualified System.IO as IO
+import System.Random (StdGen, genByteString, newStdGen)
+
+data JournalMsgStore = JournalMsgStore
+  { config :: JournalStoreConfig,
+    random :: TVar StdGen,
+    queueLocks :: TMap RecipientId Lock,
+    queues :: TMap RecipientId JournalQueue,
+    senders :: TMap SenderId RecipientId,
+    notifiers :: TMap NotifierId RecipientId,
+    storeLog :: TVar (Maybe (StoreLog 'WriteMode))
+  }
+
+data JournalStoreConfig = JournalStoreConfig
+  { storePath :: FilePath,
+    pathParts :: Int,
+    quota :: Int,
+    -- Max number of messages per journal file - ignored in STM store.
+    -- When this limit is reached, the file will be changed.
+    -- This number should be set bigger than queue quota.
+    maxMsgCount :: Int,
+    maxStateLines :: Int,
+    stateTailSize :: Int,
+    -- time in seconds after which the queue will be closed after message expiration
+    idleInterval :: Int64
+  }
+
+data JournalQueue = JournalQueue
+  { queueLock :: Lock,
+    -- To avoid race conditions and errors when restoring queues,
+    -- Nothing is written to TVar when queue is deleted.
+    queueRec :: TVar (Maybe QueueRec),
+    msgQueue_ :: TVar (Maybe JournalMsgQueue),
+    -- system time in seconds since epoch
+    activeAt :: TVar Int64,
+    -- Just True - empty, Just False - non-empty, Nothing - unknown
+    isEmpty :: TVar (Maybe Bool)
+  }
+
+data JMQueue = JMQueue
+  { queueDirectory :: FilePath,
+    statePath :: FilePath
+  }
+
+data JournalMsgQueue = JournalMsgQueue
+  { queue :: JMQueue,
+    state :: TVar MsgQueueState,
+    -- tipMsg contains last message and length incl. newline
+    -- Nothing - unknown, Just Nothing - empty queue.
+    -- It  prevents reading each message twice,
+    -- and reading it after it was just written.
+    tipMsg :: TVar (Maybe (Maybe (Message, Int64))),
+    handles :: TVar (Maybe MsgQueueHandles)
+  }
+
+data MsgQueueState = MsgQueueState
+  { readState :: JournalState 'JTRead,
+    writeState :: JournalState 'JTWrite,
+    canWrite :: Bool,
+    size :: Int
+  }
+  deriving (Show)
+
+data MsgQueueHandles = MsgQueueHandles
+  { stateHandle :: Handle, -- handle to queue state log file, rotates and removes old backups when server is restarted
+    readHandle :: Handle,
+    writeHandle :: Maybe Handle -- optional, used when write file is different from read file
+  }
+
+data JournalState t = JournalState
+  { journalType :: SJournalType t,
+    journalId :: ByteString,
+    msgPos :: Int,
+    msgCount :: Int,
+    bytePos :: Int64,
+    byteCount :: Int64
+  }
+  deriving (Show)
+
+data JournalType = JTRead | JTWrite
+
+data SJournalType (t :: JournalType) where
+  SJTRead :: SJournalType 'JTRead
+  SJTWrite :: SJournalType 'JTWrite
+
+class JournalTypeI t where sJournalType :: SJournalType t
+
+instance JournalTypeI 'JTRead where sJournalType = SJTRead
+
+instance JournalTypeI 'JTWrite where sJournalType = SJTWrite
+
+deriving instance Show (SJournalType t)
+
+newMsgQueueState :: ByteString -> MsgQueueState
+newMsgQueueState journalId =
+  MsgQueueState
+    { writeState = newJournalState journalId,
+      readState = newJournalState journalId,
+      canWrite = True,
+      size = 0
+    }
+
+newJournalState :: JournalTypeI t => ByteString -> JournalState t
+newJournalState journalId = JournalState sJournalType journalId 0 0 0 0
+
+journalFilePath :: FilePath -> ByteString -> FilePath
+journalFilePath dir journalId = dir </> (msgLogFileName <> "." <> B.unpack journalId <> logFileExt)
+
+instance StrEncoding MsgQueueState where
+  strEncode MsgQueueState {writeState, readState, canWrite, size} =
+    B.unwords
+      [ "write=" <> strEncode writeState,
+        "read=" <> strEncode readState,
+        "canWrite=" <> strEncode canWrite,
+        "size=" <> strEncode size
+      ]
+  strP = do
+    writeState <- "write=" *> strP
+    readState <- " read=" *> strP
+    canWrite <- " canWrite=" *> strP
+    size <- " size=" *> strP
+    pure MsgQueueState {writeState, readState, canWrite, size}
+
+instance JournalTypeI t => StrEncoding (JournalState t) where
+  strEncode JournalState {journalId, msgPos, msgCount, bytePos, byteCount} =
+    B.intercalate "," [journalId, e msgPos, e msgCount, e bytePos, e byteCount]
+    where
+      e :: StrEncoding a => a -> ByteString
+      e = strEncode
+  strP = do
+    journalId <- A.takeTill (== ',')
+    JournalState sJournalType journalId <$> i <*> i <*> i <*> i
+    where
+      i :: Integral a => A.Parser a
+      i = A.char ',' *> A.decimal
+
+queueLogFileName :: String
+queueLogFileName = "queue_state"
+
+msgLogFileName :: String
+msgLogFileName = "messages"
+
+logFileExt :: String
+logFileExt = ".log"
+
+newtype StoreIO a = StoreIO {unStoreIO :: IO a}
+  deriving newtype (Functor, Applicative, Monad)
+
+instance STMQueueStore JournalMsgStore where
+  queues' = queues
+  senders' = senders
+  notifiers' = notifiers
+  storeLog' = storeLog
+  mkQueue st qr = do
+    lock <- getMapLock (queueLocks st) $ recipientId qr
+    q <- newTVar $ Just qr
+    mq <- newTVar Nothing
+    activeAt <- newTVar 0
+    isEmpty <- newTVar Nothing
+    pure $ JournalQueue lock q mq activeAt isEmpty
+  msgQueue_' = msgQueue_
+
+instance MsgStoreClass JournalMsgStore where
+  type StoreMonad JournalMsgStore = StoreIO
+  type StoreQueue JournalMsgStore = JournalQueue
+  type MsgQueue JournalMsgStore = JournalMsgQueue
+  type MsgStoreConfig JournalMsgStore = JournalStoreConfig
+
+  newMsgStore :: JournalStoreConfig -> IO JournalMsgStore
+  newMsgStore config = do
+    random <- newTVarIO =<< newStdGen
+    queueLocks <- TM.emptyIO
+    queues <- TM.emptyIO
+    senders <- TM.emptyIO
+    notifiers <- TM.emptyIO
+    storeLog <- newTVarIO Nothing
+    pure JournalMsgStore {config, random, queueLocks, queues, senders, notifiers, storeLog}
+
+  setStoreLog :: JournalMsgStore -> StoreLog 'WriteMode -> IO ()
+  setStoreLog st sl = atomically $ writeTVar (storeLog st) (Just sl)
+
+  closeMsgStore st = do
+    readTVarIO (storeLog st) >>= mapM_ closeStoreLog
+    readTVarIO (queues st) >>= mapM_ closeMsgQueue
+
+  activeMsgQueues = queues
+  {-# INLINE activeMsgQueues #-}
+
+  -- This function is a "foldr" that opens and closes all queues, processes them as defined by action and accumulates the result.
+  -- It is used to export storage to a single file and also to expire messages and validate all queues when server is started.
+  -- TODO this function requires case-sensitive file system, because it uses queue directory as recipient ID.
+  -- It can be made to support case-insensite FS by supporting more than one queue per directory, by getting recipient ID from state file name.
+  withAllMsgQueues :: forall a. Monoid a => Bool -> JournalMsgStore -> (RecipientId -> JournalQueue -> IO a) -> IO a
+  withAllMsgQueues tty ms@JournalMsgStore {config} action = ifM (doesDirectoryExist storePath) processStore (pure mempty)
+    where
+      processStore = do
+        (!count, !res) <- foldQueues 0 processQueue (0, mempty) ("", storePath)
+        putStrLn $ progress count
+        pure res
+      JournalStoreConfig {storePath, pathParts} = config
+      processQueue :: (Int, a) -> (String, FilePath) -> IO (Int, a)
+      processQueue (!i, !r) (queueId, dir) = do
+        when (tty && i `mod` 100 == 0) $ putStr (progress i <> "\r") >> IO.hFlush stdout
+        r' <- case strDecode $ B.pack queueId of
+          Right rId ->
+            getQueue ms SRecipient rId >>= \case
+              Right q -> unStoreIO (getMsgQueue ms rId q) *> action rId q <* closeMsgQueue q
+              Left AUTH -> do
+                logWarn $ "STORE: processQueue, queue " <> T.pack queueId <> " was removed, removing " <> T.pack dir
+                removeQueueDirectory_ dir
+                pure mempty
+              Left e -> do
+                logError $ "STORE: processQueue, error getting queue " <> T.pack queueId <> ", " <> tshow e
+                exitFailure
+          Left e -> do
+            logError $ "STORE: processQueue, message queue directory " <> T.pack dir <> " is invalid, " <> tshow e
+            exitFailure
+        pure (i + 1, r <> r')
+      progress i = "Processed: " <> show i <> " queues"
+      foldQueues depth f acc (queueId, path) = do
+        let f' = if depth == pathParts - 1 then f else foldQueues (depth + 1) f
+        listDirs >>= foldM f' acc
+        where
+          listDirs = fmap catMaybes . mapM queuePath =<< listDirectory path
+          queuePath dir = do
+            let !path' = path </> dir
+                !queueId' = queueId <> dir
+            ifM
+              (doesDirectoryExist path')
+              (pure $ Just (queueId', path'))
+              (Nothing <$ putStrLn ("Error: path " <> path' <> " is not a directory, skipping"))
+
+  logQueueStates :: JournalMsgStore -> IO ()
+  logQueueStates ms = withActiveMsgQueues ms $ \_ -> unStoreIO . logQueueState
+
+  logQueueState :: JournalQueue -> StoreIO ()
+  logQueueState q =
+    StoreIO . void $
+      readTVarIO (msgQueue_ q)
+        $>>= \mq -> readTVarIO (handles mq)
+        $>>= (\hs -> (readTVarIO (state mq) >>= appendState (stateHandle hs)) $> Just ())
+
+  queueRec' = queueRec
+  {-# INLINE queueRec' #-}
+
+  getMsgQueue :: JournalMsgStore -> RecipientId -> JournalQueue -> StoreIO JournalMsgQueue
+  getMsgQueue ms@JournalMsgStore {random} rId JournalQueue {msgQueue_} =
+    StoreIO $ readTVarIO msgQueue_ >>= maybe newQ pure
+    where
+      newQ = do
+        let dir = msgQueueDirectory ms rId
+            statePath = msgQueueStatePath dir $ B.unpack (strEncode rId)
+            queue = JMQueue {queueDirectory = dir, statePath}
+        q <- ifM (doesDirectoryExist dir) (openMsgQueue ms queue) (createQ queue)
+        atomically $ writeTVar msgQueue_ $ Just q
+        pure q
+        where
+          createQ :: JMQueue -> IO JournalMsgQueue
+          createQ queue = do
+            -- folder and files are not created here,
+            -- to avoid file IO for queues without messages during subscription
+            journalId <- newJournalId random
+            mkJournalQueue queue (newMsgQueueState journalId) Nothing
+
+  getPeekMsgQueue :: JournalMsgStore -> RecipientId -> JournalQueue -> StoreIO (Maybe (JournalMsgQueue, Message))
+  getPeekMsgQueue ms rId q@JournalQueue {isEmpty} =
+    StoreIO (readTVarIO isEmpty) >>= \case
+      Just True -> pure Nothing
+      Just False -> peek
+      Nothing -> do
+        -- We only close the queue if we just learnt it's empty.
+        -- This is needed to reduce file descriptors and memory usage
+        -- after the server just started and many clients subscribe.
+        -- In case the queue became non-empty on write and then again empty on read
+        -- we won't be closing it, to avoid frequent open/close on active queues.
+        r <- peek
+        when (isNothing r) $ StoreIO $ closeMsgQueue q
+        pure r
+    where
+      peek = do
+        mq <- getMsgQueue ms rId q
+        (mq,) <$$> tryPeekMsg_ q mq
+
+  -- only runs action if queue is not empty
+  withIdleMsgQueue :: Int64 -> JournalMsgStore -> RecipientId -> JournalQueue -> (JournalMsgQueue -> StoreIO a) -> StoreIO (Maybe a, Int)
+  withIdleMsgQueue now ms@JournalMsgStore {config} rId q action =
+    StoreIO $ readTVarIO (msgQueue_ q) >>= \case
+      Nothing ->
+        E.bracket
+          (unStoreIO $ getPeekMsgQueue ms rId q)
+          (mapM_ $ \_ -> closeMsgQueue q)
+          (maybe (pure (Nothing, 0)) (unStoreIO . run))
+        where
+          run (mq, _) = do
+            r <- action mq
+            sz <- getQueueSize_ mq
+            pure (Just r, sz)
+      Just mq -> do
+        ts <- readTVarIO $ activeAt q
+        r <- if now - ts >= idleInterval config
+          then Just <$> unStoreIO (action mq) `E.finally` closeMsgQueue q
+          else pure Nothing
+        sz <- unStoreIO $ getQueueSize_ mq
+        pure (r, sz)
+
+  deleteQueue :: JournalMsgStore -> RecipientId -> JournalQueue -> IO (Either ErrorType QueueRec)
+  deleteQueue ms rId q =
+    fst <$$> deleteQueue_ ms rId q
+
+  deleteQueueSize :: JournalMsgStore -> RecipientId -> JournalQueue -> IO (Either ErrorType (QueueRec, Int))
+  deleteQueueSize ms rId q =
+    deleteQueue_ ms rId q >>= mapM (traverse getSize)
+    -- traverse operates on the second tuple element
+    where
+      getSize = maybe (pure (-1)) (fmap size . readTVarIO . state)
+
+  getQueueMessages_ :: Bool -> JournalMsgQueue -> StoreIO [Message]
+  getQueueMessages_ drainMsgs q = StoreIO (run [])
+    where
+      run msgs = readTVarIO (handles q) >>= maybe (pure []) (getMsg msgs)
+      getMsg msgs hs = chooseReadJournal q drainMsgs hs >>= maybe (pure msgs) readMsg
+        where
+          readMsg (rs, h) = do
+            (msg, len) <- hGetMsgAt h $ bytePos rs
+            updateReadPos q drainMsgs len hs
+            (msg :) <$> run msgs
+
+  writeMsg :: JournalMsgStore -> RecipientId -> JournalQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
+  writeMsg ms rId q' logState msg = isolateQueue rId q' "writeMsg" $ do
+    q <- getMsgQueue ms rId q'
+    StoreIO $ (`E.finally` updateActiveAt q') $ do
+      st@MsgQueueState {canWrite, size} <- readTVarIO (state q)
+      let empty = size == 0
+      if canWrite || empty
+        then do
+          atomically $ writeTVar (isEmpty q') (Just False)
+          let canWrt' = quota > size
+          if canWrt'
+            then writeToJournal q st canWrt' msg $> Just (msg, empty)
+            else writeToJournal q st canWrt' msgQuota $> Nothing
+        else pure Nothing
+    where
+      JournalStoreConfig {quota, maxMsgCount} = config ms
+      msgQuota = MessageQuota {msgId = messageId msg, msgTs = messageTs msg}
+      writeToJournal q st@MsgQueueState {writeState, readState = rs, size} canWrt' !msg' = do
+        let msgStr = strEncode msg' `B.snoc` '\n'
+            msgLen = fromIntegral $ B.length msgStr
+        hs <- maybe createQueueDir pure =<< readTVarIO handles
+        (ws, wh) <- case writeHandle hs of
+          Nothing | msgCount writeState >= maxMsgCount -> switchWriteJournal hs
+          wh_ -> pure (writeState, fromMaybe (readHandle hs) wh_)
+        let msgPos' = msgPos ws + 1
+            bytePos' = bytePos ws + msgLen
+            ws' = ws {msgPos = msgPos', msgCount = msgPos', bytePos = bytePos', byteCount = bytePos'}
+            rs' = if journalId ws == journalId rs then rs {msgCount = msgPos', byteCount = bytePos'} else rs
+            !st' = st {writeState = ws', readState = rs', canWrite = canWrt', size = size + 1}
+        hAppend wh (bytePos ws) msgStr
+        updateQueueState q logState hs st' $
+          when (size == 0) $ writeTVar (tipMsg q) $ Just (Just (msg, msgLen))
+        where
+          JournalMsgQueue {queue = JMQueue {queueDirectory, statePath}, handles} = q
+          createQueueDir = do
+            createDirectoryIfMissing True queueDirectory
+            sh <- openFile statePath AppendMode
+            B.hPutStr sh ""
+            rh <- createNewJournal queueDirectory $ journalId rs
+            let hs = MsgQueueHandles {stateHandle = sh, readHandle = rh, writeHandle = Nothing}
+            atomically $ writeTVar handles $ Just hs
+            pure hs
+          switchWriteJournal hs = do
+            journalId <- newJournalId $ random ms
+            wh <- createNewJournal queueDirectory journalId
+            atomically $ writeTVar handles $ Just $ hs {writeHandle = Just wh}
+            pure (newJournalState journalId, wh)
+
+  -- can ONLY be used while restoring messages, not while server running
+  setOverQuota_ :: JournalQueue -> IO ()
+  setOverQuota_ q =
+    readTVarIO (msgQueue_ q)
+      >>= mapM_ (\JournalMsgQueue {state} -> atomically $ modifyTVar' state $ \st -> st {canWrite = False})
+
+  getQueueSize_ :: JournalMsgQueue -> StoreIO Int
+  getQueueSize_ JournalMsgQueue {state} = StoreIO $ size <$> readTVarIO state
+
+  tryPeekMsg_ :: JournalQueue -> JournalMsgQueue -> StoreIO (Maybe Message)
+  tryPeekMsg_ q mq@JournalMsgQueue {tipMsg, handles} =
+    StoreIO $ (readTVarIO handles $>>= chooseReadJournal mq True $>>= peekMsg) >>= setEmpty
+    where
+      peekMsg (rs, h) = readTVarIO tipMsg >>= maybe readMsg (pure . fmap fst)
+        where
+          readMsg = do
+            ml@(msg, _) <- hGetMsgAt h $ bytePos rs
+            atomically $ writeTVar tipMsg $ Just (Just ml)
+            pure $ Just msg
+      setEmpty msg = do
+        atomically $ writeTVar (isEmpty q) (Just $ isNothing msg)
+        pure msg
+
+  tryDeleteMsg_ :: JournalQueue -> JournalMsgQueue -> Bool -> StoreIO ()
+  tryDeleteMsg_ q mq@JournalMsgQueue {tipMsg, handles} logState = StoreIO $ (`E.finally` when logState (updateActiveAt q)) $
+    void $
+      readTVarIO tipMsg -- if there is no cached tipMsg, do nothing
+        $>>= (pure . fmap snd)
+        $>>= \len -> readTVarIO handles
+        $>>= \hs -> updateReadPos mq logState len hs $> Just ()
+
+  isolateQueue :: RecipientId -> JournalQueue -> String -> StoreIO a -> ExceptT ErrorType IO a
+  isolateQueue rId JournalQueue {queueLock} op =
+    tryStore' op rId . withLock' queueLock op . unStoreIO
+
+updateActiveAt :: JournalQueue -> IO ()
+updateActiveAt q = atomically . writeTVar (activeAt q) . systemSeconds =<< getSystemTime
+
+tryStore' :: String -> RecipientId -> IO a -> ExceptT ErrorType IO a
+tryStore' op rId = tryStore op rId . fmap Right
+
+tryStore :: forall a. String -> RecipientId -> IO (Either ErrorType a) -> ExceptT ErrorType IO a
+tryStore op rId a = ExceptT $ E.mask_ $ E.try a >>= either storeErr pure
+  where
+    storeErr :: E.SomeException -> IO (Either ErrorType a)
+    storeErr e =
+      let e' = intercalate ", " [op, B.unpack $ strEncode rId, show e]
+       in logError ("STORE: " <> T.pack e') $> Left (STORE e')
+
+isolateQueueId :: String -> JournalMsgStore -> RecipientId -> IO (Either ErrorType a) -> ExceptT ErrorType IO a
+isolateQueueId op ms rId = tryStore op rId . withLockMap (queueLocks ms) rId op
+
+openMsgQueue :: JournalMsgStore -> JMQueue -> IO JournalMsgQueue
+openMsgQueue ms q@JMQueue {queueDirectory = dir, statePath} = do
+  (st, sh) <- readWriteQueueState ms statePath
+  (st', rh, wh_) <- closeOnException sh $ openJournals ms dir st sh
+  let hs = MsgQueueHandles {stateHandle = sh, readHandle = rh, writeHandle = wh_}
+  mkJournalQueue q st' (Just hs)
+
+mkJournalQueue :: JMQueue -> MsgQueueState -> Maybe MsgQueueHandles -> IO JournalMsgQueue
+mkJournalQueue queue st hs_ = do
+  state <- newTVarIO st
+  tipMsg <- newTVarIO Nothing
+  handles <- newTVarIO hs_
+  -- using the same queue lock which is currently locked,
+  -- to avoid map lookup on queue operations
+  pure JournalMsgQueue {queue, state, tipMsg, handles}
+
+chooseReadJournal :: JournalMsgQueue -> Bool -> MsgQueueHandles -> IO (Maybe (JournalState 'JTRead, Handle))
+chooseReadJournal q log' hs = do
+  st@MsgQueueState {writeState = ws, readState = rs} <- readTVarIO (state q)
+  case writeHandle hs of
+    Just wh | msgPos rs >= msgCount rs && journalId rs /= journalId ws -> do
+      -- switching to write journal
+      atomically $ writeTVar (handles q) $ Just hs {readHandle = wh, writeHandle = Nothing}
+      hClose $ readHandle hs
+      when log' $ removeJournal (queueDirectory $ queue q) rs
+      let !rs' = (newJournalState $ journalId ws) {msgCount = msgCount ws, byteCount = byteCount ws}
+          !st' = st {readState = rs'}
+      updateQueueState q log' hs st' $ pure ()
+      pure $ Just (rs', wh)
+    _ | msgPos rs >= msgCount rs && journalId rs == journalId ws -> pure Nothing
+    _ -> pure $ Just (rs, readHandle hs)
+
+updateQueueState :: JournalMsgQueue -> Bool -> MsgQueueHandles -> MsgQueueState -> STM () -> IO ()
+updateQueueState q log' hs st a = do
+  unless (validQueueState st) $ E.throwIO $ userError $ "updateQueueState invalid state: " <> show st
+  when log' $ appendState (stateHandle hs) st
+  atomically $ writeTVar (state q) st >> a
+
+appendState :: Handle -> MsgQueueState -> IO ()
+appendState h st = E.uninterruptibleMask_ $ B.hPutStr h $ strEncode st `B.snoc` '\n'
+
+updateReadPos :: JournalMsgQueue -> Bool -> Int64 -> MsgQueueHandles -> IO ()
+updateReadPos q log' len hs = do
+  st@MsgQueueState {readState = rs, size} <- readTVarIO (state q)
+  let JournalState {msgPos, bytePos} = rs
+  let msgPos' = msgPos + 1
+      rs' = rs {msgPos = msgPos', bytePos = bytePos + len}
+      st' = st {readState = rs', size = size - 1}
+  updateQueueState q log' hs st' $ writeTVar (tipMsg q) Nothing
+
+msgQueueDirectory :: JournalMsgStore -> RecipientId -> FilePath
+msgQueueDirectory JournalMsgStore {config = JournalStoreConfig {storePath, pathParts}} rId =
+  storePath </> B.unpack (B.intercalate "/" $ splitSegments pathParts $ strEncode rId)
+  where
+    splitSegments _ "" = []
+    splitSegments 1 s = [s]
+    splitSegments n s =
+      let (seg, s') = B.splitAt 2 s
+       in seg : splitSegments (n - 1) s'
+
+msgQueueStatePath :: FilePath -> String -> FilePath
+msgQueueStatePath dir queueId = dir </> (queueLogFileName <> "." <> queueId <> logFileExt)
+
+createNewJournal :: FilePath -> ByteString -> IO Handle
+createNewJournal dir journalId = do
+  let path = journalFilePath dir journalId -- TODO retry if file exists
+  h <- openFile path ReadWriteMode
+  B.hPutStr h ""
+  pure h
+
+newJournalId :: TVar StdGen -> IO ByteString
+newJournalId g = strEncode <$> atomically (stateTVar g $ genByteString 12)
+
+openJournals :: JournalMsgStore -> FilePath -> MsgQueueState -> Handle -> IO (MsgQueueState, Handle, Maybe Handle)
+openJournals ms dir st@MsgQueueState {readState = rs, writeState = ws} sh = do
+  let rjId = journalId rs
+      wjId = journalId ws
+  openJournal rs >>= \case
+    Left path
+      | rjId == wjId -> do
+          logError $ "STORE: openJournals, no read/write file - creating new file, " <> T.pack path
+          newReadJournal
+      | otherwise -> do
+          let rs' = (newJournalState wjId) {msgCount = msgCount ws, byteCount = byteCount ws}
+              st' = st {readState = rs', size = msgCount ws}
+          openJournal rs' >>= \case
+            Left path' -> do
+              logError $ "STORE: openJournals, no read and write files - creating new file, read: " <> T.pack path <> ", write: " <> T.pack path'
+              newReadJournal
+            Right rh -> do
+              logError $ "STORE: openJournals, no read file - switched to write file, " <> T.pack path
+              closeOnException rh $ fixFileSize rh $ bytePos ws
+              pure (st', rh, Nothing)
+    Right rh
+      | rjId == wjId -> do
+          closeOnException rh $ fixFileSize rh $ bytePos ws
+          pure (st, rh, Nothing)
+      | otherwise -> closeOnException rh $ do
+          fixFileSize rh $ byteCount rs
+          openJournal ws >>= \case
+            Left path -> do
+              let msgs = msgCount rs
+                  bytes = byteCount rs
+                  size' = msgs - msgPos rs
+                  ws' = (newJournalState rjId) {msgPos = msgs, msgCount = msgs, bytePos = bytes, byteCount = bytes}
+                  st' = st {writeState = ws', size = size'} -- we don't amend canWrite to trigger QCONT
+              logError $ "STORE: openJournals, no write file, " <> T.pack path
+              pure (st', rh, Nothing)
+            Right wh -> do
+              closeOnException wh $ fixFileSize wh $ bytePos ws
+              pure (st, rh, Just wh)
+  where
+    newReadJournal = do
+      rjId' <- newJournalId $ random ms
+      rh <- createNewJournal dir rjId'
+      let st' = newMsgQueueState rjId'
+      closeOnException rh $ appendState sh st'
+      pure (st', rh, Nothing)
+    openJournal :: JournalState t -> IO (Either FilePath Handle)
+    openJournal JournalState {journalId} =
+      let path = journalFilePath dir journalId
+       in ifM (doesFileExist path) (Right <$> openFile path ReadWriteMode) (pure $ Left path)
+    -- do that for all append operations
+
+fixFileSize :: Handle -> Int64 -> IO ()
+fixFileSize h pos = do
+  let pos' = fromIntegral pos
+  size <- IO.hFileSize h
+  if
+    | size > pos' -> do
+        name <- IO.hShow h
+        logWarn $ "STORE: fixFileSize, size " <> tshow size <> " > pos " <> tshow pos <> " - truncating, " <> T.pack name
+        IO.hSetFileSize h pos'
+    | size < pos' -> do
+        -- From code logic this can't happen.
+        name <- IO.hShow h
+        E.throwIO $ userError $ "fixFileSize size " <> show size <> " < pos " <> show pos <> " - aborting: " <> name
+    | otherwise -> pure ()
+
+removeJournal :: FilePath -> JournalState t -> IO ()
+removeJournal dir JournalState {journalId} = do
+  let path = journalFilePath dir journalId
+  removeFile path `catchAny` (\e -> logError $ "STORE: removeJournal, " <> T.pack path <> ", " <> tshow e)
+
+-- This function is supposed to be resilient to crashes while updating state files,
+-- and also resilient to crashes during its execution.
+readWriteQueueState :: JournalMsgStore -> FilePath -> IO (MsgQueueState, Handle)
+readWriteQueueState JournalMsgStore {random, config} statePath =
+  ifM
+    (doesFileExist tempBackup)
+    (renameFile tempBackup statePath >> readQueueState)
+    (ifM (doesFileExist statePath) readQueueState writeNewQueueState)
+  where
+    tempBackup = statePath <> ".bak"
+    readQueueState = do
+      ls <- B.lines <$> readFileTail
+      case ls of
+        [] -> writeNewQueueState
+        _ -> do
+          r@(st, _) <- useLastLine (length ls) True ls
+          unless (validQueueState st) $ E.throwIO $ userError $ "readWriteQueueState inconsistent state: " <> show st
+          pure r
+    writeNewQueueState = do
+      logWarn $ "STORE: readWriteQueueState, empty queue state - initialized, " <> T.pack statePath
+      st <- newMsgQueueState <$> newJournalId random
+      writeQueueState st
+    useLastLine len isLastLine ls = case strDecode $ last ls of
+      Right st
+        | len > maxStateLines config || not isLastLine ->
+            backupWriteQueueState st
+        | otherwise -> do
+            -- when state file has fewer than maxStateLines, we don't compact it
+            sh <- openFile statePath AppendMode
+            pure (st, sh)
+      Left e -- if the last line failed to parse
+        | isLastLine -> case init ls of -- or use the previous line
+            [] -> do
+              logWarn $ "STORE: readWriteQueueState, invalid 1-line queue state - initialized, " <> T.pack statePath
+              st <- newMsgQueueState <$> newJournalId random
+              backupWriteQueueState st
+            ls' -> do
+              logWarn $ "STORE: readWriteQueueState, invalid last line in queue state - using the previous line, " <> T.pack statePath
+              useLastLine len False ls'
+        | otherwise -> E.throwIO $ userError $ "readWriteQueueState invalid state " <> statePath <> ": " <> show e
+    backupWriteQueueState st = do
+      -- State backup is made in two steps to mitigate the crash during the backup.
+      -- Temporary backup file will be used when it is present.
+      renameFile statePath tempBackup -- 1) temp backup
+      r <- writeQueueState st -- 2) save state
+      ts <- getCurrentTime
+      renameFile tempBackup (statePath <> "." <> iso8601Show ts <> ".bak") -- 3) timed backup
+      pure r
+    writeQueueState st = do
+      sh <- openFile statePath AppendMode
+      closeOnException sh $ appendState sh st
+      pure (st, sh)
+    readFileTail =
+      IO.withFile statePath ReadMode $ \h -> do
+        size <- IO.hFileSize h
+        let sz = stateTailSize config
+            sz' = fromIntegral sz
+        if size > sz'
+          then IO.hSeek h AbsoluteSeek (size - sz') >> B.hGet h sz
+          else B.hGet h (fromIntegral size)
+
+validQueueState :: MsgQueueState -> Bool
+validQueueState MsgQueueState {readState = rs, writeState = ws, size}
+  | journalId rs == journalId ws =
+      alwaysValid
+        && msgPos rs <= msgPos ws
+        && msgCount rs == msgCount ws
+        && bytePos rs <= bytePos ws
+        && byteCount rs == byteCount ws
+        && size == msgCount rs - msgPos rs
+  | otherwise =
+      alwaysValid
+        && size == msgCount ws + msgCount rs - msgPos rs
+  where
+    alwaysValid =
+      msgPos rs <= msgCount rs
+        && bytePos rs <= byteCount rs
+        && msgPos ws == msgCount ws
+        && bytePos ws == byteCount ws
+
+deleteQueue_ :: JournalMsgStore -> RecipientId -> JournalQueue -> IO (Either ErrorType (QueueRec, Maybe JournalMsgQueue))
+deleteQueue_ ms rId q =
+  runExceptT $ isolateQueueId "deleteQueue_" ms rId $
+    deleteQueue' ms rId q >>= mapM remove
+  where
+    remove r@(_, mq_) = do
+      mapM_ closeMsgQueueHandles mq_
+      removeQueueDirectory ms rId
+      pure r
+
+closeMsgQueue :: JournalQueue -> IO ()
+closeMsgQueue JournalQueue {msgQueue_} = atomically (swapTVar msgQueue_ Nothing) >>= mapM_ closeMsgQueueHandles
+
+closeMsgQueueHandles :: JournalMsgQueue -> IO ()
+closeMsgQueueHandles q = readTVarIO (handles q) >>= mapM_ closeHandles
+  where
+    closeHandles (MsgQueueHandles sh rh wh_) = do
+      hClose sh
+      hClose rh
+      mapM_ hClose wh_
+
+removeQueueDirectory :: JournalMsgStore -> RecipientId -> IO ()
+removeQueueDirectory st = removeQueueDirectory_ . msgQueueDirectory st
+
+removeQueueDirectory_ :: FilePath -> IO ()
+removeQueueDirectory_ dir =
+  removePathForcibly dir `catchAny` \e ->
+    logError $ "STORE: removeQueueDirectory, " <> T.pack dir <> ", " <> tshow e
+
+hAppend :: Handle -> Int64 -> ByteString -> IO ()
+hAppend h pos s = do
+  fixFileSize h pos
+  IO.hSeek h SeekFromEnd 0
+  B.hPutStr h s
+
+hGetMsgAt :: Handle -> Int64 -> IO (Message, Int64)
+hGetMsgAt h pos = do
+  IO.hSeek h AbsoluteSeek $ fromIntegral pos
+  s <- B.hGetLine h
+  case strDecode s of
+    Right !msg ->
+      let !len = fromIntegral (B.length s) + 1
+       in pure (msg, len)
+    Left e -> E.throwIO $ userError $ "hGetMsgAt invalid message: " <> e
+
+openFile :: FilePath -> IOMode -> IO Handle
+openFile f mode = do
+  h <- IO.openFile f mode
+  IO.hSetBuffering h LineBuffering
+  pure h
+
+hClose :: Handle -> IO ()
+hClose h =
+  IO.hClose h `catchAny` \e -> do
+    name <- IO.hShow h
+    logError $ "STORE: hClose, " <> T.pack name <> ", " <> tshow e
+
+closeOnException :: Handle -> IO a -> IO a
+closeOnException h a = a `E.onException` hClose h

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

@@ -1,129 +1,175 @@
-{-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE ConstraintKinds #-}
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE FlexibleInstances #-}
+{-# LANGUAGE InstanceSigs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE TypeFamilies #-}
 {-# LANGUAGE TupleSections #-}
 
 module Simplex.Messaging.Server.MsgStore.STM
-  ( STMMsgStore,
-    MsgQueue (msgQueue),
-    newMsgStore,
-    getMsgQueue,
-    delMsgQueue,
-    delMsgQueueSize,
-    writeMsg,
-    tryPeekMsg,
-    tryPeekMsgIO,
-    tryDelMsg,
-    tryDelPeekMsg,
-    deleteExpiredMsgs,
-    getQueueSize,
+  ( STMMsgStore (..),
+    STMStoreConfig (..),
   )
 where
 
-import qualified Data.ByteString.Char8 as B
+import Control.Concurrent.STM
+import Control.Monad.IO.Class
+import Control.Monad.Trans.Except
 import Data.Functor (($>))
 import Data.Int (Int64)
-import Data.Time.Clock.System (SystemTime (systemSeconds))
-import Simplex.Messaging.Protocol (Message (..), MsgId, RecipientId)
+import Simplex.Messaging.Protocol
+import Simplex.Messaging.Server.MsgStore.Types
+import Simplex.Messaging.Server.QueueStore
+import Simplex.Messaging.Server.QueueStore.STM
+import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import UnliftIO.STM
+import Simplex.Messaging.Util ((<$$>), ($>>=))
+import System.IO (IOMode (..))
 
-data MsgQueue = MsgQueue
+data STMMsgStore = STMMsgStore
+  { storeConfig :: STMStoreConfig,
+    queues :: TMap RecipientId STMQueue,
+    senders :: TMap SenderId RecipientId,
+    notifiers :: TMap NotifierId RecipientId,
+    storeLog :: TVar (Maybe (StoreLog 'WriteMode))
+  }
+
+data STMQueue = STMQueue
+  { -- To avoid race conditions and errors when restoring queues,
+    -- Nothing is written to TVar when queue is deleted.
+    queueRec :: TVar (Maybe QueueRec),
+    msgQueue_ :: TVar (Maybe STMMsgQueue)
+  }
+
+data STMMsgQueue = STMMsgQueue
   { msgQueue :: TQueue Message,
-    quota :: Int,
     canWrite :: TVar Bool,
     size :: TVar Int
   }
 
-type STMMsgStore = TMap RecipientId MsgQueue
+data STMStoreConfig = STMStoreConfig
+  { storePath :: Maybe FilePath,
+    quota :: Int
+  }
 
-newMsgStore :: IO STMMsgStore
-newMsgStore = TM.emptyIO
+instance STMQueueStore STMMsgStore where
+  queues' = queues
+  senders' = senders
+  notifiers' = notifiers
+  storeLog' = storeLog
+  mkQueue _ qr = STMQueue <$> newTVar (Just qr) <*> newTVar Nothing
+  msgQueue_' = msgQueue_
 
--- The reason for double lookup is that majority of messaging queues exist,
--- because multiple messages are sent to the same queue,
--- so the first lookup without STM transaction will return the queue faster.
--- In case the queue does not exist, it needs to be looked-up again inside transaction.
-getMsgQueue :: STMMsgStore -> RecipientId -> Int -> IO MsgQueue
-getMsgQueue st rId quota = TM.lookupIO rId st >>= maybe (atomically maybeNewQ) pure
-  where
-    maybeNewQ = TM.lookup rId st >>= maybe newQ pure
-    newQ = do
-      msgQueue <- newTQueue
-      canWrite <- newTVar True
-      size <- newTVar 0
-      let q = MsgQueue {msgQueue, quota, canWrite, size}
-      TM.insert rId q st
-      pure q
+instance MsgStoreClass STMMsgStore where
+  type StoreMonad STMMsgStore = STM
+  type StoreQueue STMMsgStore = STMQueue
+  type MsgQueue STMMsgStore = STMMsgQueue
+  type MsgStoreConfig STMMsgStore = STMStoreConfig
 
-delMsgQueue :: STMMsgStore -> RecipientId -> IO ()
-delMsgQueue st rId = atomically $ TM.delete rId st
+  newMsgStore :: STMStoreConfig -> IO STMMsgStore
+  newMsgStore storeConfig = do
+    queues <- TM.emptyIO
+    senders <- TM.emptyIO
+    notifiers <- TM.emptyIO
+    storeLog <- newTVarIO Nothing
+    pure STMMsgStore {storeConfig, queues, senders, notifiers, storeLog}
 
-delMsgQueueSize :: STMMsgStore -> RecipientId -> IO Int
-delMsgQueueSize st rId = atomically (TM.lookupDelete rId st) >>= maybe (pure 0) (\MsgQueue {size} -> readTVarIO size)
+  setStoreLog :: STMMsgStore -> StoreLog 'WriteMode -> IO ()
+  setStoreLog st sl = atomically $ writeTVar (storeLog st) (Just sl)
 
-writeMsg :: MsgQueue -> Message -> IO (Maybe (Message, Bool))
-writeMsg MsgQueue {msgQueue = q, quota, canWrite, size} !msg = atomically $ do
-  canWrt <- readTVar canWrite
-  empty <- isEmptyTQueue q
-  if canWrt || empty
-    then do
-      canWrt' <- (quota >) <$> readTVar size
-      writeTVar canWrite $! canWrt'
-      modifyTVar' size (+ 1)
-      if canWrt'
-        then writeTQueue q msg $> Just (msg, empty)
-        else (writeTQueue q $! msgQuota) $> Nothing
-    else pure Nothing
-  where
-    msgQuota = MessageQuota {msgId = msgId msg, msgTs = msgTs msg}
+  closeMsgStore st = readTVarIO (storeLog st) >>= mapM_ closeStoreLog
 
-tryPeekMsgIO :: MsgQueue -> IO (Maybe Message)
-tryPeekMsgIO = atomically . tryPeekTQueue . msgQueue
-{-# INLINE tryPeekMsgIO #-}
+  activeMsgQueues = queues
+  {-# INLINE activeMsgQueues #-}
 
--- TODO remove once deliverToSub is split
-tryPeekMsg :: MsgQueue -> STM (Maybe Message)
-tryPeekMsg = tryPeekTQueue . msgQueue
-{-# INLINE tryPeekMsg #-}
+  withAllMsgQueues _ = withActiveMsgQueues
+  {-# INLINE withAllMsgQueues #-}
 
-tryDelMsg :: MsgQueue -> MsgId -> IO (Maybe Message)
-tryDelMsg mq msgId' = atomically $
-  tryPeekMsg mq >>= \case
-    msg_@(Just msg)
-      | msgId msg == msgId' || B.null msgId' -> tryDeleteMsg_ mq >> pure msg_
-      | otherwise -> pure Nothing
-    _ -> pure Nothing
+  logQueueStates _ = pure ()
 
--- atomic delete (== read) last and peek next message if available
-tryDelPeekMsg :: MsgQueue -> MsgId -> IO (Maybe Message, Maybe Message)
-tryDelPeekMsg mq msgId' = atomically $
-  tryPeekMsg mq >>= \case
-    msg_@(Just msg)
-      | msgId msg == msgId' || B.null msgId' -> (msg_,) <$> (tryDeleteMsg_ mq >> tryPeekMsg mq)
-      | otherwise -> pure (Nothing, msg_)
-    _ -> pure (Nothing, Nothing)
+  logQueueState _ = pure ()
 
-deleteExpiredMsgs :: MsgQueue -> Int64 -> IO Int
-deleteExpiredMsgs mq old = atomically $ loop 0
-  where
-    loop dc =
-      tryPeekMsg mq >>= \case
-        Just Message {msgTs}
-          | systemSeconds msgTs < old ->
-              tryDeleteMsg_ mq >> loop (dc + 1)
-        _ -> pure dc
+  queueRec' = queueRec
+  {-# INLINE queueRec' #-}
 
-tryDeleteMsg_ :: MsgQueue -> STM ()
-tryDeleteMsg_ MsgQueue {msgQueue = q, size} =
-  tryReadTQueue q >>= \case
-    Just _ -> modifyTVar' size (subtract 1)
-    _ -> pure ()
+  getMsgQueue :: STMMsgStore -> RecipientId -> STMQueue -> STM STMMsgQueue
+  getMsgQueue _ _ STMQueue {msgQueue_} = readTVar msgQueue_ >>= maybe newQ pure
+    where
+      newQ = do
+        msgQueue <- newTQueue
+        canWrite <- newTVar True
+        size <- newTVar 0
+        let q = STMMsgQueue {msgQueue, canWrite, size}
+        writeTVar msgQueue_ (Just q)
+        pure q
 
-getQueueSize :: MsgQueue -> IO Int
-getQueueSize MsgQueue {size} = readTVarIO size
+  getPeekMsgQueue :: STMMsgStore -> RecipientId -> STMQueue -> STM (Maybe (STMMsgQueue, Message))
+  getPeekMsgQueue _ _ q@STMQueue {msgQueue_} = readTVar msgQueue_ $>>= \mq -> (mq,) <$$> tryPeekMsg_ q mq
+
+  -- does not create queue if it does not exist, does not delete it if it does (can't just close in-memory queue)
+  withIdleMsgQueue :: Int64 -> STMMsgStore -> RecipientId -> STMQueue -> (STMMsgQueue -> STM a) -> STM (Maybe a, Int)
+  withIdleMsgQueue _ _ _ STMQueue {msgQueue_} action = readTVar msgQueue_ >>= \case
+    Just q -> do
+      r <- action q
+      sz <- getQueueSize_ q
+      pure (Just r, sz)
+    Nothing -> pure (Nothing, 0)
+
+  deleteQueue :: STMMsgStore -> RecipientId -> STMQueue -> IO (Either ErrorType QueueRec)
+  deleteQueue ms rId q = fst <$$> deleteQueue' ms rId q
+
+  deleteQueueSize :: STMMsgStore -> RecipientId -> STMQueue -> IO (Either ErrorType (QueueRec, Int))
+  deleteQueueSize ms rId q = deleteQueue' ms rId q >>= mapM (traverse getSize)
+    -- traverse operates on the second tuple element
+    where
+      getSize = maybe (pure 0) (\STMMsgQueue {size} -> readTVarIO size)
+
+  getQueueMessages_ :: Bool -> STMMsgQueue -> STM [Message]
+  getQueueMessages_ drainMsgs = (if drainMsgs then flushTQueue else snapshotTQueue) . msgQueue
+    where
+      snapshotTQueue q = do
+        msgs <- flushTQueue q
+        mapM_ (writeTQueue q) msgs
+        pure msgs
+
+  writeMsg :: STMMsgStore -> RecipientId -> STMQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
+  writeMsg ms rId q' _logState msg = liftIO $ atomically $ do
+    STMMsgQueue {msgQueue = q, canWrite, size} <- getMsgQueue ms rId q'
+    canWrt <- readTVar canWrite
+    empty <- isEmptyTQueue q
+    if canWrt || empty
+      then do
+        canWrt' <- (quota >) <$> readTVar size
+        writeTVar canWrite $! canWrt'
+        modifyTVar' size (+ 1)
+        if canWrt'
+          then (writeTQueue q $! msg) $> Just (msg, empty)
+          else (writeTQueue q $! msgQuota) $> Nothing
+      else pure Nothing
+    where
+      STMMsgStore {storeConfig = STMStoreConfig {quota}} = ms
+      msgQuota = MessageQuota {msgId = messageId msg, msgTs = messageTs msg}
+
+  setOverQuota_ :: STMQueue -> IO ()
+  setOverQuota_ q = readTVarIO (msgQueue_ q) >>= mapM_ (\mq -> atomically $ writeTVar (canWrite mq) False)
+
+  getQueueSize_ :: STMMsgQueue -> STM Int
+  getQueueSize_ STMMsgQueue {size} = readTVar size
+
+  tryPeekMsg_ :: STMQueue -> STMMsgQueue -> STM (Maybe Message)
+  tryPeekMsg_ _ = tryPeekTQueue . msgQueue
+  {-# INLINE tryPeekMsg_ #-}
+
+  tryDeleteMsg_ :: STMQueue -> STMMsgQueue -> Bool -> STM ()
+  tryDeleteMsg_ _ STMMsgQueue {msgQueue = q, size} _logState =
+    tryReadTQueue q >>= \case
+      Just _ -> modifyTVar' size (subtract 1)
+      _ -> pure ()
+
+  isolateQueue :: RecipientId -> STMQueue -> String -> STM a -> ExceptT ErrorType IO a
+  isolateQueue _ _ _ = liftIO . atomically

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

@@ -0,0 +1,141 @@
+{-# LANGUAGE BangPatterns #-}
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE MultiWayIf #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE TupleSections #-}
+{-# LANGUAGE TypeFamilyDependencies #-}
+{-# OPTIONS_GHC -Wno-unrecognised-pragmas #-}
+
+{-# HLINT ignore "Redundant multi-way if" #-}
+
+module Simplex.Messaging.Server.MsgStore.Types where
+
+import Control.Concurrent.STM
+import Control.Monad (foldM)
+import Control.Monad.Trans.Except
+import Data.Functor (($>))
+import Data.Int (Int64)
+import Data.Kind
+import qualified Data.Map.Strict as M
+import Data.Time.Clock.System (SystemTime (systemSeconds))
+import Simplex.Messaging.Protocol
+import Simplex.Messaging.Server.QueueStore
+import Simplex.Messaging.Server.StoreLog.Types
+import Simplex.Messaging.TMap (TMap)
+import Simplex.Messaging.Util ((<$$>))
+import System.IO (IOMode (..))
+
+class MsgStoreClass s => STMQueueStore s where
+  queues' :: s -> TMap RecipientId (StoreQueue s)
+  senders' :: s -> TMap SenderId RecipientId
+  notifiers' :: s -> TMap NotifierId RecipientId
+  storeLog' :: s -> TVar (Maybe (StoreLog 'WriteMode))
+  mkQueue :: s -> QueueRec -> STM (StoreQueue s)
+  msgQueue_' :: StoreQueue s -> TVar (Maybe (MsgQueue s))
+
+class Monad (StoreMonad s) => MsgStoreClass s where
+  type StoreMonad s = (m :: Type -> Type) | m -> s
+  type MsgStoreConfig s = c | c -> s
+  type StoreQueue s = q | q -> s
+  type MsgQueue s = q | q -> s
+  newMsgStore :: MsgStoreConfig s -> IO s
+  setStoreLog :: s -> StoreLog 'WriteMode -> IO ()
+  closeMsgStore :: s -> IO ()
+  activeMsgQueues :: s -> TMap RecipientId (StoreQueue s)
+  withAllMsgQueues :: Monoid a => Bool -> s -> (RecipientId -> StoreQueue s -> IO a) -> IO a
+  logQueueStates :: s -> IO ()
+  logQueueState :: StoreQueue s -> StoreMonad s ()
+  queueRec' :: StoreQueue s -> TVar (Maybe QueueRec)
+  getPeekMsgQueue :: s -> RecipientId -> StoreQueue s -> StoreMonad s (Maybe (MsgQueue s, Message))
+  getMsgQueue :: s -> RecipientId -> StoreQueue s -> StoreMonad s (MsgQueue s)
+
+  -- the journal queue will be closed after action if it was initially closed or idle longer than interval in config
+  withIdleMsgQueue :: Int64 -> s -> RecipientId -> StoreQueue s -> (MsgQueue s -> StoreMonad s a) -> StoreMonad s (Maybe a, Int)
+  deleteQueue :: s -> RecipientId -> StoreQueue s -> IO (Either ErrorType QueueRec)
+  deleteQueueSize :: s -> RecipientId -> StoreQueue s -> IO (Either ErrorType (QueueRec, Int))
+  getQueueMessages_ :: Bool -> MsgQueue s -> StoreMonad s [Message]
+  writeMsg :: s -> RecipientId -> StoreQueue s -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
+  setOverQuota_ :: StoreQueue s -> IO () -- can ONLY be used while restoring messages, not while server running
+  getQueueSize_ :: MsgQueue s -> StoreMonad s Int
+  tryPeekMsg_ :: StoreQueue s -> MsgQueue s -> StoreMonad s (Maybe Message)
+  tryDeleteMsg_ :: StoreQueue s -> MsgQueue s -> Bool -> StoreMonad s ()
+  isolateQueue :: RecipientId -> StoreQueue s -> String -> StoreMonad s a -> ExceptT ErrorType IO a
+
+data MSType = MSMemory | MSJournal
+
+data SMSType :: MSType -> Type where
+  SMSMemory :: SMSType 'MSMemory
+  SMSJournal :: SMSType 'MSJournal
+
+data AMSType = forall s. AMSType (SMSType s)
+
+withActiveMsgQueues :: (MsgStoreClass s, Monoid a) => s -> (RecipientId -> StoreQueue s -> IO a) -> IO a
+withActiveMsgQueues st f = readTVarIO (activeMsgQueues st) >>= foldM run mempty . M.assocs
+  where
+    run !acc (k, v) = do
+      r <- f k v
+      pure $! acc <> r
+
+getQueueMessages :: MsgStoreClass s => Bool -> s -> RecipientId -> StoreQueue s -> ExceptT ErrorType IO [Message]
+getQueueMessages drainMsgs st rId q = withPeekMsgQueue st rId q "getQueueSize" $ maybe (pure []) (getQueueMessages_ drainMsgs . fst)
+{-# INLINE getQueueMessages #-}
+
+getQueueSize :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> ExceptT ErrorType IO Int
+getQueueSize st rId q = withPeekMsgQueue st rId q "getQueueSize" $ maybe (pure 0) (getQueueSize_ . fst)
+{-# INLINE getQueueSize #-}
+
+tryPeekMsg :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> ExceptT ErrorType IO (Maybe Message)
+tryPeekMsg st rId q = snd <$$> withPeekMsgQueue st rId q "tryPeekMsg" pure
+{-# INLINE tryPeekMsg #-}
+
+tryDelMsg :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message)
+tryDelMsg st rId q msgId' =
+  withPeekMsgQueue st rId q "tryDelMsg" $
+    maybe (pure Nothing) $ \(mq, msg) ->
+      if
+        | messageId msg == msgId' ->
+            tryDeleteMsg_ q mq True $> Just msg
+        | otherwise -> pure Nothing
+
+-- atomic delete (== read) last and peek next message if available
+tryDelPeekMsg :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message, Maybe Message)
+tryDelPeekMsg st rId q msgId' =
+  withPeekMsgQueue st rId q "tryDelPeekMsg" $
+    maybe (pure (Nothing, Nothing)) $ \(mq, msg) ->
+      if
+        | messageId msg == msgId' -> (Just msg,) <$> (tryDeleteMsg_ q mq True >> tryPeekMsg_ q mq)
+        | otherwise -> pure (Nothing, Just msg)
+
+-- The action is called with Nothing when it is known that the queue is empty
+withPeekMsgQueue :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> String -> (Maybe (MsgQueue s, Message) -> StoreMonad s a) -> ExceptT ErrorType IO a
+withPeekMsgQueue st rId q op a = isolateQueue rId q op $ getPeekMsgQueue st rId q >>= a
+{-# INLINE withPeekMsgQueue #-}
+
+deleteExpiredMsgs :: MsgStoreClass s => s -> RecipientId -> StoreQueue s -> Int64 -> ExceptT ErrorType IO Int
+deleteExpiredMsgs st rId q old =
+  isolateQueue rId q "deleteExpiredMsgs" $
+    getMsgQueue st rId q >>= deleteExpireMsgs_ old q
+
+-- closed and idle queues will be closed after expiration
+-- returns (expired count, queue size after expiration)
+idleDeleteExpiredMsgs :: MsgStoreClass s => Int64 -> s -> RecipientId -> StoreQueue s -> Int64 -> ExceptT ErrorType IO (Maybe Int, Int)
+idleDeleteExpiredMsgs now st rId q old =
+  isolateQueue rId q "idleDeleteExpiredMsgs" $
+    withIdleMsgQueue now st rId q (deleteExpireMsgs_ old q)
+
+deleteExpireMsgs_ :: MsgStoreClass s => Int64 -> StoreQueue s -> MsgQueue s -> StoreMonad s Int
+deleteExpireMsgs_ old q mq = do
+  n <- loop 0
+  logQueueState q
+  pure n
+  where
+    loop dc =
+      tryPeekMsg_ q mq >>= \case
+        Just Message {msgTs}
+          | systemSeconds msgTs < old ->
+              tryDeleteMsg_ q mq False >> loop (dc + 1)
+        _ -> pure dc

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

@@ -1,119 +1,196 @@
 {-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE DataKinds #-}
+{-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE KindSignatures #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TupleSections #-}
 {-# LANGUAGE UndecidableInstances #-}
 
 module Simplex.Messaging.Server.QueueStore.STM
-  ( QueueStore (..),
-    newQueueStore,
-    addQueue,
+  ( addQueue,
     getQueue,
+    getQueueRec,
     secureQueue,
     addQueueNotifier,
     deleteQueueNotifier,
     suspendQueue,
     updateQueueTime,
-    deleteQueue,
+    deleteQueue',
+    readQueueStore,
+    withLog',
   )
 where
 
+import qualified Control.Exception as E
+import Control.Logger.Simple
 import Control.Monad
+import Control.Monad.IO.Class
+import Control.Monad.Trans.Except
+import Data.Bitraversable (bimapM)
+import qualified Data.ByteString.Char8 as B
+import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Functor (($>))
+import qualified Data.Text as T
+import Data.Text.Encoding (decodeLatin1)
+import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
+import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.QueueStore
-import Simplex.Messaging.TMap (TMap)
+import Simplex.Messaging.Server.StoreLog
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (ifM, ($>>=))
+import Simplex.Messaging.Util (ifM, tshow, ($>>=), (<$$))
+import System.IO
 import UnliftIO.STM
 
-data QueueStore = QueueStore
-  { queues :: TMap RecipientId (TVar QueueRec),
-    senders :: TMap SenderId RecipientId,
-    notifiers :: TMap NotifierId RecipientId
-  }
-
-newQueueStore :: IO QueueStore
-newQueueStore = do
-  queues <- TM.emptyIO
-  senders <- TM.emptyIO
-  notifiers <- TM.emptyIO
-  pure QueueStore {queues, senders, notifiers}
-
-addQueue :: QueueStore -> QueueRec -> IO (Either ErrorType ())
-addQueue QueueStore {queues, senders} q@QueueRec {recipientId = rId, senderId = sId} = atomically $ do
-  ifM hasId (pure $ Left DUPLICATE_) $ do
-    TM.insertM rId (newTVar q) queues
-    TM.insert sId rId senders
-    pure $ Right ()
+addQueue :: STMQueueStore s => s -> QueueRec -> IO (Either ErrorType (StoreQueue s))
+addQueue st qr@QueueRec {recipientId = rId, senderId = sId, notifier}=
+  atomically add
+    $>>= \q -> q <$$ withLog "addQueue" st (`logCreateQueue` qr)
   where
-    hasId = (||) <$> TM.member rId queues <*> TM.member sId senders
+    add = ifM hasId (pure $ Left DUPLICATE_) $ do
+      q <- mkQueue st qr -- STMQueue lock <$> (newTVar $! Just qr) <*> newTVar Nothing
+      TM.insert rId q $ queues' st
+      TM.insert sId rId $ senders' st
+      forM_ notifier $ \NtfCreds {notifierId} -> TM.insert notifierId rId $ notifiers' st
+      pure $ Right q
+    hasId = or <$> sequence [TM.member rId $ queues' st, TM.member sId $ senders' st, hasNotifier]
+    hasNotifier = maybe (pure False) (\NtfCreds {notifierId} -> TM.member notifierId (notifiers' st)) notifier
 
-getQueue :: DirectParty p => QueueStore -> SParty p -> QueueId -> IO (Either ErrorType QueueRec)
-getQueue QueueStore {queues, senders, notifiers} party qId =
-  toResult <$> (mapM readTVarIO =<< getVar)
+getQueue :: (STMQueueStore s, DirectParty p) => s -> SParty p -> QueueId -> IO (Either ErrorType (StoreQueue s))
+getQueue st party qId =
+  maybe (Left AUTH) Right <$> case party of
+    SRecipient -> TM.lookupIO qId $ queues' st
+    SSender -> TM.lookupIO qId (senders' st) $>>= (`TM.lookupIO` queues' st)
+    SNotifier -> TM.lookupIO qId (notifiers' st) $>>= (`TM.lookupIO` queues' st)
+
+getQueueRec :: (STMQueueStore s, DirectParty p) => s -> SParty p -> QueueId -> IO (Either ErrorType (StoreQueue s, QueueRec))
+getQueueRec st party qId =
+  getQueue st party qId
+    $>>= (\q -> maybe (Left AUTH) (Right . (q,)) <$> readTVarIO (queueRec' q))
+
+secureQueue :: STMQueueStore s => s -> StoreQueue s -> SndPublicAuthKey -> IO (Either ErrorType ())
+secureQueue st sq sKey =
+  atomically (readQueueRec qr $>>= secure)
+    $>>= \rId -> withLog "secureQueue" st $ \s -> logSecureQueue s rId sKey
   where
-    getVar = case party of
-      SRecipient -> TM.lookupIO qId queues
-      SSender -> TM.lookupIO qId senders $>>= (`TM.lookupIO` queues)
-      SNotifier -> TM.lookupIO qId notifiers $>>= (`TM.lookupIO` queues)
+    qr = queueRec' sq
+    secure q@QueueRec {recipientId = rId} = case senderKey q of
+      Just k -> pure $ if sKey == k then Right rId else Left AUTH
+      Nothing -> do
+        writeTVar qr $ Just q {senderKey = Just sKey}
+        pure $ Right rId
 
-secureQueue :: QueueStore -> RecipientId -> SndPublicAuthKey -> IO (Either ErrorType QueueRec)
-secureQueue QueueStore {queues} rId sKey = toResult <$> do
-  TM.lookupIO rId queues $>>= \qVar -> atomically $ 
-    readTVar qVar >>= \q -> case senderKey q of
-      Just k -> pure $ if sKey == k then Just q else Nothing
-      _ ->
-        let !q' = q {senderKey = Just sKey}
-         in writeTVar qVar q' $> Just q'
-
-addQueueNotifier :: QueueStore -> RecipientId -> NtfCreds -> IO (Either ErrorType (Maybe NotifierId))
-addQueueNotifier QueueStore {queues, notifiers} rId ntfCreds@NtfCreds {notifierId = nId} = do
-  TM.lookupIO rId queues >>= \case
-    Just qVar -> atomically $ ifM (TM.member nId notifiers) (pure $ Left DUPLICATE_) $ do
-      q <- readTVar qVar
-      nId_ <- forM (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId notifiers $> notifierId
+addQueueNotifier :: STMQueueStore s => s -> StoreQueue s -> NtfCreds -> IO (Either ErrorType (Maybe NotifierId))
+addQueueNotifier st sq ntfCreds@NtfCreds {notifierId = nId} =
+  atomically (readQueueRec qr $>>= add)
+    $>>= \(rId, nId_) -> nId_ <$$ withLog "addQueueNotifier" st (\s -> logAddNotifier s rId ntfCreds)
+  where
+    qr = queueRec' sq
+    add q@QueueRec {recipientId = rId} = ifM (TM.member nId (notifiers' st)) (pure $ Left DUPLICATE_) $ do
+      nId_ <- forM (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId (notifiers' st) $> notifierId
       let !q' = q {notifier = Just ntfCreds}
-      writeTVar qVar q'
-      TM.insert nId rId notifiers
-      pure $ Right nId_
-    Nothing -> pure $ Left AUTH
+      writeTVar qr $ Just q'
+      TM.insert nId rId $ notifiers' st
+      pure $ Right (rId, nId_)
 
-deleteQueueNotifier :: QueueStore -> RecipientId -> IO (Either ErrorType (Maybe NotifierId))
-deleteQueueNotifier QueueStore {queues, notifiers} rId =
-  withQueue rId queues $ \qVar -> do
-    q <- readTVar qVar
-    forM (notifier q) $ \NtfCreds {notifierId} -> do
-      TM.delete notifierId notifiers
-      writeTVar qVar $! q {notifier = Nothing}
+deleteQueueNotifier :: STMQueueStore s => s -> StoreQueue s -> IO (Either ErrorType (Maybe NotifierId))
+deleteQueueNotifier st sq =
+  atomically (readQueueRec qr >>= mapM delete)
+    $>>= \(rId, nId_) -> nId_ <$$ withLog "deleteQueueNotifier" st (`logDeleteNotifier` rId)
+  where
+    qr = queueRec' sq
+    delete q = fmap (recipientId q,) $ forM (notifier q) $ \NtfCreds {notifierId} -> do
+      TM.delete notifierId $ notifiers' st
+      writeTVar qr $! Just q {notifier = Nothing}
       pure notifierId
 
-suspendQueue :: QueueStore -> RecipientId -> IO (Either ErrorType ())
-suspendQueue QueueStore {queues} rId =
-  withQueue rId queues (`modifyTVar'` \q -> q {status = QueueOff})
+suspendQueue :: STMQueueStore s => s -> StoreQueue s -> IO (Either ErrorType ())
+suspendQueue st sq =
+  atomically (readQueueRec qr >>= mapM suspend)
+    $>>= \rId -> withLog "suspendQueue" st (`logSuspendQueue` rId)
+  where
+    qr = queueRec' sq
+    suspend q = do
+      writeTVar qr $! Just q {status = QueueOff}
+      pure $ recipientId q
 
-updateQueueTime :: QueueStore -> RecipientId -> RoundedSystemTime -> IO ()
-updateQueueTime QueueStore {queues} rId t =
-  void $ withQueue rId queues (`modifyTVar'` \q -> q {updatedAt = Just t})
+updateQueueTime :: STMQueueStore s => s -> StoreQueue s -> RoundedSystemTime -> IO (Either ErrorType QueueRec)
+updateQueueTime st sq t = atomically (readQueueRec qr >>= mapM update) $>>= log'
+  where
+    qr = queueRec' sq
+    update q@QueueRec {updatedAt}
+      | updatedAt == Just t = pure (q, False)
+      | otherwise =
+          let !q' = q {updatedAt = Just t}
+           in (writeTVar qr $! Just q') $> (q', True)
+    log' (q, changed)
+      | changed = q <$$ withLog "updateQueueTime" st (\sl -> logUpdateQueueTime sl (recipientId q) t)
+      | otherwise = pure $ Right q
 
-deleteQueue :: QueueStore -> RecipientId -> IO (Either ErrorType QueueRec)
-deleteQueue QueueStore {queues, senders, notifiers} rId = atomically $ do
-  TM.lookupDelete rId queues >>= \case
-    Just qVar ->
-      readTVar qVar >>= \q -> do
-        TM.delete (senderId q) senders
-        forM_ (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId notifiers
-        pure $ Right q
-    _ -> pure $ Left AUTH
+deleteQueue' :: STMQueueStore s => s -> RecipientId -> StoreQueue s -> IO (Either ErrorType (QueueRec, Maybe (MsgQueue s)))
+deleteQueue' st rId sq =
+  atomically (readQueueRec qr >>= mapM delete)
+    $>>= \q -> withLog "deleteQueue" st (`logDeleteQueue` rId)
+    >>= bimapM pure (\_ -> (q,) <$> atomically (swapTVar (msgQueue_' sq) Nothing))
+  where
+    qr = queueRec' sq
+    delete q = do
+      writeTVar qr Nothing
+      TM.delete (senderId q) $ senders' st
+      forM_ (notifier q) $ \NtfCreds {notifierId} -> TM.delete notifierId $ notifiers' st
+      pure q
 
-toResult :: Maybe a -> Either ErrorType a
-toResult = maybe (Left AUTH) Right
+readQueueRec :: TVar (Maybe QueueRec) -> STM (Either ErrorType QueueRec)
+readQueueRec qr = maybe (Left AUTH) Right <$> readTVar qr
+{-# INLINE readQueueRec #-}
 
-withQueue :: RecipientId -> TMap RecipientId (TVar QueueRec) -> (TVar QueueRec -> STM a) -> IO (Either ErrorType a)
-withQueue rId queues f = toResult <$> TM.lookupIO rId queues >>= atomically . mapM f
+withLog' :: String -> TVar (Maybe (StoreLog 'WriteMode)) -> (StoreLog 'WriteMode -> IO ()) -> IO (Either ErrorType ())
+withLog' name sl action =
+  readTVarIO sl
+    >>= maybe (pure $ Right ()) (E.try . action >=> bimapM logErr pure)
+  where
+    logErr :: E.SomeException -> IO ErrorType
+    logErr e = logError ("STORE: " <> T.pack err) $> STORE err
+      where
+        err = name <> ", withLog, " <> show e
+
+withLog :: STMQueueStore s => String -> s -> (StoreLog 'WriteMode -> IO ()) -> IO (Either ErrorType ())
+withLog name = withLog' name . storeLog'
+
+readQueueStore :: forall s. STMQueueStore s => FilePath -> s -> IO ()
+readQueueStore f st = withFile f ReadMode $ LB.hGetContents >=> mapM_ processLine . LB.lines
+  where
+    processLine :: LB.ByteString -> IO ()
+    processLine s' = either printError procLogRecord (strDecode s)
+      where
+        s = LB.toStrict s'
+        procLogRecord :: StoreLogRecord -> IO ()
+        procLogRecord = \case
+          CreateQueue q -> addQueue st q >>= qError (recipientId q) "CreateQueue"
+          SecureQueue qId sKey -> withQueue qId "SecureQueue" $ \q -> secureQueue st q sKey
+          AddNotifier qId ntfCreds -> withQueue qId "AddNotifier" $ \q -> addQueueNotifier st q ntfCreds
+          SuspendQueue qId -> withQueue qId "SuspendQueue" $ suspendQueue st
+          DeleteQueue qId -> withQueue qId "DeleteQueue" $ deleteQueue st qId
+          DeleteNotifier qId -> withQueue qId "DeleteNotifier" $ deleteQueueNotifier st
+          UpdateTime qId t -> withQueue qId "UpdateTime" $ \q -> updateQueueTime st q t
+        printError :: String -> IO ()
+        printError e = B.putStrLn $ "Error parsing log: " <> B.pack e <> " - " <> s
+        withQueue :: forall a. RecipientId -> T.Text -> (StoreQueue s -> IO (Either ErrorType a)) -> IO ()
+        withQueue qId op a = runExceptT go >>= qError qId op
+          where
+            go = do
+              q <- ExceptT $ getQueue st SRecipient qId
+              liftIO (readTVarIO $ queueRec' q) >>= \case
+                Nothing -> logWarn $ logPfx qId op <> "already deleted"
+                Just _ -> void $ ExceptT $ a q
+        qError qId op = \case
+          Left e -> logError $ logPfx qId op <> tshow e
+          Right _ -> pure ()
+        logPfx qId op = "STORE: " <> op <> ", stored queue " <> decodeLatin1 (strEncode qId) <> ", "

src/Simplex/Messaging/Server/StoreLog.hs

@@ -6,6 +6,7 @@
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE ScopedTypeVariables #-}
 {-# OPTIONS_GHC -fno-warn-orphans #-}
 
 module Simplex.Messaging.Server.StoreLog
@@ -23,36 +24,33 @@ module Simplex.Messaging.Server.StoreLog
     logDeleteQueue,
     logDeleteNotifier,
     logUpdateQueueTime,
-    readWriteQueueStore,
     readWriteStoreLog,
+    writeQueueStore,
   )
 where
 
 import Control.Applicative (optional, (<|>))
 import Control.Concurrent.STM
+import qualified Control.Exception as E
 import Control.Logger.Simple
 import Control.Monad
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import qualified Data.ByteString.Char8 as B
-import qualified Data.ByteString.Lazy.Char8 as LB
-import qualified Data.ByteString.Base64.URL as B64
+import qualified Data.Map.Strict as M
 import qualified Data.Text as T
 import Data.Time.Clock (getCurrentTime)
 import Data.Time.Format.ISO8601 (iso8601Show)
+import GHC.IO (catchAny)
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
+import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.QueueStore
-import Simplex.Messaging.Server.QueueStore.STM
-import Simplex.Messaging.Util (bshow, ifM, unlessM, whenM)
+import Simplex.Messaging.Server.StoreLog.Types
+import qualified Simplex.Messaging.TMap as TM
+import Simplex.Messaging.Util (ifM, tshow, unlessM, whenM)
 import System.Directory (doesFileExist, renameFile)
 import System.IO
 
--- | opaque container for file handle with a type-safe IOMode
--- constructors are not exported, openWriteStoreLog and openReadStoreLog should be used instead
-data StoreLog (a :: IOMode) where
-  ReadStoreLog :: FilePath -> Handle -> StoreLog 'ReadMode
-  WriteStoreLog :: FilePath -> Handle -> StoreLog 'WriteMode
-
 data StoreLogRecord
   = CreateQueue QueueRec
   | SecureQueue QueueId SndPublicAuthKey
@@ -159,11 +157,13 @@ storeLogFilePath = \case
 
 closeStoreLog :: StoreLog a -> IO ()
 closeStoreLog = \case
-  WriteStoreLog _ h -> hClose h
-  ReadStoreLog _ h -> hClose h
+  WriteStoreLog _ h -> close_ h
+  ReadStoreLog _ h -> close_ h
+  where
+    close_ h = hClose h `catchAny` \e -> logError ("STORE: closeStoreLog, error closing, " <> tshow e)
 
 writeStoreLogRecord :: StrEncoding r => StoreLog 'WriteMode -> r -> IO ()
-writeStoreLogRecord (WriteStoreLog _ h) r = do
+writeStoreLogRecord (WriteStoreLog _ h) r = E.uninterruptibleMask_ $ do
   B.hPut h $ strEncode r `B.snoc` '\n' -- hPutStrLn makes write non-atomic for length > 1024
   hFlush h
 
@@ -188,9 +188,6 @@ logDeleteNotifier s = writeStoreLogRecord s . DeleteNotifier
 logUpdateQueueTime :: StoreLog 'WriteMode -> QueueId -> RoundedSystemTime -> IO ()
 logUpdateQueueTime s qId t = writeStoreLogRecord s $ UpdateTime qId t
 
-readWriteQueueStore :: FilePath -> QueueStore -> IO (StoreLog 'WriteMode)
-readWriteQueueStore = readWriteStoreLog readQueues writeQueues
-
 readWriteStoreLog :: (FilePath -> s -> IO ()) -> (StoreLog 'WriteMode -> s -> IO ()) -> FilePath -> s -> IO (StoreLog 'WriteMode)
 readWriteStoreLog readStore writeStore f st =
   ifM
@@ -226,31 +223,11 @@ readWriteStoreLog readStore writeStore f st =
       renameFile tempBackup timedBackup
       logInfo $ "original state preserved as " <> T.pack timedBackup
 
-writeQueues :: StoreLog 'WriteMode -> QueueStore -> IO ()
-writeQueues s st = readTVarIO (queues st) >>= mapM_ writeQueue
+writeQueueStore :: STMQueueStore s => StoreLog 'WriteMode -> s -> IO ()
+writeQueueStore s st = readTVarIO (activeMsgQueues st) >>= mapM_ writeQueue . M.assocs
   where
-    writeQueue v = readTVarIO v >>= \q -> when (active q) $ logCreateQueue s q
+    writeQueue (rId, q) =
+      readTVarIO (queueRec' q) >>= \case
+        Just q' -> when (active q') $ logCreateQueue s q' -- TODO we should log suspended queues when we use them
+        Nothing -> atomically $ TM.delete rId $ activeMsgQueues st
     active QueueRec {status} = status == QueueActive
-
-readQueues :: FilePath -> QueueStore -> IO ()
-readQueues f st = withFile f ReadMode $ LB.hGetContents >=> mapM_ processLine . LB.lines
-  where
-    processLine :: LB.ByteString -> IO ()
-    processLine s' = either printError procLogRecord (strDecode s)
-      where
-        s = LB.toStrict s'
-        procLogRecord :: StoreLogRecord -> IO ()
-        procLogRecord = \case
-          CreateQueue q -> addQueue st q >>= qError (recipientId q)
-          SecureQueue qId sKey -> secureQueue st qId sKey >>= qError qId
-          AddNotifier qId ntfCreds -> addQueueNotifier st qId ntfCreds >>= qError qId
-          SuspendQueue qId -> suspendQueue st qId >>= qError qId
-          DeleteQueue qId -> deleteQueue st qId >>= qError qId
-          DeleteNotifier qId -> deleteQueueNotifier st qId >>= qError qId
-          UpdateTime qId t -> updateQueueTime st qId t
-        printError :: String -> IO ()
-        printError e = B.putStrLn $ "Error parsing log: " <> B.pack e <> " - " <> s
-        qError :: RecipientId -> Either ErrorType a -> IO ()
-        qError (EntityId qId) = \case
-          Left e -> B.putStrLn $ "stored queue " <> B64.encode qId <> " error: " <> bshow e
-          Right _ -> pure ()

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

@@ -0,0 +1,13 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE KindSignatures #-}
+
+module Simplex.Messaging.Server.StoreLog.Types where
+
+import System.IO (Handle, IOMode (..))
+
+-- | opaque container for file handle with a type-safe IOMode
+-- constructors are not exported, openWriteStoreLog and openReadStoreLog should be used instead
+data StoreLog (a :: IOMode) where
+  ReadStoreLog :: FilePath -> Handle -> StoreLog 'ReadMode
+  WriteStoreLog :: FilePath -> Handle -> StoreLog 'WriteMode

src/Simplex/Messaging/Util.hs

@@ -14,11 +14,13 @@ import Data.Bifunctor (first)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
-import Data.Int (Int64)
 import Data.IORef
+import Data.Int (Int64)
 import Data.List (groupBy, sortOn)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.List.NonEmpty as L
+import Data.Map.Strict (Map)
+import qualified Data.Map.Strict as M
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeUtf8With, encodeUtf8)
@@ -33,12 +35,16 @@ raceAny_ = r []
     r as (m : ms) = withAsync m $ \a -> r (a : as) ms
     r as [] = void $ waitAnyCancel as
 
-infixl 4 <$$>, <$?>
+infixl 4 <$$>, <$$, <$?>
 
 (<$$>) :: (Functor f, Functor g) => (a -> b) -> f (g a) -> f (g b)
 (<$$>) = fmap . fmap
 {-# INLINE (<$$>) #-}
 
+(<$$) :: (Functor f, Functor g) => b -> f (g a) -> f (g b)
+(<$$) = fmap . fmap . const
+{-# INLINE (<$$) #-}
+
 (<$?>) :: MonadFail m => (a -> Either String b) -> m a -> m b
 f <$?> m = either fail pure . f =<< m
 {-# INLINE (<$?>) #-}
@@ -185,3 +191,7 @@ encodeJSON = safeDecodeUtf8 . LB.toStrict . J.encode
 
 decodeJSON :: FromJSON a => Text -> Maybe a
 decodeJSON = J.decode . LB.fromStrict . encodeUtf8
+
+traverseWithKey_ :: Monad m => (k -> v -> m ()) -> Map k v -> m ()
+traverseWithKey_ f = M.foldrWithKey (\k v -> (f k v >>)) (pure ())
+{-# INLINE traverseWithKey_ #-}

tests/AgentTests.hs

@@ -16,6 +16,7 @@ import AgentTests.FunctionalAPITests (functionalAPITests)
 import AgentTests.MigrationTests (migrationTests)
 import AgentTests.NotificationTests (notificationTests)
 import AgentTests.SQLiteTests (storeTests)
+import AgentTests.ServerChoice (serverChoiceTests)
 import Simplex.Messaging.Transport (ATransport (..))
 import Test.Hspec
 
@@ -26,4 +27,5 @@ agentTests (ATransport t) = do
   describe "Functional API" $ functionalAPITests (ATransport t)
   describe "Notification tests" $ notificationTests (ATransport t)
   describe "SQLite store" storeTests
+  describe "Chosen servers" serverChoiceTests
   describe "Migration tests" migrationTests

tests/AgentTests/FunctionalAPITests.hs

@@ -77,7 +77,7 @@ import Data.Word (Word16)
 import qualified Database.SQLite.Simple as SQL
 import GHC.Stack (withFrozenCallStack)
 import SMPAgentClient
-import SMPClient (cfg, prevRange, prevVersion, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerOn, withSmpServerProxy, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
+import SMPClient (cfg, prevRange, prevVersion, testPort, testPort2, testStoreLogFile2, testStoreMsgsDir2, withSmpServer, withSmpServerConfigOn, withSmpServerProxy, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
 import Simplex.Messaging.Agent hiding (createConnection, joinConnection, sendMessage)
 import qualified Simplex.Messaging.Agent as A
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), ServerQueueInfo (..), UserNetworkInfo (..), UserNetworkType (..), waitForUserNetwork)
@@ -96,6 +96,7 @@ import Simplex.Messaging.Protocol (BasicAuth, ErrorType (..), MsgBody, ProtocolS
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Server.Expiration
+import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..))
 import Simplex.Messaging.Server.QueueStore.QueueInfo
 import Simplex.Messaging.Transport (ATransport (..), SMPVersion, VersionSMP, authCmdsSMPVersion, basicAuthSMPVersion, batchCmdsSMPVersion, currentServerSMPRelayVersion, sndAuthKeySMPVersion, supportedSMPHandshakes)
 import Simplex.Messaging.Util (bshow, diffToMicroseconds)
@@ -530,7 +531,7 @@ testRatchetMatrix2 t runTest = do
 testServerMatrix2 :: HasCallStack => ATransport -> (InitialAgentServers -> IO ()) -> Spec
 testServerMatrix2 t runTest = do
   it "1 server" $ withSmpServer t $ runTest initAgentServers
-  it "2 servers" $ withSmpServer t . withSmpServerOn t testPort2 $ runTest initAgentServers2
+  it "2 servers" $ withSmpServer t $ withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> runTest initAgentServers2
 
 testPQMatrix2 :: HasCallStack => ATransport -> (HasCallStack => (AgentClient, InitialKeys) -> (AgentClient, PQSupport) -> AgentMsgId -> IO ()) -> Spec
 testPQMatrix2 = pqMatrix2_ True
@@ -1041,7 +1042,7 @@ testAllowConnectionClientRestart t = do
   bob <- getSMPAgentClient' 2 agentCfg initAgentServersSrv2 testDB2
   withSmpServerStoreLogOn t testPort $ \_ -> do
     (aliceId, bobId, confId) <-
-      withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \_ -> do
+      withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> do
         runRight $ do
           (bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe
           (aliceId, sqSecured) <- joinConnection bob 1 True qInfo "bob's connInfo" SMSubscribe
@@ -1063,8 +1064,7 @@ testAllowConnectionClientRestart t = do
     alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
     runRight_ $ subscribeConnection alice2 bobId
     threadDelay 500000
-
-    withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \_ -> do
+    withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> do
       runRight $ do
         ("", "", UP _ _) <- nGet bob
         get alice2 ##> ("", bobId, CON)
@@ -1282,7 +1282,7 @@ testSkippedMessages :: HasCallStack => ATransport -> IO ()
 testSkippedMessages t = do
   alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
   bob <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
-  (aliceId, bobId) <- withSmpServerStoreLogOn t testPort $ \_ -> do
+  (aliceId, bobId) <- withSmpServerConfigOn t cfg' testPort $ \_ -> do
     (aliceId, bobId) <- runRight $ makeConnection alice bob
     runRight_ $ do
       2 <- sendMessage alice bobId SMP.noMsgFlags "hello"
@@ -1310,7 +1310,7 @@ testSkippedMessages t = do
   alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB
   bob2 <- getSMPAgentClient' 4 agentCfg initAgentServers testDB2
 
-  withSmpServerStoreLogOn t testPort $ \_ -> do
+  withSmpServerConfigOn t cfg' testPort $ \_ -> do
     runRight_ $ do
       subscribeConnection bob2 aliceId
       subscribeConnection alice2 bobId
@@ -1326,6 +1326,8 @@ testSkippedMessages t = do
       ackMessage bob2 aliceId 4 Nothing
   disposeAgentClient alice2
   disposeAgentClient bob2
+  where
+    cfg' = cfg {msgStoreType = AMSType SMSMemory, storeMsgsFile = Nothing}
 
 testDeliveryAfterSubscriptionError :: HasCallStack => ATransport -> IO ()
 testDeliveryAfterSubscriptionError t = do
@@ -1429,7 +1431,7 @@ withUP a bId p =
       ]
 
 testExpireMessageQuota :: HasCallStack => ATransport -> IO ()
-testExpireMessageQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testPort $ \_ -> do
+testExpireMessageQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1, maxJournalMsgCount = 2} testPort $ \_ -> do
   a <- getSMPAgentClient' 1 agentCfg {quotaExceededTimeout = 1, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB
   b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aId, bId) <- runRight $ do
@@ -1455,7 +1457,7 @@ testExpireMessageQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testP
   disposeAgentClient a
 
 testExpireManyMessagesQuota :: ATransport -> IO ()
-testExpireManyMessagesQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testPort $ \_ -> do
+testExpireManyMessagesQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1, maxJournalMsgCount = 2} testPort $ \_ -> do
   a <- getSMPAgentClient' 1 agentCfg {quotaExceededTimeout = 2, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB
   b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2
   (aId, bId) <- runRight $ do
@@ -1944,7 +1946,7 @@ testBatchedSubscriptions nCreate nDel t =
     runServers :: ExceptT AgentErrorType IO a -> IO a
     runServers a = do
       withSmpServerStoreLogOn t testPort $ \t1 -> do
-        res <- withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \t2 ->
+        res <- withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \t2 ->
           runRight a `finally` killThread t2
         killThread t1
         pure res
@@ -2356,7 +2358,7 @@ testJoinConnectionAsyncReplyErrorV8 t = do
         ConnectionStats {rcvQueuesInfo = [], sndQueuesInfo = [SndQueueInfo {}]} <- getConnectionServers b aId
         pure (aId, bId)
       nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
-      withSmpServerOn t testPort2 $ do
+      withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> do
         get b =##> \case ("2", c, JOINED sqSecured) -> c == aId && not sqSecured; _ -> False
         confId <- withSmpServerStoreLogOn t testPort $ \_ -> do
           pGet a >>= \case
@@ -2395,7 +2397,7 @@ testJoinConnectionAsyncReplyError t = do
         ConnectionStats {rcvQueuesInfo = [], sndQueuesInfo = [SndQueueInfo {}]} <- getConnectionServers b aId
         pure (aId, bId)
       nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False
-      withSmpServerOn t testPort2 $ do
+      withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \_ -> do
         confId <- withSmpServerStoreLogOn t testPort $ \_ -> do
           -- both servers need to be online for connection to progress because of SKEY
           get b =##> \case ("2", c, JOINED sqSecured) -> c == aId && sqSecured; _ -> False
@@ -2974,7 +2976,7 @@ testDeliveryReceiptsVersion t = do
 
 testDeliveryReceiptsConcurrent :: HasCallStack => ATransport -> IO ()
 testDeliveryReceiptsConcurrent t =
-  withSmpServerConfigOn t cfg {msgQueueQuota = 256} testPort $ \_ -> do
+  withSmpServerConfigOn t cfg {msgQueueQuota = 256, maxJournalMsgCount = 512} testPort $ \_ -> do
     withAgentClients2 $ \a b -> do
       (aId, bId) <- runRight $ makeConnection a b
       t1 <- liftIO getCurrentTime
@@ -3074,8 +3076,9 @@ testTwoUsers = withAgentClients2 $ \a b -> do
     ("", "", DOWN _ _) <- nGet a
     ("", "", DOWN _ _) <- nGet a
     ("", "", DOWN _ _) <- nGet a
-    ("", "", DOWN _ _) <- nGet a
-    ("", "", UP _ _) <- nGet a
+    -- to avoice race condition
+    nGet a =##> \case ("", "", DOWN _ _) -> True; ("", "", UP _ _) -> True; _ -> False
+    nGet a =##> \case ("", "", UP _ _) -> True; ("", "", DOWN _ _) -> True; _ -> False
     ("", "", UP _ _) <- nGet a
     ("", "", UP _ _) <- nGet a
     ("", "", UP _ _) <- nGet a

tests/AgentTests/NotificationTests.hs

@@ -56,7 +56,7 @@ import Data.Text.Encoding (encodeUtf8)
 import Database.SQLite.Simple.QQ (sql)
 import NtfClient
 import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, testDB2, testNtfServer, testNtfServer2)
-import SMPClient (cfg, cfgVPrev, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
+import SMPClient (cfg, cfgVPrev, testPort, testPort2, testStoreLogFile2, testStoreMsgsDir2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn)
 import Simplex.Messaging.Agent hiding (createConnection, joinConnection, sendMessage)
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..), withStore')
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig, Env (..), InitialAgentServers)
@@ -790,7 +790,7 @@ testNotificationsSMPRestartBatch n t apns =
     runServers :: ExceptT AgentErrorType IO a -> IO a
     runServers a = do
       withSmpServerStoreLogOn t testPort $ \t1 -> do
-        res <- withSmpServerConfigOn t (cfg :: ServerConfig) {storeLogFile = Just testStoreLogFile2} testPort2 $ \t2 ->
+        res <- withSmpServerConfigOn t (cfg :: ServerConfig) {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2} testPort2 $ \t2 ->
           runRight a `finally` killThread t2
         killThread t1
         pure res

tests/AgentTests/ServerChoice.hs

@@ -0,0 +1,85 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE OverloadedLists #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module AgentTests.ServerChoice where
+
+import AgentTests.FunctionalAPITests
+import Control.Monad.IO.Class
+import Data.List.NonEmpty (NonEmpty (..))
+import qualified Data.Map.Strict as M
+import SMPAgentClient
+import Simplex.Messaging.Agent (withAgentEnv)
+import Simplex.Messaging.Agent.Client hiding (userServers)
+import Simplex.Messaging.Agent.Env.SQLite
+import Simplex.Messaging.Client (defaultNetworkConfig)
+import Simplex.Messaging.Protocol
+import Test.Hspec
+import Test.QuickCheck
+import XFTPClient (testXFTPServer)
+
+serverChoiceTests :: Spec
+serverChoiceTests = do
+  describe "Server operators" $ do
+    it "should choose server of different operator" $ ioProperty $ testChooseDifferentOperator
+
+operatorSimpleX :: Maybe OperatorId
+operatorSimpleX = Just 1
+
+operator2 :: Maybe OperatorId
+operator2 = Just 2
+
+testOp1Srv1 :: ProtoServerWithAuth 'PSMP
+testOp1Srv1 = "smp://LcJU@test1.simplex.im"
+
+testOp1Srv2 :: ProtoServerWithAuth 'PSMP
+testOp1Srv2 = "smp://LcJU@test2.simplex.im"
+
+testOp2Srv1 :: ProtoServerWithAuth 'PSMP
+testOp2Srv1 = "smp://LcJU@srv1.example.com"
+
+testOp2Srv2 :: ProtoServerWithAuth 'PSMP
+testOp2Srv2 = "smp://LcJU@srv2.example.com"
+
+testSMPServers :: NonEmpty (ServerCfg 'PSMP)
+testSMPServers =
+  [ presetServerCfg True allRoles operatorSimpleX testOp1Srv1,
+    presetServerCfg True allRoles operatorSimpleX testOp1Srv2,
+    presetServerCfg True proxyOnly operator2 testOp2Srv1,
+    presetServerCfg True proxyOnly operator2 testOp2Srv2
+  ]
+
+storageOnly :: ServerRoles
+storageOnly = ServerRoles {storage = True, proxy = False}
+
+proxyOnly :: ServerRoles
+proxyOnly = ServerRoles {storage = False, proxy = True}
+
+initServers :: InitialAgentServers
+initServers =
+  InitialAgentServers
+    { smp = M.fromList [(1, testSMPServers)],
+      ntf = [testNtfServer],
+      xftp = userServers [testXFTPServer],
+      netCfg = defaultNetworkConfig
+    }
+
+testChooseDifferentOperator :: IO ()
+testChooseDifferentOperator = do
+  c <- getSMPAgentClient' 1 agentCfg initServers testDB
+  runRight_ $ do
+    -- chooses the only operator with storage role
+    srv1 <- withAgentEnv c $ getNextServer c 1 storageSrvs []
+    liftIO $ srv1 == testOp1Srv1 || srv1 == testOp1Srv2 `shouldBe` True
+    -- chooses another server for storage
+    srv2 <- withAgentEnv c $ getNextServer c 1 storageSrvs [protoServer testOp1Srv1]
+    liftIO $ srv2 `shouldBe` testOp1Srv2
+    -- chooses another operator for proxy
+    srv3 <- withAgentEnv c $ getNextServer c 1 proxySrvs [protoServer srv1]
+    liftIO $ srv3 == testOp2Srv1 || srv3 == testOp2Srv2 `shouldBe` True
+    -- chooses another operator for proxy
+    srv3' <- withAgentEnv c $ getNextServer c 1 proxySrvs [protoServer testOp1Srv1, protoServer testOp1Srv2]
+    liftIO $ srv3' == testOp2Srv1 || srv3' == testOp2Srv2 `shouldBe` True
+    -- chooses any other server
+    srv4 <- withAgentEnv c $ getNextServer c 1 proxySrvs [protoServer testOp1Srv1, protoServer testOp2Srv1]
+    liftIO $ srv4 == testOp1Srv2 || srv4 == testOp2Srv2 `shouldBe` True

tests/CoreTests/MsgStoreTests.hs

@@ -0,0 +1,414 @@
+{-# LANGUAGE DataKinds #-}
+{-# LANGUAGE DuplicateRecordFields #-}
+{-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE GADTs #-}
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE PatternSynonyms #-}
+{-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE StandaloneDeriving #-}
+{-# LANGUAGE TypeApplications #-}
+{-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
+{-# OPTIONS_GHC -Wno-orphans #-}
+
+module CoreTests.MsgStoreTests where
+
+import AgentTests.FunctionalAPITests (runRight, runRight_)
+import Control.Concurrent.STM
+import Control.Exception (bracket)
+import Control.Monad
+import Control.Monad.IO.Class
+import Control.Monad.Trans.Except
+import Crypto.Random (ChaChaDRG)
+import Data.ByteString.Char8 (ByteString)
+import qualified Data.ByteString.Char8 as B
+import qualified Data.ByteString.Base64.URL as B64
+import Data.Maybe (fromJust)
+import Data.Time.Clock.System (getSystemTime)
+import Simplex.Messaging.Crypto (pattern MaxLenBS)
+import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Protocol (EntityId (..), Message (..), RecipientId, SParty (..), noMsgFlags)
+import Simplex.Messaging.Server (MessageStats (..), exportMessages, importMessages, printMessageStats)
+import Simplex.Messaging.Server.Env.STM (journalMsgStoreDepth, readWriteQueueStore)
+import Simplex.Messaging.Server.MsgStore.Journal
+import Simplex.Messaging.Server.MsgStore.STM
+import Simplex.Messaging.Server.MsgStore.Types
+import Simplex.Messaging.Server.QueueStore
+import Simplex.Messaging.Server.QueueStore.STM
+import Simplex.Messaging.Server.StoreLog (closeStoreLog, logCreateQueue)
+import SMPClient (testStoreLogFile, testStoreMsgsDir, testStoreMsgsDir2, testStoreMsgsFile, testStoreMsgsFile2)
+import System.Directory (copyFile, createDirectoryIfMissing, listDirectory, removeFile, renameFile)
+import System.FilePath ((</>))
+import System.IO (IOMode (..), hClose, withFile)
+import Test.Hspec
+
+msgStoreTests :: Spec
+msgStoreTests = do
+  around (withMsgStore testSMTStoreConfig) $ describe "STM message store" someMsgStoreTests
+  around (withMsgStore testJournalStoreCfg) $ describe "Journal message store" $ do
+    someMsgStoreTests
+    it "should export and import journal store" testExportImportStore
+    describe "queue state" $ do
+      it "should restore queue state from the last line" testQueueState
+      it "should recover when message is written and state is not" testMessageState
+    describe "missing files" $ do
+      it "should create read file when missing" testReadFileMissing
+      it "should switch to write file when read file missing" testReadFileMissingSwitch
+      it "should create write file when missing" testWriteFileMissing
+      it "should create read file when read and write files are missing" testReadAndWriteFilesMissing
+  where
+    someMsgStoreTests :: STMQueueStore s => SpecWith s
+    someMsgStoreTests = do
+      it "should get queue and store/read messages" testGetQueue
+      it "should not fail on EOF when changing read journal" testChangeReadJournal
+
+withMsgStore :: STMQueueStore s => MsgStoreConfig s -> (s -> IO ()) -> IO ()
+withMsgStore cfg = bracket (newMsgStore cfg) closeMsgStore
+
+testSMTStoreConfig :: STMStoreConfig
+testSMTStoreConfig = STMStoreConfig {storePath = Nothing, quota = 3}
+
+testJournalStoreCfg :: JournalStoreConfig
+testJournalStoreCfg =
+  JournalStoreConfig
+    { storePath = testStoreMsgsDir,
+      pathParts = journalMsgStoreDepth,
+      quota = 3,
+      maxMsgCount = 4,
+      maxStateLines = 2,
+      stateTailSize = 256,
+      idleInterval = 21600
+    }
+
+mkMessage :: MonadIO m => ByteString -> m Message
+mkMessage body = liftIO $ do
+  g <- C.newRandom
+  msgTs <- getSystemTime
+  msgId <- atomically $ C.randomBytes 24 g
+  pure Message {msgId, msgTs, msgFlags = noMsgFlags, msgBody = C.unsafeMaxLenBS body}
+
+pattern Msg :: ByteString -> Maybe Message
+pattern Msg s <- Just Message {msgBody = MaxLenBS s}
+
+deriving instance Eq MsgQueueState
+
+deriving instance Eq (JournalState t)
+
+deriving instance Eq (SJournalType t)
+
+testNewQueueRec :: TVar ChaChaDRG -> Bool -> IO (RecipientId, QueueRec)
+testNewQueueRec g sndSecure = do
+  rId <- atomically $ EntityId <$> C.randomBytes 24 g
+  senderId <- atomically $ EntityId <$> C.randomBytes 24 g
+  (recipientKey, _) <- atomically $ C.generateAuthKeyPair C.SX25519 g
+  (k, pk) <- atomically $ C.generateKeyPair @'C.X25519 g
+  let qr =
+        QueueRec
+          { recipientId = rId,
+            recipientKey,
+            rcvDhSecret = C.dh' k pk,
+            senderId,
+            senderKey = Nothing,
+            sndSecure,
+            notifier = Nothing,
+            status = QueueActive,
+            updatedAt = Nothing
+          }
+  pure (rId, qr)
+
+testGetQueue :: STMQueueStore s => s -> IO ()
+testGetQueue ms = do
+  g <- C.newRandom
+  (rId, qr) <- testNewQueueRec g True
+  runRight_ $ do
+    q <- ExceptT $ addQueue ms qr
+    let write s = writeMsg ms rId q True =<< mkMessage s
+    Just (Message {msgId = mId1}, True) <- write "message 1"
+    Just (Message {msgId = mId2}, False) <- write "message 2"
+    Just (Message {msgId = mId3}, False) <- write "message 3"
+    Msg "message 1" <- tryPeekMsg ms rId q
+    Msg "message 1" <- tryPeekMsg ms rId q
+    Nothing <- tryDelMsg ms rId q mId2
+    Msg "message 1" <- tryDelMsg ms rId q mId1
+    Nothing <- tryDelMsg ms rId q mId1
+    Msg "message 2" <- tryPeekMsg ms rId q
+    Nothing <- tryDelMsg ms rId q mId1
+    (Nothing, Msg "message 2") <- tryDelPeekMsg ms rId q mId1
+    (Msg "message 2", Msg "message 3") <- tryDelPeekMsg ms rId q mId2
+    (Nothing, Msg "message 3") <- tryDelPeekMsg ms rId q mId2
+    Msg "message 3" <- tryPeekMsg ms rId q
+    (Msg "message 3", Nothing) <- tryDelPeekMsg ms rId q mId3
+    Nothing <- tryDelMsg ms rId q mId2
+    Nothing <- tryDelMsg ms rId q mId3
+    Nothing <- tryPeekMsg ms rId q
+    Just (Message {msgId = mId4}, True) <- write "message 4"
+    Msg "message 4" <- tryPeekMsg ms rId q
+    Just (Message {msgId = mId5}, False) <- write "message 5"
+    (Nothing, Msg "message 4") <- tryDelPeekMsg ms rId q mId3
+    (Msg "message 4", Msg "message 5") <- tryDelPeekMsg ms rId q mId4
+    Just (Message {msgId = mId6}, False) <- write "message 6"
+    Just (Message {msgId = mId7}, False) <- write "message 7"
+    Nothing <- write "message 8"
+    Msg "message 5" <- tryPeekMsg ms rId q
+    (Nothing, Msg "message 5") <- tryDelPeekMsg ms rId q mId4
+    (Msg "message 5", Msg "message 6") <- tryDelPeekMsg ms rId q mId5
+    (Msg "message 6", Msg "message 7") <- tryDelPeekMsg ms rId q mId6
+    (Msg "message 7", Just MessageQuota {msgId = mId8}) <- tryDelPeekMsg ms rId q mId7
+    (Just MessageQuota {}, Nothing) <- tryDelPeekMsg ms rId q mId8
+    (Nothing, Nothing) <- tryDelPeekMsg ms rId q mId8
+    void $ ExceptT $ deleteQueue ms rId q
+
+testChangeReadJournal :: STMQueueStore s => s -> IO ()
+testChangeReadJournal ms = do
+  g <- C.newRandom
+  (rId, qr) <- testNewQueueRec g True
+  runRight_ $ do
+    q <- ExceptT $ addQueue ms qr
+    let write s = writeMsg ms rId q True =<< mkMessage s
+    Just (Message {msgId = mId1}, True) <- write "message 1"
+    (Msg "message 1", Nothing) <- tryDelPeekMsg ms rId q mId1
+    Just (Message {msgId = mId2}, True) <- write "message 2"
+    (Msg "message 2", Nothing) <- tryDelPeekMsg ms rId q mId2
+    Just (Message {msgId = mId3}, True) <- write "message 3"
+    (Msg "message 3", Nothing) <- tryDelPeekMsg ms rId q mId3
+    Just (Message {msgId = mId4}, True) <- write "message 4"
+    (Msg "message 4", Nothing) <- tryDelPeekMsg ms rId q mId4
+    Just (Message {msgId = mId5}, True) <- write "message 5"
+    (Msg "message 5", Nothing) <- tryDelPeekMsg ms rId q mId5
+    void $ ExceptT $ deleteQueue ms rId q
+
+testExportImportStore :: JournalMsgStore -> IO ()
+testExportImportStore ms = do
+  g <- C.newRandom
+  (rId1, qr1) <- testNewQueueRec g True
+  (rId2, qr2) <- testNewQueueRec g True
+  sl <- readWriteQueueStore testStoreLogFile ms
+  runRight_ $ do
+    let write rId q s = writeMsg ms rId q True =<< mkMessage s
+    q1 <- ExceptT $ addQueue ms qr1
+    liftIO $ logCreateQueue sl qr1
+    Just (Message {}, True) <- write rId1 q1 "message 1"
+    Just (Message {}, False) <- write rId1 q1 "message 2"
+    q2 <- ExceptT $ addQueue ms qr2
+    liftIO $ logCreateQueue sl qr2
+    Just (Message {msgId = mId3}, True) <- write rId2 q2 "message 3"
+    Just (Message {msgId = mId4}, False) <- write rId2 q2 "message 4"
+    (Msg "message 3", Msg "message 4") <- tryDelPeekMsg ms rId2 q2 mId3
+    (Msg "message 4", Nothing) <- tryDelPeekMsg ms rId2 q2 mId4
+    Just (Message {}, True) <- write rId2 q2 "message 5"
+    Just (Message {}, False) <- write rId2 q2 "message 6"
+    Just (Message {}, False) <- write rId2 q2 "message 7"
+    Nothing <- write rId2 q2 "message 8"
+    pure ()
+  length <$> listDirectory (msgQueueDirectory ms rId1) `shouldReturn` 2
+  length <$> listDirectory (msgQueueDirectory ms rId2) `shouldReturn` 3
+  exportMessages False ms testStoreMsgsFile False
+  renameFile testStoreMsgsFile (testStoreMsgsFile <> ".copy")
+  closeMsgStore ms
+  closeStoreLog sl
+  exportMessages False ms testStoreMsgsFile False
+  (B.readFile testStoreMsgsFile `shouldReturn`) =<< B.readFile (testStoreMsgsFile <> ".copy")
+  let cfg = (testJournalStoreCfg :: JournalStoreConfig) {storePath = testStoreMsgsDir2}
+  ms' <- newMsgStore cfg
+  readWriteQueueStore testStoreLogFile ms' >>= closeStoreLog
+  stats@MessageStats {storedMsgsCount = 5, expiredMsgsCount = 0, storedQueues = 2} <-
+    importMessages False ms' testStoreMsgsFile Nothing
+  printMessageStats "Messages" stats
+  length <$> listDirectory (msgQueueDirectory ms rId1) `shouldReturn` 2
+  length <$> listDirectory (msgQueueDirectory ms rId2) `shouldReturn` 4 -- state file is backed up, 2 message files
+  exportMessages False ms' testStoreMsgsFile2 False
+  (B.readFile testStoreMsgsFile2 `shouldReturn`) =<< B.readFile (testStoreMsgsFile <> ".bak")
+  stmStore <- newMsgStore testSMTStoreConfig
+  readWriteQueueStore testStoreLogFile stmStore >>= closeStoreLog
+  MessageStats {storedMsgsCount = 5, expiredMsgsCount = 0, storedQueues = 2} <-
+    importMessages False stmStore testStoreMsgsFile2 Nothing
+  exportMessages False stmStore testStoreMsgsFile False
+  (B.sort <$> B.readFile testStoreMsgsFile `shouldReturn`) =<< (B.sort <$> B.readFile (testStoreMsgsFile2 <> ".bak"))
+
+testQueueState :: JournalMsgStore -> IO ()
+testQueueState ms = do
+  g <- C.newRandom
+  rId <- EntityId <$> atomically (C.randomBytes 24 g)
+  let dir = msgQueueDirectory ms rId
+      statePath = msgQueueStatePath dir $ B.unpack (B64.encode $ unEntityId rId)
+  createDirectoryIfMissing True dir
+  state <- newMsgQueueState <$> newJournalId (random ms)
+  withFile statePath WriteMode (`appendState` state)
+  length . lines <$> readFile statePath `shouldReturn` 1
+  readQueueState statePath `shouldReturn` state
+  length <$> listDirectory dir `shouldReturn` 1 -- no backup
+
+  let state1 =
+        state
+          { size = 1,
+            readState = (readState state) {msgCount = 1, byteCount = 100},
+            writeState = (writeState state) {msgPos = 1, msgCount = 1, bytePos = 100, byteCount = 100}
+          }
+  withFile statePath AppendMode (`appendState` state1)
+  length . lines <$> readFile statePath `shouldReturn` 2
+  readQueueState statePath `shouldReturn` state1
+  length <$> listDirectory dir `shouldReturn` 1 -- no backup
+
+  let state2 =
+        state
+          { size = 2,
+            readState = (readState state) {msgCount = 2, byteCount = 200},
+            writeState = (writeState state) {msgPos = 2, msgCount = 2, bytePos = 200, byteCount = 200}
+          }
+  withFile statePath AppendMode (`appendState` state2)
+  length . lines <$> readFile statePath `shouldReturn` 3
+  copyFile statePath (statePath <> ".2")
+  readQueueState statePath `shouldReturn` state2
+  length <$> listDirectory dir `shouldReturn` 3 -- new state, copy + backup
+  length . lines <$> readFile statePath `shouldReturn` 1
+
+  -- corrupt the only line
+  corruptFile statePath
+  newState <- readQueueState statePath
+  newState `shouldBe` newMsgQueueState (journalId $ writeState newState)
+
+  -- corrupt the last line
+  renameFile (statePath <> ".2") statePath
+  removeOtherFiles dir statePath
+  length . lines <$> readFile statePath `shouldReturn` 3
+  corruptFile statePath
+  readQueueState statePath `shouldReturn` state1
+  length <$> listDirectory dir `shouldReturn` 2
+  length . lines <$> readFile statePath `shouldReturn` 1
+  where
+    readQueueState statePath = do
+      (state, h) <- readWriteQueueState ms statePath
+      hClose h
+      pure state
+    corruptFile f = do
+      s <- readFile f
+      removeFile f
+      writeFile f $ take (length s - 4) s
+    removeOtherFiles dir keep = do
+      names <- listDirectory dir
+      forM_ names $ \name ->
+        let f = dir </> name
+         in unless (f == keep) $ removeFile f
+
+testMessageState :: JournalMsgStore -> IO ()
+testMessageState ms = do
+  g <- C.newRandom
+  (rId, qr) <- testNewQueueRec g True
+  let dir = msgQueueDirectory ms rId
+      statePath = msgQueueStatePath dir $ B.unpack (B64.encode $ unEntityId rId)
+      write q s = writeMsg ms rId q True =<< mkMessage s
+
+  mId1 <- runRight $ do
+    q <- ExceptT $ addQueue ms qr
+    Just (Message {msgId = mId1}, True) <- write q "message 1"
+    Just (Message {}, False) <- write q "message 2"
+    liftIO $ closeMsgQueue q
+    pure mId1
+
+  ls <- B.lines <$> B.readFile statePath
+  B.writeFile statePath $ B.unlines $ take (length ls - 1) ls
+
+  runRight_ $ do
+    q <- ExceptT $ getQueue ms SRecipient rId
+    Just (Message {msgId = mId3}, False) <- write q "message 3"
+    (Msg "message 1", Msg "message 3") <- tryDelPeekMsg ms rId q mId1
+    (Msg "message 3", Nothing) <- tryDelPeekMsg ms rId q mId3
+    liftIO $ closeMsgQueue q
+
+testReadFileMissing :: JournalMsgStore -> IO ()
+testReadFileMissing ms = do
+  g <- C.newRandom
+  (rId, qr) <- testNewQueueRec g True
+  let write q s = writeMsg ms rId q True =<< mkMessage s
+  q <- runRight $ do
+    q <- ExceptT $ addQueue ms qr
+    Just (Message {}, True) <- write q "message 1"
+    Msg "message 1" <- tryPeekMsg ms rId q
+    pure q
+
+  mq <- fromJust <$> readTVarIO (msgQueue_' q)
+  MsgQueueState {readState = rs} <- readTVarIO $ state mq
+  closeMsgStore ms
+  let path = journalFilePath (queueDirectory $ queue mq) $ journalId rs
+  removeFile path
+
+  runRight_ $ do
+    q' <- ExceptT $ getQueue ms SRecipient rId
+    Nothing <- tryPeekMsg ms rId q'
+    Just (Message {}, True) <- write q' "message 2"
+    Msg "message 2" <- tryPeekMsg ms rId q'
+    pure ()
+
+testReadFileMissingSwitch :: JournalMsgStore -> IO ()
+testReadFileMissingSwitch ms = do
+  g <- C.newRandom
+  (rId, qr) <- testNewQueueRec g True
+  q <- writeMessages ms rId qr
+
+  mq <- fromJust <$> readTVarIO (msgQueue_' q)
+  MsgQueueState {readState = rs} <- readTVarIO $ state mq
+  closeMsgStore ms
+  let path = journalFilePath (queueDirectory $ queue mq) $ journalId rs
+  removeFile path
+
+  runRight_ $ do
+    q' <- ExceptT $ getQueue ms SRecipient rId
+    Just (Message {}, False) <- writeMsg ms rId q' True =<< mkMessage "message 6"
+    Msg "message 5" <- tryPeekMsg ms rId q'
+    pure ()
+
+testWriteFileMissing :: JournalMsgStore -> IO ()
+testWriteFileMissing ms = do
+  g <- C.newRandom
+  (rId, qr) <- testNewQueueRec g True
+  q <- writeMessages ms rId qr
+
+  mq <- fromJust <$> readTVarIO (msgQueue_' q)
+  MsgQueueState {writeState = ws} <- readTVarIO $ state mq
+  closeMsgStore ms
+  let path = journalFilePath (queueDirectory $ queue mq) $ journalId ws
+  print path
+  removeFile path
+
+  runRight_ $ do
+    q' <- ExceptT $ getQueue ms SRecipient rId
+    Just Message {msgId = mId3} <- tryPeekMsg ms rId q'
+    (Msg "message 3", Msg "message 4") <- tryDelPeekMsg ms rId q' mId3
+    Just Message {msgId = mId4} <- tryPeekMsg ms rId q'
+    (Msg "message 4", Nothing) <- tryDelPeekMsg ms rId q' mId4
+    Just (Message {}, True) <- writeMsg ms rId q' True =<< mkMessage "message 6"
+    Msg "message 6" <- tryPeekMsg ms rId q'
+    pure ()
+
+testReadAndWriteFilesMissing :: JournalMsgStore -> IO ()
+testReadAndWriteFilesMissing ms = do
+  g <- C.newRandom
+  (rId, qr) <- testNewQueueRec g True
+  q <- writeMessages ms rId qr
+
+  mq <- fromJust <$> readTVarIO (msgQueue_' q)
+  MsgQueueState {readState = rs, writeState = ws} <- readTVarIO $ state mq
+  closeMsgStore ms
+  removeFile $ journalFilePath (queueDirectory $ queue mq) $ journalId rs
+  removeFile $ journalFilePath (queueDirectory $ queue mq) $ journalId ws
+
+  runRight_ $ do
+    q' <- ExceptT $ getQueue ms SRecipient rId
+    Nothing <- tryPeekMsg ms rId q'
+    Just (Message {}, True) <- writeMsg ms rId q' True =<< mkMessage "message 6"
+    Msg "message 6" <- tryPeekMsg ms rId q'
+    pure ()
+
+writeMessages :: JournalMsgStore -> RecipientId -> QueueRec -> IO JournalQueue
+writeMessages ms rId qr = runRight $ do
+  q <- ExceptT $ addQueue ms qr
+  let write s = writeMsg ms rId q True =<< mkMessage s
+  Just (Message {msgId = mId1}, True) <- write "message 1"
+  Just (Message {msgId = mId2}, False) <- write "message 2"
+  Just (Message {}, False) <- write "message 3"
+  (Msg "message 1", Msg "message 2") <- tryDelPeekMsg ms rId q mId1
+  (Msg "message 2", Msg "message 3") <- tryDelPeekMsg ms rId q mId2
+  Just (Message {}, False) <- write "message 4"
+  Just (Message {}, False) <- write "message 5"
+  pure q

tests/CoreTests/RetryIntervalTests.hs

@@ -66,7 +66,7 @@ testRetryIntervalSameMode =
 
 testRetryIntervalSwitchMode :: Spec
 testRetryIntervalSwitchMode =
-  it "should increase elapased time and interval when the mode stays the same" $ do
+  it "should increase elapased time and interval when the mode switches" $ do
     lock <- newEmptyTMVarIO
     intervals <- newTVarIO []
     reportedIntervals <- newTVarIO []

tests/CoreTests/StoreLogTests.hs

@@ -16,32 +16,17 @@ import Data.Either (partitionEithers)
 import qualified Data.Map.Strict as M
 import SMPClient
 import AgentTests.SQLiteTests
+import CoreTests.MsgStoreTests
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
-import Simplex.Messaging.Server.StoreLog
+import Simplex.Messaging.Server.Env.STM (readWriteQueueStore)
+import Simplex.Messaging.Server.MsgStore.Journal
+import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.QueueStore
-import Simplex.Messaging.Server.QueueStore.STM (QueueStore (..), newQueueStore)
+import Simplex.Messaging.Server.StoreLog
 import Test.Hspec
 
-testNewQueueRec :: TVar ChaChaDRG -> Bool -> IO QueueRec
-testNewQueueRec g sndSecure = do
-  recipientId <- atomically $ EntityId <$> C.randomBytes 24 g
-  senderId <- atomically $ EntityId <$> C.randomBytes 24 g
-  (recipientKey, _) <- atomically $ C.generateAuthKeyPair C.SX25519 g
-  (k, pk) <- atomically $ C.generateKeyPair @'C.X25519 g
-  pure QueueRec
-    { recipientId,
-      recipientKey,
-      rcvDhSecret = C.dh' k pk,
-      senderId,
-      senderKey = Nothing,
-      sndSecure,
-      notifier = Nothing,
-      status = QueueActive,
-      updatedAt = Nothing
-    }
-
 testNtfCreds :: TVar ChaChaDRG -> IO NtfCreds
 testNtfCreds g = do
   (notifierKey, _) <- atomically $ C.generateAuthKeyPair C.SX25519 g
@@ -66,7 +51,7 @@ deriving instance Eq NtfCreds
 storeLogTests :: Spec
 storeLogTests =
   forM_ [False, True] $ \sndSecure -> do
-    (qr, ntfCreds, date) <- runIO $ do
+    ((rId, qr), ntfCreds, date) <- runIO $ do
       g <- C.newRandom
       (,,) <$> testNewQueueRec g sndSecure <*> testNtfCreds g <*> getSystemDate
     testSMPStoreLog ("SMP server store log, sndSecure = " <> show sndSecure)
@@ -74,37 +59,37 @@ storeLogTests =
           { name = "create new queue",
             saved = [CreateQueue qr],
             compacted = [CreateQueue qr],
-            state = M.fromList [(recipientId qr, qr)]
+            state = M.fromList [(rId, qr)]
           },
         SLTC
           { name = "secure queue",
-            saved = [CreateQueue qr, SecureQueue (recipientId qr) testPublicAuthKey],
+            saved = [CreateQueue qr, SecureQueue rId testPublicAuthKey],
             compacted = [CreateQueue qr {senderKey = Just testPublicAuthKey}],
-            state = M.fromList [(recipientId qr, qr {senderKey = Just testPublicAuthKey})]
+            state = M.fromList [(rId, qr {senderKey = Just testPublicAuthKey})]
           },          
         SLTC
           { name = "create and delete queue",
-            saved = [CreateQueue qr, DeleteQueue $ recipientId qr],
+            saved = [CreateQueue qr, DeleteQueue rId],
             compacted = [],
             state = M.fromList []
           },
         SLTC
           { name = "create queue and add notifier",
-            saved = [CreateQueue qr, AddNotifier (recipientId qr) ntfCreds],
+            saved = [CreateQueue qr, AddNotifier rId ntfCreds],
             compacted = [CreateQueue $ qr {notifier = Just ntfCreds}],
-            state = M.fromList [(recipientId qr, qr {notifier = Just ntfCreds})]
+            state = M.fromList [(rId, qr {notifier = Just ntfCreds})]
           },
         SLTC
           { name = "delete notifier",
-            saved = [CreateQueue qr, AddNotifier (recipientId qr) ntfCreds, DeleteNotifier (recipientId qr)],
+            saved = [CreateQueue qr, AddNotifier rId ntfCreds, DeleteNotifier rId],
             compacted = [CreateQueue qr],
-            state = M.fromList [(recipientId qr, qr)]
+            state = M.fromList [(rId, qr)]
           },
         SLTC
           { name = "update time",
-            saved = [CreateQueue qr, UpdateTime (recipientId qr) date],
+            saved = [CreateQueue qr, UpdateTime rId date],
             compacted = [CreateQueue qr {updatedAt = Just date}],
-            state = M.fromList [(recipientId qr, qr {updatedAt = Just date})]
+            state = M.fromList [(rId, qr {updatedAt = Just date})]
           }          
       ]
 
@@ -117,11 +102,11 @@ testSMPStoreLog testSuite tests =
     replicateM_ 3 $ testReadWrite t
   where
     testReadWrite SLTC {compacted, state} = do
-      st <- newQueueStore
+      st <- newMsgStore testJournalStoreCfg
       l <- readWriteQueueStore testStoreLogFile st
       storeState st `shouldReturn` state
       closeStoreLog l
       ([], compacted') <- partitionEithers . map strDecode . B.lines <$> B.readFile testStoreLogFile
       compacted' `shouldBe` compacted
-    storeState :: QueueStore -> IO (M.Map RecipientId QueueRec)
-    storeState st = readTVarIO (queues st) >>= mapM readTVarIO
+    storeState :: JournalMsgStore -> IO (M.Map RecipientId QueueRec)
+    storeState st = M.mapMaybe id <$> (readTVarIO (queues st) >>= mapM (readTVarIO . queueRec'))

tests/NtfServerTests.hs

@@ -104,7 +104,7 @@ testNotificationSubscription (ATransport t) =
     (dhPub, dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
     let tkn = DeviceToken PPApnsTest "abcd"
     withAPNSMockServer $ \apns ->
-      smpTest2 t $ \rh sh ->
+      smpTest2' t $ \rh sh ->
         ntfTest t $ \nh -> do
           -- create queue
           (sId, rId, rKey, rcvDhSecret) <- createAndSecureQueue rh sPub

tests/SMPAgentClient.hs

@@ -38,7 +38,7 @@ testSMPServer :: SMPServer
 testSMPServer = "smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5001"
 
 testSMPServer2 :: SMPServer
-testSMPServer2 = "smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:5002"
+testSMPServer2 = "smp://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@127.0.0.1:5002"
 
 testNtfServer :: NtfServer
 testNtfServer = "ntf://LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI=@localhost:6001"
@@ -92,7 +92,7 @@ userServers :: NonEmpty (ProtocolServer p) -> Map UserId (NonEmpty (ServerCfg p)
 userServers = userServers' . L.map noAuthSrv
 
 userServers' :: NonEmpty (ProtoServerWithAuth p) -> Map UserId (NonEmpty (ServerCfg p))
-userServers' srvs = M.fromList [(1, L.map (presetServerCfg True) srvs)]
+userServers' srvs = M.fromList [(1, L.map (presetServerCfg True (ServerRoles True True) (Just 1)) srvs)]
 
 noAuthSrvCfg :: ProtocolServer p -> ServerCfg p
-noAuthSrvCfg = presetServerCfg True . noAuthSrv
+noAuthSrvCfg = presetServerCfg True (ServerRoles True True) (Just 1) . noAuthSrv

tests/SMPClient.hs

@@ -1,6 +1,7 @@
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
+{-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE NumericUnderscores #-}
@@ -23,6 +24,7 @@ import Simplex.Messaging.Encoding
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server (runSMPServerBlocking)
 import Simplex.Messaging.Server.Env.STM
+import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..))
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client
 import qualified Simplex.Messaging.Transport.Client as Client
@@ -41,6 +43,9 @@ import Util
 testHost :: NonEmpty TransportHost
 testHost = "localhost"
 
+testHost2 :: NonEmpty TransportHost
+testHost2 = "127.0.0.1"
+
 testPort :: ServiceName
 testPort = "5001"
 
@@ -62,6 +67,12 @@ testStoreMsgsFile = "tests/tmp/smp-server-messages.log"
 testStoreMsgsFile2 :: FilePath
 testStoreMsgsFile2 = "tests/tmp/smp-server-messages.log.2"
 
+testStoreMsgsDir :: FilePath
+testStoreMsgsDir = "tests/tmp/messages"
+
+testStoreMsgsDir2 :: FilePath
+testStoreMsgsDir2 = "tests/tmp/messages.2"
+
 testStoreNtfsFile :: FilePath
 testStoreNtfsFile = "tests/tmp/smp-server-ntfs.log"
 
@@ -100,22 +111,32 @@ testSMPClient_ host port vr client = do
       | otherwise = Nothing
 
 cfg :: ServerConfig
-cfg =
+cfg = cfgMS (AMSType SMSJournal)
+
+cfgMS :: AMSType -> ServerConfig
+cfgMS msType =
   ServerConfig
     { transports = [],
       smpHandshakeTimeout = 60000000,
       tbqSize = 1,
+      msgStoreType = msType,
       msgQueueQuota = 4,
+      maxJournalMsgCount = 5,
+      maxJournalStateLines = 2,
       queueIdBytes = 24,
       msgIdBytes = 24,
-      storeLogFile = Nothing,
-      storeMsgsFile = Nothing,
+      storeLogFile = Just testStoreLogFile,
+      storeMsgsFile = Just $ case msType of
+        AMSType SMSJournal -> testStoreMsgsDir
+        AMSType SMSMemory -> testStoreMsgsFile,
       storeNtfsFile = Nothing,
       allowNewQueues = True,
       newQueueBasicAuth = Nothing,
       controlPortUserAuth = Nothing,
       controlPortAdminAuth = Nothing,
       messageExpiration = Just defaultMessageExpiration,
+      expireMessagesOnStart = True,
+      idleQueueInterval = defaultIdleQueueInterval,
       notificationExpiration = defaultNtfExpiration,
       inactiveClientExpiration = Just defaultInactiveClientExpiration,
       logStatsInterval = Nothing,
@@ -168,18 +189,17 @@ proxyVRangeV8 :: VersionRangeSMP
 proxyVRangeV8 = mkVersionRange batchCmdsSMPVersion sendingProxySMPVersion
 
 withSmpServerStoreMsgLogOn :: HasCallStack => ATransport -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
-withSmpServerStoreMsgLogOn t =
-  withSmpServerConfigOn
-    t
-    cfg
-      { storeLogFile = Just testStoreLogFile,
-        storeMsgsFile = Just testStoreMsgsFile,
-        storeNtfsFile = Just testStoreNtfsFile,
-        serverStatsBackupFile = Just testServerStatsBackupFile
-      }
+withSmpServerStoreMsgLogOn = (`withSmpServerStoreMsgLogOnMS` AMSType SMSJournal)
+
+withSmpServerStoreMsgLogOnMS :: HasCallStack => ATransport -> AMSType -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
+withSmpServerStoreMsgLogOnMS t msType =
+  withSmpServerConfigOn t (cfgMS msType) {storeNtfsFile = Just testStoreNtfsFile, serverStatsBackupFile = Just testServerStatsBackupFile}
 
 withSmpServerStoreLogOn :: HasCallStack => ATransport -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
-withSmpServerStoreLogOn t = withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile, serverStatsBackupFile = Just testServerStatsBackupFile}
+withSmpServerStoreLogOn = (`withSmpServerStoreLogOnMS` AMSType SMSJournal)
+
+withSmpServerStoreLogOnMS :: HasCallStack => ATransport -> AMSType -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
+withSmpServerStoreLogOnMS t msType = withSmpServerConfigOn t (cfgMS msType) {serverStatsBackupFile = Just testServerStatsBackupFile}
 
 withSmpServerConfigOn :: HasCallStack => ATransport -> ServerConfig -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
 withSmpServerConfigOn t cfg' port' =
@@ -212,11 +232,11 @@ withSmpServer t = withSmpServerOn t testPort
 withSmpServerProxy :: HasCallStack => ATransport -> IO a -> IO a
 withSmpServerProxy t = withSmpServerConfigOn t proxyCfg testPort . const
 
-runSmpTest :: forall c a. (HasCallStack, Transport c) => (HasCallStack => THandleSMP c 'TClient -> IO a) -> IO a
-runSmpTest test = withSmpServer (transport @c) $ testSMPClient test
+runSmpTest :: forall c a. (HasCallStack, Transport c) => AMSType -> (HasCallStack => THandleSMP c 'TClient -> IO a) -> IO a
+runSmpTest msType test = withSmpServerConfigOn (transport @c) (cfgMS msType) testPort $ \_ -> testSMPClient test
 
-runSmpTestN :: forall c a. (HasCallStack, Transport c) => Int -> (HasCallStack => [THandleSMP c 'TClient] -> IO a) -> IO a
-runSmpTestN = runSmpTestNCfg cfg supportedClientSMPRelayVRange
+runSmpTestN :: forall c a. (HasCallStack, Transport c) => AMSType -> Int -> (HasCallStack => [THandleSMP c 'TClient] -> IO a) -> IO a
+runSmpTestN msType = runSmpTestNCfg (cfgMS msType) supportedClientSMPRelayVRange
 
 runSmpTestNCfg :: forall c a. (HasCallStack, Transport c) => ServerConfig -> VersionRangeSMP -> Int -> (HasCallStack => [THandleSMP c 'TClient] -> IO a) -> IO a
 runSmpTestNCfg srvCfg clntVR nClients test = withSmpServerConfigOn (transport @c) srvCfg testPort $ \_ -> run nClients []
@@ -231,7 +251,7 @@ smpServerTest ::
   TProxy c ->
   (Maybe TransmissionAuth, ByteString, ByteString, smp) ->
   IO (Maybe TransmissionAuth, ByteString, ByteString, BrokerMsg)
-smpServerTest _ t = runSmpTest $ \h -> tPut' h t >> tGet' h
+smpServerTest _ t = runSmpTest (AMSType SMSJournal) $ \h -> tPut' h t >> tGet' h
   where
     tPut' :: THandleSMP c 'TClient -> (Maybe TransmissionAuth, ByteString, ByteString, smp) -> IO ()
     tPut' h@THandle {params = THandleParams {sessionId, implySessId}} (sig, corrId, queueId, smp) = do
@@ -242,14 +262,17 @@ smpServerTest _ t = runSmpTest $ \h -> tPut' h t >> tGet' h
       [(Nothing, _, (CorrId corrId, EntityId qId, Right cmd))] <- tGet h
       pure (Nothing, corrId, qId, cmd)
 
-smpTest :: (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandleSMP c 'TClient -> IO ()) -> Expectation
-smpTest _ test' = runSmpTest test' `shouldReturn` ()
+smpTest :: (HasCallStack, Transport c) => TProxy c -> AMSType -> (HasCallStack => THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest _ msType test' = runSmpTest msType test' `shouldReturn` ()
 
-smpTestN :: (HasCallStack, Transport c) => Int -> (HasCallStack => [THandleSMP c 'TClient] -> IO ()) -> Expectation
-smpTestN n test' = runSmpTestN n test' `shouldReturn` ()
+smpTestN :: (HasCallStack, Transport c) => AMSType -> Int -> (HasCallStack => [THandleSMP c 'TClient] -> IO ()) -> Expectation
+smpTestN msType n test' = runSmpTestN msType n test' `shouldReturn` ()
 
-smpTest2 :: forall c. (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
-smpTest2 = smpTest2Cfg cfg supportedClientSMPRelayVRange
+smpTest2' :: forall c. (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest2' = (`smpTest2` AMSType SMSJournal)
+
+smpTest2 :: forall c. (HasCallStack, Transport c) => TProxy c -> AMSType -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest2 t msType = smpTest2Cfg (cfgMS msType) supportedClientSMPRelayVRange t
 
 smpTest2Cfg :: forall c. (HasCallStack, Transport c) => ServerConfig -> VersionRangeSMP -> TProxy c -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
 smpTest2Cfg srvCfg clntVR _ test' = runSmpTestNCfg srvCfg clntVR 2 _test `shouldReturn` ()
@@ -258,15 +281,15 @@ smpTest2Cfg srvCfg clntVR _ test' = runSmpTestNCfg srvCfg clntVR 2 _test `should
     _test [h1, h2] = test' h1 h2
     _test _ = error "expected 2 handles"
 
-smpTest3 :: forall c. (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
-smpTest3 _ test' = smpTestN 3 _test
+smpTest3 :: forall c. (HasCallStack, Transport c) => TProxy c -> AMSType -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest3 _ msType test' = smpTestN msType 3 _test
   where
     _test :: HasCallStack => [THandleSMP c 'TClient] -> IO ()
     _test [h1, h2, h3] = test' h1 h2 h3
     _test _ = error "expected 3 handles"
 
-smpTest4 :: forall c. (HasCallStack, Transport c) => TProxy c -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
-smpTest4 _ test' = smpTestN 4 _test
+smpTest4 :: forall c. (HasCallStack, Transport c) => TProxy c -> AMSType -> (HasCallStack => THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO ()) -> Expectation
+smpTest4 _ msType test' = smpTestN msType 4 _test
   where
     _test :: HasCallStack => [THandleSMP c 'TClient] -> IO ()
     _test [h1, h2, h3, h4] = test' h1 h2 h3 h4

tests/SMPProxyTests.hs

@@ -61,7 +61,7 @@ smpProxyTests = do
     xit "batching proxy requests" todo
   describe "deliver message via SMP proxy" $ do
     let srv1 = SMPServer testHost testPort testKeyHash
-        srv2 = SMPServer testHost testPort2 testKeyHash
+        srv2 = SMPServer testHost2 testPort2 testKeyHash
     describe "client API" $ do
       let maxLen = maxMessageLength encryptedBlockSMPVersion
       describe "one server" $ do
@@ -135,14 +135,15 @@ smpProxyTests = do
         let deliver nAgents nMsgs = agentDeliverMessagesViaProxyConc (replicate nAgents [srv1]) (map bshow [1 :: Int .. nMsgs])
         it "25 agents, 300 pairs, 17 messages" . oneServer . withNumCapabilities 4 $ deliver 25 17
   where
-    oneServer = withSmpServerConfigOn (transport @TLS) proxyCfg {msgQueueQuota = 128} testPort . const
+    oneServer = withSmpServerConfigOn (transport @TLS) proxyCfg {msgQueueQuota = 128, maxJournalMsgCount = 256} testPort . const
     twoServers = twoServers_ proxyCfg proxyCfg
-    twoServersFirstProxy = twoServers_ proxyCfg cfgV8 {msgQueueQuota = 128}
-    twoServersMoreConc = twoServers_ proxyCfg {serverClientConcurrency = 128} cfgV8 {msgQueueQuota = 128}
-    twoServersNoConc = twoServers_ proxyCfg {serverClientConcurrency = 1} cfgV8 {msgQueueQuota = 128}
+    twoServersFirstProxy = twoServers_ proxyCfg cfgV8 {msgQueueQuota = 128, maxJournalMsgCount = 256}
+    twoServersMoreConc = twoServers_ proxyCfg {serverClientConcurrency = 128} cfgV8 {msgQueueQuota = 128, maxJournalMsgCount = 256}
+    twoServersNoConc = twoServers_ proxyCfg {serverClientConcurrency = 1} cfgV8 {msgQueueQuota = 128, maxJournalMsgCount = 256}
     twoServers_ cfg1 cfg2 runTest =
       withSmpServerConfigOn (transport @TLS) cfg1 testPort $ \_ ->
-        withSmpServerConfigOn (transport @TLS) cfg2 testPort2 $ const runTest
+        let cfg2' = cfg2 {storeLogFile = Just testStoreLogFile2, storeMsgsFile = Just testStoreMsgsDir2}
+         in withSmpServerConfigOn (transport @TLS) cfg2' testPort2 $ const runTest
 
 deliverMessageViaProxy :: (C.AlgorithmI a, C.AuthAlgorithm a) => SMPServer -> SMPServer -> C.SAlgorithm a -> ByteString -> ByteString -> IO ()
 deliverMessageViaProxy proxyServ relayServ alg msg msg' = deliverMessagesViaProxy proxyServ relayServ alg [msg] [msg']
@@ -315,7 +316,7 @@ agentViaProxyVersionError :: IO ()
 agentViaProxyVersionError =
   withAgent 1 agentCfg (servers [SMPServer testHost testPort testKeyHash]) testDB $ \alice -> do
     Left (A.BROKER _ (TRANSPORT TEVersion)) <-
-      withAgent 2 agentCfg (servers [SMPServer testHost testPort2 testKeyHash]) testDB2 $ \bob -> runExceptT $ do
+      withAgent 2 agentCfg (servers [SMPServer testHost2 testPort2 testKeyHash]) testDB2 $ \bob -> runExceptT $ do
         (_bobId, qInfo) <- A.createConnection alice 1 True SCMInvitation Nothing (CR.IKNoPQ PQSupportOn) SMSubscribe
         aliceId <- A.prepareConnectionToJoin bob 1 True qInfo PQSupportOn
         A.joinConnection bob 1 aliceId True qInfo "bob's connInfo" PQSupportOn SMSubscribe
@@ -379,11 +380,11 @@ agentViaProxyRetryOffline = do
             ackMessage alice bobId (baseId + 4) Nothing
   where
     withServer :: (ThreadId -> IO a) -> IO a
-    withServer = withServer_ testStoreLogFile testStoreMsgsFile testStoreNtfsFile testPort
+    withServer = withServer_ testStoreLogFile testStoreMsgsDir testStoreNtfsFile testPort
     withServer2 :: (ThreadId -> IO a) -> IO a
-    withServer2 = withServer_ testStoreLogFile2 testStoreMsgsFile2 testStoreNtfsFile2 testPort2
-    withServer_ storeLog storeMsgs storeNtfs port =
-      withSmpServerConfigOn (transport @TLS) proxyCfg {storeLogFile = Just storeLog, storeMsgsFile = Just storeMsgs, storeNtfsFile = Just storeNtfs} port
+    withServer2 = withServer_ testStoreLogFile2 testStoreMsgsDir2 testStoreNtfsFile2 testPort2
+    withServer_ storeLog storeMsgs storeNtfs =
+      withSmpServerConfigOn (transport @TLS) proxyCfg {storeLogFile = Just storeLog, storeMsgsFile = Just storeMsgs, storeNtfsFile = Just storeNtfs}
     a `up` cId = nGet a =##> \case ("", "", UP _ [c]) -> c == cId; _ -> False
     a `down` cId = nGet a =##> \case ("", "", DOWN _ [c]) -> c == cId; _ -> False
     aCfg = agentCfg {messageRetryInterval = fastMessageRetryInterval}

tests/ServerTests.hs

@@ -16,6 +16,7 @@
 module ServerTests where
 
 import AgentTests.NotificationTests (removeFileIfExists)
+import CoreTests.MsgStoreTests (testJournalStoreCfg)
 import Control.Concurrent (ThreadId, killThread, threadDelay)
 import Control.Concurrent.STM
 import Control.Exception (SomeException, try)
@@ -35,43 +36,47 @@ import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Protocol
-import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
+import Simplex.Messaging.Server (exportMessages)
+import Simplex.Messaging.Server.Env.STM (ServerConfig (..), readWriteQueueStore)
 import Simplex.Messaging.Server.Expiration
+import Simplex.Messaging.Server.MsgStore.Journal (JournalStoreConfig (..))
+import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..), newMsgStore)
 import Simplex.Messaging.Server.Stats (PeriodStatsData (..), ServerStatsData (..))
+import Simplex.Messaging.Server.StoreLog (closeStoreLog)
 import Simplex.Messaging.Transport
+import Simplex.Messaging.Util (whenM)
 import Simplex.Messaging.Version (mkVersionRange)
-import System.Directory (removeFile)
+import System.Directory (doesDirectoryExist, removeDirectoryRecursive, removeFile)
 import System.TimeIt (timeItT)
 import System.Timeout
 import Test.HUnit
 import Test.Hspec
 
-serverTests :: ATransport -> Spec
-serverTests t@(ATransport t') = do
-  describe "SMP syntax" $ syntaxTests t
+serverTests :: SpecWith (ATransport, AMSType)
+serverTests  = do
   describe "SMP queues" $ do
-    describe "NEW and KEY commands, SEND messages" $ testCreateSecure t
+    describe "NEW and KEY commands, SEND messages" testCreateSecure
     describe "NEW and SKEY commands" $ do
-      testCreateSndSecure t
-      testSndSecureProhibited t
-    describe "NEW, OFF and DEL commands, SEND messages" $ testCreateDelete t
-    describe "Stress test" $ stressTest t
-    describe "allowNewQueues setting" $ testAllowNewQueues t'
+      testCreateSndSecure
+      testSndSecureProhibited
+    describe "NEW, OFF and DEL commands, SEND messages" testCreateDelete
+    describe "Stress test" stressTest
+    describe "allowNewQueues setting" testAllowNewQueues
   describe "SMP messages" $ do
-    describe "duplex communication over 2 SMP connections" $ testDuplex t
-    describe "switch subscription to another TCP connection" $ testSwitchSub t
-    describe "GET command" $ testGetCommand t'
-    describe "GET & SUB commands" $ testGetSubCommands t'
-    describe "Exceeding queue quota" $ testExceedQueueQuota t'
-  describe "Store log" $ testWithStoreLog t
-  describe "Restore messages" $ testRestoreMessages t
-  describe "Restore messages (old / v2)" $ testRestoreExpireMessages t
-  describe "Timing of AUTH error" $ testTiming t
-  describe "Message notifications" $ testMessageNotifications t
+    describe "duplex communication over 2 SMP connections" testDuplex
+    describe "switch subscription to another TCP connection" testSwitchSub
+    describe "GET command" testGetCommand
+    describe "GET & SUB commands" testGetSubCommands
+    describe "Exceeding queue quota" testExceedQueueQuota
+  describe "Store log" testWithStoreLog
+  describe "Restore messages" testRestoreMessages
+  describe "Restore messages (old / v2)" testRestoreExpireMessages
+  describe "Timing of AUTH error" testTiming
+  describe "Message notifications" testMessageNotifications
   describe "Message expiration" $ do
-    testMsgExpireOnSend t'
-    testMsgExpireOnInterval t'
-    testMsgNOTExpireOnInterval t'
+    testMsgExpireOnSend
+    testMsgExpireOnInterval
+    testMsgNOTExpireOnInterval
 
 pattern Resp :: CorrId -> QueueId -> BrokerMsg -> SignedTransmission ErrorType BrokerMsg
 pattern Resp corrId queueId command <- (_, _, (corrId, queueId, Right command))
@@ -131,10 +136,10 @@ decryptMsgV3 dhShared nonce body =
     Right ClientRcvMsgQuota {} -> Left "ClientRcvMsgQuota"
     Left e -> Left e
 
-testCreateSecure :: ATransport -> Spec
-testCreateSecure (ATransport t) =
-  it "should create (NEW) and secure (KEY) queue" $
-    smpTest2 t $ \r s -> do
+testCreateSecure :: SpecWith (ATransport, AMSType)
+testCreateSecure =
+  it "should create (NEW) and secure (KEY) queue" $ \(ATransport t, msType) ->
+    smpTest2 t msType $ \r s -> do
       g <- C.newRandom
       (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
       (dhPub, dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
@@ -196,10 +201,10 @@ testCreateSecure (ATransport t) =
       Resp "bcda" _ (ERR LARGE_MSG) <- signSendRecv s sKey ("bcda", sId, _SEND biggerMessage)
       pure ()
 
-testCreateSndSecure :: ATransport -> Spec
-testCreateSndSecure (ATransport t) =
-  it "should create (NEW) and secure (SKEY) queue by sender" $
-    smpTest2 t $ \r s -> do
+testCreateSndSecure :: SpecWith (ATransport, AMSType)
+testCreateSndSecure =
+  it "should create (NEW) and secure (SKEY) queue by sender" $ \(ATransport t, msType) ->
+    smpTest2 t msType $ \r s -> do
       g <- C.newRandom
       (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
       (dhPub, dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
@@ -243,10 +248,10 @@ testCreateSndSecure (ATransport t) =
       Resp "bcda" _ (ERR LARGE_MSG) <- signSendRecv s sKey ("bcda", sId, _SEND biggerMessage)
       pure ()
 
-testSndSecureProhibited :: ATransport -> Spec
-testSndSecureProhibited (ATransport t) =
-  it "should create (NEW) without allowing sndSecure and fail to and secure queue by sender (SKEY)" $
-    smpTest2 t $ \r s -> do
+testSndSecureProhibited :: SpecWith (ATransport, AMSType)
+testSndSecureProhibited =
+  it "should create (NEW) without allowing sndSecure and fail to and secure queue by sender (SKEY)" $ \(ATransport t, msType) ->
+    smpTest2 t msType $ \r s -> do
       g <- C.newRandom
       (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
       (dhPub, _dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
@@ -258,10 +263,10 @@ testSndSecureProhibited (ATransport t) =
       (sId2, sId) #== "secures queue, same queue ID in response"
       (err, ERR AUTH) #== "rejects SKEY when not allowed in NEW command"
 
-testCreateDelete :: ATransport -> Spec
-testCreateDelete (ATransport t) =
-  it "should create (NEW), suspend (OFF) and delete (DEL) queue" $
-    smpTest2 t $ \rh sh -> do
+testCreateDelete :: SpecWith (ATransport, AMSType)
+testCreateDelete =
+  it "should create (NEW), suspend (OFF) and delete (DEL) queue" $ \(ATransport t, msType) ->
+    smpTest2 t msType $ \rh sh -> do
       g <- C.newRandom
       (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
       (dhPub, dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
@@ -329,10 +334,10 @@ testCreateDelete (ATransport t) =
       Resp "cdab" _ err10 <- signSendRecv rh rKey ("cdab", rId, SUB)
       (err10, ERR AUTH) #== "rejects SUB when deleted"
 
-stressTest :: ATransport -> Spec
-stressTest (ATransport t) =
-  it "should create many queues, disconnect and re-connect" $
-    smpTest3 t $ \h1 h2 h3 -> do
+stressTest :: SpecWith (ATransport, AMSType)
+stressTest =
+  it "should create many queues, disconnect and re-connect" $ \(ATransport t, msType) ->
+    smpTest3 t msType $ \h1 h2 h3 -> do
       g <- C.newRandom
       (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
       (dhPub, _ :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
@@ -347,10 +352,10 @@ stressTest (ATransport t) =
       closeConnection $ connection h2
       subscribeQueues h3
 
-testAllowNewQueues :: forall c. Transport c => TProxy c -> Spec
-testAllowNewQueues t =
-  it "should prohibit creating new queues with allowNewQueues = False" $ do
-    withSmpServerConfigOn (ATransport t) cfg {allowNewQueues = False} testPort $ \_ ->
+testAllowNewQueues :: SpecWith (ATransport, AMSType)
+testAllowNewQueues =
+  it "should prohibit creating new queues with allowNewQueues = False" $ \(ATransport (t :: TProxy c), msType) ->
+    withSmpServerConfigOn (ATransport t) (cfgMS msType) {allowNewQueues = False} testPort $ \_ ->
       testSMPClient @c $ \h -> do
         g <- C.newRandom
         (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
@@ -358,10 +363,10 @@ testAllowNewQueues t =
         Resp "abcd" NoEntity (ERR AUTH) <- signSendRecv h rKey ("abcd", NoEntity, NEW rPub dhPub Nothing SMSubscribe False)
         pure ()
 
-testDuplex :: ATransport -> Spec
-testDuplex (ATransport t) =
-  it "should create 2 simplex connections and exchange messages" $
-    smpTest2 t $ \alice bob -> do
+testDuplex :: SpecWith (ATransport, AMSType)
+testDuplex =
+  it "should create 2 simplex connections and exchange messages" $ \(ATransport t, msType) ->
+    smpTest2 t msType $ \alice bob -> do
       g <- C.newRandom
       (arPub, arKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
       (aDhPub, aDhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
@@ -413,10 +418,10 @@ testDuplex (ATransport t) =
       Resp "bcda" _ OK <- signSendRecv bob brKey ("bcda", bRcv, ACK mId5)
       (bDec mId5 msg5, Right "how are you bob") #== "message received from alice"
 
-testSwitchSub :: ATransport -> Spec
-testSwitchSub (ATransport t) =
-  it "should create simplex connections and switch subscription to another TCP connection" $
-    smpTest3 t $ \rh1 rh2 sh -> do
+testSwitchSub :: SpecWith (ATransport, AMSType)
+testSwitchSub =
+  it "should create simplex connections and switch subscription to another TCP connection" $ \(ATransport t, msType) ->
+    smpTest3 t msType $ \rh1 rh2 sh -> do
       g <- C.newRandom
       (rPub, rKey) <- atomically $ C.generateAuthKeyPair C.SEd448 g
       (dhPub, dhPriv :: C.PrivateKeyX25519) <- atomically $ C.generateKeyPair g
@@ -458,12 +463,12 @@ testSwitchSub (ATransport t) =
         Nothing -> return ()
         Just _ -> error "nothing else is delivered to the 1st TCP connection"
 
-testGetCommand :: forall c. Transport c => TProxy c -> Spec
-testGetCommand t =
-  it "should retrieve messages from the queue using GET command" $ do
+testGetCommand :: SpecWith (ATransport, AMSType)
+testGetCommand =
+  it "should retrieve messages from the queue using GET command" $ \(ATransport (t :: TProxy c), msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-    smpTest t $ \sh -> do
+    smpTest t msType $ \sh -> do
       queue <- newEmptyTMVarIO
       testSMPClient @c $ \rh ->
         atomically . putTMVar queue =<< createAndSecureQueue rh sPub
@@ -477,12 +482,12 @@ testGetCommand t =
         Resp "4" _ OK <- signSendRecv rh rKey ("4", rId, GET)
         pure ()
 
-testGetSubCommands :: forall c. Transport c => TProxy c -> Spec
-testGetSubCommands t =
-  it "should retrieve messages with GET and receive with SUB, only one ACK would work" $ do
+testGetSubCommands :: SpecWith (ATransport, AMSType)
+testGetSubCommands =
+  it "should retrieve messages with GET and receive with SUB, only one ACK would work" $ \(ATransport t, msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-    smpTest3 t $ \rh1 rh2 sh -> do
+    smpTest3 t msType $ \rh1 rh2 sh -> do
       (sId, rId, rKey, dhShared) <- createAndSecureQueue rh1 sPub
       let dec = decryptMsgV3 dhShared
       Resp "1" _ OK <- signSendRecv sh sKey ("1", sId, _SEND "hello 1")
@@ -527,10 +532,10 @@ testGetSubCommands t =
       Resp "12" _ OK <- signSendRecv rh2 rKey ("12", rId, GET)
       pure ()
 
-testExceedQueueQuota :: forall c. Transport c => TProxy c -> Spec
-testExceedQueueQuota t =
-  it "should reply with ERR QUOTA to sender and send QUOTA message to the recipient" $ do
-    withSmpServerConfigOn (ATransport t) cfg {msgQueueQuota = 2} testPort $ \_ ->
+testExceedQueueQuota :: SpecWith (ATransport, AMSType)
+testExceedQueueQuota =
+  it "should reply with ERR QUOTA to sender and send QUOTA message to the recipient" $ \(ATransport (t :: TProxy c), msType) -> do
+    withSmpServerConfigOn (ATransport t) (cfgMS msType) {msgQueueQuota = 2} testPort $ \_ ->
       testSMPClient @c $ \sh -> testSMPClient @c $ \rh -> do
         g <- C.newRandom
         (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
@@ -554,9 +559,9 @@ testExceedQueueQuota t =
         Resp "10" _ OK <- signSendRecv rh rKey ("10", rId, ACK mId4)
         pure ()
 
-testWithStoreLog :: ATransport -> Spec
-testWithStoreLog at@(ATransport t) =
-  it "should store simplex queues to log and restore them after server restart" $ do
+testWithStoreLog :: SpecWith (ATransport, AMSType)
+testWithStoreLog =
+  it "should store simplex queues to log and restore them after server restart" $ \(at@(ATransport t), msType) -> do
     g <- C.newRandom
     (sPub1, sKey1) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
     (sPub2, sKey2) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
@@ -568,7 +573,7 @@ testWithStoreLog at@(ATransport t) =
     senderId2 <- newTVarIO NoEntity
     notifierId <- newTVarIO NoEntity
 
-    withSmpServerStoreLogOn at testPort . runTest t $ \h -> runClient t $ \h1 -> do
+    withSmpServerStoreLogOnMS at msType testPort . runTest t $ \h -> runClient t $ \h1 -> do
       (sId1, rId1, rKey1, dhShared) <- createAndSecureQueue h sPub1
       (rcvNtfPubDhKey, _) <- atomically $ C.generateKeyPair g
       Resp "abcd" _ (NID nId _) <- signSendRecv h rKey1 ("abcd", rId1, NKEY nPub rcvNtfPubDhKey)
@@ -601,13 +606,14 @@ testWithStoreLog at@(ATransport t) =
 
     logSize testStoreLogFile `shouldReturn` 6
 
-    withSmpServerThreadOn at testPort . runTest t $ \h -> do
+    let cfg' = (cfgMS msType) {msgStoreType = AMSType SMSMemory, storeLogFile = Nothing, storeMsgsFile = Nothing}
+    withSmpServerConfigOn at cfg' testPort . runTest t $ \h -> do
       sId1 <- readTVarIO senderId1
       -- fails if store log is disabled
       Resp "bcda" _ (ERR AUTH) <- signSendRecv h sKey1 ("bcda", sId1, _SEND "hello")
       pure ()
 
-    withSmpServerStoreLogOn at testPort . runTest t $ \h -> runClient t $ \h1 -> do
+    withSmpServerStoreLogOnMS at msType testPort . runTest t $ \h -> runClient t $ \h1 -> do
       -- this queue is restored
       rId1 <- readTVarIO recipientId1
       Just rKey1 <- readTVarIO recipientKey1
@@ -641,11 +647,12 @@ logSize f =
     Right l -> pure l
     Left (_ :: SomeException) -> logSize f
 
-testRestoreMessages :: ATransport -> Spec
-testRestoreMessages at@(ATransport t) =
-  it "should store messages on exit and restore on start" $ do
+testRestoreMessages :: SpecWith (ATransport, AMSType)
+testRestoreMessages =
+  it "should store messages on exit and restore on start" $ \(at@(ATransport t), msType) -> do
     removeFileIfExists testStoreLogFile
     removeFileIfExists testStoreMsgsFile
+    whenM (doesDirectoryExist testStoreMsgsDir) $ removeDirectoryRecursive testStoreMsgsDir
     removeFileIfExists testServerStatsBackupFile
 
     g <- C.newRandom
@@ -655,7 +662,7 @@ testRestoreMessages at@(ATransport t) =
     dhShared <- newTVarIO Nothing
     senderId <- newTVarIO NoEntity
 
-    withSmpServerStoreMsgLogOn at testPort . runTest t $ \h -> do
+    withSmpServerStoreMsgLogOnMS at msType testPort . runTest t $ \h -> do
       runClient t $ \h1 -> do
         (sId, rId, rKey, dh) <- createAndSecureQueue h1 sPub
         atomically $ do
@@ -679,12 +686,12 @@ testRestoreMessages at@(ATransport t) =
     rId <- readTVarIO recipientId
 
     logSize testStoreLogFile `shouldReturn` 2
-    logSize testStoreMsgsFile `shouldReturn` 5
+    -- logSize testStoreMsgsFile `shouldReturn` 5
     logSize testServerStatsBackupFile `shouldReturn` 74
     Right stats1 <- strDecode <$> B.readFile testServerStatsBackupFile
     checkStats stats1 [rId] 5 1
 
-    withSmpServerStoreMsgLogOn at testPort . runTest t $ \h -> do
+    withSmpServerStoreMsgLogOnMS at msType testPort . runTest t $ \h -> do
       Just rKey <- readTVarIO recipientKey
       Just dh <- readTVarIO dhShared
       let dec = decryptMsgV3 dh
@@ -697,12 +704,12 @@ testRestoreMessages at@(ATransport t) =
 
     logSize testStoreLogFile `shouldReturn` 1
     -- the last message is not removed because it was not ACK'd
-    logSize testStoreMsgsFile `shouldReturn` 3
+    -- logSize testStoreMsgsFile `shouldReturn` 3
     logSize testServerStatsBackupFile `shouldReturn` 74
     Right stats2 <- strDecode <$> B.readFile testServerStatsBackupFile
     checkStats stats2 [rId] 5 3
 
-    withSmpServerStoreMsgLogOn at testPort . runTest t $ \h -> do
+    withSmpServerStoreMsgLogOnMS at msType testPort . runTest t $ \h -> do
       Just rKey <- readTVarIO recipientKey
       Just dh <- readTVarIO dhShared
       let dec = decryptMsgV3 dh
@@ -714,15 +721,15 @@ testRestoreMessages at@(ATransport t) =
       (dec mId6 msg6, Left "ClientRcvMsgQuota") #== "restored message delivered"
       Resp "7" _ OK <- signSendRecv h rKey ("7", rId, ACK mId6)
       pure ()
-
     logSize testStoreLogFile `shouldReturn` 1
-    logSize testStoreMsgsFile `shouldReturn` 0
+    -- logSize testStoreMsgsFile `shouldReturn` 0
     logSize testServerStatsBackupFile `shouldReturn` 74
     Right stats3 <- strDecode <$> B.readFile testServerStatsBackupFile
     checkStats stats3 [rId] 5 5
 
     removeFile testStoreLogFile
-    removeFile testStoreMsgsFile
+    removeFileIfExists testStoreMsgsFile
+    whenM (doesDirectoryExist testStoreMsgsDir) $ removeDirectoryRecursive testStoreMsgsDir
     removeFile testServerStatsBackupFile
   where
     runTest :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation
@@ -749,9 +756,9 @@ checkStats s qs sent received = do
   IS.toList _week `shouldBe` map (hash . unEntityId) qs
   IS.toList _month `shouldBe` map (hash . unEntityId) qs
 
-testRestoreExpireMessages :: ATransport -> Spec
-testRestoreExpireMessages at@(ATransport t) =
-  it "should store messages on exit and restore on start" $ do
+testRestoreExpireMessages :: SpecWith (ATransport, AMSType)
+testRestoreExpireMessages =
+  it "should store messages on exit and restore on start" $ \(at@(ATransport t), msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
     recipientId <- newTVarIO NoEntity
@@ -759,7 +766,7 @@ testRestoreExpireMessages at@(ATransport t) =
     dhShared <- newTVarIO Nothing
     senderId <- newTVarIO NoEntity
 
-    withSmpServerStoreMsgLogOn at testPort . runTest t $ \h -> do
+    withSmpServerStoreMsgLogOnMS at msType testPort . runTest t $ \h -> do
       runClient t $ \h1 -> do
         (sId, rId, rKey, dh) <- createAndSecureQueue h1 sPub
         atomically $ do
@@ -776,29 +783,40 @@ testRestoreExpireMessages at@(ATransport t) =
       pure ()
 
     logSize testStoreLogFile `shouldReturn` 2
+    exportStoreMessages msType
     msgs <- B.readFile testStoreMsgsFile
     length (B.lines msgs) `shouldBe` 4
 
     let expCfg1 = Just ExpirationConfig {ttl = 86400, checkInterval = 43200}
-        cfg1 = cfg {storeLogFile = Just testStoreLogFile, storeMsgsFile = Just testStoreMsgsFile, messageExpiration = expCfg1, serverStatsBackupFile = Just testServerStatsBackupFile}
+        cfg1 = (cfgMS msType) {messageExpiration = expCfg1, serverStatsBackupFile = Just testServerStatsBackupFile}
     withSmpServerConfigOn at cfg1 testPort . runTest t $ \_ -> pure ()
 
     logSize testStoreLogFile `shouldReturn` 1
+    exportStoreMessages msType
     msgs' <- B.readFile testStoreMsgsFile
     msgs' `shouldBe` msgs
 
     let expCfg2 = Just ExpirationConfig {ttl = 2, checkInterval = 43200}
-        cfg2 = cfg {storeLogFile = Just testStoreLogFile, storeMsgsFile = Just testStoreMsgsFile, messageExpiration = expCfg2, serverStatsBackupFile = Just testServerStatsBackupFile}
+        cfg2 = (cfgMS msType) {messageExpiration = expCfg2, serverStatsBackupFile = Just testServerStatsBackupFile}
     withSmpServerConfigOn at cfg2 testPort . runTest t $ \_ -> pure ()
 
     logSize testStoreLogFile `shouldReturn` 1
     -- two messages expired
+    exportStoreMessages msType
     msgs'' <- B.readFile testStoreMsgsFile
     length (B.lines msgs'') `shouldBe` 2
     B.lines msgs'' `shouldBe` drop 2 (B.lines msgs)
     Right ServerStatsData {_msgExpired} <- strDecode <$> B.readFile testServerStatsBackupFile
     _msgExpired `shouldBe` 2
   where
+    exportStoreMessages :: AMSType -> IO ()
+    exportStoreMessages = \case
+      AMSType SMSJournal -> do
+        ms <- newMsgStore testJournalStoreCfg {quota = 4}
+        readWriteQueueStore testStoreLogFile ms >>= closeStoreLog
+        removeFileIfExists testStoreMsgsFile
+        exportMessages False ms testStoreMsgsFile False
+      AMSType SMSMemory -> pure ()
     runTest :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> ThreadId -> Expectation
     runTest _ test' server = do
       testSMPClient test' `shouldReturn` ()
@@ -818,12 +836,12 @@ createAndSecureQueue h sPub = do
   (rId', rId) #== "same queue ID"
   pure (sId, rId, rKey, dhShared)
 
-testTiming :: ATransport -> Spec
-testTiming (ATransport t) =
+testTiming :: SpecWith (ATransport, AMSType)
+testTiming =
   describe "should have similar time for auth error, whether queue exists or not, for all key types" $
     forM_ timingTests $ \tst ->
-      it (testName tst) $
-        smpTest2Cfg cfg (mkVersionRange batchCmdsSMPVersion authCmdsSMPVersion) t $ \rh sh ->
+      it (testName tst) $ \(ATransport t, msType) ->
+        smpTest2Cfg (cfgMS msType) (mkVersionRange batchCmdsSMPVersion authCmdsSMPVersion) t $ \rh sh ->
           testSameTiming rh sh tst
   where
     testName :: (C.AuthAlg, C.AuthAlg, Int) -> String
@@ -888,13 +906,13 @@ testTiming (ATransport t) =
             ]
           ok `shouldBe` True
 
-testMessageNotifications :: ATransport -> Spec
-testMessageNotifications (ATransport t) =
-  it "should create simplex connection, subscribe notifier and deliver notifications" $ do
+testMessageNotifications :: SpecWith (ATransport, AMSType)
+testMessageNotifications =
+  it "should create simplex connection, subscribe notifier and deliver notifications" $ \(ATransport t, msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
     (nPub, nKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-    smpTest4 t $ \rh sh nh1 nh2 -> do
+    smpTest4 t msType $ \rh sh nh1 nh2 -> do
       (sId, rId, rKey, dhShared) <- createAndSecureQueue rh sPub
       let dec = decryptMsgV3 dhShared
       (rcvNtfPubDhKey, _) <- atomically $ C.generateKeyPair g
@@ -928,12 +946,12 @@ testMessageNotifications (ATransport t) =
         Nothing -> pure ()
         Just _ -> error "nothing else should be delivered to the 2nd notifier's TCP connection"
 
-testMsgExpireOnSend :: forall c. Transport c => TProxy c -> Spec
-testMsgExpireOnSend t =
-  it "should expire messages that are not received before messageTTL on SEND" $ do
+testMsgExpireOnSend :: SpecWith (ATransport, AMSType)
+testMsgExpireOnSend =
+  it "should expire messages that are not received before messageTTL on SEND" $ \(ATransport (t :: TProxy c), msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-    let cfg' = cfg {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 10000}}
+    let cfg' = (cfgMS msType) {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 10000}}
     withSmpServerConfigOn (ATransport t) cfg' testPort $ \_ ->
       testSMPClient @c $ \sh -> do
         (sId, rId, rKey, dhShared) <- testSMPClient @c $ \rh -> createAndSecureQueue rh sPub
@@ -948,13 +966,13 @@ testMsgExpireOnSend t =
             Nothing -> return ()
             Just _ -> error "nothing else should be delivered"
 
-testMsgExpireOnInterval :: forall c. Transport c => TProxy c -> Spec
-testMsgExpireOnInterval t =
+testMsgExpireOnInterval :: SpecWith (ATransport, AMSType)
+testMsgExpireOnInterval =
   -- fails on ubuntu
-  xit' "should expire messages that are not received before messageTTL after expiry interval" $ do
+  xit' "should expire messages that are not received before messageTTL after expiry interval" $ \(ATransport (t :: TProxy c), msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-    let cfg' = cfg {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}}
+    let cfg' = (cfgMS msType) {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}, idleQueueInterval = 1}
     withSmpServerConfigOn (ATransport t) cfg' testPort $ \_ ->
       testSMPClient @c $ \sh -> do
         (sId, rId, rKey, _) <- testSMPClient @c $ \rh -> createAndSecureQueue rh sPub
@@ -968,12 +986,12 @@ testMsgExpireOnInterval t =
             Nothing -> return ()
             Just _ -> error "nothing should be delivered"
 
-testMsgNOTExpireOnInterval :: forall c. Transport c => TProxy c -> Spec
-testMsgNOTExpireOnInterval t =
-  it "should NOT expire messages that are not received before messageTTL if expiry interval is large" $ do
+testMsgNOTExpireOnInterval :: SpecWith (ATransport, AMSType)
+testMsgNOTExpireOnInterval =
+  it "should NOT expire messages that are not received before messageTTL if expiry interval is large" $ \(ATransport (t :: TProxy c), msType) -> do
     g <- C.newRandom
     (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-    let cfg' = cfg {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 10000}}
+    let cfg' = (cfgMS msType) {messageExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 10000}}
     withSmpServerConfigOn (ATransport t) cfg' testPort $ \_ ->
       testSMPClient @c $ \sh -> do
         (sId, rId, rKey, dhShared) <- testSMPClient @c $ \rh -> createAndSecureQueue rh sPub
@@ -1006,8 +1024,8 @@ instance Eq C.ASignature where
     Just Refl -> s == s'
     _ -> False
 
-syntaxTests :: ATransport -> Spec
-syntaxTests (ATransport t) = do
+serverSyntaxTests :: ATransport -> Spec
+serverSyntaxTests (ATransport t) = do
   it "unknown command" $ ("", "abcd", "1234", ('H', 'E', 'L', 'L', 'O')) >#> ("", "abcd", "1234", ERR $ CMD UNKNOWN)
   describe "NEW" $ do
     it "no parameters" $ (sampleSig, "bcda", "", NEW_) >#> ("", "bcda", "", ERR $ CMD SYNTAX)

tests/Test.hs

@@ -11,6 +11,7 @@ import CoreTests.BatchingTests
 import CoreTests.CryptoFileTests
 import CoreTests.CryptoTests
 import CoreTests.EncodingTests
+import CoreTests.MsgStoreTests
 import CoreTests.RetryIntervalTests
 import CoreTests.SOCKSSettings
 import CoreTests.StoreLogTests
@@ -24,8 +25,9 @@ import NtfServerTests (ntfServerTests)
 import RemoteControl (remoteControlTests)
 import SMPProxyTests (smpProxyTests)
 import ServerTests
+import Simplex.Messaging.Server.MsgStore.Types (AMSType (..), SMSType (..))
 import Simplex.Messaging.Transport (TLS, Transport (..))
-import Simplex.Messaging.Transport.WebSockets (WS)
+-- import Simplex.Messaging.Transport.WebSockets (WS)
 import System.Directory (createDirectoryIfMissing, removeDirectoryRecursive)
 import System.Environment (setEnv)
 import Test.Hspec
@@ -53,13 +55,20 @@ main = do
           describe "Version range" versionRangeTests
           describe "Encryption tests" cryptoTests
           describe "Encrypted files tests" cryptoFileTests
+          describe "Message store tests" msgStoreTests
           describe "Retry interval tests" retryIntervalTests
           describe "SOCKS settings tests" socksSettingsTests
           describe "Store log tests" storeLogTests
           describe "TRcvQueues tests" tRcvQueuesTests
           describe "Util tests" utilTests
-        describe "SMP server via TLS" $ serverTests (transport @TLS)
-        describe "SMP server via WebSockets" $ serverTests (transport @WS)
+        describe "SMP server via TLS, jornal message store" $ do
+          describe "SMP syntax" $ serverSyntaxTests (transport @TLS)
+          before (pure (transport @TLS, AMSType SMSJournal)) serverTests
+        describe "SMP server via TLS, memory message store" $
+          before (pure (transport @TLS, AMSType SMSMemory)) serverTests
+        -- xdescribe "SMP server via WebSockets" $ do
+        --   describe "SMP syntax" $ serverSyntaxTests (transport @WS)
+        --   before (pure (transport @WS, AMSType SMSJournal)) serverTests
         describe "Notifications server" $ ntfServerTests (transport @TLS)
         describe "SMP client agent" $ agentTests (transport @TLS)
         describe "SMP proxy" smpProxyTests

tests/XFTPAgent.hs

@@ -24,8 +24,8 @@ import SMPClient (xit'')
 import Simplex.FileTransfer.Client (XFTPClientConfig (..))
 import Simplex.FileTransfer.Description (FileChunk (..), FileDescription (..), FileDescriptionURI (..), ValidFileDescription, fileDescriptionURI, kb, mb, qrSizeLimit, pattern ValidFileDescription)
 import Simplex.FileTransfer.Protocol (FileParty (..))
-import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..), supportedXFTPhandshakes)
-import Simplex.FileTransfer.Transport (XFTPErrorType (AUTH))
+import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..))
+import Simplex.FileTransfer.Transport (XFTPErrorType (AUTH), supportedXFTPhandshakes)
 import Simplex.FileTransfer.Types (RcvFileId, SndFileId)
 import Simplex.Messaging.Agent (AgentClient, testProtocolServer, xftpDeleteRcvFile, xftpDeleteSndFileInternal, xftpDeleteSndFileRemote, xftpReceiveFile, xftpSendDescription, xftpSendFile, xftpStartWorkers)
 import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..))

tests/XFTPClient.hs

@@ -14,8 +14,8 @@ import SMPClient (serverBracket)
 import Simplex.FileTransfer.Client
 import Simplex.FileTransfer.Description
 import Simplex.FileTransfer.Server (runXFTPServerBlocking)
-import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..), defaultFileExpiration, defaultInactiveClientExpiration, supportedXFTPhandshakes)
-import Simplex.FileTransfer.Transport (supportedFileServerVRange)
+import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..), defaultFileExpiration, defaultInactiveClientExpiration)
+import Simplex.FileTransfer.Transport (supportedFileServerVRange, supportedXFTPhandshakes)
 import Simplex.Messaging.Protocol (XFTPServer)
 import Simplex.Messaging.Transport (ALPN)
 import Simplex.Messaging.Transport.Server