Merge branch 'master' into ep/disable-background

* smp server: log prometheus metrics

* save metrics

* diff

* lines

* version

* do not include Prometheus into client

* corrections

Co-authored-by: sh <37271604+shumvgolove@users.noreply.github.com>

* corrections

Co-authored-by: sh <37271604+shumvgolove@users.noreply.github.com>

* corrections

Co-authored-by: sh <37271604+shumvgolove@users.noreply.github.com>

* add timestamp to metrics

* remove type

* remove version

---------

Co-authored-by: sh <37271604+shumvgolove@users.noreply.github.com>

CHANGELOG.md

@@ -1,3 +1,19 @@
+# 6.2.0
+
+Version 6.2.0.7
+
+Build:
+- client_library flag to build only used modules in the clients, remove package yaml
+
+SMP server:
+- journal storage for messages (BETA).
+- prevent race condition when deleting queue and to avoid "orphan" messages (#1395).
+
+SMP agent:
+- support SMP and XFTP server roles (storage/proxy) and operators (#1343).
+- treat blocked STM and other critical errors that offer restart as temporary for message delivery (#1405).
+- fix inconsistent state after app restart while accepting contact request (#1412).
+
 # 6.1.3
 
 SMP server: fix restoring notification credentials.

package.yaml

@@ -1,224 +0,0 @@
-name: simplexmq
-version: 6.1.3.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
-      # - -fhpc
-
-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-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.3.0
+version:        6.2.1.100
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -66,24 +62,21 @@ 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
       Simplex.FileTransfer.Chunks
       Simplex.FileTransfer.Client
       Simplex.FileTransfer.Client.Agent
-      Simplex.FileTransfer.Client.Main
       Simplex.FileTransfer.Client.Presets
       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,36 +147,12 @@ 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.Journal
-      Simplex.Messaging.Server.MsgStore.STM
-      Simplex.Messaging.Server.MsgStore.Types
-      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
@@ -197,7 +166,6 @@ library
       Simplex.Messaging.Transport.HTTP2.Server
       Simplex.Messaging.Transport.KeepAlive
       Simplex.Messaging.Transport.Server
-      Simplex.Messaging.Transport.WebSockets
       Simplex.Messaging.Util
       Simplex.Messaging.Version
       Simplex.Messaging.Version.Internal
@@ -206,13 +174,50 @@ library
       Simplex.RemoteControl.Discovery.Multicast
       Simplex.RemoteControl.Invitation
       Simplex.RemoteControl.Types
+  if !flag(client_library)
+      exposed-modules:
+          Simplex.FileTransfer.Client.Main
+          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.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.Server
+          Simplex.Messaging.Server.CLI
+          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.Prometheus
+          Simplex.Messaging.Server.QueueStore
+          Simplex.Messaging.Server.QueueStore.STM
+          Simplex.Messaging.Server.Stats
+          Simplex.Messaging.Server.StoreLog
+          Simplex.Messaging.Server.StoreLog.Types
+          Simplex.Messaging.Transport.WebSockets
   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:
@@ -222,14 +227,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.*
@@ -242,7 +245,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
@@ -269,24 +271,28 @@ library
     , 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 !flag(client_library)
+    build-depends:
+        case-insensitive ==1.2.*
+      , hashable ==1.4.*
+      , websockets ==0.12.*
   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
@@ -294,75 +300,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
@@ -373,80 +320,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
@@ -454,75 +348,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
@@ -530,75 +364,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:
@@ -610,6 +385,7 @@ test-suite simplexmq-test
       AgentTests.MigrationTests
       AgentTests.NotificationTests
       AgentTests.SchemaDump
+      AgentTests.ServerChoice
       AgentTests.SQLiteTests
       CLITests
       CoreTests.BatchingTests
@@ -644,87 +420,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

@@ -51,8 +51,7 @@ import Data.Text (Text)
 import Data.Time.Clock (getCurrentTime)
 import Data.Time.Format (defaultTimeLocale, formatTime)
 import Simplex.FileTransfer.Chunks (toKB)
-import Simplex.FileTransfer.Client (XFTPChunkSpec (..))
-import Simplex.FileTransfer.Client.Main
+import Simplex.FileTransfer.Client (XFTPChunkSpec (..), getChunkDigest, prepareChunkSizes, prepareChunkSpecs, singleChunkSize)
 import Simplex.FileTransfer.Crypto
 import Simplex.FileTransfer.Description
 import Simplex.FileTransfer.Protocol (FileParty (..), SFileParty (..))
@@ -461,14 +460,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 +476,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 +545,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

@@ -20,16 +20,19 @@ import Data.Bifunctor (first)
 import Data.ByteString.Builder (Builder, byteString)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
+import qualified Data.ByteString.Lazy as LB
 import Data.Int (Int64)
+import Data.List (foldl')
 import Data.List.NonEmpty (NonEmpty (..))
+import Data.Maybe (listToMaybe)
 import Data.Time.Clock (UTCTime)
 import Data.Word (Word32)
 import qualified Data.X509 as X
 import qualified Data.X509.Validation as XV
 import qualified Network.HTTP.Types as N
 import qualified Network.HTTP2.Client as H
+import Simplex.FileTransfer.Chunks
 import Simplex.FileTransfer.Protocol
-import Simplex.FileTransfer.Server.Env (supportedXFTPhandshakes)
 import Simplex.FileTransfer.Transport
 import Simplex.Messaging.Client
   ( NetworkConfig (..),
@@ -299,3 +302,41 @@ noFile HTTP2Body {bodyPart} a = case bodyPart of
 
 -- FACK :: FileCommand Recipient
 -- PING :: FileCommand Recipient
+
+singleChunkSize :: Int64 -> Maybe Word32
+singleChunkSize size' =
+  listToMaybe $ dropWhile (< chunkSize) serverChunkSizes
+  where
+    chunkSize = fromIntegral size'
+
+prepareChunkSizes :: Int64 -> [Word32]
+prepareChunkSizes size' = prepareSizes size'
+  where
+    (smallSize, bigSize)
+      | size' > size34 chunkSize3 = (chunkSize2, chunkSize3)
+      | size' > size34 chunkSize2 = (chunkSize1, chunkSize2)
+      | otherwise = (chunkSize0, chunkSize1)
+    size34 sz = (fromIntegral sz * 3) `div` 4
+    prepareSizes 0 = []
+    prepareSizes size
+      | size >= fromIntegral bigSize = replicate (fromIntegral n1) bigSize <> prepareSizes remSz
+      | size > size34 bigSize = [bigSize]
+      | otherwise = replicate (fromIntegral n2') smallSize
+      where
+        (n1, remSz) = size `divMod` fromIntegral bigSize
+        n2' = let (n2, remSz2) = (size `divMod` fromIntegral smallSize) in if remSz2 == 0 then n2 else n2 + 1
+
+prepareChunkSpecs :: FilePath -> [Word32] -> [XFTPChunkSpec]
+prepareChunkSpecs filePath chunkSizes = reverse . snd $ foldl' addSpec (0, []) chunkSizes
+  where
+    addSpec :: (Int64, [XFTPChunkSpec]) -> Word32 -> (Int64, [XFTPChunkSpec])
+    addSpec (chunkOffset, specs) sz =
+      let spec = XFTPChunkSpec {filePath, chunkOffset, chunkSize = sz}
+       in (chunkOffset + fromIntegral sz, spec : specs)
+
+getChunkDigest :: XFTPChunkSpec -> IO ByteString
+getChunkDigest XFTPChunkSpec {filePath = chunkPath, chunkOffset, chunkSize} =
+  withFile chunkPath ReadMode $ \h -> do
+    hSeek h AbsoluteSeek $ fromIntegral chunkOffset
+    chunk <- LB.hGet h (fromIntegral chunkSize)
+    pure $! LC.sha256Hash chunk

src/Simplex/FileTransfer/Client/Main.hs

@@ -19,11 +19,7 @@ module Simplex.FileTransfer.Client.Main
     singleChunkSize,
     prepareChunkSizes,
     prepareChunkSpecs,
-    maxFileSize,
-    maxFileSizeHard,
-    fileSizeLen,
     getChunkDigest,
-    SentRecipientReplica (..),
   )
 where
 
@@ -34,7 +30,6 @@ import Control.Monad.Trans.Except
 import Crypto.Random (ChaChaDRG)
 import qualified Data.Attoparsec.ByteString.Char8 as A
 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.Char (toLower)
@@ -45,7 +40,7 @@ import Data.List.NonEmpty (NonEmpty (..), nonEmpty)
 import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
-import Data.Maybe (fromMaybe, listToMaybe)
+import Data.Maybe (fromMaybe)
 import qualified Data.Text as T
 import Data.Word (Word32)
 import GHC.Records (HasField (getField))
@@ -80,20 +75,6 @@ import UnliftIO.Directory
 xftpClientVersion :: String
 xftpClientVersion = "1.0.1"
 
--- | Soft limit for XFTP clients. Should be checked and reported to user.
-maxFileSize :: Int64
-maxFileSize = gb 1
-
-maxFileSizeStr :: String
-maxFileSizeStr = B.unpack . strEncode $ FileSize maxFileSize
-
--- | Hard internal limit for XFTP agent after which it refuses to prepare chunks.
-maxFileSizeHard :: Int64
-maxFileSizeHard = gb 5
-
-fileSizeLen :: Int64
-fileSizeLen = 8
-
 newtype CLIError = CLIError String
   deriving (Eq, Show, Exception)
 
@@ -231,16 +212,6 @@ data SentFileChunkReplica = SentFileChunkReplica
   }
   deriving (Show)
 
-data SentRecipientReplica = SentRecipientReplica
-  { chunkNo :: Int,
-    server :: XFTPServer,
-    rcvNo :: Int,
-    replicaId :: ChunkReplicaId,
-    replicaKey :: C.APrivateAuthKey,
-    digest :: FileDigest,
-    chunkSize :: FileSize Word32
-  }
-
 logCfg :: LogConfig
 logCfg = LogConfig {lc_file = Nothing, lc_stderr = True}
 
@@ -414,13 +385,6 @@ cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, re
       B.writeFile fdSndPath $ strEncode fdSnd
       pure (fdRcvPaths, fdSndPath)
 
-getChunkDigest :: XFTPChunkSpec -> IO ByteString
-getChunkDigest XFTPChunkSpec {filePath = chunkPath, chunkOffset, chunkSize} =
-  withFile chunkPath ReadMode $ \h -> do
-    hSeek h AbsoluteSeek $ fromIntegral chunkOffset
-    chunk <- LB.hGet h (fromIntegral chunkSize)
-    pure $! LC.sha256Hash chunk
-
 cliReceiveFile :: ReceiveOptions -> ExceptT CLIError IO ()
 cliReceiveFile ReceiveOptions {fileDescription, filePath, retryCount, tempPath, verbose, yes} =
   getFileDescription' fileDescription >>= receive
@@ -536,37 +500,6 @@ getFileDescription' path =
   getFileDescription path >>= \case
     AVFD fd -> either (throwE . CLIError) pure $ checkParty fd
 
-singleChunkSize :: Int64 -> Maybe Word32
-singleChunkSize size' =
-  listToMaybe $ dropWhile (< chunkSize) serverChunkSizes
-  where
-    chunkSize = fromIntegral size'
-
-prepareChunkSizes :: Int64 -> [Word32]
-prepareChunkSizes size' = prepareSizes size'
-  where
-    (smallSize, bigSize)
-      | size' > size34 chunkSize3 = (chunkSize2, chunkSize3)
-      | size' > size34 chunkSize2 = (chunkSize1, chunkSize2)
-      | otherwise = (chunkSize0, chunkSize1)
-    size34 sz = (fromIntegral sz * 3) `div` 4
-    prepareSizes 0 = []
-    prepareSizes size
-      | size >= fromIntegral bigSize = replicate (fromIntegral n1) bigSize <> prepareSizes remSz
-      | size > size34 bigSize = [bigSize]
-      | otherwise = replicate (fromIntegral n2') smallSize
-      where
-        (n1, remSz) = size `divMod` fromIntegral bigSize
-        n2' = let (n2, remSz2) = (size `divMod` fromIntegral smallSize) in if remSz2 == 0 then n2 else n2 + 1
-
-prepareChunkSpecs :: FilePath -> [Word32] -> [XFTPChunkSpec]
-prepareChunkSpecs filePath chunkSizes = reverse . snd $ foldl' addSpec (0, []) chunkSizes
-  where
-    addSpec :: (Int64, [XFTPChunkSpec]) -> Word32 -> (Int64, [XFTPChunkSpec])
-    addSpec (chunkOffset, specs) sz =
-      let spec = XFTPChunkSpec {filePath, chunkOffset, chunkSize = sz}
-       in (chunkOffset + fromIntegral sz, spec : specs)
-
 getEncPath :: MonadIO m => Maybe FilePath -> String -> m FilePath
 getEncPath path name = (`uniqueCombine` (name <> ".encrypted")) =<< maybe (liftIO getCanonicalTemporaryDirectory) pure path
 

src/Simplex/FileTransfer/Description.hs

@@ -37,6 +37,10 @@ module Simplex.FileTransfer.Description
     FileClientData,
     fileDescriptionURI,
     qrSizeLimit,
+    maxFileSize,
+    maxFileSizeStr,
+    maxFileSizeHard,
+    fileSizeLen,
   )
 where
 
@@ -266,6 +270,21 @@ instance StrEncoding FileDescriptionURI where
 qrSizeLimit :: Int
 qrSizeLimit = 1002 -- ~2 chunks in URLencoded YAML with some spare size for server hosts
 
+-- | Soft limit for XFTP clients. Should be checked and reported to user.
+maxFileSize :: Int64
+maxFileSize = gb 1
+
+maxFileSizeStr :: String
+maxFileSizeStr = B.unpack . strEncode $ FileSize maxFileSize
+
+-- | Hard internal limit for XFTP agent after which it refuses to prepare chunks.
+maxFileSizeHard :: Int64
+maxFileSizeHard = gb 5
+
+fileSizeLen :: Int64
+fileSizeLen = 8
+
+
 instance (Integral a, Show a) => StrEncoding (FileSize a) where
   strEncode (FileSize b)
     | b' /= 0 = bshow b

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/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/FileTransfer/Types.hs

@@ -246,6 +246,16 @@ data DeletedSndChunkReplica = DeletedSndChunkReplica
   }
   deriving (Show)
 
+data SentRecipientReplica = SentRecipientReplica
+  { chunkNo :: Int,
+    server :: XFTPServer,
+    rcvNo :: Int,
+    replicaId :: ChunkReplicaId,
+    replicaKey :: C.APrivateAuthKey,
+    digest :: FileDigest,
+    chunkSize :: FileSize Word32
+  }
+
 data FileErrorType
   = -- | cannot proceed with download from not approved relays without proxy
     NOT_APPROVED

src/Simplex/Messaging/Agent.hs

@@ -120,6 +120,8 @@ module Simplex.Messaging.Agent
     debugAgentLocks,
     getAgentSubscriptions,
     logConnection,
+    -- for tests
+    withAgentEnv,
   )
 where
 
@@ -219,13 +221,14 @@ getSMPAgentClient_ clientId cfg initServers@InitialAgentServers {smp, xftp} stor
       | backgroundMode = run c "subscriber" $ subscriber c
       | otherwise = do
           restoreServersStats c
-          raceAny_
-            [ run c "subscriber" $ subscriber c,
-              run c "runNtfSupervisor" $ runNtfSupervisor c,
-              run c "cleanupManager" $ cleanupManager c,
-              run c "logServersStats" $ logServersStats c
-            ]
-            `E.finally` saveServersStats c
+          run c "subscriber" $ subscriber c
+          -- raceAny_
+          --   [ run c "subscriber" $ subscriber c
+          --     run c "runNtfSupervisor" $ runNtfSupervisor c,
+          --     run c "cleanupManager" $ cleanupManager c,
+          --     run c "logServersStats" $ logServersStats c
+          --   ]
+          --   `E.finally` saveServersStats c
     run AgentClient {subQ, acThread} name a =
       a `E.catchAny` \e -> whenM (isJust <$> readTVarIO acThread) $ do
         logError $ "Agent thread " <> name <> " crashed: " <> tshow e
@@ -685,9 +688,15 @@ allowConnectionAsync' c corrId connId confId ownConnInfo =
       enqueueCommand c corrId connId (Just server) $ AClientCommand $ LET confId ownConnInfo
     _ -> throwE $ CMD PROHIBITED "allowConnectionAsync"
 
+-- TODO
+-- Unlike `acceptContact` (synchronous version), `acceptContactAsync` uses `unacceptInvitation` in case of error,
+-- because we're not taking lock here. In practice it is less likely to fail because it doesn't involve network IO,
+-- and also it can't be triggered by user concurrently several times in a row. It could be improved similarly to
+-- `acceptContact` by creating a new map for invitation locks and taking lock here, and removing `unacceptInvitation`
+-- while marking invitation as accepted inside "lock level transaction" after successful `joinConnAsync`.
 acceptContactAsync' :: AgentClient -> ACorrId -> Bool -> InvitationId -> ConnInfo -> PQSupport -> SubscriptionMode -> AM ConnId
 acceptContactAsync' c corrId enableNtfs invId ownConnInfo pqSupport subMode = do
-  Invitation {contactConnId, connReq} <- withStore c (`getInvitation` invId)
+  Invitation {contactConnId, connReq} <- withStore c $ \db -> getInvitation db "acceptContactAsync'" invId
   withStore c (`getConn` contactConnId) >>= \case
     SomeConn _ (ContactConnection ConnData {userId} _) -> do
       withStore' c $ \db -> acceptInvitation db invId ownConnInfo
@@ -807,7 +816,7 @@ newConnToJoin c userId connId enableNtfs cReq pqSup = case cReq of
 
 newConnToAccept :: AgentClient -> ConnId -> Bool -> ConfirmationId -> PQSupport -> AM ConnId
 newConnToAccept c connId enableNtfs invId pqSup = do
-  Invitation {connReq, contactConnId} <- withStore c (`getInvitation` invId)
+  Invitation {connReq, contactConnId} <- withStore c $ \db -> getInvitation db "newConnToAccept" invId
   withStore c (`getConn` contactConnId) >>= \case
     SomeConn _ (ContactConnection ConnData {userId} _) ->
       newConnToJoin c userId connId enableNtfs connReq pqSup
@@ -815,11 +824,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 =
@@ -937,13 +948,12 @@ allowConnection' c connId confId ownConnInfo = withConnLock c connId "allowConne
 -- | Accept contact (ACPT command) in Reader monad
 acceptContact' :: AgentClient -> ConnId -> Bool -> InvitationId -> ConnInfo -> PQSupport -> SubscriptionMode -> AM SndQueueSecured
 acceptContact' c connId enableNtfs invId ownConnInfo pqSupport subMode = withConnLock c connId "acceptContact" $ do
-  Invitation {contactConnId, connReq} <- withStore c (`getInvitation` invId)
+  Invitation {contactConnId, connReq} <- withStore c $ \db -> getInvitation db "acceptContact'" invId
   withStore c (`getConn` contactConnId) >>= \case
     SomeConn _ (ContactConnection ConnData {userId} _) -> do
+      sqSecured <- joinConn c userId connId enableNtfs connReq ownConnInfo pqSupport subMode
       withStore' c $ \db -> acceptInvitation db invId ownConnInfo
-      joinConn c userId connId enableNtfs connReq ownConnInfo pqSupport subMode `catchAgentError` \err -> do
-        withStore' c (`unacceptInvitation` invId)
-        throwE err
+      pure sqSecured
     _ -> throwE $ CMD PROHIBITED "acceptContact"
 
 -- | Reject contact (RJCT command) in Reader monad
@@ -1194,14 +1204,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 +1658,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 +2167,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

@@ -149,7 +149,6 @@ module Simplex.Messaging.Agent.Client
     userServers,
     pickServer,
     getNextServer,
-    withUserServers,
     withNextSrv,
     incSMPServerStat,
     incSMPServerStat',
@@ -193,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)
@@ -264,7 +263,6 @@ import Simplex.Messaging.Protocol
     VersionSMPC,
     XFTPServer,
     XFTPServerWithAuth,
-    sameSrvAddr',
     pattern NoEntity,
   )
 import qualified Simplex.Messaging.Protocol as SMP
@@ -619,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
@@ -1074,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
@@ -1093,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
@@ -1388,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
@@ -2021,7 +2020,7 @@ storeError = \case
   SEConnDuplicate -> CONN DUPLICATE
   SEBadConnType CRcv -> CONN SIMPLEX
   SEBadConnType CSnd -> CONN SIMPLEX
-  SEInvitationNotFound -> CMD PROHIBITED "SEInvitationNotFound"
+  SEInvitationNotFound cxt invId -> CMD PROHIBITED $ "SEInvitationNotFound " <> cxt <> ", invitationId = " <> show invId
   -- this error is never reported as store error,
   -- it is used to wrap agent operations when "transaction-like" store access is needed
   -- NOTE: network IO should NOT be used inside AgentStoreMonad
@@ -2035,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/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/Agent/Store.hs

@@ -623,7 +623,7 @@ data StoreError
   | -- | Confirmation not found.
     SEConfirmationNotFound
   | -- | Invitation not found
-    SEInvitationNotFound
+    SEInvitationNotFound String InvitationId
   | -- | Message not found
     SEMsgNotFound
   | -- | Command not found

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

@@ -926,9 +926,9 @@ createInvitation db gVar NewInvitation {contactConnId, connReq, recipientConnInf
       |]
       (invitationId, contactConnId, connReq, recipientConnInfo)
 
-getInvitation :: DB.Connection -> InvitationId -> IO (Either StoreError Invitation)
-getInvitation db invitationId =
-  firstRow invitation SEInvitationNotFound $
+getInvitation :: DB.Connection -> String -> InvitationId -> IO (Either StoreError Invitation)
+getInvitation db cxt invitationId =
+  firstRow invitation (SEInvitationNotFound cxt invitationId) $
     DB.query
       db
       [sql|

src/Simplex/Messaging/Client.hs

@@ -145,7 +145,6 @@ import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client (SocksAuth (..), SocksProxyWithAuth (..), TransportClientConfig (..), TransportHost (..), defaultSMPPort, defaultTcpConnectTimeout, runTransportClient)
 import Simplex.Messaging.Transport.KeepAlive
-import Simplex.Messaging.Transport.WebSockets (WS)
 import Simplex.Messaging.Util (bshow, diffToMicroseconds, ifM, liftEitherWith, raceAny_, threadDelay', tshow, whenM)
 import Simplex.Messaging.Version
 import System.Mem.Weak (Weak, deRefWeak)
@@ -544,7 +543,6 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
       "" -> case protocolTypeI @(ProtoType msg) of
         SPSMP | smpWebPort -> ("443", transport @TLS)
         _ -> defaultTransport cfg
-      "80" -> ("80", transport @WS)
       p -> (p, transport @TLS)
 
     client :: forall c. Transport c => TProxy c -> PClient v err msg -> TMVar (Either (ProtocolClientError err) (ProtocolClient v err msg)) -> c -> IO ()

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/Protocol.hs

@@ -538,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/Env/STM.hs

@@ -43,7 +43,7 @@ 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
@@ -79,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),
@@ -93,6 +96,9 @@ data ServerConfig = ServerConfig
     serverStatsLogFile :: FilePath,
     -- | file to save and restore stats
     serverStatsBackupFile :: Maybe FilePath,
+    -- | interval and file to save prometheus metrics
+    prometheusInterval :: Maybe Int,
+    prometheusMetricsFile :: FilePath,
     -- | notification delivery interval
     ntfDeliveryInterval :: Int,
     -- | interval between sending pending END events to unsubscribed clients, seconds
@@ -120,9 +126,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
 
@@ -155,23 +164,24 @@ 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 :: 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
   }
 
@@ -179,9 +189,9 @@ type family MsgStore s where
   MsgStore 'MSMemory = STMMsgStore
   MsgStore 'MSJournal = JournalMsgStore
 
-data AMsgStore = forall s. MsgStoreClass (MsgStore s) => AMS (SMSType s) (MsgStore s)
+data AMsgStore = forall s. (STMQueueStore (MsgStore s), MsgStoreClass (MsgStore s)) => AMS (SMSType s) (MsgStore s)
 
-data AMsgQueue = forall s. MsgStoreClass (MsgStore s) => AMQ (SMSType s) (MsgQueue (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))
 
@@ -193,11 +203,11 @@ 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
@@ -209,11 +219,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,
@@ -249,8 +264,8 @@ newServer = do
   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
@@ -276,23 +291,22 @@ newProhibitedSub = do
   return Sub {subThread = ProhibitSub, delivered}
 
 newEnv :: ServerConfig -> IO Env
-newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, msgStoreType, storeMsgsFile, smpAgentCfg, information, messageExpiration, msgQueueQuota, maxJournalMsgCount, maxJournalStateLines} = 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 <- case msgStoreType of
+  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}
+        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
@@ -303,7 +317,7 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, msgSt
   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
@@ -347,3 +361,6 @@ 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

@@ -46,10 +46,11 @@ 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, doesDirectoryExist, doesFileExist)
 import System.Exit (exitFailure)
@@ -85,6 +86,14 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
     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
@@ -99,6 +108,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                 ("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
@@ -116,6 +126,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                 ("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
@@ -137,7 +148,7 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
       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}
+    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"
@@ -238,10 +249,13 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
                 <> ("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")
-                <> "[AUTH]\n\
+                <> "# Log interval for real-time Prometheus metrics\n\
+                   \# prometheus_interval: 300\n\n\
+                   \[AUTH]\n\
                    \# Set new_queues option to off to completely prohibit creating new messaging queues.\n\
                    \# This can be useful when you want to decommission the server, but not all connections are switched yet.\n\
                    \new_queues: on\n\n\
@@ -403,6 +417,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
@@ -417,6 +433,8 @@ smpServerCLI_ generateSite serveStaticFiles attachStaticFiles cfgPath logPath =
               logStatsStartTime = 0, -- seconds from 00:00 UTC
               serverStatsLogFile = combine logPath "smp-server-stats.daily.log",
               serverStatsBackupFile = logStats $> combine logPath "smp-server-stats.log",
+              prometheusInterval = eitherToMaybe $ read . T.unpack <$> lookupValue "STORE_LOG" "prometheus_interval" ini,
+              prometheusMetricsFile = combine logPath "smp-server-metrics.txt",
               pendingENDInterval = 15000000, -- 15 seconds
               ntfDeliveryInterval = 3000000, -- 3 seconds
               smpServerVRange = supportedServerSMPRelayVRange,

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

@@ -12,13 +12,14 @@
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StandaloneDeriving #-}
 {-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE TupleSections #-}
 
 module Simplex.Messaging.Server.MsgStore.Journal
-  ( JournalMsgStore (msgQueues, random),
-    JournalMsgQueue (queue),
+  ( JournalMsgStore (queues, senders, notifiers, random),
+    JournalQueue,
+    JournalMsgQueue (queue, state),
     JMQueue (queueDirectory, statePath),
     JournalStoreConfig (..),
-    getQueueMessages,
     closeMsgQueue,
     closeMsgQueueHandles,
     -- below are exported for tests
@@ -32,6 +33,7 @@ module Simplex.Messaging.Server.MsgStore.Journal
     newJournalId,
     appendState,
     queueLogFileName,
+    journalFilePath,
     logFileExt,
   )
 where
@@ -42,27 +44,28 @@ import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Trans.Except
 import qualified Data.Attoparsec.ByteString.Char8 as A
-import Data.Bitraversable (bimapM)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.List (intercalate)
-import qualified Data.Map.Strict as M
-import Data.Maybe (catMaybes, fromMaybe)
+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 (ErrorType (..), Message (..), 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.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (ifM, tshow, ($>>=))
+import Simplex.Messaging.Server.StoreLog
+import Simplex.Messaging.Util (ifM, tshow, ($>>=), (<$$>))
 import System.Directory
 import System.Exit
 import System.FilePath ((</>))
@@ -74,7 +77,10 @@ data JournalMsgStore = JournalMsgStore
   { config :: JournalStoreConfig,
     random :: TVar StdGen,
     queueLocks :: TMap RecipientId Lock,
-    msgQueues :: TMap RecipientId JournalMsgQueue
+    queues :: TMap RecipientId JournalQueue,
+    senders :: TMap SenderId RecipientId,
+    notifiers :: TMap NotifierId RecipientId,
+    storeLog :: TVar (Maybe (StoreLog 'WriteMode))
   }
 
 data JournalStoreConfig = JournalStoreConfig
@@ -85,12 +91,26 @@ data JournalStoreConfig = JournalStoreConfig
     -- When this limit is reached, the file will be changed.
     -- This number should be set bigger than queue quota.
     maxMsgCount :: Int,
-    maxStateLines :: 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,
-    queueLock :: Lock,
     statePath :: FilePath
   }
 
@@ -198,8 +218,23 @@ 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
 
@@ -207,39 +242,51 @@ instance MsgStoreClass JournalMsgStore where
   newMsgStore config = do
     random <- newTVarIO =<< newStdGen
     queueLocks <- TM.emptyIO
-    msgQueues <- TM.emptyIO
-    pure JournalMsgStore {config, random, queueLocks, msgQueues}
+    queues <- TM.emptyIO
+    senders <- TM.emptyIO
+    notifiers <- TM.emptyIO
+    storeLog <- newTVarIO Nothing
+    pure JournalMsgStore {config, random, queueLocks, queues, senders, notifiers, storeLog}
 
-  closeMsgStore st = atomically (swapTVar (msgQueues st) M.empty) >>= mapM_ closeMsgQueueHandles
+  setStoreLog :: JournalMsgStore -> StoreLog 'WriteMode -> IO ()
+  setStoreLog st sl = atomically $ writeTVar (storeLog st) (Just sl)
 
-  activeMsgQueues = msgQueues
+  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 -> JournalMsgQueue -> IO a) -> IO a
+  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
-        closeMsgStore ms
-        lock <- createLockIO -- the same lock is used for all queues
-        (!count, !res) <- foldQueues 0 (processQueue lock) (0, mempty) ("", storePath)
+        (!count, !res) <- foldQueues 0 processQueue (0, mempty) ("", storePath)
         putStrLn $ progress count
         pure res
       JournalStoreConfig {storePath, pathParts} = config
-      processQueue :: Lock -> (Int, a) -> (String, FilePath) -> IO (Int, a)
-      processQueue queueLock (!i, !r) (queueId, dir) = do
+      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
-        let statePath = msgQueueStatePath dir queueId
-        q <- openMsgQueue ms JMQueue {queueDirectory = dir, queueLock, statePath}
         r' <- case strDecode $ B.pack queueId of
-          Right rId -> action rId q
+          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
-            putStrLn ("Error: message queue directory " <> dir <> " is invalid: " <> e)
+            logError $ "STORE: processQueue, message queue directory " <> T.pack dir <> " is invalid, " <> tshow e
             exitFailure
-        closeMsgQueueHandles q
         pure (i + 1, r <> r')
       progress i = "Processed: " <> show i <> " queues"
       foldQueues depth f acc (queueId, path) = do
@@ -256,25 +303,28 @@ instance MsgStoreClass JournalMsgStore where
               (Nothing <$ putStrLn ("Error: path " <> path' <> " is not a directory, skipping"))
 
   logQueueStates :: JournalMsgStore -> IO ()
-  logQueueStates ms = withActiveMsgQueues ms $ \_ -> logQueueState
+  logQueueStates ms = withActiveMsgQueues ms $ \_ -> unStoreIO . logQueueState
 
-  logQueueState :: JournalMsgQueue -> IO ()
-  logQueueState q = 
-    readTVarIO (handles q)
-      >>= maybe (pure ()) (\hs -> readTVarIO (state q) >>= appendState (stateHandle hs))
+  logQueueState :: JournalQueue -> StoreIO ()
+  logQueueState q =
+    StoreIO . void $
+      readTVarIO (msgQueue_ q)
+        $>>= \mq -> readTVarIO (handles mq)
+        $>>= (\hs -> (readTVarIO (state mq) >>= appendState (stateHandle hs)) $> Just ())
 
-  getMsgQueue :: JournalMsgStore -> RecipientId -> ExceptT ErrorType IO JournalMsgQueue
-  getMsgQueue ms@JournalMsgStore {queueLocks, msgQueues, random} rId =
-    tryStore "getMsgQueue" (B.unpack $ strEncode rId) $ withLockMap queueLocks rId "getMsgQueue" $
-      TM.lookupIO rId msgQueues >>= maybe newQ pure
+  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
-        queueLock <- atomically $ getMapLock queueLocks rId
         let dir = msgQueueDirectory ms rId
             statePath = msgQueueStatePath dir $ B.unpack (strEncode rId)
-            queue = JMQueue {queueDirectory = dir, queueLock, statePath}
+            queue = JMQueue {queueDirectory = dir, statePath}
         q <- ifM (doesDirectoryExist dir) (openMsgQueue ms queue) (createQ queue)
-        atomically $ TM.insert rId q msgQueues
+        atomically $ writeTVar msgQueue_ $ Just q
         pure q
         where
           createQ :: JMQueue -> IO JournalMsgQueue
@@ -282,23 +332,62 @@ instance MsgStoreClass JournalMsgStore where
             -- 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)
+            mkJournalQueue queue (newMsgQueueState journalId) Nothing
 
-  delMsgQueue :: JournalMsgStore -> RecipientId -> IO ()
-  delMsgQueue ms rId = withLockMap (queueLocks ms) rId "delMsgQueue" $ do
-    closeMsgQueue ms rId
-    removeQueueDirectory ms rId
+  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
 
-  delMsgQueueSize :: JournalMsgStore -> RecipientId -> IO Int
-  delMsgQueueSize ms rId = withLockMap (queueLocks ms) rId "delMsgQueue" $ do
-    st_ <-
-      atomically (TM.lookupDelete rId (msgQueues ms))
-        >>= mapM (\q -> closeMsgQueueHandles q >> readTVarIO (state q))
-    removeQueueDirectory ms rId
-    pure $ maybe (-1) size st_
+  -- 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)
 
-  getQueueMessages :: Bool -> JournalMsgQueue -> IO [Message]
-  getQueueMessages drainMsgs q = run []
+  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
@@ -308,22 +397,24 @@ instance MsgStoreClass JournalMsgStore where
             updateReadPos q drainMsgs len hs
             (msg :) <$> run msgs
 
-  writeMsg :: JournalMsgStore -> JournalMsgQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
-  writeMsg ms q@JournalMsgQueue {queue = JMQueue {queueDirectory, statePath}, handles} logState msg =
-    isolateQueue q "writeMsg" $ StoreIO $ do
+  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 st canWrt' msg $> Just (msg, empty)
-            else writeToJournal st canWrt' msgQuota $> Nothing
+            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 = msgId msg, msgTs = msgTs msg}
-      writeToJournal st@MsgQueueState {writeState, readState = rs, size} canWrt' !msg' = do
+      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
@@ -339,6 +430,7 @@ instance MsgStoreClass JournalMsgStore where
         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
@@ -354,15 +446,17 @@ instance MsgStoreClass JournalMsgStore where
             pure (newJournalState journalId, wh)
 
   -- can ONLY be used while restoring messages, not while server running
-  setOverQuota_ :: JournalMsgQueue -> IO ()
-  setOverQuota_ JournalMsgQueue {state} = atomically $ modifyTVar' state $ \st -> st {canWrite = False}
+  setOverQuota_ :: JournalQueue -> IO ()
+  setOverQuota_ q =
+    readTVarIO (msgQueue_ q)
+      >>= mapM_ (\JournalMsgQueue {state} -> atomically $ modifyTVar' state $ \st -> st {canWrite = False})
 
-  getQueueSize :: JournalMsgQueue -> IO Int
-  getQueueSize JournalMsgQueue {state} = size <$> readTVarIO state
+  getQueueSize_ :: JournalMsgQueue -> StoreIO Int
+  getQueueSize_ JournalMsgQueue {state} = StoreIO $ size <$> readTVarIO state
 
-  tryPeekMsg_ :: JournalMsgQueue -> StoreIO (Maybe Message)
-  tryPeekMsg_ q@JournalMsgQueue {tipMsg, handles} =
-    StoreIO $ readTVarIO handles $>>= chooseReadJournal q True $>>= peekMsg
+  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
@@ -370,36 +464,48 @@ instance MsgStoreClass JournalMsgStore where
             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_ :: JournalMsgQueue -> Bool -> StoreIO ()
-  tryDeleteMsg_ q@JournalMsgQueue {tipMsg, handles} logState = StoreIO $
+  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 q logState len hs $> Just ()
+        $>>= \hs -> updateReadPos mq logState len hs $> Just ()
 
-  isolateQueue :: JournalMsgQueue -> String -> StoreIO a -> ExceptT ErrorType IO a
-  isolateQueue JournalMsgQueue {queue = q} op =
-    tryStore op (queueDirectory q) . withLock' (queueLock q) op . unStoreIO
+  isolateQueue :: RecipientId -> JournalQueue -> String -> StoreIO a -> ExceptT ErrorType IO a
+  isolateQueue rId JournalQueue {queueLock} op =
+    tryStore' op rId . withLock' queueLock op . unStoreIO
 
-tryStore :: String -> String -> IO a -> ExceptT ErrorType IO a
-tryStore op qId a = ExceptT $ E.mask_ $ E.try a >>= bimapM storeErr pure
+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 ErrorType
+    storeErr :: E.SomeException -> IO (Either ErrorType a)
     storeErr e =
-      let e' = intercalate ", " [op, qId, show e]
-       in logError ("STORE: " <> T.pack e') $> STORE 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 dir st sh
+  (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 q st' (Just hs)
 
-mkJournalQueue :: JMQueue -> (MsgQueueState, Maybe MsgQueueHandles) -> IO JournalMsgQueue
-mkJournalQueue queue (st, hs_) = do
+mkJournalQueue :: JMQueue -> MsgQueueState -> Maybe MsgQueueHandles -> IO JournalMsgQueue
+mkJournalQueue queue st hs_ = do
   state <- newTVarIO st
   tipMsg <- newTVarIO Nothing
   handles <- newTVarIO hs_
@@ -464,17 +570,26 @@ createNewJournal dir journalId = do
 newJournalId :: TVar StdGen -> IO ByteString
 newJournalId g = strEncode <$> atomically (stateTVar g $ genByteString 12)
 
-openJournals :: FilePath -> MsgQueueState -> Handle -> IO (MsgQueueState, Handle, Maybe Handle)
-openJournals dir st@MsgQueueState {readState = rs, writeState = ws} sh = do
+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 -> do
-      logError $ "STORE: openJournals, no read file - creating new file, " <> T.pack path
-      rh <- createNewJournal dir rjId
-      let st' = newMsgQueueState rjId
-      closeOnException rh $ appendState sh st'
-      pure (st', rh, Nothing)
+    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
@@ -483,16 +598,23 @@ openJournals dir st@MsgQueueState {readState = rs, writeState = ws} sh = do
           fixFileSize rh $ byteCount rs
           openJournal ws >>= \case
             Left path -> do
-              logError $ "STORE: openJournals, no write file - creating new file, " <> T.pack path
-              wh <- createNewJournal dir wjId
-              let size' = msgCount rs - msgPos rs
-                  st' = st {writeState = newJournalState wjId, size = size'} -- we don't amend canWrite to trigger QCONT
-              closeOnException wh $ appendState sh st'
-              pure (st', rh, Just wh)
+              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
@@ -530,7 +652,7 @@ readWriteQueueState JournalMsgStore {random, config} statePath =
   where
     tempBackup = statePath <> ".bak"
     readQueueState = do
-      ls <- LB.lines <$> LB.readFile statePath
+      ls <- B.lines <$> readFileTail
       case ls of
         [] -> writeNewQueueState
         _ -> do
@@ -541,7 +663,7 @@ readWriteQueueState JournalMsgStore {random, config} statePath =
       logWarn $ "STORE: readWriteQueueState, empty queue state - initialized, " <> T.pack statePath
       st <- newMsgQueueState <$> newJournalId random
       writeQueueState st
-    useLastLine len isLastLine ls = case strDecode $ LB.toStrict $ last ls of
+    useLastLine len isLastLine ls = case strDecode $ last ls of
       Right st
         | len > maxStateLines config || not isLastLine ->
             backupWriteQueueState st
@@ -571,6 +693,14 @@ readWriteQueueState JournalMsgStore {random, config} statePath =
       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}
@@ -591,10 +721,18 @@ validQueueState MsgQueueState {readState = rs, writeState = ws, size}
         && msgPos ws == msgCount ws
         && bytePos ws == byteCount ws
 
-closeMsgQueue :: JournalMsgStore -> RecipientId -> IO ()
-closeMsgQueue ms rId =
-  atomically (TM.lookupDelete rId (msgQueues ms))
-    >>= mapM_ closeMsgQueueHandles
+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
@@ -605,9 +743,12 @@ closeMsgQueueHandles q = readTVarIO (handles q) >>= mapM_ closeHandles
       mapM_ hClose wh_
 
 removeQueueDirectory :: JournalMsgStore -> RecipientId -> IO ()
-removeQueueDirectory st rId =
-  let dir = msgQueueDirectory st rId
-   in removePathForcibly dir `catchAny` (\e -> logError $ "STORE: removeQueueDirectory, " <> T.pack dir <> ", " <> tshow e)
+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
@@ -632,7 +773,10 @@ openFile f mode = do
   pure h
 
 hClose :: Handle -> IO ()
-hClose h = IO.hClose h `catchAny` (\e -> logError $ "STORE: hClose, error closing file, " <> tshow e)
+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,5 +1,5 @@
-{-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE ConstraintKinds #-}
+{-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE FlexibleInstances #-}
@@ -8,10 +8,10 @@
 {-# LANGUAGE MultiParamTypeClasses #-}
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE TypeFamilies #-}
+{-# LANGUAGE TupleSections #-}
 
 module Simplex.Messaging.Server.MsgStore.STM
   ( STMMsgStore (..),
-    STMMsgQueue (msgQueue),
     STMStoreConfig (..),
   )
 where
@@ -20,21 +20,36 @@ import Control.Concurrent.STM
 import Control.Monad.IO.Class
 import Control.Monad.Trans.Except
 import Data.Functor (($>))
-import Simplex.Messaging.Protocol (ErrorType, Message (..), RecipientId)
+import Data.Int (Int64)
+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
-
-data STMMsgQueue = STMMsgQueue
-  { msgQueue :: TQueue Message,
-    quota :: Int,
-    canWrite :: TVar Bool,
-    size :: TVar Int
-  }
+import Simplex.Messaging.Util ((<$$>), ($>>=))
+import System.IO (IOMode (..))
 
 data STMMsgStore = STMMsgStore
   { storeConfig :: STMStoreConfig,
-    msgQueues :: TMap RecipientId STMMsgQueue
+    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,
+    canWrite :: TVar Bool,
+    size :: TVar Int
   }
 
 data STMStoreConfig = STMStoreConfig
@@ -42,19 +57,34 @@ data STMStoreConfig = STMStoreConfig
     quota :: Int
   }
 
+instance STMQueueStore STMMsgStore where
+  queues' = queues
+  senders' = senders
+  notifiers' = notifiers
+  storeLog' = storeLog
+  mkQueue _ qr = STMQueue <$> newTVar (Just qr) <*> newTVar Nothing
+  msgQueue_' = msgQueue_
+
 instance MsgStoreClass STMMsgStore where
   type StoreMonad STMMsgStore = STM
+  type StoreQueue STMMsgStore = STMQueue
   type MsgQueue STMMsgStore = STMMsgQueue
   type MsgStoreConfig STMMsgStore = STMStoreConfig
 
   newMsgStore :: STMStoreConfig -> IO STMMsgStore
   newMsgStore storeConfig = do
-    msgQueues <- TM.emptyIO
-    pure STMMsgStore {storeConfig, msgQueues}
+    queues <- TM.emptyIO
+    senders <- TM.emptyIO
+    notifiers <- TM.emptyIO
+    storeLog <- newTVarIO Nothing
+    pure STMMsgStore {storeConfig, queues, senders, notifiers, storeLog}
 
-  closeMsgStore _ = pure ()
+  setStoreLog :: STMMsgStore -> StoreLog 'WriteMode -> IO ()
+  setStoreLog st sl = atomically $ writeTVar (storeLog st) (Just sl)
 
-  activeMsgQueues = msgQueues
+  closeMsgStore st = readTVarIO (storeLog st) >>= mapM_ closeStoreLog
+
+  activeMsgQueues = queues
   {-# INLINE activeMsgQueues #-}
 
   withAllMsgQueues _ = withActiveMsgQueues
@@ -64,39 +94,52 @@ instance MsgStoreClass STMMsgStore where
 
   logQueueState _ = pure ()
 
-  -- 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 -> ExceptT ErrorType IO STMMsgQueue
-  getMsgQueue STMMsgStore {msgQueues = qs, storeConfig = STMStoreConfig {quota}} rId =
-    liftIO $ TM.lookupIO rId qs >>= maybe (atomically maybeNewQ) pure
+  queueRec' = queueRec
+  {-# INLINE queueRec' #-}
+
+  getMsgQueue :: STMMsgStore -> RecipientId -> STMQueue -> STM STMMsgQueue
+  getMsgQueue _ _ STMQueue {msgQueue_} = readTVar msgQueue_ >>= maybe newQ pure
     where
-      maybeNewQ = TM.lookup rId qs >>= maybe newQ pure
       newQ = do
         msgQueue <- newTQueue
         canWrite <- newTVar True
         size <- newTVar 0
-        let q = STMMsgQueue {msgQueue, quota, canWrite, size}
-        TM.insert rId q qs
+        let q = STMMsgQueue {msgQueue, canWrite, size}
+        writeTVar msgQueue_ (Just q)
         pure q
 
-  delMsgQueue :: STMMsgStore -> RecipientId -> IO ()
-  delMsgQueue st rId = atomically $ TM.delete rId $ msgQueues st
+  getPeekMsgQueue :: STMMsgStore -> RecipientId -> STMQueue -> STM (Maybe (STMMsgQueue, Message))
+  getPeekMsgQueue _ _ q@STMQueue {msgQueue_} = readTVar msgQueue_ $>>= \mq -> (mq,) <$$> tryPeekMsg_ q mq
 
-  delMsgQueueSize :: STMMsgStore -> RecipientId -> IO Int
-  delMsgQueueSize st rId = atomically (TM.lookupDelete rId $ msgQueues st) >>= maybe (pure 0) (\STMMsgQueue {size} -> readTVarIO size)
+  -- 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)
 
-  getQueueMessages :: Bool -> STMMsgQueue -> IO [Message]
-  getQueueMessages drainMsgs = atomically . (if drainMsgs then flushTQueue else snapshotTQueue) . msgQueue
+  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 -> STMMsgQueue -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
-  writeMsg _ STMMsgQueue {msgQueue = q, quota, canWrite, size} _logState msg = liftIO $ atomically $ do
+  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
@@ -109,23 +152,24 @@ instance MsgStoreClass STMMsgStore where
           else (writeTQueue q $! msgQuota) $> Nothing
       else pure Nothing
     where
-      msgQuota = MessageQuota {msgId = msgId msg, msgTs = msgTs msg}
+      STMMsgStore {storeConfig = STMStoreConfig {quota}} = ms
+      msgQuota = MessageQuota {msgId = messageId msg, msgTs = messageTs msg}
 
-  setOverQuota_ :: STMMsgQueue -> IO ()
-  setOverQuota_ q = atomically $ writeTVar (canWrite q) False
+  setOverQuota_ :: STMQueue -> IO ()
+  setOverQuota_ q = readTVarIO (msgQueue_ q) >>= mapM_ (\mq -> atomically $ writeTVar (canWrite mq) False)
 
-  getQueueSize :: STMMsgQueue -> IO Int
-  getQueueSize STMMsgQueue {size} = readTVarIO size
+  getQueueSize_ :: STMMsgQueue -> STM Int
+  getQueueSize_ STMMsgQueue {size} = readTVar size
 
-  tryPeekMsg_ :: STMMsgQueue -> STM (Maybe Message)
-  tryPeekMsg_ = tryPeekTQueue . msgQueue
+  tryPeekMsg_ :: STMQueue -> STMMsgQueue -> STM (Maybe Message)
+  tryPeekMsg_ _ = tryPeekTQueue . msgQueue
   {-# INLINE tryPeekMsg_ #-}
 
-  tryDeleteMsg_ :: STMMsgQueue -> Bool -> STM ()
-  tryDeleteMsg_ STMMsgQueue {msgQueue = q, size} _logState =
+  tryDeleteMsg_ :: STMQueue -> STMMsgQueue -> Bool -> STM ()
+  tryDeleteMsg_ _ STMMsgQueue {msgQueue = q, size} _logState =
     tryReadTQueue q >>= \case
       Just _ -> modifyTVar' size (subtract 1)
       _ -> pure ()
 
-  isolateQueue :: STMMsgQueue -> String -> STM a -> ExceptT ErrorType IO a
-  isolateQueue _ _ = liftIO . atomically
+  isolateQueue :: RecipientId -> STMQueue -> String -> STM a -> ExceptT ErrorType IO a
+  isolateQueue _ _ _ = liftIO . atomically

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

@@ -1,44 +1,69 @@
 {-# 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 (ErrorType, Message (..), MsgId, RecipientId)
+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 (MsgQueue s)
-  withAllMsgQueues :: Monoid a => Bool -> s -> (RecipientId -> MsgQueue s -> IO a) -> IO a
+  activeMsgQueues :: s -> TMap RecipientId (StoreQueue s)
+  withAllMsgQueues :: Monoid a => Bool -> s -> (RecipientId -> StoreQueue s -> IO a) -> IO a
   logQueueStates :: s -> IO ()
-  logQueueState :: MsgQueue s -> IO ()
-  getMsgQueue :: s -> RecipientId -> ExceptT ErrorType IO (MsgQueue s)
-  delMsgQueue :: s -> RecipientId -> IO ()
-  delMsgQueueSize :: s -> RecipientId -> IO Int
-  getQueueMessages :: Bool -> MsgQueue s -> IO [Message]
-  writeMsg :: s -> MsgQueue s -> Bool -> Message -> ExceptT ErrorType IO (Maybe (Message, Bool))
-  setOverQuota_ :: MsgQueue s -> IO () -- can ONLY be used while restoring messages, not while server running
-  getQueueSize :: MsgQueue s -> IO Int
-  tryPeekMsg_ :: MsgQueue s -> StoreMonad s (Maybe Message)
-  tryDeleteMsg_ :: MsgQueue s -> Bool -> StoreMonad s ()
-  isolateQueue :: MsgQueue s -> String -> StoreMonad s a -> ExceptT ErrorType IO a
+  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
 
@@ -48,42 +73,69 @@ data SMSType :: MSType -> Type where
 
 data AMSType = forall s. AMSType (SMSType s)
 
-withActiveMsgQueues :: (MsgStoreClass s, Monoid a) => s -> (RecipientId -> MsgQueue s -> IO a) -> IO a
+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
 
-tryPeekMsg :: MsgStoreClass s => MsgQueue s -> ExceptT ErrorType IO (Maybe Message)
-tryPeekMsg mq = isolateQueue mq "tryPeekMsg" $ tryPeekMsg_ mq
+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 => MsgQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message)
-tryDelMsg mq msgId' =
-  isolateQueue mq "tryDelMsg" $
-    tryPeekMsg_ mq >>= \case
-      msg_@(Just msg)
-        | msgId msg == msgId' ->
-            tryDeleteMsg_ mq True >> pure msg_
-      _ -> pure Nothing
+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 => MsgQueue s -> MsgId -> ExceptT ErrorType IO (Maybe Message, Maybe Message)
-tryDelPeekMsg mq msgId' =
-  isolateQueue mq "tryDelPeekMsg" $
-    tryPeekMsg_ mq >>= \case
-      msg_@(Just msg)
-        | msgId msg == msgId' -> (msg_,) <$> (tryDeleteMsg_ mq True >> tryPeekMsg_ mq)
-        | otherwise -> pure (Nothing, msg_)
-      _ -> pure (Nothing, Nothing)
+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)
 
-deleteExpiredMsgs :: MsgStoreClass s => MsgQueue s -> Bool -> Int64 -> ExceptT ErrorType IO Int
-deleteExpiredMsgs mq logState old = isolateQueue mq "deleteExpiredMsgs" $ loop 0
+-- 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_ mq >>= \case
+      tryPeekMsg_ q mq >>= \case
         Just Message {msgTs}
           | systemSeconds msgTs < old ->
-              tryDeleteMsg_ mq logState >> loop (dc + 1)
+              tryDeleteMsg_ q mq False >> loop (dc + 1)
         _ -> pure dc

src/Simplex/Messaging/Server/Prometheus.hs

@@ -0,0 +1,379 @@
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedStrings #-}
+{-# LANGUAGE TypeApplications #-}
+{-# OPTIONS_GHC -fno-warn-unrecognised-pragmas #-}
+
+module Simplex.Messaging.Server.Prometheus where
+
+import Data.Int (Int64)
+import Data.Text (Text)
+import qualified Data.Text as T
+import Data.Time.Clock (UTCTime (..), diffUTCTime)
+import Data.Time.Clock.System (systemEpochDay)
+import Data.Time.Format.ISO8601 (iso8601Show)
+import Network.Socket (ServiceName)
+import Simplex.Messaging.Server.Stats
+import Simplex.Messaging.Transport.Server (SocketStats (..))
+
+data ServerMetrics = ServerMetrics
+  { statsData :: ServerStatsData,
+    activeQueueCounts :: PeriodStatCounts,
+    activeNtfCounts :: PeriodStatCounts,
+    queueCount :: Int,
+    notifierCount :: Int
+  }
+
+data RealTimeMetrics = RealTimeMetrics
+  { socketStats :: [(ServiceName, SocketStats)],
+    threadsCount :: Int,
+    clientsCount :: Int,
+    smpSubsCount :: Int,
+    smpSubClientsCount :: Int,
+    ntfSubsCount :: Int,
+    ntfSubClientsCount :: Int
+  }
+
+{-# FOURMOLU_DISABLE\n#-}
+prometheusMetrics :: ServerMetrics -> RealTimeMetrics -> UTCTime -> Text
+prometheusMetrics sm rtm ts =
+  time <> queues <> subscriptions <> messages <> ntfMessages <> ntfs <> relays <> info
+  where
+    ServerMetrics {statsData, activeQueueCounts = ps, activeNtfCounts = psNtf, queueCount, notifierCount} = sm
+    RealTimeMetrics
+      { socketStats,
+        threadsCount,
+        clientsCount,
+        smpSubsCount,
+        smpSubClientsCount,
+        ntfSubsCount,
+        ntfSubClientsCount
+      } = rtm
+    ServerStatsData
+      { _fromTime,
+        _qCreated,
+        _qSecured,
+        _qDeletedAll,
+        _qDeletedAllB,
+        _qDeletedNew,
+        _qDeletedSecured,
+        _qSub,
+        _qSubAllB,
+        _qSubAuth,
+        _qSubDuplicate,
+        _qSubProhibited,
+        _qSubEnd,
+        _qSubEndB,
+        _ntfCreated,
+        _ntfDeleted,
+        _ntfDeletedB,
+        _ntfSub,
+        _ntfSubB,
+        _ntfSubAuth,
+        _ntfSubDuplicate,
+        _msgSent,
+        _msgSentAuth,
+        _msgSentQuota,
+        _msgSentLarge,
+        _msgRecv,
+        _msgRecvGet,
+        _msgGet,
+        _msgGetNoMsg,
+        _msgGetAuth,
+        _msgGetDuplicate,
+        _msgGetProhibited,
+        _msgExpired,
+        _activeQueues,
+        _msgSentNtf,
+        _msgRecvNtf,
+        _activeQueuesNtf,
+        _msgNtfs,
+        _msgNtfsB,
+        _msgNtfNoSub,
+        _msgNtfLost,
+        _msgNtfExpired,
+        _pRelays,
+        _pRelaysOwn,
+        _pMsgFwds,
+        _pMsgFwdsOwn,
+        _pMsgFwdsRecv,
+        _qCount,
+        _msgCount,
+        _ntfCount
+      } = statsData
+    time =
+      "# Recorded at: " <> T.pack (iso8601Show ts) <> "\n\
+      \# Stats from: " <> T.pack (iso8601Show _fromTime) <> "\n\
+      \\n"
+    queues =
+      "# Queues\n\
+      \# ------\n\
+      \\n\
+      \# HELP simplex_smp_queues_created Created queues\n\
+      \# TYPE simplex_smp_queues_created counter\n\
+      \simplex_smp_queues_created " <> mshow _qCreated <> "\n# qCreated\n\
+      \\n\
+      \# HELP simplex_smp_queues_secured Secured queues\n\
+      \# TYPE simplex_smp_queues_secured counter\n\
+      \simplex_smp_queues_secured " <> mshow _qSecured <> "\n# qSecured\n\
+      \\n\
+      \# HELP simplex_smp_queues_deleted Deleted queues\n\
+      \# TYPE simplex_smp_queues_deleted counter\n\
+      \simplex_smp_queues_deleted{type=\"all\"} " <> mshow _qDeletedAll <> "\n# qDeleted\n\
+      \simplex_smp_queues_deleted{type=\"new\"} " <> mshow _qDeletedNew <> "\n# qDeletedNew\n\
+      \simplex_smp_queues_deleted{type=\"secured\"} " <> mshow _qDeletedSecured <> "\n# qDeletedSecured\n\
+      \\n\
+      \# HELP simplex_smp_queues_deleted_batch Batched requests to delete queues\n\
+      \# TYPE simplex_smp_queues_deleted_batch counter\n\
+      \simplex_smp_queues_deleted_batch " <> mshow _qDeletedAllB <> "\n# qDeletedAllB\n\
+      \\n\
+      \# HELP simplex_smp_queues_total1 Total number of stored queues (first type of count).\n\
+      \# TYPE simplex_smp_queues_total1 gauge\n\
+      \simplex_smp_queues_total1 " <> mshow _qCount <> "\n# qCount\n\
+      \\n\
+      \# HELP simplex_smp_queues_total2 Total number of stored queues (second type of count).\n\
+      \# TYPE simplex_smp_queues_total2 gauge\n\
+      \simplex_smp_queues_total2 " <> mshow queueCount <> "\n# qCount2\n\
+      \\n\
+      \# HELP simplex_smp_queues_daily Daily active queues.\n\
+      \# TYPE simplex_smp_queues_daily gauge\n\
+      \simplex_smp_queues_daily " <> mstr (dayCount ps) <> "\n# dayMsgQueues\n\
+      \\n\
+      \# HELP simplex_smp_queues_weekly Weekly active queues.\n\
+      \# TYPE simplex_smp_queues_weekly gauge\n\
+      \simplex_smp_queues_weekly " <> mstr (weekCount ps) <> "\n# weekMsgQueues\n\
+      \\n\
+      \# HELP simplex_smp_queues_monthly Monthly active queues.\n\
+      \# TYPE simplex_smp_queues_monthly gauge\n\
+      \simplex_smp_queues_monthly " <> mstr (monthCount ps) <> "\n# monthMsgQueues\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_daily Daily active queues with notifications.\n\
+      \# TYPE simplex_smp_queues_notify_daily gauge\n\
+      \simplex_smp_queues_notify_daily " <> mstr (dayCount psNtf) <> "\n# dayCountNtf\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_weekly Weekly active queues with notifications.\n\
+      \# TYPE simplex_smp_queues_notify_weekly gauge\n\
+      \simplex_smp_queues_notify_weekly " <> mstr (weekCount psNtf) <> "\n# weekCountNtf\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_monthly Monthly active queues with notifications.\n\
+      \# TYPE simplex_smp_queues_notify_monthly gauge\n\
+      \simplex_smp_queues_notify_monthly " <> mstr (monthCount psNtf) <> "\n# monthCountNtf\n\
+      \\n"
+    subscriptions =
+      "# Subscriptions\n\
+      \# -------------\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_successes Successful subscriptions.\n\
+      \# TYPE simplex_smp_subscribtion_successes counter\n\
+      \simplex_smp_subscribtion_successes " <> mshow _qSub <> "\n# qSub\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_successes_batch Batched successful subscriptions.\n\
+      \# TYPE simplex_smp_subscribtion_successes_batch counter\n\
+      \simplex_smp_subscribtion_successes_batch " <> mshow _qSubAllB <> "\n# qSubAllB\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_end Ended subscriptions.\n\
+      \# TYPE simplex_smp_subscribtion_end counter\n\
+      \simplex_smp_subscribtion_end " <> mshow _qSubEnd <> "\n# qSubEnd\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_end_batch Batched ended subscriptions.\n\
+      \# TYPE simplex_smp_subscribtion_end_batch counter\n\
+      \simplex_smp_subscribtion_end_batch " <> mshow _qSubEndB <> "\n# qSubEndB\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_errors Subscription errors.\n\
+      \# TYPE simplex_smp_subscribtion_errors counter\n\
+      \simplex_smp_subscribtion_errors{type=\"auth\"} " <> mshow _qSubAuth <> "\n# qSubAuth\n\
+      \simplex_smp_subscribtion_errors{type=\"duplicate\"} " <> mshow _qSubDuplicate <> "\n# qSubDuplicate\n\
+      \simplex_smp_subscribtion_errors{type=\"prohibited\"} " <> mshow _qSubProhibited <> "\n# qSubProhibited\n\
+      \\n"
+    messages =
+      "# Messages\n\
+      \# --------\n\
+      \\n\
+      \# HELP simplex_smp_messages_sent Sent messages.\n\
+      \# TYPE simplex_smp_messages_sent counter\n\
+      \simplex_smp_messages_sent " <> mshow _msgSent <> "\n# msgSent\n\
+      \\n\
+      \# HELP simplex_smp_messages_sent_errors Total number of messages errors by type.\n\
+      \# TYPE simplex_smp_messages_sent_errors counter\n\
+      \simplex_smp_messages_sent_errors{type=\"auth\"} " <> mshow _msgSentAuth <> "\n# msgSentAuth\n\
+      \simplex_smp_messages_sent_errors{type=\"quota\"} " <> mshow _msgSentQuota <> "\n# msgSentQuota\n\
+      \simplex_smp_messages_sent_errors{type=\"large\"} " <> mshow _msgSentLarge <> "\n# msgSentLarge\n\
+      \\n\
+      \# HELP simplex_smp_messages_received Received messages.\n\
+      \# TYPE simplex_smp_messages_received counter\n\
+      \simplex_smp_messages_received " <> mshow _msgRecv <> "\n# msgRecv\n\
+      \\n\
+      \# HELP simplex_smp_messages_expired Expired messages.\n\
+      \# TYPE simplex_smp_messages_expired counter\n\
+      \simplex_smp_messages_expired " <> mshow _msgExpired <> "\n# msgExpired\n\
+      \\n\
+      \# HELP simplex_smp_messages_total Total number of messages stored.\n\
+      \# TYPE simplex_smp_messages_total gauge\n\
+      \simplex_smp_messages_total " <> mshow _msgCount <> "\n# msgCount\n\
+      \\n"
+    ntfMessages =
+      "# Notification messages (client)\n\
+      \# ------------------------------\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_sent Sent messages with notification flag (cleint).\n\
+      \# TYPE simplex_smp_messages_notify_sent counter\n\
+      \simplex_smp_messages_notify_sent " <> mshow _msgSentNtf <> "\n# msgSentNtf\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_received Received messages with notification flag (client).\n\
+      \# TYPE simplex_smp_messages_notify_received counter\n\
+      \simplex_smp_messages_notify_received " <> mshow _msgRecvNtf <> "\n# msgRecvNtf\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_get_sent Requests to get messages with notification flag (client).\n\
+      \# TYPE simplex_smp_messages_notify_get_sent counter\n\
+      \simplex_smp_messages_notify_get_sent " <> mshow _msgGet <> "\n# msgGet\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_get_received Succesfully received get requests messages with notification flag (client).\n\
+      \# TYPE simplex_smp_messages_notify_get_received counter\n\
+      \simplex_smp_messages_notify_get_received " <> mshow _msgRecvGet <> "\n# msgRecvGet\n\
+      \\n\
+      \# HELP simplex_smp_messages_notify_get_errors Error events with messages with notification flag (client). \n\
+      \# TYPE simplex_smp_messages_notify_get_errors counter\n\
+      \simplex_smp_messages_notify_get_errors{type=\"nomsg\"} " <> mshow _msgGetNoMsg <> "\n# msgGetNoMsg\n\
+      \simplex_smp_messages_notify_get_errors{type=\"auth\"} " <> mshow _msgGetAuth <> "\n# msgGetAuth\n\
+      \simplex_smp_messages_notify_get_errors{type=\"duplicate\"} " <> mshow _msgGetDuplicate <> "\n# msgGetDuplicate\n\
+      \simplex_smp_messages_notify_get_errors{type=\"prohibited\"} " <> mshow _msgGetProhibited <> "\n# msgGetProhibited\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_created Created queues with notification flag (client).\n\
+      \# TYPE simplex_smp_queues_notify_created counter\n\
+      \simplex_smp_queues_notify_created " <> mshow _ntfCreated <> "\n# ntfCreated\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_deleted Deleted queues with notification flag (client).\n\
+      \# TYPE simplex_smp_queues_notify_deleted counter\n\
+      \simplex_smp_queues_notify_deleted " <> mshow _ntfDeleted <> "\n# ntfDeleted\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_deleted_batch Deleted batched queues with notification flag (client).\n\
+      \# TYPE simplex_smp_queues_notify_deleted_batch counter\n\
+      \simplex_smp_queues_notify_deleted_batch " <> mshow _ntfDeletedB <> "\n# ntfDeletedB\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_total1 Total number of stored queues with notification flag (first type of count).\n\
+      \# TYPE simplex_smp_queues_notify_total1 gauge\n\
+      \simplex_smp_queues_notify_total1 " <> mshow _ntfCount <> "\n# ntfCount1\n\
+      \\n\
+      \# HELP simplex_smp_queues_notify_total2 Total number of stored queues with notification flag (second type of count).\n\
+      \# TYPE simplex_smp_queues_notify_total2 gauge\n\
+      \simplex_smp_queues_notify_total2 " <> mshow notifierCount <> "\n# ntfCount2\n\
+      \\n"
+    ntfs =
+      "# Notifications (server)\n\
+      \# ----------------------\n\
+      \\n\
+      \# HELP simplex_smp_messages_ntf_successes Successful events with notification messages (to ntf server). \n\
+      \# TYPE simplex_smp_messages_ntf_successes counter\n\
+      \simplex_smp_messages_ntf_successes " <> mshow _msgNtfs <> "\n# msgNtfs\n\
+      \\n\
+      \# HELP simplex_smp_messages_ntf_successes_batch Successful batched events with notification messages (to ntf server). \n\
+      \# TYPE simplex_smp_messages_ntf_successes_batch counter\n\
+      \simplex_smp_messages_ntf_successes_batch " <> mshow _msgNtfsB <> "\n# msgNtfsB\n\
+      \\n\
+      \# HELP simplex_smp_messages_ntf_errors Error events with notification messages (to ntf server). \n\
+      \# TYPE simplex_smp_messages_ntf_errors counter\n\
+      \simplex_smp_messages_ntf_errors{type=\"nosub\"} " <> mshow _msgNtfNoSub <> "\n# msgNtfNoSub\n\
+      \simplex_smp_messages_ntf_errors{type=\"lost\"} " <> mshow _msgNtfLost <> "\n# msgNtfLost\n\
+      \simplex_smp_messages_ntf_errors{type=\"expired\"} " <> mshow _msgNtfExpired <> "\n# msgNtfExpired\n\
+      \\n\
+      \# HELP simplex_smp_subscription_ntf_requests Subscription requests with notification flag (from ntf server). \n\
+      \# TYPE simplex_smp_subscription_ntf_requests counter\n\
+      \simplex_smp_subscription_ntf_requests " <> mshow _ntfSub <> "\n# ntfSub\n\
+      \\n\
+      \# HELP simplex_smp_subscription_ntf_requests_batch Batched subscription requests with notification flag (from ntf server). \n\
+      \# TYPE simplex_smp_subscription_ntf_requests_batch counter\n\
+      \simplex_smp_subscription_ntf_requests_batch " <> mshow _ntfSubB <> "\n# ntfSubB\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_ntf_errors Subscription errors with notification flag (from ntf server). \n\
+      \# TYPE simplex_smp_subscribtion_ntf_errors counter\n\
+      \simplex_smp_subscribtion_ntf_errors{type=\"auth\"} " <> mshow _ntfSubAuth <> "\n# ntfSubAuth\n\
+      \simplex_smp_subscribtion_ntf_errors{type=\"duplicate\"} " <> mshow _ntfSubDuplicate <> "\n# ntfSubDuplicate\n\
+      \\n"
+    relays =
+      "# Relays\n\
+      \# ------\n\
+      \\n\
+      \# HELP simplex_smp_relay_sessions_requests Session requests through relay.\n\
+      \# TYPE simplex_smp_relay_sessions_requests counter\n\
+      \simplex_smp_relay_sessions_requests{source=\"all\"} " <> mshow (_pRequests _pRelays) <> "\n# pRelays_pRequests\n\
+      \simplex_smp_relay_sessions_requests{source=\"own\"} " <> mshow (_pRequests _pRelaysOwn) <> "\n# pRelaysOwn_pRequests\n\
+      \\n\
+      \# HELP simplex_smp_relay_sessions_successes Successful session events through relay.\n\
+      \# TYPE simplex_smp_relay_sessions_successes counter\n\
+      \simplex_smp_relay_sessions_successes{source=\"all\"} " <> mshow (_pSuccesses _pRelays) <> "\n# pRelays_pSuccesses\n\
+      \simplex_smp_relay_sessions_successes{source=\"own\"} " <> mshow (_pSuccesses _pRelaysOwn) <> "\n# pRelaysOwn_pSuccesses\n\
+      \\n\
+      \# HELP simplex_smp_relay_sessions_errors Error session events through relay.\n\
+      \# TYPE simplex_smp_relay_sessions_errors counter\n\
+      \simplex_smp_relay_sessions_errors{source=\"all\",type=\"connect\"} " <> mshow (_pErrorsConnect _pRelays) <> "\n# pRelays_pErrorsConnect\n\
+      \simplex_smp_relay_sessions_errors{source=\"all\",type=\"compat\"} " <> mshow (_pErrorsCompat _pRelays) <> "\n# pRelays_pErrorsCompat\n\
+      \simplex_smp_relay_sessions_errors{source=\"all\",type=\"other\"} " <> mshow (_pErrorsOther _pRelays) <> "\n# pRelays_pErrorsOther\n\
+      \simplex_smp_relay_sessions_errors{source=\"own\",type=\"connect\"} " <> mshow (_pErrorsConnect _pRelaysOwn) <> "\n# pRelaysOwn_pErrorsConnect\n\
+      \simplex_smp_relay_sessions_errors{source=\"own\",type=\"compat\"} " <> mshow (_pErrorsCompat _pRelaysOwn) <> "\n# pRelaysOwn_pErrorsCompat\n\
+      \simplex_smp_relay_sessions_errors{source=\"own\",type=\"other\"} " <> mshow (_pErrorsOther _pRelaysOwn) <> "\n# pRelaysOwn_pErrorsOther\n\
+      \\n\
+      \# HELP simplex_smp_relay_messages_requests Message requests sent through relay.\n\
+      \# TYPE simplex_smp_relay_messages_requests counter\n\
+      \simplex_smp_relay_messages_requests{source=\"all\"} " <> mshow (_pRequests _pMsgFwds) <> "\n# pMsgFwds_pRequests\n\
+      \simplex_smp_relay_messages_requests{source=\"own\"} " <> mshow (_pRequests _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pRequests\n\
+      \\n\
+      \# HELP simplex_smp_relay_messages_successes Successful messages sent through relay.\n\
+      \# TYPE simplex_smp_relay_messages_successes counter\n\
+      \simplex_smp_relay_messages_successes{source=\"all\"} " <> mshow (_pSuccesses _pMsgFwds) <> "\n# pMsgFwds_pSuccesses\n\
+      \simplex_smp_relay_messages_successes{source=\"own\"} " <> mshow (_pSuccesses _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pSuccesses\n\
+      \\n\
+      \# HELP simplex_smp_relay_messages_errors Error events with messages sent through relay.\n\
+      \# TYPE simplex_smp_relay_messages_errors counter\n\
+      \simplex_smp_relay_messages_errors{source=\"all\",type=\"connect\"} " <> mshow (_pErrorsConnect _pMsgFwds) <> "\n# pMsgFwds_pErrorsConnect\n\
+      \simplex_smp_relay_messages_errors{source=\"all\",type=\"compat\"} " <> mshow (_pErrorsCompat _pMsgFwds) <> "\n# pMsgFwds_pErrorsCompat\n\
+      \simplex_smp_relay_messages_errors{source=\"all\",type=\"other\"} " <> mshow (_pErrorsOther _pMsgFwds) <> "\n# pMsgFwds_pErrorsOther\n\
+      \simplex_smp_relay_messages_errors{source=\"own\",type=\"connect\"} " <> mshow (_pErrorsConnect _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pErrorsConnect\n\
+      \simplex_smp_relay_messages_errors{source=\"own\",type=\"compat\"} " <> mshow (_pErrorsCompat _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pErrorsCompat\n\
+      \simplex_smp_relay_messages_errors{source=\"own\",type=\"other\"} " <> mshow (_pErrorsOther _pMsgFwdsOwn) <> "\n# pMsgFwdsOwn_pErrorsOther\n\
+      \\n\
+      \# HELP simplex_smp_relay_messages_received Relay messages statistics.\n\
+      \# TYPE simplex_smp_relay_messages_received counter\n\
+      \simplex_smp_relay_messages_received " <> mshow _pMsgFwdsRecv <> "\n# pMsgFwdsRecv\n\
+      \\n"
+    info =
+      "# Info\n\
+      \# ----\n\
+      \\n"
+      <> socketsMetric socketsAccepted "simplex_smp_sockets_accepted" "Accepted sockets"
+      <> socketsMetric socketsClosed "simplex_smp_sockets_closed" "Closed sockets"
+      <> socketsMetric socketsActive "simplex_smp_sockets_active" "Active sockets"
+      <> socketsMetric socketsLeaked "simplex_smp_sockets_leaked" "Leaked sockets"
+      <> "# HELP simplex_smp_threads_total Threads\n\
+      \# TYPE simplex_smp_threads_total gauge\n\
+      \simplex_smp_threads_total " <> mshow threadsCount <> "\n\
+      \\n\
+      \# HELP simplex_smp_clients_total Clients\n\
+      \# TYPE simplex_smp_clients_total gauge\n\
+      \simplex_smp_clients_total " <> mshow clientsCount <> "\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_total Total subscriptions\n\
+      \# TYPE simplex_smp_subscribtion_total gauge\n\
+      \simplex_smp_subscribtion_total " <> mshow smpSubsCount <> "\n# smpSubs\n\
+      \\n\
+      \# HELP simplex_smp_subscribtion_clients_total Subscribed clients, first counting method\n\
+      \# TYPE simplex_smp_subscribtion_clients_total gauge\n\
+      \simplex_smp_subscribtion_clients_total " <> mshow smpSubClientsCount <> "\n# smpSubClients\n\
+      \\n\
+      \# HELP simplex_smp_subscription_ntf_total Total notification subscripbtions (from ntf server)\n\
+      \# TYPE simplex_smp_subscription_ntf_total gauge\n\
+      \simplex_smp_subscription_ntf_total " <> mshow ntfSubsCount <> "\n# ntfSubs\n\
+      \\n\
+      \# HELP simplex_smp_subscription_ntf_clients_total Total subscribed NTF servers, first counting method\n\
+      \# TYPE simplex_smp_subscription_ntf_clients_total gauge\n\
+      \simplex_smp_subscription_ntf_clients_total " <> mshow ntfSubClientsCount <> "\n# ntfSubClients\n"
+    socketsMetric :: (SocketStats -> Int) -> Text -> Text -> Text
+    socketsMetric sel metric descr =
+      "# HELP " <> metric <> " " <> descr <> "\n"
+        <> "# TYPE " <> metric <> " gauge\n"
+        <> T.concat (map (\(port, ss) -> metric <> "{port=\"" <> T.pack port <> "\"} " <> mshow (sel ss) <> "\n") socketStats)
+        <> "\n"
+    mstr a = T.pack a <> " " <> tsEpoch
+    mshow :: Show a => a -> Text
+    mshow = mstr . show
+    tsEpoch = T.pack $ show @Int64 $ floor @Double $ realToFrac (ts `diffUTCTime` epoch) * 1000
+    epoch = UTCTime systemEpochDay 0
+{-# FOURMOLU_ENABLE\n#-}

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

@@ -1,120 +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, notifiers} q@QueueRec {recipientId = rId, senderId = sId, notifier} = atomically $ do
-  ifM hasId (pure $ Left DUPLICATE_) $ do
-    TM.insertM rId (newTVar q) queues
-    TM.insert sId rId senders
-    forM_ notifier $ \NtfCreds {notifierId} -> TM.insert notifierId rId notifiers
-    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/Stats.hs

@@ -638,6 +638,14 @@ data PeriodStatCounts = PeriodStatCounts
     monthCount :: String
   }
 
+periodStatDataCounts :: PeriodStatsData -> PeriodStatCounts
+periodStatDataCounts PeriodStatsData {_day, _week, _month} =
+  PeriodStatCounts
+    { dayCount = show $ IS.size _day,
+      weekCount = show $ IS.size _week,
+      monthCount = show $ IS.size _month
+    }
+
 periodStatCounts :: PeriodStats -> UTCTime -> IO PeriodStatCounts
 periodStatCounts ps ts = do
   let d = utctDay ts

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 "create" (recipientId q)
-          SecureQueue qId sKey -> secureQueue st qId sKey >>= qError "secure" qId
-          AddNotifier qId ntfCreds -> addQueueNotifier st qId ntfCreds >>= qError "addNotifier" qId
-          SuspendQueue qId -> suspendQueue st qId >>= qError "suspend" qId
-          DeleteQueue qId -> deleteQueue st qId >>= qError "delete" qId
-          DeleteNotifier qId -> deleteQueueNotifier st qId >>= qError "deleteNotifier" qId
-          UpdateTime qId t -> updateQueueTime st qId t
-        printError :: String -> IO ()
-        printError e = B.putStrLn $ "Error parsing log: " <> B.pack e <> " - " <> s
-        qError :: B.ByteString -> RecipientId -> Either ErrorType a -> IO ()
-        qError op (EntityId qId) = \case
-          Left e -> B.putStrLn $ op <> " 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/Transport/Server.hs

@@ -12,7 +12,9 @@ module Simplex.Messaging.Transport.Server
     runTransportServerState,
     runTransportServerState_,
     SocketState,
+    SocketStats (..),
     newSocketState,
+    getSocketStats,
     runTransportServer,
     runTransportServerSocket,
     runLocalTCPServer,
@@ -163,9 +165,24 @@ safeAccept sock =
 
 type SocketState = (TVar Int, TVar Int, TVar (IntMap (Weak ThreadId)))
 
+data SocketStats = SocketStats
+  { socketsAccepted :: Int,
+    socketsClosed :: Int,
+    socketsActive :: Int,
+    socketsLeaked :: Int
+  }
+
 newSocketState :: IO SocketState
 newSocketState = (,,) <$> newTVarIO 0 <*> newTVarIO 0 <*> newTVarIO mempty
 
+getSocketStats :: SocketState -> IO SocketStats
+getSocketStats (accepted, closed, active) = do
+  socketsAccepted <- readTVarIO accepted
+  socketsClosed <- readTVarIO closed
+  socketsActive <- IM.size <$> readTVarIO active
+  let socketsLeaked = socketsAccepted - socketsClosed - socketsActive
+  pure SocketStats {socketsAccepted, socketsClosed, socketsActive, socketsLeaked}
+
 closeServer :: TMVar Bool -> TVar (IntMap (Weak ThreadId)) -> Socket -> IO ()
 closeServer started clients sock = do
   close sock

src/Simplex/Messaging/Util.hs

@@ -35,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 (<$?>) #-}

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

@@ -3076,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/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

@@ -1,3 +1,4 @@
+{-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleContexts #-}
 {-# LANGUAGE GADTs #-}
@@ -7,6 +8,7 @@
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE StandaloneDeriving #-}
+{-# LANGUAGE TypeApplications #-}
 {-# OPTIONS_GHC -fno-warn-ambiguous-fields #-}
 {-# OPTIONS_GHC -Wno-orphans #-}
 
@@ -17,19 +19,25 @@ 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 (..), noMsgFlags)
+import Simplex.Messaging.Protocol (EntityId (..), Message (..), RecipientId, SParty (..), noMsgFlags)
 import Simplex.Messaging.Server (MessageStats (..), exportMessages, importMessages, printMessageStats)
-import Simplex.Messaging.Server.Env.STM (journalMsgStoreDepth)
+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 SMPClient (testStoreMsgsDir, testStoreMsgsDir2, testStoreMsgsFile, testStoreMsgsFile2)
+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)
@@ -44,13 +52,18 @@ msgStoreTests = do
     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 :: MsgStoreClass s => SpecWith s
+    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 :: MsgStoreClass s => MsgStoreConfig s -> (s -> IO ()) -> IO ()
+withMsgStore :: STMQueueStore s => MsgStoreConfig s -> (s -> IO ()) -> IO ()
 withMsgStore cfg = bracket (newMsgStore cfg) closeMsgStore
 
 testSMTStoreConfig :: STMStoreConfig
@@ -63,7 +76,9 @@ testJournalStoreCfg =
       pathParts = journalMsgStoreDepth,
       quota = 3,
       maxMsgCount = 4,
-      maxStateLines = 2
+      maxStateLines = 2,
+      stateTailSize = 256,
+      idleInterval = 21600
     }
 
 mkMessage :: MonadIO m => ByteString -> m Message
@@ -82,98 +97,121 @@ deriving instance Eq (JournalState t)
 
 deriving instance Eq (SJournalType t)
 
-testGetQueue :: MsgStoreClass s => s -> IO ()
+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 <- EntityId <$> atomically (C.randomBytes 24 g)
+  (rId, qr) <- testNewQueueRec g True
   runRight_ $ do
-    q <- getMsgQueue ms rId
-    let write s = writeMsg ms q True =<< mkMessage s
+    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 q
-    Msg "message 1" <- tryPeekMsg q
-    Nothing <- tryDelMsg q mId2
-    Msg "message 1" <- tryDelMsg q mId1
-    Nothing <- tryDelMsg q mId1
-    Msg "message 2" <- tryPeekMsg q
-    Nothing <- tryDelMsg q mId1
-    (Nothing, Msg "message 2") <- tryDelPeekMsg q mId1
-    (Msg "message 2", Msg "message 3") <- tryDelPeekMsg q mId2
-    (Nothing, Msg "message 3") <- tryDelPeekMsg q mId2
-    Msg "message 3" <- tryPeekMsg q
-    (Msg "message 3", Nothing) <- tryDelPeekMsg q mId3
-    Nothing <- tryDelMsg q mId2
-    Nothing <- tryDelMsg q mId3
-    Nothing <- tryPeekMsg q
+    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 q
+    Msg "message 4" <- tryPeekMsg ms rId q
     Just (Message {msgId = mId5}, False) <- write "message 5"
-    (Nothing, Msg "message 4") <- tryDelPeekMsg q mId3
-    (Msg "message 4", Msg "message 5") <- tryDelPeekMsg q mId4
+    (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 q
-    (Nothing, Msg "message 5") <- tryDelPeekMsg q mId4
-    (Msg "message 5", Msg "message 6") <- tryDelPeekMsg q mId5
-    (Msg "message 6", Msg "message 7") <- tryDelPeekMsg q mId6
-    (Msg "message 7", Just MessageQuota {msgId = mId8}) <- tryDelPeekMsg q mId7
-    (Just MessageQuota {}, Nothing) <- tryDelPeekMsg q mId8
-    (Nothing, Nothing) <- tryDelPeekMsg q mId8
-    pure ()
-  delMsgQueue ms rId
+    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 :: MsgStoreClass s => s -> IO ()
+testChangeReadJournal :: STMQueueStore s => s -> IO ()
 testChangeReadJournal ms = do
   g <- C.newRandom
-  rId <- EntityId <$> atomically (C.randomBytes 24 g)
+  (rId, qr) <- testNewQueueRec g True
   runRight_ $ do
-    q <- getMsgQueue ms rId
-    let write s = writeMsg ms q True =<< mkMessage s
+    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 q mId1
+    (Msg "message 1", Nothing) <- tryDelPeekMsg ms rId q mId1
     Just (Message {msgId = mId2}, True) <- write "message 2"
-    (Msg "message 2", Nothing) <- tryDelPeekMsg q mId2
+    (Msg "message 2", Nothing) <- tryDelPeekMsg ms rId q mId2
     Just (Message {msgId = mId3}, True) <- write "message 3"
-    (Msg "message 3", Nothing) <- tryDelPeekMsg q mId3
+    (Msg "message 3", Nothing) <- tryDelPeekMsg ms rId q mId3
     Just (Message {msgId = mId4}, True) <- write "message 4"
-    (Msg "message 4", Nothing) <- tryDelPeekMsg q mId4
+    (Msg "message 4", Nothing) <- tryDelPeekMsg ms rId q mId4
     Just (Message {msgId = mId5}, True) <- write "message 5"
-    (Msg "message 5", Nothing) <- tryDelPeekMsg q mId5
-    pure ()
-  delMsgQueue ms rId
+    (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 <- EntityId <$> atomically (C.randomBytes 24 g)
-  rId2 <- EntityId <$> atomically (C.randomBytes 24 g)
+  (rId1, qr1) <- testNewQueueRec g True
+  (rId2, qr2) <- testNewQueueRec g True
+  sl <- readWriteQueueStore testStoreLogFile ms
   runRight_ $ do
-    let write q s = writeMsg ms q True =<< mkMessage s
-    q1 <- getMsgQueue ms rId1
-    Just (Message {}, True) <- write q1 "message 1"
-    Just (Message {}, False) <- write q1 "message 2"
-    q2 <- getMsgQueue ms rId2
-    Just (Message {msgId = mId3}, True) <- write q2 "message 3"
-    Just (Message {msgId = mId4}, False) <- write q2 "message 4"
-    (Msg "message 3", Msg "message 4") <- tryDelPeekMsg q2 mId3
-    (Msg "message 4", Nothing) <- tryDelPeekMsg q2 mId4
-    Just (Message {}, True) <- write q2 "message 5"
-    Just (Message {}, False) <- write q2 "message 6"
-    Just (Message {}, False) <- write q2 "message 7"
-    Nothing <- write q2 "message 8"
+    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
@@ -182,6 +220,7 @@ testExportImportStore ms = do
   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
@@ -255,24 +294,121 @@ testQueueState ms = do
 testMessageState :: JournalMsgStore -> IO ()
 testMessageState ms = do
   g <- C.newRandom
-  rId <- EntityId <$> atomically (C.randomBytes 24 g)
+  (rId, qr) <- testNewQueueRec g True
   let dir = msgQueueDirectory ms rId
       statePath = msgQueueStatePath dir $ B.unpack (B64.encode $ unEntityId rId)
-      write q s = writeMsg ms q True =<< mkMessage s
+      write q s = writeMsg ms rId q True =<< mkMessage s
 
   mId1 <- runRight $ do
-    q <- getMsgQueue ms rId
+    q <- ExceptT $ addQueue ms qr
     Just (Message {msgId = mId1}, True) <- write q "message 1"
     Just (Message {}, False) <- write q "message 2"
-    liftIO $ closeMsgQueue ms rId
+    liftIO $ closeMsgQueue q
     pure mId1
 
   ls <- B.lines <$> B.readFile statePath
   B.writeFile statePath $ B.unlines $ take (length ls - 1) ls
 
   runRight_ $ do
-    q <- getMsgQueue ms rId
+    q <- ExceptT $ getQueue ms SRecipient rId
     Just (Message {msgId = mId3}, False) <- write q "message 3"
-    (Msg "message 1", Msg "message 3") <- tryDelPeekMsg q mId1
-    (Msg "message 3", Nothing) <- tryDelPeekMsg q mId3
-    liftIO $ closeMsgQueueHandles q
+    (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/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

@@ -43,6 +43,9 @@ import Util
 testHost :: NonEmpty TransportHost
 testHost = "localhost"
 
+testHost2 :: NonEmpty TransportHost
+testHost2 = "127.0.0.1"
+
 testPort :: ServiceName
 testPort = "5001"
 
@@ -76,6 +79,9 @@ testStoreNtfsFile = "tests/tmp/smp-server-ntfs.log"
 testStoreNtfsFile2 :: FilePath
 testStoreNtfsFile2 = "tests/tmp/smp-server-ntfs.log.2"
 
+testPrometheusMetricsFile :: FilePath
+testPrometheusMetricsFile = "tests/tmp/smp-server-metrics.txt"
+
 testServerStatsBackupFile :: FilePath
 testServerStatsBackupFile = "tests/tmp/smp-server-stats.log"
 
@@ -132,12 +138,16 @@ cfgMS msType =
       controlPortUserAuth = Nothing,
       controlPortAdminAuth = Nothing,
       messageExpiration = Just defaultMessageExpiration,
+      expireMessagesOnStart = True,
+      idleQueueInterval = defaultIdleQueueInterval,
       notificationExpiration = defaultNtfExpiration,
       inactiveClientExpiration = Just defaultInactiveClientExpiration,
       logStatsInterval = Nothing,
       logStatsStartTime = 0,
-      serverStatsLogFile = "tests/smp-server-stats.daily.log",
+      serverStatsLogFile = "tests/tmp/smp-server-stats.daily.log",
       serverStatsBackupFile = Nothing,
+      prometheusInterval = Nothing,
+      prometheusMetricsFile = testPrometheusMetricsFile,
       pendingENDInterval = 500000,
       ntfDeliveryInterval = 200000,
       smpCredentials =

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
@@ -316,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

tests/ServerTests.hs

@@ -37,15 +37,16 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server (exportMessages)
-import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
+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 (doesDirectoryExist, removeDirectoryRecursive, removeFile)
+import System.Directory (doesDirectoryExist, doesFileExist, removeDirectoryRecursive, removeFile)
 import System.TimeIt (timeItT)
 import System.Timeout
 import Test.HUnit
@@ -70,6 +71,7 @@ serverTests  = do
   describe "Store log" testWithStoreLog
   describe "Restore messages" testRestoreMessages
   describe "Restore messages (old / v2)" testRestoreExpireMessages
+  describe "Save prometheus metrics" testPrometheusMetrics
   describe "Timing of AUTH error" testTiming
   describe "Message notifications" testMessageNotifications
   describe "Message expiration" $ do
@@ -812,6 +814,7 @@ testRestoreExpireMessages =
     exportStoreMessages = \case
       AMSType SMSJournal -> do
         ms <- newMsgStore testJournalStoreCfg {quota = 4}
+        readWriteQueueStore testStoreLogFile ms >>= closeStoreLog
         removeFileIfExists testStoreMsgsFile
         exportMessages False ms testStoreMsgsFile False
       AMSType SMSMemory -> pure ()
@@ -823,6 +826,13 @@ testRestoreExpireMessages =
     runClient :: Transport c => TProxy c -> (THandleSMP c 'TClient -> IO ()) -> Expectation
     runClient _ test' = testSMPClient test' `shouldReturn` ()
 
+testPrometheusMetrics :: SpecWith (ATransport, AMSType)
+testPrometheusMetrics =
+  it "should save Prometheus metrics" $ \(at, msType) -> do
+    let cfg' = (cfgMS msType) {prometheusInterval = Just 1}
+    withSmpServerConfigOn at cfg' testPort $ \_ -> threadDelay 1000000
+    doesFileExist testPrometheusMetricsFile `shouldReturn` True
+
 createAndSecureQueue :: Transport c => THandleSMP c 'TClient -> SndPublicAuthKey -> IO (SenderId, RecipientId, RcvPrivateAuthKey, RcvDhSecret)
 createAndSecureQueue h sPub = do
   g <- C.newRandom
@@ -970,7 +980,7 @@ testMsgExpireOnInterval =
   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' = (cfgMS msType) {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

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