6.1.3.0

* smp server: log operation of restore error

* space

* rename

* revert version

CHANGELOG.md

@@ -1,5 +1,27 @@
+# 6.1.3
+
+SMP server: fix restoring notification credentials.
+
+# 6.1.2
+
+Servers: more reliable restoring of state.
+
+SMP server: reduced memory usage and faster start.
+
+Notifications: compensate for iOS notifications being droppted by Apple while device is offline (#1378):
+- Ntf server: send multiple SMP notifications in one iOS notification.
+- Agent: get multiple messages for one iOS notification.
+
+# 6.1.1
+
+SMP:
+- stop server faster (#1371)
+- add STORE error (#1372)
+
 # 6.1.0
 
+Version 6.1.0.7
+
 SMP server and client:
 - transport block encryption (#1317).
 

README.md

@@ -149,8 +149,15 @@ On Linux, you can deploy smp and xftp server using Docker. This will download im
 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 'c90886104cd640b2ed64921dba80e90691db36788e8d6dcc13d8f33f92f0ea54 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
+curl --proto '=https' --tlsv1.2 -sSf https://raw.githubusercontent.com/simplex-chat/simplexmq/stable/install.sh -o simplex-server-install.sh &&\
+if echo '53fcdb4ceab324316e2c4cda7e84dbbb344f32550a65975a7895425e5a1be757 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

install.sh

@@ -2,9 +2,6 @@
 set -eu
 
 # Links to scripts/configs
-bin="https://github.com/simplex-chat/simplexmq/releases/latest/download"
-remote_version="$(curl --proto '=https' --tlsv1.2 -sSf -L https://api.github.com/repos/simplex-chat/simplexmq/releases/latest | grep -i "tag_name" | awk -F \" '{print $4}')"
-
 scripts="https://raw.githubusercontent.com/simplex-chat/simplexmq/stable/scripts/main"
 scripts_systemd_smp="$scripts/smp-server.service"
 scripts_systemd_xftp="$scripts/xftp-server.service"
@@ -57,7 +54,7 @@ ${GRN}1.${NC} Install latest binaries from GitHub releases:
 ${GRN}2.${NC} Create server directories:
     - smp: ${YLW}${path_conf_smp}${NC}
     - xftp: ${YLW}${path_conf_xftp}${NC}
-${GRN}3.${NC} Setup user for each server:
+${GRN}3.${NC} Setup user for server:
     - xmp: ${YLW}${user_smp}${NC}
     - xftp: ${YLW}${user_xftp}${NC}
 ${GRN}4.${NC} Create systemd services:
@@ -67,9 +64,8 @@ ${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 installing both xftp and smp servers
-  - ${GRN}1${NC} to install only smp server
-  - ${GRN}2${NC} to install only xftp server
+  - ${GRN}1${NC} to install smp server
+  - ${GRN}2${NC} to install xftp server
   - ${RED}Ctrl+C${NC} to cancel installation
   
 Selection: "
@@ -83,6 +79,21 @@ Please checkout our server guides:
 To uninstall with full clean-up, simply run: ${YLW}sudo /usr/local/bin/simplex-servers-uninstall${NC}
 "
 
+set_version() {
+  ver="${VER:-latest}"
+
+  case "$ver" in 
+    latest)
+      bin="https://github.com/simplex-chat/simplexmq/releases/latest/download"
+      remote_version="$(curl --proto '=https' --tlsv1.2 -sSf -L https://api.github.com/repos/simplex-chat/simplexmq/releases/latest | grep -i "tag_name" | awk -F \" '{print $4}')"
+      ;;
+    *)
+      bin="https://github.com/simplex-chat/simplexmq/releases/download/${ver}"
+      remote_version="${ver}"
+      ;;
+  esac
+}
+
 os_test() {
  . /etc/os-release
 
@@ -155,6 +166,7 @@ checks() {
   exit 1
  fi
  
+ set_version
  os_test
 
  mkdir -p $path_conf_info  
@@ -166,13 +178,11 @@ main() {
  printf "%b\n%b" "${BLU}$logo${NC}" "$welcome"
  read ans
 
- if [ "$ans" = '1' ]; then
-  setup='smp'
- elif [ "$ans" = '2' ]; then
-  setup='xftp'
- else
-  setup='smp xftp'
- fi
+ case "$ans" in
+  1) setup='smp' ;;
+  2) setup='xftp' ;;
+  *) printf 'Installation aborted.\n' && exit 0 ;;
+ esac
 
  printf "Installing binaries..."
  

package.yaml

@@ -1,5 +1,5 @@
 name: simplexmq
-version: 6.1.0.7
+version: 6.1.3.0
 synopsis: SimpleXMQ message broker
 description: |
   This package includes <./docs/Simplex-Messaging-Server.html server>,

protocol/xftp.md

@@ -559,11 +559,13 @@ In current implementation of XFTP protocol in SimpleX Chat clients don't use FAC
 
  - perform traffic correlation attacks against senders and recipients and correlate senders and recipients within the monitored set, frustrated by the number of users on the servers.
 
- - observe how much traffic is being sent, and make guesses as to its purpose
+ - observe how much traffic is being sent, and make guesses as to its purpose.
 
-*cannot, even in case of a compromised transport protocol:*
+ - in case of a compromised transport protocol, correlate file senders and receivers.
 
- - perform traffic correlation attacks with any increase in efficiency over a non-compromised transport protocol
+*cannot, in case of a non-compromised transport protocol:*
+
+ - perform traffic correlation attacks.
 
 #### XFTP server
 

rfcs/2024-09-25-ios-notifications-2.md

@@ -0,0 +1,49 @@
+# iOS notifications delivery
+
+## Problem
+
+For iOS notifications to be delivered the client has to create credentials for notification subscription on SMP server using NKEY command and after that create a subscription on notification server using SNEW command. These two commands are sent in sequence, after the connections are created, and for it to happen the client needs to be online and in foreground.
+
+iOS users tend to close the app when it is not used, and iOS has very limited permissions for background activities, so these notification subscriptions are created with a substantial delay, and notifications do not work.
+
+This problem is distinct from and probably more common than other problems affecting notifications delivery described [here](./2024-07-06-ios-notifications.md).
+
+## Solution
+
+1. When the new connection is created, the client already knows if it needs to create notification subscription or not, based on the conversation setting (e.g., if the group is muted, the client will not create notification subscription as well.). We should extend NEW command to avoid the need to send additional NKEY command with an option to create notification subscription at the point where connection is created. NDEL would still be used to disable this notification, and NKEY will be used to re-enable it.
+
+2. In the same way we stopped using SDEL command (NDEL sends notification DELD to subscribed notification server) to delete notificaiton subscriptions from notification server, we should delegate creating notification subscription on notification server to SMP servers. Clients could use keys agreed with ntf server for e2e encryption and for command authorization to encrypt and sign instruction to create notification subscription that will be forwarded to notification server using protocol similar to SMP proxies. This will avoid the need for clients to separately contact notification servers that won't happen until they are online.
+
+3. Instead of making Ntf server trust DELD notifications, we could send deletion instructions signed by the client, which will only fail to send in case notification server is down (and they won't be sent later after server restart).
+
+Cons:
+- If SMP servers were to retain in the storage the information about which notification server is used for which queue, it would reduce metadata privacy. While currently it is not an issue, as all notification servers are known and operated by us, once there are other client apps, this can be used for app users fingerprinting, which would act as a deterrence from using new apps – but only if app users use servers of operators who are different from the app provider. To mitigate it, we could only store it in server memory and include notification instruction in subscription commands (SUB) and include notification subscription status in SUB responses. We don't need to mitigate the problem of server being able to store this information, as messaging servers can observe which notification servers connect to them anyway.
+- If SMP server is restarted before the subscription request is forwared to the notification server, then it will have to be forwarded again, once the client subscribes. The problem here is that if the client is offline, it will neither subscribe to the queue to send notification subscription request, nor receive notifications from this queue. Storing notification server and subscription request would mitigate that, as in this case we could send all pending requests on server start, without depending on client subscriptions.
+- "Small" agent will need to support connections to ntf servers and manage workers that retry sending pending subscription requests.
+- Until the client learns the public keys of notification server, it will not be able to decrypt notifications. It potentially can be mitigated by using the public key of the server returned when token is created, in this way different client keys (per-queue) will be combined with the same ntf server key (per-token).
+
+## Implementation details
+
+1. NEW and NKEY commands will need to be extended to include notification subscription request. As the notifier ID needs to be sent to notification server, this notifier ID will have to be client-generated and supplied as part of NEW command.
+
+now:
+
+```haskell
+NEW :: RcvPublicAuthKey -> RcvPublicDhKey -> Maybe BasicAuth -> SubscriptionMode -> SenderCanSecure -> Command Recipient
+NKEY :: NtfPublicAuthKey -> RcvNtfPublicDhKey -> Command Recipient
+```
+
+extended:
+
+```haskell
+NEW :: RcvPublicAuthKey -> RcvPublicDhKey -> Maybe BasicAuth -> SubscriptionMode -> SenderCanSecure -> Maybe NtfRequest -> Command Recipient
+
+data NtfRequest = NtfRequest NotifierId NtfPublicAuthKey RcvNtfPublicDhKey NtfServerRequest
+
+data NtfServerRequest = NtfServerRequest NtfServer EncSingedNtfCmd
+
+NKEY :: NtfPublicAuthKey -> RcvNtfPublicDhKey -> Maybe NtfServerRequest -> Command Recipient
+-- NotifierID is passed in entity ID field of the transmission
+```
+
+2. Notification server will need to support an additional command to receive "proxied" subscription commands, `SFWD`, that would include `NtfServerRequest`. This command can include both `SNEW` and `SDEL` commands.

rfcs/done/2023-10-25-remote-control.md

@@ -53,7 +53,7 @@ The session invitation contains this data:
 - CA TLS certificate fingerprint of the controller - this is part of long term identity of the controller established during the first session, and repeated in the subsequent session announcements.
 - Session Ed25519 public key used to verify the announcement and commands - this mitigates the compromise of the long term signature key, as the controller will have to sign each command with this key first.
 - Long-term Ed25519 public key used to verify the announcement and commands - this is part of the long term controller identity.
-- Session X25519 DH key and sntrup761 KEM encapsulation key to agree session encryption (both for multicast announcement and for commands and responses in TLS), as described in https://datatracker.ietf.org/doc/draft-josefsson-ntruprime-hybrid/. The new keys are used for each session, and if client key is already available (from the previous session), the computed shared secret will be used to encrypt the announcement multicast packet. The out-of-band invitation is unencrypted. These DH public key and KEM encapsulation key are always sent unencrypted. NaCL Cryptobox is used for encryption.
+- Session X25519 DH key and sntrup761 KEM encapsulation key to agree session encryption (both for multicast announcement and for commands and responses in TLS), as described in https://datatracker.ietf.org/doc/draft-josefsson-ntruprime-hybrid/. The new keys are used for each session, and if client key is already available (from the previous session), the computed shared secret will be used to encrypt the announcement multicast packet. The out-of-band invitation is unencrypted. This DH public key is always sent unencrypted. NaCL Cryptobox is used for encryption.
 
 Host device decrypts (except the first session) and validates the invitation:
 - Session signature is valid.

scripts/main/simplex-servers-update

@@ -2,8 +2,6 @@
 set -eu
 
 # Links to scripts/configs
-bin="https://github.com/simplex-chat/simplexmq/releases/latest/download"
-
 scripts="https://raw.githubusercontent.com/simplex-chat/simplexmq/stable/scripts/main"
 scripts_systemd_smp="$scripts/smp-server.service"
 scripts_systemd_xftp="$scripts/xftp-server.service"
@@ -76,6 +74,21 @@ installed_test() {
   set -u
 }
 
+set_version() {
+  ver="${VER:-latest}"
+
+  case "$ver" in 
+    latest)
+      bin="https://github.com/simplex-chat/simplexmq/releases/latest/download"
+      remote_version="$(curl --proto '=https' --tlsv1.2 -sSf -L https://api.github.com/repos/simplex-chat/simplexmq/releases/latest | grep -i "tag_name" | awk -F \" '{print $4}')"
+      ;;
+    *)
+      bin="https://github.com/simplex-chat/simplexmq/releases/download/${ver}"
+      remote_version="${ver}"
+      ;;
+  esac
+}
+
 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"
@@ -137,8 +150,6 @@ update_bins() {
   eval "bin=\$bin_${1}"
   eval "path_bin=\$path_bin_${1}"
 
-  remote_version="$(curl --proto '=https' --tlsv1.2 -sSf -L https://api.github.com/repos/simplex-chat/simplexmq/releases/latest | grep -i "tag_name" | awk -F \" '{print $4}')"
-
   set_ver() {
     local_version='unset'
     sed -i -- "s/local_version_${1}=.*/local_version_${1}='${remote_version}'/" "$path_conf_info/release"
@@ -192,6 +203,7 @@ checks() {
     exit 1
   fi
 
+  set_version
   os_test
   installed_test
 

scripts/main/smp-server.service

@@ -9,7 +9,9 @@ ExecStart=/usr/local/bin/smp-server start +RTS -N -RTS
 ExecStopPost=/usr/local/bin/simplex-servers-stopscript smp-server
 LimitNOFILE=65535
 KillSignal=SIGINT
+TimeoutStartSec=infinity
 TimeoutStopSec=infinity
+AmbientCapabilities=CAP_NET_BIND_SERVICE
 
 [Install]
 WantedBy=multi-user.target

scripts/main/xftp-server.service

@@ -9,6 +9,7 @@ ExecStart=/usr/local/bin/xftp-server start +RTS -N -RTS
 ExecStopPost=/usr/local/bin/simplex-servers-stopscript xftp-server
 LimitNOFILE=65535
 KillSignal=SIGINT
+TimeoutStartSec=infinity
 TimeoutStopSec=infinity
 AmbientCapabilities=CAP_NET_BIND_SERVICE
    

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        6.1.0.7
+version:        6.1.3.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -136,6 +136,7 @@ library
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240624_snd_secure
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240702_servers_stats
       Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240930_ntf_tokens_to_delete
+      Simplex.Messaging.Agent.Store.SQLite.Migrations.M20241007_rcv_queues_last_broker_ts
       Simplex.Messaging.Agent.TRcvQueues
       Simplex.Messaging.Client
       Simplex.Messaging.Client.Agent

src/Simplex/FileTransfer/Server/StoreLog.hs

@@ -34,8 +34,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (RcvPublicAuthKey, RecipientId, SenderId)
 import Simplex.Messaging.Server.QueueStore (RoundedSystemTime)
 import Simplex.Messaging.Server.StoreLog
-import Simplex.Messaging.Util (bshow, whenM)
-import System.Directory (doesFileExist, renameFile)
+import Simplex.Messaging.Util (bshow)
 import System.IO
 
 data FileStoreLogRecord
@@ -81,13 +80,7 @@ logAckFile :: StoreLog 'WriteMode -> RecipientId -> IO ()
 logAckFile s = logFileStoreRecord s . AckFile
 
 readWriteFileStore :: FilePath -> FileStore -> IO (StoreLog 'WriteMode)
-readWriteFileStore f st = do
-  whenM (doesFileExist f) $ do
-    readFileStore f st
-    renameFile f $ f <> ".bak"
-  s <- openWriteStoreLog f
-  writeFileStore s st
-  pure s
+readWriteFileStore = readWriteStoreLog readFileStore writeFileStore
 
 readFileStore :: FilePath -> FileStore -> IO ()
 readFileStore f st = mapM_ (addFileLogRecord . LB.toStrict) . LB.lines =<< LB.readFile f

src/Simplex/Messaging/Agent.hs

@@ -63,8 +63,8 @@ module Simplex.Messaging.Agent
     rejectContact,
     subscribeConnection,
     subscribeConnections,
-    getConnectionMessage,
-    getNotificationMessage,
+    getConnectionMessages,
+    getNotificationConns,
     resubscribeConnection,
     resubscribeConnections,
     sendMessage,
@@ -134,7 +134,7 @@ import Data.Bifunctor (bimap, first, second)
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Composition ((.:), (.:.), (.::), (.::.))
-import Data.Either (isRight, rights)
+import Data.Either (isRight, partitionEithers, rights)
 import Data.Foldable (foldl', toList)
 import Data.Functor (($>))
 import Data.Functor.Identity
@@ -175,8 +175,7 @@ import Simplex.Messaging.Crypto.Ratchet (PQEncryption, PQSupport (..), pattern P
 import qualified Simplex.Messaging.Crypto.Ratchet as CR
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Notifications.Protocol (DeviceToken, NtfRegCode (NtfRegCode), NtfTknStatus (..), NtfTokenId)
-import Simplex.Messaging.Notifications.Server.Push.APNS (PNMessageData (..), pnMessagesP)
+import Simplex.Messaging.Notifications.Protocol (DeviceToken, NtfRegCode (NtfRegCode), NtfTknStatus (..), NtfTokenId, PNMessageData (..), pnMessagesP)
 import Simplex.Messaging.Notifications.Types
 import Simplex.Messaging.Parsers (parse)
 import Simplex.Messaging.Protocol (BrokerMsg, Cmd (..), ErrorType (AUTH), MsgBody, MsgFlags (..), NtfServer, ProtoServerWithAuth, ProtocolType (..), ProtocolTypeI (..), SMPMsgMeta, SParty (..), SProtocolType (..), SndPublicAuthKey, SubscriptionMode (..), UserProtocol, VersionSMPC, sndAuthKeySMPClientVersion)
@@ -383,15 +382,15 @@ subscribeConnections :: AgentClient -> [ConnId] -> AE (Map ConnId (Either AgentE
 subscribeConnections c = withAgentEnv c . subscribeConnections' c
 {-# INLINE subscribeConnections #-}
 
--- | Get connection message (GET command)
-getConnectionMessage :: AgentClient -> ConnId -> AE (Maybe SMPMsgMeta)
-getConnectionMessage c = withAgentEnv c . getConnectionMessage' c
-{-# INLINE getConnectionMessage #-}
+-- | Get messages for connections (GET commands)
+getConnectionMessages :: AgentClient -> NonEmpty ConnId -> IO (NonEmpty (Maybe SMPMsgMeta))
+getConnectionMessages c = withAgentEnv' c . getConnectionMessages' c
+{-# INLINE getConnectionMessages #-}
 
--- | Get connection message for received notification
-getNotificationMessage :: AgentClient -> C.CbNonce -> ByteString -> AE (NotificationInfo, Maybe SMPMsgMeta)
-getNotificationMessage c = withAgentEnv c .: getNotificationMessage' c
-{-# INLINE getNotificationMessage #-}
+-- | Get connections for received notification
+getNotificationConns :: AgentClient -> C.CbNonce -> ByteString -> AE (NonEmpty NotificationInfo)
+getNotificationConns c = withAgentEnv c .: getNotificationConns' c
+{-# INLINE getNotificationConns #-}
 
 resubscribeConnection :: AgentClient -> ConnId -> AE ()
 resubscribeConnection c = withAgentEnv c . resubscribeConnection' c
@@ -1047,29 +1046,57 @@ resubscribeConnections' c connIds = do
   -- union is left-biased, so results returned by subscribeConnections' take precedence
   (`M.union` r) <$> subscribeConnections' c connIds'
 
-getConnectionMessage' :: AgentClient -> ConnId -> AM (Maybe SMPMsgMeta)
-getConnectionMessage' c connId = do
-  whenM (atomically $ hasActiveSubscription c connId) . throwE $ CMD PROHIBITED "getConnectionMessage: subscribed"
-  SomeConn _ conn <- withStore c (`getConn` connId)
-  case conn of
-    DuplexConnection _ (rq :| _) _ -> getQueueMessage c rq
-    RcvConnection _ rq -> getQueueMessage c rq
-    ContactConnection _ rq -> getQueueMessage c rq
-    SndConnection _ _ -> throwE $ CONN SIMPLEX
-    NewConnection _ -> throwE $ CMD PROHIBITED "getConnectionMessage: NewConnection"
+getConnectionMessages' :: AgentClient -> NonEmpty ConnId -> AM' (NonEmpty (Maybe SMPMsgMeta))
+getConnectionMessages' c = mapM getMsg
+  where
+    getMsg :: ConnId -> AM' (Maybe SMPMsgMeta)
+    getMsg connId =
+      getConnectionMessage connId `catchAgentError'` \e -> do
+        logError $ "Error loading message: " <> tshow e
+        pure Nothing
+    getConnectionMessage :: ConnId -> AM (Maybe SMPMsgMeta)
+    getConnectionMessage connId = do
+      whenM (atomically $ hasActiveSubscription c connId) . throwE $ CMD PROHIBITED "getConnectionMessage: subscribed"
+      SomeConn _ conn <- withStore c (`getConn` connId)
+      case conn of
+        DuplexConnection _ (rq :| _) _ -> getQueueMessage c rq
+        RcvConnection _ rq -> getQueueMessage c rq
+        ContactConnection _ rq -> getQueueMessage c rq
+        SndConnection _ _ -> throwE $ CONN SIMPLEX
+        NewConnection _ -> throwE $ CMD PROHIBITED "getConnectionMessage: NewConnection"
 
-getNotificationMessage' :: AgentClient -> C.CbNonce -> ByteString -> AM (NotificationInfo, Maybe SMPMsgMeta)
-getNotificationMessage' c nonce encNtfInfo = do
+getNotificationConns' :: AgentClient -> C.CbNonce -> ByteString -> AM (NonEmpty NotificationInfo)
+getNotificationConns' c nonce encNtfInfo =
   withStore' c getActiveNtfToken >>= \case
     Just NtfToken {ntfDhSecret = Just dhSecret} -> do
-      ntfData <- agentCbDecrypt dhSecret nonce encNtfInfo
+      ntfData <- liftEither $ agentCbDecrypt dhSecret nonce encNtfInfo
       pnMsgs <- liftEither (parse pnMessagesP (INTERNAL "error parsing PNMessageData") ntfData)
-      let PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta} = L.last pnMsgs
-      (ntfConnId, rcvNtfDhSecret) <- withStore c (`getNtfRcvQueue` smpQueue)
-      ntfMsgMeta <- (eitherToMaybe . smpDecode <$> agentCbDecrypt rcvNtfDhSecret nmsgNonce encNMsgMeta) `catchAgentError` \_ -> pure Nothing
-      msgMeta <- getConnectionMessage' c ntfConnId
-      pure (NotificationInfo {ntfConnId, ntfTs, ntfMsgMeta}, msgMeta)
-    _ -> throwE $ CMD PROHIBITED "getNotificationMessage"
+      let (initNtfs, lastNtf) = (L.init pnMsgs, L.last pnMsgs)
+      rs <-
+        lift $ withStoreBatch c $ \db ->
+          let initNtfInfos = map (getInitNtfInfo db) initNtfs
+              lastNtfInfo = Just . fst <$$> getNtfInfo db lastNtf
+           in initNtfInfos <> [lastNtfInfo]
+      let (errs, ntfInfos_) = partitionEithers rs
+      logError $ "Error(s) loading notifications: " <> tshow errs
+      case L.nonEmpty $ catMaybes ntfInfos_ of
+        Just r -> pure r
+        Nothing -> throwE $ INTERNAL "getNotificationConns: couldn't get conn info"
+    _ -> throwE $ CMD PROHIBITED "getNotificationConns"
+  where
+    getNtfInfo :: DB.Connection -> PNMessageData -> IO (Either AgentErrorType (NotificationInfo, Maybe UTCTime))
+    getNtfInfo db PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta} = runExceptT $ do
+      (ntfConnId, rcvNtfDhSecret, lastBrokerTs_) <- liftError' storeError $ getNtfRcvQueue db smpQueue
+      let ntfMsgMeta = eitherToMaybe $ smpDecode =<< first show (C.cbDecrypt rcvNtfDhSecret nmsgNonce encNMsgMeta)
+          ntfInfo = NotificationInfo {ntfConnId, ntfTs, ntfMsgMeta}
+      pure (ntfInfo, lastBrokerTs_)
+    getInitNtfInfo :: DB.Connection -> PNMessageData -> IO (Either AgentErrorType (Maybe NotificationInfo))
+    getInitNtfInfo db msgData = runExceptT $ do
+      (nftInfo, lastBrokerTs_) <- ExceptT $ getNtfInfo db msgData
+      pure $ case (ntfMsgMeta nftInfo, lastBrokerTs_) of
+        (Just SMP.NMsgMeta {msgTs}, Just lastBrokerTs)
+          | systemToUTCTime msgTs > lastBrokerTs -> Just nftInfo
+        _ -> Nothing
 
 -- | Send message to the connection (SEND command) in Reader monad
 sendMessage' :: AgentClient -> ConnId -> PQEncryption -> MsgFlags -> MsgBody -> AM (AgentMsgId, PQEncryption)
@@ -2479,7 +2506,7 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
 
           decryptClientMessage :: C.DhSecretX25519 -> SMP.ClientMsgEnvelope -> AM (SMP.PrivHeader, AgentMsgEnvelope)
           decryptClientMessage e2eDh SMP.ClientMsgEnvelope {cmNonce, cmEncBody} = do
-            clientMsg <- agentCbDecrypt e2eDh cmNonce cmEncBody
+            clientMsg <- liftEither $ agentCbDecrypt e2eDh cmNonce cmEncBody
             SMP.ClientMessage privHeader clientBody <- parseMessage clientMsg
             agentEnvelope <- parseMessage clientBody
             -- Version check is removed here, because when connecting via v1 contact address the agent still sends v2 message,

src/Simplex/Messaging/Agent/Client.hs

@@ -1590,7 +1590,7 @@ getQueueMessage c rq@RcvQueue {server, rcvId, rcvPrivateKey} = do
 
 decryptSMPMessage :: RcvQueue -> SMP.RcvMessage -> AM SMP.ClientRcvMsgBody
 decryptSMPMessage rq SMP.RcvMessage {msgId, msgBody = SMP.EncRcvMsgBody body} =
-  liftEither . parse SMP.clientRcvMsgBodyP (AGENT A_MESSAGE) =<< decrypt body
+  liftEither $ parse SMP.clientRcvMsgBodyP (AGENT A_MESSAGE) =<< decrypt body
   where
     decrypt = agentCbDecrypt (rcvDhSecret rq) (C.cbNonce msgId)
 
@@ -1830,9 +1830,9 @@ agentCbEncryptOnce clientVersion dhRcvPubKey msg = do
 
 -- | NaCl crypto-box decrypt - both for messages received from the server
 -- and per-queue E2E encrypted messages from the sender that were inside.
-agentCbDecrypt :: C.DhSecretX25519 -> C.CbNonce -> ByteString -> AM ByteString
+agentCbDecrypt :: C.DhSecretX25519 -> C.CbNonce -> ByteString -> Either AgentErrorType ByteString
 agentCbDecrypt dhSecret nonce msg =
-  liftEither . first cryptoError $
+  first cryptoError $
     C.cbDecrypt dhSecret nonce msg
 
 cryptoError :: C.CryptoError -> AgentErrorType

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

@@ -976,10 +976,11 @@ updateRcvIds db connId = do
   pure (internalId, internalRcvId, lastExternalSndId, lastRcvHash)
 
 createRcvMsg :: DB.Connection -> ConnId -> RcvQueue -> RcvMsgData -> IO ()
-createRcvMsg db connId rq rcvMsgData@RcvMsgData {msgMeta = MsgMeta {sndMsgId}, internalRcvId, internalHash} = do
+createRcvMsg db connId rq@RcvQueue {dbQueueId} rcvMsgData@RcvMsgData {msgMeta = MsgMeta {sndMsgId, broker = (_, brokerTs)}, internalRcvId, internalHash} = do
   insertRcvMsgBase_ db connId rcvMsgData
   insertRcvMsgDetails_ db connId rq rcvMsgData
   updateRcvMsgHash db connId sndMsgId internalRcvId internalHash
+  DB.execute db "UPDATE rcv_queues SET last_broker_ts = ? WHERE conn_id = ? AND rcv_queue_id = ?" (brokerTs, connId, dbQueueId)
 
 updateSndIds :: DB.Connection -> ConnId -> IO (Either StoreError (InternalId, InternalSndId, PrevSndMsgHash))
 updateSndIds db connId = runExceptT $ do
@@ -1817,19 +1818,19 @@ getActiveNtfToken db =
           ntfMode = fromMaybe NMPeriodic ntfMode_
        in NtfToken {deviceToken = DeviceToken provider dt, ntfServer, ntfTokenId, ntfPubKey, ntfPrivKey, ntfDhKeys, ntfDhSecret, ntfTknStatus, ntfTknAction, ntfMode}
 
-getNtfRcvQueue :: DB.Connection -> SMPQueueNtf -> IO (Either StoreError (ConnId, RcvNtfDhSecret))
+getNtfRcvQueue :: DB.Connection -> SMPQueueNtf -> IO (Either StoreError (ConnId, RcvNtfDhSecret, Maybe UTCTime))
 getNtfRcvQueue db SMPQueueNtf {smpServer = (SMPServer host port _), notifierId} =
   firstRow' res SEConnNotFound $
     DB.query
       db
       [sql|
-        SELECT conn_id, rcv_ntf_dh_secret
+        SELECT conn_id, rcv_ntf_dh_secret, last_broker_ts
         FROM rcv_queues
         WHERE host = ? AND port = ? AND ntf_id = ? AND deleted = 0
       |]
       (host, port, notifierId)
   where
-    res (connId, Just rcvNtfDhSecret) = Right (connId, rcvNtfDhSecret)
+    res (connId, Just rcvNtfDhSecret, lastBrokerTs_) = Right (connId, rcvNtfDhSecret, lastBrokerTs_)
     res _ = Left SEConnNotFound
 
 setConnectionNtfs :: DB.Connection -> ConnId -> Bool -> IO ()

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

@@ -75,6 +75,7 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240417_rcv_files_appro
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240624_snd_secure
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240702_servers_stats
 import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20240930_ntf_tokens_to_delete
+import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20241007_rcv_queues_last_broker_ts
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (dropPrefix, sumTypeJSON)
 import Simplex.Messaging.Transport.Client (TransportHost)
@@ -118,7 +119,8 @@ schemaMigrations =
     ("m20240417_rcv_files_approved_relays", m20240417_rcv_files_approved_relays, Just down_m20240417_rcv_files_approved_relays),
     ("m20240624_snd_secure", m20240624_snd_secure, Just down_m20240624_snd_secure),
     ("m20240702_servers_stats", m20240702_servers_stats, Just down_m20240702_servers_stats),
-    ("m20240930_ntf_tokens_to_delete", m20240930_ntf_tokens_to_delete, Just down_m20240930_ntf_tokens_to_delete)
+    ("m20240930_ntf_tokens_to_delete", m20240930_ntf_tokens_to_delete, Just down_m20240930_ntf_tokens_to_delete),
+    ("m20241007_rcv_queues_last_broker_ts", m20241007_rcv_queues_last_broker_ts, Just down_m20241007_rcv_queues_last_broker_ts)
   ]
 
 -- | The list of migrations in ascending order by date

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

@@ -0,0 +1,18 @@
+{-# LANGUAGE QuasiQuotes #-}
+
+module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20241007_rcv_queues_last_broker_ts where
+
+import Database.SQLite.Simple (Query)
+import Database.SQLite.Simple.QQ (sql)
+
+m20241007_rcv_queues_last_broker_ts :: Query
+m20241007_rcv_queues_last_broker_ts =
+    [sql|
+ALTER TABLE rcv_queues ADD COLUMN last_broker_ts TEXT;
+|]
+
+down_m20241007_rcv_queues_last_broker_ts :: Query
+down_m20241007_rcv_queues_last_broker_ts =
+    [sql|
+ALTER TABLE rcv_queues DROP COLUMN last_broker_ts;
+|]

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

@@ -56,6 +56,7 @@ CREATE TABLE rcv_queues(
   switch_status TEXT,
   deleted INTEGER NOT NULL DEFAULT 0,
   snd_secure INTEGER NOT NULL DEFAULT 0,
+  last_broker_ts TEXT,
   PRIMARY KEY(host, port, rcv_id),
   FOREIGN KEY(host, port) REFERENCES servers
   ON DELETE RESTRICT ON UPDATE CASCADE,

src/Simplex/Messaging/Notifications/Protocol.hs

@@ -20,8 +20,11 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import Data.Functor (($>))
 import Data.Kind
+import Data.List.NonEmpty (NonEmpty (..))
+import qualified Data.List.NonEmpty as L
 import Data.Maybe (isNothing)
 import Data.Text.Encoding (decodeLatin1, encodeUtf8)
+import Data.Time.Clock.System
 import Data.Type.Equality
 import Data.Word (Word16)
 import Database.SQLite.Simple.FromField (FromField (..))
@@ -426,6 +429,30 @@ instance FromJSON DeviceToken where
     t <- encodeUtf8 <$> o .: "token"
     pure $ DeviceToken pp t
 
+-- List of PNMessageData uses semicolon-separated encoding instead of strEncode,
+-- because strEncode of NonEmpty list uses comma for separator,
+-- and encoding of PNMessageData's smpQueue has comma in list of hosts
+encodePNMessages :: NonEmpty PNMessageData -> ByteString
+encodePNMessages = B.intercalate ";" . map strEncode . L.toList
+
+pnMessagesP :: A.Parser (NonEmpty PNMessageData)
+pnMessagesP = L.fromList <$> strP `A.sepBy1` A.char ';'
+
+data PNMessageData = PNMessageData
+  { smpQueue :: SMPQueueNtf,
+    ntfTs :: SystemTime,
+    nmsgNonce :: C.CbNonce,
+    encNMsgMeta :: EncNMsgMeta
+  }
+  deriving (Show)
+
+instance StrEncoding PNMessageData where
+  strEncode PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta} =
+    strEncode (smpQueue, ntfTs, nmsgNonce, encNMsgMeta)
+  strP = do
+    (smpQueue, ntfTs, nmsgNonce, encNMsgMeta) <- strP
+    pure PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta}
+
 type NtfEntityId = EntityId
 
 type NtfSubscriptionId = NtfEntityId

src/Simplex/Messaging/Notifications/Server.hs

@@ -19,8 +19,11 @@ import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.Reader
+import Data.Bifunctor (first)
+import qualified Data.ByteString.Builder as BLD
 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.IORef
@@ -46,7 +49,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Notifications.Server.Control
 import Simplex.Messaging.Notifications.Server.Env
-import Simplex.Messaging.Notifications.Server.Push.APNS (PNMessageData (..), PushNotification (..), PushProviderError (..))
+import Simplex.Messaging.Notifications.Server.Push.APNS (PushNotification (..), PushProviderError (..))
 import Simplex.Messaging.Notifications.Server.Stats
 import Simplex.Messaging.Notifications.Server.Store
 import Simplex.Messaging.Notifications.Server.StoreLog
@@ -86,6 +89,7 @@ type M a = ReaderT NtfEnv IO a
 
 ntfServer :: NtfServerConfig -> TMVar Bool -> M ()
 ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
+  restoreServerLastNtfs
   restoreServerStats
   s <- asks subscriber
   ps <- asks pushServer
@@ -111,12 +115,14 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
 
     stopServer :: M ()
     stopServer = do
-      withNtfLog closeStoreLog
-      saveServerStats
+      saveServer
       NtfSubscriber {smpSubscribers, smpAgent} <- asks subscriber
       liftIO $ readTVarIO smpSubscribers >>= mapM_ (\SMPSubscriber {subThreadId} -> readTVarIO subThreadId >>= mapM_ (deRefWeak >=> mapM_ killThread))
       liftIO $ closeSMPClientAgent smpAgent
 
+    saveServer :: M ()
+    saveServer = withNtfLog closeStoreLog >> saveServerLastNtfs >> saveServerStats
+
     serverStatsThread_ :: NtfServerConfig -> [M ()]
     serverStatsThread_ NtfServerConfig {logStatsInterval = Just interval, logStatsStartTime, serverStatsLogFile} =
       [logServerStats logStatsStartTime interval serverStatsLogFile]
@@ -356,8 +362,11 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
               NtfPushServer {pushQ} <- asks pushServer
               stats <- asks serverStats
               liftIO $ updatePeriodStats (activeSubs stats) ntfId
-              atomically (findNtfSubscriptionToken st smpQueue)
-                >>= mapM_ (\tkn -> atomically (writeTBQueue pushQ (tkn, PNMessage (PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta} :| []))))
+              tkn_ <- atomically (findNtfSubscriptionToken st smpQueue)
+              forM_ tkn_ $ \tkn -> do
+                let newNtf = PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta}
+                lastNtfs <- liftIO $ addTokenLastNtf st (ntfTknId tkn) newNtf
+                atomically (writeTBQueue pushQ (tkn, PNMessage lastNtfs))
               incNtfStat ntfReceived
             Right SMP.END ->
               whenM (atomically $ activeClientSession' ca sessionId srv) $
@@ -725,6 +734,44 @@ incNtfStat statSel = do
   stats <- asks serverStats
   liftIO $ atomicModifyIORef'_ (statSel stats) (+ 1)
 
+saveServerLastNtfs :: M ()
+saveServerLastNtfs = asks (storeLastNtfsFile . config) >>= mapM_ saveLastNtfs
+  where
+    saveLastNtfs f = do
+      logInfo $ "saving last notifications to file " <> T.pack f
+      NtfStore {tokenLastNtfs} <- asks store
+      liftIO . withFile f WriteMode $ \h ->
+        readTVarIO tokenLastNtfs >>= mapM_ (saveTokenLastNtfs h) . M.assocs
+      logInfo "notifications saved"
+      where
+        -- reverse on save, to save notifications in order, will become reversed again when restoring.
+        saveTokenLastNtfs h (tknId, v) = BLD.hPutBuilder h . encodeLastNtfs tknId . L.reverse =<< readTVarIO v
+        encodeLastNtfs tknId = mconcat . L.toList . L.map (\ntf -> BLD.byteString (strEncode $ TNMRv1 tknId ntf) <> BLD.char8 '\n')
+
+restoreServerLastNtfs :: M ()
+restoreServerLastNtfs =
+  asks (storeLastNtfsFile . config) >>= mapM_ restoreLastNtfs
+  where
+    restoreLastNtfs f =
+      whenM (doesFileExist f) $ do
+        logInfo $ "restoring last notifications from file " <> T.pack f
+        st <- asks store
+        runExceptT (liftIO (LB.readFile f) >>= mapM (restoreNtf st) . LB.lines) >>= \case
+          Left e -> do
+            logError . T.pack $ "error restoring last notifications: " <> e
+            liftIO exitFailure
+          Right _ -> do
+            renameFile f $ f <> ".bak"
+            logInfo "last notifications restored"
+      where
+        restoreNtf st s' = do
+          TNMRv1 tknId ntf <- liftEither . first (ntfErr "parsing") $ strDecode s
+          liftIO $ storeTokenLastNtf st tknId ntf
+          where
+            s = LB.toStrict s'
+            ntfErr :: Show e => String -> e -> String
+            ntfErr op e = op <> " error (" <> show e <> "): " <> B.unpack (B.take 100 s)
+
 saveServerStats :: M ()
 saveServerStats =
   asks (serverStatsBackupFile . config)

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

@@ -53,6 +53,7 @@ data NtfServerConfig = NtfServerConfig
     subsBatchSize :: Int,
     inactiveClientExpiration :: Maybe ExpirationConfig,
     storeLogFile :: Maybe FilePath,
+    storeLastNtfsFile :: Maybe FilePath,
     ntfCredentials :: ServerCredentials,
     -- stats config - see SMP server config
     logStatsInterval :: Maybe Int64,

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

@@ -85,6 +85,9 @@ ntfServerCLI cfgPath logPath =
           \# and restoring it when the server is started.\n\
           \# Log is compacted on start (deleted objects are removed).\n"
             <> ("enable: " <> onOff enableStoreLog <> "\n\n")
+            <> "# Last notifications are optionally saved and restored when the server restarts,\n\
+               \# they are preserved in the .bak file until the next restart.\n"
+            <> ("restore_last_notifications: " <> onOff enableStoreLog <> "\n\n")
             <> "log_stats: off\n\n\
                \[AUTH]\n\
                \# control_port_admin_password:\n\
@@ -134,6 +137,11 @@ ntfServerCLI cfgPath logPath =
         enableStoreLog = settingIsOn "STORE_LOG" "enable" ini
         logStats = settingIsOn "STORE_LOG" "log_stats" ini
         c = combine cfgPath . ($ defaultX509Config)
+        restoreLastNtfsFile path = case iniOnOff "STORE_LOG" "restore_last_notifications" ini of
+          Just True -> Just path
+          Just False -> Nothing
+          -- if the setting is not set, it is enabled when store log is enabled
+          _ -> enableStoreLog $> path
         serverConfig =
           NtfServerConfig
             { transports = iniTransports ini,
@@ -170,6 +178,7 @@ ntfServerCLI cfgPath logPath =
                       checkInterval = readStrictIni "INACTIVE_CLIENTS" "check_interval" ini
                     },
               storeLogFile = enableStoreLog $> storeLogFilePath,
+              storeLastNtfsFile = restoreLastNtfsFile $ combine logPath "ntf-server-last-notifications.log",
               ntfCredentials =
                 ServerCredentials
                   { caCertificateFile = Just $ c caCrtFile,

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

@@ -28,16 +28,13 @@ import Data.Aeson (ToJSON, (.=))
 import qualified Data.Aeson as J
 import qualified Data.Aeson.Encoding as JE
 import qualified Data.Aeson.TH as JQ
-import qualified Data.Attoparsec.ByteString.Char8 as A
 import Data.Bifunctor (first)
 import qualified Data.ByteString.Base64.URL as U
 import Data.ByteString.Builder (lazyByteString)
 import Data.ByteString.Char8 (ByteString)
-import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
 import Data.Int (Int64)
 import Data.List.NonEmpty (NonEmpty (..))
-import qualified Data.List.NonEmpty as L
 import Data.Map.Strict (Map)
 import Data.Maybe (isNothing)
 import Data.Text (Text)
@@ -51,12 +48,10 @@ import Network.HTTP2.Client (Request)
 import qualified Network.HTTP2.Client as H
 import Network.Socket (HostName, ServiceName)
 import qualified Simplex.Messaging.Crypto as C
-import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Notifications.Server.Push.APNS.Internal
 import Simplex.Messaging.Notifications.Server.Store (NtfTknData (..))
 import Simplex.Messaging.Parsers (defaultJSON)
-import Simplex.Messaging.Protocol (EncNMsgMeta)
 import Simplex.Messaging.Transport.HTTP2 (HTTP2Body (..))
 import Simplex.Messaging.Transport.HTTP2.Client
 import Simplex.Messaging.Util (safeDecodeUtf8, tshow)
@@ -112,30 +107,6 @@ data PushNotification
     PNCheckMessages
   deriving (Show)
 
--- List of PNMessageData uses semicolon-separated encoding instead of strEncode,
--- because strEncode of NonEmpty list uses comma for separator,
--- and encoding of PNMessageData's smpQueue has comma in list of hosts
-encodePNMessages :: NonEmpty PNMessageData -> ByteString
-encodePNMessages = B.intercalate ";" . map strEncode . L.toList
-
-pnMessagesP :: A.Parser (NonEmpty PNMessageData)
-pnMessagesP = L.fromList <$> strP `A.sepBy1` A.char ';'
-
-data PNMessageData = PNMessageData
-  { smpQueue :: SMPQueueNtf,
-    ntfTs :: SystemTime,
-    nmsgNonce :: C.CbNonce,
-    encNMsgMeta :: EncNMsgMeta
-  }
-  deriving (Show)
-
-instance StrEncoding PNMessageData where
-  strEncode PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta} =
-    strEncode (smpQueue, ntfTs, nmsgNonce, encNMsgMeta)
-  strP = do
-    (smpQueue, ntfTs, nmsgNonce, encNMsgMeta) <- strP
-    pure PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta}
-
 data APNSNotification = APNSNotification {aps :: APNSNotificationBody, notificationData :: Maybe J.Value}
   deriving (Show)
 
@@ -221,7 +192,7 @@ defaultAPNSPushClientConfig =
       authKeyFileEnv = "APNS_KEY_FILE", -- the environment variables APNS_KEY_FILE and APNS_KEY_ID must be set, or the server would fail to start
       authKeyAlg = "ES256",
       authKeyIdEnv = "APNS_KEY_ID",
-      paddedNtfLength = 512,
+      paddedNtfLength = 3072,
       appName = "chat.simplex.app",
       appTeamId = "5NN7GUYB6T",
       apnsPort = "443",

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

@@ -1,9 +1,12 @@
+{-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE KindSignatures #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedLists #-}
+{-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE ScopedTypeVariables #-}
 
 module Simplex.Messaging.Notifications.Server.Store where
@@ -12,12 +15,14 @@ import Control.Concurrent.STM
 import Control.Monad
 import Data.ByteString.Char8 (ByteString)
 import Data.Functor (($>))
+import Data.List.NonEmpty (NonEmpty (..), (<|))
 import qualified Data.Map.Strict as M
 import Data.Maybe (catMaybes)
 import Data.Set (Set)
 import qualified Data.Set as S
 import Data.Word (Word16)
 import qualified Simplex.Messaging.Crypto as C
+import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Protocol (NtfPrivateAuthKey, NtfPublicAuthKey, SMPServer)
 import Simplex.Messaging.TMap (TMap)
@@ -30,7 +35,8 @@ data NtfStore = NtfStore
     tokenRegistrations :: TMap DeviceToken (TMap ByteString NtfTokenId),
     subscriptions :: TMap NtfSubscriptionId NtfSubData,
     tokenSubscriptions :: TMap NtfTokenId (TVar (Set NtfSubscriptionId)),
-    subscriptionLookup :: TMap SMPQueueNtf NtfSubscriptionId
+    subscriptionLookup :: TMap SMPQueueNtf NtfSubscriptionId,
+    tokenLastNtfs :: TMap NtfTokenId (TVar (NonEmpty PNMessageData))
   }
 
 newNtfStore :: IO NtfStore
@@ -40,7 +46,8 @@ newNtfStore = do
   subscriptions <- TM.emptyIO
   tokenSubscriptions <- TM.emptyIO
   subscriptionLookup <- TM.emptyIO
-  pure NtfStore {tokens, tokenRegistrations, subscriptions, tokenSubscriptions, subscriptionLookup}
+  tokenLastNtfs <- TM.emptyIO
+  pure NtfStore {tokens, tokenRegistrations, subscriptions, tokenSubscriptions, subscriptionLookup, tokenLastNtfs}
 
 data NtfTknData = NtfTknData
   { ntfTknId :: NtfTokenId,
@@ -111,6 +118,7 @@ removeInactiveTokenRegistrations st NtfTknData {ntfTknId = tId, token} =
       forM_ tIds $ \(regKey, tId') -> do
         TM.delete regKey tknRegs
         TM.delete tId' $ tokens st
+        TM.delete tId' $ tokenLastNtfs st
         void $ deleteTokenSubs st tId'
       pure $ map snd tIds
 
@@ -130,6 +138,7 @@ deleteNtfToken st tknId = do
       TM.lookup token regs $>>= \tIds ->
         TM.delete (regKey tknVerifyKey) tIds
           >> whenM (TM.null tIds) (TM.delete token regs) $> Just ()
+  TM.delete tknId $ tokenLastNtfs st
   deleteTokenSubs st tknId
   where
     regs = tokenRegistrations st
@@ -196,3 +205,42 @@ deleteNtfSubscription st subId = do
           ts_ <- TM.lookup tokenId (tokenSubscriptions st)
           forM_ ts_ $ \ts -> modifyTVar' ts $ S.delete subId
       )
+
+addTokenLastNtf :: NtfStore -> NtfTokenId -> PNMessageData -> IO (NonEmpty PNMessageData)
+addTokenLastNtf st tknId newNtf =
+  TM.lookupIO tknId (tokenLastNtfs st) >>= maybe (atomically maybeNewTokenLastNtfs) (atomically . addNtf)
+  where
+    maybeNewTokenLastNtfs =
+      TM.lookup tknId (tokenLastNtfs st) >>= maybe newTokenLastNtfs addNtf
+    newTokenLastNtfs = do
+      v <- newTVar [newNtf]
+      TM.insert tknId v $ tokenLastNtfs st
+      pure [newNtf]
+    addNtf v =
+      stateTVar v $ \ntfs -> let !ntfs' = rebuildList ntfs in (ntfs', ntfs')
+      where
+        rebuildList :: NonEmpty PNMessageData -> NonEmpty PNMessageData
+        rebuildList = foldr keepPrevNtf [newNtf]
+          where
+            PNMessageData {smpQueue = newNtfQ} = newNtf
+            keepPrevNtf ntf@PNMessageData {smpQueue} ntfs
+              | smpQueue /= newNtfQ && length ntfs < maxNtfs = ntf <| ntfs
+              | otherwise = ntfs
+        maxNtfs = 6
+
+-- This function is expected to be called after store log is read,
+-- as it checks for token existence when adding last notification.
+storeTokenLastNtf :: NtfStore -> NtfTokenId -> PNMessageData -> IO ()
+storeTokenLastNtf (NtfStore {tokens, tokenLastNtfs}) tknId ntf = do
+  TM.lookupIO tknId tokenLastNtfs >>= atomically . maybe newTokenLastNtfs (`modifyTVar'` (ntf <|))
+  where
+    newTokenLastNtfs = TM.lookup tknId tokenLastNtfs >>= maybe insertForExistingToken (`modifyTVar'` (ntf <|))
+    insertForExistingToken =
+      whenM (TM.member tknId tokens) $
+        TM.insertM tknId (newTVar [ntf]) tokenLastNtfs
+
+data TokenNtfMessageRecord = TNMRv1 NtfTokenId PNMessageData
+
+instance StrEncoding TokenNtfMessageRecord where
+  strEncode (TNMRv1 tknId ntf) = strEncode (Str "v1", tknId, ntf)
+  strP = "v1 " *> (TNMRv1 <$> strP_ <*> strP)

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

@@ -37,8 +37,7 @@ import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Notifications.Server.Store
 import Simplex.Messaging.Protocol (NtfPrivateAuthKey)
 import Simplex.Messaging.Server.StoreLog
-import Simplex.Messaging.Util (safeDecodeUtf8, whenM)
-import System.Directory (doesFileExist, renameFile)
+import Simplex.Messaging.Util (safeDecodeUtf8)
 import System.IO
 
 data NtfStoreLogRecord
@@ -186,13 +185,7 @@ logDeleteSubscription :: StoreLog 'WriteMode -> NtfSubscriptionId -> IO ()
 logDeleteSubscription s subId = logNtfStoreRecord s $ DeleteSubscription subId
 
 readWriteNtfStore :: FilePath -> NtfStore -> IO (StoreLog 'WriteMode)
-readWriteNtfStore f st = do
-  whenM (doesFileExist f) $ do
-    readNtfStore f st
-    renameFile f $ f <> ".bak"
-  s <- openWriteStoreLog f
-  writeNtfStore s st
-  pure s
+readWriteNtfStore = readWriteStoreLog readNtfStore writeNtfStore
 
 readNtfStore :: FilePath -> NtfStore -> IO ()
 readNtfStore f st = mapM_ (addNtfLogRecord . LB.toStrict) . LB.lines =<< LB.readFile f

src/Simplex/Messaging/Protocol.hs

@@ -1198,6 +1198,8 @@ data ErrorType
     CRYPTO
   | -- | SMP queue capacity is exceeded on the server
     QUOTA
+  | -- | SMP server storage error
+    STORE {storeErr :: String}
   | -- | ACK command is sent without message to be acknowledged
     NO_MSG
   | -- | sent message is too large (> maxMessageLength = 16088 bytes)
@@ -1513,6 +1515,7 @@ instance Encoding ErrorType where
     AUTH -> "AUTH"
     CRYPTO -> "CRYPTO"
     QUOTA -> "QUOTA"
+    STORE err -> "STORE " <> smpEncode err
     EXPIRED -> "EXPIRED"
     NO_MSG -> "NO_MSG"
     LARGE_MSG -> "LARGE_MSG"
@@ -1528,6 +1531,7 @@ instance Encoding ErrorType where
       "AUTH" -> pure AUTH
       "CRYPTO" -> pure CRYPTO
       "QUOTA" -> pure QUOTA
+      "STORE" -> STORE <$> _smpP
       "EXPIRED" -> pure EXPIRED
       "NO_MSG" -> pure NO_MSG
       "LARGE_MSG" -> pure LARGE_MSG

src/Simplex/Messaging/Server.hs

@@ -160,7 +160,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
     runServer (tcpPort, ATransport t, addHTTP) = do
       smpCreds <- asks tlsServerCreds
       httpCreds_ <- asks httpServerCreds
-      ss <- asks sockets
+      ss <- liftIO newSocketState
+      asks sockets >>= atomically . (`modifyTVar'` ((tcpPort, ss) :))
       serverSignKey <- either fail pure $ fromTLSCredentials smpCreds
       env <- ask
       liftIO $ case (httpCreds_, attachHTTP_) of
@@ -190,6 +191,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
 
     stopServer :: Server -> M ()
     stopServer s = do
+      asks serverActive >>= atomically . (`writeTVar` False)
       logInfo "Saving server state..."
       withLock' (savingLock s) "final" $ saveServer False >> closeServer
       logInfo "Server stopped"
@@ -657,25 +659,28 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} attachHT
 #else
                 hPutStrLn h "Not available on GHC 8.10"
 #endif
-              CPSockets -> withUserRole $ do
-                (accepted', closed', active') <- unliftIO u $ asks sockets
-                (accepted, closed, active) <- (,,) <$> readTVarIO accepted' <*> readTVarIO closed' <*> readTVarIO active'
-                hPutStrLn h "Sockets: "
-                hPutStrLn h $ "accepted: " <> show accepted
-                hPutStrLn h $ "closed: " <> show closed
-                hPutStrLn h $ "active: " <> show (IM.size active)
-                hPutStrLn h $ "leaked: " <> show (accepted - closed - IM.size active)
+              CPSockets -> withUserRole $ unliftIO u (asks sockets) >>= readTVarIO >>= mapM_ putSockets
+                where
+                  putSockets (tcpPort, (accepted', closed', active')) = do
+                    (accepted, closed, active) <- (,,) <$> readTVarIO accepted' <*> readTVarIO closed' <*> readTVarIO active'
+                    hPutStrLn h $ "Sockets for port " <> tcpPort <> ":"
+                    hPutStrLn h $ "accepted: " <> show accepted
+                    hPutStrLn h $ "closed: " <> show closed
+                    hPutStrLn h $ "active: " <> show (IM.size active)
+                    hPutStrLn h $ "leaked: " <> show (accepted - closed - IM.size active)
               CPSocketThreads -> withAdminRole $ do
 #if MIN_VERSION_base(4,18,0)
-                (_, _, active') <- unliftIO u $ asks sockets
-                active <- readTVarIO active'
-                forM_ (IM.toList active) $ \(sid, tid') ->
-                  deRefWeak tid' >>= \case
-                    Nothing -> hPutStrLn h $ intercalate "," [show sid, "", "gone", ""]
-                    Just tid -> do
-                      label <- threadLabel tid
-                      status <- threadStatus tid
-                      hPutStrLn h $ intercalate "," [show sid, show tid, show status, fromMaybe "" label]
+                unliftIO u (asks sockets) >>= readTVarIO >>= mapM_ putSocketThreads
+                where
+                  putSocketThreads (tcpPort, (_, _, active')) = do
+                    active <- readTVarIO active'
+                    forM_ (IM.toList active) $ \(sid, tid') ->
+                      deRefWeak tid' >>= \case
+                        Nothing -> hPutStrLn h $ intercalate "," [tcpPort, show sid, "", "gone", ""]
+                        Just tid -> do
+                          label <- threadLabel tid
+                          status <- threadStatus tid
+                          hPutStrLn h $ intercalate "," [tcpPort, show sid, show tid, show status, fromMaybe "" label]
 #else
                 hPutStrLn h "Not available on GHC 8.10"
 #endif
@@ -830,12 +835,13 @@ clientDisconnected c@Client {clientId, subscriptions, ntfSubscriptions, connecte
   subs <- atomically $ swapTVar subscriptions M.empty
   ntfSubs <- atomically $ swapTVar ntfSubscriptions M.empty
   liftIO $ mapM_ cancelSub subs
-  Server {subscribers, notifiers, subClients, ntfSubClients} <- asks server
-  liftIO $ updateSubscribers subs subscribers
-  liftIO $ updateSubscribers ntfSubs notifiers
-  asks clients >>= atomically . (`modifyTVar'` IM.delete clientId)
-  atomically $ modifyTVar' subClients $ IM.delete clientId
-  atomically $ modifyTVar' ntfSubClients $ IM.delete clientId
+  whenM (asks serverActive >>= readTVarIO) $ do
+    Server {subscribers, notifiers, subClients, ntfSubClients} <- asks server
+    liftIO $ updateSubscribers subs subscribers
+    liftIO $ updateSubscribers ntfSubs notifiers
+    asks clients >>= atomically . (`modifyTVar'` IM.delete clientId)
+    atomically $ modifyTVar' subClients $ IM.delete clientId
+    atomically $ modifyTVar' ntfSubClients $ IM.delete clientId
   tIds <- atomically $ swapTVar endThreads IM.empty
   liftIO $ mapM_ (mapM_ killThread <=< deRefWeak) tIds
   where
@@ -862,8 +868,10 @@ cancelSub s = case subThread s of
 receive :: Transport c => THandleSMP c 'TServer -> Client -> M ()
 receive h@THandle {params = THandleParams {thAuth}} Client {rcvQ, sndQ, rcvActiveAt, sessionId} = do
   labelMyThread . B.unpack $ "client $" <> encode sessionId <> " receive"
+  sa <- asks serverActive
   forever $ do
     ts <- L.toList <$> liftIO (tGet h)
+    unlessM (readTVarIO sa) $ throwIO $ userError "server stopped"
     atomically . (writeTVar rcvActiveAt $!) =<< liftIO getSystemTime
     stats <- asks serverStats
     (errs, cmds) <- partitionEithers <$> mapM (cmdAction stats) ts

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

@@ -19,8 +19,6 @@ import Data.IntMap.Strict (IntMap)
 import qualified Data.IntMap.Strict as IM
 import Data.List (intercalate)
 import Data.List.NonEmpty (NonEmpty)
-import Data.Map.Strict (Map)
-import qualified Data.Map.Strict as M
 import Data.Maybe (isJust, isNothing)
 import qualified Data.Text as T
 import Data.Time.Clock (getCurrentTime)
@@ -39,7 +37,7 @@ import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Server.Information
 import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.NtfStore
-import Simplex.Messaging.Server.QueueStore (NtfCreds (..), QueueRec (..))
+import Simplex.Messaging.Server.QueueStore (QueueRec (..))
 import Simplex.Messaging.Server.QueueStore.STM
 import Simplex.Messaging.Server.Stats
 import Simplex.Messaging.Server.StoreLog
@@ -138,6 +136,7 @@ defaultProxyClientConcurrency = 32
 
 data Env = Env
   { config :: ServerConfig,
+    serverActive :: TVar Bool,
     serverInfo :: ServerInformation,
     server :: Server,
     serverIdentity :: KeyHash,
@@ -149,7 +148,7 @@ data Env = Env
     tlsServerCreds :: T.Credential,
     httpServerCreds :: Maybe T.Credential,
     serverStats :: ServerStats,
-    sockets :: SocketState,
+    sockets :: TVar [(ServiceName, SocketState)],
     clientSeq :: TVar ClientId,
     clients :: TVar (IntMap (Maybe Client)),
     proxyAgent :: ProxyAgent -- senders served on this proxy
@@ -243,6 +242,7 @@ newProhibitedSub = do
 
 newEnv :: ServerConfig -> IO Env
 newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, smpAgentCfg, information, messageExpiration} = do
+  serverActive <- newTVarIO True
   server <- newServer
   queueStore <- newQueueStore
   msgStore <- newMsgStore
@@ -251,18 +251,18 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, smpAg
   storeLog <-
     forM storeLogFile $ \f -> do
       logInfo $ "restoring queues from file " <> T.pack f
-      restoreQueues queueStore f
+      readWriteQueueStore f queueStore
   tlsServerCreds <- getCredentials "SMP" smpCredentials
   httpServerCreds <- mapM (getCredentials "HTTPS") httpCredentials
   mapM_ checkHTTPSCredentials httpServerCreds
   Fingerprint fp <- loadFingerprint smpCredentials
   let serverIdentity = KeyHash fp
   serverStats <- newServerStats =<< getCurrentTime
-  sockets <- newSocketState
+  sockets <- newTVarIO []
   clientSeq <- newTVarIO 0
   clients <- newTVarIO mempty
   proxyAgent <- newSMPProxyAgent smpAgentCfg random
-  pure Env {config, serverInfo, server, serverIdentity, queueStore, msgStore, ntfStore, random, storeLog, tlsServerCreds, httpServerCreds, serverStats, sockets, clientSeq, clients, proxyAgent}
+  pure Env {serverActive, config, serverInfo, server, serverIdentity, queueStore, msgStore, ntfStore, random, storeLog, tlsServerCreds, httpServerCreds, serverStats, sockets, clientSeq, clients, proxyAgent}
   where
     getCredentials protocol creds = do
       files <- missingCreds
@@ -284,19 +284,6 @@ newEnv config@ServerConfig {smpCredentials, httpCredentials, storeLogFile, smpAg
           putStrLn $ "Error: unsupported HTTPS credentials, required 4096-bit RSA\n" <> letsEncrypt
           exitFailure
     letsEncrypt = "Use Let's Encrypt to generate: certbot certonly --standalone -d yourdomainname --key-type rsa --rsa-key-size 4096"
-    restoreQueues :: QueueStore -> FilePath -> IO (StoreLog 'WriteMode)
-    restoreQueues QueueStore {queues, senders, notifiers} f = do
-      (qs, s) <- readWriteStoreLog f
-      atomically . writeTVar queues =<< mapM newTVarIO qs
-      atomically $ writeTVar senders $! M.foldr' addSender M.empty qs
-      atomically $ writeTVar notifiers $! M.foldr' addNotifier M.empty qs
-      pure s
-    addSender :: QueueRec -> Map SenderId RecipientId -> Map SenderId RecipientId
-    addSender q = M.insert (senderId q) (recipientId q)
-    addNotifier :: QueueRec -> Map NotifierId RecipientId -> Map NotifierId RecipientId
-    addNotifier q = case notifier q of
-      Nothing -> id
-      Just NtfCreds {notifierId} -> M.insert notifierId (recipientId q)
     serverInfo =
       ServerInformation
         { information,

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

@@ -47,11 +47,11 @@ newQueueStore = do
   pure QueueStore {queues, senders, notifiers}
 
 addQueue :: QueueStore -> QueueRec -> IO (Either ErrorType ())
-addQueue QueueStore {queues, senders} q@QueueRec {recipientId = rId, senderId = sId} = atomically $ do
+addQueue QueueStore {queues, senders, notifiers} q@QueueRec {recipientId = rId, senderId = sId, notifier} = atomically $ do
   ifM hasId (pure $ Left DUPLICATE_) $ do
-    qVar <- newTVar q
-    TM.insert rId qVar queues
+    TM.insertM rId (newTVar q) queues
     TM.insert sId rId senders
+    forM_ notifier $ \NtfCreds {notifierId} -> TM.insert notifierId rId notifiers
     pure $ Right ()
   where
     hasId = (||) <$> TM.member rId queues <*> TM.member sId senders

src/Simplex/Messaging/Server/StoreLog.hs

@@ -1,3 +1,4 @@
+{-# LANGUAGE BangPatterns #-}
 {-# LANGUAGE DataKinds #-}
 {-# LANGUAGE DuplicateRecordFields #-}
 {-# LANGUAGE GADTs #-}
@@ -22,23 +23,27 @@ module Simplex.Messaging.Server.StoreLog
     logDeleteQueue,
     logDeleteNotifier,
     logUpdateQueueTime,
+    readWriteQueueStore,
     readWriteStoreLog,
   )
 where
 
 import Control.Applicative (optional, (<|>))
-import Control.Monad (foldM, unless, when)
+import Control.Concurrent.STM
+import Control.Logger.Simple
+import Control.Monad
 import qualified Data.Attoparsec.ByteString.Char8 as A
 import qualified Data.ByteString.Char8 as B
 import qualified Data.ByteString.Lazy.Char8 as LB
-import Data.Functor (($>))
-import Data.Map.Strict (Map)
-import qualified Data.Map.Strict as M
+import qualified Data.ByteString.Base64.URL as B64
+import qualified Data.Text as T
+import Data.Time.Clock (getCurrentTime)
+import Data.Time.Format.ISO8601 (iso8601Show)
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.QueueStore
-import Simplex.Messaging.Transport.Buffer (trimCR)
-import Simplex.Messaging.Util (ifM)
+import Simplex.Messaging.Server.QueueStore.STM
+import Simplex.Messaging.Util (bshow, ifM, unlessM, whenM)
 import System.Directory (doesFileExist, renameFile)
 import System.IO
 
@@ -144,7 +149,7 @@ openWriteStoreLog f = do
 
 openReadStoreLog :: FilePath -> IO (StoreLog 'ReadMode)
 openReadStoreLog f = do
-  doesFileExist f >>= (`unless` writeFile f "")
+  unlessM (doesFileExist f) (writeFile f "")
   ReadStoreLog f <$> openFile f ReadMode
 
 storeLogFilePath :: StoreLog a -> FilePath
@@ -183,37 +188,69 @@ logDeleteNotifier s = writeStoreLogRecord s . DeleteNotifier
 logUpdateQueueTime :: StoreLog 'WriteMode -> QueueId -> RoundedSystemTime -> IO ()
 logUpdateQueueTime s qId t = writeStoreLogRecord s $ UpdateTime qId t
 
-readWriteStoreLog :: FilePath -> IO (Map RecipientId QueueRec, StoreLog 'WriteMode)
-readWriteStoreLog f = do
-  qs <- ifM (doesFileExist f) readQS (pure M.empty)
-  s <- openWriteStoreLog f
-  writeQueues s qs
-  pure (qs, s)
-  where
-    readQS = readQueues f <* renameFile f (f <> ".bak")
+readWriteQueueStore :: FilePath -> QueueStore -> IO (StoreLog 'WriteMode)
+readWriteQueueStore = readWriteStoreLog readQueues writeQueues
 
-writeQueues :: StoreLog 'WriteMode -> Map RecipientId QueueRec -> IO ()
-writeQueues s = mapM_ $ \q -> when (active q) $ logCreateQueue s q
+readWriteStoreLog :: (FilePath -> s -> IO ()) -> (StoreLog 'WriteMode -> s -> IO ()) -> FilePath -> s -> IO (StoreLog 'WriteMode)
+readWriteStoreLog readStore writeStore f st =
+  ifM
+    (doesFileExist tempBackup)
+    (useTempBackup >> readWriteLog)
+    (ifM (doesFileExist f) readWriteLog (writeLog "creating store log..."))
   where
+    f' = T.pack f
+    tempBackup = f <> ".start"
+    useTempBackup = do
+      -- preserve current file, use temp backup
+      logWarn $ "Server terminated abnormally on last start, restoring state from " <> T.pack tempBackup
+      whenM (doesFileExist f) $ do
+        renameFile f (f <> ".bak")
+        logInfo $ "preserved incomplete state " <> f' <> " as " <> (f' <> ".bak")
+      renameFile tempBackup f
+    readWriteLog = do
+      -- log backup is made in two steps to mitigate the crash during the compacting.
+      -- Temporary backup file .start will be used when it is present.
+      readStore f st
+      renameFile f tempBackup -- 1) make temp backup
+      s <- writeLog "compacting store log (do not terminate)..." -- 2) save state
+      renameBackup -- 3) timed backup
+      pure s
+    writeLog msg = do
+      s <- openWriteStoreLog f
+      logInfo msg
+      writeStore s st
+      pure s
+    renameBackup = do
+      ts <- getCurrentTime
+      let timedBackup = f <> "." <> iso8601Show ts
+      renameFile tempBackup timedBackup
+      logInfo $ "original state preserved as " <> T.pack timedBackup
+
+writeQueues :: StoreLog 'WriteMode -> QueueStore -> IO ()
+writeQueues s st = readTVarIO (queues st) >>= mapM_ writeQueue
+  where
+    writeQueue v = readTVarIO v >>= \q -> when (active q) $ logCreateQueue s q
     active QueueRec {status} = status == QueueActive
 
-readQueues :: FilePath -> IO (Map RecipientId QueueRec)
-readQueues f = foldM processLine M.empty . LB.lines =<< LB.readFile f
+readQueues :: FilePath -> QueueStore -> IO ()
+readQueues f st = withFile f ReadMode $ LB.hGetContents >=> mapM_ processLine . LB.lines
   where
-    processLine :: Map RecipientId QueueRec -> LB.ByteString -> IO (Map RecipientId QueueRec)
-    processLine m s' = case strDecode $ trimCR s of
-      Right r -> pure $ procLogRecord r
-      Left e -> printError e $> m
+    processLine :: LB.ByteString -> IO ()
+    processLine s' = either printError procLogRecord (strDecode s)
       where
         s = LB.toStrict s'
-        procLogRecord :: StoreLogRecord -> Map RecipientId QueueRec
+        procLogRecord :: StoreLogRecord -> IO ()
         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
-          UpdateTime qId t -> M.adjust (\q -> q {updatedAt = Just t}) qId m
+          CreateQueue q -> addQueue st q >>= qError "create" (recipientId q)
+          SecureQueue qId sKey -> secureQueue st qId sKey >>= qError "secure" qId
+          AddNotifier qId ntfCreds -> addQueueNotifier st qId ntfCreds >>= qError "addNotifier" qId
+          SuspendQueue qId -> suspendQueue st qId >>= qError "suspend" qId
+          DeleteQueue qId -> deleteQueue st qId >>= qError "delete" qId
+          DeleteNotifier qId -> deleteQueueNotifier st qId >>= qError "deleteNotifier" qId
+          UpdateTime qId t -> updateQueueTime st qId t
         printError :: String -> IO ()
         printError e = B.putStrLn $ "Error parsing log: " <> B.pack e <> " - " <> s
+        qError :: B.ByteString -> RecipientId -> Either ErrorType a -> IO ()
+        qError op (EntityId qId) = \case
+          Left e -> B.putStrLn $ op <> " stored queue " <> B64.encode qId <> " error: " <> bshow e
+          Right _ -> pure ()

src/Simplex/Messaging/Transport/Server.hs

@@ -132,7 +132,7 @@ runTCPServerSocket :: SocketState -> TMVar Bool -> IO Socket -> (Socket -> IO ()
 runTCPServerSocket (accepted, gracefullyClosed, clients) started getSocket server =
   E.bracket getSocket (closeServer started clients) $ \sock ->
     forever . E.bracketOnError (safeAccept sock) (close . fst) $ \(conn, _peer) -> do
-      cId <- atomically $ stateTVar accepted $ \cId -> let cId' = cId + 1 in cId `seq` (cId', cId')
+      cId <- atomically $ stateTVar accepted $ \cId -> let cId' = cId + 1 in cId' `seq` (cId', cId')
       let closeConn _ = do
             atomically $ modifyTVar' clients $ IM.delete cId
             gracefulClose conn 5000 `catchAll_` pure () -- catchAll_ is needed here in case the connection was closed earlier
@@ -168,8 +168,8 @@ newSocketState = (,,) <$> newTVarIO 0 <*> newTVarIO 0 <*> newTVarIO mempty
 
 closeServer :: TMVar Bool -> TVar (IntMap (Weak ThreadId)) -> Socket -> IO ()
 closeServer started clients sock = do
-  readTVarIO clients >>= mapM_ (deRefWeak >=> mapM_ killThread)
   close sock
+  readTVarIO clients >>= mapM_ (deRefWeak >=> mapM_ killThread)
   void . atomically $ tryPutTMVar started False
 
 startTCPServer :: TMVar Bool -> Maybe HostName -> ServiceName -> IO Socket

tests/AgentTests/FunctionalAPITests.hs

@@ -1712,7 +1712,7 @@ testOnlyCreatePullSlowHandshake = withAgentClientsCfg2 agentProxyCfgV8 agentProx
 getMsg :: AgentClient -> ConnId -> ExceptT AgentErrorType IO a -> ExceptT AgentErrorType IO a
 getMsg c cId action = do
   liftIO $ noMessages c "nothing should be delivered before GET"
-  Just _ <- getConnectionMessage c cId
+  [Just _] <- lift $ getConnectionMessages c [cId]
   action
 
 getMSGNTF :: AgentClient -> ConnId -> ExceptT AgentErrorType IO ()

tests/AgentTests/NotificationTests.hs

@@ -3,6 +3,7 @@
 {-# LANGUAGE GADTs #-}
 {-# LANGUAGE LambdaCase #-}
 {-# LANGUAGE NamedFieldPuns #-}
+{-# LANGUAGE OverloadedLists #-}
 {-# LANGUAGE OverloadedStrings #-}
 {-# LANGUAGE PatternSynonyms #-}
 {-# LANGUAGE QuasiQuotes #-}
@@ -49,6 +50,7 @@ import Data.Bifunctor (bimap, first)
 import qualified Data.ByteString.Base64.URL as U
 import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
+import Data.List.NonEmpty (NonEmpty (..))
 import qualified Data.List.NonEmpty as L
 import Data.Text.Encoding (encodeUtf8)
 import Database.SQLite.Simple.QQ (sql)
@@ -490,8 +492,11 @@ testNotificationSubscriptionExistingConnection apns baseId alice@AgentClient {ag
     (nonce, message) <- messageNotification apns tkn
     pure (bobId, aliceId, nonce, message)
 
-  -- alice client already has subscription for the connection
-  Left (CMD PROHIBITED _) <- runExceptT $ getNotificationMessage alice nonce message
+  Right [NotificationInfo {ntfConnId = cId}] <- runExceptT $ getNotificationConns alice nonce message
+  cId `shouldBe` bobId
+  -- alice client already has subscription for the connection,
+  -- so get fails with CMD PROHIBITED (transformed into Nothing in catch)
+  [Nothing] <- getConnectionMessages alice [cId]
 
   threadDelay 500000
   suspendAgent alice 0
@@ -500,8 +505,8 @@ testNotificationSubscriptionExistingConnection apns baseId alice@AgentClient {ag
   putStrLn "before opening the database from another agent"
 
   -- aliceNtf client doesn't have subscription and is allowed to get notification message
-  withAgent 3 aliceCfg initAgentServers testDB $ \aliceNtf -> runRight_ $ do
-    (_, Just SMPMsgMeta {msgFlags = MsgFlags True}) <- getNotificationMessage aliceNtf nonce message
+  withAgent 3 aliceCfg initAgentServers testDB $ \aliceNtf -> do
+    (Just SMPMsgMeta {msgFlags = MsgFlags True}) :| _ <- getConnectionMessages aliceNtf [cId]
     pure ()
 
   threadDelay 1000000

tests/CoreTests/StoreLogTests.hs

@@ -21,6 +21,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol
 import Simplex.Messaging.Server.StoreLog
 import Simplex.Messaging.Server.QueueStore
+import Simplex.Messaging.Server.QueueStore.STM (QueueStore (..), newQueueStore)
 import Test.Hspec
 
 testNewQueueRec :: TVar ChaChaDRG -> Bool -> IO QueueRec
@@ -116,8 +117,11 @@ testSMPStoreLog testSuite tests =
     replicateM_ 3 $ testReadWrite t
   where
     testReadWrite SLTC {compacted, state} = do
-      (state', l) <- readWriteStoreLog testStoreLogFile
-      state' `shouldBe` state
+      st <- newQueueStore
+      l <- readWriteQueueStore testStoreLogFile st
+      storeState st `shouldReturn` state
       closeStoreLog l
       ([], compacted') <- partitionEithers . map strDecode . B.lines <$> B.readFile testStoreLogFile
       compacted' `shouldBe` compacted
+    storeState :: QueueStore -> IO (M.Map RecipientId QueueRec)
+    storeState st = readTVarIO (queues st) >>= mapM readTVarIO

tests/NtfClient.hs

@@ -75,6 +75,9 @@ testKeyHash = "LcJUMfVhwD8yxjAiSaDzzGF3-kLG4Uh0Fl_ZIjrRwjI="
 ntfTestStoreLogFile :: FilePath
 ntfTestStoreLogFile = "tests/tmp/ntf-server-store.log"
 
+ntfTestStoreLastNtfsFile :: FilePath
+ntfTestStoreLastNtfsFile = "tests/tmp/ntf-server-last-notifications.log"
+
 testNtfClient :: Transport c => (THandleNTF c 'TClient -> IO a) -> IO a
 testNtfClient client = do
   Right host <- pure $ chooseTransportHost defaultNetworkConfig testHost
@@ -104,6 +107,7 @@ ntfServerCfg =
       subsBatchSize = 900,
       inactiveClientExpiration = Just defaultInactiveClientExpiration,
       storeLogFile = Nothing,
+      storeLastNtfsFile = Nothing,
       ntfCredentials =
         ServerCredentials
           { caCertificateFile = Just "tests/fixtures/ca.crt",
@@ -131,7 +135,7 @@ ntfServerCfgVPrev =
     serverVRange' = serverVRange smpCfg'
 
 withNtfServerStoreLog :: ATransport -> (ThreadId -> IO a) -> IO a
-withNtfServerStoreLog t = withNtfServerCfg ntfServerCfg {storeLogFile = Just ntfTestStoreLogFile, transports = [(ntfTestPort, t, False)]}
+withNtfServerStoreLog t = withNtfServerCfg ntfServerCfg {storeLogFile = Just ntfTestStoreLogFile, storeLastNtfsFile = Just ntfTestStoreLastNtfsFile, transports = [(ntfTestPort, t, False)]}
 
 withNtfServerThreadOn :: HasCallStack => ATransport -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
 withNtfServerThreadOn t port' = withNtfServerCfg ntfServerCfg {transports = [(port', t, False)]}

tests/NtfServerTests.hs

@@ -38,7 +38,6 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Notifications.Server.Push.APNS
-import qualified Simplex.Messaging.Notifications.Server.Push.APNS as APNS
 import Simplex.Messaging.Notifications.Transport (THandleNTF)
 import Simplex.Messaging.Parsers (parse, parseAll)
 import Simplex.Messaging.Protocol hiding (notification)
@@ -136,7 +135,7 @@ testNotificationSubscription (ATransport t) =
               Right message = ntfData' .-> "message"
               Right ntfDataDecrypted = C.cbDecrypt dhSecret nonce' message
               Right pnMsgs1 = parse pnMessagesP (AP.INTERNAL "error parsing PNMessageData") ntfDataDecrypted
-              APNS.PNMessageData {smpQueue = SMPQueueNtf {smpServer, notifierId}, nmsgNonce, encNMsgMeta} = L.last pnMsgs1
+              PNMessageData {smpQueue = SMPQueueNtf {smpServer, notifierId}, nmsgNonce, encNMsgMeta} = L.last pnMsgs1
               Right nMsgMeta = C.cbDecrypt rcvNtfDhSecret nmsgNonce encNMsgMeta
               Right NMsgMeta {msgId, msgTs} = parse smpP (AP.INTERNAL "error parsing NMsgMeta") nMsgMeta
           smpServer `shouldBe` srv
@@ -167,6 +166,6 @@ testNotificationSubscription (ATransport t) =
               Right message3 = ntfData3 .-> "message"
               Right ntfDataDecrypted3 = C.cbDecrypt dhSecret nonce3 message3
               Right pnMsgs2 = parse pnMessagesP (AP.INTERNAL "error parsing PNMessageData") ntfDataDecrypted3
-              APNS.PNMessageData {smpQueue = SMPQueueNtf {smpServer = smpServer3, notifierId = notifierId3}} = L.last pnMsgs2
+              PNMessageData {smpQueue = SMPQueueNtf {smpServer = smpServer3, notifierId = notifierId3}} = L.last pnMsgs2
           smpServer3 `shouldBe` srv
           notifierId3 `shouldBe` nId