diff --git a/src/Simplex/Messaging/Server/MsgStore/Journal.hs b/src/Simplex/Messaging/Server/MsgStore/Journal.hs index 8e9005091..40ff03af0 100644 --- a/src/Simplex/Messaging/Server/MsgStore/Journal.hs +++ b/src/Simplex/Messaging/Server/MsgStore/Journal.hs @@ -303,7 +303,7 @@ instance StoreQueueClass (JournalQueue s) where {-# INLINE withQueueLock #-} removeQueueLock :: JournalQueue s -> IO () removeQueueLock JournalQueue {recipientId', queueLock, queueLocks'} = - atomically $ TM.lookup recipientId' queueLocks' >>= \l -> when (l == Just queueLock) $ TM.delete recipientId' queueLocks' + atomically $ whenM ((Just queueLock ==) <$> TM.lookup recipientId' queueLocks') $ TM.delete recipientId' queueLocks' instance QueueStoreClass (JournalQueue s) (QStore s) where type QueueStoreCfg (QStore s) = QStoreCfg s diff --git a/src/Simplex/Messaging/Server/QueueStore/Postgres.hs b/src/Simplex/Messaging/Server/QueueStore/Postgres.hs index a0bea33f0..7e419a916 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Postgres.hs +++ b/src/Simplex/Messaging/Server/QueueStore/Postgres.hs @@ -176,7 +176,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where withQueueLock sq "addQueue_" $ E.uninterruptibleMask_ $ runExceptT $ do void $ withDB "addQueue_" st $ \db -> E.try (DB.execute db insertQueueQuery $ queueRecToRow (rId, qr)) - >>= bimapM handleDuplicate pure + >>= bimapM (\e -> unless (isRecipientIdViolation e) (removeQueueLock sq) >> handleDuplicate e) pure when useCache $ do atomically $ TM.insert rId sq queues atomically $ TM.insert (senderId qr) rId senders @@ -185,6 +185,8 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where pure sq where PostgresQueueStore {queues, senders, notifiers, useCache} = st + -- the lock is kept when another queue has the same recipient ID + isRecipientIdViolation e = constraintViolation e == Just (UniqueViolation "msg_queues_pkey") -- Not doing duplicate checks in maps as the probability of duplicates is very low. -- It needs to be reconsidered when IDs are supplied by the users. -- hasId = anyM [TM.memberIO rId queues, TM.memberIO senderId senders, hasNotifier] @@ -209,20 +211,19 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where loadRcvQueue = do (rId, qRec) <- loadQueue " WHERE recipient_id = ?" cacheQueue rId qRec $ \_ -> pure () -- recipient map already checked, not caching sender ref - loadSndQueue = do - (rId, qRec) <- loadQueue " WHERE sender_id = ?" - -- checking recipient map first, sender ref is only cached for a queue in the map - atomically (TM.lookup rId queues >>= mapM (\sq -> sq <$ cacheSender rId)) - >>= maybe (cacheQueue rId qRec cacheSender) pure + loadSndQueue = loadSndQueue_ " WHERE sender_id = ?" cacheSender -- link IDs are supplied by clients, they are not cached to prevent collisions with sender IDs - loadLinkQueue = do - (rId, qRec) <- loadQueue " WHERE link_id = ?" - liftIO (TM.lookupIO rId queues) >>= maybe (cacheQueue rId qRec $ \_ -> pure ()) pure + loadLinkQueue = loadSndQueue_ " WHERE link_id = ?" $ \_ -> pure () loadNtfQueue = do (rId, qRec) <- loadQueue " WHERE notifier_id = ?" liftIO $ TM.lookupIO rId queues -- checking recipient map first, not creating lock in map, not caching queue >>= maybe (mkQ False rId qRec) pure + loadSndQueue_ condition insertRef = do + (rId, qRec) <- loadQueue condition + -- checking recipient map first, ref is only cached for the queue in the map + atomically (TM.lookup rId queues >>= mapM (\sq -> sq <$ insertRef rId)) + >>= maybe (cacheQueue rId qRec insertRef) pure loadQueueNoCache cond = mask $ loadQueue cond >>= liftIO . uncurry (mkQ True) mask = E.uninterruptibleMask_ . runExceptT cacheSender rId = TM.insert qId rId senders diff --git a/src/Simplex/Messaging/Server/QueueStore/STM.hs b/src/Simplex/Messaging/Server/QueueStore/STM.hs index 4c32505e6..418c34647 100644 --- a/src/Simplex/Messaging/Server/QueueStore/STM.hs +++ b/src/Simplex/Messaging/Server/QueueStore/STM.hs @@ -47,7 +47,7 @@ import Simplex.Messaging.SystemTime import Simplex.Messaging.TMap (TMap) import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Transport (SMPServiceRole (..)) -import Simplex.Messaging.Util (anyM, ifM, tshow, ($>>), ($>>=), (<$$), (<$$>)) +import Simplex.Messaging.Util (anyM, ifM, tshow, unlessM, ($>>), ($>>=), (<$$), (<$$>)) import System.IO import UnliftIO.STM @@ -119,7 +119,10 @@ instance StoreQueueClass q => QueueStoreClass q (STMQueueStore q) where addQueue_ :: STMQueueStore q -> (RecipientId -> QueueRec -> IO q) -> RecipientId -> QueueRec -> IO (Either ErrorType q) 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 + add sq >>= \case + Right () -> withLog "addStoreQueue" st (\s -> logCreateQueue s rId qr) $> Right sq + -- the lock is kept when another queue has the same recipient ID + Left e -> Left e <$ unlessM (TM.memberIO rId queues) (removeQueueLock sq) where STMQueueStore {queues, senders, notifiers, links} = st add q = atomically $ ifM hasId (pure $ Left DUPLICATE_) $ Right () <$ do diff --git a/src/Simplex/Messaging/Server/QueueStore/Types.hs b/src/Simplex/Messaging/Server/QueueStore/Types.hs index d105357bf..84cd7b4a7 100644 --- a/src/Simplex/Messaging/Server/QueueStore/Types.hs +++ b/src/Simplex/Messaging/Server/QueueStore/Types.hs @@ -27,7 +27,7 @@ class StoreQueueClass q where recipientId :: q -> RecipientId queueRec :: q -> TVar (Maybe QueueRec) withQueueLock :: q -> Text -> IO a -> IO a - -- must only be called for deleted queues + -- must only be called for deleted or not added queues removeQueueLock :: q -> IO () class StoreQueueClass q => QueueStoreClass q s where diff --git a/tests/CoreTests/MsgStoreTests.hs b/tests/CoreTests/MsgStoreTests.hs index d56084cbe..2f34a8509 100644 --- a/tests/CoreTests/MsgStoreTests.hs +++ b/tests/CoreTests/MsgStoreTests.hs @@ -75,6 +75,7 @@ msgStoreTests = do journalMsgStoreTests it "should export and import journal store" testExportImportStore it "should remove deleted queues from queue store maps" $ testDeleteQueueMaps stmQueueMapSizes (Just stmLinksSize) + it "should not leave queue lock when queue is not added" testAddDuplicateQueueLock #if defined(dbServerPostgres) around_ (postgressBracket testServerDBConnectInfo) $ do around (withMsgStore $ testJournalStoreCfg $ PQStoreCfg testPostgresStoreCfg) $ @@ -83,6 +84,7 @@ msgStoreTests = do journalMsgStoreTests it "should remove deleted queues from queue cache maps" $ testDeleteQueueMaps postgresQueueMapSizes Nothing it "should not cache queue deleted while loading" testDeletedQueueNotCached + it "should not leave queue lock when queue is not added" testAddDuplicateQueueLock it "should not keep link data in queue records" testQueueRecNoLinkData around (withMsgStore testPostgresStoreConfig) $ describe "Postgres-only message store" $ do @@ -366,13 +368,11 @@ testDeleteQueueMaps :: forall s. MsgStoreClass s => (s -> IO QueueMapSizes) -> M testDeleteQueueMaps mapSizes linksSize_ ms = do g <- C.newRandom ntfCreds <- testNtfCreds g - let qd = (EncDataBytes "fixed data", EncDataBytes "user data") - newLinkId = atomically $ EntityId <$> C.randomBytes 24 g - lnkId1 <- newLinkId - lnkId2 <- newLinkId - lnkId3 <- newLinkId + lnkId1 <- testLinkId g + lnkId2 <- testLinkId g + lnkId3 <- testLinkId g (rId1, qr1) <- testNewQueueRec g QMMessaging - (rId2, qr2) <- testNewQueueRecData g QMContact (Just (lnkId1, qd)) + (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] @@ -384,8 +384,8 @@ testDeleteQueueMaps mapSizes linksSize_ ms = do q2 <- ExceptT $ addQueue ms rId2 qr2 q3 <- ExceptT $ addQueue ms rId3 qr3 q4 <- ExceptT $ addQueue ms rId4 qr4 - ExceptT $ addQueueLinkData (queueStore ms) q3 lnkId2 qd - ExceptT $ addQueueLinkData (queueStore ms) q4 lnkId3 qd + 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 $ sizesShouldBe (4, 4, 1) 3 @@ -403,6 +403,26 @@ testDeleteQueueMaps mapSizes linksSize_ ms = do mapSizes ms `shouldReturn` sizes forM_ linksSize_ $ \f -> f ms `shouldReturn` linksSize +testAddDuplicateQueueLock :: JournalMsgStore s -> IO () +testAddDuplicateQueueLock ms = do + g <- C.newRandom + (rId, qr) <- testNewQueueRec g QMMessaging + (rId', qr') <- testNewQueueRec g QMMessaging + void $ runRight $ ExceptT $ addQueue ms rId qr + -- duplicate sender ID + addQueue ms rId' qr >>= expectError + -- duplicate recipient ID, the lock of the existing queue is kept + addQueue ms rId qr' >>= expectError + queueLockCount <$> loadedQueueCounts ms `shouldReturn` 1 + where + expectError = either (\_ -> pure ()) (\_ -> expectationFailure "duplicate queue is added") + +testLinkId :: TVar ChaChaDRG -> IO LinkId +testLinkId g = atomically $ EntityId <$> C.randomBytes 24 g + +testLinkData :: QueueLinkData +testLinkData = (EncDataBytes "fixed data", EncDataBytes "user data") + testLinkIdSenderIdCollision :: MsgStoreClass s => s -> IO () testLinkIdSenderIdCollision ms = do g <- C.newRandom @@ -412,7 +432,7 @@ testLinkIdSenderIdCollision ms = do runRight_ $ do void $ ExceptT $ addQueue ms rIdV qrV qA <- ExceptT $ addQueue ms rIdA qrA - ExceptT $ addQueueLinkData (queueStore ms) qA sIdV (EncDataBytes "fixed data", EncDataBytes "user data") + ExceptT $ addQueueLinkData (queueStore ms) qA sIdV testLinkData qA' <- ExceptT $ getQueue ms SSenderLink sIdV liftIO $ recipientId qA' `shouldBe` rIdA qV <- ExceptT $ getQueue ms SSender sIdV @@ -421,22 +441,20 @@ testLinkIdSenderIdCollision ms = do testQueueRecNoLinkData :: MsgStoreClass s => s -> IO () testQueueRecNoLinkData ms = do g <- C.newRandom - let qd = (EncDataBytes "fixed data", EncDataBytes "user data") - qd' = (EncDataBytes "fixed data", EncDataBytes "updated user data") + let qd' = (EncDataBytes "fixed data", EncDataBytes "updated user data") noData = (EncDataBytes "", EncDataBytes "") - newLinkId = atomically $ EntityId <$> C.randomBytes 24 g - lnkId1 <- newLinkId - lnkId2 <- newLinkId - (rId1, qr1) <- testNewQueueRecData g QMContact (Just (lnkId1, qd)) + lnkId1 <- testLinkId g + lnkId2 <- testLinkId g + (rId1, qr1) <- testNewQueueRecData g QMContact (Just (lnkId1, testLinkData)) (rId2, qr2) <- testNewQueueRec g QMContact runRight_ $ do q1 <- ExceptT $ addQueue ms rId1 qr1 q2 <- ExceptT $ addQueue ms rId2 qr2 - ExceptT $ addQueueLinkData (queueStore ms) q2 lnkId2 qd + ExceptT $ addQueueLinkData (queueStore ms) q2 lnkId2 testLinkData liftIO $ queueLinkData q1 `shouldReturn` Just (lnkId1, noData) liftIO $ queueLinkData q2 `shouldReturn` Just (lnkId2, noData) - ExceptT (getQueueLinkData (queueStore ms) q1 lnkId1) >>= liftIO . (`shouldBe` qd) - ExceptT (getQueueLinkData (queueStore ms) q2 lnkId2) >>= liftIO . (`shouldBe` qd) + ExceptT (getQueueLinkData (queueStore ms) q1 lnkId1) >>= liftIO . (`shouldBe` testLinkData) + ExceptT (getQueueLinkData (queueStore ms) q2 lnkId2) >>= liftIO . (`shouldBe` testLinkData) ExceptT $ addQueueLinkData (queueStore ms) q2 lnkId2 qd' liftIO $ queueLinkData q2 `shouldReturn` Just (lnkId2, noData) ExceptT (getQueueLinkData (queueStore ms) q2 lnkId2) >>= liftIO . (`shouldBe` qd') @@ -464,7 +482,7 @@ testDeletedQueueNotCached ms = do evictQueue rId sId = TM.delete rId queues >> TM.delete sId senders loadWhileDeleting g load evict = replicateM_ 100 $ do (rId, qr) <- testNewQueueRec g QMMessaging - q <- either (fail . show) pure =<< addQueue ms rId qr + q <- runRight $ ExceptT $ addQueue ms rId qr atomically $ evict rId (senderId qr) (q_, deleted) <- concurrently (load rId (senderId qr)) (deleteQueue ms q) deleted `shouldSatisfy` isRight