dandri/simplexmq
1
0
Fork 0
mirror of https://github.com/simplex-chat/simplexmq.git synced 2026-08-28 11:34:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

smp server: wrap all queries in transactions (#1603)

Browse Source 
* smp server: wrap all queries in transactions

* fix test

* fix schema test
This commit is contained in:
Evgeny
2025-08-17 10:38:20 +01:00
committed by GitHub
parent 2cedb66667
commit 96e8b4a146
4 changed files with 22 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Simplex/Messaging/Server/QueueStore/Postgres.hs
+7 -7
View File
@@ -142,7 +142,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
getEntityCounts :: PostgresQueueStore q -> IO EntityCounts
getEntityCounts st =
withConnection (dbStore st) $ \db -> do
withTransaction (dbStore st) $ \db -> do
(queueCount, notifierCount, rcvServiceCount, ntfServiceCount, rcvServiceQueuesCount, ntfServiceQueuesCount) : _ <-
DB.query
db
@@ -496,7 +496,7 @@ instance StoreQueueClass q => QueueStoreClass q (PostgresQueueStore q) where
batchInsertServices :: [STMService] -> PostgresQueueStore q -> IO Int64
batchInsertServices services' toStore =
withConnection (dbStore toStore) $ \db ->
withTransaction (dbStore toStore) $ \db ->
DB.executeMany db insertServiceQuery $ map (serviceRecToRow . serviceRec) services'
batchInsertQueues :: StoreQueueClass q => Bool -> M.Map RecipientId q -> PostgresQueueStore q' -> IO Int64
@@ -505,7 +505,7 @@ batchInsertQueues tty queues toStore = do
putStrLn $ "Importing " <> show (length qs) <> " queues..."
let st = dbStore toStore
count <-
withConnection st $ \db -> do
withTransaction st $ \db -> do
DB.copy_
db
[sql|
@@ -514,7 +514,7 @@ batchInsertQueues tty queues toStore = do
|]
mapM_ (putQueue db) (zip [1..] qs)
DB.putCopyEnd db
Only qCnt : _ <- withConnection st (`DB.query_` "SELECT count(*) FROM msg_queues")
Only qCnt : _ <- withTransaction st (`DB.query_` "SELECT count(*) FROM msg_queues")
putStrLn $ progress count
pure qCnt
where
@@ -541,13 +541,13 @@ insertServiceQuery =
foldServiceRecs :: forall a q. Monoid a => PostgresQueueStore q -> (ServiceRec -> IO a) -> IO a
foldServiceRecs st f =
withConnection (dbStore st) $ \db ->
withTransaction (dbStore st) $ \db ->
DB.fold_ db "SELECT service_id, service_role, service_cert, service_cert_hash, created_at FROM services" mempty $
\ !acc -> fmap (acc <>) . f . rowToServiceRec
foldQueueRecs :: forall a q. Monoid a => Bool -> Bool -> PostgresQueueStore q -> Maybe Int64 -> ((RecipientId, QueueRec) -> IO a) -> IO a
foldQueueRecs tty withData st skipOld_ f = do
(n, r) <- withConnection (dbStore st) $ \db ->
(n, r) <- withTransaction (dbStore st) $ \db ->
foldRecs db (0 :: Int, mempty) $ \(i, acc) qr -> do
r <- f qr
let !i' = i + 1
@@ -686,7 +686,7 @@ withDB' op st action = withDB op st $ fmap Right . action
withDB :: forall a q. Text -> PostgresQueueStore q -> (DB.Connection -> IO (Either ErrorType a)) -> ExceptT ErrorType IO a
withDB op st action =
ExceptT $ E.try (withConnection (dbStore st) action) >>= either logErr pure
ExceptT $ E.try (withTransaction (dbStore st) action) >>= either logErr pure
where
logErr :: E.SomeException -> IO (Either ErrorType a)
logErr e = logError ("STORE: " <> err) $> Left (STORE err)