6.1.2.0

* smp server: reduce memory usage when starting

* servers: prevent losing state when compacting is incomplete (terminated abnormally)

* fix test

* fix/refactor

* message

* remove .bak

* simpler

CHANGELOG.md

@@ -1,3 +1,13 @@
+# 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:

package.yaml

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

simplexmq.cabal

@@ -5,7 +5,7 @@ cabal-version: 1.12
 -- see: https://github.com/sol/hpack
 
 name:           simplexmq
-version:        6.1.1.0
+version:        6.1.2.0
 synopsis:       SimpleXMQ message broker
 description:    This package includes <./docs/Simplex-Messaging-Server.html server>,
                 <./docs/Simplex-Messaging-Client.html client> and
@@ -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/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
@@ -253,7 +251,7 @@ 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
@@ -286,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

@@ -49,8 +49,7 @@ newQueueStore = do
 addQueue :: QueueStore -> QueueRec -> IO (Either ErrorType ())
 addQueue QueueStore {queues, senders} q@QueueRec {recipientId = rId, senderId = sId} = 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
     pure $ Right ()
   where

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 (recipientId q)
+          SecureQueue qId sKey -> secureQueue st qId sKey >>= qError qId
+          AddNotifier qId ntfCreds -> addQueueNotifier st qId ntfCreds >>= qError qId
+          SuspendQueue qId -> suspendQueue st qId >>= qError qId
+          DeleteQueue qId -> deleteQueue st qId >>= qError qId
+          DeleteNotifier qId -> deleteQueueNotifier st qId >>= qError qId
+          UpdateTime qId t -> updateQueueTime st qId t
         printError :: String -> IO ()
         printError e = B.putStrLn $ "Error parsing log: " <> B.pack e <> " - " <> s
+        qError :: RecipientId -> Either ErrorType a -> IO ()
+        qError (EntityId qId) = \case
+          Left e -> B.putStrLn $ "stored queue " <> B64.encode qId <> " error: " <> bshow e
+          Right _ -> pure ()

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