Merge branch 'master' into f/files-tests

* server: control port

* do not remove messages when saving via control port

* remove unused record fields

* fix tests

CHANGELOG.md

@@ -1,3 +1,47 @@
+# 5.2.0 (NTF server 1.5.0)
+
+Agent:
+- treat agent INACTIVE error as temporary - fixes failed message delivery in some race conditions.
+- restore connection confirmations after client restart - fixes failed connections.
+- ratchet resynchronization protocol and API.
+- increase connection version to mutually supported by both peers on each received message.
+
+Client:
+- make timeout for batched functions dependent on the number of batches - fixes expiry on large batches.
+
+Servers:
+- add timeout in case of sending TCP traffic and in case of partial delivery of requested blocks to avoid resource leaks.
+
+# 5.1.2, 5.1.3 (NTF server 1.4.1, 1.4.2)
+
+Agent:
+- ACK message on decryption error (fixes stuck message delivery bug)
+- more robust connection switching logic, API to abort switching the address
+
+Notification server:
+- batch subscriptions to SMP servers
+
+# 5.1.1 (NTF server 1.4.0)
+
+Agent:
+- store and check hashes of previous encrypted messages to differentiate between duplicates and decryption errors
+
+Server:
+- larger processing queues
+- expire messages when restoring them
+
+# 5.1.0
+
+XFTP client:
+- check encrypted file exists when uploading
+- remove user ID from deletion API
+
+Agent:
+- vacuum database on migrations
+
+SMP server:
+- configure message expiration time in INI file
+
 # 5.0.0
 
 SimpleX File Transfer Protocol (XFTP):

README.md

@@ -150,7 +150,7 @@ You can install and setup servers automatically using our script:
 
 ```sh
 curl --proto '=https' --tlsv1.2 -sSf https://raw.githubusercontent.com/simplex-chat/simplexmq/stable/install.sh -o simplex-server-install.sh \
-&& if echo '1268d605e90bca1a8c7ef476038a8bd9aa9b1a28f79ea0a1485669ccf8fc23cd simplex-server-install.sh' | sha256sum -c; then chmod +x ./simplex-server-install.sh && ./simplex-server-install.sh; rm /simplex-server-install.sh; else echo "SHA-256 checksum is incorrect!" && rm ./simplex-server-install.sh; fi
+&& if echo 'b8cf2be103f21f9461d9a500bcd3db06ab7d01d68871b07f4bd245195cbead1d simplex-server-install.sh' | sha256sum -c; then chmod +x ./simplex-server-install.sh && ./simplex-server-install.sh; rm ./simplex-server-install.sh; else echo "SHA-256 checksum is incorrect!" && rm ./simplex-server-install.sh; fi
 ```
 
 ### Build from source

cabal.project

@@ -3,14 +3,6 @@ packages: .
 -- packages: . ../hs-socks
 -- packages: . ../http2
 
-package *
-  coverage: True
-  library-coverage: True
-
-package attoparsec
-  coverage: False
-  library-coverage: False
-
 source-repository-package
     type: git
     location: https://github.com/simplex-chat/aeson.git

install.sh

@@ -11,6 +11,7 @@ scripts_systemd_smp="$scripts/smp-server.service"
 scripts_systemd_xftp="$scripts/xftp-server.service"
 scripts_update="$scripts/simplex-servers-update"
 scripts_uninstall="$scripts/simplex-servers-uninstall"
+scripts_stopscript="$scripts/simplex-servers-stopscript"
 
 # Default installation paths
 path_bin="/usr/local/bin"
@@ -18,6 +19,7 @@ path_bin_smp="$path_bin/smp-server"
 path_bin_xftp="$path_bin/xftp-server"
 path_bin_update="$path_bin/simplex-servers-update"
 path_bin_uninstall="$path_bin/simplex-servers-uninstall"
+path_bin_stopscript="$path_bin/simplex-servers-stopscript"
 
 path_conf_etc="/etc/opt"
 path_conf_var="/var/opt"
@@ -60,8 +62,8 @@ ${GRN}3.${NC} Setup user for each server:
 ${GRN}4.${NC} Create systemd services:
     - smp: ${YLW}${path_systemd_smp}${NC}
     - xftp: ${YLW}${path_systemd_xftp}${NC}
-${GRN}5.${NC} Install update and uninstallation script:
-    - all: ${YLW}${path_bin_update}${NC}, ${YLW}${path_bin_uninstall}${NC}
+${GRN}5.${NC} Install stopscript (systemd), update and uninstallation script:
+    - all: ${YLW}${path_bin_update}${NC}, ${YLW}${path_bin_uninstall}${NC}, ${YLW}${path_bin_stopscript}${NC}
 
 Press ${GRN}ENTER${NC} to continue or ${RED}Ctrl+C${NC} to cancel installation"
 
@@ -87,9 +89,9 @@ setup_users() {
 setup_dirs() {
  # Unquoted varibles, so field splitting can occur
  mkdir -p $path_conf_smp
- chown "$user_smp":"$user_smp" "$path_conf_smp"
+ chown "$user_smp":"$user_smp" $path_conf_smp
  mkdir -p $path_conf_xftp
- chown "$user_xftp":"$user_xftp" "$path_conf_xftp"
+ chown "$user_xftp":"$user_xftp" $path_conf_xftp
 }
 
 setup_systemd() {
@@ -100,6 +102,7 @@ setup_systemd() {
 setup_scripts() {
  curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_update" -o "$path_bin_update" && chmod +x "$path_bin_update"
  curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_uninstall" -o "$path_bin_uninstall" && chmod +x "$path_bin_uninstall"
+ curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_stopscript" -o "$path_bin_stopscript" && chmod +x "$path_bin_stopscript"
  }
 
 checks() {
@@ -131,7 +134,7 @@ main() {
  setup_systemd
  printf "${GRN} Done!${NC}\n"
 
- printf "Installing update and uninstallation script..."
+ printf "Installing stopscript, update and uninstallation script..."
  setup_scripts
  printf "${GRN} Done!${NC}\n"
 

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 5.1.1
+version: 5.3.0.0
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,
@@ -144,8 +144,6 @@ tests:
       - silently == 1.2.*
       - main-tester == 0.2.*
       - timeit == 2.0.*
-    ghc-options:
-      - -fhpc
 
 ghc-options:
   # - -haddock

rfcs/2023-05-03-delivery-receipts.md

@@ -0,0 +1,170 @@
+# Delivery receipts
+
+## Problems
+
+User experience - users need to know that the messages are delivered to the recipient, as this confirms that the system is functioning.
+
+The downside of communicating message delivery as it confirms that the recipient was online, and, unless there is a delay in confirming, can be used to track the location via the variation in network latency. So delivery receipts should be delayed with a randomized interval and should be opt in or opt out.
+
+Another problem of message receipts is that they increase network traffic and server load. This could be avoided if delivery receipts are communicated as part of normal message delivery flow.
+
+Some other existing and planned features implicitely confirm message delivery and, possibly, should depend on message delivery being enabled:
+- agent message to resume delivery when quota was exceeded (implemented, [rfc](./2022-12-27-queue-quota.md))
+- agent message to re-deliver skipped messages or to re-negotiate double ratchet.
+
+## Solution
+
+There are three layers where delivery receipts can be implemented:
+- chat protocol. Pro: logic of when to deliver it is decoupled from the message flow, Con: extra traffic, can only work in duplex connections.
+- agent client protocol. Pro: can be automated and combined with the protocol to re-deliver skipped messages. Con: extra traffic.
+- SMP protocol. Pro: minimal extra traffic, Con: complicates server design as it would require pushing receipts when there is no next message.
+
+The last approach seems the most promising for avoiding additional traffic:
+- modify client ACK command to include whether delivery receipt should be provided to sender, and, possibly, any e2e encrypted data that should be included in the receipt (e.g., that the receiving client already saw this message in case we use "feedback" variant of roumor-mongering protocol for groups).
+- server would manage delaying of the receipts, by randomizing the time after which the receipt will be available to the sender, and by combining the receipts when possible.
+- modify response to SEND command to include any available delivery receipts.
+- add a separate delivery receipt that will be pushed to the sender in the connection where the message was received by the server.
+
+## SMP protocol changes
+
+```haskell
+data Command (p :: Party) where
+  -- ...
+  ACK :: MsgId -> Maybe ByteString -> Command Recipient
+  -- the presense of ByteString in ACK indicates that the delivery needs to be confirmed.
+  -- the protocol does not define the format of this confirmation, it is application specific, and can be -- an empty string.
+  -- And open question is how to e2e encrypt information in this string - this probably can be handled on Agent client protocol level, and could be the same ratchet key that was used to encrypt and decrypt the message. The downside of this approach is that this key currently is not stored, and storing it requires additional logic to clear these keys if unused after some time.
+  -- TODO consider what could be a better approach.
+  SENT :: MsgId -> [(MsgId, UTCTime, ByteString)] -> Command Sender
+  -- or
+  -- SENT :: MsgId -> Command Sender
+  -- in case we just batch
+  -- this response will be sent to SEND command and will include a sender's message ID generated by the server (currently it does not exist), and posibly an empty list of delivery receipts with the same message IDs as in responses to SEND, timestamps when these receipts became available, and e2e encrypted ByteString passed in ACK command.
+  -- The ID in this response should be different from the ID used in MSG, to keep the promise of not having shared identifiers in sent/received traffic even inside TLS tunnel.
+  -- Keeping the quality of shared ciphertext also requires adding additional encryption layer between the server and the sender, this can be achieved in one of two ways:
+  -- 1) passing a separate DH key in each SEND command, and server including additional DH key in each SENT response, with computed DH secret per message later used to encrypt and decrypt the delivery receipt payloads. This is probably a bad idea as it would increase a cryptographic load on both the server and the client.
+  -- 2) agree a key per queue, in the same way it is done for the recipient. Possibly, it requires additional DH key in confirmation message that the recipient then uses to secure the queue, and passing this key in KEY (secure queue) command. The response to this secure command would the include server's DH key returned to the recipient that would be passed to the sender in HELLO message. Even though recipient could observe both public DH keys, they won't know the computed shared secret. Recipient that controls the server could perform MITM attack on this key exchange, but it doesn't give any benefit over what recipient can do when they have access to the server - the threat model remains the same. The downside of this approach is that it also requires additional changes in client protocol level (confirmation message format and HELLO message).
+  -- 3) also agree on a key per queue, but via separate commands between the sender and the server, once the sender was notified that the queue is secured. This approach is probably better, and the server would simply delay the delivery of delivery receipts until the shared secret is agreed.
+  SKEY :: C.PublicKeyX25519 -> Command Sender
+  SBKEY :: C.PublicKeyX25519 -> BrokerMsg
+  -- these are the command and response to agree secret to encrypt delivery receipt payloads for option 3
+  SSUB :: Command Sender
+  -- subscribe to receive delivery receipts for a given queue - will be sent when the conversation is opened (unless there is an active subscription already), not all queues at once, and won't be re-subscribed on losing the server connection (TBC).
+  RCVD :: MsgId -> UTCTime -> ByteString -> Command Recipient
+  -- delivery receipt. UTCTime is the time when it became available, not the time when ACK was sent by the recipient, to avoid leaking location via network latency.
+```
+
+Possibly, there is no need to include delivery receipts into SENT response and instead just use batching of responses that is already supported. As server responses are not signed, there is no per-response overhead that is substantial, and a lot of receipts that are available can be packed into one block (depending on the size of payload that has to be fixed not to leak metadata).
+
+This all seems rather complicated for SMP protocol, and the approach of doing it on a higher level seems more attractive than initially. Possibly we should reconsider, and reduce traffic by reducing block sizes... Reducing block sizes unfortunately requires supporting variable block sizes, and would leak some metadata during the transition period.
+
+## Another approach
+
+Above represents substantial complexity, and at least doubles server code complexity for the feature that is definitely not doubling the value of server software. Moving to variable block size is simpler, but also has a lot of complexity, reduces metadata privacy (at least for the duration of migration period), reduces image preview quality, and requires postponing this feature for multiple releases, until all clients migrate.
+
+Given that the main traffic is generated by the groups, and direct messages do not create a lot of traffic, a much simpler and better solution is to simply send delivery receipts as the message, in direct conversations only, either as chat protocol message or as agent client protocol message (either on the message or on the envelope layer).
+
+### Comparison of these two approaches:
+
+**Chat protocol message**
+
+Pros:
+- simpler, more contained change - SMP layer is not aware of this feature
+- easier to extend protocol with additional application specific payload, e.g. references to group DAG
+Cons:
+- ?
+
+```json
+  // ...
+  "x.msg.delivered": {
+    "properties": {
+      "msgId": {"ref": "base64url"},
+      "params": {
+        "properties": {
+          "msgId": {"ref": "base64url"},
+        },
+        "optionalProperties": {
+          "data": {} // possibly the initial protocol does not need it, with JSON can be added later
+        }
+      }
+    }
+  },
+  // ...
+```
+
+**Agent client protocol envelope**
+
+Pros:
+- possibility of using it in a wider range of the applications
+- possibility to include received message hash to increase communication integrity - the sending client would be then notified, and it can be exposed in the UI, that the received message is not the same as sent.
+
+Cons:
+- additional implementation complexity - requires additional events to communicate between chat and agent.
+
+```haskell
+data AMessage =
+  -- ...
+  | A_RCVD AgentMsgId MsgHash ByteString -- references to the received message
+  -- ...
+```
+
+The weirdness of the above design is that it refers to the data present in the header of another message, the alternative would be to have a separate envelope for delivery receipt:
+
+```haskell
+data AgentMessage =
+  -- ...
+  AgentMessageRcvd APrivHeader AgentMsgId MsgHash ByteString -- references to the received message
+  -- ...
+```
+
+But probably the first one is a bit better, TBC.
+
+In any case there should be an additional event to notify chat client:
+
+```haskell
+data ACommand (p :: AParty) (e :: AEntity) where
+  -- ...
+  RCVD AgentMsgId MsgMeta ByteString -> ACommand Agent AEConn
+  -- ...
+```
+
+On the balance of things, implementing on the level of Agent Client protocol seems better, as the additional complexity is marginal, but it allows for wider range of applications, and also allows for additional delivery integrity validation. The format for payload still requires chat protocol message encoding once we want to add it, but initially it could be just an empty string.
+
+## Implementation plan
+
+Currently, we delete sent messages once delivered to the server. It would be helpful if we could keep the records in snd_messages table, and then use them to process delivery receipts, although it may be insufficient (we could add fields). Probably it is not possible to keep them as there is a foreign key constraint with `on delete cascade`.
+
+The new table will be used to track sent message hashes to correlate their IDs with delivery receipts (that will also be stored in messages/rcv_messages). Receipts need to be processed in chat in the same way as normal messages, so they would be sent to chat with MsgMeta, and the chat will need to ack them once processed.
+
+Agent ackMessage function will be also used to automatically schedule sending delivery receipts if they are enabled for connection, only for normal messages - no sending receipts to receipts.
+
+There can be receipt re-delivery to the chat in the same cases as when normal message can be re-delivered (in case of AGENT A_DUPLICATE when it was not ack'd by the user).
+
+## Other considerations
+
+### How clients decide whether to send delivery receipts
+
+Two options are possible - local settings per conversation or chat preferences framework, that allows to have mutual on/off. The latter seems preferable as without knowing whether the other party is sending receipts, it is not possible to uderstand what the absense of the receipt means - network malfunction or receipts disabled.
+
+Groups are a special case, as while some groups may enable sending the delivery receipts, the group members should be able to disable it locally. This should probably be done via a separate conversation setting in the same way as enabling notifications or favourites. In this case the receipt would only be sent to the group if it is enabled in the group and not disabled by the member. The toggle may be located in the same page as chat preferences, but it should be a separate setting. We might want though to communicate somehow whether a given member sends delivery receipts so that other members know whether to expect them or not.
+
+### How this functionality is released
+
+5.2:
+- support for sending and receiving delivery receipts preference, both in direct messages and in groups (for forward compatibility).
+- support for sending and receiving delivery receipts in direct chats only, but disable sending them
+
+5.3:
+- show receipt preferences in the UI
+- enable sending receipts in direct chats where they are enabled
+
+A separate question is how to enable this functionality for the existing contacts. Possible options are:
+
+1. Enable (as per default) for all contacts, show notification to the user when they open the app for update that delivery notifications are now sent by default to all contacts. Pro: no extra logic to implement. Con: may be perceived as a privacy violation, as to some contacts the delivery receipts will be sent before the user had chance to disable them.
+2. Ask the user when the new version first runs whether they want to use delivery receipts and offer these options:
+  1) keep enabled for all profiles (and for all contacts)
+  2) enable for all profiles in ~12 or in ~24 hours giving users the chance to review all contacts / profiles and disable some of them. The problem here is also in possibility of the correlation in case they all start sending receipts at the same time. Possibly the option could be to set a random time in 12-15 or 24-30 hours range to avoid the possibility of such correlation.
+  3) disable for all profiles – it will require sending profile updates to all contacts, so the delivery receipts should be kept disabled and profile updates should be sent after random intervals, one by one (not scheduled all at once, as the time may pass while the app is off).
+3. Offer an option to enable globally later - we could keep it as a one-off option, available only to existing users, and visible on the top level of the Settings - once enabled, the option will disappear and it won't be possible to disable again. The downside here is that the new contacts would be receiving the profile with enabled notifications but they still won't be delivered...
+4. Another option is to have all new contacts decided based on a global user default (that can be set in the settings and in the dialog on first start), but for the existing contacts keep in unset state that is not interpreted as either on or off, but interpreted as unknown until the user makes a choice... That might be an optimal solution for the users but it would probably require changing the preferences framework or some ad-hoc hacks. That still keeps the question open how to avoid correlation between profiles.
+5. That might be the case for version agreement too - the availability of the option per contact will depend on the version. It doesn't answer the question what to do with global defaults though...

rfcs/2023-06-08-resync-ratchets.md

@@ -0,0 +1,366 @@
+# Re-sync encryption ratchets
+
+## Problem
+
+See https://github.com/simplex-chat/simplexmq/pull/743/files for problem and high-level solution.
+
+## Implementation
+
+### Diagnosing ratchet de-synchronization
+
+Message decryption happens in `agentClientMsg`, in `agentRatchetDecrypt`, which can return decryption result or error. Decryption error can be differentiated in `agentClientMsg` result pattern match, in Left cases, where we already differentiate duplicate error (`AGENT A_DUPLICATE`).
+
+Question: Which decryption errors can be diagnosed as ratchet de-synchronization?
+
+Possibly any `AGENT A_CRYPTO` error. Definitely on `RATCHET_HEADER`, TBC other. See `cryptoError :: C.CryptoError -> AgentErrorType` for conversion from decryption errors to `A_CRYPTO` or other agent errors. We're only interested in crypto errors, as other are either other client implementation errors, internal errors, or already processed duplicate error.
+
+Proposed classification of crypto errors, based on `AgentCryptoError`:
+
+`DECRYPT_AES` -> re-sync allowed (recommended/required?)
+
+`DECRYPT_CB` -> re-sync allowed (recommended/required?)
+
+`RATCHET_HEADER` -> **re-sync required**
+
+`RATCHET_EARLIER` -> re-sync allowed
+
+`RATCHET_SKIPPED` -> **re-sync required**
+
+Ratchet re-synchronization could be started automatically on diagnosing de-synchronization, based on these errors. As a potentially dangerous feature (e.g., implementation error could lead to infinite re-sync loop causing large traffic consumption), initially it will be available via agent functional api for client to call. Ratchet de-synchronization will instead produce an event prompting client to re-synchronize.
+
+Diagnosing possible ratchet de-synchronization also will be recorded as connection state - `ratchet_desync_state` field in `connections` table. Client should be prohibited to start ratchet re-synchronization unless `ratchet_desync_state` is set.
+
+Event should not be repeated for following received messages that can't be decrypted - based on `ratchet_desync_state`. If a received message can be decrypted, `ratchet_desync_state` should be set to NULL and a new event sent, indicating ratchet has healed.
+
+New event - `RDESYNC :: RatchetDesyncState -> ConnectionStats -> ACommand Agent AEConn`
+
+```haskell
+data RatchetDesyncState
+  = RDResyncAllowed
+  | RDResyncRequired
+  | RDHealed
+```
+
+New field should be added to `ConnectionStats` - `ratchetDesyncState :: Maybe RatchetDesyncState`, based on `ratchet_desync_state`.
+
+> On `RDESYNC` events chat should create chat item, prompting ratchet re-synchronization or notifying it has healed.
+> If connection has diagnosed ratchet de-sync, chat item should have a button to start ratchet re-sync.
+> We'd have to get `ConnectionStats` on chat level for this instead of chat info.
+> This wouldn't work for groups. One option is to add `ConnectionStats` to `GroupMember` type and update on events.
+> Same could be done for `Contact` then.
+
+To consider - allow to start ratchet re-synchronization at any time regardless of this field as an experimental feature. In chat it could be behind "Developer tools" + additional "Experimental" toggle. Agent api would have `force :: Bool` as parameter, allowing to bypass `ratchet_desync_state`. Should `ratchet_resync_state` (see below) still be honored in this case?
+
+### Re-synchronization process
+
+\*\*\*\*\*
+
+Basic idea is the following:
+
+Both agents send new ratchet keys and compute a new shared secret. Agent that starts re-synchronization should record this fact in the connection state. Agent that receives a new key should respond with a key of its own, unless it has recorded that it itself started re-synchronization in the connection state.
+
+It can happen that both agents start re-synchronizing simultaneously. In this case they both would record it in the connection state and would not respond with a new message - instead they would use each other's already sent keys.
+
+Agent has both keys if:
+
+- It initiates with the first key, and then receives the second key;
+- It receives the first key and then generates its own in response.
+
+After agent has both keys, it initiates new ratchet depending on keys ordering. The agent that sent the lower key should use `initRcvRatchet` function, the agent that sent the greater key should use `initSndRatchet` (or vice versa - but they should deterministically choose different sides).
+
+\*\*\*\*\*
+
+State whether the ratchet re-synchronization is in progress should be tracked in database via `connections` table new `ratchet_resync_state` field.
+
+New functional api:
+
+```haskell
+resyncConnectionRatchet :: AgentErrorMonad m => AgentClient -> ConnId -> m ConnectionStats
+```
+
+or if we want to allow re-synchronizing ratchet at any time even if de-synchronization wasn't diagnosed:
+
+```haskell
+resyncConnectionRatchet :: AgentErrorMonad m => AgentClient -> ConnId -> Bool -> m ConnectionStats
+resyncConnectionRatchet c connId force = ...
+```
+
+Possibly client command?
+
+``` haskell
+data ACommand (p :: AParty) (e :: AEntity) where
+  ...
+  RESYNC_RATCHET :: Bool -> ACommand Client AEConn
+```
+
+New event - `RRESYNC :: RatchetResyncState -> ConnectionStats -> ACommand Agent AEConn`
+
+```haskell
+data RatchetResyncState
+  = RRStarted
+  | RRAgreedSnd
+  | RRAgreedRcv
+  | RRComplete
+```
+
+New `ConnectionStats` field - `ratchetResyncState :: Maybe RatchetResyncState`.
+
+When called, it should:
+
+- Generate new keys.
+- Update database connection state.
+  - Set `ratchet_desync_state` to NULL.
+  - Set `ratchet_resync_state` to `RRStarted`.
+  - Delete old ratchet from `ratchets` (is it safe?), create new ratchet.
+- Send `AgentRatchetKey` message.
+- Return updated `ConnectionStats` to client.
+
+> On `RRESYNC` events chat should create chat item, and reset connection verification.
+> Parameterized `RRESYNC` allows to distinguish: start and end of re-synchronization for initiating party; chat item direction - `RRESYNC RRStarted` is snd, `RRESYNC RRAgreedSnd/Rcv` and `RRESYNC RRComplete` are rcv (`RRESYNC RRAgreedSnd/Rcv` chat item could be omitted).
+
+AgentRatchetKey is a new message on the level of AgentMsgEnvelope - encrypted with queue level e2e encryption, but not with connection level e2e encryption (since ratchet de-synchronized).
+
+```haskell
+data AgentMsgEnvelope
+  = ...
+  | AgentRatchetKey
+      { agentVersion :: Version,
+        e2eEncryption :: E2ERatchetParams 'C.X448,
+        info :: ByteString -- for extension
+      }
+```
+
+On receiving `AgentRatchetKey`, if the receiving client hasn't started the ratchet re-synchronization itself (check `ratchet_resync_state`), it should:
+
+- Generate new keys and compute new shared secret, initializing ratchet based on keys comparison.
+- Update database connection state.
+  - Set `ratchet_resync_state` to `RRAgreedSnd/Rcv` (depending on whether ratchet was initialized as sending or receiving).
+  - Delete old ratchet from `ratchets`, create new ratchet.
+- Reply with its own `AgentRatchetKey`.
+- Notify client with `RRESYNC RRAgreedSnd/Rcv`.
+- If ratchet was initialized as sending, send `EREADY` message, notifying other agent ratchet is re-synced.
+
+New agent message:
+
+```haskell
+data AMessage
+  = ...
+  | -- ratchet re-synchronization is complete, with last decrypted sender message id
+    EREADY PrevExternalSndId
+```
+
+On receiving `AgentRatchetKey`, if the receiving client started re-sync:
+
+- Compute new shared secret, initializing ratchet based on keys comparison.
+- Update database connection state.
+  - Set `ratchet_resync_state` to `RRAgreedSnd/Rcv` (depending on whether ratchet was initialized as sending or receiving).
+  - Update ratchet.
+- Notify client with `RRESYNC RRAgreedSnd/Rcv`.
+- If ratchet was initialized as sending, send `EREADY` message.
+
+After agent receives `EREADY` (or any other message that successfully decrypts):
+
+- Reset `ratchet_resync_state` to NULL.
+- Notify client with `RRESYNC RRComplete`.
+- If ratchet was initialized as receiving, send reply `EREADY` message.
+
+### State transitions
+
+For initiating party:
+
+```
+    +------------+
+    | Ratchet ok |
+    +------------+
+          |
+          | message received, decryption error
+   * ---->|-----------------------------------+
+          |                                   |
+          V          new (clarifying)         V
+  +-----------------+     error      +------------------+
+  | Re-sync allowed |--------------->| Re-sync required |
+  +-----------------+                +------------------+
+          |                                   |
+          |-----------------------------------|
+          |                                   | alternative - message received,
+          | re-sync started by client         | successfully decrypted
+          V                                   V
+  +-----------------+                   +------------+
+  | Re-sync started |                   | Ratchet ok |
+  +-----------------+                   +------------+
+          |
+          | other party replied with new ratchet key
+          V
+  +----------------+
+  | Re-sync agreed |----> * message received, decryption error
+  |   snd / rcv    |        (should remember agreed state for reply EREADY?)
+  +----------------+
+          |
+          | message received, successfully decrypted
+          | (can be, but not necessarily, EREADY)
+          V
+    +------------+
+    | Ratchet ok |
+    +------------+
+```
+
+For replying party:
+
+```
+    +------------+
+    | Ratchet ok |
+    +------------+
+          |
+          | other party sent new ratchet key
+          V
+  +----------------+
+  | Re-sync agreed |
+  |   snd / rcv    |
+  +----------------+
+          |
+          | message received, successfully decrypted
+          | (can be, but not necessarily, EREADY)
+          V
+    +------------+
+    | Ratchet ok |
+    +------------+
+```
+
+### Ratchet state model
+
+#### 2 state variables
+
+Above we considered model with separate de-sync and re-sync state.
+
+| Desync \ Resync      | Nothing | RRStarted | RRAgreedSnd | RRAgreedRcv |
+| ---                  |  :---:  |   :---:   |    :---:    |    :---:    |
+| **Nothing**          | 1       | 3         | 4           | 4           |
+| **RDResyncAllowed**  | 2       |           | 5           | 5           |
+| **RDResyncRequired** | 2       |           | 5           | 5           |
+
+1: Ratchet is ok.
+
+2: Re-sync diagnosed, not in progress.
+
+3: Rs-sync started, diagnosing de-sync is prohibited.
+
+4: Re-sync agreed.
+
+5: Re-sync agreed, new de-sync is diagnosed.
+
+Combination 5 is possible in case de-sync was diagnosed before message that could be decrypted is received, for example if `EREADY` failed to deliver and no other decryptable message followed. We shouldn't prohibit diagnosing de-sync in this case, because agent may never exit "Agreed" state (if new decryptable message is never received). We also shouldn't overwrite/forget state of re-sync, even if we diagnose new possible de-sync, because if the decryptable `EREADY` is received and ratchet is in `RRAgreedRcv` state, it should respond with reply `EREADY`.
+
+Some combinations should be impossible:
+
+  - `RDResyncAllowed` with `RRStarted`.
+
+  - `RDResyncRequired` with `RRStarted`.
+
+`RDHealed` is equivalent to `Nothing` and only used for `RDESYNC` event, `Maybe RatchetDesyncState` can be replaced with `RatchetDesyncState`, with single new constructor `RDNoDesync` replacing `Nothing` and `RDHealed`.
+
+`RRComplete` is equivalent to `Nothing` and only used for `RRESYNC` event, `Maybe RatchetResyncState` can be replaced with `RatchetResyncState`, with single new constructor `RDNoResync` replacing `Nothing` and `RRComplete`.
+
+#### Single state variable
+
+Another option is two have a single state variable describing ratchet.
+
+```haskell
+data RatchetState
+  = RSOk
+  | RSResyncAllowed
+  | RSResyncRequired
+  | RSResyncStarted
+  | RRResyncAgreedSnd
+  | RRResyncAgreedRcv
+
+-- When `resyncConnectionRatchet` is not prohibited. Can override with `force`.
+-- Currently we check:`(isJust ratchetDesyncState || force) && ratchetResyncState /= Just RRStarted`.
+resyncConnectionRatchetAllowed :: RatchetState -> Bool
+resyncConnectionRatchetAllowed = \case
+  RSOk -> False
+  RSResyncAllowed -> True
+  RSResyncRequired -> True
+  RSResyncStarted -> False -- `force` shouldn't override
+  RRResyncAgreedSnd -> False
+  RRResyncAgreedRcv -> False
+
+-- When we register and notify about ratchet de-synchronization.
+-- Currently we check: `(isNothing ratchetDesyncState && ratchetResyncState /= Just RRStarted)`.
+-- We should also allow to update from Allowed to Required.
+shouldNotifyRDESYNC :: RatchetState -> Bool
+shouldNotifyRDESYNC = \case
+  RSOk -> True
+  RSResyncAllowed -> False -- only if new error implies Required
+  RSResyncRequired -> False
+  RSResyncStarted -> False
+  RRResyncAgreedSnd -> True
+  RRResyncAgreedRcv -> True
+
+-- When we prohibit connection switch, for `checkRatchetDesync`.
+-- Currently we check: `(ratchetDesyncState == Just RDResyncRequired || ratchetResyncState == Just RRStarted)`
+-- Also use in `runSmpQueueMsgDelivery` to pause delivery?
+ratchetDesynced :: RatchetState -> Bool
+ratchetDesynced = \case
+  RSOk -> False
+  RSResyncAllowed -> False
+  RSResyncRequired -> True
+  RSResyncStarted -> True
+  RRResyncAgreedSnd -> False
+  RRResyncAgreedRcv -> False
+```
+
+Having a single state variable limits differentiation described for combination 5 in matrix. It also limits possible differentiations in client between events when ratchet is healed on its own, and when ratchet re-sync is completed after agents negotiation. Overall, since matrix is not very sparse and allows for more fine-grained decision-making, having separate state variables for de-sync and re-sync seems preferred.
+
+#### Single state variable simplified (final version)
+
+```haskell
+data RatchetSyncState
+  = RSOk
+  | RSAllowed
+  | RSRequired
+  | RSStarted
+  | RSAgreed
+
+-- event
+RSYNC :: RatchetSyncState -> ConnectionStats -> ACommand Agent AEConn`
+
+-- ConnectionStats field
+ratchetSyncState :: RatchetSyncState
+```
+
+Updated design decisions:
+
+1. Single constructor for "Agreed" state. Differentiating `RRResyncAgreedSnd` and `RRResyncAgreedRcv` allowed for easier processing of `EREADY` by helping to determine whether reply `EREADY` has to be sent. However, it duplicated information already present in ratchet's state, and can be instead worked around by remembering and analyzing ratchet state pre decryption.
+
+2. Prohibit transition from "Agreed" state to "Desync" states. This would make possible edge-cases that leave ratchet in de-synchronized state without ability to progress (e.g. failed delivery of `AgentRatchetKey`), but would simplify state machine by removing dedicated "Desync" variable. Besides, there's still a recovery way with a `force` option.
+
+3. Treat "Agreed" as unfinished state - prohibit new messages to be enqueued, etc. Reception of any decryptable message transitions ratchet to "Ok" state.
+
+Possible improvements:
+
+- Repeatedly triggering re-synchronization while in "Started"/"Agreed" state re-sends same keys and EREADY.
+- Cooldown period, during which repeat re-synchronization is prohibited.
+
+### Skipped messages
+
+Options:
+
+1. Ignore skipped messages.
+2. Stop sending new messages while connection re-synchronizes (can use `ratchet_resync_state`).
+
+- Initiator shouldn't send new messages until receives `AgentRatchetKey` from second party.
+- Second party knows new shared secret immediately after processing first `AgentRatchetKey`, so it's not necessary to limit?
+
+3. 2 + Re-send skipped messages first.
+
+- Add `last_external_snd_msg_id` to `AgentRatchetKey`? + see link above
+
+4. Re-send only messages after the latest ratchet step. \*
+
+It may be okay to ignore skipped messages, or at most implement option 2, as ratchet de-synchronization is usually caused by misuse (human error) - the most common cause of ratchet de-sync seems to be sending and receiving messages after running agent with old database backup. In this case user has already seen most skipped messages, and it can be expected to not have them after switching to an old backup. So in this case the only "really skipped" messages are those that were sent during the latest ratchet step and failed to decrypt, triggering ratchet re-sync (\* another option is to only re-send those).
+
+Besides, depending on time of backup there may be an arbitrary large number of skipped messages, which may consume a lot of traffic and may halt delivery of up-to-date messages for some time.
+
+It may be better to have request for repeat delivery as a separate feature, that can be requested in necessary contexts - for example for group stability.
+
+Can servers delivery failure lead to de-sync? If message is lost on server and never delivered, ratchet wouldn't advance, so there's no room for de-sync? If yes, re-evaluate.

scripts/main/simplex-servers-stopscript

@@ -0,0 +1,30 @@
+#!/usr/bin/env sh
+set -eu
+
+path_conf_var="/var/opt"
+path_conf_smp="$path_conf_var/simplex"
+path_conf_xftp="$path_conf_var/simplex-xftp"
+path_conf_storelog_smp="$path_conf_smp/smp-server-store.log"
+path_conf_storelog_xftp="$path_conf_xftp/file-server-store.log"
+date="$(date -u '+%Y-%m-%dT%H:%M:%S')"
+
+backup_smp() {
+ if [ -e "$path_conf_storelog_smp" ]; then
+  cp "$path_conf_storelog_smp" "${path_conf_storelog_smp}.${date:-date-failed}"
+ fi
+}
+
+backup_xftp() {
+ if [ -e "$path_conf_storelog_xftp" ]; then
+  cp "$path_conf_storelog_xftp" "${path_conf_storelog_xftp}.${date:-date-failed}"
+ fi
+}
+
+if [ "$1" = 'smp-server' ]; then
+ backup_smp
+elif [ "$1" = 'xftp-server' ]; then
+ backup_xftp
+else
+ backup_smp
+ backup_xftp
+fi

scripts/main/simplex-servers-uninstall

@@ -13,6 +13,11 @@ fi
 printf "${RED}This action will permanently remove all configs, directories, binaries from Installation Script. Please backup any relevant configs if they are needed.${NC}\n\nPress ${GRN}ENTER${NC} to continue or ${RED}Ctrl+C${NC} to cancel installation"
 read ans
 
-rm -rf /var/opt/simplex /etc/opt/simplex /var/opt/simplex-xftp /etc/opt/simplex-xftp /srv/xftp /etc/systemd/system/smp-server.service /etc/systemd/system/xftp-server.service /usr/local/bin/smp-server /usr/local/bin/xftp-server /usr/local/bin/simplex-servers-update /usr/local/bin/simplex-servers-uninstall && userdel smp && userdel xftp
+systemctl stop smp-server
+systemctl stop xftp-server
+
+rm -rf /var/opt/simplex /etc/opt/simplex /var/opt/simplex-xftp /etc/opt/simplex-xftp /srv/xftp /etc/systemd/system/smp-server.service /etc/systemd/system/xftp-server.service /usr/local/bin/smp-server /usr/local/bin/xftp-server /usr/local/bin/simplex-servers-update /usr/local/bin/simplex-servers-uninstall
+
+userdel smp && userdel xftp
 
 printf "Uninstallation is complete! Thanks for trying out SimpleX!\n"

scripts/main/simplex-servers-update

@@ -11,6 +11,7 @@ scripts_systemd_smp="$scripts/smp-server.service"
 scripts_systemd_xftp="$scripts/xftp-server.service"
 scripts_update="$scripts/simplex-servers-update"
 scripts_uninstall="$scripts/simplex-servers-uninstall"
+scripts_stopscript="$scripts/simplex-servers-stopscript"
 
 # Default installation paths
 path_bin="/usr/local/bin"
@@ -18,6 +19,7 @@ path_bin_smp="$path_bin/smp-server"
 path_bin_xftp="$path_bin/xftp-server"
 path_bin_update="$path_bin/simplex-servers-update"
 path_bin_uninstall="$path_bin/simplex-servers-uninstall"
+path_bin_stopscript="$path_bin/simplex-servers-stopscript"
 
 path_systemd="/etc/systemd/system"
 path_systemd_smp="$path_systemd/smp-server.service"
@@ -27,6 +29,7 @@ path_systemd_xftp="$path_systemd/xftp-server.service"
 path_tmp_bin="$(mktemp -d)"
 path_tmp_bin_update="$path_tmp_bin/simplex-servers-update"
 path_tmp_bin_uninstall="$path_tmp_bin/simplex-servers-uninstall"
+path_tmp_bin_stopscript="$path_tmp_bin/simplex-servers-stopscript"
 path_tmp_systemd_smp="$path_tmp_bin/smp-server.service"
 path_tmp_systemd_xftp="$path_tmp_bin/xftp-server.service"
 
@@ -43,8 +46,9 @@ remote_version="$(curl --proto '=https' --tlsv1.2 -sSf -L https://api.github.com
 update_scripts() {
   curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_update" -o "$path_tmp_bin_update" && chmod +x "$path_tmp_bin_update"
   curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_uninstall" -o "$path_tmp_bin_uninstall" && chmod +x "$path_tmp_bin_uninstall"
+  curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_stopscript" -o "$path_tmp_bin_stopscript" && chmod +x "$path_tmp_bin_stopscript"
 
-  if diff -q "$path_bin_uninstall" "$path_tmp_bin_uninstall" > /dev/null; then
+  if diff -q "$path_bin_uninstall" "$path_tmp_bin_uninstall" > /dev/null 2>&1; then
     printf -- "- ${YLW}Uninstall script is up-to-date${NC}.\n"
     rm "$path_tmp_bin_uninstall"
   else
@@ -52,7 +56,15 @@ update_scripts() {
     mv "$path_tmp_bin_uninstall" "$path_bin_uninstall"
     printf "${GRN}Done!${NC}\n"
   fi
-  if diff -q "$path_bin_update" "$path_tmp_bin_update" > /dev/null; then
+  if diff -q "$path_bin_stopscript" "$path_tmp_bin_stopscript" > /dev/null 2>&1; then
+    printf -- "- ${YLW}Stopscript script is up-to-date${NC}.\n"
+    rm "$path_tmp_bin_stopscript"
+  else
+    printf -- "- Updating stopscript script..."
+    mv "$path_tmp_bin_stopscript" "$path_bin_stopscript"
+    printf "${GRN}Done!${NC}\n"
+  fi
+  if diff -q "$path_bin_update" "$path_tmp_bin_update" > /dev/null 2>&1; then
     printf -- "- ${YLW}Update script is up-to-date${NC}.\n"
     rm "$path_tmp_bin_update"
   else
@@ -68,7 +80,7 @@ update_systemd() {
   curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_systemd_smp" -o "$path_tmp_systemd_smp"
   curl --proto '=https' --tlsv1.2 -sSf -L "$scripts_systemd_xftp" -o "$path_tmp_systemd_xftp"
 
-  if diff -q "$path_systemd_smp" "$path_tmp_systemd_smp" > /dev/null; then
+  if diff -q "$path_systemd_smp" "$path_tmp_systemd_smp" > /dev/null 2>&1; then
     printf -- "- ${YLW}smp-server service is up-to-date${NC}.\n"
     rm "$path_tmp_systemd_smp"
   else
@@ -77,7 +89,7 @@ update_systemd() {
     systemctl daemon-reload
     printf "${GRN}Done!${NC}\n"
   fi
-  if diff -q "$path_systemd_xftp" "$path_tmp_systemd_xftp" > /dev/null; then
+  if diff -q "$path_systemd_xftp" "$path_tmp_systemd_xftp" > /dev/null 2>&1; then
     printf -- "- ${YLW}xftp-server service is up-to-date${NC}.\n"
     rm "$path_tmp_systemd_xftp"
   else
@@ -95,16 +107,16 @@ update_bins() {
       systemctl stop smp-server
       printf "${GRN}Done!${NC}\n"
 
-      print -- "- Updating smp-server bin to %s..." "$remote_version"
-      curl --proto '=https' --tlsv1.2 -sSf -L "$bin_smp" -o "$bin_path_smp" && chmod +x "$bin_path_smp"
+      printf -- "- Updating smp-server bin to %s..." "$remote_version"
+      curl --proto '=https' --tlsv1.2 -sSf -L "$bin_smp" -o "$path_bin_smp" && chmod +x "$path_bin_smp"
       printf "${GRN}Done!${NC}\n"
       
       printf -- "- Starting smp-server service..."
-      systemctl stop smp-server
+      systemctl start smp-server
       printf "${GRN}Done!${NC}\n"
     else
-      print -- "- Updating smp-server bin..."
-      curl --proto '=https' --tlsv1.2 -sSf -L "$bin_smp" -o "$bin_path_smp" && chmod +x "$bin_path_smp"
+      printf -- "- Updating smp-server bin..."
+      curl --proto '=https' --tlsv1.2 -sSf -L "$bin_smp" -o "$path_bin_smp" && chmod +x "$path_bin_smp"
       printf "${GRN}Done!${NC}\n"
     fi
 
@@ -113,16 +125,16 @@ update_bins() {
       systemctl stop xftp-server
       printf "${GRN}Done!${NC}\n"
 
-      print -- "- Updating xftp-server bin to %s..." "$remote_version"
-      curl --proto '=https' --tlsv1.2 -sSf -L "$bin_xftp" -o "$bin_path_xftp" && chmod +x "$bin_path_xftp"
+      printf -- "- Updating xftp-server bin to %s..." "$remote_version"
+      curl --proto '=https' --tlsv1.2 -sSf -L "$bin_xftp" -o "$path_bin_xftp" && chmod +x "$path_bin_xftp"
       printf "${GRN}Done!${NC}\n"
       
       printf -- "- Starting xftp-server service..."
-      systemctl stop xftp-server
+      systemctl start xftp-server
       printf "${GRN}Done!${NC}\n"
     else
-      print -- "- Updating xftp-server bin..."
-      curl --proto '=https' --tlsv1.2 -sSf -L "$bin_smp" -o "$bin_path_xftp" && chmod +x "$bin_path_xftp"
+      printf -- "- Updating xftp-server bin..."
+      curl --proto '=https' --tlsv1.2 -sSf -L "$bin_smp" -o "$path_bin_xftp" && chmod +x "$path_bin_xftp"
       printf "${GRN}Done!${NC}\n"
     fi
   else
@@ -155,6 +167,8 @@ main() {
   
   printf "Updating simplex server binaries...\n"
   update_bins
+
+  rm -rf "$path_tmp_bin"
 }
 
 main "$@"

scripts/main/smp-server.service

@@ -6,7 +6,7 @@ User=smp
 Group=smp
 Type=simple
 ExecStart=/usr/local/bin/smp-server start +RTS -N -RTS 
-ExecStopPost=/usr/bin/env sh -c '[ -e "/var/opt/simplex/smp-server-store.log" ] && cp "/var/opt/simplex/smp-server-store.log" "/var/opt/simplex/smp-server-store.log.$(date +%FT%T)'
+ExecStopPost=/usr/local/bin/simplex-servers-stopscript smp-server
 LimitNOFILE=65535
 KillSignal=SIGINT
 TimeoutStopSec=infinity

scripts/main/xftp-server.service

@@ -6,7 +6,7 @@ User=xftp
 Group=xftp
 Type=simple
 ExecStart=/usr/local/bin/xftp-server start +RTS -N -RTS
-ExecStopPost=/usr/bin/env sh -c '[ -e "/var/opt/simplex-xftp/file-server-store.log" ] && cp "/var/opt/simplex-xftp/file-server-store.log" "/var/opt/simplex-xftp/file-server-store.log.$(date +%FT%T)'
+ExecStopPost=/usr/local/bin/simplex-servers-stopscript xftp-server
 LimitNOFILE=65535
 KillSignal=SIGINT
 TimeoutStopSec=infinity

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        5.1.1
+version:        5.3.0.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -34,7 +34,6 @@ flag swift
 
 library
   exposed-modules:
-      Simplex.FileTransfer
       Simplex.FileTransfer.Agent
       Simplex.FileTransfer.Client
       Simplex.FileTransfer.Client.Agent
@@ -63,6 +62,7 @@ library
       Simplex.Messaging.Agent.Server
       Simplex.Messaging.Agent.Store
       Simplex.Messaging.Agent.Store.SQLite
+      Simplex.Messaging.Agent.Store.SQLite.Common
       Simplex.Messaging.Agent.Store.SQLite.Migrations
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys
@@ -82,6 +82,9 @@ library
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230401_snd_files
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230510_files_pending_replicas_indexes
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230516_encrypted_rcv_message_hashes
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230531_switch_status
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230615_ratchet_sync
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230701_delivery_receipts
       Simplex.Messaging.Agent.TAsyncs
       Simplex.Messaging.Agent.TRcvQueues
       Simplex.Messaging.Client
@@ -106,6 +109,7 @@ library
       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.Main
@@ -531,6 +535,7 @@ test-suite simplexmq-test
       CoreTests.EncodingTests
       CoreTests.ProtocolErrorTests
       CoreTests.RetryIntervalTests
+      CoreTests.UtilTests
       CoreTests.VersionRangeTests
       FileDescriptionTests
       NtfClient
@@ -545,7 +550,7 @@ test-suite simplexmq-test
       Paths_simplexmq
   hs-source-dirs:
       tests
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -fhpc
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns
   build-depends:
       HUnit ==1.6.*
     , QuickCheck ==2.14.*

src/Simplex/FileTransfer.hs

@@ -1,14 +0,0 @@
-module Simplex.FileTransfer where
-
--- TODO
--- Protocol
--- Store (in memory storage)
--- StoreLog (append only log)
--- FileDescription
--- Server
--- Client
--- Server/Main (server CLI)
--- Client/Main (client CLI)
---
--- Transport for HTTP2 ?
--- streaming Crypto

src/Simplex/FileTransfer/Agent.hs

@@ -71,7 +71,6 @@ import System.FilePath (takeFileName, (</>))
 import UnliftIO
 import UnliftIO.Concurrent
 import UnliftIO.Directory
-import qualified UnliftIO.Exception as E
 
 startWorkers :: AgentMonad m => AgentClient -> Maybe FilePath -> m ()
 startWorkers c workDir = do
@@ -162,7 +161,7 @@ addWorker c wsSel runWorker runWorkerNoSrv srv_ = do
       let runWorker' = case srv_ of
             Just srv -> runWorker c srv doWork
             Nothing -> runWorkerNoSrv c doWork
-      worker <- async $ runWorker' `E.finally` atomically (TM.delete srv_ ws)
+      worker <- async $ runWorker' `agentFinally` atomically (TM.delete srv_ ws)
       atomically $ TM.insert srv_ (doWork, worker) ws
     Just (doWork, _) ->
       void . atomically $ tryPutTMVar doWork ()
@@ -187,10 +186,10 @@ runXFTPRcvWorker c srv doWork = do
           let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay
           withRetryInterval ri' $ \delay' loop ->
             downloadFileChunk fc replica
-              `catchError` \e -> retryOnError "XFTP rcv worker" (retryLoop loop e delay') (retryDone e) e
+              `catchAgentError` \e -> retryOnError "XFTP rcv worker" (retryLoop loop e delay') (retryDone e) e
           where
             retryLoop loop e replicaDelay = do
-              flip catchError (\_ -> pure ()) $ do
+              flip catchAgentError (\_ -> pure ()) $ do
                 notifyOnRetry <- asks (xftpNotifyErrsOnRetry . config)
                 when notifyOnRetry $ notify c rcvFileEntityId $ RFERR e
                 closeXFTPServerClient c userId server digest
@@ -249,7 +248,7 @@ runXFTPRcvLocalWorker c doWork = do
       case nextFile of
         Nothing -> noWorkToDo
         Just f@RcvFile {rcvFileId, rcvFileEntityId, tmpPath} ->
-          decryptFile f `catchError` (rcvWorkerInternalError c rcvFileId rcvFileEntityId tmpPath . show)
+          decryptFile f `catchAgentError` (rcvWorkerInternalError c rcvFileId rcvFileEntityId tmpPath . show)
     noWorkToDo = void . atomically $ tryTakeTMVar doWork
     decryptFile :: RcvFile -> m ()
     decryptFile RcvFile {rcvFileId, rcvFileEntityId, key, nonce, tmpPath, savePath, status, chunks} = do
@@ -300,7 +299,7 @@ sendFileExperimental c@AgentClient {xftpServers} userId filePath numRecipients =
       createDirectory outputDir
       let tempPath = workPath </> "snd"
       createDirectoryIfMissing False tempPath
-      runSend fileName outputDir tempPath `catchError` \e -> do
+      runSend fileName outputDir tempPath `catchAgentError` \e -> do
         cleanup outputDir tempPath
         notify c sndFileId $ SFERR e
       where
@@ -370,7 +369,7 @@ runXFTPSndPrepareWorker c doWork = do
       case nextFile of
         Nothing -> noWorkToDo
         Just f@SndFile {sndFileId, sndFileEntityId, prefixPath} ->
-          prepareFile f `catchError` (sndWorkerInternalError c sndFileId sndFileEntityId prefixPath . show)
+          prepareFile f `catchAgentError` (sndWorkerInternalError c sndFileId sndFileEntityId prefixPath . show)
     noWorkToDo = void . atomically $ tryTakeTMVar doWork
     prepareFile :: SndFile -> m ()
     prepareFile SndFile {prefixPath = Nothing} =
@@ -424,7 +423,7 @@ runXFTPSndPrepareWorker c doWork = do
               usedSrvs <- newTVarIO ([] :: [XFTPServer])
               withRetryInterval (riFast ri) $ \_ loop ->
                 createWithNextSrv usedSrvs
-                  `catchError` \e -> retryOnError "XFTP prepare worker" (retryLoop loop) (throwError e) e
+                  `catchAgentError` \e -> retryOnError "XFTP prepare worker" (retryLoop loop) (throwError e) e
               where
                 retryLoop loop = atomically (assertAgentForeground c) >> loop
             createWithNextSrv usedSrvs = do
@@ -460,10 +459,10 @@ runXFTPSndWorker c srv doWork = do
           let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay
           withRetryInterval ri' $ \delay' loop ->
             uploadFileChunk fc replica
-              `catchError` \e -> retryOnError "XFTP snd worker" (retryLoop loop e delay') (retryDone e) e
+              `catchAgentError` \e -> retryOnError "XFTP snd worker" (retryLoop loop e delay') (retryDone e) e
           where
             retryLoop loop e replicaDelay = do
-              flip catchError (\_ -> pure ()) $ do
+              flip catchAgentError (\_ -> pure ()) $ do
                 notifyOnRetry <- asks (xftpNotifyErrsOnRetry . config)
                 when notifyOnRetry $ notify c sndFileEntityId $ SFERR e
                 closeXFTPServerClient c userId server digest
@@ -579,8 +578,8 @@ deleteSndFileInternal c sndFileEntityId = do
 
 deleteSndFileRemote :: forall m. AgentMonad m => AgentClient -> UserId -> SndFileId -> ValidFileDescription 'FSender -> m ()
 deleteSndFileRemote c userId sndFileEntityId (ValidFileDescription FileDescription {chunks}) = do
-  deleteSndFileInternal c sndFileEntityId `catchError` (notify c sndFileEntityId . SFERR)
-  forM_ chunks $ \ch -> deleteFileChunk ch `catchError` (notify c sndFileEntityId . SFERR)
+  deleteSndFileInternal c sndFileEntityId `catchAgentError` (notify c sndFileEntityId . SFERR)
+  forM_ chunks $ \ch -> deleteFileChunk ch `catchAgentError` (notify c sndFileEntityId . SFERR)
   where
     deleteFileChunk :: FileChunk -> m ()
     deleteFileChunk FileChunk {digest, replicas = replica@FileChunkReplica {server} : _} = do
@@ -594,7 +593,7 @@ addXFTPDelWorker c srv = do
   atomically (TM.lookup srv ws) >>= \case
     Nothing -> do
       doWork <- newTMVarIO ()
-      worker <- async $ runXFTPDelWorker c srv doWork `E.finally` atomically (TM.delete srv ws)
+      worker <- async $ runXFTPDelWorker c srv doWork `agentFinally` atomically (TM.delete srv ws)
       atomically $ TM.insert srv (doWork, worker) ws
     Just (doWork, _) ->
       void . atomically $ tryPutTMVar doWork ()
@@ -619,10 +618,10 @@ runXFTPDelWorker c srv doWork = do
           let ri' = maybe ri (\d -> ri {initialInterval = d, increaseAfter = 0}) delay
           withRetryInterval ri' $ \delay' loop ->
             deleteChunkReplica replica
-              `catchError` \e -> retryOnError "XFTP del worker" (retryLoop loop e delay') (retryDone e) e
+              `catchAgentError` \e -> retryOnError "XFTP del worker" (retryLoop loop e delay') (retryDone e) e
           where
             retryLoop loop e replicaDelay = do
-              flip catchError (\_ -> pure ()) $ do
+              flip catchAgentError (\_ -> pure ()) $ do
                 notifyOnRetry <- asks (xftpNotifyErrsOnRetry . config)
                 when notifyOnRetry $ notify c "" $ SFERR e
                 closeXFTPServerClient c userId server chunkDigest

src/Simplex/FileTransfer/Client/Main.hs

@@ -38,9 +38,8 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Char (toLower)
-import Data.Function (on)
 import Data.Int (Int64)
-import Data.List (foldl', groupBy, sortOn)
+import Data.List (foldl', sortOn)
 import Data.List.NonEmpty (NonEmpty (..), nonEmpty)
 import qualified Data.List.NonEmpty as L
 import Data.Map (Map)
@@ -66,7 +65,7 @@ import Simplex.Messaging.Encoding.String (StrEncoding (..))
 import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Protocol (ProtoServerWithAuth (..), SenderId, SndPrivateSignKey, XFTPServer, XFTPServerWithAuth)
 import Simplex.Messaging.Server.CLI (getCliCommand')
-import Simplex.Messaging.Util (ifM, tshow, whenM)
+import Simplex.Messaging.Util (groupAllOn, ifM, tshow, whenM)
 import System.Exit (exitFailure)
 import System.FilePath (splitFileName, (</>))
 import System.IO.Temp (getCanonicalTemporaryDirectory)
@@ -75,7 +74,7 @@ import UnliftIO
 import UnliftIO.Directory
 
 xftpClientVersion :: String
-xftpClientVersion = "0.1.0"
+xftpClientVersion = "1.0.1"
 
 chunkSize1 :: Word32
 chunkSize1 = kb 256
@@ -316,7 +315,7 @@ cliSendFileOpts SendOptions {filePath, outputDir, numRecipients, xftpServers, re
       let xftpSrvs = fromMaybe defaultXFTPServers (nonEmpty xftpServers)
       srvs <- liftIO $ replicateM (length chunks) $ getXFTPServer gen xftpSrvs
       let thd3 (_, _, x) = x
-          chunks' = groupBy ((==) `on` thd3) $ sortOn thd3 $ zip3 [1 ..] chunks srvs
+          chunks' = groupAllOn thd3 $ zip3 [1 ..] chunks srvs
       -- TODO shuffle/unshuffle chunks
       -- the reason we don't do pooled downloads here within one server is that http2 library doesn't handle cleint concurrency, even though
       -- upload doesn't allow other requests within the same client until complete (but download does allow).
@@ -428,7 +427,7 @@ cliReceiveFile ReceiveOptions {fileDescription, filePath, retryCount, tempPath,
       liftIO $ printNoNewLine "Downloading file..."
       downloadedChunks <- newTVarIO []
       let srv FileChunk {replicas} = server (head replicas :: FileChunkReplica)
-          srvChunks = groupBy ((==) `on` srv) $ sortOn srv chunks
+          srvChunks = groupAllOn srv chunks
       chunkPaths <- map snd . sortOn fst . concat <$> pooledForConcurrentlyN 16 srvChunks (mapM $ downloadFileChunk a encPath size downloadedChunks)
       encDigest <- liftIO $ LC.sha512Hash <$> readChunks chunkPaths
       when (encDigest /= unFileDigest digest) $ throwError $ CLIError "File digest mismatch"

src/Simplex/FileTransfer/Description.hs

@@ -42,9 +42,8 @@ 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 Data.Function (on)
 import Data.Int (Int64)
-import Data.List (foldl', groupBy, sortOn)
+import Data.List (foldl', sortOn)
 import Data.Map (Map)
 import qualified Data.Map as M
 import Data.Maybe (fromMaybe)
@@ -59,7 +58,7 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll)
 import Simplex.Messaging.Protocol (XFTPServer)
-import Simplex.Messaging.Util (bshow, (<$?>))
+import Simplex.Messaging.Util (bshow, groupAllOn, (<$?>))
 
 data FileDescription (p :: FileParty) = FileDescription
   { party :: SFileParty p,
@@ -258,9 +257,7 @@ instance (ToField a) => ToField (FileSize a) where toField (FileSize s) = toFiel
 
 groupReplicasByServer :: FileSize Word32 -> [FileChunk] -> [[FileServerReplica]]
 groupReplicasByServer defChunkSize =
-  groupBy ((==) `on` replicaServer)
-    . sortOn replicaServer
-    . unfoldChunksToReplicas defChunkSize
+  groupAllOn replicaServer . unfoldChunksToReplicas defChunkSize
 
 encodeFileReplicas :: FileSize Word32 -> [FileChunk] -> [YAMLServerReplicas]
 encodeFileReplicas defChunkSize =
@@ -268,7 +265,7 @@ encodeFileReplicas defChunkSize =
   where
     encodeServerReplicas fs =
       YAMLServerReplicas
-        { server = replicaServer $ head fs, -- groupBy guarantees that fs is not empty
+        { server = replicaServer $ head fs, -- groupAllOn guarantees that fs is not empty
           chunks = map (B.unpack . encodeServerReplica) fs
         }
 

src/Simplex/FileTransfer/Server.hs

@@ -70,7 +70,7 @@ runXFTPServerBlocking :: TMVar Bool -> XFTPServerConfig -> IO ()
 runXFTPServerBlocking started cfg = newXFTPServerEnv cfg >>= runReaderT (xftpServer cfg started)
 
 xftpServer :: XFTPServerConfig -> TMVar Bool -> M ()
-xftpServer cfg@XFTPServerConfig {xftpPort, logTLSErrors} started = do
+xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig} started = do
   restoreServerStats
   raceAny_ (runServer : expireFilesThread_ cfg <> serverStatsThread_ cfg) `finally` stopServer
   where
@@ -79,7 +79,7 @@ xftpServer cfg@XFTPServerConfig {xftpPort, logTLSErrors} started = do
       serverParams <- asks tlsServerParams
       env <- ask
       liftIO $
-        runHTTP2Server started xftpPort defaultHTTP2BufferSize serverParams logTLSErrors $ \sessionId r sendResponse -> do
+        runHTTP2Server started xftpPort defaultHTTP2BufferSize serverParams transportConfig $ \sessionId r sendResponse -> do
           reqBody <- getHTTP2Body r xftpBlockSize
           processRequest HTTP2Request {sessionId, request = r, reqBody, sendResponse} `runReaderT` env
 

src/Simplex/FileTransfer/Server/Env.hs

@@ -26,7 +26,7 @@ import Simplex.FileTransfer.Server.StoreLog
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol (BasicAuth, RcvPublicVerifyKey)
 import Simplex.Messaging.Server.Expiration
-import Simplex.Messaging.Transport.Server (loadFingerprint, loadTLSServerParams)
+import Simplex.Messaging.Transport.Server (loadFingerprint, loadTLSServerParams, TransportServerConfig)
 import Simplex.Messaging.Util (tshow)
 import System.IO (IOMode (..))
 import UnliftIO.STM
@@ -55,7 +55,7 @@ data XFTPServerConfig = XFTPServerConfig
     logStatsStartTime :: Int64,
     serverStatsLogFile :: FilePath,
     serverStatsBackupFile :: Maybe FilePath,
-    logTLSErrors :: Bool
+    transportConfig :: TransportServerConfig
   }
 
 data XFTPEnv = XFTPEnv

src/Simplex/FileTransfer/Server/Main.hs

@@ -25,13 +25,14 @@ import Simplex.Messaging.Protocol (ProtoServerWithAuth (..), pattern XFTPServer)
 import Simplex.Messaging.Server.CLI
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Transport.Client (TransportHost (..))
+import Simplex.Messaging.Transport.Server (TransportServerConfig (..), defaultTransportServerConfig)
 import System.Directory (createDirectoryIfMissing, doesFileExist)
 import System.FilePath (combine)
 import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
 import Text.Read (readMaybe)
 
 xftpServerVersion :: String
-xftpServerVersion = "1.0.0"
+xftpServerVersion = "1.0.1"
 
 xftpServerCLI :: FilePath -> FilePath -> IO ()
 xftpServerCLI cfgPath logPath = do
@@ -151,7 +152,10 @@ xftpServerCLI cfgPath logPath = do
               logStatsStartTime = 0, -- seconds from 00:00 UTC
               serverStatsLogFile = combine logPath "file-server-stats.daily.log",
               serverStatsBackupFile = logStats $> combine logPath "file-server-stats.log",
-              logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini
+              transportConfig =
+                defaultTransportServerConfig
+                  { logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini
+                  }
             }
 
 data CliCommand

src/Simplex/Messaging/Agent/Client.hs

@@ -41,7 +41,9 @@ module Simplex.Messaging.Agent.Client
     temporaryOrHostError,
     secureQueue,
     enableQueueNotifications,
+    enableQueuesNtfs,
     disableQueueNotifications,
+    disableQueuesNtfs,
     sendAgentMessage,
     agentNtfRegisterToken,
     agentNtfVerifyToken,
@@ -161,9 +163,6 @@ import Simplex.Messaging.Protocol
     ErrorType,
     MsgFlags (..),
     MsgId,
-    NotifierId,
-    NtfPrivateSignKey,
-    NtfPublicVerifyKey,
     NtfServer,
     ProtoServer,
     ProtoServerWithAuth (..),
@@ -400,7 +399,7 @@ instance ProtocolServerClient XFTPErrorType FileResponse where
 
 getSMPServerClient :: forall m. AgentMonad m => AgentClient -> SMPTransportSession -> m SMPClient
 getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv, _) = do
-  unlessM (readTVarIO active) . throwError $ INTERNAL "agent is stopped"
+  unlessM (readTVarIO active) . throwError $ INACTIVE
   atomically (getClientVar tSess smpClients)
     >>= either
       (newProtocolClient c tSess smpClients connectClient reconnectSMPClient)
@@ -444,7 +443,7 @@ reconnectServer c tSess = newAsyncAction tryReconnectSMPClient $ reconnections c
     tryReconnectSMPClient aId = do
       ri <- asks $ reconnectInterval . config
       withRetryInterval ri $ \_ loop ->
-        reconnectSMPClient c tSess `catchError` const loop
+        reconnectSMPClient c tSess `catchAgentError` const loop
       atomically . removeAsyncAction aId $ reconnections c
 
 reconnectSMPClient :: forall m. AgentMonad m => AgentClient -> SMPTransportSession -> m ()
@@ -468,7 +467,7 @@ reconnectSMPClient c tSess@(_, srv, _) =
 
 getNtfServerClient :: forall m. AgentMonad m => AgentClient -> NtfTransportSession -> m NtfClient
 getNtfServerClient c@AgentClient {active, ntfClients} tSess@(userId, srv, _) = do
-  unlessM (readTVarIO active) . throwError $ INTERNAL "agent is stopped"
+  unlessM (readTVarIO active) . throwError $ INACTIVE
   atomically (getClientVar tSess ntfClients)
     >>= either
       (newProtocolClient c tSess ntfClients connectClient $ \_ _ -> pure ())
@@ -488,7 +487,7 @@ getNtfServerClient c@AgentClient {active, ntfClients} tSess@(userId, srv, _) = d
 
 getXFTPServerClient :: forall m. AgentMonad m => AgentClient -> XFTPTransportSession -> m XFTPClient
 getXFTPServerClient c@AgentClient {active, xftpClients, useNetworkConfig} tSess@(userId, srv, _) = do
-  unlessM (readTVarIO active) . throwError $ INTERNAL "agent is stopped"
+  unlessM (readTVarIO active) . throwError $ INACTIVE
   atomically (getClientVar tSess xftpClients)
     >>= either
       (newProtocolClient c tSess xftpClients connectClient $ \_ _ -> pure ())
@@ -641,7 +640,7 @@ withLockMap_ locks key = withGetLock $ TM.lookup key locks >>= maybe newLock pur
 withClient_ :: forall a m err msg. (AgentMonad m, ProtocolServerClient err msg) => AgentClient -> TransportSession msg -> ByteString -> (Client msg -> m a) -> m a
 withClient_ c tSess@(userId, srv, _) statCmd action = do
   cl <- getProtocolServerClient c tSess
-  (action cl <* stat cl "OK") `catchError` logServerError cl
+  (action cl <* stat cl "OK") `catchAgentError` logServerError cl
   where
     stat cl = liftIO . incClientStat c userId cl statCmd
     logServerError :: Client msg -> AgentErrorType -> m a
@@ -841,6 +840,7 @@ newRcvQueue c userId connId (ProtoServerWithAuth srv auth) vRange = do
             dbQueueId = 0,
             primary = True,
             dbReplaceQueueId = Nothing,
+            rcvSwchStatus = Nothing,
             smpClientVersion = maxVersion vRange,
             clientNtfCreds = Nothing,
             deleteErrors = 0
@@ -860,6 +860,7 @@ temporaryAgentError :: AgentErrorType -> Bool
 temporaryAgentError = \case
   BROKER _ NETWORK -> True
   BROKER _ TIMEOUT -> True
+  INACTIVE -> True
   _ -> False
 
 temporaryOrHostError :: AgentErrorType -> Bool
@@ -875,11 +876,13 @@ subscribeQueues c qs = do
     modifyTVar (subscrConns c) $ S.insert connId
     RQ.addQueue rq $ pendingSubs c
   u <- askUnliftIO
-  (errs <>) <$> sendTSessionBatches "SUB" 90 (subscribeQueues_ u) c qs
+  -- only "checked" queues are subscribed
+  (errs <>) <$> sendTSessionBatches "SUB" 90 id (subscribeQueues_ u) c qs'
   where
     checkQueue rq@RcvQueue {rcvId, server} = do
       prohibited <- atomically . TM.member (server, rcvId) $ getMsgLocks c
       pure $ if prohibited then Left (rq, Left $ CMD PROHIBITED) else Right rq
+    subscribeQueues_ :: UnliftIO m -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError ())
     subscribeQueues_ u smp qs' = do
       rs <- sendBatch subscribeSMPQueues smp qs'
       mapM_ (uncurry $ processSubResult c) rs
@@ -887,25 +890,25 @@ subscribeQueues c qs = do
         unliftIO u $ reconnectServer c $ transportSession' smp
       pure rs
 
-type BatchResponses e = (NonEmpty (RcvQueue, Either e ()))
+type BatchResponses e r = (NonEmpty (RcvQueue, Either e r))
 
 -- statBatchSize is not used to batch the commands, only for traffic statistics
-sendTSessionBatches :: forall m. AgentMonad m => ByteString -> Int -> (SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError)) -> AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
-sendTSessionBatches statCmd statBatchSize action c qs =
+sendTSessionBatches :: forall m q r. AgentMonad m => ByteString -> Int -> (q -> RcvQueue) -> (SMPClient -> NonEmpty q -> IO (BatchResponses SMPClientError r)) -> AgentClient -> [q] -> m [(RcvQueue, Either AgentErrorType r)]
+sendTSessionBatches statCmd statBatchSize toRQ action c qs =
   concatMap L.toList <$> (mapConcurrently sendClientBatch =<< batchQueues)
   where
-    batchQueues :: m [(SMPTransportSession, NonEmpty RcvQueue)]
+    batchQueues :: m [(SMPTransportSession, NonEmpty q)]
     batchQueues = do
       mode <- sessionMode <$> readTVarIO (useNetworkConfig c)
       pure . M.assocs $ foldl' (batch mode) M.empty qs
       where
-        batch mode m rq =
-          let tSess = mkSMPTSession rq mode
-           in M.alter (Just . maybe [rq] (rq <|)) tSess m
-    sendClientBatch :: (SMPTransportSession, NonEmpty RcvQueue) -> m (BatchResponses AgentErrorType)
+        batch mode m q =
+          let tSess = mkSMPTSession (toRQ q) mode
+           in M.alter (Just . maybe [q] (q <|)) tSess m
+    sendClientBatch :: (SMPTransportSession, NonEmpty q) -> m (BatchResponses AgentErrorType r)
     sendClientBatch (tSess@(userId, srv, _), qs') =
       tryError (getSMPServerClient c tSess) >>= \case
-        Left e -> pure $ L.map (,Left e) qs'
+        Left e -> pure $ L.map ((,Left e) . toRQ) qs'
         Right smp -> liftIO $ do
           logServer "-->" c srv (bshow (length qs') <> " queues") statCmd
           rs <- L.map agentError <$> action smp qs'
@@ -915,9 +918,9 @@ sendTSessionBatches statCmd statBatchSize action c qs =
             agentError = second . first $ protocolClientError SMP $ clientServer smp
             statBatch =
               let n = (length qs - 1) `div` statBatchSize + 1
-               in incClientStatN c userId smp n (statCmd <> "S") "OK"
+               in incClientStatN c userId smp n statCmd "OK"
 
-sendBatch :: (SMPClient -> NonEmpty (SMP.RcvPrivateSignKey, SMP.RecipientId) -> IO (NonEmpty (Either SMPClientError ()))) -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError)
+sendBatch :: (SMPClient -> NonEmpty (SMP.RcvPrivateSignKey, SMP.RecipientId) -> IO (NonEmpty (Either SMPClientError ()))) -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError ())
 sendBatch smpCmdFunc smp qs = L.zip qs <$> smpCmdFunc smp (L.map queueCreds qs)
   where
     queueCreds RcvQueue {rcvPrivateKey, rcvId} = (rcvPrivateKey, rcvId)
@@ -1000,16 +1003,28 @@ secureQueue c rq@RcvQueue {rcvId, rcvPrivateKey} senderKey =
   withSMPClient c rq "KEY <key>" $ \smp ->
     secureSMPQueue smp rcvPrivateKey rcvId senderKey
 
-enableQueueNotifications :: AgentMonad m => AgentClient -> RcvQueue -> NtfPublicVerifyKey -> RcvNtfPublicDhKey -> m (NotifierId, RcvNtfPublicDhKey)
+enableQueueNotifications :: AgentMonad m => AgentClient -> RcvQueue -> SMP.NtfPublicVerifyKey -> SMP.RcvNtfPublicDhKey -> m (SMP.NotifierId, SMP.RcvNtfPublicDhKey)
 enableQueueNotifications c rq@RcvQueue {rcvId, rcvPrivateKey} notifierKey rcvNtfPublicDhKey =
   withSMPClient c rq "NKEY <nkey>" $ \smp ->
     enableSMPQueueNotifications smp rcvPrivateKey rcvId notifierKey rcvNtfPublicDhKey
 
+enableQueuesNtfs :: forall m. AgentMonad m => AgentClient -> [(RcvQueue, SMP.NtfPublicVerifyKey, SMP.RcvNtfPublicDhKey)] -> m [(RcvQueue, Either AgentErrorType (SMP.NotifierId, SMP.RcvNtfPublicDhKey))]
+enableQueuesNtfs = sendTSessionBatches "NKEY" 90 fst3 enableQueues_
+  where
+    fst3 (x, _, _) = x
+    enableQueues_ :: SMPClient -> NonEmpty (RcvQueue, SMP.NtfPublicVerifyKey, SMP.RcvNtfPublicDhKey) -> IO (NonEmpty (RcvQueue, Either (ProtocolClientError ErrorType) (SMP.NotifierId, RcvNtfPublicDhKey)))
+    enableQueues_ smp qs' = L.zipWith ((,) . fst3) qs' <$> enableSMPQueuesNtfs smp (L.map queueCreds qs')
+    queueCreds :: (RcvQueue, SMP.NtfPublicVerifyKey, SMP.RcvNtfPublicDhKey) -> (SMP.RcvPrivateSignKey, SMP.RecipientId, SMP.NtfPublicVerifyKey, SMP.RcvNtfPublicDhKey)
+    queueCreds (RcvQueue {rcvPrivateKey, rcvId}, notifierKey, rcvNtfPublicDhKey) = (rcvPrivateKey, rcvId, notifierKey, rcvNtfPublicDhKey)
+
 disableQueueNotifications :: AgentMonad m => AgentClient -> RcvQueue -> m ()
 disableQueueNotifications c rq@RcvQueue {rcvId, rcvPrivateKey} =
   withSMPClient c rq "NDEL" $ \smp ->
     disableSMPQueueNotifications smp rcvPrivateKey rcvId
 
+disableQueuesNtfs :: forall m. AgentMonad m => AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
+disableQueuesNtfs = sendTSessionBatches "NDEL" 90 id $ sendBatch disableSMPQueuesNtfs
+
 sendAck :: AgentMonad m => AgentClient -> RcvQueue -> MsgId -> m ()
 sendAck c rq@RcvQueue {rcvId, rcvPrivateKey} msgId = do
   withSMPClient c rq "ACK" $ \smp ->
@@ -1031,7 +1046,7 @@ deleteQueue c rq@RcvQueue {rcvId, rcvPrivateKey} = do
     deleteSMPQueue smp rcvPrivateKey rcvId
 
 deleteQueues :: forall m. AgentMonad m => AgentClient -> [RcvQueue] -> m [(RcvQueue, Either AgentErrorType ())]
-deleteQueues = sendTSessionBatches "DEL" 90 $ sendBatch deleteSMPQueues
+deleteQueues = sendTSessionBatches "DEL" 90 id $ sendBatch deleteSMPQueues
 
 sendAgentMessage :: AgentMonad m => AgentClient -> SndQueue -> MsgFlags -> ByteString -> m ()
 sendAgentMessage c sq@SndQueue {sndId, sndPrivateKey} msgFlags agentMsg =
@@ -1064,7 +1079,7 @@ agentNtfEnableCron :: AgentMonad m => AgentClient -> NtfTokenId -> NtfToken -> W
 agentNtfEnableCron c tknId NtfToken {ntfServer, ntfPrivKey} interval =
   withNtfClient c ntfServer tknId "TCRN" $ \ntf -> ntfEnableCron ntf ntfPrivKey tknId interval
 
-agentNtfCreateSubscription :: AgentMonad m => AgentClient -> NtfTokenId -> NtfToken -> SMPQueueNtf -> NtfPrivateSignKey -> m NtfSubscriptionId
+agentNtfCreateSubscription :: AgentMonad m => AgentClient -> NtfTokenId -> NtfToken -> SMPQueueNtf -> SMP.NtfPrivateSignKey -> m NtfSubscriptionId
 agentNtfCreateSubscription c tknId NtfToken {ntfServer, ntfPrivKey} smpQueue nKey =
   withNtfClient c ntfServer tknId "SNEW" $ \ntf -> ntfCreateSubscription ntf ntfPrivKey (NewNtfSub tknId smpQueue nKey)
 

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

@@ -6,6 +6,7 @@
 {-# LANGUAGE NamedFieldPuns #-}
 {-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE RankNTypes #-}
+{-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications #-}
 {-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-}
 
@@ -17,6 +18,9 @@ module Simplex.Messaging.Agent.Env.SQLite
     NetworkConfig (..),
     defaultAgentConfig,
     defaultReconnectInterval,
+    tryAgentError,
+    catchAgentError,
+    agentFinally,
     Env (..),
     newSMPAgentEnv,
     createAgentStore,
@@ -52,9 +56,10 @@ import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (TLS, Transport (..))
 import Simplex.Messaging.Transport.Client (defaultSMPPort)
+import Simplex.Messaging.Util (allFinally, catchAllErrors, tryAllErrors)
 import Simplex.Messaging.Version
 import System.Random (StdGen, newStdGen)
-import UnliftIO (Async)
+import UnliftIO (Async, SomeException)
 import UnliftIO.STM
 
 type AgentMonad' m = (MonadUnliftIO m, MonadReader Env m)
@@ -82,7 +87,8 @@ data AgentConfig = AgentConfig
     helloTimeout :: NominalDiffTime,
     initialCleanupDelay :: Int64,
     cleanupInterval :: Int64,
-    rcvMsgHashesTTL :: NominalDiffTime,
+    cleanupStepInterval :: Int,
+    storedMsgDataTTL :: NominalDiffTime,
     rcvFilesTTL :: NominalDiffTime,
     sndFilesTTL :: NominalDiffTime,
     xftpNotifyErrsOnRetry :: Bool,
@@ -146,7 +152,8 @@ defaultAgentConfig =
       helloTimeout = 2 * nominalDay,
       initialCleanupDelay = 30 * 1000000, -- 30 seconds
       cleanupInterval = 30 * 60 * 1000000, -- 30 minutes
-      rcvMsgHashesTTL = 30 * nominalDay,
+      cleanupStepInterval = 200000, -- 200ms
+      storedMsgDataTTL = 21 * nominalDay,
       rcvFilesTTL = 2 * nominalDay,
       sndFilesTTL = nominalDay,
       xftpNotifyErrsOnRetry = True,
@@ -223,3 +230,19 @@ newXFTPAgent = do
   xftpSndWorkers <- TM.empty
   xftpDelWorkers <- TM.empty
   pure XFTPAgent {xftpWorkDir, xftpRcvWorkers, xftpSndWorkers, xftpDelWorkers}
+
+tryAgentError :: AgentMonad m => m a -> m (Either AgentErrorType a)
+tryAgentError = tryAllErrors mkInternal
+{-# INLINE tryAgentError #-}
+
+catchAgentError :: AgentMonad m => m a -> (AgentErrorType -> m a) -> m a
+catchAgentError = catchAllErrors mkInternal
+{-# INLINE catchAgentError #-}
+
+agentFinally :: AgentMonad m => m a -> m b -> m a
+agentFinally = allFinally mkInternal
+{-# INLINE agentFinally #-}
+
+mkInternal :: SomeException -> AgentErrorType
+mkInternal = INTERNAL . show
+{-# INLINE mkInternal #-}

src/Simplex/Messaging/Agent/NtfSubSupervisor.hs

@@ -147,7 +147,7 @@ processNtfSub c (connId, cmd) = do
       atomically (TM.lookup srv ws) >>= \case
         Nothing -> do
           doWork <- newTMVarIO ()
-          worker <- async $ runWorker c srv doWork `E.finally` atomically (TM.delete srv ws)
+          worker <- async $ runWorker c srv doWork `agentFinally` atomically (TM.delete srv ws)
           atomically $ TM.insert srv (doWork, worker) ws
         Just (doWork, _) ->
           void . atomically $ tryPutTMVar doWork ()
@@ -173,7 +173,7 @@ runNtfWorker c srv doWork = do
           ri <- asks $ reconnectInterval . config
           withRetryInterval ri $ \_ loop ->
             processAction a
-              `catchError` retryOnError c "NtfWorker" loop (workerInternalError c connId . show)
+              `catchAgentError` retryOnError c "NtfWorker" loop (workerInternalError c connId . show)
     noWorkToDo = void . atomically $ tryTakeTMVar doWork
     processAction :: (NtfSubscription, NtfSubNTFAction, NtfActionTs) -> m ()
     processAction (sub@NtfSubscription {connId, smpServer, ntfSubId}, action, actionTs) = do
@@ -213,7 +213,7 @@ runNtfWorker c srv doWork = do
           NSADelete -> case ntfSubId of
             Just nSubId ->
               (getNtfToken >>= mapM_ (agentNtfDeleteSubscription c nSubId))
-                `E.finally` continueDeletion
+                `agentFinally` continueDeletion
             _ -> continueDeletion
             where
               continueDeletion = do
@@ -224,7 +224,7 @@ runNtfWorker c srv doWork = do
           NSARotate -> case ntfSubId of
             Just nSubId ->
               (getNtfToken >>= mapM_ (agentNtfDeleteSubscription c nSubId))
-                `E.finally` deleteCreate
+                `agentFinally` deleteCreate
             _ -> deleteCreate
             where
               deleteCreate = do
@@ -257,7 +257,7 @@ runNtfSMPWorker c srv doWork = do
           ri <- asks $ reconnectInterval . config
           withRetryInterval ri $ \_ loop ->
             processAction a
-              `catchError` retryOnError c "NtfSMPWorker" loop (workerInternalError c connId . show)
+              `catchAgentError` retryOnError c "NtfSMPWorker" loop (workerInternalError c connId . show)
     noWorkToDo = void . atomically $ tryTakeTMVar doWork
     processAction :: (NtfSubscription, NtfSubSMPAction, NtfActionTs) -> m ()
     processAction (sub@NtfSubscription {connId, ntfServer}, smpAction, actionTs) = do

src/Simplex/Messaging/Agent/Protocol.hs

@@ -55,15 +55,24 @@ module Simplex.Messaging.Agent.Protocol
     AEntityI (..),
     MsgHash,
     MsgMeta (..),
+    RcvQueueInfo (..),
+    SndQueueInfo (..),
     ConnectionStats (..),
     SwitchPhase (..),
+    RcvSwitchStatus (..),
+    SndSwitchStatus (..),
     QueueDirection (..),
+    RatchetSyncState (..),
     SMPConfirmation (..),
     AgentMsgEnvelope (..),
     AgentMessage (..),
     AgentMessageType (..),
     APrivHeader (..),
     AMessage (..),
+    AMessageReceipt (..),
+    MsgReceipt (..),
+    MsgReceiptInfo,
+    MsgReceiptStatus (..),
     SndQAddr,
     SMPServer,
     pattern SMPServer,
@@ -94,6 +103,7 @@ module Simplex.Messaging.Agent.Protocol
     BrokerErrorType (..),
     SMPAgentError (..),
     AgentCryptoError (..),
+    cryptoErrToSyncState,
     ATransmission,
     ATransmissionOrError,
     ARawTransmission,
@@ -155,7 +165,7 @@ import qualified Data.Map as M
 import Data.Maybe (isJust)
 import Data.Text (Text)
 import qualified Data.Text as T
-import Data.Text.Encoding (encodeUtf8)
+import Data.Text.Encoding (decodeLatin1, encodeUtf8)
 import Data.Time.Clock (UTCTime)
 import Data.Time.Clock.System (SystemTime)
 import Data.Time.ISO8601
@@ -205,7 +215,7 @@ import Text.Read
 import UnliftIO.Exception (Exception)
 
 currentSMPAgentVersion :: Version
-currentSMPAgentVersion = 2
+currentSMPAgentVersion = 4
 
 supportedSMPAgentVRange :: VersionRange
 supportedSMPAgentVRange = mkVersionRange 1 currentSMPAgentVersion
@@ -308,7 +318,7 @@ data ACommand (p :: AParty) (e :: AEntity) where
   JOIN :: Bool -> AConnectionRequestUri -> ConnInfo -> ACommand Client AEConn -- response OK
   CONF :: ConfirmationId -> [SMPServer] -> ConnInfo -> ACommand Agent AEConn -- ConnInfo is from sender, [SMPServer] will be empty only in v1 handshake
   LET :: ConfirmationId -> ConnInfo -> ACommand Client AEConn -- ConnInfo is from client
-  REQ :: InvitationId -> L.NonEmpty SMPServer -> ConnInfo -> ACommand Agent AEConn -- ConnInfo is from sender
+  REQ :: InvitationId -> NonEmpty SMPServer -> ConnInfo -> ACommand Agent AEConn -- ConnInfo is from sender
   ACPT :: InvitationId -> ConnInfo -> ACommand Client AEConn -- ConnInfo is from client
   RJCT :: InvitationId -> ACommand Client AEConn
   INFO :: ConnInfo -> ACommand Agent AEConn
@@ -320,12 +330,14 @@ data ACommand (p :: AParty) (e :: AEntity) where
   DOWN :: SMPServer -> [ConnId] -> ACommand Agent AENone
   UP :: SMPServer -> [ConnId] -> ACommand Agent AENone
   SWITCH :: QueueDirection -> SwitchPhase -> ConnectionStats -> ACommand Agent AEConn
+  RSYNC :: RatchetSyncState -> Maybe AgentCryptoError -> ConnectionStats -> ACommand Agent AEConn
   SEND :: MsgFlags -> MsgBody -> ACommand Client AEConn
   MID :: AgentMsgId -> ACommand Agent AEConn
   SENT :: AgentMsgId -> ACommand Agent AEConn
   MERR :: AgentMsgId -> AgentErrorType -> ACommand Agent AEConn
   MSG :: MsgMeta -> MsgFlags -> MsgBody -> ACommand Agent AEConn
-  ACK :: AgentMsgId -> ACommand Client AEConn
+  ACK :: AgentMsgId -> Maybe MsgReceiptInfo -> ACommand Client AEConn
+  RCVD :: MsgMeta -> NonEmpty MsgReceipt -> ACommand Agent AEConn 
   SWCH :: ACommand Client AEConn
   OFF :: ACommand Client AEConn
   DEL :: ACommand Client AEConn
@@ -378,12 +390,14 @@ data ACommandTag (p :: AParty) (e :: AEntity) where
   DOWN_ :: ACommandTag Agent AENone
   UP_ :: ACommandTag Agent AENone
   SWITCH_ :: ACommandTag Agent AEConn
+  RSYNC_ :: ACommandTag Agent AEConn
   SEND_ :: ACommandTag Client AEConn
   MID_ :: ACommandTag Agent AEConn
   SENT_ :: ACommandTag Agent AEConn
   MERR_ :: ACommandTag Agent AEConn
   MSG_ :: ACommandTag Agent AEConn
   ACK_ :: ACommandTag Client AEConn
+  RCVD_ :: ACommandTag Agent AEConn
   SWCH_ :: ACommandTag Client AEConn
   OFF_ :: ACommandTag Client AEConn
   DEL_ :: ACommandTag Client AEConn
@@ -429,12 +443,14 @@ aCommandTag = \case
   DOWN {} -> DOWN_
   UP {} -> UP_
   SWITCH {} -> SWITCH_
+  RSYNC {} -> RSYNC_
   SEND {} -> SEND_
   MID _ -> MID_
   SENT _ -> SENT_
   MERR {} -> MERR_
   MSG {} -> MSG_
-  ACK _ -> ACK_
+  ACK {} -> ACK_
+  RCVD {} -> RCVD_
   SWCH -> SWCH_
   OFF -> OFF_
   DEL -> DEL_
@@ -473,18 +489,20 @@ instance ToJSON QueueDirection where
 instance FromJSON QueueDirection where
   parseJSON = strParseJSON "QueueDirection"
 
-data SwitchPhase = SPStarted | SPConfirmed | SPCompleted
+data SwitchPhase = SPStarted | SPConfirmed | SPSecured | SPCompleted
   deriving (Eq, Show)
 
 instance StrEncoding SwitchPhase where
   strEncode = \case
     SPStarted -> "started"
     SPConfirmed -> "confirmed"
+    SPSecured -> "secured"
     SPCompleted -> "completed"
   strP =
     A.takeTill (== ' ') >>= \case
       "started" -> pure SPStarted
       "confirmed" -> pure SPConfirmed
+      "secured" -> pure SPSecured
       "completed" -> pure SPCompleted
       _ -> fail "bad SwitchPhase"
 
@@ -495,19 +513,158 @@ instance ToJSON SwitchPhase where
 instance FromJSON SwitchPhase where
   parseJSON = strParseJSON "SwitchPhase"
 
+data RcvSwitchStatus
+  = RSSwitchStarted
+  | RSSendingQADD
+  | RSSendingQUSE
+  | RSReceivedMessage
+  deriving (Eq, Show)
+
+instance StrEncoding RcvSwitchStatus where
+  strEncode = \case
+    RSSwitchStarted -> "switch_started"
+    RSSendingQADD -> "sending_qadd"
+    RSSendingQUSE -> "sending_quse"
+    RSReceivedMessage -> "received_message"
+  strP =
+    A.takeTill (== ' ') >>= \case
+      "switch_started" -> pure RSSwitchStarted
+      "sending_qadd" -> pure RSSendingQADD
+      "sending_quse" -> pure RSSendingQUSE
+      "received_message" -> pure RSReceivedMessage
+      _ -> fail "bad RcvSwitchStatus"
+
+instance ToField RcvSwitchStatus where toField = toField . decodeLatin1 . strEncode
+
+instance FromField RcvSwitchStatus where fromField = fromTextField_ $ eitherToMaybe . strDecode . encodeUtf8
+
+instance ToJSON RcvSwitchStatus where
+  toEncoding = strToJEncoding
+  toJSON = strToJSON
+
+instance FromJSON RcvSwitchStatus where
+  parseJSON = strParseJSON "RcvSwitchStatus"
+
+data SndSwitchStatus
+  = SSSendingQKEY
+  | SSSendingQTEST
+  deriving (Eq, Show)
+
+instance StrEncoding SndSwitchStatus where
+  strEncode = \case
+    SSSendingQKEY -> "sending_qkey"
+    SSSendingQTEST -> "sending_qtest"
+  strP =
+    A.takeTill (== ' ') >>= \case
+      "sending_qkey" -> pure SSSendingQKEY
+      "sending_qtest" -> pure SSSendingQTEST
+      _ -> fail "bad SndSwitchStatus"
+
+instance ToField SndSwitchStatus where toField = toField . decodeLatin1 . strEncode
+
+instance FromField SndSwitchStatus where fromField = fromTextField_ $ eitherToMaybe . strDecode . encodeUtf8
+
+instance ToJSON SndSwitchStatus where
+  toEncoding = strToJEncoding
+  toJSON = strToJSON
+
+instance FromJSON SndSwitchStatus where
+  parseJSON = strParseJSON "SndSwitchStatus"
+
+data RatchetSyncState
+  = RSOk
+  | RSAllowed
+  | RSRequired
+  | RSStarted
+  | RSAgreed
+  deriving (Eq, Show)
+
+instance StrEncoding RatchetSyncState where
+  strEncode = \case
+    RSOk -> "ok"
+    RSAllowed -> "allowed"
+    RSRequired -> "required"
+    RSStarted -> "started"
+    RSAgreed -> "agreed"
+  strP =
+    A.takeTill (== ' ') >>= \case
+      "ok" -> pure RSOk
+      "allowed" -> pure RSAllowed
+      "required" -> pure RSRequired
+      "started" -> pure RSStarted
+      "agreed" -> pure RSAgreed
+      _ -> fail "bad RatchetSyncState"
+
+instance FromField RatchetSyncState where fromField = fromTextField_ $ eitherToMaybe . strDecode . encodeUtf8
+
+instance ToField RatchetSyncState where toField = toField . decodeLatin1 . strEncode
+
+instance ToJSON RatchetSyncState where
+  toEncoding = strToJEncoding
+  toJSON = strToJSON
+
+instance FromJSON RatchetSyncState where
+  parseJSON = strParseJSON "RatchetSyncState"
+
+data RcvQueueInfo = RcvQueueInfo
+  { rcvServer :: SMPServer,
+    rcvSwitchStatus :: Maybe RcvSwitchStatus,
+    canAbortSwitch :: Bool
+  }
+  deriving (Eq, Show, Generic)
+
+instance ToJSON RcvQueueInfo where toEncoding = J.genericToEncoding J.defaultOptions {J.omitNothingFields = True}
+
+instance StrEncoding RcvQueueInfo where
+  strEncode RcvQueueInfo {rcvServer, rcvSwitchStatus, canAbortSwitch} =
+    "srv=" <> strEncode rcvServer
+      <> maybe "" (\switch -> ";switch=" <> strEncode switch) rcvSwitchStatus
+      <> (";can_abort_switch=" <> strEncode canAbortSwitch)
+  strP = do
+    rcvServer <- "srv=" *> strP
+    rcvSwitchStatus <- optional $ ";switch=" *> strP
+    canAbortSwitch <- ";can_abort_switch=" *> strP
+    pure RcvQueueInfo {rcvServer, rcvSwitchStatus, canAbortSwitch}
+
+data SndQueueInfo = SndQueueInfo
+  { sndServer :: SMPServer,
+    sndSwitchStatus :: Maybe SndSwitchStatus
+  }
+  deriving (Eq, Show, Generic)
+
+instance ToJSON SndQueueInfo where toEncoding = J.genericToEncoding J.defaultOptions {J.omitNothingFields = True}
+
+instance StrEncoding SndQueueInfo where
+  strEncode SndQueueInfo {sndServer, sndSwitchStatus} =
+    "srv=" <> strEncode sndServer <> maybe "" (\switch -> ";switch=" <> strEncode switch) sndSwitchStatus
+  strP = do
+    sndServer <- "srv=" *> strP
+    sndSwitchStatus <- optional $ ";switch=" *> strP
+    pure SndQueueInfo {sndServer, sndSwitchStatus}
+
 data ConnectionStats = ConnectionStats
-  { rcvServers :: [SMPServer],
-    sndServers :: [SMPServer]
+  { connAgentVersion :: Version,
+    rcvQueuesInfo :: [RcvQueueInfo],
+    sndQueuesInfo :: [SndQueueInfo],
+    ratchetSyncState :: RatchetSyncState,
+    ratchetSyncSupported :: Bool
   }
   deriving (Eq, Show, Generic)
 
 instance StrEncoding ConnectionStats where
-  strEncode ConnectionStats {rcvServers, sndServers} =
-    "rcv=" <> strEncodeList rcvServers <> " snd=" <> strEncodeList sndServers
+  strEncode ConnectionStats {connAgentVersion, rcvQueuesInfo, sndQueuesInfo, ratchetSyncState, ratchetSyncSupported} =
+    "agent_version=" <> strEncode connAgentVersion
+      <> (" rcv=" <> strEncodeList rcvQueuesInfo)
+      <> (" snd=" <> strEncodeList sndQueuesInfo)
+      <> (" sync=" <> strEncode ratchetSyncState)
+      <> (" sync_supported=" <> strEncode ratchetSyncSupported)
   strP = do
-    rcvServers <- "rcv=" *> strListP
-    sndServers <- " snd=" *> strListP
-    pure ConnectionStats {rcvServers, sndServers}
+    connAgentVersion <- "agent_version=" *> strP
+    rcvQueuesInfo <- " rcv=" *> strListP
+    sndQueuesInfo <- " snd=" *> strListP
+    ratchetSyncState <- " sync=" *> strP
+    ratchetSyncSupported <- " sync_supported=" *> strP
+    pure ConnectionStats {connAgentVersion, rcvQueuesInfo, sndQueuesInfo, ratchetSyncState, ratchetSyncSupported}
 
 instance ToJSON ConnectionStats where toEncoding = J.genericToEncoding J.defaultOptions
 
@@ -593,6 +750,25 @@ data MsgMeta = MsgMeta
   }
   deriving (Eq, Show)
 
+instance StrEncoding MsgMeta where
+  strEncode MsgMeta {integrity, recipient = (rmId, rTs), broker = (bmId, bTs), sndMsgId} =
+    B.unwords
+      [ strEncode integrity,
+        "R=" <> bshow rmId <> "," <> showTs rTs,
+        "B=" <> encode bmId <> "," <> showTs bTs,
+        "S=" <> bshow sndMsgId
+      ]
+    where
+      showTs = B.pack . formatISO8601Millis
+  strP = do
+    integrity <- strP
+    recipient <- " R=" *> partyMeta A.decimal
+    broker <- " B=" *> partyMeta base64P
+    sndMsgId <- " S=" *> A.decimal
+    pure MsgMeta {integrity, recipient, broker, sndMsgId}
+    where
+      partyMeta idParser = (,) <$> idParser <* A.char ',' <*> tsISO8601P
+
 data SMPConfirmation = SMPConfirmation
   { -- | sender's public key to use for authentication of sender's commands at the recepient's server
     senderKey :: SndPublicVerifyKey,
@@ -610,7 +786,7 @@ data SMPConfirmation = SMPConfirmation
 data AgentMsgEnvelope
   = AgentConfirmation
       { agentVersion :: Version,
-        e2eEncryption :: Maybe (E2ERatchetParams 'C.X448),
+        e2eEncryption_ :: Maybe (E2ERatchetParams 'C.X448),
         encConnInfo :: ByteString
       }
   | AgentMsgEnvelope
@@ -622,22 +798,29 @@ data AgentMsgEnvelope
         connReq :: ConnectionRequestUri 'CMInvitation,
         connInfo :: ByteString -- this message is only encrypted with per-queue E2E, not with double ratchet,
       }
+  | AgentRatchetKey
+      { agentVersion :: Version,
+        e2eEncryption :: E2ERatchetParams 'C.X448,
+        info :: ByteString
+      }
   deriving (Show)
 
 instance Encoding AgentMsgEnvelope where
   smpEncode = \case
-    AgentConfirmation {agentVersion, e2eEncryption, encConnInfo} ->
-      smpEncode (agentVersion, 'C', e2eEncryption, Tail encConnInfo)
+    AgentConfirmation {agentVersion, e2eEncryption_, encConnInfo} ->
+      smpEncode (agentVersion, 'C', e2eEncryption_, Tail encConnInfo)
     AgentMsgEnvelope {agentVersion, encAgentMessage} ->
       smpEncode (agentVersion, 'M', Tail encAgentMessage)
     AgentInvitation {agentVersion, connReq, connInfo} ->
       smpEncode (agentVersion, 'I', Large $ strEncode connReq, Tail connInfo)
+    AgentRatchetKey {agentVersion, e2eEncryption, info} ->
+      smpEncode (agentVersion, 'R', e2eEncryption, Tail info)
   smpP = do
     agentVersion <- smpP
     smpP >>= \case
       'C' -> do
-        (e2eEncryption, Tail encConnInfo) <- smpP
-        pure AgentConfirmation {agentVersion, e2eEncryption, encConnInfo}
+        (e2eEncryption_, Tail encConnInfo) <- smpP
+        pure AgentConfirmation {agentVersion, e2eEncryption_, encConnInfo}
       'M' -> do
         Tail encAgentMessage <- smpP
         pure AgentMsgEnvelope {agentVersion, encAgentMessage}
@@ -645,15 +828,21 @@ instance Encoding AgentMsgEnvelope where
         connReq <- strDecode . unLarge <$?> smpP
         Tail connInfo <- smpP
         pure AgentInvitation {agentVersion, connReq, connInfo}
+      'R' -> do
+        e2eEncryption <- smpP
+        Tail info <- smpP
+        pure AgentRatchetKey {agentVersion, e2eEncryption, info}
       _ -> fail "bad AgentMsgEnvelope"
 
 -- SMP agent message formats (after double ratchet decryption,
 -- or in case of AgentInvitation - in plain text body)
+-- AgentRatchetInfo is not encrypted with double ratchet, but with per-queue E2E encryption
 data AgentMessage
   = AgentConnInfo ConnInfo
   | -- AgentConnInfoReply is only used in duplexHandshake mode (v2), allowing to include reply queue(s) in the initial confirmation.
     -- It makes REPLY message unnecessary.
-    AgentConnInfoReply (L.NonEmpty SMPQueueInfo) ConnInfo
+    AgentConnInfoReply (NonEmpty SMPQueueInfo) ConnInfo
+  | AgentRatchetInfo ByteString
   | AgentMessage APrivHeader AMessage
   deriving (Show)
 
@@ -661,46 +850,56 @@ instance Encoding AgentMessage where
   smpEncode = \case
     AgentConnInfo cInfo -> smpEncode ('I', Tail cInfo)
     AgentConnInfoReply smpQueues cInfo -> smpEncode ('D', smpQueues, Tail cInfo) -- 'D' stands for "duplex"
+    AgentRatchetInfo info -> smpEncode ('R', Tail info)
     AgentMessage hdr aMsg -> smpEncode ('M', hdr, aMsg)
   smpP =
     smpP >>= \case
       'I' -> AgentConnInfo . unTail <$> smpP
       'D' -> AgentConnInfoReply <$> smpP <*> (unTail <$> smpP)
+      'R' -> AgentRatchetInfo . unTail <$> smpP
       'M' -> AgentMessage <$> smpP <*> smpP
       _ -> fail "bad AgentMessage"
 
 data AgentMessageType
   = AM_CONN_INFO
   | AM_CONN_INFO_REPLY
+  | AM_RATCHET_INFO
   | AM_HELLO_
   | AM_REPLY_
   | AM_A_MSG_
+  | AM_A_RCVD_
   | AM_QCONT_
   | AM_QADD_
   | AM_QKEY_
   | AM_QUSE_
   | AM_QTEST_
+  | AM_EREADY_
   deriving (Eq, Show)
 
 instance Encoding AgentMessageType where
   smpEncode = \case
     AM_CONN_INFO -> "C"
     AM_CONN_INFO_REPLY -> "D"
+    AM_RATCHET_INFO -> "S"
     AM_HELLO_ -> "H"
     AM_REPLY_ -> "R"
     AM_A_MSG_ -> "M"
+    AM_A_RCVD_ -> "V"
     AM_QCONT_ -> "QC"
     AM_QADD_ -> "QA"
     AM_QKEY_ -> "QK"
     AM_QUSE_ -> "QU"
     AM_QTEST_ -> "QT"
+    AM_EREADY_ -> "E"
   smpP =
     A.anyChar >>= \case
       'C' -> pure AM_CONN_INFO
       'D' -> pure AM_CONN_INFO_REPLY
+      'S' -> pure AM_RATCHET_INFO
       'H' -> pure AM_HELLO_
       'R' -> pure AM_REPLY_
       'M' -> pure AM_A_MSG_
+      'V' -> pure AM_A_RCVD_
       'Q' ->
         A.anyChar >>= \case
           'C' -> pure AM_QCONT_
@@ -709,12 +908,14 @@ instance Encoding AgentMessageType where
           'U' -> pure AM_QUSE_
           'T' -> pure AM_QTEST_
           _ -> fail "bad AgentMessageType"
+      'E' -> pure AM_EREADY_
       _ -> fail "bad AgentMessageType"
 
 agentMessageType :: AgentMessage -> AgentMessageType
 agentMessageType = \case
   AgentConnInfo _ -> AM_CONN_INFO
   AgentConnInfoReply {} -> AM_CONN_INFO_REPLY
+  AgentRatchetInfo _ -> AM_RATCHET_INFO
   AgentMessage _ aMsg -> case aMsg of
     -- HELLO is used both in v1 and in v2, but differently.
     -- - in v1 (and, possibly, in v2 for simplex connections) can be sent multiple times,
@@ -724,11 +925,13 @@ agentMessageType = \case
     -- REPLY is only used in v1
     REPLY _ -> AM_REPLY_
     A_MSG _ -> AM_A_MSG_
+    A_RCVD {} -> AM_A_RCVD_
     QCONT _ -> AM_QCONT_
     QADD _ -> AM_QADD_
     QKEY _ -> AM_QKEY_
     QUSE _ -> AM_QUSE_
     QTEST _ -> AM_QTEST_
+    EREADY _ -> AM_EREADY_
 
 data APrivHeader = APrivHeader
   { -- | sequential ID assigned by the sending agent
@@ -747,11 +950,13 @@ data AMsgType
   = HELLO_
   | REPLY_
   | A_MSG_
+  | A_RCVD_
   | QCONT_
   | QADD_
   | QKEY_
   | QUSE_
   | QTEST_
+  | EREADY_
   deriving (Eq)
 
 instance Encoding AMsgType where
@@ -759,16 +964,19 @@ instance Encoding AMsgType where
     HELLO_ -> "H"
     REPLY_ -> "R"
     A_MSG_ -> "M"
+    A_RCVD_ -> "V"
     QCONT_ -> "QC"
     QADD_ -> "QA"
     QKEY_ -> "QK"
     QUSE_ -> "QU"
     QTEST_ -> "QT"
+    EREADY_ -> "E"
   smpP =
     A.anyChar >>= \case
       'H' -> pure HELLO_
       'R' -> pure REPLY_
       'M' -> pure A_MSG_
+      'V' -> pure A_RCVD_
       'Q' ->
         A.anyChar >>= \case
           'C' -> pure QCONT_
@@ -777,6 +985,7 @@ instance Encoding AMsgType where
           'U' -> pure QUSE_
           'T' -> pure QTEST_
           _ -> fail "bad AMsgType"
+      'E' -> pure EREADY_
       _ -> fail "bad AMsgType"
 
 -- | Messages sent between SMP agents once SMP queue is secured.
@@ -786,21 +995,60 @@ data AMessage
   = -- | the first message in the queue to validate it is secured
     HELLO
   | -- | reply queues information
-    REPLY (L.NonEmpty SMPQueueInfo)
+    REPLY (NonEmpty SMPQueueInfo)
   | -- | agent envelope for the client message
     A_MSG MsgBody
+  | -- | agent envelope for delivery receipt 
+    A_RCVD (NonEmpty AMessageReceipt)
   | -- | the message instructing the client to continue sending messages (after ERR QUOTA)
     QCONT SndQAddr
   | -- add queue to connection (sent by recipient), with optional address of the replaced queue
-    QADD (L.NonEmpty (SMPQueueUri, Maybe SndQAddr))
+    QADD (NonEmpty (SMPQueueUri, Maybe SndQAddr))
   | -- key to secure the added queues and agree e2e encryption key (sent by sender)
-    QKEY (L.NonEmpty (SMPQueueInfo, SndPublicVerifyKey))
+    QKEY (NonEmpty (SMPQueueInfo, SndPublicVerifyKey))
   | -- inform that the queues are ready to use (sent by recipient)
-    QUSE (L.NonEmpty (SndQAddr, Bool))
+    QUSE (NonEmpty (SndQAddr, Bool))
   | -- sent by the sender to test new queues and to complete switching
-    QTEST (L.NonEmpty SndQAddr)
+    QTEST (NonEmpty SndQAddr)
+  | -- ratchet re-synchronization is complete, with last decrypted sender message id (recipient's `last_external_snd_msg_id`)
+    EREADY AgentMsgId
   deriving (Show)
 
+-- | this type is used to send as part of the protocol between different clients
+-- TODO possibly, rename fields and types referring to external and internal IDs to make them different
+data AMessageReceipt = AMessageReceipt
+  { agentMsgId :: AgentMsgId, -- this is an external snd message ID referenced by the message recipient
+    msgHash :: MsgHash,
+    rcptInfo :: MsgReceiptInfo
+  }
+  deriving (Show)
+
+-- | this type is used as part of agent protocol to communicate with the user application
+data MsgReceipt = MsgReceipt
+  { agentMsgId :: AgentMsgId, -- this is an internal agent message ID of received message
+    msgRcptStatus :: MsgReceiptStatus
+  }
+  deriving (Eq, Show)
+
+data MsgReceiptStatus = MROk | MRBadMsgHash
+  deriving (Eq, Show)
+
+instance StrEncoding MsgReceiptStatus where
+  strEncode = \case
+    MROk -> "ok"
+    MRBadMsgHash -> "badMsgHash"
+  strP =
+    A.takeWhile1 (/= ' ') >>= \ case
+      "ok" -> pure MROk
+      "badMsgHash" -> pure MRBadMsgHash
+      _ -> fail "bad MsgReceiptStatus"
+
+instance ToJSON MsgReceiptStatus where
+  toJSON = strToJSON
+  toEncoding = strToJEncoding
+
+type MsgReceiptInfo = ByteString
+
 type SndQAddr = (SMPServer, SMP.SenderId)
 
 instance Encoding AMessage where
@@ -808,22 +1056,41 @@ instance Encoding AMessage where
     HELLO -> smpEncode HELLO_
     REPLY smpQueues -> smpEncode (REPLY_, smpQueues)
     A_MSG body -> smpEncode (A_MSG_, Tail body)
+    A_RCVD mrs -> smpEncode (A_RCVD_, mrs)
     QCONT addr -> smpEncode (QCONT_, addr)
     QADD qs -> smpEncode (QADD_, qs)
     QKEY qs -> smpEncode (QKEY_, qs)
     QUSE qs -> smpEncode (QUSE_, qs)
     QTEST qs -> smpEncode (QTEST_, qs)
+    EREADY lastDecryptedMsgId -> smpEncode (EREADY_, lastDecryptedMsgId)
   smpP =
     smpP
       >>= \case
         HELLO_ -> pure HELLO
         REPLY_ -> REPLY <$> smpP
         A_MSG_ -> A_MSG . unTail <$> smpP
+        A_RCVD_ -> A_RCVD <$> smpP
         QCONT_ -> QCONT <$> smpP
         QADD_ -> QADD <$> smpP
         QKEY_ -> QKEY <$> smpP
         QUSE_ -> QUSE <$> smpP
         QTEST_ -> QTEST <$> smpP
+        EREADY_ -> EREADY <$> smpP
+
+instance Encoding AMessageReceipt where
+  smpEncode AMessageReceipt {agentMsgId, msgHash, rcptInfo} =
+    smpEncode (agentMsgId, msgHash, Large rcptInfo)
+  smpP = do
+    (agentMsgId, msgHash, Large rcptInfo) <- smpP
+    pure AMessageReceipt {agentMsgId, msgHash, rcptInfo}
+
+instance StrEncoding MsgReceipt where
+  strEncode MsgReceipt {agentMsgId, msgRcptStatus} =
+    strEncode agentMsgId <> ":" <> strEncode msgRcptStatus
+  strP = do
+    agentMsgId <- strP <* A.char ':'
+    msgRcptStatus <- strP
+    pure MsgReceipt {agentMsgId, msgRcptStatus}
 
 instance forall m. ConnectionModeI m => StrEncoding (ConnectionRequestUri m) where
   strEncode = \case
@@ -1054,7 +1321,7 @@ deriving instance Show AConnectionRequestUri
 data ConnReqUriData = ConnReqUriData
   { crScheme :: ConnReqScheme,
     crAgentVRange :: VersionRange,
-    crSmpQueues :: L.NonEmpty SMPQueueUri,
+    crSmpQueues :: NonEmpty SMPQueueUri,
     crClientData :: Maybe CRClientData
   }
   deriving (Eq, Show)
@@ -1116,7 +1383,7 @@ instance StrEncoding MsgIntegrity where
   strP = "OK" $> MsgOk <|> "ERR " *> (MsgError <$> strP)
   strEncode = \case
     MsgOk -> "OK"
-    MsgError e -> "ERR" <> strEncode e
+    MsgError e -> "ERR " <> strEncode e
 
 instance ToJSON MsgIntegrity where
   toJSON = J.genericToJSON $ sumTypeJSON fstToLower
@@ -1136,7 +1403,7 @@ data MsgErrorType
 instance StrEncoding MsgErrorType where
   strP =
     "ID " *> (MsgBadId <$> A.decimal)
-      <|> "IDS " *> (MsgSkipped <$> A.decimal <* A.space <*> A.decimal)
+      <|> "NO_ID " *> (MsgSkipped <$> A.decimal <* A.space <*> A.decimal)
       <|> "HASH" $> MsgBadHash
       <|> "DUPLICATE" $> MsgDuplicate
   strEncode = \case
@@ -1171,6 +1438,8 @@ data AgentErrorType
     AGENT {agentErr :: SMPAgentError}
   | -- | agent implementation or dependency errors
     INTERNAL {internalErr :: String}
+  | -- | agent inactive
+    INACTIVE
   deriving (Eq, Generic, Show, Exception)
 
 instance ToJSON AgentErrorType where
@@ -1267,6 +1536,20 @@ instance ToJSON AgentCryptoError where
   toJSON = J.genericToJSON $ sumTypeJSON id
   toEncoding = J.genericToEncoding $ sumTypeJSON id
 
+instance StrEncoding AgentCryptoError where
+  strP =
+    "DECRYPT_AES" $> DECRYPT_AES
+      <|> "DECRYPT_CB" $> DECRYPT_CB
+      <|> "RATCHET_HEADER" $> RATCHET_HEADER
+      <|> "RATCHET_EARLIER " *> (RATCHET_EARLIER <$> strP)
+      <|> "RATCHET_SKIPPED " *> (RATCHET_SKIPPED <$> strP)
+  strEncode = \case
+    DECRYPT_AES -> "DECRYPT_AES"
+    DECRYPT_CB -> "DECRYPT_CB"
+    RATCHET_HEADER -> "RATCHET_HEADER"
+    RATCHET_EARLIER n -> "RATCHET_EARLIER " <> strEncode n
+    RATCHET_SKIPPED n -> "RATCHET_SKIPPED " <> strEncode n
+
 instance ToJSON SMPAgentError where
   toJSON = J.genericToJSON $ sumTypeJSON id
   toEncoding = J.genericToEncoding $ sumTypeJSON id
@@ -1285,6 +1568,7 @@ instance StrEncoding AgentErrorType where
       <|> "AGENT QUEUE " *> (AGENT . A_QUEUE <$> parseRead A.takeByteString)
       <|> "AGENT " *> (AGENT <$> parseRead1)
       <|> "INTERNAL " *> (INTERNAL <$> parseRead A.takeByteString)
+      <|> "INACTIVE" $> INACTIVE
     where
       textP = T.unpack . safeDecodeUtf8 <$> A.takeTill (== ' ')
   strEncode = \case
@@ -1300,6 +1584,7 @@ instance StrEncoding AgentErrorType where
     AGENT (A_QUEUE e) -> "AGENT QUEUE " <> bshow e
     AGENT e -> "AGENT " <> bshow e
     INTERNAL e -> "INTERNAL " <> bshow e
+    INACTIVE -> "INACTIVE"
     where
       text = encodeUtf8 . T.pack
 
@@ -1315,6 +1600,14 @@ instance Arbitrary SMPAgentError where arbitrary = genericArbitraryU
 
 instance Arbitrary AgentCryptoError where arbitrary = genericArbitraryU
 
+cryptoErrToSyncState :: AgentCryptoError -> RatchetSyncState
+cryptoErrToSyncState = \case
+  DECRYPT_AES -> RSAllowed
+  DECRYPT_CB -> RSAllowed
+  RATCHET_HEADER -> RSRequired
+  RATCHET_EARLIER _ -> RSAllowed
+  RATCHET_SKIPPED _ -> RSRequired
+
 -- | SMP agent command and response parser for commands passed via network (only parses binary length)
 networkCommandP :: Parser ACmd
 networkCommandP = commandP A.takeByteString
@@ -1344,12 +1637,14 @@ instance StrEncoding ACmdTag where
       "DOWN" -> nt DOWN_
       "UP" -> nt UP_
       "SWITCH" -> ct SWITCH_
+      "RSYNC" -> ct RSYNC_
       "SEND" -> t SEND_
       "MID" -> ct MID_
       "SENT" -> ct SENT_
       "MERR" -> ct MERR_
       "MSG" -> ct MSG_
       "ACK" -> t ACK_
+      "RCVD" -> ct RCVD_
       "SWCH" -> t SWCH_
       "OFF" -> t OFF_
       "DEL" -> t DEL_
@@ -1397,12 +1692,14 @@ instance (APartyI p, AEntityI e) => StrEncoding (ACommandTag p e) where
     DOWN_ -> "DOWN"
     UP_ -> "UP"
     SWITCH_ -> "SWITCH"
+    RSYNC_ -> "RSYNC"
     SEND_ -> "SEND"
     MID_ -> "MID"
     SENT_ -> "SENT"
     MERR_ -> "MERR"
     MSG_ -> "MSG"
     ACK_ -> "ACK"
+    RCVD_ -> "RCVD"
     SWCH_ -> "SWCH"
     OFF_ -> "OFF"
     DEL_ -> "DEL"
@@ -1446,7 +1743,7 @@ commandP binaryP =
           RJCT_ -> s (RJCT <$> A.takeByteString)
           SUB_ -> pure SUB
           SEND_ -> s (SEND <$> smpP <* A.space <*> binaryP)
-          ACK_ -> s (ACK <$> A.decimal)
+          ACK_ -> s (ACK <$> A.decimal <*> optional (A.space *> binaryP))
           SWCH_ -> pure SWCH
           OFF_ -> pure OFF
           DEL_ -> pure DEL
@@ -1464,10 +1761,12 @@ commandP binaryP =
           DOWN_ -> s (DOWN <$> strP_ <*> connections)
           UP_ -> s (UP <$> strP_ <*> connections)
           SWITCH_ -> s (SWITCH <$> strP_ <*> strP_ <*> strP)
+          RSYNC_ -> s (RSYNC <$> strP_ <*> strP <*> strP)
           MID_ -> s (MID <$> A.decimal)
           SENT_ -> s (SENT <$> A.decimal)
           MERR_ -> s (MERR <$> A.decimal <* A.space <*> strP)
-          MSG_ -> s (MSG <$> msgMetaP <* A.space <*> smpP <* A.space <*> binaryP)
+          MSG_ -> s (MSG <$> strP <* A.space <*> smpP <* A.space <*> binaryP)
+          RCVD_ -> s (RCVD <$> strP <* A.space <*> strP)
           DEL_RCVQ_ -> s (DEL_RCVQ <$> strP_ <*> strP_ <*> strP)
           DEL_CONN_ -> pure DEL_CONN
           DEL_USER_ -> s (DEL_USER <$> strP)
@@ -1492,13 +1791,6 @@ commandP binaryP =
        in case ds of
             [] -> Left "no sender file description"
             sd : rds -> SFDONE <$> strDecode (encodeUtf8 sd) <*> mapM (strDecode . encodeUtf8) rds
-    msgMetaP = do
-      integrity <- strP
-      recipient <- " R=" *> partyMeta A.decimal
-      broker <- " B=" *> partyMeta base64P
-      sndMsgId <- " S=" *> A.decimal
-      pure MsgMeta {integrity, recipient, broker, sndMsgId}
-    partyMeta idParser = (,) <$> idParser <* A.char ',' <*> tsISO8601P
 
 parseCommand :: ByteString -> Either AgentErrorType ACmd
 parseCommand = parse (commandP A.takeByteString) $ CMD SYNTAX
@@ -1522,12 +1814,14 @@ serializeCommand = \case
   DOWN srv conns -> B.unwords [s DOWN_, s srv, connections conns]
   UP srv conns -> B.unwords [s UP_, s srv, connections conns]
   SWITCH dir phase srvs -> s (SWITCH_, dir, phase, srvs)
+  RSYNC rrState cryptoErr cstats -> s (RSYNC_, rrState, cryptoErr, cstats)
   SEND msgFlags msgBody -> B.unwords [s SEND_, smpEncode msgFlags, serializeBinary msgBody]
   MID mId -> s (MID_, Str $ bshow mId)
   SENT mId -> s (SENT_, Str $ bshow mId)
   MERR mId e -> s (MERR_, Str $ bshow mId, e)
-  MSG msgMeta msgFlags msgBody -> B.unwords [s MSG_, serializeMsgMeta msgMeta, smpEncode msgFlags, serializeBinary msgBody]
-  ACK mId -> s (ACK_, Str $ bshow mId)
+  MSG msgMeta msgFlags msgBody -> B.unwords [s MSG_, s msgMeta, smpEncode msgFlags, serializeBinary msgBody]
+  ACK mId rcptInfo_ -> s (ACK_, Str $ bshow mId) <> maybe "" (B.cons ' ' . serializeBinary) rcptInfo_
+  RCVD msgMeta rcpts -> s (RCVD_, msgMeta, rcpts)
   SWCH -> s SWCH_
   OFF -> s OFF_
   DEL -> s DEL_
@@ -1549,19 +1843,9 @@ serializeCommand = \case
   where
     s :: StrEncoding a => a -> ByteString
     s = strEncode
-    showTs :: UTCTime -> ByteString
-    showTs = B.pack . formatISO8601Millis
     connections :: [ConnId] -> ByteString
     connections = B.intercalate "," . map strEncode
     sfDone sd rds = B.intercalate fdSeparator $ strEncode sd : map strEncode rds
-    serializeMsgMeta :: MsgMeta -> ByteString
-    serializeMsgMeta MsgMeta {integrity, recipient = (rmId, rTs), broker = (bmId, bTs), sndMsgId} =
-      B.unwords
-        [ strEncode integrity,
-          "R=" <> bshow rmId <> "," <> showTs rTs,
-          "B=" <> encode bmId <> "," <> showTs bTs,
-          "S=" <> bshow sndMsgId
-        ]
 
 serializeBinary :: ByteString -> ByteString
 serializeBinary body = bshow (B.length body) <> "\n" <> body

src/Simplex/Messaging/Agent/Server.hs

@@ -23,7 +23,7 @@ import Simplex.Messaging.Agent.Env.SQLite
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore)
 import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), simplexMQVersion)
-import Simplex.Messaging.Transport.Server (loadTLSServerParams, runTransportServer)
+import Simplex.Messaging.Transport.Server (loadTLSServerParams, runTransportServer, defaultTransportServerConfig)
 import Simplex.Messaging.Util (bshow)
 import UnliftIO.Async (race_)
 import qualified UnliftIO.Exception as E
@@ -48,7 +48,7 @@ runSMPAgentBlocking (ATransport t) cfg@AgentConfig {tcpPort, caCertificateFile,
     smpAgent _ = do
       -- tlsServerParams is not in Env to avoid breaking functional API w/t key and certificate generation
       tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
-      runTransportServer started tcpPort tlsServerParams True $ \(h :: c) -> do
+      runTransportServer started tcpPort tlsServerParams defaultTransportServerConfig $ \(h :: c) -> do
         liftIO . putLn h $ "Welcome to SMP agent v" <> B.pack simplexMQVersion
         c <- getAgentClient initServers
         logConnection c True

src/Simplex/Messaging/Agent/Store.hs

@@ -21,6 +21,7 @@ import Data.Kind (Type)
 import Data.List (find)
 import Data.List.NonEmpty (NonEmpty)
 import qualified Data.List.NonEmpty as L
+import Data.Maybe (isJust)
 import Data.Time (UTCTime)
 import Data.Type.Equality
 import Simplex.Messaging.Agent.Protocol
@@ -66,12 +67,13 @@ data RcvQueue = RcvQueue
     sndId :: SMP.SenderId,
     -- | queue status
     status :: QueueStatus,
-    -- | database queue ID (within connection), can be Nothing for old queues
+    -- | database queue ID (within connection)
     dbQueueId :: Int64,
     -- | True for a primary or a next primary queue of the connection (next if dbReplaceQueueId is set)
     primary :: Bool,
     -- | database queue ID to replace, Nothing if this queue is not replacing another, `Just Nothing` is used for replacing old queues
     dbReplaceQueueId :: Maybe Int64,
+    rcvSwchStatus :: Maybe RcvSwitchStatus,
     -- | SMP client version
     smpClientVersion :: Version,
     -- | credentials used in context of notifications
@@ -80,6 +82,22 @@ data RcvQueue = RcvQueue
   }
   deriving (Eq, Show)
 
+rcvQueueInfo :: RcvQueue -> RcvQueueInfo
+rcvQueueInfo rq@RcvQueue {server, rcvSwchStatus} =
+  RcvQueueInfo {rcvServer = server, rcvSwitchStatus = rcvSwchStatus, canAbortSwitch = canAbortRcvSwitch rq}
+
+canAbortRcvSwitch :: RcvQueue -> Bool
+canAbortRcvSwitch = maybe False canAbort . rcvSwchStatus
+  where
+    canAbort = \case
+      RSSwitchStarted -> True
+      RSSendingQADD -> True
+      -- if switch is in RSSendingQUSE, a race condition with sender deleting the original queue is possible
+      RSSendingQUSE -> False
+      -- if switch is in RSReceivedMessage status, aborting switch (deleting new queue)
+      -- will break the connection because the sender would have original queue deleted
+      RSReceivedMessage -> False
+
 data ClientNtfCreds = ClientNtfCreds
   { -- | key pair to be used by the notification server to sign transmissions
     ntfPublicKey :: NtfPublicVerifyKey,
@@ -107,17 +125,22 @@ data SndQueue = SndQueue
     e2eDhSecret :: C.DhSecretX25519,
     -- | queue status
     status :: QueueStatus,
-    -- | database queue ID (within connection), can be Nothing for old queues
+    -- | database queue ID (within connection)
     dbQueueId :: Int64,
     -- | True for a primary or a next primary queue of the connection (next if dbReplaceQueueId is set)
     primary :: Bool,
     -- | ID of the queue this one is replacing
     dbReplaceQueueId :: Maybe Int64,
+    sndSwchStatus :: Maybe SndSwitchStatus,
     -- | SMP client version
     smpClientVersion :: Version
   }
   deriving (Eq, Show)
 
+sndQueueInfo :: SndQueue -> SndQueueInfo
+sndQueueInfo SndQueue {server, sndSwchStatus} =
+  SndQueueInfo {sndServer = server, sndSwitchStatus = sndSwchStatus}
+
 instance SMPQueue RcvQueue where
   qServer RcvQueue {server} = server
   {-# INLINE qServer #-}
@@ -155,10 +178,20 @@ findRQ :: (SMPServer, SMP.SenderId) -> NonEmpty RcvQueue -> Maybe RcvQueue
 findRQ sAddr = find $ sameQAddress sAddr . sndAddress
 {-# INLINE findRQ #-}
 
+switchingRQ :: NonEmpty RcvQueue -> Maybe RcvQueue
+switchingRQ = find $ isJust . rcvSwchStatus
+{-# INLINE switchingRQ #-}
+
+updatedQs :: SMPQueueRec q => q -> NonEmpty q -> NonEmpty q
+updatedQs q = L.map $ \q' -> if dbQId q == dbQId q' then q else q'
+{-# INLINE updatedQs #-}
+
 class SMPQueue q => SMPQueueRec q where
   qUserId :: q -> UserId
   qConnId :: q -> ConnId
   queueId :: q -> QueueId
+  dbQId :: q -> Int64
+  dbReplaceQId :: q -> Maybe Int64
 
 instance SMPQueueRec RcvQueue where
   qUserId = userId
@@ -167,6 +200,10 @@ instance SMPQueueRec RcvQueue where
   {-# INLINE qConnId #-}
   queueId = rcvId
   {-# INLINE queueId #-}
+  dbQId = dbQueueId
+  {-# INLINE dbQId #-}
+  dbReplaceQId = dbReplaceQueueId
+  {-# INLINE dbReplaceQId #-}
 
 instance SMPQueueRec SndQueue where
   qUserId = userId
@@ -175,6 +212,10 @@ instance SMPQueueRec SndQueue where
   {-# INLINE qConnId #-}
   queueId = sndId
   {-# INLINE queueId #-}
+  dbQId = dbQueueId
+  {-# INLINE dbQId #-}
+  dbReplaceQId = dbReplaceQueueId
+  {-# INLINE dbReplaceQId #-}
 
 -- * Connection types
 
@@ -202,14 +243,22 @@ deriving instance Eq (Connection d)
 
 deriving instance Show (Connection d)
 
-connData :: Connection d -> ConnData
-connData = \case
+toConnData :: Connection d -> ConnData
+toConnData = \case
   NewConnection cData -> cData
   RcvConnection cData _ -> cData
   SndConnection cData _ -> cData
   DuplexConnection cData _ _ -> cData
   ContactConnection cData _ -> cData
 
+updateConnection :: ConnData -> Connection d -> Connection d
+updateConnection cData = \case
+  NewConnection _ -> NewConnection cData
+  RcvConnection _ rq -> RcvConnection cData rq
+  SndConnection _ sq -> SndConnection cData sq
+  DuplexConnection _ rqs sqs -> DuplexConnection cData rqs sqs
+  ContactConnection _ rq -> ContactConnection cData rq
+
 data SConnType :: ConnType -> Type where
   SCNew :: SConnType CNew
   SCRcv :: SConnType CRcv
@@ -252,10 +301,28 @@ data ConnData = ConnData
     connAgentVersion :: Version,
     enableNtfs :: Bool,
     duplexHandshake :: Maybe Bool, -- added in agent protocol v2
-    deleted :: Bool
+    lastExternalSndId :: PrevExternalSndId,
+    deleted :: Bool,
+    ratchetSyncState :: RatchetSyncState
   }
   deriving (Eq, Show)
 
+-- this function should be mirrored in the clients
+ratchetSyncAllowed :: ConnData -> Bool
+ratchetSyncAllowed cData@ConnData {ratchetSyncState} =
+  ratchetSyncSupported' cData && (ratchetSyncState `elem` ([RSAllowed, RSRequired] :: [RatchetSyncState]))
+
+ratchetSyncSupported' :: ConnData -> Bool
+ratchetSyncSupported' ConnData {connAgentVersion} = connAgentVersion >= 3
+
+messageRcptsSupported :: ConnData -> Bool
+messageRcptsSupported ConnData {connAgentVersion} = connAgentVersion >= 4
+
+-- this function should be mirrored in the clients
+ratchetSyncSendProhibited :: ConnData -> Bool
+ratchetSyncSendProhibited ConnData {ratchetSyncState} =
+  ratchetSyncState `elem` ([RSRequired, RSStarted, RSAgreed] :: [RatchetSyncState])
+
 data PendingCommand = PendingCommand
   { corrId :: ACorrId,
     userId :: UserId,
@@ -308,6 +375,7 @@ data InternalCommand
   | ICAllowSecure SMP.RecipientId SMP.SndPublicVerifyKey
   | ICDuplexSecure SMP.RecipientId SMP.SndPublicVerifyKey
   | ICDeleteConn
+  | ICDeleteRcvQueue SMP.RecipientId
   | ICQSecure SMP.RecipientId SMP.SndPublicVerifyKey
   | ICQDelete SMP.RecipientId
 
@@ -317,6 +385,7 @@ data InternalCommandTag
   | ICAllowSecure_
   | ICDuplexSecure_
   | ICDeleteConn_
+  | ICDeleteRcvQueue_
   | ICQSecure_
   | ICQDelete_
   deriving (Show)
@@ -328,6 +397,7 @@ instance StrEncoding InternalCommand where
     ICAllowSecure rId sndKey -> strEncode (ICAllowSecure_, rId, sndKey)
     ICDuplexSecure rId sndKey -> strEncode (ICDuplexSecure_, rId, sndKey)
     ICDeleteConn -> strEncode ICDeleteConn_
+    ICDeleteRcvQueue rId -> strEncode (ICDeleteRcvQueue_, rId)
     ICQSecure rId senderKey -> strEncode (ICQSecure_, rId, senderKey)
     ICQDelete rId -> strEncode (ICQDelete_, rId)
   strP =
@@ -337,6 +407,7 @@ instance StrEncoding InternalCommand where
       ICAllowSecure_ -> ICAllowSecure <$> _strP <*> _strP
       ICDuplexSecure_ -> ICDuplexSecure <$> _strP <*> _strP
       ICDeleteConn_ -> pure ICDeleteConn
+      ICDeleteRcvQueue_ -> ICDeleteRcvQueue <$> _strP
       ICQSecure_ -> ICQSecure <$> _strP <*> _strP
       ICQDelete_ -> ICQDelete <$> _strP
 
@@ -347,6 +418,7 @@ instance StrEncoding InternalCommandTag where
     ICAllowSecure_ -> "ALLOW_SECURE"
     ICDuplexSecure_ -> "DUPLEX_SECURE"
     ICDeleteConn_ -> "DELETE_CONN"
+    ICDeleteRcvQueue_ -> "DELETE_RCV_QUEUE"
     ICQSecure_ -> "QSECURE"
     ICQDelete_ -> "QDELETE"
   strP =
@@ -356,6 +428,7 @@ instance StrEncoding InternalCommandTag where
       "ALLOW_SECURE" -> pure ICAllowSecure_
       "DUPLEX_SECURE" -> pure ICDuplexSecure_
       "DELETE_CONN" -> pure ICDeleteConn_
+      "DELETE_RCV_QUEUE" -> pure ICDeleteRcvQueue_
       "QSECURE" -> pure ICQSecure_
       "QDELETE" -> pure ICQDelete_
       _ -> fail "bad InternalCommandTag"
@@ -372,6 +445,7 @@ internalCmdTag = \case
   ICAllowSecure {} -> ICAllowSecure_
   ICDuplexSecure {} -> ICDuplexSecure_
   ICDeleteConn -> ICDeleteConn_
+  ICDeleteRcvQueue {} -> ICDeleteRcvQueue_
   ICQSecure {} -> ICQSecure_
   ICQDelete _ -> ICQDelete_
 
@@ -435,7 +509,10 @@ data RcvMsgData = RcvMsgData
 data RcvMsg = RcvMsg
   { internalId :: InternalId,
     msgMeta :: MsgMeta,
+    msgType :: AgentMessageType,
     msgBody :: MsgBody,
+    internalHash :: MsgHash,
+    msgReceipt :: Maybe MsgReceipt, -- if this message is a delivery receipt
     userAck :: Bool
   }
 
@@ -450,6 +527,14 @@ data SndMsgData = SndMsgData
     prevMsgHash :: MsgHash
   }
 
+data SndMsg = SndMsg
+  { internalId :: InternalId,
+    internalSndId :: InternalSndId,
+    msgType :: AgentMessageType,
+    internalHash :: MsgHash,
+    msgReceipt :: Maybe MsgReceipt
+  }
+
 data PendingMsgData = PendingMsgData
   { msgId :: InternalId,
     msgType :: AgentMessageType,

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

@@ -52,16 +52,26 @@ module Simplex.Messaging.Agent.Store.SQLite
     getDeletedConns,
     getConnData,
     setConnDeleted,
+    setConnAgentVersion,
     getDeletedConnIds,
+    setConnRatchetSync,
+    addProcessedRatchetKeyHash,
+    checkRatchetKeyHashExists,
+    deleteRatchetKeyHashesExpired,
     getRcvConn,
+    getRcvQueueById,
+    getSndQueueById,
     deleteConn,
     upgradeRcvConnToDuplex,
     upgradeSndConnToDuplex,
     addConnRcvQueue,
     addConnSndQueue,
     setRcvQueueStatus,
+    setRcvSwitchStatus,
+    setRcvQueueDeleted,
     setRcvQueueConfirmedE2E,
     setSndQueueStatus,
+    setSndSwitchStatus,
     setRcvQueuePrimary,
     setSndQueuePrimary,
     deleteConnRcvQueue,
@@ -69,6 +79,7 @@ module Simplex.Messaging.Agent.Store.SQLite
     deleteConnSndQueue,
     getPrimaryRcvQueue,
     getRcvQueue,
+    getDeletedRcvQueue,
     setRcvQueueNtfCreds,
     -- Confirmations
     createConfirmation,
@@ -88,20 +99,29 @@ module Simplex.Messaging.Agent.Store.SQLite
     updateSndIds,
     createSndMsg,
     createSndMsgDelivery,
+    getSndMsgViaRcpt,
+    updateSndMsgRcpt,
     getPendingMsgData,
     updatePendingMsgRIState,
     getPendingMsgs,
     deletePendingMsgs,
     setMsgUserAck,
+    getRcvMsg,
     getLastMsg,
     checkRcvMsgHashExists,
     deleteMsg,
+    deleteDeliveredSndMsg,
     deleteSndMsgDelivery,
     deleteRcvMsgHashesExpired,
+    deleteSndMsgsExpired,
     -- Double ratchet persistence
     createRatchetX3dhKeys,
     getRatchetX3dhKeys,
+    createRatchetX3dhKeys',
+    getRatchetX3dhKeys',
+    setRatchetX3dhKeys,
     createRatchet,
+    deleteRatchet,
     getRatchet,
     getSkippedMsgKeys,
     updateRatchet,
@@ -191,7 +211,6 @@ module Simplex.Messaging.Agent.Store.SQLite
   )
 where
 
-import Control.Concurrent (threadDelay)
 import Control.Concurrent.STM (stateTVar)
 import Control.Monad.Except
 import Crypto.Random (ChaChaDRG, randomBytesGenerate)
@@ -202,11 +221,10 @@ import Data.Bifunctor (second)
 import Data.ByteString (ByteString)
 import qualified Data.ByteString.Base64.URL as U
 import Data.Char (toLower)
-import Data.Function (on)
 import Data.Functor (($>))
 import Data.IORef
 import Data.Int (Int64)
-import Data.List (foldl', groupBy, intercalate, sortBy)
+import Data.List (foldl', intercalate, sortBy)
 import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
 import qualified Data.Map.Strict as M
@@ -215,7 +233,7 @@ import Data.Ord (Down (..))
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1, encodeUtf8)
-import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, diffUTCTime, getCurrentTime)
+import Data.Time.Clock (NominalDiffTime, UTCTime, addUTCTime, getCurrentTime)
 import Data.Word (Word32)
 import Database.SQLite.Simple (FromRow, NamedParam (..), Only (..), Query (..), SQLError, ToRow, field, (:.) (..))
 import qualified Database.SQLite.Simple as DB
@@ -232,6 +250,7 @@ import Simplex.FileTransfer.Types
 import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.RetryInterval (RI2State (..))
 import Simplex.Messaging.Agent.Store
+import Simplex.Messaging.Agent.Store.SQLite.Common
 import Simplex.Messaging.Agent.Store.SQLite.Migrations (DownMigration (..), MTRError, Migration (..), MigrationsToRun (..), mtrErrorDescription)
 import qualified Simplex.Messaging.Agent.Store.SQLite.Migrations as Migrations
 import qualified Simplex.Messaging.Crypto as C
@@ -244,25 +263,18 @@ import Simplex.Messaging.Parsers (blobFieldParser, dropPrefix, fromTextField_, s
 import Simplex.Messaging.Protocol
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Transport.Client (TransportHost)
-import Simplex.Messaging.Util (bshow, diffToMilliseconds, eitherToMaybe, ($>>=), (<$$>))
+import Simplex.Messaging.Util (bshow, eitherToMaybe, groupOn, ($>>=), (<$$>))
 import Simplex.Messaging.Version
 import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
 import System.Exit (exitFailure)
 import System.FilePath (takeDirectory)
 import System.IO (hFlush, stdout)
-import UnliftIO.Exception (bracket, onException)
+import UnliftIO.Exception (onException)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
 -- * SQLite Store implementation
 
-data SQLiteStore = SQLiteStore
-  { dbFilePath :: FilePath,
-    dbEncrypted :: TVar Bool,
-    dbConnection :: TMVar DB.Connection,
-    dbNew :: Bool
-  }
-
 data MigrationError
   = MEUpgrade {upMigrations :: [UpMigration]}
   | MEDowngrade {downMigrations :: [String]}
@@ -319,35 +331,35 @@ createSQLiteStore dbFilePath dbKey migrations confirmMigrations = do
     Left e -> closeSQLiteStore st $> Left e
 
 migrateSchema :: SQLiteStore -> [Migration] -> MigrationConfirmation -> IO (Either MigrationError ())
-migrateSchema st migrations confirmMigrations = withConnection st $ \db -> do
-  Migrations.initialize db
-  Migrations.get db migrations >>= \case
+migrateSchema st migrations confirmMigrations = do
+  Migrations.initialize st
+  Migrations.get st migrations >>= \case
     Left e -> do
       when (confirmMigrations == MCConsole) $ confirmOrExit ("Database state error: " <> mtrErrorDescription e)
       pure . Left $ MigrationError e
     Right MTRNone -> pure $ Right ()
     Right ms@(MTRUp ums)
-      | dbNew st -> Migrations.run db ms $> Right ()
+      | dbNew st -> Migrations.run st ms $> Right ()
       | otherwise -> case confirmMigrations of
-        MCYesUp -> run db ms
-        MCYesUpDown -> run db ms
-        MCConsole -> confirm err >> run db ms
+        MCYesUp -> run ms
+        MCYesUpDown -> run ms
+        MCConsole -> confirm err >> run ms
         MCError -> pure $ Left err
       where
         err = MEUpgrade $ map upMigration ums -- "The app has a newer version than the database.\nConfirm to back up and upgrade using these migrations: " <> intercalate ", " (map name ums)
     Right ms@(MTRDown dms) -> case confirmMigrations of
-      MCYesUpDown -> run db ms
-      MCConsole -> confirm err >> run db ms
+      MCYesUpDown -> run ms
+      MCConsole -> confirm err >> run ms
       MCYesUp -> pure $ Left err
       MCError -> pure $ Left err
       where
         err = MEDowngrade $ map downName dms
   where
     confirm err = confirmOrExit $ migrationErrorDescription err
-    run db ms = do
+    run ms = do
       let f = dbFilePath st
       copyFile f (f <> ".bak")
-      Migrations.run db ms
+      Migrations.run st ms
       pure $ Right ()
 
 confirmOrExit :: String -> IO ()
@@ -361,9 +373,10 @@ confirmOrExit s = do
 connectSQLiteStore :: FilePath -> String -> IO SQLiteStore
 connectSQLiteStore dbFilePath dbKey = do
   dbNew <- not <$> doesFileExist dbFilePath
-  dbConnection <- newTMVarIO =<< connectDB dbFilePath dbKey
+  dbConn <- dbBusyLoop $ connectDB dbFilePath dbKey
+  dbConnVar <- newTMVarIO dbConn
   dbEncrypted <- newTVarIO . not $ null dbKey
-  pure SQLiteStore {dbFilePath, dbEncrypted, dbConnection, dbNew}
+  pure SQLiteStore {dbFilePath, dbEncrypted, dbConnection = dbConnVar, dbNew}
 
 connectDB :: FilePath -> String -> IO DB.Connection
 connectDB path key = do
@@ -375,13 +388,11 @@ connectDB path key = do
     prepare db = do
       let exec = SQLite3.exec $ DB.connectionHandle db
       unless (null key) . exec $ "PRAGMA key = " <> sqlString key <> ";"
-      exec . fromQuery $
-        [sql|
-          PRAGMA foreign_keys = ON;
-          -- PRAGMA trusted_schema = OFF;
-          PRAGMA secure_delete = ON;
-          PRAGMA auto_vacuum = FULL;
-        |]
+      exec "PRAGMA busy_timeout = 100;"
+      exec "PRAGMA foreign_keys = ON;"
+      -- exec "PRAGMA trusted_schema = OFF;"
+      exec "PRAGMA secure_delete = ON;"
+      exec "PRAGMA auto_vacuum = FULL;"
 
 closeSQLiteStore :: SQLiteStore -> IO ()
 closeSQLiteStore st = atomically (takeTMVar $ dbConnection st) >>= DB.close
@@ -424,37 +435,6 @@ handleSQLError err e
   | DB.sqlError e == DB.ErrorConstraint = err
   | otherwise = SEInternal $ bshow e
 
-withConnection :: SQLiteStore -> (DB.Connection -> IO a) -> IO a
-withConnection SQLiteStore {dbConnection} =
-  bracket
-    (atomically $ takeTMVar dbConnection)
-    (atomically . putTMVar dbConnection)
-
-withTransaction :: forall a. SQLiteStore -> (DB.Connection -> IO a) -> IO a
-withTransaction = withTransactionCtx Nothing
-
-withTransactionCtx :: forall a. Maybe String -> SQLiteStore -> (DB.Connection -> IO a) -> IO a
-withTransactionCtx ctx_ st action = withConnection st $ loop 500 3_000_000
-  where
-    loop :: Int -> Int -> DB.Connection -> IO a
-    loop t tLim db =
-      transactionWithCtx `E.catch` \(e :: SQLError) ->
-        if tLim > t && DB.sqlError e == DB.ErrorBusy
-          then do
-            threadDelay t
-            loop (t * 9 `div` 8) (tLim - t) db
-          else E.throwIO e
-      where
-        transactionWithCtx = case ctx_ of
-          Nothing -> DB.withImmediateTransaction db (action db)
-          Just ctx -> do
-            t1 <- getCurrentTime
-            r <- DB.withImmediateTransaction db (action db)
-            t2 <- getCurrentTime
-            putStrLn $ "withTransactionCtx start :: " <> show t1 <> " :: " <> ctx
-            putStrLn $ "withTransactionCtx end   :: " <> show t2 <> " :: " <> ctx <> " :: duration=" <> show (diffToMilliseconds $ diffUTCTime t2 t1)
-            pure r
-
 createUserRecord :: DB.Connection -> IO UserId
 createUserRecord db = do
   DB.execute_ db "INSERT INTO users DEFAULT VALUES"
@@ -561,7 +541,7 @@ getRcvConn :: DB.Connection -> SMPServer -> SMP.RecipientId -> IO (Either StoreE
 getRcvConn db ProtocolServer {host, port} rcvId = runExceptT $ do
   rq@RcvQueue {connId} <-
     ExceptT . firstRow toRcvQueue SEConnNotFound $
-      DB.query db (rcvQueueQuery <> " WHERE q.host = ? AND q.port = ? AND q.rcv_id = ?") (host, port, rcvId)
+      DB.query db (rcvQueueQuery <> " WHERE q.host = ? AND q.port = ? AND q.rcv_id = ? AND q.deleted = 0") (host, port, rcvId)
   (rq,) <$> ExceptT (getConn db connId)
 
 deleteConn :: DB.Connection -> ConnId -> IO ()
@@ -620,6 +600,29 @@ setRcvQueueStatus db RcvQueue {rcvId, server = ProtocolServer {host, port}} stat
     |]
     [":status" := status, ":host" := host, ":port" := port, ":rcv_id" := rcvId]
 
+setRcvSwitchStatus :: DB.Connection -> RcvQueue -> Maybe RcvSwitchStatus -> IO RcvQueue
+setRcvSwitchStatus db rq@RcvQueue {rcvId, server = ProtocolServer {host, port}} rcvSwchStatus = do
+  DB.execute
+    db
+    [sql|
+      UPDATE rcv_queues
+      SET switch_status = ?
+      WHERE host = ? AND port = ? AND rcv_id = ?
+    |]
+    (rcvSwchStatus, host, port, rcvId)
+  pure rq {rcvSwchStatus}
+
+setRcvQueueDeleted :: DB.Connection -> RcvQueue -> IO ()
+setRcvQueueDeleted db RcvQueue {rcvId, server = ProtocolServer {host, port}} = do
+  DB.execute
+    db
+    [sql|
+      UPDATE rcv_queues
+      SET deleted = 1
+      WHERE host = ? AND port = ? AND rcv_id = ?
+    |]
+    (host, port, rcvId)
+
 setRcvQueueConfirmedE2E :: DB.Connection -> RcvQueue -> C.DhSecretX25519 -> Version -> IO ()
 setRcvQueueConfirmedE2E db RcvQueue {rcvId, server = ProtocolServer {host, port}} e2eDhSecret smpClientVersion =
   DB.executeNamed
@@ -651,6 +654,18 @@ setSndQueueStatus db SndQueue {sndId, server = ProtocolServer {host, port}} stat
     |]
     [":status" := status, ":host" := host, ":port" := port, ":snd_id" := sndId]
 
+setSndSwitchStatus :: DB.Connection -> SndQueue -> Maybe SndSwitchStatus -> IO SndQueue
+setSndSwitchStatus db sq@SndQueue {sndId, server = ProtocolServer {host, port}} sndSwchStatus = do
+  DB.execute
+    db
+    [sql|
+      UPDATE snd_queues
+      SET switch_status = ?
+      WHERE host = ? AND port = ? AND snd_id = ?
+    |]
+    (sndSwchStatus, host, port, sndId)
+  pure sq {sndSwchStatus}
+
 setRcvQueuePrimary :: DB.Connection -> ConnId -> RcvQueue -> IO ()
 setRcvQueuePrimary db connId RcvQueue {dbQueueId} = do
   DB.execute db "UPDATE rcv_queues SET rcv_primary = ? WHERE conn_id = ?" (False, connId)
@@ -687,7 +702,12 @@ getPrimaryRcvQueue db connId =
 getRcvQueue :: DB.Connection -> ConnId -> SMPServer -> SMP.RecipientId -> IO (Either StoreError RcvQueue)
 getRcvQueue db connId (SMPServer host port _) rcvId =
   firstRow toRcvQueue SEConnNotFound $
-    DB.query db (rcvQueueQuery <> "WHERE q.conn_id = ? AND q.host = ? AND q.port = ? AND q.rcv_id = ?") (connId, host, port, rcvId)
+    DB.query db (rcvQueueQuery <> "WHERE q.conn_id = ? AND q.host = ? AND q.port = ? AND q.rcv_id = ? AND q.deleted = 0") (connId, host, port, rcvId)
+
+getDeletedRcvQueue :: DB.Connection -> ConnId -> SMPServer -> SMP.RecipientId -> IO (Either StoreError RcvQueue)
+getDeletedRcvQueue db connId (SMPServer host port _) rcvId =
+  firstRow toRcvQueue SEConnNotFound $
+    DB.query db (rcvQueueQuery <> "WHERE q.conn_id = ? AND q.host = ? AND q.port = ? AND q.rcv_id = ? AND q.deleted = 1") (connId, host, port, rcvId)
 
 setRcvQueueNtfCreds :: DB.Connection -> ConnId -> Maybe ClientNtfCreds -> IO ()
 setRcvQueueNtfCreds db connId clientNtfCreds =
@@ -878,6 +898,31 @@ createSndMsgDelivery :: DB.Connection -> ConnId -> SndQueue -> InternalId -> IO
 createSndMsgDelivery db connId SndQueue {dbQueueId} msgId =
   DB.execute db "INSERT INTO snd_message_deliveries (conn_id, snd_queue_id, internal_id) VALUES (?, ?, ?)" (connId, dbQueueId, msgId)
 
+getSndMsgViaRcpt :: DB.Connection -> ConnId -> InternalSndId -> IO (Either StoreError SndMsg)
+getSndMsgViaRcpt db connId sndMsgId =
+  firstRow toSndMsg SEMsgNotFound $
+    DB.query
+      db
+      [sql|
+        SELECT s.internal_id, m.msg_type, s.internal_hash, s.rcpt_internal_id, s.rcpt_status
+        FROM snd_messages s
+        JOIN messages m ON s.internal_id = m.internal_id
+        WHERE s.conn_id = ? AND s.internal_snd_id = ?
+      |]
+      (connId, sndMsgId)
+  where
+    toSndMsg :: (InternalId, AgentMessageType, MsgHash, Maybe AgentMsgId, Maybe MsgReceiptStatus) -> SndMsg
+    toSndMsg (internalId, msgType, internalHash, rcptInternalId_, rcptStatus_) =
+      let msgReceipt = MsgReceipt <$> rcptInternalId_ <*> rcptStatus_
+       in SndMsg {internalId, internalSndId = sndMsgId, msgType, internalHash, msgReceipt}
+
+updateSndMsgRcpt :: DB.Connection -> ConnId -> InternalSndId -> MsgReceipt -> IO ()
+updateSndMsgRcpt db connId sndMsgId MsgReceipt {agentMsgId, msgRcptStatus} =
+  DB.execute
+    db
+    "UPDATE snd_messages SET rcpt_internal_id = ?, rcpt_status = ? WHERE conn_id = ? AND internal_snd_id = ?"
+    (agentMsgId, msgRcptStatus, connId, sndMsgId)
+
 getPendingMsgData :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError (Maybe RcvQueue, PendingMsgData))
 getPendingMsgData db connId msgId = do
   rq_ <- L.head <$$> getRcvQueuesByConnId_ db connId
@@ -914,32 +959,51 @@ deletePendingMsgs db connId SndQueue {dbQueueId} =
 
 setMsgUserAck :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError (RcvQueue, SMP.MsgId))
 setMsgUserAck db connId agentMsgId = runExceptT $ do
-  liftIO $ DB.execute db "UPDATE rcv_messages SET user_ack = ? WHERE conn_id = ? AND internal_id = ?" (True, connId, agentMsgId)
   (dbRcvId, srvMsgId) <-
     ExceptT . firstRow id SEMsgNotFound $
       DB.query db "SELECT rcv_queue_id, broker_id FROM rcv_messages WHERE conn_id = ? AND internal_id = ?" (connId, agentMsgId)
-  rq <- ExceptT $ getRcvQueueById_ db connId dbRcvId
+  rq <- ExceptT $ getRcvQueueById db connId dbRcvId
+  liftIO $ DB.execute db "UPDATE rcv_messages SET user_ack = ? WHERE conn_id = ? AND internal_id = ?" (True, connId, agentMsgId)
   pure (rq, srvMsgId)
 
-getLastMsg :: DB.Connection -> ConnId -> SMP.MsgId -> IO (Maybe RcvMsg)
-getLastMsg db connId msgId =
-  maybeFirstRow rcvMsg $
+getRcvMsg :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError RcvMsg)
+getRcvMsg db connId agentMsgId =
+  firstRow toRcvMsg SEMsgNotFound $
     DB.query
       db
       [sql|
         SELECT
-          r.internal_id, m.internal_ts, r.broker_id, r.broker_ts, r.external_snd_id, r.integrity,
-          m.msg_body, r.user_ack
+          r.internal_id, m.internal_ts, r.broker_id, r.broker_ts, r.external_snd_id, r.integrity, r.internal_hash,
+          m.msg_type, m.msg_body, s.internal_id, s.rcpt_status, r.user_ack
+        FROM rcv_messages r
+        JOIN messages m ON r.internal_id = m.internal_id
+        LEFT JOIN snd_messages s ON s.rcpt_internal_id = r.internal_id
+        WHERE r.conn_id = ? AND r.internal_id = ?
+      |]
+      (connId, agentMsgId)
+
+getLastMsg :: DB.Connection -> ConnId -> SMP.MsgId -> IO (Maybe RcvMsg)
+getLastMsg db connId msgId =
+  maybeFirstRow toRcvMsg $
+    DB.query
+      db
+      [sql|
+        SELECT
+          r.internal_id, m.internal_ts, r.broker_id, r.broker_ts, r.external_snd_id, r.integrity, r.internal_hash,
+          m.msg_type, m.msg_body, s.internal_id, s.rcpt_status, r.user_ack
         FROM rcv_messages r
         JOIN messages m ON r.internal_id = m.internal_id
         JOIN connections c ON r.conn_id = c.conn_id AND c.last_internal_msg_id = r.internal_id
+        LEFT JOIN snd_messages s ON s.rcpt_internal_id = r.internal_id
         WHERE r.conn_id = ? AND r.broker_id = ?
       |]
       (connId, msgId)
-  where
-    rcvMsg (agentMsgId, internalTs, brokerId, brokerTs, sndMsgId, integrity, msgBody, userAck) =
-      let msgMeta = MsgMeta {recipient = (agentMsgId, internalTs), broker = (brokerId, brokerTs), sndMsgId, integrity}
-       in RcvMsg {internalId = InternalId agentMsgId, msgMeta, msgBody, userAck}
+
+toRcvMsg :: (Int64, InternalTs, BrokerId, BrokerTs, AgentMsgId, MsgIntegrity, MsgHash, AgentMessageType, MsgBody, Maybe AgentMsgId, Maybe MsgReceiptStatus, Bool) -> RcvMsg    
+toRcvMsg (agentMsgId, internalTs, brokerId, brokerTs, sndMsgId, integrity, internalHash, msgType, msgBody, rcptInternalId_, rcptStatus_, userAck) =
+  let msgMeta = MsgMeta {recipient = (agentMsgId, internalTs), broker = (brokerId, brokerTs), sndMsgId, integrity}
+      msgReceipt = MsgReceipt <$> rcptInternalId_ <*> rcptStatus_
+    in RcvMsg {internalId = InternalId agentMsgId, msgMeta, msgType, msgBody, internalHash, msgReceipt, userAck}
 
 checkRcvMsgHashExists :: DB.Connection -> ConnId -> ByteString -> IO Bool
 checkRcvMsgHashExists db connId hash = do
@@ -956,20 +1020,55 @@ deleteMsg :: DB.Connection -> ConnId -> InternalId -> IO ()
 deleteMsg db connId msgId =
   DB.execute db "DELETE FROM messages WHERE conn_id = ? AND internal_id = ?;" (connId, msgId)
 
-deleteSndMsgDelivery :: DB.Connection -> ConnId -> SndQueue -> InternalId -> IO ()
-deleteSndMsgDelivery db connId SndQueue {dbQueueId} msgId = do
+deleteMsgContent :: DB.Connection -> ConnId -> InternalId -> IO ()
+deleteMsgContent db connId msgId =
+  DB.execute db "UPDATE messages SET msg_body = x'' WHERE conn_id = ? AND internal_id = ?;" (connId, msgId)
+
+deleteDeliveredSndMsg :: DB.Connection -> ConnId -> InternalId -> IO ()
+deleteDeliveredSndMsg db connId msgId = do
+  cnt <- countPendingSndDeliveries_ db connId msgId
+  when (cnt == 0) $ deleteMsg db connId msgId
+
+deleteSndMsgDelivery :: DB.Connection -> ConnId -> SndQueue -> InternalId -> Bool -> IO ()
+deleteSndMsgDelivery db connId SndQueue {dbQueueId} msgId keepForReceipt = do
   DB.execute
     db
     "DELETE FROM snd_message_deliveries WHERE conn_id = ? AND snd_queue_id = ? AND internal_id = ?"
     (connId, dbQueueId, msgId)
-  (Only (cnt :: Int) : _) <- DB.query db "SELECT count(*) FROM snd_message_deliveries WHERE conn_id = ? AND internal_id = ?" (connId, msgId)
-  when (cnt == 0) $ deleteMsg db connId msgId
+  cnt <- countPendingSndDeliveries_ db connId msgId
+  when (cnt == 0) $ do
+    del <-
+      maybeFirstRow id (DB.query db "SELECT rcpt_internal_id, rcpt_status FROM snd_messages WHERE conn_id = ? AND internal_id = ?" (connId, msgId)) >>= \case
+        Just (Just (_ :: Int64), Just MROk) -> pure deleteMsg
+        _ -> pure $ if keepForReceipt then deleteMsgContent else deleteMsg
+    del db connId msgId
+
+countPendingSndDeliveries_ :: DB.Connection -> ConnId -> InternalId -> IO Int
+countPendingSndDeliveries_ db connId msgId = do
+  (Only cnt : _) <- DB.query db "SELECT count(*) FROM snd_message_deliveries WHERE conn_id = ? AND internal_id = ?" (connId, msgId)
+  pure cnt
 
 deleteRcvMsgHashesExpired :: DB.Connection -> NominalDiffTime -> IO ()
 deleteRcvMsgHashesExpired db ttl = do
   cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime
   DB.execute db "DELETE FROM encrypted_rcv_message_hashes WHERE created_at < ?" (Only cutoffTs)
 
+deleteSndMsgsExpired :: DB.Connection -> NominalDiffTime -> IO ()
+deleteSndMsgsExpired db ttl = do
+  cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime
+  DB.execute
+    db
+    [sql|
+      DELETE FROM messages
+      WHERE internal_id IN (
+        SELECT s.internal_id
+        FROM snd_messages s
+        JOIN messages m USING (internal_id)
+        WHERE m.internal_ts < ?
+      )
+    |]
+    (Only cutoffTs)
+
 createRatchetX3dhKeys :: DB.Connection -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> IO ()
 createRatchetX3dhKeys db connId x3dhPrivKey1 x3dhPrivKey2 =
   DB.execute db "INSERT INTO ratchets (conn_id, x3dh_priv_key_1, x3dh_priv_key_2) VALUES (?, ?, ?)" (connId, x3dhPrivKey1, x3dhPrivKey2)
@@ -984,6 +1083,35 @@ getRatchetX3dhKeys db connId =
       Right (Just k1, Just k2) -> Right (k1, k2)
       _ -> Left SEX3dhKeysNotFound
 
+createRatchetX3dhKeys' :: DB.Connection -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> C.PublicKeyX448 -> C.PublicKeyX448 -> IO ()
+createRatchetX3dhKeys' db connId x3dhPrivKey1 x3dhPrivKey2 x3dhPubKey1 x3dhPubKey2 =
+  DB.execute
+    db
+    "INSERT INTO ratchets (conn_id, x3dh_priv_key_1, x3dh_priv_key_2, x3dh_pub_key_1, x3dh_pub_key_2) VALUES (?,?,?,?,?)"
+    (connId, x3dhPrivKey1, x3dhPrivKey2, x3dhPubKey1, x3dhPubKey2)
+
+getRatchetX3dhKeys' :: DB.Connection -> ConnId -> IO (Either StoreError (C.PrivateKeyX448, C.PrivateKeyX448, C.PublicKeyX448, C.PublicKeyX448))
+getRatchetX3dhKeys' db connId =
+  fmap hasKeys $
+    firstRow id SEX3dhKeysNotFound $
+      DB.query db "SELECT x3dh_priv_key_1, x3dh_priv_key_2, x3dh_pub_key_1, x3dh_pub_key_2 FROM ratchets WHERE conn_id = ?" (Only connId)
+  where
+    hasKeys = \case
+      Right (Just pk1, Just pk2, Just k1, Just k2) -> Right (pk1, pk2, k1, k2)
+      _ -> Left SEX3dhKeysNotFound
+
+-- used to remember new keys when starting ratchet re-synchronization
+setRatchetX3dhKeys :: DB.Connection -> ConnId -> C.PrivateKeyX448 -> C.PrivateKeyX448 -> C.PublicKeyX448 -> C.PublicKeyX448 -> IO ()
+setRatchetX3dhKeys db connId x3dhPrivKey1 x3dhPrivKey2 x3dhPubKey1 x3dhPubKey2 =
+  DB.execute
+    db
+    [sql|
+      UPDATE ratchets
+      SET x3dh_priv_key_1 = ?, x3dh_priv_key_2 = ?, x3dh_pub_key_1 = ?, x3dh_pub_key_2 = ?
+      WHERE conn_id = ?
+    |]
+    (x3dhPrivKey1, x3dhPrivKey2, x3dhPubKey1, x3dhPubKey2, connId)
+
 createRatchet :: DB.Connection -> ConnId -> RatchetX448 -> IO ()
 createRatchet db connId rc =
   DB.executeNamed
@@ -994,10 +1122,16 @@ createRatchet db connId rc =
       ON CONFLICT (conn_id) DO UPDATE SET
         ratchet_state = :ratchet_state,
         x3dh_priv_key_1 = NULL,
-        x3dh_priv_key_2 = NULL
+        x3dh_priv_key_2 = NULL,
+        x3dh_pub_key_1 = NULL,
+        x3dh_pub_key_2 = NULL
     |]
     [":conn_id" := connId, ":ratchet_state" := rc]
 
+deleteRatchet :: DB.Connection -> ConnId -> IO ()
+deleteRatchet db connId =
+  DB.execute db "DELETE FROM ratchets WHERE conn_id = ?" (Only connId)
+
 getRatchet :: DB.Connection -> ConnId -> IO (Either StoreError RatchetX448)
 getRatchet db connId =
   firstRow' ratchet SERatchetNotFound $ DB.query db "SELECT ratchet_state FROM ratchets WHERE conn_id = ?" (Only connId)
@@ -1046,7 +1180,9 @@ insertedRowId db = fromOnly . head <$> DB.query_ db "SELECT last_insert_rowid()"
 
 getPendingCommands :: DB.Connection -> ConnId -> IO [(Maybe SMPServer, [AsyncCmdId])]
 getPendingCommands db connId = do
-  map (\ids -> (fst $ head ids, map snd ids)) . groupBy ((==) `on` fst) . map srvCmdId
+  -- `groupOn` is used instead of `groupAllOn` to avoid extra sorting by `server + cmdId`, as the query already sorts by them.
+  -- TODO review whether this can break if, e.g., the server has another key hash.
+  map (\ids -> (fst $ head ids, map snd ids)) . groupOn fst . map srvCmdId
     <$> DB.query
       db
       [sql|
@@ -1375,7 +1511,7 @@ getNtfRcvQueue db SMPQueueNtf {smpServer = (SMPServer host port _), notifierId}
       [sql|
         SELECT conn_id, rcv_ntf_dh_secret
         FROM rcv_queues
-        WHERE host = ? AND port = ? AND ntf_id = ?
+        WHERE host = ? AND port = ? AND ntf_id = ? AND deleted = 0
       |]
       (host, port, notifierId)
   where
@@ -1452,6 +1588,10 @@ instance ToField AgentCommandTag where toField = toField . strEncode
 
 instance FromField AgentCommandTag where fromField = blobFieldParser strP
 
+instance ToField MsgReceiptStatus where toField = toField . decodeLatin1 . strEncode
+
+instance FromField MsgReceiptStatus where fromField = fromTextField_ $ eitherToMaybe . strDecode . encodeUtf8
+
 listToEither :: e -> [a] -> Either e a
 listToEither _ (x : _) = Right x
 listToEither e _ = Left e
@@ -1596,22 +1736,61 @@ getAnyConns_ deleted' db connIds = forM connIds $ E.handle handleDBError . getAn
     handleDBError = pure . Left . SEInternal . bshow
 
 getConnData :: DB.Connection -> ConnId -> IO (Maybe (ConnData, ConnectionMode))
-getConnData dbConn connId' =
-  maybeFirstRow cData $ DB.query dbConn "SELECT user_id, conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake, deleted FROM connections WHERE conn_id = ?;" (Only connId')
+getConnData db connId' =
+  maybeFirstRow cData $
+    DB.query
+      db
+      [sql|
+        SELECT
+          user_id, conn_id, conn_mode, smp_agent_version, enable_ntfs, duplex_handshake,
+          last_external_snd_msg_id, deleted, ratchet_sync_state
+        FROM connections
+        WHERE conn_id = ?
+      |]
+      (Only connId')
   where
-    cData (userId, connId, cMode, connAgentVersion, enableNtfs_, duplexHandshake, deleted) = (ConnData {userId, connId, connAgentVersion, enableNtfs = fromMaybe True enableNtfs_, duplexHandshake, deleted}, cMode)
+    cData (userId, connId, cMode, connAgentVersion, enableNtfs_, duplexHandshake, lastExternalSndId, deleted, ratchetSyncState) =
+      (ConnData {userId, connId, connAgentVersion, enableNtfs = fromMaybe True enableNtfs_, duplexHandshake, lastExternalSndId, deleted, ratchetSyncState}, cMode)
 
 setConnDeleted :: DB.Connection -> ConnId -> IO ()
 setConnDeleted db connId = DB.execute db "UPDATE connections SET deleted = ? WHERE conn_id = ?" (True, connId)
 
+setConnAgentVersion :: DB.Connection -> ConnId -> Version -> IO ()
+setConnAgentVersion db connId aVersion =
+  DB.execute db "UPDATE connections SET smp_agent_version = ? WHERE conn_id = ?" (aVersion, connId)
+
 getDeletedConnIds :: DB.Connection -> IO [ConnId]
 getDeletedConnIds db = map fromOnly <$> DB.query db "SELECT conn_id FROM connections WHERE deleted = ?" (Only True)
 
+setConnRatchetSync :: DB.Connection -> ConnId -> RatchetSyncState -> IO ()
+setConnRatchetSync db connId ratchetSyncState =
+  DB.execute db "UPDATE connections SET ratchet_sync_state = ? WHERE conn_id = ?" (ratchetSyncState, connId)
+
+addProcessedRatchetKeyHash :: DB.Connection -> ConnId -> ByteString -> IO ()
+addProcessedRatchetKeyHash db connId hash =
+  DB.execute db "INSERT INTO processed_ratchet_key_hashes (conn_id, hash) VALUES (?,?)" (connId, hash)
+
+checkRatchetKeyHashExists :: DB.Connection -> ConnId -> ByteString -> IO Bool
+checkRatchetKeyHashExists db connId hash = do
+  fromMaybe False
+    <$> maybeFirstRow
+      fromOnly
+      ( DB.query
+          db
+          "SELECT 1 FROM processed_ratchet_key_hashes WHERE conn_id = ? AND hash = ? LIMIT 1"
+          (connId, hash)
+      )
+
+deleteRatchetKeyHashesExpired :: DB.Connection -> NominalDiffTime -> IO ()
+deleteRatchetKeyHashesExpired db ttl = do
+  cutoffTs <- addUTCTime (- ttl) <$> getCurrentTime
+  DB.execute db "DELETE FROM processed_ratchet_key_hashes WHERE created_at < ?" (Only cutoffTs)
+
 -- | returns all connection queues, the first queue is the primary one
 getRcvQueuesByConnId_ :: DB.Connection -> ConnId -> IO (Maybe (NonEmpty RcvQueue))
 getRcvQueuesByConnId_ db connId =
   L.nonEmpty . sortBy primaryFirst . map toRcvQueue
-    <$> DB.query db (rcvQueueQuery <> "WHERE q.conn_id = ?") (Only connId)
+    <$> DB.query db (rcvQueueQuery <> "WHERE q.conn_id = ? AND q.deleted = 0") (Only connId)
   where
     primaryFirst RcvQueue {primary = p, dbReplaceQueueId = i} RcvQueue {primary = p', dbReplaceQueueId = i'} =
       -- the current primary queue is ordered first, the next primary - second
@@ -1622,7 +1801,7 @@ rcvQueueQuery =
   [sql|
     SELECT c.user_id, COALESCE(q.server_key_hash, s.key_hash), q.conn_id, q.host, q.port, q.rcv_id, q.rcv_private_key, q.rcv_dh_secret,
       q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.status,
-      q.rcv_queue_id, q.rcv_primary, q.replace_rcv_queue_id, q.smp_client_version, q.delete_errors,
+      q.rcv_queue_id, q.rcv_primary, q.replace_rcv_queue_id, q.switch_status, q.smp_client_version, q.delete_errors,
       q.ntf_public_key, q.ntf_private_key, q.ntf_id, q.rcv_ntf_dh_secret
     FROM rcv_queues q
     JOIN servers s ON q.host = s.host AND q.port = s.port
@@ -1631,44 +1810,62 @@ rcvQueueQuery =
 
 toRcvQueue ::
   (UserId, C.KeyHash, ConnId, NonEmpty TransportHost, ServiceName, SMP.RecipientId, SMP.RcvPrivateSignKey, SMP.RcvDhSecret, C.PrivateKeyX25519, Maybe C.DhSecretX25519, SMP.SenderId, QueueStatus)
-    :. (Int64, Bool, Maybe Int64, Maybe Version, Int)
+    :. (Int64, Bool, Maybe Int64, Maybe RcvSwitchStatus, Maybe Version, Int)
     :. (Maybe SMP.NtfPublicVerifyKey, Maybe SMP.NtfPrivateSignKey, Maybe SMP.NotifierId, Maybe RcvNtfDhSecret) ->
   RcvQueue
-toRcvQueue ((userId, keyHash, connId, host, port, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status) :. (dbQueueId, primary, dbReplaceQueueId, smpClientVersion_, deleteErrors) :. (ntfPublicKey_, ntfPrivateKey_, notifierId_, rcvNtfDhSecret_)) =
+toRcvQueue ((userId, keyHash, connId, host, port, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status) :. (dbQueueId, primary, dbReplaceQueueId, rcvSwchStatus, smpClientVersion_, deleteErrors) :. (ntfPublicKey_, ntfPrivateKey_, notifierId_, rcvNtfDhSecret_)) =
   let server = SMPServer host port keyHash
       smpClientVersion = fromMaybe 1 smpClientVersion_
       clientNtfCreds = case (ntfPublicKey_, ntfPrivateKey_, notifierId_, rcvNtfDhSecret_) of
         (Just ntfPublicKey, Just ntfPrivateKey, Just notifierId, Just rcvNtfDhSecret) -> Just $ ClientNtfCreds {ntfPublicKey, ntfPrivateKey, notifierId, rcvNtfDhSecret}
         _ -> Nothing
-   in RcvQueue {userId, connId, server, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status, dbQueueId, primary, dbReplaceQueueId, smpClientVersion, clientNtfCreds, deleteErrors}
+   in RcvQueue {userId, connId, server, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status, dbQueueId, primary, dbReplaceQueueId, rcvSwchStatus, smpClientVersion, clientNtfCreds, deleteErrors}
 
-getRcvQueueById_ :: DB.Connection -> ConnId -> Int64 -> IO (Either StoreError RcvQueue)
-getRcvQueueById_ db connId dbRcvId =
+getRcvQueueById :: DB.Connection -> ConnId -> Int64 -> IO (Either StoreError RcvQueue)
+getRcvQueueById db connId dbRcvId =
   firstRow toRcvQueue SEConnNotFound $
-    DB.query db (rcvQueueQuery <> " WHERE q.conn_id = ? AND q.rcv_queue_id = ?") (connId, dbRcvId)
+    DB.query db (rcvQueueQuery <> " WHERE q.conn_id = ? AND q.rcv_queue_id = ? AND q.deleted = 0") (connId, dbRcvId)
 
 -- | returns all connection queues, the first queue is the primary one
 getSndQueuesByConnId_ :: DB.Connection -> ConnId -> IO (Maybe (NonEmpty SndQueue))
 getSndQueuesByConnId_ dbConn connId =
-  L.nonEmpty . sortBy primaryFirst . map sndQueue
-    <$> DB.query
-      dbConn
-      [sql|
-        SELECT c.user_id, COALESCE(q.server_key_hash, s.key_hash), q.host, q.port, q.snd_id, q.snd_public_key, q.snd_private_key, q.e2e_pub_key, q.e2e_dh_secret, q.status, q.snd_queue_id, q.snd_primary, q.replace_snd_queue_id, q.smp_client_version
-        FROM snd_queues q
-        JOIN servers s ON q.host = s.host AND q.port = s.port
-        JOIN connections c ON q.conn_id = c.conn_id
-        WHERE q.conn_id = ?;
-      |]
-      (Only connId)
+  L.nonEmpty . sortBy primaryFirst . map toSndQueue
+    <$> DB.query dbConn (sndQueueQuery <> "WHERE q.conn_id = ?") (Only connId)
   where
-    sndQueue ((userId, keyHash, host, port, sndId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status) :. (dbQueueId, primary, dbReplaceQueueId, smpClientVersion)) =
-      let server = SMPServer host port keyHash
-       in SndQueue {userId, connId, server, sndId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status, dbQueueId, primary, dbReplaceQueueId, smpClientVersion}
     primaryFirst SndQueue {primary = p, dbReplaceQueueId = i} SndQueue {primary = p', dbReplaceQueueId = i'} =
       -- the current primary queue is ordered first, the next primary - second
       compare (Down p) (Down p') <> compare i i'
 
+sndQueueQuery :: Query
+sndQueueQuery =
+  [sql|
+    SELECT
+      c.user_id, COALESCE(q.server_key_hash, s.key_hash), q.conn_id, q.host, q.port, q.snd_id,
+      q.snd_public_key, q.snd_private_key, q.e2e_pub_key, q.e2e_dh_secret, q.status,
+      q.snd_queue_id, q.snd_primary, q.replace_snd_queue_id, q.switch_status, q.smp_client_version
+    FROM snd_queues q
+    JOIN servers s ON q.host = s.host AND q.port = s.port
+    JOIN connections c ON q.conn_id = c.conn_id
+  |]
+
+toSndQueue ::
+  (UserId, C.KeyHash, ConnId, NonEmpty TransportHost, ServiceName, SenderId)
+    :. (Maybe C.APublicVerifyKey, SndPrivateSignKey, Maybe C.PublicKeyX25519, C.DhSecretX25519, QueueStatus)
+    :. (Int64, Bool, Maybe Int64, Maybe SndSwitchStatus, Version) ->
+  SndQueue
+toSndQueue
+  ( (userId, keyHash, connId, host, port, sndId)
+      :. (sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status)
+      :. (dbQueueId, primary, dbReplaceQueueId, sndSwchStatus, smpClientVersion)
+    ) =
+    let server = SMPServer host port keyHash
+     in SndQueue {userId, connId, server, sndId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status, dbQueueId, primary, dbReplaceQueueId, sndSwchStatus, smpClientVersion}
+
+getSndQueueById :: DB.Connection -> ConnId -> Int64 -> IO (Either StoreError SndQueue)
+getSndQueueById db connId dbSndId =
+  firstRow toSndQueue SEConnNotFound $
+    DB.query db (sndQueueQuery <> " WHERE q.conn_id = ? AND q.snd_queue_id = ?") (connId, dbSndId)
+
 -- * updateRcvIds helpers
 
 retrieveLastIdsAndHashRcv_ :: DB.Connection -> ConnId -> IO (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash)

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

@@ -0,0 +1,61 @@
+{-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Common
+  ( SQLiteStore (..),
+    withConnection,
+    withTransaction,
+    withTransactionCtx,
+    dbBusyLoop,
+  )
+where
+
+import Control.Concurrent (threadDelay)
+import Data.Time.Clock (diffUTCTime, getCurrentTime)
+import Database.SQLite.Simple (SQLError)
+import qualified Database.SQLite.Simple as DB
+import Simplex.Messaging.Util (diffToMilliseconds)
+import UnliftIO.Exception (bracket)
+import qualified UnliftIO.Exception as E
+import UnliftIO.STM
+
+data SQLiteStore = SQLiteStore
+  { dbFilePath :: FilePath,
+    dbEncrypted :: TVar Bool,
+    dbConnection :: TMVar DB.Connection,
+    dbNew :: Bool
+  }
+
+withConnection :: SQLiteStore -> (DB.Connection -> IO a) -> IO a
+withConnection SQLiteStore {dbConnection} =
+  bracket
+    (atomically $ takeTMVar dbConnection)
+    (atomically . putTMVar dbConnection)
+
+withTransaction :: forall a. SQLiteStore -> (DB.Connection -> IO a) -> IO a
+withTransaction = withTransactionCtx Nothing
+
+withTransactionCtx :: forall a. Maybe String -> SQLiteStore -> (DB.Connection -> IO a) -> IO a
+withTransactionCtx ctx_ st action = withConnection st $ \db -> dbBusyLoop (transactionWithCtx db)
+  where
+    transactionWithCtx db = case ctx_ of
+      Nothing -> DB.withImmediateTransaction db (action db)
+      Just ctx -> do
+        t1 <- getCurrentTime
+        r <- DB.withImmediateTransaction db (action db)
+        t2 <- getCurrentTime
+        putStrLn $ "withTransactionCtx start :: " <> show t1 <> " :: " <> ctx
+        putStrLn $ "withTransactionCtx end   :: " <> show t2 <> " :: " <> ctx <> " :: duration=" <> show (diffToMilliseconds $ diffUTCTime t2 t1)
+        pure r
+
+dbBusyLoop :: forall a. IO a -> IO a
+dbBusyLoop action = loop 500 3000000
+  where
+    loop :: Int -> Int -> IO a
+    loop t tLim =
+      action `E.catch` \(e :: SQLError) ->
+        if tLim > t && DB.sqlError e == DB.ErrorBusy
+          then do
+            threadDelay t
+            loop (t * 9 `div` 8) (tLim - t)
+          else E.throwIO e

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

@@ -42,6 +42,7 @@ import Database.SQLite.Simple.QQ (sql)
 import qualified Database.SQLite3 as SQLite3
 import GHC.Generics (Generic)
 import Simplex.Messaging.Agent.Protocol (extraSMPServerHosts)
+import Simplex.Messaging.Agent.Store.SQLite.Common
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220101_initial
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220301_snd_queue_keys
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20220322_notifications
@@ -60,6 +61,9 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230320_retry_state
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230401_snd_files
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230510_files_pending_replicas_indexes
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230516_encrypted_rcv_message_hashes
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230531_switch_status
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230615_ratchet_sync
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230701_delivery_receipts
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (dropPrefix, sumTypeJSON)
 import Simplex.Messaging.Transport.Client (TransportHost)
@@ -86,7 +90,10 @@ schemaMigrations =
     ("m20230320_retry_state", m20230320_retry_state, Just down_m20230320_retry_state),
     ("m20230401_snd_files", m20230401_snd_files, Just down_m20230401_snd_files),
     ("m20230510_files_pending_replicas_indexes", m20230510_files_pending_replicas_indexes, Just down_m20230510_files_pending_replicas_indexes),
-    ("m20230516_encrypted_rcv_message_hashes", m20230516_encrypted_rcv_message_hashes, Just down_m20230516_encrypted_rcv_message_hashes)
+    ("m20230516_encrypted_rcv_message_hashes", m20230516_encrypted_rcv_message_hashes, Just down_m20230516_encrypted_rcv_message_hashes),
+    ("m20230531_switch_status", m20230531_switch_status, Just down_m20230531_switch_status),
+    ("m20230615_ratchet_sync", m20230615_ratchet_sync, Just down_m20230615_ratchet_sync),
+    ("m20230701_delivery_receipts", m20230701_delivery_receipts, Just down_m20230701_delivery_receipts)
   ]
 
 -- | The list of migrations in ascending order by date
@@ -95,44 +102,42 @@ app = sortOn name $ map migration schemaMigrations
   where
     migration (name, up, down) = Migration {name, up = fromQuery up, down = fromQuery <$> down}
 
-get :: Connection -> [Migration] -> IO (Either MTRError MigrationsToRun)
-get db migrations = migrationsToRun migrations <$> getCurrent db
+get :: SQLiteStore -> [Migration] -> IO (Either MTRError MigrationsToRun)
+get st migrations = migrationsToRun migrations <$> withTransaction st getCurrent
 
 getCurrent :: Connection -> IO [Migration]
 getCurrent db = map toMigration <$> DB.query_ db "SELECT name, down FROM migrations ORDER BY name ASC;"
   where
     toMigration (name, down) = Migration {name, up = "", down}
 
-run :: Connection -> MigrationsToRun -> IO ()
-run db = \case
+run :: SQLiteStore -> MigrationsToRun -> IO ()
+run st = \case
   MTRUp [] -> pure ()
-  MTRUp ms -> mapM_ runUp ms >> execSQL "VACUUM;"
+  MTRUp ms -> mapM_ runUp ms >> withConnection st (`execSQL` "VACUUM;")
   MTRDown ms -> mapM_ runDown $ reverse ms
   MTRNone -> pure ()
   where
-    runUp Migration {name, up, down} = do
-      when (name == "m20220811_onion_hosts") updateServers
-      DB.withImmediateTransaction db $ insert >> execSQL up
+    runUp Migration {name, up, down} = withTransaction st $ \db -> do
+      when (name == "m20220811_onion_hosts") $ updateServers db
+      insert db >> execSQL db up
       where
-        insert = DB.execute db "INSERT INTO migrations (name, down, ts) VALUES (?,?,?)" . (name,down,) =<< getCurrentTime
-        updateServers = forM_ (M.assocs extraSMPServerHosts) $ \(h, h') ->
-          DB.withImmediateTransaction db $
-            let hs = decodeLatin1 . strEncode $ ([h, h'] :: NonEmpty TransportHost)
-             in DB.execute db "UPDATE servers SET host = ? WHERE host = ?" (hs, decodeLatin1 $ strEncode h)
-    runDown DownMigration {downName, downQuery} = do
-      DB.withImmediateTransaction db $ do
-        execSQL downQuery
-        DB.execute db "DELETE FROM migrations WHERE name = ?" (Only downName)
-    execSQL = SQLite3.exec $ DB.connectionHandle db
+        insert db = DB.execute db "INSERT INTO migrations (name, down, ts) VALUES (?,?,?)" . (name,down,) =<< getCurrentTime
+        updateServers db = forM_ (M.assocs extraSMPServerHosts) $ \(h, h') ->
+          let hs = decodeLatin1 . strEncode $ ([h, h'] :: NonEmpty TransportHost)
+           in DB.execute db "UPDATE servers SET host = ? WHERE host = ?" (hs, decodeLatin1 $ strEncode h)
+    runDown DownMigration {downName, downQuery} = withTransaction st $ \db -> do
+      execSQL db downQuery
+      DB.execute db "DELETE FROM migrations WHERE name = ?" (Only downName)
+    execSQL db = SQLite3.exec $ DB.connectionHandle db
 
-initialize :: Connection -> IO ()
-initialize db = do
+initialize :: SQLiteStore -> IO ()
+initialize st = withTransaction st $ \db -> do
   cs :: [Text] <- map fromOnly <$> DB.query_ db "SELECT name FROM pragma_table_info('migrations')"
   case cs of
-    [] -> createMigrations
+    [] -> createMigrations db
     _ -> when ("down" `notElem` cs) $ DB.execute_ db "ALTER TABLE migrations ADD COLUMN down TEXT"
   where
-    createMigrations =
+    createMigrations db =
       DB.execute_
         db
         [sql|

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

@@ -0,0 +1,22 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230531_switch_status where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20230531_switch_status :: Query
+m20230531_switch_status =
+  [sql|
+ALTER TABLE rcv_queues ADD COLUMN switch_status TEXT;
+ALTER TABLE rcv_queues ADD COLUMN deleted INTEGER NOT NULL DEFAULT 0;
+ALTER TABLE snd_queues ADD COLUMN switch_status TEXT;
+|]
+
+down_m20230531_switch_status :: Query
+down_m20230531_switch_status =
+  [sql|
+ALTER TABLE snd_queues DROP COLUMN switch_status;
+ALTER TABLE rcv_queues DROP COLUMN deleted;
+ALTER TABLE rcv_queues DROP COLUMN switch_status;
+|]

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

@@ -0,0 +1,41 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230615_ratchet_sync where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+-- Ratchet public keys are saved when ratchet re-synchronization is started - upon receiving other party's public keys,
+-- keys are compared to determine ratchet initialization ordering for both parties.
+-- This solves a possible race when both parties start ratchet re-synchronization at the same time.
+m20230615_ratchet_sync :: Query
+m20230615_ratchet_sync =
+  [sql|
+ALTER TABLE connections ADD COLUMN ratchet_sync_state TEXT NOT NULL DEFAULT 'ok';
+
+ALTER TABLE ratchets ADD COLUMN x3dh_pub_key_1 BLOB;
+ALTER TABLE ratchets ADD COLUMN x3dh_pub_key_2 BLOB;
+
+CREATE TABLE processed_ratchet_key_hashes(
+  processed_ratchet_key_hash_id INTEGER PRIMARY KEY,
+  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
+  hash BLOB NOT NULL,
+  created_at TEXT NOT NULL DEFAULT (datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+CREATE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(conn_id, hash);
+|]
+
+down_m20230615_ratchet_sync :: Query
+down_m20230615_ratchet_sync =
+  [sql|
+DROP INDEX idx_processed_ratchet_key_hashes_hash;
+
+DROP TABLE processed_ratchet_key_hashes;
+
+ALTER TABLE ratchets DROP COLUMN x3dh_pub_key_2;
+ALTER TABLE ratchets DROP COLUMN x3dh_pub_key_1;
+
+ALTER TABLE connections DROP COLUMN ratchet_sync_state;
+|]

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

@@ -0,0 +1,24 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230701_delivery_receipts where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20230701_delivery_receipts :: Query
+m20230701_delivery_receipts =
+  [sql|
+ALTER TABLE snd_messages ADD COLUMN rcpt_internal_id INTEGER;
+ALTER TABLE snd_messages ADD COLUMN rcpt_status TEXT;
+
+CREATE INDEX idx_snd_messages_rcpt_internal_id ON snd_messages(conn_id, rcpt_internal_id);
+|]
+
+down_m20230701_delivery_receipts :: Query
+down_m20230701_delivery_receipts =
+  [sql|
+DROP INDEX idx_snd_messages_rcpt_internal_id;
+
+ALTER TABLE snd_messages DROP COLUMN rcpt_internal_id;
+ALTER TABLE snd_messages DROP COLUMN rcpt_status;
+|]

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

@@ -25,7 +25,8 @@ CREATE TABLE connections(
   enable_ntfs INTEGER,
   deleted INTEGER DEFAULT 0 CHECK(deleted NOT NULL),
   user_id INTEGER CHECK(user_id NOT NULL)
-  REFERENCES users ON DELETE CASCADE
+  REFERENCES users ON DELETE CASCADE,
+  ratchet_sync_state TEXT NOT NULL DEFAULT 'ok'
 ) WITHOUT ROWID;
 CREATE TABLE rcv_queues(
   host TEXT NOT NULL,
@@ -50,6 +51,8 @@ CREATE TABLE rcv_queues(
   replace_rcv_queue_id INTEGER NULL,
   delete_errors INTEGER DEFAULT 0 CHECK(delete_errors NOT NULL),
   server_key_hash BLOB,
+  switch_status TEXT,
+  deleted INTEGER NOT NULL DEFAULT 0,
   PRIMARY KEY(host, port, rcv_id),
   FOREIGN KEY(host, port) REFERENCES servers
   ON DELETE RESTRICT ON UPDATE CASCADE,
@@ -71,6 +74,7 @@ CREATE TABLE snd_queues(
   snd_primary INTEGER CHECK(snd_primary NOT NULL),
   replace_snd_queue_id INTEGER NULL,
   server_key_hash BLOB,
+  switch_status TEXT,
   PRIMARY KEY(host, port, snd_id),
   FOREIGN KEY(host, port) REFERENCES servers
   ON DELETE RESTRICT ON UPDATE CASCADE
@@ -115,6 +119,8 @@ CREATE TABLE snd_messages(
   previous_msg_hash BLOB NOT NULL DEFAULT x'',
   retry_int_slow INTEGER,
   retry_int_fast INTEGER,
+  rcpt_internal_id INTEGER,
+  rcpt_status TEXT,
   PRIMARY KEY(conn_id, internal_snd_id),
   FOREIGN KEY(conn_id, internal_id) REFERENCES messages
   ON DELETE CASCADE
@@ -151,6 +157,9 @@ CREATE TABLE ratchets(
   -- ratchet is initially empty on the receiving side(the side offering the connection)
   ratchet_state BLOB,
   e2e_version INTEGER NOT NULL DEFAULT 1
+  ,
+  x3dh_pub_key_1 BLOB,
+  x3dh_pub_key_2 BLOB
 ) WITHOUT ROWID;
 CREATE TABLE skipped_messages(
   skipped_message_id INTEGER PRIMARY KEY,
@@ -353,6 +362,13 @@ CREATE TABLE encrypted_rcv_message_hashes(
   created_at TEXT NOT NULL DEFAULT(datetime('now')),
   updated_at TEXT NOT NULL DEFAULT(datetime('now'))
 );
+CREATE TABLE processed_ratchet_key_hashes(
+  processed_ratchet_key_hash_id INTEGER PRIMARY KEY,
+  conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE,
+  hash BLOB NOT NULL,
+  created_at TEXT NOT NULL DEFAULT(datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT(datetime('now'))
+);
 CREATE UNIQUE INDEX idx_rcv_queues_ntf ON rcv_queues(host, port, ntf_id);
 CREATE UNIQUE INDEX idx_rcv_queue_id ON rcv_queues(conn_id, rcv_queue_id);
 CREATE UNIQUE INDEX idx_snd_queue_id ON snd_queues(conn_id, snd_queue_id);
@@ -443,3 +459,11 @@ CREATE INDEX idx_encrypted_rcv_message_hashes_hash ON encrypted_rcv_message_hash
   conn_id,
   hash
 );
+CREATE INDEX idx_processed_ratchet_key_hashes_hash ON processed_ratchet_key_hashes(
+  conn_id,
+  hash
+);
+CREATE INDEX idx_snd_messages_rcpt_internal_id ON snd_messages(
+  conn_id,
+  rcpt_internal_id
+);

src/Simplex/Messaging/Client.hs

@@ -11,6 +11,7 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 {-# LANGUAGE TypeApplications #-}
+{-# LANGUAGE TupleSections #-}
 
 -- |
 -- Module      : Simplex.Messaging.Client
@@ -41,6 +42,7 @@ module Simplex.Messaging.Client
     subscribeSMPQueues,
     getSMPMessage,
     subscribeSMPQueueNotifications,
+    subscribeSMPQueuesNtfs,
     secureSMPQueue,
     enableSMPQueueNotifications,
     disableSMPQueueNotifications,
@@ -81,10 +83,11 @@ import qualified Data.Aeson as J
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Either (rights)
+import Data.Foldable (foldl')
 import Data.Functor (($>))
 import Data.Int (Int64)
 import Data.List (find)
-import Data.List.NonEmpty (NonEmpty)
+import Data.List.NonEmpty (NonEmpty (..), (<|))
 import qualified Data.List.NonEmpty as L
 import Data.Maybe (fromMaybe)
 import Data.Time.Clock (UTCTime, getCurrentTime)
@@ -113,6 +116,9 @@ data ProtocolClient err msg = ProtocolClient
     sessionId :: SessionId,
     sessionTs :: UTCTime,
     thVersion :: Version,
+    timeoutPerBlock :: Int,
+    blockSize :: Int,
+    batch :: Bool,
     client_ :: PClient err msg
   }
 
@@ -121,6 +127,7 @@ data PClient err msg = PClient
     transportSession :: TransportSession msg,
     transportHost :: TransportHost,
     tcpTimeout :: Int,
+    batchDelay :: Maybe Int,
     pingErrorCount :: TVar Int,
     clientCorrId :: TVar Natural,
     sentCommands :: TMap CorrId (Request err msg),
@@ -167,6 +174,8 @@ data NetworkConfig = NetworkConfig
     tcpConnectTimeout :: Int,
     -- | timeout of protocol commands (microseconds)
     tcpTimeout :: Int,
+    -- | additional timeout per kilobyte (1024 bytes) to be sent
+    tcpTimeoutPerKb :: Int,
     -- | TCP keep-alive options, Nothing to skip enabling keep-alive
     tcpKeepAlive :: Maybe KeepAliveOpts,
     -- | period for SMP ping commands (microseconds, 0 to disable)
@@ -198,8 +207,9 @@ defaultNetworkConfig =
       hostMode = HMOnionViaSocks,
       requiredHostMode = False,
       sessionMode = TSMUser,
-      tcpConnectTimeout = 7_500_000,
-      tcpTimeout = 5_000_000,
+      tcpConnectTimeout = 15_000_000,
+      tcpTimeout = 10_000_000,
+      tcpTimeoutPerKb = 20_000, -- 20ms, should be less than 130ms to avoid Int overflow on 32 bit systems
       tcpKeepAlive = Just defaultKeepAliveOpts,
       smpPingInterval = 600_000_000, -- 10min
       smpPingCount = 3,
@@ -218,8 +228,10 @@ data ProtocolClientConfig = ProtocolClientConfig
     defaultTransport :: (ServiceName, ATransport),
     -- | network configuration
     networkConfig :: NetworkConfig,
-    -- | SMP client-server protocol version range
-    smpServerVRange :: VersionRange
+    -- | client-server protocol version range
+    serverVRange :: VersionRange,
+    -- | delay between sending batches of commands (microseconds)
+    batchDelay :: Maybe Int
   }
 
 -- | Default protocol client configuration.
@@ -229,7 +241,8 @@ defaultClientConfig =
     { qSize = 64,
       defaultTransport = ("443", transport @TLS),
       networkConfig = defaultNetworkConfig,
-      smpServerVRange = supportedSMPServerVRange
+      serverVRange = supportedSMPServerVRange,
+      batchDelay = Nothing
     }
 
 data Request err msg = Request
@@ -275,14 +288,14 @@ type TransportSession msg = (UserId, ProtoServer msg, Maybe EntityId)
 -- A single queue can be used for multiple 'SMPClient' instances,
 -- as 'SMPServerTransmission' includes server information.
 getProtocolClient :: forall err msg. Protocol err msg => TransportSession msg -> ProtocolClientConfig -> Maybe (TBQueue (ServerTransmission msg)) -> (ProtocolClient err msg -> IO ()) -> IO (Either (ProtocolClientError err) (ProtocolClient err msg))
-getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, smpServerVRange} msgQ disconnected = do
+getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize, networkConfig, serverVRange, batchDelay} msgQ disconnected = do
   case chooseTransportHost networkConfig (host srv) of
     Right useHost ->
       (atomically (mkProtocolClient useHost) >>= runClient useTransport useHost)
         `catch` \(e :: IOException) -> pure . Left $ PCEIOError e
     Left e -> pure $ Left e
   where
-    NetworkConfig {tcpConnectTimeout, tcpTimeout, smpPingInterval} = networkConfig
+    NetworkConfig {tcpConnectTimeout, tcpTimeout, tcpTimeoutPerKb, smpPingInterval} = networkConfig
     mkProtocolClient :: TransportHost -> STM (PClient err msg)
     mkProtocolClient transportHost = do
       connected <- newTVar False
@@ -297,6 +310,7 @@ getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize,
             transportSession,
             transportHost,
             tcpTimeout,
+            batchDelay,
             pingErrorCount,
             clientCorrId,
             sentCommands,
@@ -328,11 +342,12 @@ getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize,
 
     client :: forall c. Transport c => TProxy c -> PClient err msg -> TMVar (Either (ProtocolClientError err) (ProtocolClient err msg)) -> c -> IO ()
     client _ c cVar h =
-      runExceptT (protocolClientHandshake @err @msg h (keyHash srv) smpServerVRange) >>= \case
+      runExceptT (protocolClientHandshake @err @msg h (keyHash srv) serverVRange) >>= \case
         Left e -> atomically . putTMVar cVar . Left $ PCETransportError e
-        Right th@THandle {sessionId, thVersion} -> do
+        Right th@THandle {sessionId, thVersion, blockSize, batch} -> do
           sessionTs <- getCurrentTime
-          let c' = ProtocolClient {action = Nothing, client_ = c, sessionId, thVersion, sessionTs}
+          let timeoutPerBlock = (blockSize * tcpTimeoutPerKb) `div` 1024
+              c' = ProtocolClient {action = Nothing, client_ = c, sessionId, thVersion, sessionTs, timeoutPerBlock, blockSize, batch}
           atomically $ do
             writeTVar (connected c) True
             putTMVar cVar $ Right c'
@@ -340,7 +355,7 @@ getProtocolClient transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize,
             `finally` disconnected c'
 
     send :: Transport c => ProtocolClient err msg -> THandle c -> IO ()
-    send ProtocolClient {client_ = PClient {sndQ}} h = forever $ atomically (readTBQueue sndQ) >>= tPut h
+    send ProtocolClient {client_ = PClient {sndQ}} h = forever $ atomically (readTBQueue sndQ) >>= tPut h batchDelay
 
     receive :: Transport c => ProtocolClient err msg -> THandle c -> IO ()
     receive ProtocolClient {client_ = PClient {rcvQ}} h = forever $ tGet h >>= atomically . writeTBQueue rcvQ
@@ -487,6 +502,10 @@ getSMPMessage c rpKey rId =
 subscribeSMPQueueNotifications :: SMPClient -> NtfPrivateSignKey -> NotifierId -> ExceptT SMPClientError IO ()
 subscribeSMPQueueNotifications = okSMPCommand NSUB
 
+-- | Subscribe to multiple SMP queues notifications batching commands if supported.
+subscribeSMPQueuesNtfs :: SMPClient -> NonEmpty (NtfPrivateSignKey, NotifierId) -> IO (NonEmpty (Either SMPClientError ()))
+subscribeSMPQueuesNtfs = okSMPCommands NSUB
+
 -- | Secure the SMP queue by adding a sender public key.
 --
 -- https://github.com/simplex-chat/simplexmq/blob/master/protocol/simplex-messaging.md#secure-queue-command
@@ -578,32 +597,61 @@ okSMPCommands cmd c qs = L.map response <$> sendProtocolCommands c cs
 sendSMPCommand :: PartyI p => SMPClient -> Maybe C.APrivateSignKey -> QueueId -> Command p -> ExceptT SMPClientError IO BrokerMsg
 sendSMPCommand c pKey qId cmd = sendProtocolCommand c pKey qId (Cmd sParty cmd)
 
+type PCTransmission err msg = (SentRawTransmission, TMVar (Response err msg))
+
 -- | Send multiple commands with batching and collect responses
+-- It will result in Int overflow on 32 bit platform for a large number of blocks (~13.4k blocks / ~1.2m subscriptions)
+-- TODO switch to timeout or TimeManager that supports Int64
 sendProtocolCommands :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> NonEmpty (ClientCommand msg) -> IO (NonEmpty (Either (ProtocolClientError err) msg))
-sendProtocolCommands c@ProtocolClient {client_ = PClient {sndQ}} cs = do
-  ts <- mapM (runExceptT . mkTransmission c) cs
-  mapM_ (atomically . writeTBQueue sndQ . L.map fst) . L.nonEmpty . rights $ L.toList ts
-  forConcurrently ts $ \case
-    Right (_, r) -> withTimeout c $ atomically $ takeTMVar r
+sendProtocolCommands c@ProtocolClient {client_ = PClient {sndQ, tcpTimeout, batchDelay}, batch, blockSize, timeoutPerBlock} cs = do
+  (h :| ts) <- mapM (runExceptT . mkTransmission c) cs
+  let h' :: Either (ProtocolClientError err) (PCTransmission err msg, Int) = (,timeoutPerBlock) <$> h
+      batchSz = if batch then either (const 0) tSize h else 0
+      ts' :: NonEmpty (Either (ProtocolClientError err) (PCTransmission err msg, Int)) =
+        L.reverse . fst3 $ foldl' batchTimeouts ([h'], timeoutPerBlock, batchSz) ts
+      ts_ :: (Maybe (NonEmpty SentRawTransmission)) =
+        L.nonEmpty . map (fst . fst) . rights $ L.toList ts'
+  mapM_ (atomically . writeTBQueue sndQ) ts_
+  forConcurrently ts' $ \case
+    Right ((_t, r), bt) -> withTimeout c (tcpTimeout + bt) (atomically $ takeTMVar r)
     Left e -> pure $ Left e
+  where
+    fst3 (x, _, _) = x
+    -- tSize calculation matches the batching logic in tPut that does actual breaking of transmissions into blocks
+    tSize :: PCTransmission err msg -> Int
+    tSize ((sig, t), _) = maybe 0 C.signatureSize sig + B.length t + 3 -- 1 byte for signature size + 2 bytes for transmission size
+    batchTimeouts :: (NonEmpty (Either (ProtocolClientError err) (PCTransmission err msg, Int)), Int, Int) -> Either (ProtocolClientError err) (PCTransmission err msg) -> (NonEmpty (Either (ProtocolClientError err) (PCTransmission err msg, Int)), Int, Int)
+    batchTimeouts (ts, bt, batchSz) = \case
+      Left e -> (Left e <| ts, bt, batchSz)
+      Right t
+        | not batch ->
+          (Right (t, bt') <| ts, bt', 0)
+        | batchSz' + 1 > blockSize ->
+          (Right (t, bt') <| ts, bt', tSz)
+        | otherwise -> -- same block in the batch
+          (Right (t, bt) <| ts, bt, batchSz') -- 1 byte for the number of transmissions in the batch
+        where
+          batchSz' = batchSz + tSz
+          bt' = bt + timeoutPerBlock + fromMaybe 0 batchDelay
+          tSz = tSize t
 
 -- | Send Protocol command
 sendProtocolCommand :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> Maybe C.APrivateSignKey -> QueueId -> ProtoCommand msg -> ExceptT (ProtocolClientError err) IO msg
-sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}} pKey qId cmd = do
+sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ, tcpTimeout}} pKey qId cmd = do
   (t, r) <- mkTransmission c (pKey, qId, cmd)
   ExceptT $ sendRecv t r
   where
     -- two separate "atomically" needed to avoid blocking
     sendRecv :: SentRawTransmission -> TMVar (Response err msg) -> IO (Response err msg)
-    sendRecv t r = atomically (writeTBQueue sndQ [t]) >> withTimeout c (atomically $ takeTMVar r)
+    sendRecv t r = atomically (writeTBQueue sndQ [t]) >> withTimeout c tcpTimeout (atomically $ takeTMVar r)
 
-withTimeout :: ProtocolClient err msg -> IO (Either (ProtocolClientError err) msg) -> IO (Either (ProtocolClientError err) msg)
-withTimeout ProtocolClient {client_ = PClient {tcpTimeout, pingErrorCount}} a =
-  timeout tcpTimeout a >>= \case
+withTimeout :: ProtocolClient err msg -> Int -> IO (Either (ProtocolClientError err) msg) -> IO (Either (ProtocolClientError err) msg)
+withTimeout ProtocolClient {client_ = PClient {pingErrorCount}} t a = do
+  timeout t a >>= \case
     Just r -> atomically (writeTVar pingErrorCount 0) >> pure r
     _ -> pure $ Left PCEResponseTimeout
 
-mkTransmission :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> ClientCommand msg -> ExceptT (ProtocolClientError err) IO (SentRawTransmission, TMVar (Response err msg))
+mkTransmission :: forall err msg. ProtocolEncoding err (ProtoCommand msg) => ProtocolClient err msg -> ClientCommand msg -> ExceptT (ProtocolClientError err) IO (PCTransmission err msg)
 mkTransmission ProtocolClient {sessionId, thVersion, client_ = PClient {clientCorrId, sentCommands}} (pKey, qId, cmd) = do
   corrId <- liftIO $ atomically getNextCorrId
   let t = signTransmission $ encodeTransmission thVersion sessionId (corrId, qId, cmd)

src/Simplex/Messaging/Client/Agent.hs

@@ -5,6 +5,7 @@
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE RankNTypes #-}
 {-# LANGUAGE ScopedTypeVariables #-}
+{-# LANGUAGE TupleSections #-}
 {-# LANGUAGE TypeApplications #-}
 {-# OPTIONS_GHC -fno-warn-orphans #-}
 
@@ -16,24 +17,31 @@ import Control.Logger.Simple
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
 import Control.Monad.Trans.Except
+import Data.Bifunctor (first, bimap)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
+import Data.Either (partitionEithers)
+import Data.List (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 (listToMaybe)
 import Data.Set (Set)
 import Data.Text.Encoding
+import Data.Tuple (swap)
 import Numeric.Natural
 import Simplex.Messaging.Agent.RetryInterval
 import Simplex.Messaging.Client
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Protocol (BrokerMsg, ProtocolServer (..), QueueId, SMPServer)
+import Simplex.Messaging.Protocol (BrokerMsg, ProtocolServer (..), QueueId, SMPServer, NtfPrivateSignKey, NotifierId, RcvPrivateSignKey, RecipientId)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
-import Simplex.Messaging.Util (catchAll_, tryE, unlessM, ($>>=))
+import Simplex.Messaging.Util (catchAll_, tryE, ($>>=), toChunks)
 import System.Timeout (timeout)
-import UnliftIO (async, forConcurrently_)
+import UnliftIO (async)
 import UnliftIO.Exception (Exception)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
@@ -44,8 +52,8 @@ data SMPClientAgentEvent
   = CAConnected SMPServer
   | CADisconnected SMPServer (Set SMPSub)
   | CAReconnected SMPServer
-  | CAResubscribed SMPServer SMPSub
-  | CASubError SMPServer SMPSub SMPClientError
+  | CAResubscribed SMPServer (NonEmpty SMPSub)
+  | CASubError SMPServer (NonEmpty (SMPSub, SMPClientError))
 
 data SMPSubParty = SPRecipient | SPNotifier
   deriving (Eq, Ord, Show)
@@ -58,7 +66,8 @@ data SMPClientAgentConfig = SMPClientAgentConfig
   { smpCfg :: ProtocolClientConfig,
     reconnectInterval :: RetryInterval,
     msgQSize :: Natural,
-    agentQSize :: Natural
+    agentQSize :: Natural,
+    agentSubsBatchSize :: Int
   }
 
 defaultSMPClientAgentConfig :: SMPClientAgentConfig
@@ -71,8 +80,9 @@ defaultSMPClientAgentConfig =
             increaseAfter = 10 * second,
             maxInterval = 10 * second
           },
-      msgQSize = 64,
-      agentQSize = 64
+      msgQSize = 256,
+      agentQSize = 256,
+      agentSubsBatchSize = 900
     }
   where
     second = 1000000
@@ -184,9 +194,9 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ} srv =
             _ -> TM.insert srv sVar ps
 
     serverDown :: Map SMPSub C.APrivateSignKey -> IO ()
-    serverDown ss = unless (M.null ss) . void . runExceptT $ do
+    serverDown ss = unless (M.null ss) $ do
       notify . CADisconnected srv $ M.keysSet ss
-      reconnectServer
+      void $ runExceptT reconnectServer
 
     reconnectServer :: ExceptT SMPClientError IO ()
     reconnectServer = do
@@ -201,27 +211,37 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ} srv =
     reconnectClient :: ExceptT SMPClientError IO ()
     reconnectClient = do
       withSMP ca srv $ \smp -> do
-        notify $ CAReconnected srv
-        cs <- atomically $ mapM readTVar =<< TM.lookup srv (pendingSrvSubs ca)
-        forConcurrently_ (maybe [] M.assocs cs) $ \sub@(s, _) ->
-          unlessM (atomically $ hasSub (srvSubs ca) srv s) $
-            subscribe_ smp sub `catchE` handleError s
+        liftIO $ notify $ CAReconnected srv
+        cs_ <- atomically $ mapM readTVar =<< TM.lookup srv (pendingSrvSubs ca)
+        forM_ cs_ $ \cs -> do
+          subs' <- filterM (fmap not . atomically . hasSub (srvSubs ca) srv . fst) $ M.assocs cs
+          let (nSubs, rSubs) = partition (isNotifier . fst . fst) subs'
+          subscribe_ smp SPNotifier nSubs
+          subscribe_ smp SPRecipient rSubs
       where
-        subscribe_ :: SMPClient -> (SMPSub, C.APrivateSignKey) -> ExceptT SMPClientError IO ()
-        subscribe_ smp sub@(s, _) = do
-          smpSubscribe smp sub
-          atomically $ addSubscription ca srv sub
-          notify $ CAResubscribed srv s
+        isNotifier = \case
+          SPNotifier -> True
+          SPRecipient -> False
 
-        handleError :: SMPSub -> SMPClientError -> ExceptT SMPClientError IO ()
-        handleError s = \case
-          e@PCEResponseTimeout -> throwE e
-          e@PCENetworkError -> throwE e
-          e -> do
-            notify $ CASubError srv s e
-            atomically $ removePendingSubscription ca srv s
+        subscribe_ :: SMPClient -> SMPSubParty -> [(SMPSub, C.APrivateSignKey)] -> ExceptT SMPClientError IO ()
+        subscribe_ smp party = mapM_ subscribeBatch . toChunks (agentSubsBatchSize agentCfg)
+          where
+            subscribeBatch subs' = do
+              let subs'' :: (NonEmpty (QueueId, C.APrivateSignKey)) = L.map (first snd) subs'
+              rs <- liftIO $ smpSubscribeQueues party ca smp srv subs''
+              let rs' :: (NonEmpty ((SMPSub, C.APrivateSignKey), Either SMPClientError ())) =
+                    L.zipWith (first . const) subs' rs
+                  rs'' :: [Either (SMPSub, SMPClientError) (SMPSub, C.APrivateSignKey)] =
+                    map (\(sub, r) -> bimap (fst sub,) (const sub) r) $ L.toList rs'
+                  (errs, oks) = partitionEithers rs''
+                  (tempErrs, finalErrs) = partition (temporaryClientError . snd) errs
+              mapM_ (atomically . addSubscription ca srv) oks
+              mapM_ (liftIO . notify . CAResubscribed srv) $ L.nonEmpty $ map fst oks
+              mapM_ (atomically . removePendingSubscription ca srv . fst) finalErrs 
+              mapM_ (liftIO . notify . CASubError srv) $ L.nonEmpty finalErrs
+              mapM_ (throwE . snd) $ listToMaybe tempErrs
 
-    notify :: SMPClientAgentEvent -> ExceptT SMPClientError IO ()
+    notify :: SMPClientAgentEvent -> IO ()
     notify evt = atomically $ writeTBQueue (agentQ ca) evt
 
 closeSMPClientAgent :: MonadUnliftIO m => SMPClientAgent -> m ()
@@ -263,6 +283,35 @@ subscribeQueue ca srv sub = do
         removePendingSubscription ca srv $ fst sub
       throwE e
 
+subscribeQueuesSMP :: SMPClientAgent -> SMPServer -> NonEmpty (RecipientId, RcvPrivateSignKey) -> IO (NonEmpty (RecipientId, Either SMPClientError ()))
+subscribeQueuesSMP = subscribeQueues_ SPRecipient
+
+subscribeQueuesNtfs :: SMPClientAgent -> SMPServer -> NonEmpty (NotifierId, NtfPrivateSignKey) -> IO (NonEmpty (NotifierId, Either SMPClientError ()))
+subscribeQueuesNtfs = subscribeQueues_ SPNotifier
+
+subscribeQueues_ :: SMPSubParty -> SMPClientAgent -> SMPServer -> NonEmpty (QueueId, C.APrivateSignKey) -> IO (NonEmpty (QueueId, Either SMPClientError ()))
+subscribeQueues_ party ca srv subs = do
+  atomically $ forM_ subs $ addPendingSubscription ca srv . first (party,)
+  runExceptT (getSMPServerClient' ca srv) >>= \case
+    Left e -> pure $ L.map ((,Left e) . fst) subs
+    Right smp -> smpSubscribeQueues party ca smp srv subs
+
+smpSubscribeQueues :: SMPSubParty -> SMPClientAgent -> SMPClient -> SMPServer -> NonEmpty (QueueId, C.APrivateSignKey) -> IO (NonEmpty (QueueId, Either SMPClientError ()))
+smpSubscribeQueues party ca smp srv subs = do
+  rs <- L.zip subs <$> subscribe smp (L.map swap subs)
+  atomically $ forM rs $ \(sub, r) -> (fst sub,) <$> case r of
+    Right () -> do
+      addSubscription ca srv $ first (party,) sub
+      pure $ Right ()
+    Left e -> do
+      when (e /= PCENetworkError && e /= PCEResponseTimeout) $
+        removePendingSubscription ca srv $ (party,) $ fst sub
+      pure $ Left e
+  where
+    subscribe = case party of
+      SPRecipient -> subscribeSMPQueues
+      SPNotifier -> subscribeSMPQueuesNtfs
+
 showServer :: SMPServer -> ByteString
 showServer ProtocolServer {host, port} =
   strEncode host <> B.pack (if null port then "" else ':' : port)

src/Simplex/Messaging/Crypto.hs

@@ -680,6 +680,9 @@ instance SignatureSize (Signature a) where
     SignatureEd25519 _ -> Ed25519.signatureSize
     SignatureEd448 _ -> Ed448.signatureSize
 
+instance SignatureSize ASignature where
+  signatureSize (ASignature _ s) = signatureSize s
+
 instance SignatureSize APrivateSignKey where
   signatureSize (APrivateSignKey _ k) = signatureSize k
 

src/Simplex/Messaging/Notifications/Protocol.hs

@@ -429,7 +429,7 @@ data NtfSubStatus
     NSAuth
   | -- | SMP error other than AUTH
     NSErr ByteString
-  deriving (Eq, Show)
+  deriving (Eq, Ord, Show)
 
 ntfShouldSubscribe :: NtfSubStatus -> Bool
 ntfShouldSubscribe = \case

src/Simplex/Messaging/Notifications/Server.hs

@@ -16,12 +16,17 @@ import Control.Concurrent.STM (stateTVar)
 import Control.Logger.Simple
 import Control.Monad.Except
 import Control.Monad.Reader
+import Data.Bifunctor (second)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Functor (($>))
 import Data.Int (Int64)
-import Data.List (intercalate)
+import Data.List (intercalate, sort)
+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)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1)
 import Data.Time.Clock (UTCTime (..), diffTimeToPicoseconds, getCurrentTime)
@@ -51,7 +56,7 @@ import System.Exit (exitFailure)
 import System.IO (BufferMode (..), hPutStrLn, hSetBuffering)
 import System.Mem.Weak (deRefWeak)
 import UnliftIO (IOMode (..), async, uninterruptibleCancel, withFile)
-import UnliftIO.Concurrent (forkIO, killThread, mkWeakThreadId, threadDelay)
+import UnliftIO.Concurrent (forkIO, killThread, mkWeakThreadId)
 import UnliftIO.Directory (doesFileExist, renameFile)
 import UnliftIO.Exception
 import UnliftIO.STM
@@ -67,7 +72,7 @@ runNtfServerBlocking started cfg = runReaderT (ntfServer cfg started) =<< newNtf
 type M a = ReaderT NtfEnv IO a
 
 ntfServer :: NtfServerConfig -> TMVar Bool -> M ()
-ntfServer cfg@NtfServerConfig {transports, logTLSErrors} started = do
+ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
   restoreServerStats
   s <- asks subscriber
   ps <- asks pushServer
@@ -78,7 +83,7 @@ ntfServer cfg@NtfServerConfig {transports, logTLSErrors} started = do
     runServer :: (ServiceName, ATransport) -> M ()
     runServer (tcpPort, ATransport t) = do
       serverParams <- asks tlsServerParams
-      runTransportServer started tcpPort serverParams logTLSErrors (runClient t)
+      runTransportServer started tcpPort serverParams tCfg (runClient t)
 
     runClient :: Transport c => TProxy c -> c -> M ()
     runClient _ h = do
@@ -141,24 +146,27 @@ ntfServer cfg@NtfServerConfig {transports, logTLSErrors} started = do
 
 resubscribe :: NtfSubscriber -> Map NtfSubscriptionId NtfSubData -> M ()
 resubscribe NtfSubscriber {newSubQ} subs = do
-  d <- asks $ resubscribeDelay . config
-  forM_ subs $ \sub@NtfSubData {} ->
-    whenM (ntfShouldSubscribe <$> readTVarIO (subStatus sub)) $ do
-      atomically $ writeTBQueue newSubQ $ NtfSub sub
-      threadDelay d
-  liftIO $ logInfo "SMP connections resubscribed"
+  subs' <- atomically $ filterM (fmap ntfShouldSubscribe . readTVar . subStatus) $ M.elems subs
+  atomically . writeTBQueue newSubQ $ map NtfSub subs'
+  liftIO $ logInfo $ "SMP resubscriptions queued (" <> tshow (length subs') <> " subscriptions)"
 
 ntfSubscriber :: NtfSubscriber -> M ()
 ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAgent {msgQ, agentQ}} = do
   raceAny_ [subscribe, receiveSMP, receiveAgent]
   where
     subscribe :: M ()
-    subscribe =
-      forever $
-        atomically (readTBQueue newSubQ) >>= \case
-          sub@(NtfSub NtfSubData {smpQueue = SMPQueueNtf {smpServer}}) -> do
-            SMPSubscriber {newSubQ = subscriberSubQ} <- getSMPSubscriber smpServer
-            atomically $ writeTQueue subscriberSubQ sub
+    subscribe = forever $ do
+      subs <- atomically (readTBQueue newSubQ)
+      let ss = L.groupAllWith server subs
+      batchSize <- asks $ subsBatchSize . config
+      forM_ ss $ \serverSubs -> do
+        let srv = server $ L.head serverSubs
+            batches = toChunks batchSize $ L.toList serverSubs
+        SMPSubscriber {newSubQ = subscriberSubQ} <- getSMPSubscriber srv
+        mapM_ (atomically . writeTQueue subscriberSubQ) batches
+
+    server :: NtfEntityRec 'Subscription -> SMPServer
+    server (NtfSub sub) = ntfSubServer sub
 
     getSMPSubscriber :: SMPServer -> M SMPSubscriber
     getSMPSubscriber smpServer =
@@ -173,21 +181,36 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
 
     runSMPSubscriber :: SMPSubscriber -> M ()
     runSMPSubscriber SMPSubscriber {newSubQ = subscriberSubQ} =
-      forever $
-        atomically (peekTQueue subscriberSubQ)
-          >>= \(NtfSub NtfSubData {smpQueue, notifierKey}) -> do
-            updateSubStatus smpQueue NSPending
-            let SMPQueueNtf {smpServer, notifierId} = smpQueue
-            liftIO (runExceptT $ subscribeQueue ca smpServer ((SPNotifier, notifierId), notifierKey)) >>= \case
-              Right _ -> do
-                updateSubStatus smpQueue NSActive
-                void . atomically $ readTQueue subscriberSubQ
-              Left err -> do
-                handleSubError smpQueue err
-                case err of
-                  PCEResponseTimeout -> pure ()
-                  PCENetworkError -> pure ()
-                  _ -> void . atomically $ readTQueue subscriberSubQ
+      forever $ do
+        subs <- atomically (peekTQueue subscriberSubQ)
+        let subs' = L.map (\(NtfSub sub) -> sub) subs
+            srv = server $ L.head subs
+        logSubStatus srv "subscribing" $ length subs
+        mapM_ (\NtfSubData {smpQueue} -> updateSubStatus smpQueue NSPending) subs'
+        rs <- liftIO $ subscribeQueues srv subs'
+        (subs'', oks, errs) <- foldM process ([], 0, []) rs
+        atomically $ do
+          void $ readTQueue subscriberSubQ
+          mapM_ (writeTQueue subscriberSubQ . L.map NtfSub) $ L.nonEmpty subs''
+        logSubStatus srv "retrying" $ length subs''
+        logSubStatus srv "subscribed" oks
+        logSubErrors srv errs
+      where
+        process :: ([NtfSubData], Int, [NtfSubStatus]) -> (NtfSubData, Either SMPClientError ()) -> M ([NtfSubData], Int, [NtfSubStatus])
+        process (subs, oks, errs) (sub@NtfSubData {smpQueue}, r) = case r of
+          Right _ -> updateSubStatus smpQueue NSActive $> (subs, oks + 1, errs)
+          Left e -> update <$> handleSubError smpQueue e
+          where
+            update = \case
+              Just err -> (subs, oks, err : errs) -- permanent error, log and don't retry subscription
+              Nothing -> (sub : subs, oks, errs) -- temporary error, retry subscription
+
+    -- | Subscribe to queues. The list of results can have a different order.
+    subscribeQueues :: SMPServer -> NonEmpty NtfSubData -> IO (NonEmpty (NtfSubData, Either SMPClientError ()))
+    subscribeQueues srv subs =
+      L.map (second snd) . L.zip subs <$> subscribeQueuesNtfs ca srv (L.map sub subs)
+      where
+        sub NtfSubData {smpQueue = SMPQueueNtf {notifierId}, notifierKey} = (notifierId, notifierKey)
 
     receiveSMP :: M ()
     receiveSMP = forever $ do
@@ -212,37 +235,43 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
         atomically (readTBQueue agentQ) >>= \case
           CAConnected _ -> pure ()
           CADisconnected srv subs -> do
-            logInfo $ "SMP server disconnected " <> showServer' srv <> " (" <> tshow (length subs) <> ") subscriptions"
+            logSubStatus srv "disconnected" $ length subs
             forM_ subs $ \(_, ntfId) -> do
               let smpQueue = SMPQueueNtf srv ntfId
               updateSubStatus smpQueue NSInactive
           CAReconnected srv ->
             logInfo $ "SMP server reconnected " <> showServer' srv
-          CAResubscribed srv sub -> do
-            let ntfId = snd sub
-                smpQueue = SMPQueueNtf srv ntfId
-            updateSubStatus smpQueue NSActive
-          CASubError srv (_, ntfId) err -> do
-            logError $ "SMP subscription error on server " <> showServer' srv <> ": " <> tshow err
-            handleSubError (SMPQueueNtf srv ntfId) err
-      where
-        showServer' = decodeLatin1 . strEncode . host
+          CAResubscribed srv subs -> do
+            forM_ subs $ \(_, ntfId) -> updateSubStatus (SMPQueueNtf srv ntfId) NSActive
+            logSubStatus srv "resubscribed" $ length subs
+          CASubError srv errs ->
+            forM errs (\((_, ntfId), err) -> handleSubError (SMPQueueNtf srv ntfId) err)
+              >>= logSubErrors srv . catMaybes . L.toList
 
-    handleSubError :: SMPQueueNtf -> SMPClientError -> M ()
+    logSubStatus srv event n = when (n > 0) $
+      logInfo $ "SMP server " <> event <> " " <> showServer' srv <> " (" <> tshow n <> " subscriptions)"
+
+    logSubErrors :: SMPServer -> [NtfSubStatus] -> M ()        
+    logSubErrors srv errs = forM_ (L.group $ sort errs) $ \errs' -> do
+      logError $ "SMP subscription errors on server " <> showServer' srv <> ": " <> tshow (L.head errs') <> " (" <> tshow (length errs') <> " errors)"
+
+    showServer' = decodeLatin1 . strEncode . host
+
+    handleSubError :: SMPQueueNtf -> SMPClientError -> M (Maybe NtfSubStatus)
     handleSubError smpQueue = \case
-      PCEProtocolError AUTH -> updateSubStatus smpQueue NSAuth
+      PCEProtocolError AUTH -> updateSubStatus smpQueue NSAuth $> Just NSAuth
       PCEProtocolError e -> updateErr "SMP error " e
-      PCEIOError e -> updateErr "IOError " e
       PCEResponseError e -> updateErr "ResponseError " e
       PCEUnexpectedResponse r -> updateErr "UnexpectedResponse " r
       PCETransportError e -> updateErr "TransportError " e
       PCECryptoError e -> updateErr "CryptoError " e
-      PCEIncompatibleHost -> updateSubStatus smpQueue $ NSErr "IncompatibleHost"
-      PCEResponseTimeout -> pure ()
-      PCENetworkError -> pure ()
+      PCEIncompatibleHost -> let e = NSErr "IncompatibleHost" in updateSubStatus smpQueue e $> Just e
+      PCEResponseTimeout -> pure Nothing
+      PCENetworkError -> pure Nothing
+      PCEIOError _ -> pure Nothing
       where
-        updateErr :: Show e => ByteString -> e -> M ()
-        updateErr errType e = updateSubStatus smpQueue . NSErr $ errType <> bshow e
+        updateErr :: Show e => ByteString -> e -> M (Maybe NtfSubStatus)
+        updateErr errType e = updateSubStatus smpQueue (NSErr $ errType <> bshow e) $> Just (NSErr errType)
 
     updateSubStatus smpQueue status = do
       st <- asks store
@@ -336,7 +365,7 @@ receive th NtfServerClient {rcvQ, sndQ, activeAt} = forever $ do
 send :: Transport c => THandle c -> NtfServerClient -> IO ()
 send h@THandle {thVersion = v} NtfServerClient {sndQ, sessionId, activeAt} = forever $ do
   t <- atomically $ readTBQueue sndQ
-  void . liftIO $ tPut h [(Nothing, encodeTransmission v sessionId t)]
+  void . liftIO $ tPut h Nothing [(Nothing, encodeTransmission v sessionId t)]
   atomically . writeTVar activeAt =<< liftIO getSystemTime
 
 -- instance Show a => Show (TVar a) where
@@ -504,7 +533,7 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
         sub <- atomically $ mkNtfSubData subId newSub
         resp <-
           atomically (addNtfSubscription st subId sub) >>= \case
-            Just _ -> atomically (writeTBQueue newSubQ $ NtfSub sub) $> NRSubId subId
+            Just _ -> atomically (writeTBQueue newSubQ [NtfSub sub]) $> NRSubId subId
             _ -> pure $ NRErr AUTH
         withNtfLog (`logCreateSubscription` sub)
         incNtfStat subCreated

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

@@ -12,6 +12,7 @@ import Control.Monad.IO.Unlift
 import Crypto.Random
 import Data.ByteString.Char8 (ByteString)
 import Data.Int (Int64)
+import Data.List.NonEmpty (NonEmpty)
 import Data.Time.Clock (getCurrentTime)
 import Data.Time.Clock.System (SystemTime)
 import Data.Word (Word16)
@@ -31,7 +32,7 @@ import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (ATransport)
-import Simplex.Messaging.Transport.Server (loadFingerprint, loadTLSServerParams)
+import Simplex.Messaging.Transport.Server (loadFingerprint, loadTLSServerParams, TransportServerConfig)
 import System.IO (IOMode (..))
 import System.Mem.Weak (Weak)
 import UnliftIO.STM
@@ -45,9 +46,9 @@ data NtfServerConfig = NtfServerConfig
     pushQSize :: Natural,
     smpAgentCfg :: SMPClientAgentConfig,
     apnsConfig :: APNSPushClientConfig,
+    subsBatchSize :: Int,
     inactiveClientExpiration :: Maybe ExpirationConfig,
     storeLogFile :: Maybe FilePath,
-    resubscribeDelay :: Int, -- microseconds
     -- CA certificate private key is not needed for initialization
     caCertificateFile :: FilePath,
     privateKeyFile :: FilePath,
@@ -57,7 +58,7 @@ data NtfServerConfig = NtfServerConfig
     logStatsStartTime :: Int64,
     serverStatsLogFile :: FilePath,
     serverStatsBackupFile :: Maybe FilePath,
-    logTLSErrors :: Bool
+    transportConfig :: TransportServerConfig
   }
 
 defaultInactiveClientExpiration :: ExpirationConfig
@@ -93,7 +94,7 @@ newNtfServerEnv config@NtfServerConfig {subQSize, pushQSize, smpAgentCfg, apnsCo
 
 data NtfSubscriber = NtfSubscriber
   { smpSubscribers :: TMap SMPServer SMPSubscriber,
-    newSubQ :: TBQueue (NtfEntityRec 'Subscription),
+    newSubQ :: TBQueue [NtfEntityRec 'Subscription],
     smpAgent :: SMPClientAgent
   }
 
@@ -105,7 +106,7 @@ newNtfSubscriber qSize smpAgentCfg = do
   pure NtfSubscriber {smpSubscribers, newSubQ, smpAgent}
 
 data SMPSubscriber = SMPSubscriber
-  { newSubQ :: TQueue (NtfEntityRec 'Subscription),
+  { newSubQ :: TQueue (NonEmpty (NtfEntityRec 'Subscription)),
     subThreadId :: TVar (Maybe (Weak ThreadId))
   }
 

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

@@ -14,7 +14,8 @@ import Data.Maybe (fromMaybe)
 import qualified Data.Text as T
 import Network.Socket (HostName)
 import Options.Applicative
-import Simplex.Messaging.Client.Agent (defaultSMPClientAgentConfig)
+import Simplex.Messaging.Client (ProtocolClientConfig (..))
+import Simplex.Messaging.Client.Agent (SMPClientAgentConfig (..), defaultSMPClientAgentConfig)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Notifications.Server (runNtfServer)
 import Simplex.Messaging.Notifications.Server.Env (NtfServerConfig (..))
@@ -22,13 +23,17 @@ import Simplex.Messaging.Notifications.Server.Push.APNS (defaultAPNSPushClientCo
 import Simplex.Messaging.Protocol (ProtoServerWithAuth (..), pattern NtfServer)
 import Simplex.Messaging.Server.CLI
 import Simplex.Messaging.Transport.Client (TransportHost (..))
+import Simplex.Messaging.Transport.Server (TransportServerConfig (..), defaultTransportServerConfig)
 import System.Directory (createDirectoryIfMissing, doesFileExist)
 import System.FilePath (combine)
 import System.IO (BufferMode (..), hSetBuffering, stderr, stdout)
 import Text.Read (readMaybe)
 
 ntfServerVersion :: String
-ntfServerVersion = "1.4.0"
+ntfServerVersion = "1.5.1"
+
+defaultSMPBatchDelay :: Int
+defaultSMPBatchDelay = 10000
 
 ntfServerCLI :: FilePath -> FilePath -> IO ()
 ntfServerCLI cfgPath logPath =
@@ -80,7 +85,9 @@ ntfServerCLI cfgPath logPath =
             <> ("host: " <> host <> "\n")
             <> ("port: " <> defaultServerPort <> "\n")
             <> "log_tls_errors: off\n\
-               \websockets: off\n"
+               \# delay between command batches sent to SMP relays (microseconds), 0 to disable\n"
+            <> ("smp_batch_delay: " <> show defaultSMPBatchDelay <> "\n")
+            <> "websockets: off\n"
     runServer ini = do
       hSetBuffering stdout LineBuffering
       hSetBuffering stderr LineBuffering
@@ -96,19 +103,21 @@ ntfServerCLI cfgPath logPath =
         enableStoreLog = settingIsOn "STORE_LOG" "enable" ini
         logStats = settingIsOn "STORE_LOG" "log_stats" ini
         c = combine cfgPath . ($ defaultX509Config)
+        smpBatchDelay = readIniDefault defaultSMPBatchDelay "TRANSPORT" "smp_batch_delay" ini
+        batchDelay = if smpBatchDelay <= 0 then Nothing else Just smpBatchDelay
         serverConfig =
           NtfServerConfig
             { transports = iniTransports ini,
               subIdBytes = 24,
               regCodeBytes = 32,
-              clientQSize = 16,
-              subQSize = 64,
-              pushQSize = 128,
-              smpAgentCfg = defaultSMPClientAgentConfig,
+              clientQSize = 64,
+              subQSize = 512,
+              pushQSize = 1048,
+              smpAgentCfg = defaultSMPClientAgentConfig {smpCfg = (smpCfg defaultSMPClientAgentConfig) {batchDelay}},
               apnsConfig = defaultAPNSPushClientConfig,
+              subsBatchSize = 900,
               inactiveClientExpiration = Nothing,
               storeLogFile = enableStoreLog $> storeLogFilePath,
-              resubscribeDelay = 50000, -- 50ms
               caCertificateFile = c caCrtFile,
               privateKeyFile = c serverKeyFile,
               certificateFile = c serverCrtFile,
@@ -116,7 +125,10 @@ ntfServerCLI cfgPath logPath =
               logStatsStartTime = 0, -- seconds from 00:00 UTC
               serverStatsLogFile = combine logPath "ntf-server-stats.daily.log",
               serverStatsBackupFile = logStats $> combine logPath "ntf-server-stats.log",
-              logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini
+              transportConfig =
+                defaultTransportServerConfig
+                  { logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini
+                  }
             }
 
 data CliCommand

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

@@ -20,7 +20,7 @@ import qualified Data.Set as S
 import Data.Word (Word16)
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Notifications.Protocol
-import Simplex.Messaging.Protocol (NtfPrivateSignKey)
+import Simplex.Messaging.Protocol (NtfPrivateSignKey, SMPServer)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Util (whenM, ($>>=))
@@ -68,6 +68,9 @@ data NtfSubData = NtfSubData
     subStatus :: TVar NtfSubStatus
   }
 
+ntfSubServer :: NtfSubData -> SMPServer
+ntfSubServer NtfSubData {smpQueue = SMPQueueNtf {smpServer}} = smpServer
+
 data NtfEntityRec (e :: NtfEntity) where
   NtfTkn :: NtfTknData -> NtfEntityRec 'Token
   NtfSub :: NtfSubData -> NtfEntityRec 'Subscription

src/Simplex/Messaging/Protocol.hs

@@ -146,6 +146,7 @@ module Simplex.Messaging.Protocol
 where
 
 import Control.Applicative (optional, (<|>))
+import Control.Concurrent (threadDelay)
 import Control.Monad.Except
 import Data.Aeson (FromJSON (..), ToJSON (..))
 import qualified Data.Aeson as J
@@ -1244,8 +1245,8 @@ instance Encoding CommandError where
       _ -> fail "bad command error type"
 
 -- | Send signed SMP transmission to TCP transport.
-tPut :: Transport c => THandle c -> NonEmpty SentRawTransmission -> IO [Either TransportError ()]
-tPut th trs
+tPut :: Transport c => THandle c -> Maybe Int -> NonEmpty SentRawTransmission -> IO [Either TransportError ()]
+tPut th delay_ trs 
   | batch th = tPutBatch [] $ L.map tEncode trs
   | otherwise = forM (L.toList trs) $ tPutLog . tEncode
   where
@@ -1255,7 +1256,7 @@ tPut th trs
       r <- if n == 0 then largeMsg else replicate n <$> tPutLog (tEncodeBatch n s)
       let rs' = rs <> r
       case ts_ of
-        Just ts' -> tPutBatch rs' ts'
+        Just ts' -> mapM_ threadDelay delay_ >> tPutBatch rs' ts'
         _ -> pure rs'
     largeMsg = putStrLn "tPut error: large message" >> pure [Left TELargeMsg]
     tPutLog s = do

src/Simplex/Messaging/Server.hs

@@ -58,11 +58,13 @@ import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Time.Format.ISO8601 (iso8601Show)
 import Data.Type.Equality
 import GHC.TypeLits (KnownNat)
-import Network.Socket (ServiceName)
+import Network.Socket (ServiceName, Socket, socketToHandle)
+import Simplex.Messaging.Agent.Lock
 import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding (Encoding (smpEncode))
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
+import Simplex.Messaging.Server.Control
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.MsgStore
@@ -74,10 +76,11 @@ import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
+import Simplex.Messaging.Transport.Buffer (trimCR)
 import Simplex.Messaging.Transport.Server
 import Simplex.Messaging.Util
 import System.Exit (exitFailure)
-import System.IO (hPutStrLn)
+import System.IO (hPutStrLn, hSetNewlineMode, universalNewlineMode)
 import System.Mem.Weak (deRefWeak)
 import UnliftIO.Concurrent
 import UnliftIO.Directory (doesFileExist, renameFile)
@@ -103,26 +106,29 @@ runSMPServerBlocking started cfg = newEnv cfg >>= runReaderT (smpServer started
 type M a = ReaderT Env IO a
 
 smpServer :: TMVar Bool -> ServerConfig -> M ()
-smpServer started cfg@ServerConfig {transports, logTLSErrors} = do
+smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
   s <- asks server
   restoreServerMessages
   restoreServerStats
   raceAny_
     ( serverThread s subscribedQ subscribers subscriptions cancelSub :
       serverThread s ntfSubscribedQ notifiers ntfSubscriptions (\_ -> pure ()) :
-      map runServer transports <> expireMessagesThread_ cfg <> serverStatsThread_ cfg
+      map runServer transports <> expireMessagesThread_ cfg <> serverStatsThread_ cfg <> controlPortThread_ cfg
     )
-    `finally` (withLog closeStoreLog >> saveServerMessages >> saveServerStats)
+    `finally` withLock (savingLock s) "final" (saveServer False)
   where
     runServer :: (ServiceName, ATransport) -> M ()
     runServer (tcpPort, ATransport t) = do
       serverParams <- asks tlsServerParams
-      runTransportServer started tcpPort serverParams logTLSErrors (runClient t)
+      runTransportServer started tcpPort serverParams tCfg (runClient t)
+
+    saveServer :: Bool -> M ()
+    saveServer keepMsgs = withLog closeStoreLog >> saveServerMessages keepMsgs >> saveServerStats
 
     serverThread ::
       forall s.
       Server ->
-      (Server -> TBQueue (QueueId, Client)) ->
+      (Server -> TQueue (QueueId, Client)) ->
       (Server -> TMap QueueId Client) ->
       (Client -> TMap QueueId s) ->
       (s -> IO ()) ->
@@ -134,7 +140,7 @@ smpServer started cfg@ServerConfig {transports, logTLSErrors} = do
       where
         updateSubscribers :: STM (Maybe (QueueId, Client))
         updateSubscribers = do
-          (qId, clnt) <- readTBQueue $ subQ s
+          (qId, clnt) <- readTQueue $ subQ s
           let clientToBeNotified = \c' ->
                 if sameClientSession clnt c'
                   then pure Nothing
@@ -223,6 +229,57 @@ smpServer started cfg@ServerConfig {transports, logTLSErrors} = do
         Right th -> runClientTransport th
         Left _ -> pure ()
 
+    controlPortThread_ :: ServerConfig -> [M ()]
+    controlPortThread_ ServerConfig {controlPort = Just port} = [runCPServer port]
+    controlPortThread_ _ = []
+
+    runCPServer :: ServiceName -> M ()
+    runCPServer port = do
+      srv <- asks server
+      cpStarted <- newEmptyTMVarIO
+      u <- askUnliftIO
+      liftIO $ runTCPServer cpStarted port $ runCPClient u srv
+      where
+        runCPClient :: UnliftIO (ReaderT Env IO) -> Server -> Socket -> IO ()
+        runCPClient u srv sock = do
+          h <- socketToHandle sock ReadWriteMode
+          hSetBuffering h LineBuffering
+          hSetNewlineMode h universalNewlineMode
+          hPutStrLn h "SMP server control port\n'help' for supported commands"
+          cpLoop h
+          where
+            cpLoop h = do
+              s <- B.hGetLine h
+              case strDecode $ trimCR s of
+                Right CPQuit -> hClose h
+                Right cmd -> processCP h cmd >> cpLoop h
+                Left err -> hPutStrLn h ("error: " <> err) >> cpLoop h
+            processCP h = \case
+              CPSuspend -> hPutStrLn h "suspend not implemented"
+              CPResume -> hPutStrLn h "resume not implemented"
+              CPClients -> hPutStrLn h "clients not implemented"
+              CPStats -> do
+                ServerStats {fromTime, qCreated, qSecured, qDeleted, msgSent, msgRecv, msgSentNtf, msgRecvNtf, qCount, msgCount} <- unliftIO u $ asks serverStats
+                putStat "fromTime" fromTime
+                putStat "qCreated" qCreated
+                putStat "qSecured" qSecured
+                putStat "qDeleted" qDeleted
+                putStat "msgSent" msgSent
+                putStat "msgRecv" msgRecv
+                putStat "msgSentNtf" msgSentNtf
+                putStat "msgRecvNtf" msgRecvNtf
+                putStat "qCount" qCount
+                putStat "msgCount" msgCount
+                where
+                  putStat :: Show a => String -> TVar a -> IO ()
+                  putStat label var = readTVarIO var >>= \v -> hPutStrLn h $ label <> ": " <> show v
+              CPSave -> withLock (savingLock srv) "control" $ do
+                hPutStrLn h "saving server state..."
+                unliftIO u $ saveServer True
+                hPutStrLn h "server state saved!"
+              CPHelp -> hPutStrLn h "commands: stats, save, help, quit"
+              CPQuit -> pure ()
+
 runClientTransport :: Transport c => THandle c -> M ()
 runClientTransport th@THandle {thVersion, sessionId} = do
   q <- asks $ tbqSize . config
@@ -281,12 +338,13 @@ receive th Client {rcvQ, sndQ, activeAt} = forever $ do
 send :: Transport c => THandle c -> Client -> IO ()
 send h@THandle {thVersion = v} Client {sndQ, sessionId, activeAt} = forever $ do
   ts <- atomically $ L.sortWith tOrder <$> readTBQueue sndQ
-  void . liftIO . tPut h $ L.map ((Nothing,) . encodeTransmission v sessionId) ts
+  void . liftIO . tPut h Nothing $ L.map ((Nothing,) . encodeTransmission v sessionId) ts
   atomically . writeTVar activeAt =<< liftIO getSystemTime
   where
     tOrder :: Transmission BrokerMsg -> Int
     tOrder (_, _, cmd) = case cmd of
       MSG {} -> 0
+      NMSG {} -> 0
       _ -> 1
 
 disconnectTransport :: Transport c => THandle c -> client -> (client -> TVar SystemTime) -> ExpirationConfig -> IO ()
@@ -476,7 +534,7 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
           where
             newSub :: m (TVar Sub)
             newSub = time "SUB newSub" . atomically $ do
-              writeTBQueue subscribedQ (rId, clnt)
+              writeTQueue subscribedQ (rId, clnt)
               sub <- newTVar =<< newSubscription NoSub
               TM.insert rId sub subscriptions
               pure sub
@@ -521,7 +579,7 @@ client clnt@Client {thVersion, subscriptions, ntfSubscriptions, rcvQ, sndQ} Serv
         subscribeNotifications :: m (Transmission BrokerMsg)
         subscribeNotifications = time "NSUB" . atomically $ do
           unlessM (TM.member queueId ntfSubscriptions) $ do
-            writeTBQueue ntfSubscribedQ (queueId, clnt)
+            writeTQueue ntfSubscribedQ (queueId, clnt)
             TM.insert queueId () ntfSubscriptions
           pure ok
 
@@ -719,8 +777,8 @@ randomId n = do
   gVar <- asks idsDrg
   atomically (C.pseudoRandomBytes n gVar)
 
-saveServerMessages :: (MonadUnliftIO m, MonadReader Env m) => m ()
-saveServerMessages = asks (storeMsgsFile . config) >>= mapM_ saveMessages
+saveServerMessages :: (MonadUnliftIO m, MonadReader Env m) => Bool -> m ()
+saveServerMessages keepMsgs = asks (storeMsgsFile . config) >>= mapM_ saveMessages
   where
     saveMessages f = do
       logInfo $ "saving messages to file " <> T.pack f
@@ -729,8 +787,9 @@ saveServerMessages = asks (storeMsgsFile . config) >>= mapM_ saveMessages
         readTVarIO ms >>= mapM_ (saveQueueMsgs ms h) . M.keys
       logInfo "messages saved"
       where
+        getMessages = if keepMsgs then snapshotMsgQueue else flushMsgQueue
         saveQueueMsgs ms h rId =
-          atomically (flushMsgQueue ms rId)
+          atomically (getMessages ms rId)
             >>= mapM_ (B.hPutStrLn h . strEncode . MLRv3 rId)
 
 restoreServerMessages :: forall m. (MonadUnliftIO m, MonadReader Env m) => m ()

src/Simplex/Messaging/Server/Control.hs

@@ -0,0 +1,36 @@
+{-# LANGUAGE LambdaCase #-}
+{-# LANGUAGE OverloadedStrings #-}
+
+module Simplex.Messaging.Server.Control where
+
+import qualified Data.Attoparsec.ByteString.Char8 as A
+import Simplex.Messaging.Encoding.String
+
+data ControlProtocol
+  = CPSuspend
+  | CPResume
+  | CPClients
+  | CPStats
+  | CPSave
+  | CPHelp
+  | CPQuit
+
+instance StrEncoding ControlProtocol where
+  strEncode = \case
+    CPSuspend -> "suspend"
+    CPResume -> "resume"
+    CPClients -> "clients"
+    CPStats -> "stats"
+    CPSave -> "save"
+    CPHelp -> "help"
+    CPQuit -> "quit"
+  strP =
+    A.takeTill (== ' ') >>= \case
+      "suspend" -> pure CPSuspend
+      "resume" -> pure CPResume
+      "clients" -> pure CPClients
+      "stats" -> pure CPStats
+      "save" -> pure CPSave
+      "help" -> pure CPHelp
+      "quit" -> pure CPQuit
+      _ -> fail "bad ControlProtocol command"

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

@@ -19,6 +19,7 @@ import Data.X509.Validation (Fingerprint (..))
 import Network.Socket (ServiceName)
 import qualified Network.TLS as T
 import Numeric.Natural
+import Simplex.Messaging.Agent.Lock
 import Simplex.Messaging.Crypto (KeyHash (..))
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.Expiration
@@ -30,7 +31,7 @@ import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport (ATransport)
-import Simplex.Messaging.Transport.Server (loadFingerprint, loadTLSServerParams)
+import Simplex.Messaging.Transport.Server (loadFingerprint, loadTLSServerParams, TransportServerConfig)
 import Simplex.Messaging.Version
 import System.IO (IOMode (..))
 import System.Mem.Weak (Weak)
@@ -39,7 +40,7 @@ import UnliftIO.STM
 data ServerConfig = ServerConfig
   { transports :: [(ServiceName, ATransport)],
     tbqSize :: Natural,
-    serverTbqSize :: Natural,
+    -- serverTbqSize :: Natural,
     msgQueueQuota :: Int,
     queueIdBytes :: Int,
     msgIdBytes :: Int,
@@ -69,7 +70,10 @@ data ServerConfig = ServerConfig
     certificateFile :: FilePath,
     -- | SMP client-server protocol version range
     smpServerVRange :: VersionRange,
-    logTLSErrors :: Bool
+    -- | TCP transport config
+    transportConfig :: TransportServerConfig,
+    -- | run listener on control port
+    controlPort :: Maybe ServiceName
   }
 
 defMsgExpirationDays :: Int64
@@ -102,10 +106,11 @@ data Env = Env
   }
 
 data Server = Server
-  { subscribedQ :: TBQueue (RecipientId, Client),
+  { subscribedQ :: TQueue (RecipientId, Client),
     subscribers :: TMap RecipientId Client,
-    ntfSubscribedQ :: TBQueue (NotifierId, Client),
-    notifiers :: TMap NotifierId Client
+    ntfSubscribedQ :: TQueue (NotifierId, Client),
+    notifiers :: TMap NotifierId Client,
+    savingLock :: Lock
   }
 
 data Client = Client
@@ -126,13 +131,14 @@ data Sub = Sub
     delivered :: TMVar MsgId
   }
 
-newServer :: Natural -> STM Server
-newServer qSize = do
-  subscribedQ <- newTBQueue qSize
+newServer :: STM Server
+newServer = do
+  subscribedQ <- newTQueue
   subscribers <- TM.empty
-  ntfSubscribedQ <- newTBQueue qSize
+  ntfSubscribedQ <- newTQueue
   notifiers <- TM.empty
-  return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers}
+  savingLock <- createLock
+  return Server {subscribedQ, subscribers, ntfSubscribedQ, notifiers, savingLock}
 
 newClient :: Natural -> Version -> ByteString -> SystemTime -> STM Client
 newClient qSize thVersion sessionId ts = do
@@ -151,26 +157,25 @@ newSubscription subThread = do
 
 newEnv :: forall m. (MonadUnliftIO m, MonadRandom m) => ServerConfig -> m Env
 newEnv config@ServerConfig {caCertificateFile, certificateFile, privateKeyFile, storeLogFile} = do
-  server <- atomically $ newServer (serverTbqSize config)
+  server <- atomically newServer
   queueStore <- atomically newQueueStore
   msgStore <- atomically newMsgStore
   idsDrg <- drgNew >>= newTVarIO
-  storeLog <- liftIO $ openReadStoreLog `mapM` storeLogFile
-  s' <- restoreQueues queueStore `mapM` storeLog
+  storeLog <- restoreQueues queueStore `mapM` storeLogFile
   tlsServerParams <- liftIO $ loadTLSServerParams caCertificateFile certificateFile privateKeyFile
   Fingerprint fp <- liftIO $ loadFingerprint caCertificateFile
   let serverIdentity = KeyHash fp
   serverStats <- atomically . newServerStats =<< liftIO getCurrentTime
-  return Env {config, server, serverIdentity, queueStore, msgStore, idsDrg, storeLog = s', tlsServerParams, serverStats}
+  return Env {config, server, serverIdentity, queueStore, msgStore, idsDrg, storeLog, tlsServerParams, serverStats}
   where
-    restoreQueues :: QueueStore -> StoreLog 'ReadMode -> m (StoreLog 'WriteMode)
-    restoreQueues QueueStore {queues, senders, notifiers} s = do
-      (qs, s') <- liftIO $ readWriteStoreLog s
+    restoreQueues :: QueueStore -> FilePath -> m (StoreLog 'WriteMode)
+    restoreQueues QueueStore {queues, senders, notifiers} f = do
+      (qs, s) <- liftIO $ readWriteStoreLog f
       atomically $ do
         writeTVar queues =<< mapM newTVar qs
         writeTVar senders $! M.foldr' addSender M.empty qs
         writeTVar notifiers $! M.foldr' addNotifier M.empty qs
-      pure s'
+      pure s
     addSender :: QueueRec -> Map SenderId RecipientId -> Map SenderId RecipientId
     addSender q = M.insert (senderId q) (recipientId q)
     addNotifier :: QueueRec -> Map NotifierId RecipientId -> Map NotifierId RecipientId

src/Simplex/Messaging/Server/Main.hs

@@ -28,6 +28,7 @@ import Simplex.Messaging.Server.Env.STM (ServerConfig (..), defaultInactiveClien
 import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Transport (simplexMQVersion, supportedSMPServerVRange)
 import Simplex.Messaging.Transport.Client (TransportHost (..))
+import Simplex.Messaging.Transport.Server (TransportServerConfig (..), defaultTransportServerConfig)
 import Simplex.Messaging.Util (safeDecodeUtf8)
 import System.Directory (createDirectoryIfMissing, doesFileExist)
 import System.FilePath (combine)
@@ -129,7 +130,8 @@ smpServerCLI cfgPath logPath =
                 <> ("host: " <> host <> "\n")
                 <> ("port: " <> defaultServerPort <> "\n")
                 <> "log_tls_errors: off\n\
-                   \websockets: off\n\n\
+                   \websockets: off\n\
+                   \# control_port: 5224\n\n\
                    \[INACTIVE_CLIENTS]\n\
                    \# TTL and interval to check inactive clients\n\
                    \disconnect: off\n"
@@ -164,8 +166,8 @@ smpServerCLI cfgPath logPath =
         serverConfig =
           ServerConfig
             { transports = iniTransports ini,
-              tbqSize = 32,
-              serverTbqSize = 1024,
+              tbqSize = 64,
+              -- serverTbqSize = 1024,
               msgQueueQuota = 128,
               queueIdBytes = 24,
               msgIdBytes = 24, -- must be at least 24 bytes, it is used as 192-bit nonce for XSalsa20
@@ -198,7 +200,11 @@ smpServerCLI cfgPath logPath =
               serverStatsLogFile = combine logPath "smp-server-stats.daily.log",
               serverStatsBackupFile = logStats $> combine logPath "smp-server-stats.log",
               smpServerVRange = supportedSMPServerVRange,
-              logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini
+              transportConfig =
+                defaultTransportServerConfig
+                  { logTLSErrors = fromMaybe False $ iniOnOff "TRANSPORT" "log_tls_errors" ini
+                  },
+              controlPort = either (const Nothing) (Just . T.unpack) $ lookupValue "TRANSPORT" "control_port" ini
             }
 
 data CliCommand

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

@@ -13,6 +13,7 @@ module Simplex.Messaging.Server.MsgStore.STM
     getMsgQueue,
     delMsgQueue,
     flushMsgQueue,
+    snapshotMsgQueue,
     writeMsg,
     tryPeekMsg,
     peekMsg,
@@ -62,6 +63,14 @@ delMsgQueue st rId = TM.delete rId st
 flushMsgQueue :: STMMsgStore -> RecipientId -> STM [Message]
 flushMsgQueue st rId = TM.lookupDelete rId st >>= maybe (pure []) (flushTQueue . msgQueue)
 
+snapshotMsgQueue :: STMMsgStore -> RecipientId -> STM [Message]
+snapshotMsgQueue st rId  = TM.lookup rId st >>= maybe (pure []) (snapshotTQueue . msgQueue)
+  where
+    snapshotTQueue q = do
+      msgs <- flushTQueue q
+      mapM_ (writeTQueue q) msgs
+      pure msgs
+
 writeMsg :: MsgQueue -> Message -> STM (Maybe Message)
 writeMsg MsgQueue {msgQueue = q, quota, canWrite, size} msg = do
   canWrt <- readTVar canWrite

src/Simplex/Messaging/Server/Stats.hs

@@ -94,7 +94,7 @@ setServerStats s d = do
   writeTVar (msgRecvNtf s) $! _msgRecvNtf d
   setPeriodStats (activeQueuesNtf s) (_activeQueuesNtf d)
   writeTVar (qCount s) $! _qCount d
-  writeTVar (msgCount s) $! _qCount d
+  writeTVar (msgCount s) $! _msgCount d
 
 instance StrEncoding ServerStatsData where
   strEncode ServerStatsData {_fromTime, _qCreated, _qSecured, _qDeleted, _msgSent, _msgRecv, _msgSentNtf, _msgRecvNtf, _activeQueues, _activeQueuesNtf} =

src/Simplex/Messaging/Server/StoreLog.hs

@@ -25,20 +25,17 @@ module Simplex.Messaging.Server.StoreLog
 where
 
 import Control.Applicative (optional, (<|>))
-import Control.Monad (unless)
-import Data.Bifunctor (first, second)
+import Control.Monad (foldM, unless, when)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
-import qualified Data.ByteString.Lazy.Char8 as LB
-import Data.Either (partitionEithers)
 import Data.Functor (($>))
-import Data.List (foldl')
 import Data.Map.Strict (Map)
 import qualified Data.Map.Strict as M
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.QueueStore (NtfCreds (..), QueueRec (..), ServerQueueStatus (..))
 import Simplex.Messaging.Transport.Buffer (trimCR)
+import Simplex.Messaging.Util (ifM)
 import System.Directory (doesFileExist)
 import System.IO
 
@@ -139,37 +136,33 @@ logDeleteQueue s = writeStoreLogRecord s . DeleteQueue
 logDeleteNotifier :: StoreLog 'WriteMode -> QueueId -> IO ()
 logDeleteNotifier s = writeStoreLogRecord s . DeleteNotifier
 
-readWriteStoreLog :: StoreLog 'ReadMode -> IO (Map RecipientId QueueRec, StoreLog 'WriteMode)
-readWriteStoreLog s@(ReadStoreLog f _) = do
-  qs <- readQueues s
-  closeStoreLog s
-  s' <- openWriteStoreLog f
-  writeQueues s' qs
-  pure (qs, s')
+readWriteStoreLog :: FilePath -> IO (Map RecipientId QueueRec, StoreLog 'WriteMode)
+readWriteStoreLog f = do
+  qs <- ifM (doesFileExist f) (readQueues f) (pure M.empty)
+  s <- openWriteStoreLog f
+  writeQueues s qs
+  pure (qs, s)
 
 writeQueues :: StoreLog 'WriteMode -> Map RecipientId QueueRec -> IO ()
-writeQueues s = mapM_ (writeStoreLogRecord s . CreateQueue) . M.filter active
+writeQueues s = mapM_ $ \q -> when (active q) $ logCreateQueue s q
   where
     active QueueRec {status} = status == QueueActive
 
-type LogParsingError = (String, ByteString)
-
-readQueues :: StoreLog 'ReadMode -> IO (Map RecipientId QueueRec)
-readQueues (ReadStoreLog _ h) = LB.hGetContents h >>= returnResult . procStoreLog
+readQueues :: FilePath -> IO (Map RecipientId QueueRec)
+readQueues f = foldM processLine M.empty . B.lines =<< B.readFile f
   where
-    procStoreLog :: LB.ByteString -> ([LogParsingError], Map RecipientId QueueRec)
-    procStoreLog = second (foldl' procLogRecord M.empty) . partitionEithers . map parseLogRecord . LB.lines
-    returnResult :: ([LogParsingError], Map RecipientId QueueRec) -> IO (Map RecipientId QueueRec)
-    returnResult (errs, res) = mapM_ printError errs $> res
-    parseLogRecord :: LB.ByteString -> Either LogParsingError StoreLogRecord
-    parseLogRecord = (\s -> first (,s) $ strDecode s) . trimCR . LB.toStrict
-    procLogRecord :: Map RecipientId QueueRec -> StoreLogRecord -> Map RecipientId QueueRec
-    procLogRecord m = \case
-      CreateQueue q -> M.insert (recipientId q) q m
-      SecureQueue qId sKey -> M.adjust (\q -> q {senderKey = Just sKey}) qId m
-      AddNotifier qId ntfCreds -> M.adjust (\q -> q {notifier = Just ntfCreds}) qId m
-      SuspendQueue qId -> M.adjust (\q -> q {status = QueueOff}) qId m
-      DeleteQueue qId -> M.delete qId m
-      DeleteNotifier qId -> M.adjust (\q -> q {notifier = Nothing}) qId m
-    printError :: LogParsingError -> IO ()
-    printError (e, s) = B.putStrLn $ "Error parsing log: " <> B.pack e <> " - " <> s
+    processLine :: Map RecipientId QueueRec -> ByteString -> IO (Map RecipientId QueueRec)
+    processLine m s = case strDecode $ trimCR s of
+      Right r -> pure $ procLogRecord r
+      Left e -> printError e $> m
+      where
+        procLogRecord :: StoreLogRecord -> Map RecipientId QueueRec
+        procLogRecord = \case
+          CreateQueue q -> M.insert (recipientId q) q m
+          SecureQueue qId sKey -> M.adjust (\q -> q {senderKey = Just sKey}) qId m
+          AddNotifier qId ntfCreds -> M.adjust (\q -> q {notifier = Just ntfCreds}) qId m
+          SuspendQueue qId -> M.adjust (\q -> q {status = QueueOff}) qId m
+          DeleteQueue qId -> M.delete qId m
+          DeleteNotifier qId -> M.adjust (\q -> q {notifier = Nothing}) qId m
+        printError :: String -> IO ()
+        printError e = B.putStrLn $ "Error parsing log: " <> B.pack e <> " - " <> s

src/Simplex/Messaging/Transport.hs

@@ -29,6 +29,7 @@ module Simplex.Messaging.Transport
     supportedSMPServerVRange,
     simplexMQVersion,
     smpBlockSize,
+    TransportConfig (..),
 
     -- * Transport connection class
     Transport (..),
@@ -104,6 +105,11 @@ simplexMQVersion = showVersion SMQ.version
 
 -- * Transport connection class
 
+data TransportConfig = TransportConfig
+  { logTLSErrors :: Bool,
+    transportTimeout :: Maybe Int
+  }
+
 class Transport c where
   transport :: ATransport
   transport = ATransport (TProxy @c)
@@ -112,11 +118,13 @@ class Transport c where
 
   transportPeer :: c -> TransportPeer
 
+  transportConfig :: c -> TransportConfig
+
   -- | Upgrade server TLS context to connection (used in the server)
-  getServerConnection :: T.Context -> IO c
+  getServerConnection :: TransportConfig -> T.Context -> IO c
 
   -- | Upgrade client TLS context to connection (used in the client)
-  getClientConnection :: T.Context -> IO c
+  getClientConnection :: TransportConfig -> T.Context -> IO c
 
   -- | tls-unique channel binding per RFC5929
   tlsUnique :: c -> SessionId
@@ -150,24 +158,25 @@ data TLS = TLS
   { tlsContext :: T.Context,
     tlsPeer :: TransportPeer,
     tlsUniq :: ByteString,
-    tlsBuffer :: TBuffer
+    tlsBuffer :: TBuffer,
+    tlsTransportConfig :: TransportConfig
   }
 
-connectTLS :: T.TLSParams p => Maybe HostName -> Bool -> p -> Socket -> IO T.Context
-connectTLS host_ logErrors params sock =
+connectTLS :: T.TLSParams p => Maybe HostName -> TransportConfig -> p -> Socket -> IO T.Context
+connectTLS host_ TransportConfig {logTLSErrors} params sock =
   E.bracketOnError (T.contextNew sock params) closeTLS $ \ctx ->
     logHandshakeErrors (T.handshake ctx) $> ctx
   where
-    logHandshakeErrors = if logErrors then (`catchAll` logThrow) else id
+    logHandshakeErrors = if logTLSErrors then (`catchAll` logThrow) else id
     logThrow e = putStrLn ("TLS error" <> host <> ": " <> show e) >> E.throwIO e
     host = maybe "" (\h -> " (" <> h <> ")") host_
 
-getTLS :: TransportPeer -> T.Context -> IO TLS
-getTLS tlsPeer cxt = withTlsUnique tlsPeer cxt newTLS
+getTLS :: TransportPeer -> TransportConfig -> T.Context -> IO TLS
+getTLS tlsPeer cfg cxt = withTlsUnique tlsPeer cxt newTLS
   where
     newTLS tlsUniq = do
       tlsBuffer <- atomically newTBuffer
-      pure TLS {tlsContext = cxt, tlsPeer, tlsUniq, tlsBuffer}
+      pure TLS {tlsContext = cxt, tlsTransportConfig = cfg, tlsPeer, tlsUniq, tlsBuffer}
 
 withTlsUnique :: TransportPeer -> T.Context -> (ByteString -> IO c) -> IO c
 withTlsUnique peer cxt f =
@@ -199,6 +208,7 @@ supportedParameters =
 instance Transport TLS where
   transportName _ = "TLS"
   transportPeer = tlsPeer
+  transportConfig = tlsTransportConfig
   getServerConnection = getTLS TServer
   getClientConnection = getTLS TClient
   tlsUnique = tlsUniq
@@ -207,10 +217,12 @@ instance Transport TLS where
   -- https://hackage.haskell.org/package/tls-1.6.0/docs/Network-TLS.html#v:recvData
   -- this function may return less than requested number of bytes
   cGet :: TLS -> Int -> IO ByteString
-  cGet TLS {tlsContext, tlsBuffer} n = getBuffered tlsBuffer n (T.recvData tlsContext)
-
+  cGet TLS {tlsContext, tlsBuffer, tlsTransportConfig = TransportConfig {transportTimeout = t_}} n =
+    getBuffered tlsBuffer n t_ (T.recvData tlsContext)
+      
   cPut :: TLS -> ByteString -> IO ()
-  cPut tls = T.sendData (tlsContext tls) . BL.fromStrict
+  cPut TLS {tlsContext, tlsTransportConfig = TransportConfig {transportTimeout = t_}} s =
+    withTimedErr t_ . T.sendData tlsContext $ BL.fromStrict s
 
   getLn :: TLS -> IO ByteString
   getLn TLS {tlsContext, tlsBuffer} = do

src/Simplex/Messaging/Transport/Buffer.hs

@@ -8,6 +8,8 @@ import Control.Concurrent.STM
 import qualified Control.Exception as E
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
+import System.Timeout (timeout)
+import GHC.IO.Exception (ioException, IOException (..), IOErrorType (..))
 
 data TBuffer = TBuffer
   { buffer :: TVar ByteString,
@@ -26,22 +28,33 @@ withBufferLock TBuffer {getLock} =
     (atomically $ takeTMVar getLock)
     (atomically $ putTMVar getLock ())
 
-getBuffered :: TBuffer -> Int -> IO ByteString -> IO ByteString
-getBuffered tb@TBuffer {buffer} n getChunk = withBufferLock tb $ do
-  b <- readChunks =<< readTVarIO buffer
+getBuffered :: TBuffer -> Int -> Maybe Int -> IO ByteString -> IO ByteString
+getBuffered tb@TBuffer {buffer} n t_ getChunk = withBufferLock tb $ do
+  b <- readChunks True =<< readTVarIO buffer
   let (s, b') = B.splitAt n b
   atomically $ writeTVar buffer $! b'
   -- This would prevent the need to pad auth tag in HTTP2
   -- threadDelay 150
   pure s
   where
-    readChunks :: ByteString -> IO ByteString
-    readChunks b
+    readChunks :: Bool -> ByteString -> IO ByteString
+    readChunks firstChunk b
       | B.length b >= n = pure b
       | otherwise =
-        getChunk >>= \case
+        get >>= \case
           "" -> pure b
-          s -> readChunks $ b <> s
+          s -> readChunks False $ b <> s
+      where
+        get
+          | firstChunk = getChunk
+          | otherwise = withTimedErr t_ getChunk
+
+withTimedErr :: Maybe Int -> IO a -> IO a
+withTimedErr t_ a = case t_ of
+  Just t -> timeout t a >>= maybe err pure
+  Nothing -> a
+  where
+    err = ioException (IOError Nothing TimeExpired "" "get timeout" Nothing Nothing)
 
 -- This function is only used in test and needs to be improved before it can be used in production,
 -- it will never complete if TLS connection is closed before there is newline.

src/Simplex/Messaging/Transport/Client.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE FlexibleInstances #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
@@ -114,12 +115,16 @@ data TransportClientConfig = TransportClientConfig
 defaultTransportClientConfig :: TransportClientConfig
 defaultTransportClientConfig = TransportClientConfig Nothing (Just defaultKeepAliveOpts) True
 
+clientTransportConfig :: TransportClientConfig -> TransportConfig
+clientTransportConfig TransportClientConfig {logTLSErrors} =
+  TransportConfig {logTLSErrors, transportTimeout = Nothing}
+
 -- | Connect to passed TCP host:port and pass handle to the client.
 runTransportClient :: (Transport c, MonadUnliftIO m) => TransportClientConfig -> Maybe ByteString -> TransportHost -> ServiceName -> Maybe C.KeyHash -> (c -> m a) -> m a
 runTransportClient = runTLSTransportClient supportedParameters Nothing
 
 runTLSTransportClient :: (Transport c, MonadUnliftIO m) => T.Supported -> Maybe XS.CertificateStore -> TransportClientConfig -> Maybe ByteString -> TransportHost -> ServiceName -> Maybe C.KeyHash -> (c -> m a) -> m a
-runTLSTransportClient tlsParams caStore_ TransportClientConfig {socksProxy, tcpKeepAlive, logTLSErrors} proxyUsername host port keyHash client = do
+runTLSTransportClient tlsParams caStore_ cfg@TransportClientConfig {socksProxy, tcpKeepAlive} proxyUsername host port keyHash client = do
   let hostName = B.unpack $ strEncode host
       clientParams = mkTLSClientParams tlsParams caStore_ hostName port keyHash
       connectTCP = case socksProxy of
@@ -128,7 +133,8 @@ runTLSTransportClient tlsParams caStore_ TransportClientConfig {socksProxy, tcpK
   c <- liftIO $ do
     sock <- connectTCP port
     mapM_ (setSocketKeepAlive sock) tcpKeepAlive
-    connectTLS (Just hostName) logTLSErrors clientParams sock >>= getClientConnection
+    let tCfg = clientTransportConfig cfg
+    connectTLS (Just hostName) tCfg clientParams sock >>= getClientConnection tCfg
   client c `E.finally` liftIO (closeConnection c)
   where
     hostAddr = \case

src/Simplex/Messaging/Transport/HTTP2.hs

@@ -73,7 +73,7 @@ instance HTTP2BodyChunk HS.Request where
 getHTTP2Body :: HTTP2BodyChunk a => a -> Int -> IO HTTP2Body
 getHTTP2Body r n = do
   bodyBuffer <- atomically newTBuffer
-  let getPart n' = getBuffered bodyBuffer n' $ getBodyChunk r
+  let getPart n' = getBuffered bodyBuffer n' Nothing $ getBodyChunk r
   bodyHead <- getPart n
   let bodySize = fromMaybe 0 $ getBodySize r
       -- TODO check bodySize once it is set

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

@@ -14,7 +14,7 @@ import qualified Network.TLS as T
 import Numeric.Natural (Natural)
 import Simplex.Messaging.Transport (SessionId)
 import Simplex.Messaging.Transport.HTTP2
-import Simplex.Messaging.Transport.Server (loadSupportedTLSServerParams, runTransportServer)
+import Simplex.Messaging.Transport.Server (TransportServerConfig (..), loadSupportedTLSServerParams, runTransportServer)
 
 type HTTP2ServerFunc = SessionId -> Request -> (Response -> IO ()) -> IO ()
 
@@ -27,7 +27,7 @@ data HTTP2ServerConfig = HTTP2ServerConfig
     caCertificateFile :: FilePath,
     privateKeyFile :: FilePath,
     certificateFile :: FilePath,
-    logTLSErrors :: Bool
+    transportConfig :: TransportServerConfig
   }
   deriving (Show)
 
@@ -45,12 +45,12 @@ data HTTP2Server = HTTP2Server
 
 -- This server is for testing only, it processes all requests in a single queue.
 getHTTP2Server :: HTTP2ServerConfig -> IO HTTP2Server
-getHTTP2Server HTTP2ServerConfig {qSize, http2Port, bufferSize, bodyHeadSize, serverSupported, caCertificateFile, certificateFile, privateKeyFile, logTLSErrors} = do
+getHTTP2Server HTTP2ServerConfig {qSize, http2Port, bufferSize, bodyHeadSize, serverSupported, caCertificateFile, certificateFile, privateKeyFile, transportConfig} = do
   tlsServerParams <- loadSupportedTLSServerParams serverSupported caCertificateFile certificateFile privateKeyFile
   started <- newEmptyTMVarIO
   reqQ <- newTBQueueIO qSize
   action <- async $
-    runHTTP2Server started http2Port bufferSize tlsServerParams logTLSErrors $ \sessionId r sendResponse -> do
+    runHTTP2Server started http2Port bufferSize tlsServerParams transportConfig $ \sessionId r sendResponse -> do
       reqBody <- getHTTP2Body r bodyHeadSize
       atomically $ writeTBQueue reqQ HTTP2Request {sessionId, request = r, reqBody, sendResponse}
   void . atomically $ takeTMVar started
@@ -59,8 +59,8 @@ getHTTP2Server HTTP2ServerConfig {qSize, http2Port, bufferSize, bodyHeadSize, se
 closeHTTP2Server :: HTTP2Server -> IO ()
 closeHTTP2Server = uninterruptibleCancel . action
 
-runHTTP2Server :: TMVar Bool -> ServiceName -> BufferSize -> T.ServerParams -> Bool -> HTTP2ServerFunc -> IO ()
-runHTTP2Server started port bufferSize serverParams logTLSErrors http2Server =
-  runTransportServer started port serverParams logTLSErrors $ withHTTP2 bufferSize run
+runHTTP2Server :: TMVar Bool -> ServiceName -> BufferSize -> T.ServerParams -> TransportServerConfig -> HTTP2ServerFunc -> IO ()
+runHTTP2Server started port bufferSize serverParams transportConfig http2Server =
+  runTransportServer started port serverParams transportConfig $ withHTTP2 bufferSize run
   where
     run cfg sessId = H.run cfg $ \req _aux sendResp -> http2Server sessId req (`sendResp` [])

src/Simplex/Messaging/Transport/Server.hs

@@ -7,6 +7,8 @@
 module Simplex.Messaging.Transport.Server
   ( runTransportServer,
     runTCPServer,
+    TransportServerConfig (..),
+    defaultTransportServerConfig,
     loadSupportedTLSServerParams,
     loadTLSServerParams,
     loadFingerprint,
@@ -38,15 +40,33 @@ import UnliftIO.Concurrent
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
 
+data TransportServerConfig = TransportServerConfig
+  { logTLSErrors :: Bool,
+    transportTimeout :: Int
+  }
+  deriving (Eq, Show)
+
+defaultTransportServerConfig :: TransportServerConfig
+defaultTransportServerConfig = TransportServerConfig
+  { logTLSErrors = True,
+    transportTimeout = 40000000
+  }
+
+serverTransportConfig :: TransportServerConfig -> TransportConfig
+serverTransportConfig TransportServerConfig {logTLSErrors} =
+  -- TransportConfig {logTLSErrors, transportTimeout = Just transportTimeout}
+  TransportConfig {logTLSErrors, transportTimeout = Nothing}
+
 -- | Run transport server (plain TCP or WebSockets) on passed TCP port and signal when server started and stopped via passed TMVar.
 --
 -- All accepted connections are passed to the passed function.
-runTransportServer :: forall c m. (Transport c, MonadUnliftIO m) => TMVar Bool -> ServiceName -> T.ServerParams -> Bool -> (c -> m ()) -> m ()
-runTransportServer started port serverParams logTLSErrors server = do
+runTransportServer :: forall c m. (Transport c, MonadUnliftIO m) => TMVar Bool -> ServiceName -> T.ServerParams -> TransportServerConfig -> (c -> m ()) -> m ()
+runTransportServer started port serverParams cfg server = do
   u <- askUnliftIO
+  let tCfg = serverTransportConfig cfg
   liftIO . runTCPServer started port $ \conn ->
     E.bracket
-      (connectTLS Nothing logTLSErrors serverParams conn >>= getServerConnection)
+      (connectTLS Nothing tCfg serverParams conn >>= getServerConnection tCfg)
       closeConnection
       (unliftIO u . server)
 

src/Simplex/Messaging/Transport/WebSockets.hs

@@ -17,6 +17,7 @@ import Simplex.Messaging.Transport
     Transport (..),
     TransportError (..),
     TransportPeer (..),
+    TransportConfig (..),
     closeTLS,
     smpBlockSize,
     withTlsUnique,
@@ -27,7 +28,8 @@ data WS = WS
   { wsPeer :: TransportPeer,
     tlsUniq :: ByteString,
     wsStream :: Stream,
-    wsConnection :: Connection
+    wsConnection :: Connection,
+    wsTransportConfig :: TransportConfig
   }
 
 websocketsOpts :: ConnectionOptions
@@ -45,10 +47,13 @@ instance Transport WS where
   transportPeer :: WS -> TransportPeer
   transportPeer = wsPeer
 
-  getServerConnection :: T.Context -> IO WS
+  transportConfig :: WS -> TransportConfig
+  transportConfig = wsTransportConfig
+
+  getServerConnection :: TransportConfig -> T.Context -> IO WS
   getServerConnection = getWS TServer
 
-  getClientConnection :: T.Context -> IO WS
+  getClientConnection :: TransportConfig -> T.Context -> IO WS
   getClientConnection = getWS TClient
 
   tlsUnique :: WS -> ByteString
@@ -74,13 +79,13 @@ instance Transport WS where
       then E.throwIO TEBadBlock
       else pure $ B.init s
 
-getWS :: TransportPeer -> T.Context -> IO WS
-getWS wsPeer cxt = withTlsUnique wsPeer cxt connectWS
+getWS :: TransportPeer -> TransportConfig -> T.Context -> IO WS
+getWS wsPeer cfg cxt = withTlsUnique wsPeer cxt connectWS
   where
     connectWS tlsUniq = do
       s <- makeTLSContextStream cxt
       wsConnection <- connectPeer wsPeer s
-      pure $ WS {wsPeer, tlsUniq, wsStream = s, wsConnection}
+      pure $ WS {wsPeer, tlsUniq, wsStream = s, wsConnection, wsTransportConfig = cfg}
     connectPeer :: TransportPeer -> Stream -> IO Connection
     connectPeer TServer = acceptClientRequest
     connectPeer TClient = sendClientRequest

src/Simplex/Messaging/Util.hs

@@ -1,4 +1,3 @@
-{-# LANGUAGE NumericUnderscores #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 
@@ -13,11 +12,15 @@ import Data.Bifunctor (first)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Int (Int64)
+import Data.List (groupBy, sortOn)
+import Data.List.NonEmpty (NonEmpty)
+import qualified Data.List.NonEmpty as L
 import Data.Text (Text)
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeUtf8With)
-import Data.Time (NominalDiffTime, nominalDiffTimeToSeconds)
+import Data.Time (NominalDiffTime)
 import UnliftIO.Async
+import qualified UnliftIO.Exception as UE
 
 raceAny_ :: MonadUnliftIO m => [m a] -> m ()
 raceAny_ = r []
@@ -98,10 +101,44 @@ catchAll_ :: IO a -> IO a -> IO a
 catchAll_ a = catchAll a . const
 {-# INLINE catchAll_ #-}
 
+tryAllErrors :: (MonadUnliftIO m, MonadError e m) => (E.SomeException -> e) -> m a -> m (Either e a)
+tryAllErrors err action = tryError action `UE.catch` (pure . Left . err)
+{-# INLINE tryAllErrors #-}
+
+catchAllErrors :: (MonadUnliftIO m, MonadError e m) => (E.SomeException -> e) -> m a -> (e -> m a) -> m a
+catchAllErrors err action handle = tryAllErrors err action >>= either handle pure
+{-# INLINE catchAllErrors #-}
+
+catchThrow :: (MonadUnliftIO m, MonadError e m) => m a -> (E.SomeException -> e) -> m a
+catchThrow action err  = catchAllErrors err action throwError
+{-# INLINE catchThrow #-}
+
+allFinally :: (MonadUnliftIO m, MonadError e m) => (E.SomeException -> e) -> m a -> m b -> m a
+allFinally err action final = tryAllErrors err action >>= \r -> final >> either throwError pure r
+{-# INLINE allFinally #-}
+
 eitherToMaybe :: Either a b -> Maybe b
 eitherToMaybe = either (const Nothing) Just
 {-# INLINE eitherToMaybe #-}
 
+groupOn :: Eq k => (a -> k) -> [a] -> [[a]]
+groupOn = groupBy . eqOn
+  -- it is equivalent to groupBy ((==) `on` f),
+  -- but it redefines `on` to avoid duplicate computation for most values.
+  -- source: https://hackage.haskell.org/package/extra-1.7.13/docs/src/Data.List.Extra.html#groupOn
+  -- the on2 in this package is specialized to only use `==` as the function, `eqOn f` is equivalent to `(==) `on` f`
+  where
+    eqOn f = \x -> let fx = f x in \y -> fx == f y
+
+groupAllOn :: Ord k => (a -> k) -> [a] -> [[a]]
+groupAllOn f = groupOn f . sortOn f
+
+toChunks :: Int -> [a] -> [NonEmpty a]
+toChunks _ [] = []
+toChunks n xs =
+  let (ys, xs') = splitAt n xs
+  in maybe id (:) (L.nonEmpty ys) (toChunks n xs')
+
 safeDecodeUtf8 :: ByteString -> Text
 safeDecodeUtf8 = decodeUtf8With onError
   where

tests/AgentTests/ConnectionRequestTests.hs

@@ -70,8 +70,8 @@ connectionRequest =
   ACR SCMInvitation $
     CRInvitationUri connReqData testE2ERatchetParams
 
-connectionRequest12 :: AConnectionRequestUri
-connectionRequest12 =
+connectionRequestCurrentRange :: AConnectionRequestUri
+connectionRequestCurrentRange =
   ACR SCMInvitation $
     CRInvitationUri
       connReqData {crAgentVRange = supportedSMPAgentVRange, crSmpQueues = [queueV1, queueV1]}
@@ -113,8 +113,8 @@ connectionRequestTests =
         `shouldBe` "https://simplex.chat/invitation#/?v=1&smp=smp%3A%2F%2F1234-w%3D%3D%40smp.simplex.im%3A5223%2F3456-w%3D%3D%23%2F%3Fv%3D1%26dh%3D"
         <> urlEncode True testDhKeyStrUri
         <> "&e2e=v%3D1%26x3dh%3DMEIwBQYDK2VvAzkAmKuSYeQ_m0SixPDS8Wq8VBaTS1cW-Lp0n0h4Diu-kUpR-qXx4SDJ32YGEFoGFGSbGPry5Ychr6U%3D%2CMEIwBQYDK2VvAzkAmKuSYeQ_m0SixPDS8Wq8VBaTS1cW-Lp0n0h4Diu-kUpR-qXx4SDJ32YGEFoGFGSbGPry5Ychr6U%3D"
-      strEncode connectionRequest12
-        `shouldBe` "https://simplex.chat/invitation#/?v=1-2&smp=smp%3A%2F%2F1234-w%3D%3D%40smp.simplex.im%3A5223%2F3456-w%3D%3D%23%2F%3Fv%3D1%26dh%3D"
+      strEncode connectionRequestCurrentRange
+        `shouldBe` "https://simplex.chat/invitation#/?v=1-4&smp=smp%3A%2F%2F1234-w%3D%3D%40smp.simplex.im%3A5223%2F3456-w%3D%3D%23%2F%3Fv%3D1%26dh%3D"
         <> urlEncode True testDhKeyStrUri
         <> "%2Csmp%3A%2F%2F1234-w%3D%3D%40smp.simplex.im%3A5223%2F3456-w%3D%3D%23%2F%3Fv%3D1%26dh%3D"
         <> urlEncode True testDhKeyStrUri
@@ -158,9 +158,9 @@ connectionRequestTests =
             <> testDhKeyStrUri
             <> "&e2e=extra_key%3Dnew%26v%3D1-2%26x3dh%3DMEIwBQYDK2VvAzkAmKuSYeQ_m0SixPDS8Wq8VBaTS1cW-Lp0n0h4Diu-kUpR-qXx4SDJ32YGEFoGFGSbGPry5Ychr6U%3D%2CMEIwBQYDK2VvAzkAmKuSYeQ_m0SixPDS8Wq8VBaTS1cW-Lp0n0h4Diu-kUpR-qXx4SDJ32YGEFoGFGSbGPry5Ychr6U%3D"
             <> "&some_new_param=abc"
-            <> "&v=1-2"
+            <> "&v=1-4"
         )
-        `shouldBe` Right connectionRequest12
+        `shouldBe` Right connectionRequestCurrentRange
       strDecode
         ( "https://simplex.chat/invitation#/?smp=smp%3A%2F%2F1234-w%3D%3D%40smp.simplex.im%3A5223%2F3456-w%3D%3D%23%2F%3Fv%3D1%26dh%3D"
             <> testDhKeyStrUri

tests/AgentTests/NotificationTests.hs

@@ -19,8 +19,8 @@ import qualified Data.ByteString.Base64.URL as U
 import Data.ByteString.Char8 (ByteString)
 import Data.Text.Encoding (encodeUtf8)
 import NtfClient
-import SMPAgentClient (agentCfg, initAgentServers, testDB, testDB2)
-import SMPClient (testPort, withSmpServer, withSmpServerStoreLogOn, xit')
+import SMPAgentClient (agentCfg, initAgentServers, initAgentServers2, testDB, testDB2)
+import SMPClient (cfg, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerStoreLogOn, xit')
 import Simplex.Messaging.Agent
 import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..), InitialAgentServers)
 import Simplex.Messaging.Agent.Protocol
@@ -31,6 +31,7 @@ import Simplex.Messaging.Notifications.Server.Push.APNS
 import Simplex.Messaging.Notifications.Types (NtfToken (..))
 import Simplex.Messaging.Protocol (ErrorType (AUTH), MsgFlags (MsgFlags), SMPMsgMeta (..))
 import qualified Simplex.Messaging.Protocol as SMP
+import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Transport (ATransport)
 import Simplex.Messaging.Util (tryE)
 import System.Directory (doesFileExist, removeFile)
@@ -85,6 +86,10 @@ notificationTests t =
       it "should resume subscriptions after SMP server is restarted" $ \_ ->
         withAPNSMockServer $ \apns ->
           withNtfServer t $ testNotificationsSMPRestart t apns
+    describe "Notifications after SMP server restart" $
+      it "should resume batched subscriptions after SMP server is restarted" $ \_ ->
+        withAPNSMockServer $ \apns ->
+          withNtfServer t $ testNotificationsSMPRestartBatch 100 t apns
     describe "should switch notifications to the new queue" $
       testServerMatrix2 t $ \servers ->
         withAPNSMockServer $ \apns ->
@@ -248,7 +253,7 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} = do
 
   runRight_ $ do
     get alice =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False
-    ackMessage alice bobId $ baseId + 1
+    ackMessage alice bobId (baseId + 1) Nothing
     -- delete notification subscription
     toggleConnectionNtfs alice bobId False
     liftIO $ threadDelay 250000
@@ -291,13 +296,13 @@ testNotificationSubscriptionNewConnection APNSMockServer {apnsQ} = do
     get bob ##> ("", aliceId, SENT $ baseId + 1)
     void $ messageNotification apnsQ
     get alice =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False
-    ackMessage alice bobId $ baseId + 1
+    ackMessage alice bobId (baseId + 1) Nothing
     -- alice sends message
     2 <- msgId <$> sendMessage alice bobId (SMP.MsgFlags True) "hey there"
     get alice ##> ("", bobId, SENT $ baseId + 2)
     void $ messageNotification apnsQ
     get bob =##> \case ("", c, Msg "hey there") -> c == aliceId; _ -> False
-    ackMessage bob aliceId $ baseId + 2
+    ackMessage bob aliceId (baseId + 2) Nothing
     -- no unexpected notifications should follow
     noNotification apnsQ
   where
@@ -338,7 +343,7 @@ testChangeNotificationsMode APNSMockServer {apnsQ} = do
     get bob ##> ("", aliceId, SENT $ baseId + 1)
     void $ messageNotification apnsQ
     get alice =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False
-    ackMessage alice bobId $ baseId + 1
+    ackMessage alice bobId (baseId + 1) Nothing
     -- set mode to NMPeriodic
     NTActive <- registerNtfToken alice tkn NMPeriodic
     -- send message, no notification
@@ -347,7 +352,7 @@ testChangeNotificationsMode APNSMockServer {apnsQ} = do
     get bob ##> ("", aliceId, SENT $ baseId + 2)
     noNotification apnsQ
     get alice =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False
-    ackMessage alice bobId $ baseId + 2
+    ackMessage alice bobId (baseId + 2) Nothing
     -- set mode to NMInstant
     NTActive <- registerNtfToken alice tkn NMInstant
     -- send message, receive notification
@@ -356,7 +361,7 @@ testChangeNotificationsMode APNSMockServer {apnsQ} = do
     get bob ##> ("", aliceId, SENT $ baseId + 3)
     void $ messageNotification apnsQ
     get alice =##> \case ("", c, Msg "hello there") -> c == bobId; _ -> False
-    ackMessage alice bobId $ baseId + 3
+    ackMessage alice bobId (baseId + 3) Nothing
     -- turn off notifications
     deleteNtfToken alice tkn
     -- send message, no notification
@@ -365,7 +370,7 @@ testChangeNotificationsMode APNSMockServer {apnsQ} = do
     get bob ##> ("", aliceId, SENT $ baseId + 4)
     noNotification apnsQ
     get alice =##> \case ("", c, Msg "why hello there") -> c == bobId; _ -> False
-    ackMessage alice bobId $ baseId + 4
+    ackMessage alice bobId (baseId + 4) Nothing
     -- turn on notifications, set mode to NMInstant
     void $ registerTestToken alice "abcd" NMInstant apnsQ
     -- send message, receive notification
@@ -374,7 +379,7 @@ testChangeNotificationsMode APNSMockServer {apnsQ} = do
     get bob ##> ("", aliceId, SENT $ baseId + 5)
     void $ messageNotification apnsQ
     get alice =##> \case ("", c, Msg "hey") -> c == bobId; _ -> False
-    ackMessage alice bobId $ baseId + 5
+    ackMessage alice bobId (baseId + 5) Nothing
     -- no notifications should follow
     noNotification apnsQ
   where
@@ -402,7 +407,7 @@ testChangeToken APNSMockServer {apnsQ} = do
     get bob ##> ("", aliceId, SENT $ baseId + 1)
     void $ messageNotification apnsQ
     get alice =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False
-    ackMessage alice bobId $ baseId + 1
+    ackMessage alice bobId (baseId + 1) Nothing
     pure (aliceId, bobId)
   disconnectAgentClient alice
 
@@ -417,7 +422,7 @@ testChangeToken APNSMockServer {apnsQ} = do
     get bob ##> ("", aliceId, SENT $ baseId + 2)
     void $ messageNotification apnsQ
     get alice1 =##> \case ("", c, Msg "hello there") -> c == bobId; _ -> False
-    ackMessage alice1 bobId $ baseId + 2
+    ackMessage alice1 bobId (baseId + 2) Nothing
     -- no notifications should follow
     noNotification apnsQ
   where
@@ -436,7 +441,7 @@ testNotificationsStoreLog t APNSMockServer {apnsQ} = do
     get bob ##> ("", aliceId, SENT 4)
     void $ messageNotification apnsQ
     get alice =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False
-    ackMessage alice bobId 4
+    ackMessage alice bobId 4 Nothing
     liftIO $ killThread threadId
     pure (aliceId, bobId)
 
@@ -462,7 +467,7 @@ testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
     get bob ##> ("", aliceId, SENT 4)
     void $ messageNotification apnsQ
     get alice =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False
-    ackMessage alice bobId 4
+    ackMessage alice bobId 4 Nothing
     liftIO $ killThread threadId
     pure (aliceId, bobId)
 
@@ -480,6 +485,52 @@ testNotificationsSMPRestart t APNSMockServer {apnsQ} = do
     get alice =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False
     liftIO $ killThread threadId
 
+testNotificationsSMPRestartBatch :: Int -> ATransport -> APNSMockServer -> IO ()
+testNotificationsSMPRestartBatch n t APNSMockServer {apnsQ} = do
+  a <- getSMPAgentClient' agentCfg initAgentServers2 testDB
+  b <- getSMPAgentClient' agentCfg initAgentServers2 testDB2
+  conns <- runServers $ do
+    conns <- forM [1 .. n :: Int] . const $ makeConnection a b
+    _ <- registerTestToken a "abcd" NMInstant apnsQ
+    liftIO $ threadDelay 1500000
+    forM_ conns $ \(aliceId, bobId) -> do
+      msgId <- sendMessage b aliceId (SMP.MsgFlags True) "hello"
+      get b ##> ("", aliceId, SENT msgId)
+      void $ messageNotification apnsQ
+      get a =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False
+      ackMessage a bobId msgId Nothing
+    pure conns
+
+  runRight_ @AgentErrorType $ do
+    ("", "", DOWN _ bcs1) <- nGet a
+    ("", "", DOWN _ bcs2) <- nGet a
+    liftIO $ length (bcs1 <> bcs2) `shouldBe` length conns
+    ("", "", DOWN _ acs1) <- nGet b
+    ("", "", DOWN _ acs2) <- nGet b
+    liftIO $ length (acs1 <> acs2) `shouldBe` length conns
+
+  runServers $ do
+    ("", "", UP _ bcs1) <- nGet a
+    ("", "", UP _ bcs2) <- nGet a
+    liftIO $ length (bcs1 <> bcs2) `shouldBe` length conns
+    ("", "", UP _ acs1) <- nGet b
+    ("", "", UP _ acs2) <- nGet b
+    liftIO $ length (acs1 <> acs2) `shouldBe` length conns
+    liftIO $ threadDelay 1500000
+    forM_ conns $ \(aliceId, bobId) -> do
+      msgId <- sendMessage b aliceId (SMP.MsgFlags True) "hello again"
+      get b ##> ("", aliceId, SENT msgId)
+      _ <- messageNotificationData a apnsQ
+      get a =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False
+  where
+    runServers :: ExceptT AgentErrorType IO a -> IO a
+    runServers a = do
+      withSmpServerStoreLogOn t testPort $ \t1 -> do
+        res <- withSmpServerConfigOn t cfg {storeLogFile = Just testStoreLogFile2} testPort2 $ \t2 ->
+          runRight a `finally` killThread t2
+        killThread t1
+        pure res
+
 testSwitchNotifications :: InitialAgentServers -> APNSMockServer -> IO ()
 testSwitchNotifications servers APNSMockServer {apnsQ} = do
   a <- getSMPAgentClient' agentCfg servers testDB
@@ -494,9 +545,9 @@ testSwitchNotifications servers APNSMockServer {apnsQ} = do
           get b ##> ("", aId, SENT msgId)
           void $ messageNotification apnsQ
           get a =##> \case ("", c, Msg msg') -> c == bId && msg == msg'; _ -> False
-          ackMessage a bId msgId
+          ackMessage a bId msgId Nothing
     testMessage "hello"
-    switchConnectionAsync a "" bId
+    _ <- switchConnectionAsync a "" bId
     switchComplete a bId b aId
     liftIO $ threadDelay 500000
     testMessage "hello again"

tests/AgentTests/SQLiteTests.hs

@@ -140,7 +140,7 @@ testForeignKeysEnabled =
       `shouldThrow` (\e -> DB.sqlError e == DB.ErrorConstraint)
 
 cData1 :: ConnData
-cData1 = ConnData {userId = 1, connId = "conn1", connAgentVersion = 1, enableNtfs = True, duplexHandshake = Nothing, deleted = False}
+cData1 = ConnData {userId = 1, connId = "conn1", connAgentVersion = 1, enableNtfs = True, duplexHandshake = Nothing, lastExternalSndId = 0, deleted = False, ratchetSyncState = RSOk}
 
 testPrivateSignKey :: C.APrivateSignKey
 testPrivateSignKey = C.APrivateSignKey C.SEd25519 "MC4CAQAwBQYDK2VwBCIEIDfEfevydXXfKajz3sRkcQ7RPvfWUPoq6pu1TYHV1DEe"
@@ -167,6 +167,7 @@ rcvQueue1 =
       dbQueueId = 1,
       primary = True,
       dbReplaceQueueId = Nothing,
+      rcvSwchStatus = Nothing,
       smpClientVersion = 1,
       clientNtfCreds = Nothing,
       deleteErrors = 0
@@ -187,6 +188,7 @@ sndQueue1 =
       dbQueueId = 1,
       primary = True,
       dbReplaceQueueId = Nothing,
+      sndSwchStatus = Nothing,
       smpClientVersion = 1
     }
 
@@ -324,6 +326,7 @@ testUpgradeRcvConnToDuplex =
               e2eDhSecret = testDhSecret,
               status = New,
               dbQueueId = 1,
+              sndSwchStatus = Nothing,
               primary = True,
               dbReplaceQueueId = Nothing,
               smpClientVersion = 1
@@ -352,6 +355,7 @@ testUpgradeSndConnToDuplex =
               sndId = "4567",
               status = New,
               dbQueueId = 1,
+              rcvSwchStatus = Nothing,
               primary = True,
               dbReplaceQueueId = Nothing,
               smpClientVersion = 1,

tests/AgentTests/SchemaDump.hs

@@ -50,13 +50,13 @@ testSchemaMigrations = do
       putStrLn $ "down migration " <> name m
       let downMigr = fromJust $ toDownMigration m
       schema <- getSchema testDB testSchema
-      withConnection st (`Migrations.run` MTRUp [m])
+      Migrations.run st $ MTRUp [m]
       schema' <- getSchema testDB testSchema
       schema' `shouldNotBe` schema
-      withConnection st (`Migrations.run` MTRDown [downMigr])
+      Migrations.run st $ MTRDown [downMigr]
       schema'' <- getSchema testDB testSchema
       schema'' `shouldBe` schema
-      withConnection st (`Migrations.run` MTRUp [m])
+      Migrations.run st $ MTRUp [m]
       schema''' <- getSchema testDB testSchema
       schema''' `shouldBe` schema'
 

tests/CoreTests/UtilTests.hs

@@ -0,0 +1,119 @@
+{-# LANGUAGE ScopedTypeVariables #-}
+
+module CoreTests.UtilTests where
+
+import Control.Exception (Exception, SomeException, throwIO)
+import Control.Monad.Except
+import Data.IORef
+import Simplex.Messaging.Util
+import Simplex.Messaging.Client.Agent ()
+import Test.Hspec
+import qualified UnliftIO.Exception as UE
+
+utilTests :: Spec
+utilTests = do
+  describe "problems of lifted try, catch and finally (don't use them)" $ do
+    describe "lifted try" $ do
+      it "does not catch errors" $ do
+        runExceptT (UE.try throwTestError >>= either handleCatch pure) `shouldReturn` Left (TestError "error")
+        runExceptT (UE.try throwTestException >>= either handleCatch pure) `shouldThrow` (\(e :: IOError) -> show e == "user error (error)")
+      it "with SomeException catches all errors but wraps ExceptT errors" $ do
+        runExceptT (UE.try throwTestError >>= either handleException pure) `shouldReturn` Right "caught InternalException {unInternalException = TestError \"error\"}"
+        runExceptT (UE.try throwTestException >>= either handleException pure) `shouldReturn` Right "caught user error (error)"
+    describe "lifted catch" $ do
+      it "does not catch" $ do
+        runExceptT (throwTestError `UE.catch` handleCatch) `shouldReturn` Left (TestError "error")
+        runExceptT (throwTestException `UE.catch` handleCatch) `shouldThrow` (\(e :: IOError) -> show e == "user error (error)")
+      it "with SomeException catches all errors but wraps ExceptT errors" $ do
+        runExceptT (throwTestError `UE.catch` handleException) `shouldReturn` Right "caught InternalException {unInternalException = TestError \"error\"}"
+        runExceptT (throwTestException `UE.catch` handleException) `shouldReturn` Right "caught user error (error)"
+    describe "lifted finally" $ do
+      it "with ExceptT error executes final action and stays in ExceptT monad" $ withFinal $ \final ->
+        runExceptT (throwTestError `UE.finally` final) `shouldReturn` Left (TestError "error")
+      it "with exception executes final action (not always - race condition?) and throws exception" $ withFinal $ \final ->
+        runExceptT (throwTestException `UE.finally` final) `shouldThrow` (\(e :: IOError) -> show e == "user error (error)")
+  describe "problems of tryError and catchError (don't use them)" $ do
+    describe "tryError" $ do
+      it "catches ExceptT errors but not Exceptions" $ do
+        runExceptT (tryError throwTestError >>= either handleCatch pure) `shouldReturn` Right "caught TestError \"error\""
+        runExceptT (tryError throwTestException >>= either handleCatch pure) `shouldThrow` (\(e :: IOError) -> show e == "user error (error)")
+    describe "catchError" $ do
+      it "catches ExceptT errors but not Exceptions" $ do
+        runExceptT (throwTestError `catchError` handleCatch) `shouldReturn` Right "caught TestError \"error\""
+        runExceptT (throwTestException `catchError` handleCatch) `shouldThrow` (\(e :: IOError) -> show e == "user error (error)")
+  describe "tryAllErrors" $ do
+    it "should return ExceptT error as Left" $
+      runExceptT (tryAllErrors testErr throwTestError) `shouldReturn` Right (Left (TestError "error"))
+    it "should return SomeException as Left" $
+      runExceptT (tryAllErrors testErr throwTestException) `shouldReturn` Right (Left (TestException "user error (error)"))
+    it "should return no errors as Right" $
+      runExceptT (tryAllErrors testErr noErrors) `shouldReturn` Right (Right "no errors")
+  describe "tryAllErrors specialized as tryTestError" $ do
+    let tryTestError = tryAllErrors testErr
+    it "should return ExceptT error as Left" $
+      runExceptT (tryTestError throwTestError) `shouldReturn` Right (Left (TestError "error"))
+    it "should return SomeException as Left" $
+      runExceptT (tryTestError throwTestException) `shouldReturn` Right (Left (TestException "user error (error)"))
+    it "should return no errors as Right" $
+      runExceptT (tryTestError noErrors) `shouldReturn` Right (Right "no errors")
+  describe "catchAllErrors" $ do
+    it "should catch ExceptT error" $
+      runExceptT (catchAllErrors testErr throwTestError handleCatch) `shouldReturn` Right "caught TestError \"error\""
+    it "should catch SomeException" $
+      runExceptT (catchAllErrors testErr throwTestException handleCatch) `shouldReturn` Right "caught TestException \"user error (error)\""
+    it "should not throw if there are no errors" $
+      runExceptT (catchAllErrors testErr noErrors throwError) `shouldReturn` Right "no errors"
+  describe "catchAllErrors specialized as catchTestError" $ do
+    let catchTestError = catchAllErrors testErr
+    it "should catch ExceptT error" $
+      runExceptT (throwTestError `catchTestError` handleCatch) `shouldReturn` Right "caught TestError \"error\""
+    it "should catch SomeException" $
+      runExceptT (throwTestException `catchTestError` handleCatch) `shouldReturn` Right "caught TestException \"user error (error)\""
+    it "should not throw if there are no errors" $
+      runExceptT (noErrors `catchTestError` throwError) `shouldReturn` Right "no errors"
+  describe "catchThrow" $ do
+    it "should re-throw ExceptT error" $
+      runExceptT (throwTestError `catchThrow` testErr) `shouldReturn` Left (TestError "error")
+    it "should catch SomeException and throw as ExceptT error" $
+      runExceptT (throwTestException `catchThrow` testErr) `shouldReturn` Left (TestException "user error (error)")
+    it "should not throw if there are no exceptions" $
+      runExceptT (noErrors `catchThrow` testErr) `shouldReturn` Right "no errors"
+  describe "allFinally should run final action" $ do
+    it "then throw ExceptT error" $ withFinal $ \final ->
+      runExceptT (allFinally testErr throwTestError final) `shouldReturn` Left (TestError "error")
+    it "then throw SomeException as ExceptT error" $ withFinal $ \final ->
+      runExceptT (allFinally testErr throwTestException final) `shouldReturn` Left (TestException "user error (error)")
+    it "and should not throw if there are no exceptions" $ withFinal $ \final ->
+      runExceptT (allFinally testErr noErrors final) `shouldReturn` Right "no errors"
+  describe "allFinally specialized as testFinally should run final action" $ do
+    let testFinally = allFinally testErr
+    it "then throw ExceptT error" $ withFinal $ \final ->
+      runExceptT (throwTestError `testFinally` final) `shouldReturn` Left (TestError "error")
+    it "then throw SomeException as ExceptT error" $ withFinal $ \final ->
+      runExceptT (throwTestException `testFinally` final) `shouldReturn` Left (TestException "user error (error)")
+    it "and should not throw if there are no exceptions" $ withFinal $ \final ->
+      runExceptT (noErrors `testFinally` final) `shouldReturn` Right "no errors"
+  where
+    throwTestError :: ExceptT TestError IO String
+    throwTestError = throwError $ TestError "error"
+    throwTestException :: ExceptT TestError IO String
+    throwTestException = liftIO $ throwIO $ userError "error"
+    noErrors :: ExceptT TestError IO String
+    noErrors = pure "no errors"
+    testErr :: SomeException -> TestError
+    testErr = TestException . show
+    handleCatch :: TestError -> ExceptT TestError IO String
+    handleCatch e = pure $ "caught " <> show e
+    handleException :: SomeException -> ExceptT TestError IO String
+    handleException e = pure $ "caught " <> show e
+    withFinal :: (ExceptT TestError IO String -> IO ()) -> IO ()
+    withFinal test = do
+      r <- newIORef False
+      let final = liftIO $ writeIORef r True >> pure "final"
+      test final
+      readIORef r `shouldReturn` True
+
+data TestError = TestError String | TestException String
+  deriving (Eq, Show)
+
+instance Exception TestError

tests/NtfClient.hs

@@ -44,6 +44,7 @@ import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client
 import Simplex.Messaging.Transport.HTTP2 (HTTP2Body (..), http2TLSParams)
 import Simplex.Messaging.Transport.HTTP2.Server
+import Simplex.Messaging.Transport.Server
 import Test.Hspec
 import UnliftIO.Async
 import UnliftIO.Concurrent
@@ -88,9 +89,9 @@ ntfServerCfg =
           { apnsPort = apnsTestPort,
             caStoreFile = "tests/fixtures/ca.crt"
           },
+      subsBatchSize = 900,
       inactiveClientExpiration = Just defaultInactiveClientExpiration,
       storeLogFile = Nothing,
-      resubscribeDelay = 1000,
       -- CA certificate private key is not needed for initialization
       caCertificateFile = "tests/fixtures/ca.crt",
       privateKeyFile = "tests/fixtures/server.key",
@@ -100,7 +101,7 @@ ntfServerCfg =
       logStatsStartTime = 0,
       serverStatsLogFile = "tests/ntf-server-stats.daily.log",
       serverStatsBackupFile = Nothing,
-      logTLSErrors = True
+      transportConfig = defaultTransportServerConfig
     }
 
 withNtfServerStoreLog :: ATransport -> (ThreadId -> IO a) -> IO a
@@ -134,7 +135,7 @@ ntfServerTest _ t = runNtfTest $ \h -> tPut' h t >> tGet' h
   where
     tPut' h (sig, corrId, queueId, smp) = do
       let t' = smpEncode (sessionId (h :: THandle c), corrId, queueId, smp)
-      [Right ()] <- tPut h [(sig, t')]
+      [Right ()] <- tPut h Nothing [(sig, t')]
       pure ()
     tGet' h = do
       [(Nothing, _, (CorrId corrId, qId, Right cmd))] <- tGet h
@@ -167,7 +168,7 @@ apnsMockServerConfig =
       caCertificateFile = "tests/fixtures/ca.crt",
       privateKeyFile = "tests/fixtures/server.key",
       certificateFile = "tests/fixtures/server.crt",
-      logTLSErrors = True
+      transportConfig = defaultTransportServerConfig
     }
 
 withAPNSMockServer :: (APNSMockServer -> IO ()) -> IO ()

tests/SMPClient.hs

@@ -24,6 +24,7 @@ import Simplex.Messaging.Server (runSMPServerBlocking)
 import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client
+import Simplex.Messaging.Transport.Server
 import Simplex.Messaging.Version
 import System.Environment (lookupEnv)
 import System.Info (os)
@@ -81,7 +82,7 @@ cfg =
   ServerConfig
     { transports = undefined,
       tbqSize = 1,
-      serverTbqSize = 1,
+      -- serverTbqSize = 1,
       msgQueueQuota = 4,
       queueIdBytes = 24,
       msgIdBytes = 24,
@@ -99,7 +100,8 @@ cfg =
       privateKeyFile = "tests/fixtures/server.key",
       certificateFile = "tests/fixtures/server.crt",
       smpServerVRange = supportedSMPServerVRange,
-      logTLSErrors = True
+      transportConfig = defaultTransportServerConfig,
+      controlPort = Nothing
     }
 
 withSmpServerStoreMsgLogOnV2 :: HasCallStack => ATransport -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
@@ -159,7 +161,7 @@ smpServerTest _ t = runSmpTest $ \h -> tPut' h t >> tGet' h
   where
     tPut' h (sig, corrId, queueId, smp) = do
       let t' = smpEncode (sessionId (h :: THandle c), corrId, queueId, smp)
-      [Right ()] <- tPut h [(sig, t')]
+      [Right ()] <- tPut h Nothing [(sig, t')]
       pure ()
     tGet' h = do
       [(Nothing, _, (CorrId corrId, qId, Right cmd))] <- tGet h

tests/ServerTests.hs

@@ -88,7 +88,7 @@ signSendRecv h@THandle {thVersion, sessionId} pk (corrId, qId, cmd) = do
 
 tPut1 :: Transport c => THandle c -> SentRawTransmission -> IO (Either TransportError ())
 tPut1 h t = do
-  [r] <- tPut h [t]
+  [r] <- tPut h Nothing [t]
   pure r
 
 tGet1 :: (ProtocolEncoding err cmd, Transport c, MonadIO m, MonadFail m) => THandle c -> m (SignedTransmission err cmd)

tests/Test.hs

@@ -7,6 +7,7 @@ import CoreTests.CryptoTests
 import CoreTests.EncodingTests
 import CoreTests.ProtocolErrorTests
 import CoreTests.RetryIntervalTests
+import CoreTests.UtilTests
 import CoreTests.VersionRangeTests
 import FileDescriptionTests (fileDescriptionTests)
 import NtfServerTests (ntfServerTests)
@@ -39,6 +40,7 @@ main = do
           describe "Version range" versionRangeTests
           describe "Encryption tests" cryptoTests
           describe "Retry interval tests" retryIntervalTests
+          describe "Util tests" utilTests
         describe "SMP server via TLS" $ serverTests (transport @TLS)
         describe "SMP server via WebSockets" $ serverTests (transport @WS)
         describe "Notifications server" $ ntfServerTests (transport @TLS)

tests/XFTPAgent.hs

@@ -37,7 +37,7 @@ xftpAgentTests = around_ testBracket . describe "Functional API" $ do
   it "should send and receive file" testXFTPAgentSendReceive
   it "should resume receiving file after restart" testXFTPAgentReceiveRestore
   it "should cleanup rcv tmp path after permanent error" testXFTPAgentReceiveCleanup
-  xit'' "should resume sending file after restart" testXFTPAgentSendRestore
+  it "should resume sending file after restart" testXFTPAgentSendRestore
   it "should cleanup snd prefix path after permanent error" testXFTPAgentSendCleanup
   it "should delete sent file on server" testXFTPAgentDelete
   it "should resume deleting file after restart" testXFTPAgentDeleteRestore
@@ -255,6 +255,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do
     liftIO $ sfId' `shouldBe` sfId
 
     -- prefix path should be removed after sending file
+    threadDelay 100000
     doesDirectoryExist prefixPath `shouldReturn` False
     doesFileExist encPath `shouldReturn` False
 

tests/XFTPClient.hs

@@ -14,6 +14,7 @@ import Simplex.FileTransfer.Description
 import Simplex.FileTransfer.Server (runXFTPServerBlocking)
 import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..), defaultFileExpiration)
 import Simplex.Messaging.Protocol (XFTPServer)
+import Simplex.Messaging.Transport.Server
 import Test.Hspec
 
 xftpTest :: HasCallStack => (HasCallStack => XFTPClient -> IO ()) -> Expectation
@@ -111,7 +112,7 @@ testXFTPServerConfig =
       logStatsStartTime = 0,
       serverStatsLogFile = "tests/tmp/xftp-server-stats.daily.log",
       serverStatsBackupFile = Nothing,
-      logTLSErrors = True
+      transportConfig = defaultTransportServerConfig
     }
 
 testXFTPClientConfig :: XFTPClientConfig