From 77bf76e1f95f0ff9083e843a70c15f6e4d9357c1 Mon Sep 17 00:00:00 2001 From: Evgeny Date: Mon, 1 Sep 2025 17:45:31 +0100 Subject: [PATCH] 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 --- .../Messaging/Server/QueueStore/STM.hs | 7 ++- tests/ServerTests.hs | 47 ++++++++++++++++++- tests/Test.hs | 8 ++-- 3 files changed, 55 insertions(+), 7 deletions(-) diff --git a/src/Simplex/Messaging/Server/QueueStore/STM.hs b/src/Simplex/Messaging/Server/QueueStore/STM.hs index 4dd6240a8..0cbd33b0c 100644 --- a/src/Simplex/Messaging/Server/QueueStore/STM.hs +++ b/src/Simplex/Messaging/Server/QueueStore/STM.hs @@ -114,7 +114,7 @@ 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 @@ -122,8 +122,11 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where 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 diff --git a/tests/ServerTests.hs b/tests/ServerTests.hs index 560ac63d7..649d259d7 100644 --- a/tests/ServerTests.hs +++ b/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 diff --git a/tests/Test.hs b/tests/Test.hs index 4598bb8e4..b88eb3bf5 100644 --- a/tests/Test.hs +++ b/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