smp-server: remove deleted queues from store maps (#1947)

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

@@ -277,9 +277,11 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where
     where
       rId = recipientId sq
       qr = queueRec sq
-      delete q@QueueRec {senderId, rcvServiceId} = do
+      delete q@QueueRec {senderId, queueData, rcvServiceId} = do
         writeTVar qr Nothing
+        TM.delete rId $ queues st
         TM.delete senderId $ senders st
+        forM_ queueData $ \(lnkId, _) -> TM.delete lnkId $ links st
         mapM_ (removeServiceQueue st serviceRcvQueues rId) rcvServiceId
         mapM_ (removeNotifier st) $ notifier q
         pure q

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

@@ -14,9 +14,6 @@ module Simplex.Messaging.Server.StoreLog.ReadWrite
 
 import Control.Concurrent.STM
 import Control.Logger.Simple
-import Control.Monad
-import Control.Monad.IO.Class
-import Control.Monad.Trans.Except
 import qualified Data.ByteString.Char8 as B
 import qualified Data.Text as T
 import Data.Text.Encoding (decodeLatin1)
@@ -26,7 +23,7 @@ import Simplex.Messaging.Server.QueueStore (QueueRec, ServiceRec (..))
 import Simplex.Messaging.Server.QueueStore.STM (STMQueueStore (..), STMService (..), withLoadedQueues)
 import Simplex.Messaging.Server.QueueStore.Types
 import Simplex.Messaging.Server.StoreLog
-import Simplex.Messaging.Util (tshow)
+import Simplex.Messaging.Util (tshow, ($>>=))
 import System.IO
 
 writeQueueStore :: forall q. StoreQueueClass q => StoreLog 'WriteMode -> STMQueueStore q -> IO ()
@@ -68,13 +65,7 @@ readQueueStore tty mkQ f st = readLogLines tty f $ \_ -> processLine
         printError :: String -> IO ()
         printError e = B.putStrLn $ "Error parsing log: " <> B.pack e <> " - " <> s
         withQueue :: forall a. RecipientId -> T.Text -> (q -> IO (Either ErrorType a)) -> IO ()
-        withQueue qId op a = runExceptT go >>= qError qId op
-          where
-            go = do
-              q <- ExceptT $ getQueue_ st mkQ SRecipient qId
-              liftIO (readTVarIO $ queueRec q) >>= \case
-                Nothing -> logWarn $ logPfx qId op <> "already deleted"
-                Just _ -> void $ ExceptT $ a q
+        withQueue qId op a = (getQueue_ st mkQ SRecipient qId $>>= a) >>= qError qId op
         qError qId op = \case
           Left e -> logError $ logPfx qId op <> tshow e
           Right _ -> pure ()

tests/CoreTests/MsgStoreTests.hs

@@ -23,6 +23,7 @@ import Control.Monad.Trans.Except
 import Crypto.Random (ChaChaDRG)
 import Data.ByteString.Char8 (ByteString)
 import Data.Int (Int64)
+import qualified Data.Map.Strict as M
 import Data.Time.Clock.System (getSystemTime)
 import Simplex.Messaging.Crypto (pattern MaxLenBS)
 import qualified Simplex.Messaging.Crypto as C
@@ -31,7 +32,9 @@ import Simplex.Messaging.Server.MsgStore.STM
 import Simplex.Messaging.Server.MsgStore.Types
 import Simplex.Messaging.Server.QueueStore
 import Simplex.Messaging.Server.QueueStore.QueueInfo
+import Simplex.Messaging.Server.QueueStore.STM (STMQueueStore (..))
 import Simplex.Messaging.Server.QueueStore.Types
+import Simplex.Messaging.TMap (TMap)
 import Test.Hspec hiding (fit, it)
 import Util
 
@@ -50,7 +53,9 @@ import SMPClient (testStoreDBOpts)
 
 msgStoreTests :: Spec
 msgStoreTests = do
-  describe "STM message store" $ someMsgStoreTests testSMTStoreConfig
+  describe "STM message store" $ do
+    someMsgStoreTests testSMTStoreConfig
+    it "should remove deleted queues from queue store maps" $ withMsgStore testSMTStoreConfig testDeleteQueueMaps
 #if defined(dbServerPostgres)
   describe "Postgres-only message store" $ do
     someMsgStoreTests testPostgresStoreConfig
@@ -126,6 +131,18 @@ testNewQueueRecData g qm queueData = do
   where
     rndId = atomically $ EntityId <$> C.randomBytes 24 g
 
+testNtfCreds :: TVar ChaChaDRG -> IO NtfCreds
+testNtfCreds g = do
+  (notifierKey, _) <- atomically $ C.generateAuthKeyPair C.SX25519 g
+  (k, pk) <- atomically $ C.generateKeyPair @'C.X25519 g
+  pure
+    NtfCreds
+      { notifierId = EntityId "ijkl",
+        notifierKey,
+        rcvNtfDhSecret = C.dh' k pk,
+        ntfServiceId = Nothing
+      }
+
 testGetQueue :: MsgStoreClass s => s -> IO ()
 testGetQueue ms = do
   g <- C.newRandom
@@ -217,6 +234,56 @@ testLinkDataSecuredQueue ms = do
     void $ ExceptT $ deleteQueue ms q
     void $ ExceptT $ deleteQueue ms cq
 
+-- sizes of queues, senders, links and notifiers maps
+type QueueMapSizes = (Int, Int, Int, Int)
+
+queueMapSizes :: STMMsgStore -> IO QueueMapSizes
+queueMapSizes ms = (,,,) <$> size queues <*> size senders <*> size links <*> size notifiers
+  where
+    STMQueueStore {queues, senders, links, notifiers} = queueStore ms
+    size :: TMap k v -> IO Int
+    size = fmap M.size . readTVarIO
+
+testDeleteQueueMaps :: STMMsgStore -> IO ()
+testDeleteQueueMaps ms = do
+  g <- C.newRandom
+  ntfCreds <- testNtfCreds g
+  let newLinkId = atomically $ EntityId <$> C.randomBytes 24 g
+  lnkId1 <- newLinkId
+  lnkId2 <- newLinkId
+  lnkId3 <- newLinkId
+  (rId1, qr1) <- testNewQueueRec g QMMessaging
+  (rId2, qr2) <- testNewQueueRecData g QMContact (Just (lnkId1, testLinkData))
+  (rId3, qr3) <- testNewQueueRec g QMMessaging
+  (rId4, qr4) <- testNewQueueRec g QMMessaging
+  let rIds = [rId1, rId2, rId3, rId4] :: [RecipientId]
+      sIds = map senderId [qr1, qr2, qr3, qr4]
+      lnkIds = [lnkId1, lnkId2, lnkId3] :: [LinkId]
+  queueMapSizes ms `shouldReturn` (0, 0, 0, 0)
+  runRight_ $ do
+    q1 <- ExceptT $ addQueue ms rId1 qr1 {notifier = Just ntfCreds}
+    q2 <- ExceptT $ addQueue ms rId2 qr2
+    q3 <- ExceptT $ addQueue ms rId3 qr3
+    q4 <- ExceptT $ addQueue ms rId4 qr4
+    ExceptT $ addQueueLinkData (queueStore ms) q3 lnkId2 testLinkData
+    ExceptT $ addQueueLinkData (queueStore ms) q4 lnkId3 testLinkData
+    forM_ sIds $ void . ExceptT . getQueue ms SSender
+    forM_ lnkIds $ void . ExceptT . getQueue ms SSenderLink
+    liftIO $ queueMapSizes ms `shouldReturn` (4, 4, 3, 1)
+    ExceptT $ deleteQueueLinkData (queueStore ms) q3
+    liftIO $ queueMapSizes ms `shouldReturn` (4, 4, 2, 1)
+    forM_ ([q1, q2, q3, q4] :: [STMQueue]) $ void . ExceptT . deleteQueue ms
+  queueMapSizes ms `shouldReturn` (0, 0, 0, 0)
+  forM_ rIds $ \rId -> getQueue ms SRecipient rId >>= expectAuth
+  forM_ sIds $ \sId -> getQueue ms SSender sId >>= expectAuth
+  forM_ lnkIds $ \lnkId -> getQueue ms SSenderLink lnkId >>= expectAuth
+  queueMapSizes ms `shouldReturn` (0, 0, 0, 0)
+  where
+    expectAuth = either (`shouldBe` AUTH) (\_ -> expectationFailure "deleted queue is still found")
+
+testLinkData :: QueueLinkData
+testLinkData = (EncDataBytes "fixed data", EncDataBytes "user data")
+
 #if defined(dbServerPostgres)
 testUpdateMessageCounts :: PostgresMsgStore -> IO ()
 testUpdateMessageCounts ms = do

tests/CoreTests/StoreLogTests.hs

@@ -48,18 +48,6 @@ import Simplex.Messaging.Server.Main
 testPublicAuthKey :: C.APublicAuthKey
 testPublicAuthKey = C.APublicAuthKey C.SEd25519 (C.publicKey "MC4CAQAwBQYDK2VwBCIEIDfEfevydXXfKajz3sRkcQ7RPvfWUPoq6pu1TYHV1DEe")
 
-testNtfCreds :: TVar ChaChaDRG -> IO NtfCreds
-testNtfCreds g = do
-  (notifierKey, _) <- atomically $ C.generateAuthKeyPair C.SX25519 g
-  (k, pk) <- atomically $ C.generateKeyPair @'C.X25519 g
-  pure
-    NtfCreds
-      { notifierId = EntityId "ijkl",
-        notifierKey,
-        rcvNtfDhSecret = C.dh' k pk,
-        ntfServiceId = Nothing
-      }
-
 data StoreLogTestCase r s = SLTC {name :: String, saved :: [r], state :: s, compacted :: [r]}
 
 type SMPStoreLogTestCase = StoreLogTestCase StoreLogRecord (M.Map RecipientId QueueRec)