smp server: fix in-memory server not restoring queue/service associations after 2+ restarts (#1618)

* smp server: fix in-memory server not restoring queue/service associations after 2+ restarts

* fix test, do not reuse database

Author: epoberezkin

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

@@ -114,16 +114,19 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where
       serviceQueuesCount serviceSel = foldM (\n s -> (n +) . S.size <$> readTVarIO (serviceSel s)) 0
 
   addQueue_ :: STMQueueStore q -> (RecipientId -> QueueRec -> IO q) -> RecipientId -> QueueRec -> IO (Either ErrorType q)
-  addQueue_ st mkQ rId qr@QueueRec {senderId = sId, notifier, queueData} = do
+  addQueue_ st mkQ rId qr@QueueRec {senderId = sId, notifier, queueData, rcvServiceId} = do
     sq <- mkQ rId qr
     add sq $>> withLog "addStoreQueue" st (\s -> logCreateQueue s rId qr) $> Right sq
     where
       STMQueueStore {queues, senders, notifiers, links} = st
       add q = atomically $ ifM hasId (pure $ Left DUPLICATE_) $ Right () <$ do
         TM.insert rId q queues
         TM.insert sId rId senders
-        forM_ notifier $ \NtfCreds {notifierId} -> TM.insert notifierId rId notifiers
+        forM_ notifier $ \NtfCreds {notifierId = nId, ntfServiceId} -> do
+          TM.insert nId rId notifiers
+          mapM_ (addServiceQueue st serviceNtfQueues nId) ntfServiceId
         forM_ queueData $ \(lnkId, _) -> TM.insert lnkId rId links
+        mapM_ (addServiceQueue st serviceRcvQueues rId) rcvServiceId
       hasId = anyM [TM.member rId queues, TM.member sId senders, hasNotifier, hasLink]
       hasNotifier = maybe (pure False) (\NtfCreds {notifierId} -> TM.member notifierId notifiers) notifier
       hasLink = maybe (pure False) (\(lnkId, _) -> TM.member lnkId links) queueData

tests/ServerTests.hs

@@ -86,6 +86,7 @@ serverTests = do
   describe "Message notifications" $ do
     testMessageNotifications
     testMessageServiceNotifications
+    testServiceNotificationsTwoRestarts
   describe "Message expiration" $ do
     testMsgExpireOnSend
     testMsgExpireOnInterval
@@ -1091,7 +1092,6 @@ testMessageServiceNotifications =
       (rcvNtfPubDhKey, _) <- atomically $ C.generateKeyPair g
       Resp "1" _ (NID nId _) <- signSendRecv rh rKey ("1", rId, NKEY nPub rcvNtfPubDhKey)
       serviceKeys@(_, servicePK) <- atomically $ C.generateKeyPair g
-      -- TODO [certs] we need to get certificate fingerprint and include it into signed over for NSUB commands
       testNtfServiceClient t serviceKeys $ \nh1 -> do
         -- can't subscribe without service signature in service connection
         Resp "2a" _ (ERR SERVICE) <- signSendRecv nh1 nKey ("2a", nId, NSUB)
@@ -1155,6 +1155,51 @@ testMessageServiceNotifications =
         Resp "" _ (NMSG _ _) <- tGet1 nh
         pure ()
 
+testServiceNotificationsTwoRestarts :: SpecWith (ASrvTransport, AStoreType)
+testServiceNotificationsTwoRestarts =
+  it "subscribe notifier as service and deliver notifications after two restarts" $ \ps@(ATransport t, _) -> do
+    g <- C.newRandom
+    (sPub, sKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
+    (nPub, nKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
+    serviceKeys@(_, servicePK) <- atomically $ C.generateKeyPair g
+    (rcvNtfPubDhKey, _) <- atomically $ C.generateKeyPair g
+    (rId, rKey, sId, dec, serviceId) <- withSmpServerStoreLogOn ps testPort $ runTest2 t $ \sh rh -> do
+      (sId, rId, rKey, dhShared) <- createAndSecureQueue rh sPub
+      let dec = decryptMsgV3 dhShared
+      Resp "0" _ (NID nId _) <- signSendRecv rh rKey ("0", rId, NKEY nPub rcvNtfPubDhKey)
+      testNtfServiceClient t serviceKeys $ \nh -> do
+        Resp "1" _ (SOK (Just serviceId)) <- serviceSignSendRecv nh nKey servicePK ("1", nId, NSUB)
+        deliverMessage rh rId rKey sh sId sKey nh "hello" dec
+        pure (rId, rKey, sId, dec, serviceId)
+    threadDelay 250000
+    withSmpServerStoreLogOn ps testPort $ runTest2 t $ \sh rh ->
+      testNtfServiceClient t serviceKeys $ \nh -> do
+        Resp "2.1" serviceId' (SOKS n) <- signSendRecv nh (C.APrivateAuthKey C.SEd25519 servicePK) ("2.1", serviceId, NSUBS)
+        n `shouldBe` 1
+        Resp "2.2" _ (SOK Nothing) <- signSendRecv rh rKey ("2.2", rId, SUB)
+        serviceId' `shouldBe` serviceId
+        deliverMessage rh rId rKey sh sId sKey nh "hello 2" dec
+    threadDelay 250000
+    withSmpServerStoreLogOn ps testPort $ runTest2 t $ \sh rh ->
+      testNtfServiceClient t serviceKeys $ \nh -> do
+        Resp "3.1" _ (SOKS n) <- signSendRecv nh (C.APrivateAuthKey C.SEd25519 servicePK) ("3.1", serviceId, NSUBS)
+        n `shouldBe` 1
+        Resp "3.2" _ (SOK Nothing) <- signSendRecv rh rKey ("3.2", rId, SUB)
+        deliverMessage rh rId rKey sh sId sKey nh "hello 3" dec
+    where
+      runTest2 :: Transport c => TProxy c 'TServer -> (THandleSMP c 'TClient -> THandleSMP c 'TClient -> IO a) -> ThreadId -> IO a
+      runTest2 _ test' server = do
+        a <- testSMPClient $ \h1 -> testSMPClient $ \h2 -> test' h1 h2
+        killThread server
+        pure a
+      deliverMessage rh rId rKey sh sId sKey nh msgText dec = do
+        Resp "msg-1" _ OK <- signSendRecv sh sKey ("msg-1", sId, _SEND' msgText)
+        Resp "" _ (Msg mId msg) <- tGet1 rh
+        Resp "msg-2" _ OK <- signSendRecv rh rKey ("msg-2", rId, ACK mId)
+        (dec mId msg, Right msgText) #== "delivered from queue"
+        Resp "" _ (NMSG _ _) <- tGet1 nh
+        pure ()
+
 testMsgExpireOnSend :: SpecWith (ASrvTransport, AStoreType)
 testMsgExpireOnSend =
   it "should expire messages that are not received before messageTTL on SEND" $ \(ATransport (t :: TProxy c 'TServer), msType) -> do

tests/Test.hs

@@ -102,7 +102,7 @@ main = do
               ] -- skipComparisonForDownMigrations
               testStoreDBOpts
               "src/Simplex/Messaging/Server/QueueStore/Postgres/server_schema.sql"
-        aroundAll_ (postgressBracket testServerDBConnectInfo) $
+        around_ (postgressBracket testServerDBConnectInfo) $
           describe "SMP server via TLS, postgres+jornal message store" $
             before (pure (transport @TLS, ASType SQSPostgres SMSJournal)) serverTests
 #endif
@@ -122,13 +122,13 @@ main = do
               [] -- skipComparisonForDownMigrations
               ntfTestStoreDBOpts
               "src/Simplex/Messaging/Notifications/Server/Store/ntf_server_schema.sql"
-        aroundAll_ (postgressBracket ntfTestServerDBConnectInfo) $ do
+        around_ (postgressBracket ntfTestServerDBConnectInfo) $ do
           describe "Notifications server (SMP server: jornal store)" $
             ntfServerTests (transport @TLS, ASType SQSMemory SMSJournal)
-          aroundAll_ (postgressBracket testServerDBConnectInfo) $
+          around_ (postgressBracket testServerDBConnectInfo) $
             describe "Notifications server (SMP server: postgres+jornal store)" $
               ntfServerTests (transport @TLS, ASType SQSPostgres SMSJournal)
-        aroundAll_ (postgressBracket testServerDBConnectInfo) $ do
+        around_ (postgressBracket testServerDBConnectInfo) $ do
           describe "SMP client agent, postgres+jornal message store" $ agentTests (transport @TLS, ASType SQSPostgres SMSJournal)
           describe "SMP proxy, postgres+jornal message store" $
             before (pure $ ASType SQSPostgres SMSJournal) smpProxyTests