expect DOWN

fix some tests
mark waitForWork
2026-08-30 14:08:22 +00:00 · 2024-04-09 22:43:47 +03:00 · 2024-04-09 21:44:23 +03:00 · 2024-04-09 20:26:32 +03:00 · 2024-04-09 20:16:50 +03:00
41 changed files with 501 additions and 487 deletions
@@ -114,45 +114,30 @@ executables:
    main: Main.hs
    dependencies:
      - simplexmq
-    ghc-options:
-      - -threaded
-      - -rtsopts

  ntf-server:
    source-dirs: apps/ntf-server
    main: Main.hs
    dependencies:
      - simplexmq
-    ghc-options:
-      - -threaded
-      - -rtsopts

  xftp-server:
    source-dirs: apps/xftp-server
    main: Main.hs
    dependencies:
      - simplexmq
-    ghc-options:
-      - -threaded
-      - -rtsopts

  smp-agent:
    source-dirs: apps/smp-agent
    main: Main.hs
    dependencies:
      - simplexmq
-    ghc-options:
-      - -threaded
-      - -rtsopts

  xftp:
    source-dirs: apps/xftp
    main: Main.hs
    dependencies:
      - simplexmq
-    ghc-options:
-      - -threaded
-      - -rtsopts

 tests:
  simplexmq-test:
@@ -180,6 +165,8 @@ ghc-options:
  - -Wincomplete-uni-patterns
  - -Wunused-type-patterns
  - -O2
+  - -threaded
+  - -rtsopts

 default-extensions:
  - StrictData
@@ -173,7 +173,7 @@ library
      src
  default-extensions:
      StrictData
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
  include-dirs:
      cbits
  c-sources:
@@ -661,7 +661,7 @@ test-suite simplexmq-test
      tests
  default-extensions:
      StrictData
-  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2
+  ghc-options: -Wall -Wcompat -Werror=incomplete-patterns -Wredundant-constraints -Wincomplete-record-updates -Wincomplete-uni-patterns -Wunused-type-patterns -O2 -threaded -rtsopts
  build-depends:
      HUnit ==1.6.*
    , QuickCheck ==2.14.*
@@ -70,7 +70,7 @@ import qualified Simplex.Messaging.Crypto.Lazy as LC
 import Simplex.Messaging.Encoding
 import Simplex.Messaging.Encoding.String (strDecode, strEncode)
 import Simplex.Messaging.Protocol (EntityId, XFTPServer)
-import Simplex.Messaging.Util (catchAll_, liftError, tshow, unlessM, whenM)
+import Simplex.Messaging.Util (catchAll_, liftError, tshow, unlessM, whenM, atomically')
 import System.FilePath (takeFileName, (</>))
 import UnliftIO
 import UnliftIO.Directory
@@ -110,7 +110,7 @@ closeXFTPAgent a = do
  stopWorkers $ xftpSndWorkers a
  stopWorkers $ xftpDelWorkers a
  where
-    stopWorkers workers = atomically (swapTVar workers M.empty) >>= mapM_ (liftIO . cancelWorker)
+    stopWorkers workers = atomically' (swapTVar workers M.empty) >>= mapM_ (liftIO . cancelWorker)

 xftpReceiveFile' :: AgentClient -> UserId -> ValidFileDescription 'FRecipient -> Maybe CryptoFileArgs -> AM RcvFileId
 xftpReceiveFile' c userId (ValidFileDescription fd@FileDescription {chunks, redirect}) cfArgs = do
@@ -131,7 +131,7 @@ xftpReceiveFile' c userId (ValidFileDescription fd@FileDescription {chunks, redi
          relSavePathRedirect = relPrefixPath </> "xftp.redirect-decrypted"
      lift $ createDirectory =<< toFSFilePath relTmpPathRedirect
      lift $ createEmptyFile =<< toFSFilePath relSavePathRedirect
-      cfArgsRedirect <- atomically $ CF.randomArgs g
+      cfArgsRedirect <- atomically' $ CF.randomArgs g
      let saveFileRedirect = CryptoFile relSavePathRedirect $ Just cfArgsRedirect
      -- create download tasks
      withStore c $ \db -> createRcvFileRedirect db g userId fd relPrefixPath relTmpPathRedirect saveFileRedirect relTmpPath saveFile
@@ -170,7 +170,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do
  cfg <- asks config
  forever $ do
    lift $ waitForWork doWork
-    atomically $ assertAgentForeground c
+    atomically' $ assertAgentForeground c
    runXFTPOperation cfg
  where
    runXFTPOperation :: AgentConfig -> AM ()
@@ -188,7 +188,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do
                when notifyOnRetry $ notify c rcvFileEntityId $ RFERR e
                liftIO $ closeXFTPServerClient c userId server digest
                withStore' c $ \db -> updateRcvChunkReplicaDelay db rcvChunkReplicaId replicaDelay
-              atomically $ assertAgentForeground c
+              atomically' $ assertAgentForeground c
              loop
            retryDone e = rcvWorkerInternalError c rcvFileId rcvFileEntityId (Just fileTmpPath) (show e)
    downloadFileChunk :: RcvFileChunk -> RcvFileChunkReplica -> AM ()
@@ -198,7 +198,7 @@ runXFTPRcvWorker c srv Worker {doWork} = do
      let chunkSpec = XFTPRcvChunkSpec chunkPath (unFileSize chunkSize) (unFileDigest digest)
          relChunkPath = fileTmpPath </> takeFileName chunkPath
      agentXFTPDownloadChunk c userId digest replica chunkSpec
-      atomically $ waitUntilForeground c
+      atomically' $ waitUntilForeground c
      (entityId, complete, progress) <- withStore c $ \db -> runExceptT $ do
        liftIO $ updateRcvFileChunkReceived db (rcvChunkReplicaId replica) rcvChunkId relChunkPath
        RcvFile {size = FileSize currentSize, chunks, redirect} <- ExceptT $ getRcvFile db rcvFileId
@@ -244,7 +244,7 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
  cfg <- asks config
  forever $ do
    lift $ waitForWork doWork
-    atomically $ assertAgentForeground c
+    atomically' $ assertAgentForeground c
    runXFTPOperation cfg
  where
    runXFTPOperation :: AgentConfig -> AM ()
@@ -270,12 +270,12 @@ runXFTPRcvLocalWorker c Worker {doWork} = do
        Nothing -> do
          notify c rcvFileEntityId $ RFDONE fsSavePath
          lift $ forM_ tmpPath (removePath <=< toFSFilePath)
-          atomically $ waitUntilForeground c
+          atomically' $ waitUntilForeground c
          withStore' c (`updateRcvFileComplete` rcvFileId)
        Just RcvFileRedirect {redirectFileInfo, redirectDbId} -> do
          let RedirectFileInfo {size = redirectSize, digest = redirectDigest} = redirectFileInfo
          lift $ forM_ tmpPath (removePath <=< toFSFilePath)
-          atomically $ waitUntilForeground c
+          atomically' $ waitUntilForeground c
          withStore' c (`updateRcvFileComplete` rcvFileId)
          -- proceed with redirect
          yaml <- liftError (INTERNAL . show) (CF.readFile $ CryptoFile fsSavePath cfArgs) `agentFinally` (lift $ toFSFilePath fsSavePath >>= removePath)
@@ -339,7 +339,7 @@ xftpSendDescription' c userId (ValidFileDescription fdDirect@FileDescription {si
  createDirectory prefixPath
  let relPrefixPath = takeFileName prefixPath
  let directYaml = prefixPath </> "direct.yaml"
-  cfArgs <- atomically $ CF.randomArgs g
+  cfArgs <- atomically' $ CF.randomArgs g
  let file = CryptoFile directYaml (Just cfArgs)
  liftError (INTERNAL . show) $ CF.writeFile file (LB.fromStrict $ strEncode fdDirect)
  key <- atomically $ C.randomSbKey g
@@ -362,7 +362,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
  cfg <- asks config
  forever $ do
    lift $ waitForWork doWork
-    atomically $ assertAgentForeground c
+    atomically' $ assertAgentForeground c
    runXFTPOperation cfg
  where
    runXFTPOperation :: AgentConfig -> AM ()
@@ -415,7 +415,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
          pure (FileDigest digest, zip chunkSpecs $ coerce chunkDigests)
        createChunk :: Int -> SndFileChunk -> AM ()
        createChunk numRecipients' ch = do
-          atomically $ assertAgentForeground c
+          atomically' $ assertAgentForeground c
          (replica, ProtoServerWithAuth srv _) <- tryCreate
          withStore' c $ \db -> createSndFileReplica db ch replica
          lift . void $ getXFTPSndWorker True c (Just srv)
@@ -426,7 +426,7 @@ runXFTPSndPrepareWorker c Worker {doWork} = do
                createWithNextSrv usedSrvs
                  `catchAgentError` \e -> retryOnError "XFTP prepare worker" (retryLoop loop) (throwError e) e
              where
-                retryLoop loop = atomically (assertAgentForeground c) >> loop
+                retryLoop loop = atomically' (assertAgentForeground c) >> loop
            createWithNextSrv usedSrvs = do
              deleted <- withStore' c $ \db -> getSndFileDeleted db sndFileId
              when deleted $ throwError $ INTERNAL "file deleted, aborting chunk creation"
@@ -445,7 +445,7 @@ runXFTPSndWorker c srv Worker {doWork} = do
  cfg <- asks config
  forever $ do
    lift $ waitForWork doWork
-    atomically $ assertAgentForeground c
+    atomically' $ assertAgentForeground c
    runXFTPOperation cfg
  where
    runXFTPOperation :: AgentConfig -> AM ()
@@ -463,7 +463,7 @@ runXFTPSndWorker c srv Worker {doWork} = do
                when notifyOnRetry $ notify c sndFileEntityId $ SFERR e
                liftIO $ closeXFTPServerClient c userId server digest
                withStore' c $ \db -> updateSndChunkReplicaDelay db sndChunkReplicaId replicaDelay
-              atomically $ assertAgentForeground c
+              atomically' $ assertAgentForeground c
              loop
            retryDone e = sndWorkerInternalError c sndFileId sndFileEntityId (Just filePrefixPath) (show e)
    uploadFileChunk :: AgentConfig -> SndFileChunk -> SndFileChunkReplica -> AM ()
@@ -472,9 +472,9 @@ runXFTPSndWorker c srv Worker {doWork} = do
      fsFilePath <- lift $ toFSFilePath filePath
      unlessM (doesFileExist fsFilePath) $ throwError $ INTERNAL "encrypted file doesn't exist on upload"
      let chunkSpec' = chunkSpec {filePath = fsFilePath} :: XFTPChunkSpec
-      atomically $ assertAgentForeground c
+      atomically' $ assertAgentForeground c
      agentXFTPUploadChunk c userId chunkDigest replica' chunkSpec'
-      atomically $ waitUntilForeground c
+      atomically' $ waitUntilForeground c
      sf@SndFile {sndFileEntityId, prefixPath, chunks} <- withStore c $ \db -> do
        updateSndChunkReplicaStatus db sndChunkReplicaId SFRSUploaded
        getSndFile db sndFileId
@@ -610,7 +610,7 @@ runXFTPDelWorker c srv Worker {doWork} = do
  cfg <- asks config
  forever $ do
    lift $ waitForWork doWork
-    atomically $ assertAgentForeground c
+    atomically' $ assertAgentForeground c
    runXFTPOperation cfg
  where
    runXFTPOperation :: AgentConfig -> AM ()
@@ -629,7 +629,7 @@ runXFTPDelWorker c srv Worker {doWork} = do
                when notifyOnRetry $ notify c "" $ SFERR e
                liftIO $ closeXFTPServerClient c userId server chunkDigest
                withStore' c $ \db -> updateDeletedSndChunkReplicaDelay db deletedSndChunkReplicaId replicaDelay
-              atomically $ assertAgentForeground c
+              atomically' $ assertAgentForeground c
              loop
            retryDone = delWorkerInternalError c deletedSndChunkReplicaId
            deleteChunkReplica = do
@@ -24,7 +24,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (ProtocolServer (..), XFTPServer)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (catchAll_)
+import Simplex.Messaging.Util (catchAll_, atomically')
 import UnliftIO

 type XFTPClientVar = TMVar (Either XFTPClientAgentError XFTPClient)
@@ -63,7 +63,7 @@ type ME a = ExceptT XFTPClientAgentError IO a

 getXFTPServerClient :: TVar ChaChaDRG -> XFTPClientAgent -> XFTPServer -> ME XFTPClient
 getXFTPServerClient g XFTPClientAgent {xftpClients, config} srv = do
-  atomically getClientVar >>= either newXFTPClient waitForXFTPClient
+  atomically' getClientVar >>= either newXFTPClient waitForXFTPClient
  where
    connectClient :: ME XFTPClient
    connectClient =
@@ -88,7 +88,7 @@ getXFTPServerClient g XFTPClientAgent {xftpClients, config} srv = do
    waitForXFTPClient :: XFTPClientVar -> ME XFTPClient
    waitForXFTPClient clientVar = do
      let XFTPClientConfig {xftpNetworkConfig = NetworkConfig {tcpConnectTimeout}} = xftpConfig config
-      client_ <- liftIO $ tcpConnectTimeout `timeout` atomically (readTMVar clientVar)
+      client_ <- liftIO $ tcpConnectTimeout `timeout` atomically' (readTMVar clientVar)
      liftEither $ case client_ of
        Just (Right c) -> Right c
        Just (Left e) -> Left e
@@ -102,12 +102,12 @@ getXFTPServerClient g XFTPClientAgent {xftpClients, config} srv = do
          tryError connectClient >>= \r -> case r of
            Right client -> do
              logInfo $ "connected to " <> showServer srv
-              atomically $ putTMVar clientVar r
+              atomically' $ putTMVar clientVar r
              pure client
            Left e@(XFTPClientAgentError _ e') -> do
              if temporaryClientError e'
                then retryAction
-                else atomically $ do
+                else atomically' $ do
                  putTMVar clientVar r
                  TM.delete srv xftpClients
              throwError e
@@ -125,6 +125,6 @@ closeXFTPServerClient XFTPClientAgent {xftpClients, config} srv =
  where
    closeClient cVar = do
      let NetworkConfig {tcpConnectTimeout} = xftpNetworkConfig $ xftpConfig config
-      tcpConnectTimeout `timeout` atomically (readTMVar cVar) >>= \case
+      tcpConnectTimeout `timeout` atomically' (readTMVar cVar) >>= \case
        Just (Right client) -> closeXFTPClient client `catchAll_` pure ()
        _ -> pure ()
@@ -107,7 +107,7 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
        Right pk' -> pure pk'
        Left e -> putStrLn ("servers has no valid key: " <> show e) >> exitFailure
      env <- ask
-      sessions <- atomically TM.empty
+      sessions <- atomically' TM.empty
      let cleanup sessionId = atomically $ TM.delete sessionId sessions
      liftIO . runHTTP2Server started xftpPort defaultHTTP2BufferSize serverParams transportConfig inactiveClientExpiration cleanup $ \sessionId sessionALPN r sendResponse -> do
        reqBody <- getHTTP2Body r xftpBlockSize
@@ -191,15 +191,15 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
        withFile statsFilePath AppendMode $ \h -> liftIO $ do
          hSetBuffering h LineBuffering
          ts <- getCurrentTime
-          fromTime' <- atomically $ swapTVar fromTime ts
-          filesCreated' <- atomically $ swapTVar filesCreated 0
-          fileRecipients' <- atomically $ swapTVar fileRecipients 0
-          filesUploaded' <- atomically $ swapTVar filesUploaded 0
-          filesExpired' <- atomically $ swapTVar filesExpired 0
-          filesDeleted' <- atomically $ swapTVar filesDeleted 0
-          files <- atomically $ periodStatCounts filesDownloaded ts
-          fileDownloads' <- atomically $ swapTVar fileDownloads 0
-          fileDownloadAcks' <- atomically $ swapTVar fileDownloadAcks 0
+          fromTime' <- atomically' $ swapTVar fromTime ts
+          filesCreated' <- atomically' $ swapTVar filesCreated 0
+          fileRecipients' <- atomically' $ swapTVar fileRecipients 0
+          filesUploaded' <- atomically' $ swapTVar filesUploaded 0
+          filesExpired' <- atomically' $ swapTVar filesExpired 0
+          filesDeleted' <- atomically' $ swapTVar filesDeleted 0
+          files <- atomically' $ periodStatCounts filesDownloaded ts
+          fileDownloads' <- atomically' $ swapTVar fileDownloads 0
+          fileDownloadAcks' <- atomically' $ swapTVar fileDownloadAcks 0
          filesCount' <- readTVarIO filesCount
          filesSize' <- readTVarIO filesSize
          hPutStrLn h $
@@ -268,8 +268,8 @@ xftpServer cfg@XFTPServerConfig {xftpPort, transportConfig, inactiveClientExpira
              CPDelete fileId -> withUserRole $ unliftIO u $ do
                fs <- asks store
                r <- runExceptT $ do
-                  let asSender = ExceptT . atomically $ getFile fs SFSender fileId
-                  let asRecipient = ExceptT . atomically $ getFile fs SFRecipient fileId
+                  let asSender = ExceptT . atomically' $ getFile fs SFSender fileId
+                  let asRecipient = ExceptT . atomically' $ getFile fs SFRecipient fileId
                  (fr, _) <- asSender `catchError` const asRecipient
                  ExceptT $ deleteServerFile_ fr
                liftIO . hPutStrLn h $ either (\e -> "error: " <> show e) (\() -> "ok") r
@@ -336,7 +336,7 @@ verifyXFTPTransmission auth_ tAuth authorized fId cmd =
    verifyCmd :: SFileParty p -> M VerificationResult
    verifyCmd party = do
      st <- asks store
-      atomically $ verify <$> getFile st party fId
+      atomically' $ verify <$> getFile st party fId
      where
        verify = \case
          Right (fr, k) -> XFTPReqCmd fId fr cmd `verifyWith` k
@@ -397,7 +397,7 @@ processXFTPRequest HTTP2Body {bodyPart} = \case
    retryAdd 0 _ = pure $ Left INTERNAL
    retryAdd n add = do
      fId <- getFileId
-      atomically (add fId) >>= \case
+      atomically' (add fId) >>= \case
        Left DUPLICATE_ -> retryAdd (n - 1) add
        r -> pure r
    addRecipients :: XFTPFileId -> NonEmpty RcvPublicAuthKey -> M FileResponse
@@ -447,7 +447,7 @@ processXFTPRequest HTTP2Body {bodyPart} = \case
                pure FROk
              Left e -> do
                us <- asks $ usedStorage . store
-                atomically . modifyTVar' us $ subtract (fromIntegral size)
+                atomically' . modifyTVar' us $ subtract (fromIntegral size)
                liftIO $ whenM (doesFileExist fPath) (removeFile fPath) `catch` logFileError
                pure $ FRErr e
          receiveChunk spec = do
@@ -465,7 +465,7 @@ processXFTPRequest HTTP2Body {bodyPart} = \case
            Right sbState -> do
              stats <- asks serverStats
              atomically $ modifyTVar' (fileDownloads stats) (+ 1)
-              atomically $ updatePeriodStats (filesDownloaded stats) senderId
+              atomically' $ updatePeriodStats (filesDownloaded stats) senderId
              pure (FRFile sDhKey cbNonce, Just ServerFile {filePath = path, fileSize = size, sbState})
            _ -> pure (FRErr INTERNAL, Nothing)
        _ -> pure (FRErr NO_FILE, Nothing)
@@ -480,7 +480,7 @@ processXFTPRequest HTTP2Body {bodyPart} = \case
    ackFileReception rId fr = do
      withFileLog (`logAckFile` rId)
      st <- asks store
-      atomically $ deleteRecipient st rId fr
+      atomically' $ deleteRecipient st rId fr
      stats <- asks serverStats
      atomically $ modifyTVar' (fileDownloadAcks stats) (+ 1)
      pure FROk
@@ -493,7 +493,7 @@ deleteServerFile_ FileRec {senderId, fileInfo, filePath} = do
    stats <- asks serverStats
    ExceptT $ first (\(_ :: SomeException) -> FILE_IO) <$> try (forM_ path $ \p -> whenM (doesFileExist p) (removeFile p >> deletedStats stats))
    st <- asks store
-    void $ atomically $ deleteFile st senderId
+    void $ atomically' $ deleteFile st senderId
    atomically $ modifyTVar' (filesDeleted stats) (+ 1)
  where
    deletedStats stats = do
@@ -509,7 +509,7 @@ expireServerFiles itemDelay expCfg = do
  logInfo $ "Expiration check: " <> tshow (M.size files') <> " files"
  forM_ (M.keys files') $ \sId -> do
    mapM_ threadDelay itemDelay
-    atomically (expiredFilePath st sId old)
+    atomically' (expiredFilePath st sId old)
      >>= mapM_ (maybeRemove $ delete st sId)
  usedEnd <- readTVarIO $ usedStorage st
  logInfo $ "Used " <> mbs usedStart <> " -> " <> mbs usedEnd <> ", " <> mbs (usedStart - usedEnd) <> " reclaimed."
@@ -523,7 +523,7 @@ expireServerFiles itemDelay expCfg = do
        del
    delete st sId = do
      withFileLog (`logDeleteFile` sId)
-      void . atomically $ deleteFile st sId -- will not update usedStorage if sId isn't in store
+      void . atomically' $ deleteFile st sId -- will not update usedStorage if sId isn't in store
      FileServerStats {filesExpired} <- asks serverStats
      atomically $ modifyTVar' filesExpired (+ 1)

@@ -546,7 +546,7 @@ incFileStat statSel = do
 saveServerStats :: M ()
 saveServerStats =
  asks (serverStatsBackupFile . config)
-    >>= mapM_ (\f -> asks serverStats >>= atomically . getFileServerStatsData >>= liftIO . saveStats f)
+    >>= mapM_ (\f -> asks serverStats >>= atomically' . getFileServerStatsData >>= liftIO . saveStats f)
  where
    saveStats f stats = do
      logInfo $ "saving server stats to file " <> T.pack f
@@ -564,7 +564,7 @@ restoreServerStats = asks (serverStatsBackupFile . config) >>= mapM_ restoreStat
          FileStore {files, usedStorage} <- asks store
          _filesCount <- M.size <$> readTVarIO files
          _filesSize <- readTVarIO usedStorage
-          atomically $ setFileServerStats s d {_filesCount, _filesSize}
+          atomically' $ setFileServerStats s d {_filesCount, _filesSize}
          renameFile f $ f <> ".bak"
          logInfo "server stats restored"
          when (statsFilesCount /= _filesCount) $ logWarn $ "Files count differs: stats: " <> tshow statsFilesCount <> ", store: " <> tshow _filesCount
@@ -34,7 +34,7 @@ import Simplex.FileTransfer.Server.Store
 import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Protocol (RcvPublicAuthKey, RecipientId, SenderId)
 import Simplex.Messaging.Server.StoreLog
-import Simplex.Messaging.Util (bshow, whenM)
+import Simplex.Messaging.Util (bshow, whenM, atomically')
 import System.Directory (doesFileExist, renameFile)
 import System.IO

@@ -94,7 +94,7 @@ readFileStore f st = mapM_ (addFileLogRecord . LB.toStrict) . LB.lines =<< LB.re
    addFileLogRecord s = case strDecode s of
      Left e -> B.putStrLn $ "Log parsing error (" <> B.pack e <> "): " <> B.take 100 s
      Right lr ->
-        atomically (addToStore lr) >>= \case
+        atomically' (addToStore lr) >>= \case
          Left e -> B.putStrLn $ "Log processing error (" <> bshow e <> "): " <> B.take 100 s
          _ -> pure ()
    addToStore = \case
@@ -197,7 +197,7 @@ getSMPAgentClient_ clientId cfg initServers store backgroundMode =
  liftIO $ newSMPAgentEnv cfg store >>= runReaderT runAgent
  where
    runAgent = do
-      c@AgentClient {acThread}  <- atomically . newAgentClient clientId initServers =<< ask
+      c@AgentClient {acThread}  <- atomically' . newAgentClient clientId initServers =<< ask
      t <- runAgentThreads c `forkFinally` const (liftIO $ disconnectAgentClient c)
      atomically . writeTVar acThread . Just =<< mkWeakThreadId t
      pure c
@@ -224,7 +224,7 @@ disconnectAgentClient c@AgentClient {agentEnv = Env {ntfSupervisor = ns, xftpAge
 -- only used in the tests
 disposeAgentClient :: AgentClient -> IO ()
 disposeAgentClient c@AgentClient {acThread, agentEnv = Env {store}} = do
-  t_ <- atomically (swapTVar acThread Nothing) $>>= (liftIO . deRefWeak)
+  t_ <- atomically' (swapTVar acThread Nothing) $>>= (liftIO . deRefWeak)
  disconnectAgentClient c
  mapM_ killThread t_
  liftIO $ closeSQLiteStore store
@@ -405,7 +405,7 @@ testProtocolServer c userId srv = withAgentEnv' c $ case protocolTypeI @p of
 -- | set SOCKS5 proxy on/off and optionally set TCP timeout
 setNetworkConfig :: AgentClient -> NetworkConfig -> IO ()
 setNetworkConfig c cfg' = do
-  cfg <- atomically $ do
+  cfg <- atomically' $ do
    swapTVar (useNetworkConfig c) cfg'
  when (cfg /= cfg') $ reconnectAllServers c

@@ -522,7 +522,7 @@ getAgentStats :: AgentClient -> IO [(AgentStatsKey, Int)]
 getAgentStats c = readTVarIO (agentStats c) >>= mapM (\(k, cnt) -> (k,) <$> readTVarIO cnt) . M.assocs

 resetAgentStats :: AgentClient -> IO ()
-resetAgentStats = atomically . TM.clear . agentStats
+resetAgentStats = atomically' . TM.clear . agentStats
 {-# INLINE resetAgentStats #-}

 withAgentEnv' :: AgentClient -> AM' a -> IO a
@@ -545,9 +545,9 @@ runAgentClient c = race_ (subscriber c) (client c)

 client :: AgentClient -> AM' ()
 client c@AgentClient {rcvQ, subQ} = forever $ do
-  (corrId, entId, cmd) <- atomically $ readTBQueue rcvQ
+  (corrId, entId, cmd) <- atomically' $ readTBQueue rcvQ
  runExceptT (processCommand c (entId, cmd))
-    >>= atomically . writeTBQueue subQ . \case
+    >>= atomically' . writeTBQueue subQ . \case
      Left e -> (corrId, entId, APC SAEConn $ ERR e)
      Right (entId', resp) -> (corrId, entId', resp)

@@ -587,7 +587,7 @@ deleteUser' c userId delSMPQueues = do
  atomically $ TM.delete userId $ smpServers c
  where
    delUser =
-      whenM (withStore' c (`deleteUserWithoutConns` userId)) . atomically $
+      whenM (withStore' c (`deleteUserWithoutConns` userId)) . atomically' $
        writeTBQueue (subQ c) ("", "", APC SAENone $ DEL_USER userId)

 newConnAsync :: ConnectionModeI c => AgentClient -> UserId -> ACorrId -> Bool -> SConnectionMode c -> CR.InitialKeys -> SubscriptionMode -> AM ConnId
@@ -709,7 +709,7 @@ newRcvConnSrv c userId connId enableNtfs cMode clientData pqInitKeys subMode srv
    SMSubscribe -> addSubscription c rq'
  when enableNtfs $ do
    ns <- asks ntfSupervisor
-    atomically $ sendNtfSubCommand ns (connId, NSCCreate)
+    atomically' $ sendNtfSubCommand ns (connId, NSCCreate)
  let pqEnc = CR.connPQEncryption pqInitKeys
      crData = ConnReqUriData SSSimplex (smpAgentVRange pqEnc) [qUri] clientData
      e2eVRange = e2eEncryptVRange pqEnc
@@ -769,7 +769,7 @@ compatibleContactUri (CRContactUri ConnReqUriData {crAgentVRange, crSmpQueues =
  AgentConfig {smpClientVRange, smpAgentVRange} <- asks config
  pure $
    (,)
-      <$> (qUri `compatibleVersion` smpClientVRange) 
+      <$> (qUri `compatibleVersion` smpClientVRange)
      <*> (crAgentVRange `compatibleVersion` smpAgentVRange pqSup)

 versionPQSupport_ :: VersionSMPA -> Maybe CR.VersionE2E -> PQSupport
@@ -820,7 +820,7 @@ createReplyQueue c ConnData {userId, connId, enableNtfs} SndQueue {smpClientVers
    SMSubscribe -> addSubscription c rq'
  when enableNtfs $ do
    ns <- asks ntfSupervisor
-    atomically $ sendNtfSubCommand ns (connId, NSCCreate)
+    atomically' $ sendNtfSubCommand ns (connId, NSCCreate)
  pure qInfo

 -- | Approve confirmation (LET command) in Reader monad
@@ -930,7 +930,7 @@ subscribeConnections' c connIds = do
    notifyResultError rs = do
      let actual = M.size rs
          expected = length connIds
-      when (actual /= expected) . atomically $
+      when (actual /= expected) . atomically' $
        writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ INTERNAL $ "subscribeConnections result size: " <> show actual <> ", expected " <> show expected)

 resubscribeConnection' :: AgentClient -> ConnId -> AM ()
@@ -941,13 +941,13 @@ resubscribeConnections' :: AgentClient -> [ConnId] -> AM (Map ConnId (Either Age
 resubscribeConnections' _ [] = pure M.empty
 resubscribeConnections' c connIds = do
  let r = M.fromList . zip connIds . repeat $ Right ()
-  connIds' <- filterM (fmap not . atomically . hasActiveSubscription c) connIds
+  connIds' <- filterM (fmap not . atomically' . hasActiveSubscription c) connIds
  -- union is left-biased, so results returned by subscribeConnections' take precedence
  (`M.union` r) <$> subscribeConnections' c connIds'

 getConnectionMessage' :: AgentClient -> ConnId -> AM (Maybe SMPMsgMeta)
 getConnectionMessage' c connId = do
-  whenM (atomically $ hasActiveSubscription c connId) . throwError $ CMD PROHIBITED
+  whenM (atomically' $ hasActiveSubscription c connId) . throwError $ CMD PROHIBITED
  SomeConn _ conn <- withStore c (`getConn` connId)
  case conn of
    DuplexConnection _ (rq :| _) _ -> getQueueMessage c rq
@@ -1033,7 +1033,7 @@ resumeConnCmds c connId =
    withStore' c (`getPendingCommandServers` connId)
      >>= mapM_ (lift . resumeSrvCmds c)
  where
-    connQueued = atomically $ isJust <$> TM.lookupInsert connId True (connCmdsQueued c)
+    connQueued = atomically' $ isJust <$> TM.lookupInsert connId True (connCmdsQueued c)

 getAsyncCmdWorker :: Bool -> AgentClient -> Maybe SMPServer -> AM' Worker
 getAsyncCmdWorker hasWork c server =
@@ -1043,10 +1043,10 @@ runCommandProcessing :: AgentClient -> Maybe SMPServer -> Worker -> AM ()
 runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do
  ri <- asks $ messageRetryInterval . config -- different retry interval?
  forever $ do
-    atomically $ endAgentOperation c AOSndNetwork
+    atomically' $ endAgentOperation c AOSndNetwork
    lift $ waitForWork doWork
-    atomically $ throwWhenInactive c
-    atomically $ beginAgentOperation c AOSndNetwork
+    atomically' $ throwWhenInactive c
+    atomically' $ beginAgentOperation c AOSndNetwork
    withWork c doWork (`getPendingServerCommand` server_) $ processCmd (riFast ri)
  where
    processCmd :: RetryInterval -> PendingCommand -> AM ()
@@ -1126,7 +1126,7 @@ runCommandProcessing c@AgentClient {subQ} server_ Worker {doWork} = do
                    withStore' c $ \db -> deleteConnRcvQueue db rq'
                    when (enableNtfs cData) $ do
                      ns <- asks ntfSupervisor
-                      atomically $ sendNtfSubCommand ns (connId, NSCCreate)
+                      atomically' $ sendNtfSubCommand ns (connId, NSCCreate)
                    let conn' = DuplexConnection cData (rq'' :| rqs') sqs
                    notify $ SWITCH QDRcv SPCompleted $ connectionStats conn'
              _ -> internalErr "ICQDelete: cannot delete the only queue in connection"
@@ -1256,14 +1256,14 @@ runSmpQueueMsgDelivery :: AgentClient -> ConnData -> SndQueue -> (Worker, TMVar
 runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork}, qLock) = do
  AgentConfig {messageRetryInterval = ri, messageTimeout, helloTimeout, quotaExceededTimeout} <- asks config
  forever $ do
-    atomically $ endAgentOperation c AOSndNetwork
+    atomically' $ endAgentOperation c AOSndNetwork
    lift $ waitForWork doWork
-    atomically $ throwWhenInactive c
-    atomically $ throwWhenNoDelivery c sq
-    atomically $ beginAgentOperation c AOSndNetwork
+    atomically' $ throwWhenInactive c
+    atomically' $ throwWhenNoDelivery c sq
+    atomically' $ beginAgentOperation c AOSndNetwork
    withWork c doWork (\db -> getPendingQueueMsg db connId sq) $
      \(rq_, PendingMsgData {msgId, msgType, msgBody, pqEncryption, msgFlags, msgRetryState, internalTs}) -> do
-        atomically $ endAgentOperation c AOMsgDelivery -- this operation begins in submitPendingMsg
+        atomically' $ endAgentOperation c AOMsgDelivery -- this operation begins in submitPendingMsg
        let mId = unId msgId
            ri' = maybe id updateRetryInterval2 msgRetryState ri
        withRetryLock2 ri' qLock $ \riState loop -> do
@@ -1399,10 +1399,10 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork

 retrySndOp :: AgentClient -> AM () -> AM ()
 retrySndOp c loop = do
-  -- end... is in a separate atomically because if begin... blocks, SUSPENDED won't be sent
-  atomically $ endAgentOperation c AOSndNetwork
-  atomically $ throwWhenInactive c
-  atomically $ beginAgentOperation c AOSndNetwork
+  -- end... is in a separate atomically' because if begin... blocks, SUSPENDED won't be sent
+  atomically' $ endAgentOperation c AOSndNetwork
+  atomically' $ throwWhenInactive c
+  atomically' $ beginAgentOperation c AOSndNetwork
  loop

 ackMessage' :: AgentClient -> ConnId -> AgentMsgId -> Maybe MsgReceiptInfo -> AM ()
@@ -1545,7 +1545,7 @@ connRcvQueues = \case

 disableConn :: AgentClient -> ConnId -> AM' ()
 disableConn c connId = do
-  atomically $ removeSubscription c connId
+  atomically' $ removeSubscription c connId
  ns <- asks ntfSupervisor
  atomically $ writeTBQueue (ntfSubQ ns) (connId, NSCDelete)

@@ -1589,7 +1589,7 @@ prepareDeleteConnections_ getConnections c waitDelivery connIds = do
    rcvQueues (SomeConn _ conn) = case connRcvQueues conn of
      [] -> Left $ Right ()
      rqs -> Right rqs
-    notify = atomically . writeTBQueue (subQ c)
+    notify = atomically' . writeTBQueue (subQ c)

 deleteConnQueues :: AgentClient -> Bool -> Bool -> [RcvQueue] -> AM' (Map ConnId (Either AgentErrorType ()))
 deleteConnQueues c waitDelivery ntf rqs = do
@@ -1618,7 +1618,7 @@ deleteConnQueues c waitDelivery ntf rqs = do
            | temporaryOrHostError e && deleteErrors rq + 1 < maxErrs -> incRcvDeleteErrors db rq $> ((rq, r), Nothing)
            | otherwise -> deleteConnRcvQueue db rq $> ((rq, Right ()), Just (notifyRQ rq (Just e)))
    notifyRQ rq e_ = notify ("", qConnId rq, APC SAEConn $ DEL_RCVQ (qServer rq) (queueId rq) e_)
-    notify = when ntf . atomically . writeTBQueue (subQ c)
+    notify = when ntf . atomically' . writeTBQueue (subQ c)
    connResults :: [(RcvQueue, Either AgentErrorType ())] -> Map ConnId (Either AgentErrorType ())
    connResults = M.map snd . foldl' addResult M.empty
      where
@@ -1653,7 +1653,7 @@ deleteConnections_ getConnections ntf waitDelivery c connIds = do
    notifyResultError rs = do
      let actual = M.size rs
          expected = length connIds
-      when (actual /= expected) . atomically $
+      when (actual /= expected) . atomically' $
        writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ INTERNAL $ "deleteConnections result size: " <> show actual <> ", expected " <> show expected)

 getConnectionServers' :: AgentClient -> ConnId -> AM ConnectionStats
@@ -1713,7 +1713,7 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
        (Just tknId, Just NTACheck)
          | savedDeviceToken == suppliedDeviceToken -> do
              ns <- asks ntfSupervisor
-              atomically $ nsUpdateToken ns tkn {ntfMode = suppliedNtfMode}
+              atomically' $ nsUpdateToken ns tkn {ntfMode = suppliedNtfMode}
              when (ntfTknStatus == NTActive) $ do
                cron <- asks $ ntfCron . config
                agentNtfEnableCron c tknId tkn cron
@@ -1726,7 +1726,7 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
          agentNtfDeleteToken c tknId tkn
          withStore' c (`removeNtfToken` tkn)
          ns <- asks ntfSupervisor
-          atomically $ nsRemoveNtfToken ns
+          atomically' $ nsRemoveNtfToken ns
          pure NTExpired
        _ -> pure ntfTknStatus
      withStore' c $ \db -> updateNtfMode db tkn suppliedNtfMode
@@ -1740,13 +1740,13 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
              then throwError e
              else do
                withStore' c $ \db -> removeNtfToken db tkn
-                atomically $ nsRemoveNtfToken ns
+                atomically' $ nsRemoveNtfToken ns
                createToken
          where
            tryReplace ns = do
              agentNtfReplaceToken c tknId tkn suppliedDeviceToken
              withStore' c $ \db -> updateDeviceToken db tkn suppliedDeviceToken
-              atomically $ nsUpdateToken ns tkn {deviceToken = suppliedDeviceToken, ntfTknStatus = NTRegistered, ntfMode = suppliedNtfMode}
+              atomically' $ nsUpdateToken ns tkn {deviceToken = suppliedDeviceToken, ntfTknStatus = NTRegistered, ntfMode = suppliedNtfMode}
              pure NTRegistered
    _ -> createToken
  where
@@ -1771,7 +1771,7 @@ registerNtfToken' c suppliedDeviceToken suppliedNtfMode =
      let dhSecret = C.dh' srvPubDhKey privDhKey
      withStore' c $ \db -> updateNtfTokenRegistration db tkn tknId dhSecret
      ns <- asks ntfSupervisor
-      atomically $ nsUpdateToken ns tkn {deviceToken = suppliedDeviceToken, ntfTknStatus = NTRegistered, ntfMode = suppliedNtfMode}
+      atomically' $ nsUpdateToken ns tkn {deviceToken = suppliedDeviceToken, ntfTknStatus = NTRegistered, ntfMode = suppliedNtfMode}

 verifyNtfToken' :: AgentClient -> DeviceToken -> C.CbNonce -> ByteString -> AM ()
 verifyNtfToken' c deviceToken nonce code =
@@ -1834,7 +1834,7 @@ toggleConnectionNtfs' c connId enable = do
          withStore' c $ \db -> setConnectionNtfs db connId enable
          ns <- asks ntfSupervisor
          let cmd = if enable then NSCCreate else NSCDelete
-          atomically $ sendNtfSubCommand ns (connId, cmd)
+          atomically' $ sendNtfSubCommand ns (connId, cmd)

 deleteToken_ :: AgentClient -> NtfToken -> AM ()
 deleteToken_ c tkn@NtfToken {ntfTokenId, ntfTknStatus} = do
@@ -1842,28 +1842,28 @@ deleteToken_ c tkn@NtfToken {ntfTokenId, ntfTknStatus} = do
  forM_ ntfTokenId $ \tknId -> do
    let ntfTknAction = Just NTADelete
    withStore' c $ \db -> updateNtfToken db tkn ntfTknStatus ntfTknAction
-    atomically $ nsUpdateToken ns tkn {ntfTknStatus, ntfTknAction}
+    atomically' $ nsUpdateToken ns tkn {ntfTknStatus, ntfTknAction}
    agentNtfDeleteToken c tknId tkn `catchAgentError` \case
      NTF AUTH -> pure ()
      e -> throwError e
  withStore' c $ \db -> removeNtfToken db tkn
-  atomically $ nsRemoveNtfToken ns
+  atomically' $ nsRemoveNtfToken ns

 withToken :: AgentClient -> NtfToken -> Maybe (NtfTknStatus, NtfTknAction) -> (NtfTknStatus, Maybe NtfTknAction) -> AM a -> AM NtfTknStatus
 withToken c tkn@NtfToken {deviceToken, ntfMode} from_ (toStatus, toAction_) f = do
  ns <- asks ntfSupervisor
  forM_ from_ $ \(status, action) -> do
    withStore' c $ \db -> updateNtfToken db tkn status (Just action)
-    atomically $ nsUpdateToken ns tkn {ntfTknStatus = status, ntfTknAction = Just action}
+    atomically' $ nsUpdateToken ns tkn {ntfTknStatus = status, ntfTknAction = Just action}
  tryError f >>= \case
    Right _ -> do
      withStore' c $ \db -> updateNtfToken db tkn toStatus toAction_
      let updatedToken = tkn {ntfTknStatus = toStatus, ntfTknAction = toAction_}
-      atomically $ nsUpdateToken ns updatedToken
+      atomically' $ nsUpdateToken ns updatedToken
      pure toStatus
    Left e@(NTF AUTH) -> do
      withStore' c $ \db -> removeNtfToken db tkn
-      atomically $ nsRemoveNtfToken ns
+      atomically' $ nsRemoveNtfToken ns
      void $ registerNtfToken' c deviceToken ntfMode
      throwError e
    Left e -> throwError e
@@ -1875,13 +1875,13 @@ initializeNtfSubs c = sendNtfConnCommands c NSCCreate
 deleteNtfSubs :: AgentClient -> NtfSupervisorCommand -> AM ()
 deleteNtfSubs c deleteCmd = do
  ns <- asks ntfSupervisor
-  void . atomically . flushTBQueue $ ntfSubQ ns
+  void . atomically' . flushTBQueue $ ntfSubQ ns
  sendNtfConnCommands c deleteCmd

 sendNtfConnCommands :: AgentClient -> NtfSupervisorCommand -> AM ()
 sendNtfConnCommands c cmd = do
  ns <- asks ntfSupervisor
-  connIds <- atomically $ getSubscriptions c
+  connIds <- atomically' $ getSubscriptions c
  forM_ connIds $ \connId -> do
    withStore' c (`getConnData` connId) >>= \case
      Just (ConnData {enableNtfs}, _) ->
@@ -1910,7 +1910,7 @@ suspendAgent c 0 = do
    suspend opSel = atomically $ modifyTVar' (opSel c) $ \s -> s {opSuspended = True}
 suspendAgent c@AgentClient {agentState = as} maxDelay = do
  state <-
-    atomically $ do
+    atomically' $ do
      writeTVar as ASSuspending
      suspendOperation c AONtfNetwork $ pure ()
      suspendOperation c AORcvNetwork $
@@ -1920,7 +1920,7 @@ suspendAgent c@AgentClient {agentState = as} maxDelay = do
  when (state == ASSuspending) . void . forkIO $ do
    threadDelay maxDelay
    -- liftIO $ putStrLn "suspendAgent after timeout"
-    atomically . whenSuspending c $ do
+    atomically' . whenSuspending c $ do
      -- unsafeIOToSTM $ putStrLn $ "in timeout: suspendSendingAndDatabase"
      suspendSendingAndDatabase c

@@ -1934,10 +1934,10 @@ debugAgentLocks :: AgentClient -> IO AgentLocks
 debugAgentLocks AgentClient {connLocks = cs, invLocks = is, deleteLock = d} = do
  connLocks <- getLocks cs
  invLocks <- getLocks is
-  delLock <- atomically $ tryReadTMVar d
+  delLock <- atomically' $ tryReadTMVar d
  pure AgentLocks {connLocks, invLocks, delLock}
  where
-    getLocks ls = atomically $ M.mapKeys (B.unpack . strEncode) . M.mapMaybe id <$> (mapM tryReadTMVar =<< readTVar ls)
+    getLocks ls = atomically' $ M.mapKeys (B.unpack . strEncode) . M.mapMaybe id <$> (mapM tryReadTMVar =<< readTVar ls)

 getSMPServer :: AgentClient -> UserId -> AM SMPServerWithAuth
 getSMPServer c userId = withUserServers c userId pickServer
@@ -1945,7 +1945,7 @@ getSMPServer c userId = withUserServers c userId pickServer

 subscriber :: AgentClient -> AM' ()
 subscriber c@AgentClient {msgQ} = forever $ do
-  t <- atomically $ readTBQueue msgQ
+  t <- atomically' $ readTBQueue msgQ
  agentOperationBracket c AORcvNetwork waitUntilActive $
    runExceptT (processSMPTransmission c t) >>= \case
      Left e -> liftIO $ print e
@@ -1977,7 +1977,7 @@ cleanupManager c@AgentClient {subQ} = do
      step <- asks $ cleanupStepInterval . config
      liftIO $ threadDelay step
    -- we are catching it to avoid CRITICAL errors in tests when this is the only remaining handle to active
-    waitActive a = liftIO (E.tryAny . atomically $ waitUntilActive c) >>= either (\_ -> pure ()) (\_ -> void a)
+    waitActive a = liftIO (E.tryAny . atomically' $ waitUntilActive c) >>= either (\_ -> pure ()) (\_ -> void a)
    deleteConns =
      withLock (deleteLock c) "cleanupManager" $ do
        void $ withStore' c getDeletedConnIds >>= deleteDeletedConns c
@@ -2042,7 +2042,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
              ack' <- handleNotifyAck $ case msg' of
                SMP.ClientRcvMsgBody {msgTs = srvTs, msgFlags, msgBody} -> processClientMsg srvTs msgFlags msgBody
                SMP.ClientRcvMsgQuota {} -> queueDrained >> ack
-              whenM (atomically $ hasGetLock c rq) $
+              whenM (atomically' $ hasGetLock c rq) $
                notify (MSGNTF $ SMP.rcvMessageMeta srvMsgId msg')
              pure ack'
            where
@@ -2204,7 +2204,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
            notify . ERR $ BROKER (B.unpack $ strEncode srv) UNEXPECTED
        where
          notify :: forall e m. MonadIO m => AEntityI e => ACommand 'Agent e -> m ()
-          notify = atomically . notify'
+          notify = atomically' . notify'

          notify' :: forall e. AEntityI e => ACommand 'Agent e -> STM ()
          notify' msg = writeTBQueue subQ ("", connId, APC (sAEntity @e) msg)
@@ -2307,7 +2307,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), _v,
            case findQ addr sqs of
              Just sq -> do
                logServer "<--" c srv rId $ "MSG <QCONT>:" <> logSecret srvMsgId
-                atomically $
+                atomically' $
                  TM.lookup (qAddress sq) (smpDeliveryWorkers c)
                    >>= mapM_ (\(_, retryLock) -> tryPutTMVar retryLock ())
              Nothing -> qError "QCONT: queue address not found"
@@ -162,6 +162,7 @@ import Data.Text.Encoding
 import Data.Time (UTCTime, defaultTimeLocale, formatTime, getCurrentTime)
 import Data.Time.Clock.System (getSystemTime)
 import Data.Word (Word16)
+import GHC.Stack (HasCallStack, withFrozenCallStack)
 import Network.Socket (HostName)
 import Simplex.FileTransfer.Client (XFTPChunkSpec (..), XFTPClient, XFTPClientConfig (..), XFTPClientError)
 import qualified Simplex.FileTransfer.Client as X
@@ -300,7 +301,7 @@ getAgentWorker = getAgentWorker' id pure

 getAgentWorker' :: forall a k. (Ord k, Show k) => (a -> Worker) -> (Worker -> STM a) -> String -> Bool -> AgentClient -> k -> TMap k a -> (a -> AM ()) -> AM' a
 getAgentWorker' toW fromW name hasWork c key ws work = do
-  atomically (getWorker >>= maybe createWorker whenExists) >>= \w -> runWorker w $> w
+  atomically' (getWorker >>= maybe createWorker whenExists) >>= \w -> runWorker w $> w
  where
    getWorker = TM.lookup key ws
    createWorker = do
@@ -319,7 +320,7 @@ getAgentWorker' toW fromW name hasWork c key ws work = do
          t <- liftIO getSystemTime
          maxRestarts <- asks $ maxWorkerRestartsPerMin . config
          -- worker may terminate because it was deleted from the map (getWorker returns Nothing), then it won't restart
-          restart <- atomically $ getWorker >>= maybe (pure False) (shouldRestart e_ (toW w) t maxRestarts)
+          restart <- atomically' $ getWorker >>= maybe (pure False) (shouldRestart e_ (toW w) t maxRestarts)
          when restart runWork
        shouldRestart e_ Worker {workerId = wId, doWork, action, restarts} t maxRestarts w'
          | wId == workerId (toW w') =
@@ -355,11 +356,11 @@ newWorker c = do
 runWorkerAsync :: Worker -> AM' () -> AM' ()
 runWorkerAsync Worker {action} work =
  E.bracket
-    (atomically $ takeTMVar action) -- get current action, locking to avoid race conditions
-    (atomically . tryPutTMVar action) -- if it was running (or if start crashes), put it back and unlock (don't lock if it was just started)
+    (atomically' $ takeTMVar action) -- get current action, locking to avoid race conditions
+    (atomically' . tryPutTMVar action) -- if it was running (or if start crashes), put it back and unlock (don't lock if it was just started)
    (\a -> when (isNothing a) start) -- start worker if it's not running
  where
-    start = atomically . putTMVar action . Just =<< async work
+    start = atomically' . putTMVar action . Just =<< async work

 data AgentOperation = AONtfNetwork | AORcvNetwork | AOMsgDelivery | AOSndNetwork | AODatabase
  deriving (Eq, Show)
@@ -517,7 +518,7 @@ instance ProtocolServerClient XFTPVersion XFTPErrorType FileResponse where
 getSMPServerClient :: AgentClient -> SMPTransportSession -> AM SMPClient
 getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv, _) = do
  unlessM (readTVarIO active) . throwError $ INACTIVE
-  atomically (getTSessVar c tSess smpClients)
+  atomically' (getTSessVar c tSess smpClients)
    >>= either newClient (waitForProtocolClient c tSess)
  where
    -- we resubscribe only on newClient error, but not on waitForProtocolClient error,
@@ -544,7 +545,7 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv,
        -- because we can have a race condition when a new current client could have already
        -- made subscriptions active, and the old client would be processing diconnection later.
        removeClientAndSubs :: IO ([RcvQueue], [ConnId])
-        removeClientAndSubs = atomically $ ifM currentActiveClient removeSubs $ pure ([], [])
+        removeClientAndSubs = atomically' $ ifM currentActiveClient removeSubs $ pure ([], [])
          where
            currentActiveClient = (&&) <$> removeTSessVar' v tSess smpClients <*> readTVar active
            removeSubs = do
@@ -558,7 +559,7 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv,
          notifySub "" $ hostEvent DISCONNECT client
          unless (null conns) $ notifySub "" $ DOWN srv conns
          unless (null qs) $ do
-            atomically $ mapM_ (releaseGetLock c) qs
+            atomically' $ mapM_ (releaseGetLock c) qs
            runReaderT (resubscribeSMPSession c tSess) env

        notifySub :: forall e. AEntityI e => ConnId -> ACommand 'Agent e -> IO ()
@@ -566,7 +567,7 @@ getSMPServerClient c@AgentClient {active, smpClients, msgQ} tSess@(userId, srv,

 resubscribeSMPSession :: AgentClient -> SMPTransportSession -> AM' ()
 resubscribeSMPSession c@AgentClient {smpSubWorkers} tSess =
-  atomically getWorkerVar >>= mapM_ (either newSubWorker (\_ -> pure ()))
+  atomically' getWorkerVar >>= mapM_ (either newSubWorker (\_ -> pure ()))
  where
    getWorkerVar =
      ifM
@@ -574,13 +575,13 @@ resubscribeSMPSession c@AgentClient {smpSubWorkers} tSess =
        (pure Nothing) -- prevent race with cleanup and adding pending queues in another call
        (Just <$> getTSessVar c tSess smpSubWorkers)
    newSubWorker v = do
-      a <- async $ void (E.tryAny runSubWorker) >> atomically (cleanup v)
-      atomically $ putTMVar (sessionVar v) a
+      a <- async $ void (E.tryAny runSubWorker) >> atomically' (cleanup v)
+      atomically' $ putTMVar (sessionVar v) a
    runSubWorker = do
      ri <- asks $ reconnectInterval . config
      timeoutCounts <- newTVarIO 0
      withRetryInterval ri $ \_ loop -> do
-        pending <- atomically getPending
+        pending <- atomically' getPending
        forM_ (L.nonEmpty pending) $ \qs -> do
          void . tryAgentError' $ reconnectSMPClient timeoutCounts c tSess qs
          loop
@@ -626,7 +627,7 @@ reconnectSMPClient tc c tSess@(_, srv, _) qs = do
 getNtfServerClient :: AgentClient -> NtfTransportSession -> AM NtfClient
 getNtfServerClient c@AgentClient {active, ntfClients} tSess@(userId, srv, _) = do
  unlessM (readTVarIO active) . throwError $ INACTIVE
-  atomically (getTSessVar c tSess ntfClients)
+  atomically' (getTSessVar c tSess ntfClients)
    >>= either
      (newProtocolClient c tSess ntfClients connectClient)
      (waitForProtocolClient c tSess)
@@ -641,7 +642,7 @@ getNtfServerClient c@AgentClient {active, ntfClients} tSess@(userId, srv, _) = d

    clientDisconnected :: NtfClientVar -> NtfClient -> IO ()
    clientDisconnected v client = do
-      atomically $ removeTSessVar v tSess ntfClients
+      atomically' $ removeTSessVar v tSess ntfClients
      incClientStat c userId client "DISCONNECT" ""
      atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent DISCONNECT client)
      logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
@@ -649,7 +650,7 @@ getNtfServerClient c@AgentClient {active, ntfClients} tSess@(userId, srv, _) = d
 getXFTPServerClient :: AgentClient -> XFTPTransportSession -> AM XFTPClient
 getXFTPServerClient c@AgentClient {active, xftpClients, useNetworkConfig} tSess@(userId, srv, _) = do
  unlessM (readTVarIO active) . throwError $ INACTIVE
-  atomically (getTSessVar c tSess xftpClients)
+  atomically' (getTSessVar c tSess xftpClients)
    >>= either
      (newProtocolClient c tSess xftpClients connectClient)
      (waitForProtocolClient c tSess)
@@ -665,7 +666,7 @@ getXFTPServerClient c@AgentClient {active, xftpClients, useNetworkConfig} tSess@

    clientDisconnected :: XFTPClientVar -> XFTPClient -> IO ()
    clientDisconnected v client = do
-      atomically $ removeTSessVar v tSess xftpClients
+      atomically' $ removeTSessVar v tSess xftpClients
      incClientStat c userId client "DISCONNECT" ""
      atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent DISCONNECT client)
      logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv
@@ -694,7 +695,7 @@ removeTSessVar' v tSess vs =
 waitForProtocolClient :: ProtocolTypeI (ProtoType msg) => AgentClient -> TransportSession msg -> ClientVar msg -> AM (Client msg)
 waitForProtocolClient c (_, srv, _) v = do
  NetworkConfig {tcpConnectTimeout} <- readTVarIO $ useNetworkConfig c
-  client_ <- liftIO $ tcpConnectTimeout `timeout` atomically (readTMVar $ sessionVar v)
+  client_ <- liftIO $ tcpConnectTimeout `timeout` atomically' (readTMVar $ sessionVar v)
  liftEither $ case client_ of
    Just (Right smpClient) -> Right smpClient
    Just (Left e) -> Left e
@@ -714,13 +715,13 @@ newProtocolClient c tSess@(userId, srv, entityId_) clients connectClient v =
  tryAgentError (connectClient v) >>= \case
    Right client -> do
      logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv <> " (user " <> bshow userId <> maybe "" (" for entity " <>) entityId_ <> ")"
-      atomically $ putTMVar (sessionVar v) (Right client)
+      atomically' $ putTMVar (sessionVar v) (Right client)
      liftIO $ incClientStat c userId client "CLIENT" "OK"
      atomically $ writeTBQueue (subQ c) ("", "", APC SAENone $ hostEvent CONNECT client)
      pure client
    Left e -> do
      liftIO $ incServerStat c userId srv "CLIENT" $ strEncode e
-      atomically $ do
+      atomically' $ do
        removeTSessVar v tSess clients
        putTMVar (sessionVar v) (Left e)
      throwError e -- signal error to caller
@@ -740,26 +741,26 @@ closeAgentClient c = do
  closeProtocolServerClients c smpClients
  closeProtocolServerClients c ntfClients
  closeProtocolServerClients c xftpClients
-  atomically (swapTVar (smpSubWorkers c) M.empty) >>= mapM_ cancelReconnect
+  atomically' (swapTVar (smpSubWorkers c) M.empty) >>= mapM_ cancelReconnect
  clearWorkers smpDeliveryWorkers >>= mapM_ (cancelWorker . fst)
  clearWorkers asyncCmdWorkers >>= mapM_ cancelWorker
  clear connCmdsQueued
-  atomically . RQ.clear $ activeSubs c
-  atomically . RQ.clear $ pendingSubs c
+  atomically' . RQ.clear $ activeSubs c
+  atomically' . RQ.clear $ pendingSubs c
  clear subscrConns
  clear getMsgLocks
  where
    clearWorkers :: Ord k => (AgentClient -> TMap k a) -> IO (Map k a)
-    clearWorkers workers = atomically $ swapTVar (workers c) mempty
+    clearWorkers workers = atomically' $ swapTVar (workers c) mempty
    clear :: Monoid m => (AgentClient -> TVar m) -> IO ()
    clear sel = atomically $ writeTVar (sel c) mempty
    cancelReconnect :: SessionVar (Async ()) -> IO ()
-    cancelReconnect v = void . forkIO $ atomically (readTMVar $ sessionVar v) >>= uninterruptibleCancel
+    cancelReconnect v = void . forkIO $ atomically' (readTMVar $ sessionVar v) >>= uninterruptibleCancel

 cancelWorker :: Worker -> IO ()
 cancelWorker Worker {doWork, action} = do
  noWorkToDo doWork
-  atomically (tryTakeTMVar action) >>= mapM_ (mapM_ uninterruptibleCancel)
+  atomically' (tryTakeTMVar action) >>= mapM_ (mapM_ uninterruptibleCancel)

 waitUntilActive :: AgentClient -> STM ()
 waitUntilActive c = unlessM (readTVar $ active c) retry
@@ -777,7 +778,7 @@ throwWhenNoDelivery c sq =

 closeProtocolServerClients :: ProtocolServerClient v err msg => AgentClient -> (AgentClient -> TMap (TransportSession msg) (ClientVar msg)) -> IO ()
 closeProtocolServerClients c clientsSel =
-  atomically (clientsSel c `swapTVar` M.empty) >>= mapM_ (forkIO . closeClient_ c)
+  atomically' (clientsSel c `swapTVar` M.empty) >>= mapM_ (forkIO . closeClient_ c)

 reconnectServerClients :: ProtocolServerClient v err msg => AgentClient -> (AgentClient -> TMap (TransportSession msg) (ClientVar msg)) -> IO ()
 reconnectServerClients c clientsSel =
@@ -790,7 +791,7 @@ closeClient c clientSel tSess =
 closeClient_ :: ProtocolServerClient v err msg => AgentClient -> ClientVar msg -> IO ()
 closeClient_ c v = do
  NetworkConfig {tcpConnectTimeout} <- readTVarIO $ useNetworkConfig c
-  tcpConnectTimeout `timeout` atomically (readTMVar $ sessionVar v) >>= \case
+  tcpConnectTimeout `timeout` atomically' (readTMVar $ sessionVar v) >>= \case
    Just (Right client) -> closeProtocolServerClient client `catchAll_` pure ()
    _ -> pure ()

@@ -1081,7 +1082,7 @@ processSubResult :: AgentClient -> RcvQueue -> Either SMPClientError () -> IO (E
 processSubResult c rq r = do
  case r of
    Left e ->
-      unless (temporaryClientError e) . atomically $ do
+      unless (temporaryClientError e) . atomically' $ do
        RQ.deleteQueue rq (pendingSubs c)
        TM.insert (RQ.qKey rq) e (removedSubs c)
    _ -> addSubscription c rq
@@ -1105,7 +1106,7 @@ temporaryOrHostError = \case
 subscribeQueues :: AgentClient -> [RcvQueue] -> AM' [(RcvQueue, Either AgentErrorType ())]
 subscribeQueues c qs = do
  (errs, qs') <- partitionEithers <$> mapM checkQueue qs
-  atomically $ do
+  atomically' $ do
    modifyTVar' (subscrConns c) (`S.union` S.fromList (map qConnId qs'))
    RQ.batchAddQueues (pendingSubs c) qs'
  env <- ask
@@ -1113,7 +1114,7 @@ subscribeQueues c qs = do
  (errs <>) <$> sendTSessionBatches "SUB" 90 id (subscribeQueues_ env) c qs'
  where
    checkQueue rq = do
-      prohibited <- atomically $ hasGetLock c rq
+      prohibited <- atomically' $ hasGetLock c rq
      pure $ if prohibited then Left (rq, Left $ CMD PROHIBITED) else Right rq
    subscribeQueues_ :: Env -> SMPClient -> NonEmpty RcvQueue -> IO (BatchResponses SMPClientError ())
    subscribeQueues_ env smp qs' = do
@@ -1159,7 +1160,7 @@ sendBatch smpCmdFunc smp qs = L.zip qs <$> smpCmdFunc smp (L.map queueCreds qs)
    queueCreds RcvQueue {rcvPrivateKey, rcvId} = (rcvPrivateKey, rcvId)

 addSubscription :: AgentClient -> RcvQueue -> IO ()
-addSubscription c rq@RcvQueue {connId} = atomically $ do
+addSubscription c rq@RcvQueue {connId} = atomically' $ do
  modifyTVar' (subscrConns c) $ S.insert connId
  RQ.addQueue rq $ activeSubs c
  RQ.deleteQueue rq $ pendingSubs c
@@ -1216,7 +1217,7 @@ sendInvitation c userId (Compatible (SMPQueueInfo v SMPQueueAddress {smpServer,

 getQueueMessage :: AgentClient -> RcvQueue -> AM (Maybe SMPMsgMeta)
 getQueueMessage c rq@RcvQueue {server, rcvId, rcvPrivateKey} = do
-  atomically createTakeGetLock
+  atomically' createTakeGetLock
  msg_ <- withSMPClient c rq "GET" $ \smp ->
    getSMPMessage smp rcvPrivateKey rcvId
  mapM decryptMeta msg_
@@ -1266,7 +1267,7 @@ sendAck :: AgentClient -> RcvQueue -> MsgId -> AM ()
 sendAck c rq@RcvQueue {rcvId, rcvPrivateKey} msgId = do
  withSMPClient c rq ("ACK:" <> logSecret msgId) $ \smp ->
    ackSMPMessage smp rcvPrivateKey rcvId msgId
-  atomically $ releaseGetLock c rq
+  atomically' $ releaseGetLock c rq

 hasGetLock :: AgentClient -> RcvQueue -> STM Bool
 hasGetLock c RcvQueue {server, rcvId} =
@@ -1364,7 +1365,7 @@ agentXFTPDeleteChunk c userId DeletedSndChunkReplica {server, replicaId = ChunkR

 xftpRcvKeys :: Int -> AM (NonEmpty C.AAuthKeyPair)
 xftpRcvKeys n = do
-  rKeys <- atomically . replicateM n . C.generateAuthKeyPair C.SEd25519 =<< asks random
+  rKeys <- atomically' . replicateM n . C.generateAuthKeyPair C.SEd25519 =<< asks random
  case L.nonEmpty rKeys of
    Just rKeys' -> pure rKeys'
    _ -> throwError $ INTERNAL "non-positive number of recipients"
@@ -1374,7 +1375,7 @@ xftpRcvIdsKeys rIds rKeys = L.map ChunkReplicaId rIds `L.zip` L.map snd rKeys

 agentCbEncrypt :: SndQueue -> Maybe C.PublicKeyX25519 -> ByteString -> AM ByteString
 agentCbEncrypt SndQueue {e2eDhSecret, smpClientVersion} e2ePubKey msg = do
-  cmNonce <- atomically . C.randomCbNonce =<< asks random
+  cmNonce <- atomically' . C.randomCbNonce =<< asks random
  let paddedLen = maybe SMP.e2eEncMessageLength (const SMP.e2eEncConfirmationLength) e2ePubKey
  cmEncBody <-
    liftEither . first cryptoError $
@@ -1416,8 +1417,8 @@ cryptoError = \case
  where
    c = AGENT . A_CRYPTO

-waitForWork :: MonadIO m => TMVar () -> m ()
-waitForWork = void . atomically . readTMVar
+waitForWork :: (MonadIO m, HasCallStack) => TMVar () -> m ()
+waitForWork v = withFrozenCallStack $ void . atomically' $ readTMVar v
 {-# INLINE waitForWork #-}

 withWork :: AgentClient -> TMVar () -> (DB.Connection -> IO (Either StoreError (Maybe a))) -> (a -> AM ()) -> AM ()
@@ -1432,7 +1433,7 @@ withWork c doWork getWork action =
    notifyErr err e = atomically $ writeTBQueue (subQ c) ("", "", APC SAEConn $ ERR $ err $ show e)

 noWorkToDo :: TMVar () -> IO ()
-noWorkToDo = void . atomically . tryTakeTMVar
+noWorkToDo = void . atomically' . tryTakeTMVar
 {-# INLINE noWorkToDo #-}

 hasWorkToDo :: Worker -> STM ()
@@ -1499,8 +1500,8 @@ beginAgentOperation c op = do
 agentOperationBracket :: MonadUnliftIO m => AgentClient -> AgentOperation -> (AgentClient -> STM ()) -> m a -> m a
 agentOperationBracket c op check action =
  E.bracket
-    (atomically (check c) >> atomically (beginAgentOperation c op))
-    (\_ -> atomically $ endAgentOperation c op)
+    (atomically' (check c) >> atomically' (beginAgentOperation c op))
+    (\_ -> atomically' $ endAgentOperation c op)
    (const action)

 waitUntilForeground :: AgentClient -> STM ()
@@ -1561,13 +1562,13 @@ incClientStat c userId pc = incClientStatN c userId pc 1
 incServerStat :: AgentClient -> UserId -> ProtocolServer p -> ByteString -> ByteString -> IO ()
 incServerStat c userId ProtocolServer {host} cmd res = do
  threadDelay 100000
-  atomically $ incStat c 1 statsKey
+  atomically' $ incStat c 1 statsKey
  where
    statsKey = AgentStatsKey {userId, host = strEncode $ L.head host, clientTs = "", cmd, res}

 incClientStatN :: ProtocolServerClient v err msg => AgentClient -> UserId -> Client msg -> Int -> ByteString -> ByteString -> IO ()
 incClientStatN c userId pc n cmd res = do
-  atomically $ incStat c n statsKey
+  atomically' $ incStat c n statsKey
  where
    statsKey = AgentStatsKey {userId, host = strEncode $ clientTransportHost pc, clientTs = strEncode $ clientSessionTs pc, cmd, res}

@@ -1582,7 +1583,7 @@ pickServer = \case
  srv :| [] -> pure srv
  servers -> do
    gen <- asks randomServer
-    atomically $ (servers L.!!) <$> stateTVar gen (randomR (0, L.length servers - 1))
+    atomically' $ (servers L.!!) <$> stateTVar gen (randomR (0, L.length servers - 1))

 getNextServer :: forall p. (ProtocolTypeI p, UserProtocol p) => AgentClient -> UserId -> [ProtocolServer p] -> AM (ProtoServerWithAuth p)
 getNextServer c userId usedSrvs = withUserServers c userId $ \srvs ->
@@ -1600,7 +1601,7 @@ withNextSrv :: forall p a. (ProtocolTypeI p, UserProtocol p) => AgentClient -> U
 withNextSrv c userId usedSrvs initUsed action = do
  used <- readTVarIO usedSrvs
  srvAuth@(ProtoServerWithAuth srv _) <- getNextServer c userId used
-  atomically $ do
+  atomically' $ do
    srvs_ <- TM.lookup userId $ userServers c
    let unused = maybe [] ((\\ used) . map protoServer . L.toList) srvs_
        used' = if null unused then initUsed else srv : used
@@ -1688,8 +1689,8 @@ getAgentWorkersDetails AgentClient {smpClients, ntfClients, xftpClients, smpDeli
    workerStats :: StrEncoding k => Map k Worker -> IO (Map Text WorkersDetails)
    workerStats ws = fmap M.fromList . forM (M.toList ws) $ \(qa, Worker {restarts, doWork, action}) -> do
      RestartCount {restartCount} <- readTVarIO restarts
-      hasWork <- atomically $ not <$> isEmptyTMVar doWork
-      hasAction <- atomically $ not <$> isEmptyTMVar action
+      hasWork <- atomically' $ not <$> isEmptyTMVar doWork
+      hasAction <- atomically' $ not <$> isEmptyTMVar action
      pure (textKey qa, WorkersDetails {restarts = restartCount, hasWork, hasAction})
    Env {ntfSupervisor, xftpAgent} = agentEnv
    NtfSupervisor {ntfWorkers, ntfSMPWorkers} = ntfSupervisor
@@ -1754,7 +1755,7 @@ getAgentWorkersSummary AgentClient {smpClients, ntfClients, xftpClients, smpDeli
        byWork WorkersSummary {numActive, numIdle, totalRestarts} Worker {action, restarts} = do
          RestartCount {restartCount} <- readTVarIO restarts
          ifM
-            (atomically $ isJust <$> tryReadTMVar action)
+            (atomically' $ isJust <$> tryReadTMVar action)
            (pure WorkersSummary {numActive, numIdle = numIdle + 1, totalRestarts = totalRestarts + restartCount})
            (pure WorkersSummary {numActive = numActive + 1, numIdle, totalRestarts = totalRestarts + restartCount})

@@ -15,6 +15,7 @@ import Data.Functor (($>))
 import UnliftIO.Async (forConcurrently)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
+import Simplex.Messaging.Util (atomically')

 type Lock = TMVar String

@@ -29,23 +30,23 @@ withLock lock name = ExceptT . withLock' lock name . runExceptT
 withLock' :: MonadUnliftIO m => Lock -> String -> m a -> m a
 withLock' lock name =
  E.bracket_
-    (atomically $ putTMVar lock name)
-    (void . atomically $ takeTMVar lock)
+    (atomically' $ putTMVar lock name)
+    (void . atomically' $ takeTMVar lock)

 withGetLock :: MonadUnliftIO m => (k -> STM Lock) -> k -> String -> m a -> m a
 withGetLock getLock key name a =
  E.bracket
-    (atomically $ getPutLock getLock key name)
-    (atomically . takeTMVar)
+    (atomically' $ getPutLock getLock key name)
+    (atomically' . takeTMVar)
    (const a)

 withGetLocks :: MonadUnliftIO m => (k -> STM Lock) -> [k] -> String -> m a -> m a
 withGetLocks getLock keys name = E.bracket holdLocks releaseLocks . const
  where
-    holdLocks = forConcurrently keys $ \key -> atomically $ getPutLock getLock key name
+    holdLocks = forConcurrently keys $ \key -> atomically' $ getPutLock getLock key name
    -- only this withGetLocks would be holding the locks,
    -- so it's safe to combine all lock releases into one transaction
-    releaseLocks = atomically . mapM_ takeTMVar
+    releaseLocks = atomically' . mapM_ takeTMVar

 -- getLock and putTMVar can be in one transaction on the assumption that getLock doesn't write in case the lock already exists,
 -- and in case it is created and added to some shared resource (we use TMap) it also helps avoid contention for the newly created lock.
@@ -38,7 +38,7 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Notifications.Protocol (NtfSubStatus (..), NtfTknStatus (..), SMPQueueNtf (..))
 import Simplex.Messaging.Notifications.Types
 import Simplex.Messaging.Protocol (NtfServer, SMPServer, sameSrvAddr)
-import Simplex.Messaging.Util (diffToMicroseconds, threadDelay', tshow, unlessM)
+import Simplex.Messaging.Util (diffToMicroseconds, threadDelay', tshow, unlessM, atomically')
 import System.Random (randomR)
 import UnliftIO
 import UnliftIO.Concurrent (forkIO, threadDelay)
@@ -48,7 +48,7 @@ runNtfSupervisor :: AgentClient -> AM' ()
 runNtfSupervisor c = do
  ns <- asks ntfSupervisor
  forever $ do
-    cmd@(connId, _) <- atomically . readTBQueue $ ntfSubQ ns
+    cmd@(connId, _) <- atomically' . readTBQueue $ ntfSubQ ns
    handleErr connId . agentOperationBracket c AONtfNetwork waitUntilActive $
      runExceptT (processNtfSub c cmd) >>= \case
        Left e -> notifyErr connId e
@@ -265,7 +265,7 @@ runNtfSMPWorker c srv Worker {doWork} = do
                  setRcvQueueNtfCreds db connId $ Just ClientNtfCreds {ntfPublicKey, ntfPrivateKey, notifierId, rcvNtfDhSecret}
                  updateNtfSubscription db sub {ntfQueueId = Just notifierId, ntfSubStatus = NASKey} (NtfSubNTFAction NSACreate) ts
                ns <- asks ntfSupervisor
-                atomically $ sendNtfSubCommand ns (connId, NSCNtfWorker ntfServer)
+                atomically' $ sendNtfSubCommand ns (connId, NSCNtfWorker ntfServer)
              _ -> workerInternalError c connId "NSASmpKey - no active token"
          NSASmpDelete -> do
            rq_ <- withStore' c $ \db -> do
@@ -278,10 +278,10 @@ rescheduleAction :: TMVar () -> UTCTime -> UTCTime -> AM' Bool
 rescheduleAction doWork ts actionTs
  | actionTs <= ts = pure False
  | otherwise = do
-      void . atomically $ tryTakeTMVar doWork
+      void . atomically' $ tryTakeTMVar doWork
      void . forkIO $ do
        liftIO $ threadDelay' $ diffToMicroseconds $ diffUTCTime actionTs ts
-        atomically $ hasWorkToDo' doWork
+        atomically' $ hasWorkToDo' doWork
      pure True

 retryOnError :: AgentClient -> Text -> AM () -> (AgentErrorType -> AM ()) -> AgentErrorType -> AM ()
@@ -293,9 +293,9 @@ retryOnError c name loop done e = do
    _ -> done e
  where
    retryLoop = do
-      atomically $ endAgentOperation c AONtfNetwork
-      atomically $ throwWhenInactive c
-      atomically $ beginAgentOperation c AONtfNetwork
+      atomically' $ endAgentOperation c AONtfNetwork
+      atomically' $ throwWhenInactive c
+      atomically' $ beginAgentOperation c AONtfNetwork
      loop

 workerInternalError :: AgentClient -> ConnId -> String -> AM ()
@@ -334,7 +334,7 @@ closeNtfSupervisor ns = do
  stopWorkers $ ntfWorkers ns
  stopWorkers $ ntfSMPWorkers ns
  where
-    stopWorkers workers = atomically (swapTVar workers M.empty) >>= mapM_ (liftIO . cancelWorker)
+    stopWorkers workers = atomically' (swapTVar workers M.empty) >>= mapM_ (liftIO . cancelWorker)

 getNtfServer :: AgentClient -> AM' (Maybe NtfServer)
 getNtfServer c = do
@@ -18,7 +18,7 @@ import Control.Concurrent (forkIO)
 import Control.Monad (void)
 import Control.Monad.IO.Class (MonadIO, liftIO)
 import Data.Int (Int64)
-import Simplex.Messaging.Util (threadDelay', whenM)
+import Simplex.Messaging.Util (threadDelay', whenM, atomically')
 import UnliftIO.STM

 data RetryInterval = RetryInterval
@@ -82,8 +82,8 @@ withRetryLock2 RetryInterval2 {riSlow, riFast} lock action =
          waiting <- newTVarIO True
          _ <- liftIO . forkIO $ do
            threadDelay' delay
-            atomically $ whenM (readTVar waiting) $ void $ tryPutTMVar lock ()
-          atomically $ do
+            atomically' $ whenM (readTVar waiting) $ void $ tryPutTMVar lock ()
+          atomically' $ do
            takeTMVar lock
            writeTVar waiting False

@@ -24,7 +24,7 @@ import Simplex.Messaging.Agent.Protocol
 import Simplex.Messaging.Agent.Store.SQLite (SQLiteStore)
 import Simplex.Messaging.Transport (ATransport (..), TProxy, Transport (..), simplexMQVersion)
 import Simplex.Messaging.Transport.Server (defaultTransportServerConfig, loadTLSServerParams, runTransportServer)
-import Simplex.Messaging.Util (bshow)
+import Simplex.Messaging.Util (bshow, atomically')
 import UnliftIO.Async (race_)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
@@ -76,7 +76,7 @@ receive h c@AgentClient {rcvQ, subQ} = forever $ do

 send :: Transport c => c -> AgentClient -> IO ()
 send h c@AgentClient {subQ} = forever $ do
-  t <- atomically $ readTBQueue subQ
+  t <- atomically' $ readTBQueue subQ
  tPut h t
  logClient c "<--" t

@@ -278,7 +278,7 @@ import Simplex.Messaging.Parsers (blobFieldParser, defaultJSON, dropPrefix, from
 import Simplex.Messaging.Protocol
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Transport.Client (TransportHost)
-import Simplex.Messaging.Util (bshow, catchAllErrors, eitherToMaybe, ifM, safeDecodeUtf8, ($>>=), (<$$>))
+import Simplex.Messaging.Util (bshow, catchAllErrors, eitherToMaybe, ifM, safeDecodeUtf8, ($>>=), (<$$>), atomically')
 import Simplex.Messaging.Version.Internal
 import System.Directory (copyFile, createDirectoryIfMissing, doesFileExist)
 import System.Exit (exitFailure)
@@ -420,11 +420,11 @@ openSQLiteStore st@SQLiteStore {dbClosed} key keepKey =
 openSQLiteStore_ :: SQLiteStore -> ScrubbedBytes -> Bool -> IO ()
 openSQLiteStore_ SQLiteStore {dbConnection, dbFilePath, dbKey, dbClosed} key keepKey =
  bracketOnError
-    (atomically $ takeTMVar dbConnection)
-    (atomically . tryPutTMVar dbConnection)
+    (atomically' $ takeTMVar dbConnection)
+    (atomically' . tryPutTMVar dbConnection)
    $ \DB.Connection {slow} -> do
      DB.Connection {conn} <- connectDB dbFilePath key
-      atomically $ do
+      atomically' $ do
        putTMVar dbConnection DB.Connection {conn, slow}
        writeTVar dbClosed False
        writeTVar dbKey $! storeKey key keepKey
@@ -1214,7 +1214,7 @@ setRatchetX3dhKeys db connId x3dhPrivKey1 x3dhPrivKey2 pqPrivKem =
    db
    [sql|
      UPDATE ratchets
-      SET x3dh_priv_key_1 = ?, x3dh_priv_key_2 = ?, x3dh_pub_key_1 = ?, x3dh_pub_key_2 = ?, pq_priv_kem = ? 
+      SET x3dh_priv_key_1 = ?, x3dh_priv_key_2 = ?, x3dh_pub_key_1 = ?, x3dh_pub_key_2 = ?, pq_priv_kem = ?
      WHERE conn_id = ?
    |]
    (x3dhPrivKey1, x3dhPrivKey2, C.publicKey x3dhPrivKey1, C.publicKey x3dhPrivKey2, pqPrivKem, connId)
@@ -2248,7 +2248,7 @@ createWithRandomId' gVar create = tryCreate 3
          | otherwise -> pure . Left . SEInternal $ bshow e

 randomId :: TVar ChaChaDRG -> Int -> IO ByteString
-randomId gVar n = atomically $ U.encode <$> C.randomBytes n gVar
+randomId gVar n = atomically' $ U.encode <$> C.randomBytes n gVar

 ntfSubAndSMPAction :: NtfSubAction -> (Maybe NtfSubNTFAction, Maybe NtfSubSMPAction)
 ntfSubAndSMPAction (NtfSubNTFAction action) = (Just action, Nothing)
@@ -21,7 +21,7 @@ import Data.Time.Clock (diffUTCTime, getCurrentTime)
 import Database.SQLite.Simple (SQLError)
 import qualified Database.SQLite.Simple as SQL
 import qualified Simplex.Messaging.Agent.Store.SQLite.DB as DB
-import Simplex.Messaging.Util (diffToMilliseconds)
+import Simplex.Messaging.Util (diffToMilliseconds, atomically')
 import UnliftIO.Exception (bracket)
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
@@ -40,8 +40,8 @@ data SQLiteStore = SQLiteStore
 withConnection :: SQLiteStore -> (DB.Connection -> IO a) -> IO a
 withConnection SQLiteStore {dbConnection} =
  bracket
-    (atomically $ takeTMVar dbConnection)
-    (atomically . putTMVar dbConnection)
+    (atomically' $ takeTMVar dbConnection)
+    (atomically' . putTMVar dbConnection)

 withConnection' :: SQLiteStore -> (SQL.Connection -> IO a) -> IO a
 withConnection' st action = withConnection st $ action . DB.conn
@@ -28,7 +28,7 @@ import qualified Database.SQLite.Simple as SQL
 import Simplex.Messaging.Parsers (defaultJSON)
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
-import Simplex.Messaging.Util (diffToMilliseconds)
+import Simplex.Messaging.Util (diffToMilliseconds, atomically')

 data Connection = Connection
  { conn :: SQL.Connection,
@@ -48,7 +48,7 @@ timeIt slow sql a = do
  r <- a
  t' <- getCurrentTime
  let diff = diffToMilliseconds $ diffUTCTime t' t
-  atomically $ when (diff > 5) $ TM.alter (updateQueryStats diff) sql slow
+  atomically' $ when (diff > 5) $ TM.alter (updateQueryStats diff) sql slow
  pure r
  where
    updateQueryStats :: Int64 -> Maybe SlowQueryStats -> Maybe SlowQueryStats
@@ -110,7 +110,7 @@ import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client (SocksProxy, TransportClientConfig (..), TransportHost (..), runTransportClient)
 import Simplex.Messaging.Transport.KeepAlive
 import Simplex.Messaging.Transport.WebSockets (WS)
-import Simplex.Messaging.Util (bshow, raceAny_, threadDelay')
+import Simplex.Messaging.Util (bshow, raceAny_, threadDelay', atomically')
 import Simplex.Messaging.Version
 import System.Timeout (timeout)

@@ -360,8 +360,8 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
      action <-
        async $
          runTransportClient tcConfig (Just username) useHost port' (Just $ keyHash srv) (client t c cVar)
-            `finally` atomically (tryPutTMVar cVar $ Left PCENetworkError)
-      c_ <- tcpConnectTimeout `timeout` atomically (takeTMVar cVar)
+            `finally` atomically' (tryPutTMVar cVar $ Left PCENetworkError)
+      c_ <- tcpConnectTimeout `timeout` atomically' (takeTMVar cVar)
      case c_ of
        Just (Right c') -> pure $ Right c' {action = Just action}
        Just (Left e) -> pure $ Left e
@@ -377,21 +377,21 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
    client _ c cVar h = do
      ks <- atomically $ C.generateKeyPair g
      runExceptT (protocolClientHandshake @v @err @msg h ks (keyHash srv) serverVRange) >>= \case
-        Left e -> atomically . putTMVar cVar . Left $ PCETransportError e
+        Left e -> atomically' . putTMVar cVar . Left $ PCETransportError e
        Right th@THandle {params} -> do
          sessionTs <- getCurrentTime
          let c' = ProtocolClient {action = Nothing, client_ = c, thParams = params, sessionTs}
-          atomically $ do
+          atomically' $ do
            writeTVar (connected c) True
            putTMVar cVar $ Right c'
          raceAny_ ([send c' th, process c', receive c' th] <> [ping c' | smpPingInterval > 0])
            `finally` disconnected c'

    send :: Transport c => ProtocolClient v err msg -> THandle v c -> IO ()
-    send ProtocolClient {client_ = PClient {sndQ}} h = forever $ atomically (readTBQueue sndQ) >>= tPutLog h
+    send ProtocolClient {client_ = PClient {sndQ}} h = forever $ atomically' (readTBQueue sndQ) >>= tPutLog h

    receive :: Transport c => ProtocolClient v err msg -> THandle v c -> IO ()
-    receive ProtocolClient {client_ = PClient {rcvQ}} h = forever $ tGet h >>= atomically . writeTBQueue rcvQ
+    receive ProtocolClient {client_ = PClient {rcvQ}} h = forever $ tGet h >>= atomically' . writeTBQueue rcvQ

    ping :: ProtocolClient v err msg -> IO ()
    ping c@ProtocolClient {client_ = PClient {pingErrorCount}} = do
@@ -405,7 +405,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
        maxCnt = smpPingCount networkConfig

    process :: ProtocolClient v err msg -> IO ()
-    process c = forever $ atomically (readTBQueue $ rcvQ $ client_ c) >>= mapM_ (processMsg c)
+    process c = forever $ atomically' (readTBQueue $ rcvQ $ client_ c) >>= mapM_ (processMsg c)

    processMsg :: ProtocolClient v err msg -> SignedTransmission err msg -> IO ()
    processMsg c@ProtocolClient {client_ = PClient {sentCommands}} (_, _, (corrId, entId, respOrErr)) =
@@ -414,7 +414,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
        else do
          atomically (TM.lookup corrId sentCommands) >>= \case
            Nothing -> sendMsg respOrErr
-            Just Request {entityId, responseVar} -> atomically $ do
+            Just Request {entityId, responseVar} -> atomically' $ do
              TM.delete corrId sentCommands
              putTMVar responseVar $ response entityId
      where
@@ -428,7 +428,7 @@ getProtocolClient g transportSession@(_, srv, _) cfg@ProtocolClientConfig {qSize
          | otherwise = Left . PCEUnexpectedResponse $ bshow respOrErr
        sendMsg :: Either err msg -> IO ()
        sendMsg = \case
-          Right msg -> atomically $ mapM_ (`writeTBQueue` serverTransmission c entId msg) msgQ
+          Right msg -> atomically' $ mapM_ (`writeTBQueue` serverTransmission c entId msg) msgQ
          Left e -> putStrLn $ "SMP client error: " <> show e

 proxyUsername :: TransportSession msg -> ByteString
@@ -525,7 +525,7 @@ processSUBResponse c (Response rId r) = case r of
  Left e -> pure $ Left e

 writeSMPMessage :: SMPClient -> RecipientId -> BrokerMsg -> IO ()
-writeSMPMessage c rId msg = atomically $ mapM_ (`writeTBQueue` serverTransmission c rId msg) (msgQ $ client_ c)
+writeSMPMessage c rId msg = atomically' $ mapM_ (`writeTBQueue` serverTransmission c rId msg) (msgQ $ client_ c)

 serverTransmission :: ProtocolClient v err msg -> RecipientId -> msg -> ServerTransmission v msg
 serverTransmission ProtocolClient {thParams = THandleParams {thVersion, sessionId}, client_ = PClient {transportSession}} entityId message =
@@ -702,7 +702,7 @@ sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, thParams = THand
      Left e -> pure . Left $ PCETransportError e
      Right t
        | B.length s > blockSize - 2 -> pure . Left $ PCETransportError TELargeMsg
-        | otherwise -> atomically (writeTBQueue sndQ s) >> response <$> getResponse c r
+        | otherwise -> atomically' (writeTBQueue sndQ s) >> response <$> getResponse c r
        where
          s
            | batch = tEncodeBatch1 t
@@ -712,17 +712,17 @@ sendProtocolCommand c@ProtocolClient {client_ = PClient {sndQ}, thParams = THand
 getResponse :: ProtocolClient v err msg -> Request err msg -> IO (Response err msg)
 getResponse ProtocolClient {client_ = PClient {tcpTimeout, pingErrorCount}} Request {entityId, responseVar} = do
  response <-
-    timeout tcpTimeout (atomically (takeTMVar responseVar)) >>= \case
+    timeout tcpTimeout (atomically' (takeTMVar responseVar)) >>= \case
      Just r -> atomically (writeTVar pingErrorCount 0) $> r
      Nothing -> pure $ Left PCEResponseTimeout
  pure Response {entityId, response}

 mkTransmission :: forall v err msg. ProtocolEncoding v err (ProtoCommand msg) => ProtocolClient v err msg -> ClientCommand msg -> IO (PCTransmission err msg)
 mkTransmission ProtocolClient {thParams, client_ = PClient {clientCorrId, sentCommands}} (pKey_, entId, cmd) = do
-  corrId <- atomically getNextCorrId
+  corrId <- atomically' getNextCorrId
  let TransmissionForAuth {tForAuth, tToSend} = encodeTransmissionForAuth thParams (corrId, entId, cmd)
      auth = authTransmission (thAuth thParams) pKey_ corrId tForAuth
-  r <- atomically $ mkRequest corrId
+  r <- atomically' $ mkRequest corrId
  pure ((,tToSend) <$> auth, r)
  where
    getNextCorrId :: STM CorrId
@@ -43,7 +43,7 @@ import Simplex.Messaging.Protocol (BrokerMsg, NotifierId, NtfPrivateAuthKey, Pro
 import Simplex.Messaging.TMap (TMap)
 import qualified Simplex.Messaging.TMap as TM
 import Simplex.Messaging.Transport
-import Simplex.Messaging.Util (catchAll_, toChunks, ($>>=))
+import Simplex.Messaging.Util (catchAll_, toChunks, ($>>=), atomically')
 import System.Timeout (timeout)
 import UnliftIO (async)
 import UnliftIO.Exception (Exception)
@@ -154,7 +154,7 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ, randomDrg} sr
    waitForSMPClient :: SMPClientVar -> ExceptT SMPClientError IO SMPClient
    waitForSMPClient smpVar = do
      let ProtocolClientConfig {networkConfig = NetworkConfig {tcpConnectTimeout}} = smpCfg agentCfg
-      smpClient_ <- liftIO $ tcpConnectTimeout `timeout` atomically (readTMVar smpVar)
+      smpClient_ <- liftIO $ tcpConnectTimeout `timeout` atomically' (readTMVar smpVar)
      liftEither $ case smpClient_ of
        Just (Right smpClient) -> Right smpClient
        Just (Left e) -> Left e
@@ -168,12 +168,12 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ, randomDrg} sr
          tryE connectClient >>= \r -> case r of
            Right smp -> do
              logInfo . decodeUtf8 $ "Agent connected to " <> showServer srv
-              atomically $ putTMVar smpVar r
+              atomically' $ putTMVar smpVar r
              successAction smp
            Left e -> do
              if e == PCENetworkError || e == PCEResponseTimeout
                then retryAction
-                else atomically $ do
+                else atomically' $ do
                  putTMVar smpVar (Left e)
                  TM.delete srv smpClients
              throwE e
@@ -195,7 +195,7 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ, randomDrg} sr
      logInfo . decodeUtf8 $ "Agent disconnected from " <> showServer srv

    removeClientAndSubs :: IO (Maybe (Map SMPSub C.APrivateAuthKey))
-    removeClientAndSubs = atomically $ do
+    removeClientAndSubs = atomically' $ do
      TM.delete srv smpClients
      TM.lookupDelete srv (srvSubs ca) >>= mapM updateSubs
      where
@@ -229,9 +229,9 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ, randomDrg} sr
    reconnectClient = do
      withSMP ca srv $ \smp -> do
        liftIO $ notify $ CAReconnected srv
-        cs_ <- atomically $ mapM readTVar =<< TM.lookup srv (pendingSrvSubs ca)
+        cs_ <- atomically' $ mapM readTVar =<< TM.lookup srv (pendingSrvSubs ca)
        forM_ cs_ $ \cs -> do
-          subs' <- filterM (fmap not . atomically . hasSub (srvSubs ca) srv . fst) $ M.assocs cs
+          subs' <- filterM (fmap not . atomically' . hasSub (srvSubs ca) srv . fst) $ M.assocs cs
          let (nSubs, rSubs) = partition (isNotifier . fst . fst) subs'
          subscribe_ smp SPNotifier nSubs
          subscribe_ smp SPRecipient rSubs
@@ -252,9 +252,9 @@ getSMPServerClient' ca@SMPClientAgent {agentCfg, smpClients, msgQ, randomDrg} sr
                    map (\(sub, r) -> bimap (fst sub,) (const sub) r) $ L.toList rs'
                  (errs, oks) = partitionEithers rs''
                  (tempErrs, finalErrs) = partition (temporaryClientError . snd) errs
-              mapM_ (atomically . addSubscription ca srv) oks
+              mapM_ (atomically' . addSubscription ca srv) oks
              mapM_ (liftIO . notify . CAResubscribed srv) $ L.nonEmpty $ map fst oks
-              mapM_ (atomically . removePendingSubscription ca srv . fst) finalErrs
+              mapM_ (atomically' . removePendingSubscription ca srv . fst) finalErrs
              mapM_ (liftIO . notify . CASubError srv) $ L.nonEmpty finalErrs
              mapM_ (throwE . snd) $ listToMaybe tempErrs

@@ -271,7 +271,7 @@ closeSMPServerClients :: SMPClientAgent -> IO ()
 closeSMPServerClients c = readTVarIO (smpClients c) >>= mapM_ (forkIO . closeClient)
  where
    closeClient smpVar =
-      atomically (readTMVar smpVar) >>= \case
+      atomically' (readTMVar smpVar) >>= \case
        Right smp -> closeProtocolClient smp `catchAll_` pure ()
        _ -> pure ()

@@ -288,15 +288,15 @@ withSMP ca srv action = (getSMPServerClient' ca srv >>= action) `catchE` logSMPE

 subscribeQueue :: SMPClientAgent -> SMPServer -> (SMPSub, C.APrivateAuthKey) -> ExceptT SMPClientError IO ()
 subscribeQueue ca srv sub = do
-  atomically $ addPendingSubscription ca srv sub
+  atomically' $ addPendingSubscription ca srv sub
  withSMP ca srv $ \smp -> subscribe_ smp `catchE` handleErr
  where
    subscribe_ smp = do
      smpSubscribe smp sub
-      atomically $ addSubscription ca srv sub
+      atomically' $ addSubscription ca srv sub

    handleErr e = do
-      atomically . when (e /= PCENetworkError && e /= PCEResponseTimeout) $
+      atomically' . when (e /= PCENetworkError && e /= PCEResponseTimeout) $
        removePendingSubscription ca srv (fst sub)
      throwE e

@@ -308,7 +308,7 @@ subscribeQueuesNtfs = subscribeQueues_ SPNotifier

 subscribeQueues_ :: SMPSubParty -> SMPClientAgent -> SMPServer -> NonEmpty (QueueId, C.APrivateAuthKey) -> IO (NonEmpty (QueueId, Either SMPClientError ()))
 subscribeQueues_ party ca srv subs = do
-  atomically $ forM_ subs $ addPendingSubscription ca srv . first (party,)
+  atomically' $ forM_ subs $ addPendingSubscription ca srv . first (party,)
  runExceptT (getSMPServerClient' ca srv) >>= \case
    Left e -> pure $ L.map ((,Left e) . fst) subs
    Right smp -> smpSubscribeQueues party ca smp srv subs
@@ -316,7 +316,7 @@ subscribeQueues_ party ca srv subs = do
 smpSubscribeQueues :: SMPSubParty -> SMPClientAgent -> SMPClient -> SMPServer -> NonEmpty (QueueId, C.APrivateAuthKey) -> IO (NonEmpty (QueueId, Either SMPClientError ()))
 smpSubscribeQueues party ca smp srv subs = do
  rs <- L.zip subs <$> subscribe smp (L.map swap subs)
-  atomically $ forM rs $ \(sub, r) ->
+  atomically' $ forM rs $ \(sub, r) ->
    (fst sub,) <$> case r of
      Right () -> do
        addSubscription ca srv $ first (party,) sub
@@ -150,7 +150,7 @@ currentE2EEncryptVersion = VersionE2E 2
 -- TODO v5.7 remove dependency of version range on whether PQ encryption is used
 supportedE2EEncryptVRange :: PQSupport -> VersionRangeE2E
 supportedE2EEncryptVRange pq =
-  mkVersionRange kdfX3DHE2EEncryptVersion $ case pq of 
+  mkVersionRange kdfX3DHE2EEncryptVersion $ case pq of
    PQSupportOn -> pqRatchetE2EEncryptVersion
    PQSupportOff -> currentE2EEncryptVersion

@@ -116,16 +116,16 @@ ntfServer cfg@NtfServerConfig {transports, transportConfig = tCfg} started = do
        withFile statsFilePath AppendMode $ \h -> liftIO $ do
          hSetBuffering h LineBuffering
          ts <- getCurrentTime
-          fromTime' <- atomically $ swapTVar fromTime ts
-          tknCreated' <- atomically $ swapTVar tknCreated 0
-          tknVerified' <- atomically $ swapTVar tknVerified 0
-          tknDeleted' <- atomically $ swapTVar tknDeleted 0
-          subCreated' <- atomically $ swapTVar subCreated 0
-          subDeleted' <- atomically $ swapTVar subDeleted 0
-          ntfReceived' <- atomically $ swapTVar ntfReceived 0
-          ntfDelivered' <- atomically $ swapTVar ntfDelivered 0
-          tkn <- atomically $ periodStatCounts activeTokens ts
-          sub <- atomically $ periodStatCounts activeSubs ts
+          fromTime' <- atomically' $ swapTVar fromTime ts
+          tknCreated' <- atomically' $ swapTVar tknCreated 0
+          tknVerified' <- atomically' $ swapTVar tknVerified 0
+          tknDeleted' <- atomically' $ swapTVar tknDeleted 0
+          subCreated' <- atomically' $ swapTVar subCreated 0
+          subDeleted' <- atomically' $ swapTVar subDeleted 0
+          ntfReceived' <- atomically' $ swapTVar ntfReceived 0
+          ntfDelivered' <- atomically' $ swapTVar ntfDelivered 0
+          tkn <- atomically' $ periodStatCounts activeTokens ts
+          sub <- atomically' $ periodStatCounts activeSubs ts
          hPutStrLn h $
            intercalate
              ","
@@ -151,7 +151,7 @@ resubscribe NtfSubscriber {newSubQ} = do
  logInfo "Preparing SMP resubscriptions..."
  subs <- readTVarIO =<< asks (subscriptions . store)
  subs' <- filterM (fmap ntfShouldSubscribe . readTVarIO . subStatus) $ M.elems subs
-  atomically . writeTBQueue newSubQ $ map NtfSub subs'
+  atomically' . writeTBQueue newSubQ $ map NtfSub subs'
  logInfo $ "SMP resubscriptions queued (" <> tshow (length subs') <> " subscriptions)"

 ntfSubscriber :: NtfSubscriber -> M ()
@@ -160,14 +160,14 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
  where
    subscribe :: M ()
    subscribe = forever $ do
-      subs <- atomically (readTBQueue newSubQ)
+      subs <- atomically' (readTBQueue newSubQ)
      let ss = L.groupAllWith server subs
      batchSize <- asks $ subsBatchSize . config
      forM_ ss $ \serverSubs -> do
        let srv = server $ L.head serverSubs
            batches = toChunks batchSize $ L.toList serverSubs
        SMPSubscriber {newSubQ = subscriberSubQ} <- getSMPSubscriber srv
-        mapM_ (atomically . writeTQueue subscriberSubQ) batches
+        mapM_ (atomically' . writeTQueue subscriberSubQ) batches

    server :: NtfEntityRec 'Subscription -> SMPServer
    server (NtfSub sub) = ntfSubServer sub
@@ -186,14 +186,14 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
    runSMPSubscriber :: SMPSubscriber -> M ()
    runSMPSubscriber SMPSubscriber {newSubQ = subscriberSubQ} =
      forever $ do
-        subs <- atomically (peekTQueue subscriberSubQ)
+        subs <- atomically' (peekTQueue subscriberSubQ)
        let subs' = L.map (\(NtfSub sub) -> sub) subs
            srv = server $ L.head subs
        logSubStatus srv "subscribing" $ length subs
        mapM_ (\NtfSubData {smpQueue} -> updateSubStatus smpQueue NSPending) subs'
        rs <- liftIO $ subscribeQueues srv subs'
        (subs'', oks, errs) <- foldM process ([], 0, []) rs
-        atomically $ do
+        atomically' $ do
          void $ readTQueue subscriberSubQ
          mapM_ (writeTQueue subscriberSubQ . L.map NtfSub) $ L.nonEmpty subs''
        logSubStatus srv "retrying" $ length subs''
@@ -218,7 +218,7 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge

    receiveSMP :: M ()
    receiveSMP = forever $ do
-      ((_, srv, _), _, _, ntfId, msg) <- atomically $ readTBQueue msgQ
+      ((_, srv, _), _, _, ntfId, msg) <- atomically' $ readTBQueue msgQ
      let smpQueue = SMPQueueNtf srv ntfId
      case msg of
        SMP.NMSG nmsgNonce encNMsgMeta -> do
@@ -226,8 +226,8 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge
          st <- asks store
          NtfPushServer {pushQ} <- asks pushServer
          stats <- asks serverStats
-          atomically $ updatePeriodStats (activeSubs stats) ntfId
-          atomically $
+          atomically' $ updatePeriodStats (activeSubs stats) ntfId
+          atomically' $
            findNtfSubscriptionToken st smpQueue
              >>= mapM_ (\tkn -> writeTBQueue pushQ (tkn, PNMessage PNMessageData {smpQueue, ntfTs, nmsgNonce, encNMsgMeta}))
          incNtfStat ntfReceived
@@ -236,7 +236,7 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge

    receiveAgent =
      forever $
-        atomically (readTBQueue agentQ) >>= \case
+        atomically' (readTBQueue agentQ) >>= \case
          CAConnected _ -> pure ()
          CADisconnected srv subs -> do
            logSubStatus srv "disconnected" $ length subs
@@ -280,7 +280,7 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge

    updateSubStatus smpQueue status = do
      st <- asks store
-      atomically (findNtfSubscription st smpQueue) >>= mapM_ update
+      atomically' (findNtfSubscription st smpQueue) >>= mapM_ update
      where
        update NtfSubData {ntfSubId, subStatus} = do
          old <- atomically $ stateTVar subStatus (,status)
@@ -288,7 +288,7 @@ ntfSubscriber NtfSubscriber {smpSubscribers, newSubQ, smpAgent = ca@SMPClientAge

 ntfPush :: NtfPushServer -> M ()
 ntfPush s@NtfPushServer {pushQ} = forever $ do
-  (tkn@NtfTknData {ntfTknId, token = DeviceToken pp _, tknStatus}, ntf) <- atomically (readTBQueue pushQ)
+  (tkn@NtfTknData {ntfTknId, token = DeviceToken pp _, tknStatus}, ntf) <- atomically' (readTBQueue pushQ)
  liftIO $ logDebug $ "sending push notification to " <> T.pack (show pp)
  status <- readTVarIO tknStatus
  case ntf of
@@ -307,7 +307,7 @@ ntfPush s@NtfPushServer {pushQ} = forever $ do
      void $ deliverNotification pp tkn ntf
    PNMessage {} -> checkActiveTkn status $ do
      stats <- asks serverStats
-      atomically $ updatePeriodStats (activeTokens stats) ntfTknId
+      atomically' $ updatePeriodStats (activeTokens stats) ntfTknId
      void $ deliverNotification pp tkn ntf
      incNtfStat ntfDelivered
  where
@@ -343,7 +343,7 @@ runNtfClientTransport :: Transport c => THandleNTF c -> M ()
 runNtfClientTransport th@THandle {params} = do
  qSize <- asks $ clientQSize . config
  ts <- liftIO getSystemTime
-  c <- atomically $ newNtfServerClient qSize params ts
+  c <- atomically' $ newNtfServerClient qSize params ts
  s <- asks subscriber
  ps <- asks pushServer
  expCfg <- asks $ inactiveClientExpiration . config
@@ -373,7 +373,7 @@ receive th@THandle {params = THandleParams {thAuth}} NtfServerClient {rcvQ, sndQ

 send :: Transport c => THandleNTF c -> NtfServerClient -> IO ()
 send h@THandle {params} NtfServerClient {sndQ, sndActiveAt} = forever $ do
-  t <- atomically $ readTBQueue sndQ
+  t <- atomically' $ readTBQueue sndQ
  void . liftIO $ tPut h [Right (Nothing, encodeTransmission params t)]
  atomically . writeTVar sndActiveAt =<< liftIO getSystemTime

@@ -387,7 +387,7 @@ verifyNtfTransmission auth_ (tAuth, authorized, (corrId, entId, _)) cmd = do
  st <- asks store
  case cmd of
    NtfCmd SToken c@(TNEW tkn@(NewNtfTkn _ k _)) -> do
-      r_ <- atomically $ getNtfTokenRegistration st tkn
+      r_ <- atomically' $ getNtfTokenRegistration st tkn
      pure $
        if verifyCmdAuthorization auth_ tAuth authorized k
          then case r_ of
@@ -397,26 +397,26 @@ verifyNtfTransmission auth_ (tAuth, authorized, (corrId, entId, _)) cmd = do
            _ -> VRVerified (NtfReqNew corrId (ANE SToken tkn))
          else VRFailed
    NtfCmd SToken c -> do
-      t_ <- atomically $ getNtfToken st entId
+      t_ <- atomically' $ getNtfToken st entId
      verifyToken t_ (`verifiedTknCmd` c)
    NtfCmd SSubscription c@(SNEW sub@(NewNtfSub tknId smpQueue _)) -> do
-      s_ <- atomically $ findNtfSubscription st smpQueue
+      s_ <- atomically' $ findNtfSubscription st smpQueue
      case s_ of
        Nothing -> do
-          t_ <- atomically $ getActiveNtfToken st tknId
+          t_ <- atomically' $ getActiveNtfToken st tknId
          verifyToken' t_ $ VRVerified (NtfReqNew corrId (ANE SSubscription sub))
        Just s@NtfSubData {tokenId = subTknId} ->
          if subTknId == tknId
            then do
-              t_ <- atomically $ getActiveNtfToken st subTknId
+              t_ <- atomically' $ getActiveNtfToken st subTknId
              verifyToken' t_ $ verifiedSubCmd s c
            else pure $ maybe False (dummyVerifyCmd auth_ authorized) tAuth `seq` VRFailed
    NtfCmd SSubscription PING -> pure $ VRVerified $ NtfReqPing corrId entId
    NtfCmd SSubscription c -> do
-      s_ <- atomically $ getNtfSubscription st entId
+      s_ <- atomically' $ getNtfSubscription st entId
      case s_ of
        Just s@NtfSubData {tokenId = subTknId} -> do
-          t_ <- atomically $ getActiveNtfToken st subTknId
+          t_ <- atomically' $ getActiveNtfToken st subTknId
          verifyToken' t_ $ verifiedSubCmd s c
        _ -> pure $ maybe False (dummyVerifyCmd auth_ authorized) tAuth `seq` VRFailed
  where
@@ -436,9 +436,9 @@ verifyNtfTransmission auth_ (tAuth, authorized, (corrId, entId, _)) cmd = do
 client :: NtfServerClient -> NtfSubscriber -> NtfPushServer -> M ()
 client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPushServer {pushQ, intervalNotifiers} =
  forever $
-    atomically (readTBQueue rcvQ)
+    atomically' (readTBQueue rcvQ)
      >>= processCommand
-      >>= atomically . writeTBQueue sndQ
+      >>= atomically' . writeTBQueue sndQ
  where
    processCommand :: NtfRequest -> M (Transmission NtfResponse)
    processCommand = \case
@@ -449,8 +449,8 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
        let dhSecret = C.dh' dhPubKey srvDhPrivKey
        tknId <- getId
        regCode <- getRegCode
-        tkn <- atomically $ mkNtfTknData tknId newTkn ks dhSecret regCode
-        atomically $ addNtfToken st tknId tkn
+        tkn <- atomically' $ mkNtfTknData tknId newTkn ks dhSecret regCode
+        atomically' $ addNtfToken st tknId tkn
        atomically $ writeTBQueue pushQ (tkn, PNVerification regCode)
        withNtfLog (`logCreateToken` tkn)
        incNtfStatT token tknCreated
@@ -472,7 +472,7 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
                logDebug "TVFY - token verified"
                st <- asks store
                updateTknStatus tkn NTActive
-                tIds <- atomically $ removeInactiveTokenRegistrations st tkn
+                tIds <- atomically' $ removeInactiveTokenRegistrations st tkn
                forM_ tIds cancelInvervalNotifications
                incNtfStatT token tknVerified
                pure NROk
@@ -486,7 +486,7 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
            logDebug "TRPL - replace token"
            st <- asks store
            regCode <- getRegCode
-            atomically $ do
+            atomically' $ do
              removeTokenRegistration st tkn
              writeTVar tknStatus NTRegistered
              let tkn' = tkn {token = token', tknRegCode = regCode}
@@ -499,9 +499,9 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
          TDEL -> do
            logDebug "TDEL"
            st <- asks store
-            qs <- atomically $ deleteNtfToken st tknId
+            qs <- atomically' $ deleteNtfToken st tknId
            forM_ qs $ \SMPQueueNtf {smpServer, notifierId} ->
-              atomically $ removeSubscription ca smpServer (SPNotifier, notifierId)
+              atomically' $ removeSubscription ca smpServer (SPNotifier, notifierId)
            cancelInvervalNotifications tknId
            withNtfLog (`logDeleteToken` tknId)
            incNtfStatT token tknDeleted
@@ -538,10 +538,10 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
        logDebug "SNEW - new subscription"
        st <- asks store
        subId <- getId
-        sub <- atomically $ mkNtfSubData subId newSub
+        sub <- atomically' $ mkNtfSubData subId newSub
        resp <-
-          atomically (addNtfSubscription st subId sub) >>= \case
-            Just _ -> atomically (writeTBQueue newSubQ [NtfSub sub]) $> NRSubId subId
+          atomically' (addNtfSubscription st subId sub) >>= \case
+            Just _ -> atomically' (writeTBQueue newSubQ [NtfSub sub]) $> NRSubId subId
            _ -> pure $ NRErr AUTH
        withNtfLog (`logCreateSubscription` sub)
        incNtfStat subCreated
@@ -562,8 +562,8 @@ client NtfServerClient {rcvQ, sndQ} NtfSubscriber {newSubQ, smpAgent = ca} NtfPu
          SDEL -> do
            logDebug "SDEL"
            st <- asks store
-            atomically $ deleteNtfSubscription st subId
-            atomically $ removeSubscription ca smpServer (SPNotifier, notifierId)
+            atomically' $ deleteNtfSubscription st subId
+            atomically' $ removeSubscription ca smpServer (SPNotifier, notifierId)
            withNtfLog (`logDeleteSubscription` subId)
            incNtfStat subDeleted
            pure NROk
@@ -595,7 +595,7 @@ incNtfStat statSel = do
 saveServerStats :: M ()
 saveServerStats =
  asks (serverStatsBackupFile . config)
-    >>= mapM_ (\f -> asks serverStats >>= atomically . getNtfServerStatsData >>= liftIO . saveStats f)
+    >>= mapM_ (\f -> asks serverStats >>= atomically' . getNtfServerStatsData >>= liftIO . saveStats f)
  where
    saveStats f stats = do
      logInfo $ "saving server stats to file " <> T.pack f
@@ -610,7 +610,7 @@ restoreServerStats = asks (serverStatsBackupFile . config) >>= mapM_ restoreStat
      liftIO (strDecode <$> B.readFile f) >>= \case
        Right d -> do
          s <- asks serverStats
-          atomically $ setNtfServerStats s d
+          atomically' $ setNtfServerStats s d
          renameFile f $ f <> ".bak"
          logInfo "server stats restored"
        Left e -> do
@@ -37,7 +37,7 @@ import Simplex.Messaging.Notifications.Protocol
 import Simplex.Messaging.Notifications.Server.Store
 import Simplex.Messaging.Protocol (NtfPrivateAuthKey)
 import Simplex.Messaging.Server.StoreLog
-import Simplex.Messaging.Util (safeDecodeUtf8, whenM)
+import Simplex.Messaging.Util (safeDecodeUtf8, whenM, atomically')
 import System.Directory (doesFileExist, renameFile)
 import System.IO

@@ -196,7 +196,7 @@ readNtfStore f st = mapM_ (addNtfLogRecord . LB.toStrict) . LB.lines =<< LB.read
  where
    addNtfLogRecord s = case strDecode s of
      Left e -> logError $ "Log parsing error (" <> T.pack e <> "): " <> safeDecodeUtf8 (B.take 100 s)
-      Right lr -> atomically $ case lr of
+      Right lr -> atomically' $ case lr of
        CreateToken r@NtfTknRec {ntfTknId} -> do
          tkn <- mkTknData r
          addNtfToken st ntfTknId tkn
@@ -155,7 +155,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
    serverThread s label subQ subs clientSubs unsub = do
      labelMyThread label
      forever $
-        atomically updateSubscribers
+        atomically' updateSubscribers
          $>>= endPreviousSubscriptions
          >>= liftIO . mapM_ unsub
      where
@@ -176,7 +176,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
            labelMyThread $ label <> ".endPreviousSubscriptions"
            atomically $ writeTBQueue (sndQ c) [(CorrId "", qId, END)]
            atomically $ modifyTVar' (endThreads c) $ IM.delete tId
-          mkWeakThreadId t >>= atomically . modifyTVar' (endThreads c) . IM.insert tId
+          mkWeakThreadId t >>= atomically' . modifyTVar' (endThreads c) . IM.insert tId
          atomically $ TM.lookupDelete qId (clientSubs c)

    expireMessagesThread_ :: ServerConfig -> [M ()]
@@ -195,8 +195,8 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
        old <- liftIO $ expireBeforeEpoch expCfg
        rIds <- M.keysSet <$> readTVarIO ms
        forM_ rIds $ \rId -> do
-          q <- atomically (getMsgQueue ms rId quota)
-          deleted <- atomically $ deleteExpiredMsgs q old
+          q <- atomically' (getMsgQueue ms rId quota)
+          deleted <- atomically' $ deleteExpiredMsgs q old
          atomically $ modifyTVar' (msgExpired stats) (+ deleted)

    serverStatsThread_ :: ServerConfig -> [M ()]
@@ -216,19 +216,19 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
        withFile statsFilePath AppendMode $ \h -> liftIO $ do
          hSetBuffering h LineBuffering
          ts <- getCurrentTime
-          fromTime' <- atomically $ swapTVar fromTime ts
-          qCreated' <- atomically $ swapTVar qCreated 0
-          qSecured' <- atomically $ swapTVar qSecured 0
-          qDeletedAll' <- atomically $ swapTVar qDeletedAll 0
-          qDeletedNew' <- atomically $ swapTVar qDeletedNew 0
-          qDeletedSecured' <- atomically $ swapTVar qDeletedSecured 0
-          msgSent' <- atomically $ swapTVar msgSent 0
-          msgRecv' <- atomically $ swapTVar msgRecv 0
-          msgExpired' <- atomically $ swapTVar msgExpired 0
-          ps <- atomically $ periodStatCounts activeQueues ts
-          msgSentNtf' <- atomically $ swapTVar msgSentNtf 0
-          msgRecvNtf' <- atomically $ swapTVar msgRecvNtf 0
-          psNtf <- atomically $ periodStatCounts activeQueuesNtf ts
+          fromTime' <- atomically' $ swapTVar fromTime ts
+          qCreated' <- atomically' $ swapTVar qCreated 0
+          qSecured' <- atomically' $ swapTVar qSecured 0
+          qDeletedAll' <- atomically' $ swapTVar qDeletedAll 0
+          qDeletedNew' <- atomically' $ swapTVar qDeletedNew 0
+          qDeletedSecured' <- atomically' $ swapTVar qDeletedSecured 0
+          msgSent' <- atomically' $ swapTVar msgSent 0
+          msgRecv' <- atomically' $ swapTVar msgRecv 0
+          msgExpired' <- atomically' $ swapTVar msgExpired 0
+          ps <- atomically' $ periodStatCounts activeQueues ts
+          msgSentNtf' <- atomically' $ swapTVar msgSentNtf 0
+          msgRecvNtf' <- atomically' $ swapTVar msgRecvNtf 0
+          psNtf <- atomically' $ periodStatCounts activeQueuesNtf ts
          qCount' <- readTVarIO qCount
          msgCount' <- readTVarIO msgCount
          hPutStrLn h $
@@ -354,7 +354,7 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
 #endif
              CPSockets -> withAdminRole $ do
                (accepted', closed', active') <- unliftIO u $ asks sockets
-                (accepted, closed, active) <- atomically $ (,,) <$> readTVar accepted' <*> readTVar closed' <*> readTVar active'
+                (accepted, closed, active) <- atomically' $ (,,) <$> readTVar accepted' <*> readTVar closed' <*> readTVar active'
                hPutStrLn h "Sockets: "
                hPutStrLn h $ "accepted: " <> show accepted
                hPutStrLn h $ "closed: " <> show closed
@@ -377,10 +377,10 @@ smpServer started cfg@ServerConfig {transports, transportConfig = tCfg} = do
              CPDelete queueId' -> withUserRole $ unliftIO u $ do
                st <- asks queueStore
                ms <- asks msgStore
-                queueId <- atomically (getQueue st SSender queueId') >>= \case
+                queueId <- atomically' (getQueue st SSender queueId') >>= \case
                  Left _ -> pure queueId' -- fallback to using as recipientId directly
                  Right QueueRec {recipientId} -> pure recipientId
-                r <- atomically $
+                r <- atomically' $
                  deleteQueue st queueId $>>= \q ->
                    Right . (q,) <$> delMsgQueueSize ms queueId
                case r of
@@ -415,7 +415,7 @@ runClientTransport th@THandle {params = THandleParams {thVersion, sessionId}} =
  ts <- liftIO getSystemTime
  active <- asks clients
  nextClientId <- asks clientSeq
-  c <- atomically $ do
+  c <- atomically' $ do
    new@Client {clientId} <- newClient nextClientId q thVersion sessionId ts
    modifyTVar' active $ IM.insert clientId new
    pure new
@@ -427,20 +427,20 @@ runClientTransport th@THandle {params = THandleParams {thVersion, sessionId}} =
  where
    disconnectThread_ c (Just expCfg) = [liftIO $ disconnectTransport th (rcvActiveAt c) (sndActiveAt c) expCfg (noSubscriptions c)]
    disconnectThread_ _ _ = []
-    noSubscriptions c = atomically $ (&&) <$> TM.null (subscriptions c) <*> TM.null (ntfSubscriptions c)
+    noSubscriptions c = atomically' $ (&&) <$> TM.null (subscriptions c) <*> TM.null (ntfSubscriptions c)

 clientDisconnected :: Client -> M ()
 clientDisconnected c@Client {clientId, subscriptions, connected, sessionId, endThreads} = do
  labelMyThread . B.unpack $ "client $" <> encode sessionId <> " disc"
-  subs <- atomically $ do
+  subs <- atomically' $ do
    writeTVar connected False
    swapTVar subscriptions M.empty
  liftIO $ mapM_ cancelSub subs
  srvSubs <- asks $ subscribers . server
  atomically $ modifyTVar' srvSubs $ \cs ->
    M.foldrWithKey (\sub _ -> M.update deleteCurrentClient sub) cs subs
-  asks clients >>= atomically . (`modifyTVar'` IM.delete clientId)
-  tIds <- atomically $ swapTVar endThreads IM.empty
+  asks clients >>= atomically' . (`modifyTVar'` IM.delete clientId)
+  tIds <- atomically' $ swapTVar endThreads IM.empty
  liftIO $ mapM_ (mapM_ killThread <=< deRefWeak) tIds
  where
    deleteCurrentClient :: Client -> Maybe Client
@@ -476,13 +476,13 @@ receive th@THandle {params = THandleParams {thAuth}} Client {rcvQ, sndQ, rcvActi
            verified = \case
              VRVerified qr -> Right (qr, (corrId, queueId, cmd))
              VRFailed -> Left (corrId, queueId, ERR AUTH)
-    write q = mapM_ (atomically . writeTBQueue q) . L.nonEmpty
+    write q = mapM_ (atomically' . writeTBQueue q) . L.nonEmpty

 send :: Transport c => THandleSMP c -> Client -> IO ()
 send h@THandle {params} Client {sndQ, sessionId, sndActiveAt} = do
  labelMyThread . B.unpack $ "client $" <> encode sessionId <> " send"
  forever $ do
-    ts <- atomically $ L.sortWith tOrder <$> readTBQueue sndQ
+    ts <- atomically' $ L.sortWith tOrder <$> readTBQueue sndQ
    -- TODO we can authorize responses as well
    void . liftIO . tPut h $ L.map (\t -> Right (Nothing, encodeTransmission params t)) ts
    atomically . writeTVar sndActiveAt =<< liftIO getSystemTime
@@ -534,7 +534,7 @@ verifyTransmission auth_ tAuth authorized queueId cmd =
    get :: SParty p -> M (Either ErrorType QueueRec)
    get party = do
      st <- asks queueStore
-      atomically $ getQueue st party queueId
+      atomically' $ getQueue st party queueId

 verifyCmdAuthorization :: Maybe (THandleAuth, C.CbNonce) -> Maybe TransmissionAuth -> ByteString -> C.APublicAuthKey -> Bool
 verifyCmdAuthorization auth_ tAuth authorized key = maybe False (verify key) tAuth
@@ -585,9 +585,9 @@ client :: Client -> Server -> M ()
 client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Server {subscribedQ, ntfSubscribedQ, notifiers} = do
  labelMyThread . B.unpack $ "client $" <> encode sessionId <> " commands"
  forever $
-    atomically (readTBQueue rcvQ)
+    atomically' (readTBQueue rcvQ)
      >>= mapM processCommand
-      >>= atomically . writeTBQueue sndQ
+      >>= atomically' . writeTBQueue sndQ
  where
    processCommand :: (Maybe QueueRec, Transmission Cmd) -> M (Transmission BrokerMsg)
    processCommand (qr_, (corrId, queueId, cmd)) = do
@@ -642,7 +642,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
              ids@(rId, _) <- getIds
              -- create QueueRec record with these ids and keys
              let qr = qRec ids
-              atomically (addQueue st qr) >>= \case
+              atomically' (addQueue st qr) >>= \case
                Left DUPLICATE_ -> addQueueRetry (n - 1) qik qRec
                Left e -> pure $ ERR e
                Right _ -> do
@@ -657,7 +657,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv

            logCreateById :: StoreLog 'WriteMode -> RecipientId -> IO ()
            logCreateById s rId =
-              atomically (getQueue st SRecipient rId) >>= \case
+              atomically' (getQueue st SRecipient rId) >>= \case
                Right q -> logCreateQueue s q
                _ -> pure ()

@@ -671,7 +671,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
          withLog $ \s -> logSecureQueue s queueId sKey
          stats <- asks serverStats
          atomically $ modifyTVar' (qSecured stats) (+ 1)
-          atomically $ (corrId,queueId,) . either ERR (const OK) <$> secureQueue st queueId sKey
+          atomically' $ (corrId,queueId,) . either ERR (const OK) <$> secureQueue st queueId sKey

        addQueueNotifier_ :: QueueStore -> NtfPublicAuthKey -> RcvNtfPublicDhKey -> M (Transmission BrokerMsg)
        addQueueNotifier_ st notifierKey dhKey = time "NKEY" $ do
@@ -684,7 +684,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
            addNotifierRetry n rcvPublicDhKey rcvNtfDhSecret = do
              notifierId <- randomId =<< asks (queueIdBytes . config)
              let ntfCreds = NtfCreds {notifierId, notifierKey, rcvNtfDhSecret}
-              atomically (addQueueNotifier st queueId ntfCreds) >>= \case
+              atomically' (addQueueNotifier st queueId ntfCreds) >>= \case
                Left DUPLICATE_ -> addNotifierRetry (n - 1) rcvPublicDhKey rcvNtfDhSecret
                Left e -> pure $ ERR e
                Right _ -> do
@@ -694,12 +694,12 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
        deleteQueueNotifier_ :: QueueStore -> M (Transmission BrokerMsg)
        deleteQueueNotifier_ st = do
          withLog (`logDeleteNotifier` queueId)
-          okResp <$> atomically (deleteQueueNotifier st queueId)
+          okResp <$> atomically' (deleteQueueNotifier st queueId)

        suspendQueue_ :: QueueStore -> M (Transmission BrokerMsg)
        suspendQueue_ st = do
          withLog (`logSuspendQueue` queueId)
-          okResp <$> atomically (suspendQueue st queueId)
+          okResp <$> atomically' (suspendQueue st queueId)

        subscribeQueue :: QueueRec -> RecipientId -> M (Transmission BrokerMsg)
        subscribeQueue qr rId = do
@@ -712,10 +712,10 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
                  -- cannot use SUB in the same connection where GET was used
                  pure (corrId, rId, ERR $ CMD PROHIBITED)
                s ->
-                  atomically (tryTakeTMVar $ delivered s) >> deliver sub
+                  atomically' (tryTakeTMVar $ delivered s) >> deliver sub
          where
            newSub :: M (TVar Sub)
-            newSub = time "SUB newSub" . atomically $ do
+            newSub = time "SUB newSub" . atomically' $ do
              writeTQueue subscribedQ (rId, clnt)
              sub <- newTVar =<< newSubscription NoSub
              TM.insert rId sub subscriptions
@@ -723,7 +723,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
            deliver :: TVar Sub -> M (Transmission BrokerMsg)
            deliver sub = do
              q <- getStoreMsgQueue "SUB" rId
-              msg_ <- atomically $ tryPeekMsg q
+              msg_ <- atomically' $ tryPeekMsg q
              deliverMessage "SUB" qr rId sub q msg_

        getMessage :: QueueRec -> M (Transmission BrokerMsg)
@@ -734,7 +734,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
            Just sub ->
              readTVarIO sub >>= \case
                s@Sub {subThread = ProhibitSub} ->
-                  atomically (tryTakeTMVar $ delivered s)
+                  atomically' (tryTakeTMVar $ delivered s)
                    >> getMessage_ s
                -- cannot use GET in the same connection where there is an active subscription
                _ -> pure (corrId, queueId, ERR $ CMD PROHIBITED)
@@ -748,7 +748,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
            getMessage_ :: Sub -> M (Transmission BrokerMsg)
            getMessage_ s = do
              q <- getStoreMsgQueue "GET" queueId
-              atomically $
+              atomically' $
                tryPeekMsg q >>= \case
                  Just msg ->
                    let encMsg = encryptMsg qr msg
@@ -759,7 +759,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
        withQueue action = maybe (pure $ err AUTH) action qr_

        subscribeNotifications :: M (Transmission BrokerMsg)
-        subscribeNotifications = time "NSUB" . atomically $ do
+        subscribeNotifications = time "NSUB" . atomically' $ do
          unlessM (TM.member queueId ntfSubscriptions) $ do
            writeTQueue ntfSubscribedQ (queueId, clnt)
            TM.insert queueId () ntfSubscriptions
@@ -770,16 +770,16 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
          atomically (TM.lookup queueId subscriptions) >>= \case
            Nothing -> pure $ err NO_MSG
            Just sub ->
-              atomically (getDelivered sub) >>= \case
+              atomically' (getDelivered sub) >>= \case
                Just s -> do
                  q <- getStoreMsgQueue "ACK" queueId
                  case s of
                    Sub {subThread = ProhibitSub} -> do
-                      deletedMsg_ <- atomically $ tryDelMsg q msgId
+                      deletedMsg_ <- atomically' $ tryDelMsg q msgId
                      mapM_ updateStats deletedMsg_
                      pure ok
                    _ -> do
-                      (deletedMsg_, msg_) <- atomically $ tryDelPeekMsg q msgId
+                      (deletedMsg_, msg_) <- atomically' $ tryDelPeekMsg q msgId
                      mapM_ updateStats deletedMsg_
                      deliverMessage "ACK" qr queueId sub q msg_
                _ -> pure $ err NO_MSG
@@ -798,10 +798,10 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
                stats <- asks serverStats
                atomically $ modifyTVar' (msgRecv stats) (+ 1)
                atomically $ modifyTVar' (msgCount stats) (subtract 1)
-                atomically $ updatePeriodStats (activeQueues stats) queueId
+                atomically' $ updatePeriodStats (activeQueues stats) queueId
                when (notification msgFlags) $ do
                  atomically $ modifyTVar' (msgRecvNtf stats) (+ 1)
-                  atomically $ updatePeriodStats (activeQueuesNtf stats) queueId
+                  atomically' $ updatePeriodStats (activeQueuesNtf stats) queueId

        sendMessage :: QueueRec -> MsgFlags -> MsgBody -> M (Transmission BrokerMsg)
        sendMessage qr msgFlags msgBody
@@ -815,18 +815,18 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
                    msg_ <- time "SEND" $ do
                      q <- getStoreMsgQueue "SEND" $ recipientId qr
                      expireMessages q
-                      atomically . writeMsg q =<< mkMessage body
+                      atomically' . writeMsg q =<< mkMessage body
                    case msg_ of
                      Nothing -> pure $ err QUOTA
                      Just msg -> time "SEND ok" $ do
                        stats <- asks serverStats
                        when (notification msgFlags) $ do
-                          atomically . trySendNotification msg =<< asks random
+                          atomically' . trySendNotification msg =<< asks random
                          atomically $ modifyTVar' (msgSentNtf stats) (+ 1)
-                          atomically $ updatePeriodStats (activeQueuesNtf stats) (recipientId qr)
+                          atomically' $ updatePeriodStats (activeQueuesNtf stats) (recipientId qr)
                        atomically $ modifyTVar' (msgSent stats) (+ 1)
                        atomically $ modifyTVar' (msgCount stats) (+ 1)
-                        atomically $ updatePeriodStats (activeQueues stats) (recipientId qr)
+                        atomically' $ updatePeriodStats (activeQueues stats) (recipientId qr)
                        pure ok
          where
            mkMessage :: C.MaxLenBS MaxMessageLen -> M Message
@@ -840,7 +840,7 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
              msgExp <- asks $ messageExpiration . config
              old <- liftIO $ mapM expireBeforeEpoch msgExp
              stats <- asks serverStats
-              deleted <- atomically $ sum <$> mapM (deleteExpiredMsgs q) old
+              deleted <- atomically' $ sum <$> mapM (deleteExpiredMsgs q) old
              atomically $ modifyTVar' (msgExpired stats) (+ deleted)

            trySendNotification :: Message -> TVar ChaChaDRG -> STM ()
@@ -870,22 +870,22 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
              case msg_ of
                Just msg ->
                  let encMsg = encryptMsg qr msg
-                   in atomically (setDelivered s msg) $> (corrId, rId, MSG encMsg)
+                   in atomically' (setDelivered s msg) $> (corrId, rId, MSG encMsg)
                _ -> forkSub $> ok
            _ -> pure ok
          where
            forkSub :: M ()
            forkSub = do
-              atomically . modifyTVar' sub $ \s -> s {subThread = SubPending}
+              atomically' . modifyTVar' sub $ \s -> s {subThread = SubPending}
              t <- mkWeakThreadId =<< forkIO subscriber
-              atomically . modifyTVar' sub $ \case
+              atomically' . modifyTVar' sub $ \case
                s@Sub {subThread = SubPending} -> s {subThread = SubThread t}
                s -> s
              where
                subscriber = do
                  labelMyThread $ B.unpack ("client $" <> encode sessionId) <> " subscriber/" <> T.unpack name
-                  msg <- atomically $ peekMsg q
-                  time "subscriber" . atomically $ do
+                  msg <- atomically' $ peekMsg q
+                  time "subscriber" . atomically' $ do
                    let encMsg = encryptMsg qr msg
                    writeTBQueue sndQ [(CorrId "", rId, MSG encMsg)]
                    s <- readTVar sub
@@ -912,13 +912,13 @@ client clnt@Client {subscriptions, ntfSubscriptions, rcvQ, sndQ, sessionId} Serv
        getStoreMsgQueue name rId = time (name <> " getMsgQueue") $ do
          ms <- asks msgStore
          quota <- asks $ msgQueueQuota . config
-          atomically $ getMsgQueue ms rId quota
+          atomically' $ getMsgQueue ms rId quota

        delQueueAndMsgs :: QueueStore -> M (Transmission BrokerMsg)
        delQueueAndMsgs st = do
          withLog (`logDeleteQueue` queueId)
          ms <- asks msgStore
-          atomically (deleteQueue st queueId $>>= \q -> delMsgQueue ms queueId $> Right q) >>= \case
+          atomically' (deleteQueue st queueId $>>= \q -> delMsgQueue ms queueId $> Right q) >>= \case
            Right q -> updateDeletedStats q $> ok
            Left e -> pure $ err e

@@ -971,7 +971,7 @@ saveServerMessages keepMsgs = asks (storeMsgsFile . config) >>= mapM_ saveMessag
      where
        getMessages = if keepMsgs then snapshotMsgQueue else flushMsgQueue
        saveQueueMsgs ms h rId =
-          atomically (getMessages ms rId)
+          atomically' (getMessages ms rId)
            >>= mapM_ (B.hPutStrLn h . strEncode . MLRv3 rId)

 restoreServerMessages :: M Int
@@ -999,7 +999,7 @@ restoreServerMessages = asks (storeMsgsFile . config) >>= \case
          where
            s = LB.toStrict s'
            addToMsgQueue rId msg = do
-              (isExpired, logFull) <- atomically $ do
+              (isExpired, logFull) <- atomically' $ do
                q <- getMsgQueue ms rId quota
                case msg of
                  Message {msgTs}
@@ -1014,7 +1014,7 @@ restoreServerMessages = asks (storeMsgsFile . config) >>= \case
 saveServerStats :: M ()
 saveServerStats =
  asks (serverStatsBackupFile . config)
-    >>= mapM_ (\f -> asks serverStats >>= atomically . getServerStatsData >>= liftIO . saveStats f)
+    >>= mapM_ (\f -> asks serverStats >>= atomically' . getServerStatsData >>= liftIO . saveStats f)
  where
    saveStats f stats = do
      logInfo $ "saving server stats to file " <> T.pack f
@@ -1031,7 +1031,7 @@ restoreServerStats expiredWhileRestoring = asks (serverStatsBackupFile . config)
          s <- asks serverStats
          _qCount <- fmap M.size . readTVarIO . queues =<< asks queueStore
          _msgCount <- foldM (\(!n) q -> (n +) <$> readTVarIO (size q)) 0 =<< readTVarIO =<< asks msgStore
-          atomically $ setServerStats s d {_qCount, _msgCount, _msgExpired = _msgExpired d + expiredWhileRestoring}
+          atomically' $ setServerStats s d {_qCount, _msgCount, _msgExpired = _msgExpired d + expiredWhileRestoring}
          renameFile f $ f <> ".bak"
          logInfo "server stats restored"
          when (_qCount /= statsQCount) $ logWarn $ "Queue count differs: stats: " <> tshow statsQCount <> ", store: " <> tshow _qCount
@@ -11,6 +11,7 @@ import Data.ByteString.Char8 (ByteString)
 import qualified Data.ByteString.Char8 as B
 import GHC.IO.Exception (IOErrorType (..), IOException (..), ioException)
 import System.Timeout (timeout)
+import Simplex.Messaging.Util (atomically')

 data TBuffer = TBuffer
  { buffer :: TVar ByteString,
@@ -26,8 +27,8 @@ newTBuffer = do
 withBufferLock :: TBuffer -> IO a -> IO a
 withBufferLock TBuffer {getLock} =
  E.bracket_
-    (atomically $ takeTMVar getLock)
-    (atomically $ putTMVar getLock ())
+    (atomically' $ takeTMVar getLock)
+    (atomically' $ putTMVar getLock ())

 -- | Attempt to read some bytes, appending it to the existing buffer
 peekBuffered :: TBuffer -> Int -> IO ByteString -> IO (ByteString, Maybe ByteString)
@@ -50,7 +50,7 @@ import Simplex.Messaging.Encoding.String
 import Simplex.Messaging.Parsers (parseAll, parseString)
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.KeepAlive
-import Simplex.Messaging.Util (bshow, (<$?>), catchAll, tshow)
+import Simplex.Messaging.Util (bshow, (<$?>), catchAll, tshow, atomically')
 import System.IO.Error
 import Text.Read (readMaybe)
 import UnliftIO.Exception (IOException)
@@ -143,7 +143,7 @@ runTLSTransportClient tlsParams caStore_ cfg@TransportClientConfig {socksProxy,
    mapM_ (setSocketKeepAlive sock) tcpKeepAlive `catchAll` \e -> logError ("Error setting TCP keep-alive" <> tshow e)
    let tCfg = clientTransportConfig cfg
    connectTLS (Just hostName) tCfg clientParams sock >>= \tls -> do
-      chain <- atomically (tryTakeTMVar serverCert) >>= \case
+      chain <- atomically' (tryTakeTMVar serverCert) >>= \case
        Nothing -> do
          logError "onServerCertificate didn't fire or failed to get cert chain"
          closeTLS tls >> error "onServerCertificate failed"
@@ -234,7 +234,7 @@ mkTLSClientParams supported caStore_ host port cafp_ clientCreds_ alpn_ serverCe
    onServerCert _ _ _ c = do
      errs <- maybe def (\ca -> validateCertificateChain ca host p c) cafp_
      when (null errs) $
-        atomically (putTMVar serverCerts c)
+        atomically' (putTMVar serverCerts c)
      pure errs

 validateCertificateChain :: C.KeyHash -> HostName -> ByteString -> X.CertificateChain -> IO [XV.FailedReason]
@@ -23,6 +23,7 @@ import Data.X509.Validation (Fingerprint (..), getFingerprint)
 import qualified Network.TLS as TLS
 import qualified Simplex.Messaging.Crypto as C
 import qualified Time.System as Hourglass
+import Simplex.Messaging.Util (atomically')

 -- | Generate a certificate chain to be used with TLS fingerprint-pinning
 --
@@ -29,6 +29,7 @@ import Simplex.Messaging.Transport.HTTP2
 import UnliftIO.STM
 import UnliftIO.Timeout
 import qualified Data.X509 as X
+import Simplex.Messaging.Util (atomically')

 data HTTP2Client = HTTP2Client
  { action :: Maybe (Async HTTP2Response),
@@ -107,8 +108,8 @@ getVerifiedHTTP2ClientWith config host port disconnected setup =
    runClient :: HClient -> IO (Either HTTP2ClientError HTTP2Client)
    runClient c = do
      cVar <- newEmptyTMVarIO
-      action <- async $ setup (client c cVar) `E.finally` atomically (putTMVar cVar $ Left HCNetworkError)
-      c_ <- connTimeout config `timeout` atomically (takeTMVar cVar)
+      action <- async $ setup (client c cVar) `E.finally` atomically' (putTMVar cVar $ Left HCNetworkError)
+      c_ <- connTimeout config `timeout` atomically' (takeTMVar cVar)
      pure $ case c_ of
        Just (Right c') -> Right c' {action = Just action}
        Just (Left e) -> Left e
@@ -128,18 +129,18 @@ getVerifiedHTTP2ClientWith config host port disconnected setup =
                sessionId = tlsUniq tls,
                sessionALPN = tlsALPN tls
              }
-      atomically $ do
+      atomically' $ do
        writeTVar (connected c) True
        putTMVar cVar (Right c')
      process c' sendReq `E.finally` disconnected

    process :: HTTP2Client -> H.Client HTTP2Response
    process HTTP2Client {client_ = HClient {reqQ}} sendReq = forever $ do
-      (req, respVar) <- atomically $ readTBQueue reqQ
+      (req, respVar) <- atomically' $ readTBQueue reqQ
      sendReq req $ \r -> do
        respBody <- getHTTP2Body r (bodyHeadSize config)
        let resp = HTTP2Response {response = r, respBody}
-        atomically $ putTMVar respVar resp
+        atomically' $ putTMVar respVar resp
        pure resp

 -- | Disconnects client from the server and terminates client threads.
@@ -151,7 +152,7 @@ sendRequest HTTP2Client {client_ = HClient {config, reqQ}} req reqTimeout_ = do
  resp <- newEmptyTMVarIO
  atomically $ writeTBQueue reqQ (req, resp)
  let reqTimeout = http2RequestTimeout config reqTimeout_
-  maybe (Left HCResponseTimeout) Right <$> (reqTimeout `timeout` atomically (takeTMVar resp))
+  maybe (Left HCResponseTimeout) Right <$> (reqTimeout `timeout` atomically' (takeTMVar resp))

 -- | this function should not be used until HTTP2 is thread safe, use sendRequest
 sendRequestDirect :: HTTP2Client -> Request -> Maybe Int -> IO (Either HTTP2ClientError HTTP2Response)
@@ -16,7 +16,7 @@ import Simplex.Messaging.Server.Expiration
 import Simplex.Messaging.Transport (ALPN, SessionId, TLS, closeConnection, tlsALPN, tlsUniq)
 import Simplex.Messaging.Transport.HTTP2
 import Simplex.Messaging.Transport.Server (TransportServerConfig (..), loadSupportedTLSServerParams, runTransportServer)
-import Simplex.Messaging.Util (threadDelay')
+import Simplex.Messaging.Util (threadDelay', atomically')
 import UnliftIO (finally)
 import UnliftIO.Concurrent (forkIO, killThread)

@@ -58,7 +58,7 @@ getHTTP2Server HTTP2ServerConfig {qSize, http2Port, bufferSize, bodyHeadSize, se
    runHTTP2Server started http2Port bufferSize tlsServerParams transportConfig Nothing (const $ pure ()) $ \sessionId sessionALPN r sendResponse -> do
      reqBody <- getHTTP2Body r bodyHeadSize
      atomically $ writeTBQueue reqQ HTTP2Request {sessionId, sessionALPN, request = r, reqBody, sendResponse}
-  void . atomically $ takeTMVar started
+  void . atomically' $ takeTMVar started
  pure HTTP2Server {action, reqQ}

 closeHTTP2Server :: HTTP2Server -> IO ()
@@ -38,7 +38,7 @@ import qualified Data.X509.Validation as XV
 import Network.Socket
 import qualified Network.TLS as T
 import Simplex.Messaging.Transport
-import Simplex.Messaging.Util (catchAll_, labelMyThread, tshow)
+import Simplex.Messaging.Util (catchAll_, labelMyThread, tshow, atomically')
 import System.Exit (exitFailure)
 import System.Mem.Weak (Weak, deRefWeak)
 import UnliftIO (timeout)
@@ -114,7 +114,7 @@ runTCPServerSocket (accepted, gracefullyClosed, clients) started getSocket serve
    forever . E.bracketOnError (accept sock) (close . fst) $ \(conn, _peer) -> do
      cId <- atomically $ stateTVar accepted $ \cId -> let cId' = cId + 1 in cId `seq` (cId', cId')
      let closeConn _ = do
-            atomically $ modifyTVar' clients $ IM.delete cId 
+            atomically $ modifyTVar' clients $ IM.delete cId
            gracefulClose conn 5000 `catchAll_` pure () -- catchAll_ is needed here in case the connection was closed earlier
            atomically $ modifyTVar' gracefullyClosed (+1)
      tId <- mkWeakThreadId =<< server conn `forkFinally` closeConn
@@ -129,7 +129,7 @@ closeServer :: TMVar Bool -> TVar (IntMap (Weak ThreadId)) -> Socket -> IO ()
 closeServer started clients sock = do
  readTVarIO clients >>= mapM_ (deRefWeak >=> mapM_ killThread)
  close sock
-  void . atomically $ tryPutTMVar started False
+  void . atomically' $ tryPutTMVar started False

 startTCPServer :: TMVar Bool -> ServiceName -> IO Socket
 startTCPServer started port = withSocketsDo $ resolve >>= open >>= setStarted
@@ -148,7 +148,7 @@ startTCPServer started port = withSocketsDo $ resolve >>= open >>= setStarted
      bind sock $ addrAddress addr
      listen sock 1024
      pure sock
-    setStarted sock = atomically (tryPutTMVar started True) >> pure sock
+    setStarted sock = atomically' (tryPutTMVar started True) >> pure sock

 loadTLSServerParams :: FilePath -> FilePath -> FilePath -> IO T.ServerParams
 loadTLSServerParams = loadSupportedTLSServerParams supportedParameters
@@ -4,6 +4,7 @@
 module Simplex.Messaging.Util where

 import qualified Control.Exception as E
+import Control.Logger.Simple
 import Control.Monad
 import Control.Monad.Except
 import Control.Monad.IO.Unlift
@@ -19,6 +20,7 @@ import qualified Data.Text as T
 import Data.Text.Encoding (decodeUtf8With)
 import Data.Time (NominalDiffTime)
 import GHC.Conc (labelThread, myThreadId, threadDelay)
+import GHC.Stack (HasCallStack, withFrozenCallStack)
 import UnliftIO
 import qualified UnliftIO.Exception as UE

@@ -167,3 +169,11 @@ diffToMilliseconds diff = fromIntegral ((truncate $ diff * 1000) :: Integer)

 labelMyThread :: MonadIO m => String -> m ()
 labelMyThread label = liftIO $ myThreadId >>= (`labelThread` label)
+
+{-# INLINE atomically' #-}
+atomically' :: (MonadIO m, HasCallStack) => STM a -> m a
+atomically' f =
+  liftIO $
+    atomically f `UE.catch` \e@E.BlockedIndefinitelyOnSTM -> do
+      withFrozenCallStack $ logError "BlockedIndefinitelyOnSTM"
+      throwIO e
@@ -102,16 +102,16 @@ connectRCHost drg pairing@RCHostPairing {caKey, caCert, idPrivKey, knownHost} ct
  r <- newEmptyTMVarIO
  found@(RCCtrlAddress {address} :| _) <- findCtrlAddress
  c@RCHClient_ {startedPort, announcer} <- liftIO mkClient
-  hostKeys <- atomically genHostKeys
+  hostKeys <- atomically' genHostKeys
  action <- liftIO $ runClient c r hostKeys
  -- wait for the port to make invitation
-  portNum <- atomically $ readTMVar startedPort
+  portNum <- atomically' $ readTMVar startedPort
  signedInv@RCSignedInvitation {invitation} <- maybe (throwError RCETLSStartFailed) (liftIO . mkInvitation hostKeys address) portNum
  when multicast $ case knownHost of
    Nothing -> throwError RCENewController
    Just KnownHostPairing {hostDhPubKey} -> do
      ann <- liftIO . async . runExceptT $ announceRC drg 60 idPrivKey hostDhPubKey hostKeys invitation
-      atomically $ putTMVar announcer ann
+      atomically' $ putTMVar announcer ann
  pure (found, signedInv, RCHostClient {action, client_ = c}, r)
  where
    findCtrlAddress :: ExceptT RCErrorType IO (NonEmpty RCCtrlAddress)
@@ -131,33 +131,33 @@ connectRCHost drg pairing@RCHostPairing {caKey, caCert, idPrivKey, knownHost} ct
      startTLSServer port_ startedPort tlsCreds (tlsHooks r knownHost hostCAHash) $ \tls ->
        void . runExceptT $ do
          r' <- newEmptyTMVarIO
-          whenM (atomically $ tryPutTMVar r $ Right (tlsUniq tls, tls, r')) $
+          whenM (atomically' $ tryPutTMVar r $ Right (tlsUniq tls, tls, r')) $
            runSession tls r' `putRCError` r'
      where
        runSession tls r' = do
          logDebug "Incoming TLS connection"
          hostEncHello <- receiveRCPacket tls
          logDebug "Received host HELLO"
-          hostCA <- atomically $ takeTMVar hostCAHash
+          hostCA <- atomically' $ takeTMVar hostCAHash
          (ctrlEncHello, sessionKeys, helloBody, pairing') <- prepareHostSession drg hostCA pairing hostKeys hostEncHello
          sendRCPacket tls ctrlEncHello
          logDebug "Sent ctrl HELLO"
-          whenM (atomically $ tryPutTMVar r' $ Right (RCHostSession {tls, sessionKeys}, helloBody, pairing')) $ do
-            atomically (tryReadTMVar announcer) >>= mapM_ uninterruptibleCancel
+          whenM (atomically' $ tryPutTMVar r' $ Right (RCHostSession {tls, sessionKeys}, helloBody, pairing')) $ do
+            atomically' (tryReadTMVar announcer) >>= mapM_ uninterruptibleCancel
            -- can use `RCHostSession` until `endSession` is signalled
            logDebug "Holding session"
-            atomically $ takeTMVar endSession
+            atomically' $ takeTMVar endSession
    tlsHooks :: TMVar a -> Maybe KnownHostPairing -> TMVar C.KeyHash -> TLS.ServerHooks
    tlsHooks r knownHost_ hostCAHash =
      def
-        { TLS.onNewHandshake = \_ -> atomically $ isNothing <$> tryReadTMVar r,
+        { TLS.onNewHandshake = \_ -> atomically' $ isNothing <$> tryReadTMVar r,
          TLS.onClientCertificate = \(X509.CertificateChain chain) ->
            case chain of
              [_leaf, ca] -> do
                let kh = certFingerprint ca
                    accept = maybe True (\h -> hostFingerprint h == kh) knownHost_
                if accept
-                  then atomically (putTMVar hostCAHash kh) $> TLS.CertificateUsageAccept
+                  then atomically' (putTMVar hostCAHash kh) $> TLS.CertificateUsageAccept
                  else pure $ TLS.CertificateUsageReject TLS.CertificateRejectUnknownCA
              _ ->
                pure $ TLS.CertificateUsageReject TLS.CertificateRejectUnknownCA
@@ -197,8 +197,8 @@ certFingerprint caCert = C.KeyHash fp

 cancelHostClient :: RCHostClient -> IO ()
 cancelHostClient RCHostClient {action, client_ = RCHClient_ {announcer, endSession}} = do
-  atomically $ putTMVar endSession ()
-  atomically (tryTakeTMVar announcer) >>= mapM_ uninterruptibleCancel
+  atomically' $ putTMVar endSession ()
+  atomically' (tryTakeTMVar announcer) >>= mapM_ uninterruptibleCancel
  uninterruptibleCancel action

 prepareHostSession :: TVar ChaChaDRG -> C.KeyHash -> RCHostPairing -> RCHostKeys -> RCHostEncHello -> ExceptT RCErrorType IO (RCCtrlEncHello, HostSessKeys, RCHostHello, RCHostPairing)
@@ -285,9 +285,9 @@ connectRCCtrl_ drg pairing'@RCCtrlPairing {caKey, caCert} inv@RCInvitation {ca,
        liftIO $ peekBuffered tlsBuffer 100000 (TLS.recvData tlsContext) >>= logDebug . tshow -- should normally be ("", Nothing) here
        logDebug "Got TLS connection"
        r' <- newEmptyTMVarIO
-        whenM (atomically $ tryPutTMVar r $ Right (tlsUniq tls, tls, r')) $ do
+        whenM (atomically' $ tryPutTMVar r $ Right (tlsUniq tls, tls, r')) $ do
          logDebug "Waiting for session confirmation"
-          whenM (atomically $ readTMVar confirmSession) $ runSession tls r' `putRCError` r'
+          whenM (atomically' $ readTMVar confirmSession) $ runSession tls r' `putRCError` r'
      where
        runSession tls r' = do
          (sharedKey, kemPrivKey, hostEncHello) <- prepareHostHello drg pairing' inv hostAppInfo
@@ -295,11 +295,11 @@ connectRCCtrl_ drg pairing'@RCCtrlPairing {caKey, caCert} inv@RCInvitation {ca,
          ctrlEncHello <- receiveRCPacket tls
          logDebug "Received ctrl HELLO"
          ctrlSessKeys <- prepareCtrlSession pairing' inv sharedKey kemPrivKey ctrlEncHello
-          whenM (atomically $ tryPutTMVar r' $ Right (RCCtrlSession {tls, sessionKeys = ctrlSessKeys}, pairing')) $ do
+          whenM (atomically' $ tryPutTMVar r' $ Right (RCCtrlSession {tls, sessionKeys = ctrlSessKeys}, pairing')) $ do
            logDebug "Session started"
            -- release second putTMVar in confirmCtrlSession
-            void . atomically $ takeTMVar confirmSession
-            atomically $ takeTMVar endSession
+            void . atomically' $ takeTMVar confirmSession
+            atomically' $ takeTMVar endSession
            logDebug "Session ended"

 catchRCError :: ExceptT RCErrorType IO a -> (RCErrorType -> ExceptT RCErrorType IO a) -> ExceptT RCErrorType IO a
@@ -307,7 +307,7 @@ catchRCError = catchAllErrors (RCEException . show)
 {-# INLINE catchRCError #-}

 putRCError :: ExceptT RCErrorType IO a -> TMVar (Either RCErrorType b) -> ExceptT RCErrorType IO a
-a `putRCError` r = a `catchRCError` \e -> atomically (tryPutTMVar r $ Left e) >> throwError e
+a `putRCError` r = a `catchRCError` \e -> atomically' (tryPutTMVar r $ Left e) >> throwError e

 sendRCPacket :: Encoding a => TLS -> a -> ExceptT RCErrorType IO ()
 sendRCPacket tls pkt = do
@@ -411,14 +411,14 @@ findRCCtrlPairing pairings RCEncInvitation {dhPubKey, nonce, encInvitation} = do
 -- application should call this function when TMVar resolves
 confirmCtrlSession :: RCCtrlClient -> Bool -> IO ()
 confirmCtrlSession RCCtrlClient {client_ = RCCClient_ {confirmSession}} res = do
-  atomically $ putTMVar confirmSession res
+  atomically' $ putTMVar confirmSession res
  -- controler does takeTMVar, freeing the slot
  -- TODO add timeout
-  atomically $ putTMVar confirmSession res -- wait for Ctrl to take the var
+  atomically' $ putTMVar confirmSession res -- wait for Ctrl to take the var

 cancelCtrlClient :: RCCtrlClient -> IO ()
 cancelCtrlClient RCCtrlClient {action, client_ = RCCClient_ {endSession}} = do
-  atomically $ putTMVar endSession ()
+  atomically' $ putTMVar endSession ()
  uninterruptibleCancel action

 -- * Session encryption
@@ -27,7 +27,7 @@ import Simplex.Messaging.Transport (supportedParameters)
 import qualified Simplex.Messaging.Transport as Transport
 import Simplex.Messaging.Transport.Client (TransportHost (..))
 import Simplex.Messaging.Transport.Server (defaultTransportServerConfig, runTransportServerSocket, startTCPServer)
-import Simplex.Messaging.Util (ifM, tshow)
+import Simplex.Messaging.Util (ifM, tshow, atomically')
 import Simplex.RemoteControl.Discovery.Multicast (setMembership)
 import Simplex.RemoteControl.Types
 import UnliftIO
@@ -73,7 +73,7 @@ startTLSServer port_ startedOnPort credentials hooks server = async . liftIO $ d
  started <- newEmptyTMVarIO
  bracketOnError (startTCPServer started $ maybe "0" show port_) (\_e -> setPort Nothing) $ \socket ->
    ifM
-      (atomically $ readTMVar started)
+      (atomically' $ readTMVar started)
      (runServer started socket)
      (setPort Nothing)
  where
@@ -82,7 +82,7 @@ startTLSServer port_ startedOnPort credentials hooks server = async . liftIO $ d
      logInfo $ "System-assigned port: " <> tshow port
      setPort $ Just port
      runTransportServerSocket started (pure socket) "RCP TLS" serverParams defaultTransportServerConfig server
-    setPort = void . atomically . tryPutTMVar startedOnPort
+    setPort = void . atomically' . tryPutTMVar startedOnPort
    serverParams =
      def
        { TLS.serverWantClientCert = True,
@@ -112,19 +112,19 @@ closeListener subscribers sock =

 joinMulticast :: TMVar Int -> N.Socket -> N.HostAddress -> IO ()
 joinMulticast subscribers sock group = do
-  now <- atomically $ takeTMVar subscribers
+  now <- atomically' $ takeTMVar subscribers
  when (now == 0) $ do
    setMembership sock group True >>= \case
-      Left e -> atomically (putTMVar subscribers now) >> logError ("setMembership failed " <> tshow e)
-      Right () -> atomically $ putTMVar subscribers (now + 1)
+      Left e -> atomically' (putTMVar subscribers now) >> logError ("setMembership failed " <> tshow e)
+      Right () -> atomically' $ putTMVar subscribers (now + 1)

 partMulticast :: TMVar Int -> N.Socket -> N.HostAddress -> IO ()
 partMulticast subscribers sock group = do
-  now <- atomically $ takeTMVar subscribers
+  now <- atomically' $ takeTMVar subscribers
  when (now == 1) $
    setMembership sock group False >>= \case
-      Left e -> atomically (putTMVar subscribers now) >> logError ("setMembership failed " <> tshow e)
-      Right () -> atomically $ putTMVar subscribers (now - 1)
+      Left e -> atomically' (putTMVar subscribers now) >> logError ("setMembership failed " <> tshow e)
+      Right () -> atomically' $ putTMVar subscribers (now - 1)

 listenerHostAddr4 :: UDP.ListenSocket -> N.HostAddress
 listenerHostAddr4 sock = case UDP.mySockAddr sock of
@@ -64,6 +64,7 @@ import Data.Time.Clock.System (SystemTime (..), getSystemTime)
 import Data.Type.Equality
 import Data.Word (Word16)
 import qualified Database.SQLite.Simple as SQL
+import GHC.Stack (withFrozenCallStack)
 import SMPAgentClient
 import SMPClient (cfg, testPort, testPort2, testStoreLogFile2, withSmpServer, withSmpServerConfigOn, withSmpServerOn, withSmpServerStoreLogOn, withSmpServerStoreMsgLogOn, withSmpServerV7)
 import Simplex.Messaging.Agent hiding (createConnection, joinConnection, sendMessage)
@@ -76,15 +77,16 @@ import Simplex.Messaging.Agent.Store.SQLite (MigrationConfirmation (..), SQLiteS
 import Simplex.Messaging.Agent.Store.SQLite.Common (withTransaction')
 import Simplex.Messaging.Client (NetworkConfig (..), ProtocolClientConfig (..), TransportSessionMode (TSMEntity, TSMUser), defaultSMPClientConfig)
 import qualified Simplex.Messaging.Crypto as C
-import Simplex.Messaging.Crypto.Ratchet (InitialKeys (..), PQEncryption (..), PQSupport (..), pattern PQEncOn, pattern PQEncOff, pattern PQSupportOn, pattern PQSupportOff)
+import Simplex.Messaging.Crypto.Ratchet (InitialKeys (..), PQEncryption (..), PQSupport (..), pattern PQEncOff, pattern PQEncOn, pattern PQSupportOff, pattern PQSupportOn)
 import qualified Simplex.Messaging.Crypto.Ratchet as CR
 import Simplex.Messaging.Encoding.String
-import Simplex.Messaging.Notifications.Transport (NTFVersion, pattern VersionNTF, authBatchCmdsNTFVersion)
+import Simplex.Messaging.Notifications.Transport (NTFVersion, authBatchCmdsNTFVersion, pattern VersionNTF)
 import Simplex.Messaging.Protocol (BasicAuth, ErrorType (..), MsgBody, ProtocolServer (..), SubscriptionMode (..), supportedSMPClientVRange)
 import qualified Simplex.Messaging.Protocol as SMP
 import Simplex.Messaging.Server.Env.STM (ServerConfig (..))
 import Simplex.Messaging.Server.Expiration
-import Simplex.Messaging.Transport (ATransport (..), SMPVersion, VersionSMP, authCmdsSMPVersion, batchCmdsSMPVersion, basicAuthSMPVersion, currentServerSMPRelayVersion)
+import Simplex.Messaging.Transport (ATransport (..), SMPVersion, VersionSMP, authCmdsSMPVersion, basicAuthSMPVersion, batchCmdsSMPVersion, currentServerSMPRelayVersion)
+import Simplex.Messaging.Util (atomically')
 import Simplex.Messaging.Version (VersionRange (..))
 import qualified Simplex.Messaging.Version as V
 import Simplex.Messaging.Version.Internal (Version (..))
@@ -113,20 +115,20 @@ withTimeout a test =
    Nothing -> error "operation timed out"
    Just t -> liftIO $ test t

-get :: MonadIO m => AgentClient -> m (AEntityTransmission 'AEConn)
-get = get' @'AEConn
+get :: (MonadIO m, HasCallStack) => AgentClient -> m (AEntityTransmission 'AEConn)
+get c = withFrozenCallStack $ get' @'AEConn c

-rfGet :: MonadIO m => AgentClient -> m (AEntityTransmission 'AERcvFile)
-rfGet = get' @'AERcvFile
+rfGet :: (MonadIO m, HasCallStack) => AgentClient -> m (AEntityTransmission 'AERcvFile)
+rfGet c = withFrozenCallStack $ get' @'AERcvFile c

-sfGet :: MonadIO m => AgentClient -> m (AEntityTransmission 'AESndFile)
-sfGet = get' @'AESndFile
+sfGet :: (MonadIO m, HasCallStack) => AgentClient -> m (AEntityTransmission 'AESndFile)
+sfGet c = withFrozenCallStack $ get' @'AESndFile c

-nGet :: MonadIO m => AgentClient -> m (AEntityTransmission 'AENone)
-nGet = get' @'AENone
+nGet :: (MonadIO m, HasCallStack) => AgentClient -> m (AEntityTransmission 'AENone)
+nGet c = withFrozenCallStack $ get' @'AENone c

-get' :: forall e m. (MonadIO m, AEntityI e) => AgentClient -> m (AEntityTransmission e)
-get' c = do
+get' :: forall e m. (MonadIO m, AEntityI e, HasCallStack) => AgentClient -> m (AEntityTransmission e)
+get' c = withFrozenCallStack $ do
  (corrId, connId, APC e cmd) <- pGet c
  case testEquality e (sAEntity @e) of
    Just Refl -> pure (corrId, connId, cmd)
@@ -134,7 +136,7 @@ get' c = do

 pGet :: forall m. MonadIO m => AgentClient -> m (ATransmission 'Agent)
 pGet c = do
-  t@(_, _, APC _ cmd) <- atomically (readTBQueue $ subQ c)
+  t@(_, _, APC _ cmd) <- atomically' (readTBQueue $ subQ c)
  case cmd of
    CONNECT {} -> pGet c
    DISCONNECT {} -> pGet c
@@ -219,11 +221,11 @@ runRight action =
    Left e -> error $ "Unexpected error: " <> show e

 getInAnyOrder :: HasCallStack => AgentClient -> [ATransmission 'Agent -> Bool] -> Expectation
-getInAnyOrder c = inAnyOrder (pGet c)
+getInAnyOrder c ts = withFrozenCallStack $ inAnyOrder (pGet c) ts

 inAnyOrder :: (Show a, MonadIO m, HasCallStack) => m a -> [a -> Bool] -> m ()
 inAnyOrder _ [] = pure ()
-inAnyOrder g rs = do
+inAnyOrder g rs = withFrozenCallStack $ do
  r <- g
  let rest = filter (not . expected r) rs
  if length rest < length rs
@@ -280,7 +282,7 @@ functionalAPITests t = do
        testIncreaseConnAgentVersionMaxCompatible t
      it "should increase when connection was negotiated on different versions" $
        testIncreaseConnAgentVersionStartDifferentVersion t
-      -- TODO PQ tests for upgrading connection to PQ encryption
+    -- TODO PQ tests for upgrading connection to PQ encryption
    it "should deliver message after client restart" $
      testDeliverClientRestart t
    it "should deliver messages to the user once, even if repeat delivery is made by the server (no ACK)" $
@@ -440,7 +442,7 @@ canCreateQueue allowNew (srvAuth, srvVersion) (clntAuth, clntVersion) =

 testMatrix2 :: ATransport -> (PQSupport -> AgentClient -> AgentClient -> AgentMsgId -> IO ()) -> Spec
 testMatrix2 t runTest = do
-  it "v7" $ withSmpServerV7 t $ runTestCfg2 agentCfgV7 agentCfgV7 3 $ runTest PQSupportOn 
+  it "v7" $ withSmpServerV7 t $ runTestCfg2 agentCfgV7 agentCfgV7 3 $ runTest PQSupportOn
  it "v7 to current" $ withSmpServerV7 t $ runTestCfg2 agentCfgV7 agentCfg 3 $ runTest PQSupportOn
  it "current to v7" $ withSmpServerV7 t $ runTestCfg2 agentCfg agentCfgV7 3 $ runTest PQSupportOn
  it "current with v7 server" $ withSmpServerV7 t $ runTestCfg2 agentCfg agentCfg 3 $ runTest PQSupportOn
@@ -451,10 +453,10 @@ testMatrix2 t runTest = do

 testRatchetMatrix2 :: ATransport -> (PQSupport -> AgentClient -> AgentClient -> AgentMsgId -> IO ()) -> Spec
 testRatchetMatrix2 t runTest = do
-  it "ratchet next" $ withSmpServerV7 t $ runTestCfg2 agentCfgV7 agentCfgV7 3 $ runTest PQSupportOn 
-  it "ratchet next to current" $ withSmpServerV7 t $ runTestCfg2 agentCfgV7 agentCfg 3 $ runTest PQSupportOn 
-  it "ratchet current to next" $ withSmpServerV7 t $ runTestCfg2 agentCfg agentCfgV7 3 $ runTest PQSupportOn 
-  it "ratchet current" $ withSmpServer t $ runTestCfg2 agentCfg agentCfg 3 $ runTest PQSupportOn 
+  it "ratchet next" $ withSmpServerV7 t $ runTestCfg2 agentCfgV7 agentCfgV7 3 $ runTest PQSupportOn
+  it "ratchet next to current" $ withSmpServerV7 t $ runTestCfg2 agentCfgV7 agentCfg 3 $ runTest PQSupportOn
+  it "ratchet current to next" $ withSmpServerV7 t $ runTestCfg2 agentCfg agentCfgV7 3 $ runTest PQSupportOn
+  it "ratchet current" $ withSmpServer t $ runTestCfg2 agentCfg agentCfg 3 $ runTest PQSupportOn
  it "ratchet prev" $ withSmpServer t $ runTestCfg2 agentCfgRatchetVPrev agentCfgRatchetVPrev 3 $ runTest PQSupportOff
  it "ratchets prev to current" $ withSmpServer t $ runTestCfg2 agentCfgRatchetVPrev agentCfg 3 $ runTest PQSupportOff
  it "ratchets current to prev" $ withSmpServer t $ runTestCfg2 agentCfg agentCfgRatchetVPrev 3 $ runTest PQSupportOff
@@ -1366,8 +1368,8 @@ testInactiveNoSubs t = do
  withSmpServerConfigOn t cfg' testPort $ \_ -> do
    alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB
    runRight_ . void $ createConnection alice 1 True SCMInvitation Nothing SMOnlyCreate -- do not subscribe to pass noSubscriptions check
-    Just (_, _, APC SAENone (CONNECT _ _)) <- timeout 2000000 $ atomically (readTBQueue $ subQ alice)
-    Just (_, _, APC SAENone (DISCONNECT _ _)) <- timeout 5000000 $ atomically (readTBQueue $ subQ alice)
+    Just (_, _, APC SAENone (CONNECT _ _)) <- timeout 2000000 $ atomically' (readTBQueue $ subQ alice)
+    Just (_, _, APC SAENone (DISCONNECT _ _)) <- timeout 5000000 $ atomically' (readTBQueue $ subQ alice)
    disposeAgentClient alice

 testInactiveWithSubs :: ATransport -> IO ()
@@ -1654,6 +1656,7 @@ testDeleteConnectionAsync t = do
    pure ([bId1, bId2, bId3] :: [ConnId])
  runRight_ $ do
    deleteConnectionsAsync a False connIds
+    nGet a =##> \case ("", "", DOWN {}) -> True; _ -> False
    get a =##> \case ("", c, DEL_RCVQ _ _ (Just (BROKER _ e))) -> c `elem` connIds && (e == TIMEOUT || e == NETWORK); _ -> False
    get a =##> \case ("", c, DEL_RCVQ _ _ (Just (BROKER _ e))) -> c `elem` connIds && (e == TIMEOUT || e == NETWORK); _ -> False
    get a =##> \case ("", c, DEL_RCVQ _ _ (Just (BROKER _ e))) -> c `elem` connIds && (e == TIMEOUT || e == NETWORK); _ -> False
@@ -68,6 +68,7 @@ import System.Directory (doesFileExist, removeFile)
 import Test.Hspec
 import UnliftIO
 import Util
+import Simplex.Messaging.Util (atomically')

 removeFileIfExists :: FilePath -> IO ()
 removeFileIfExists filePath = do
@@ -170,7 +171,7 @@ testNotificationToken APNSMockServer {apnsQ} = do
    let tkn = DeviceToken PPApnsTest "abcd"
    NTRegistered <- registerNtfToken a tkn NMPeriodic
    APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
-      atomically $ readTBQueue apnsQ
+      atomically' $ readTBQueue apnsQ
    verification <- ntfData .-> "verification"
    nonce <- C.cbNonce <$> ntfData .-> "nonce"
    liftIO $ sendApnsResponse APNSRespOk
@@ -198,13 +199,13 @@ testNtfTokenRepeatRegistration APNSMockServer {apnsQ} = do
    let tkn = DeviceToken PPApnsTest "abcd"
    NTRegistered <- registerNtfToken a tkn NMPeriodic
    APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
-      atomically $ readTBQueue apnsQ
+      atomically' $ readTBQueue apnsQ
    verification <- ntfData .-> "verification"
    nonce <- C.cbNonce <$> ntfData .-> "nonce"
    liftIO $ sendApnsResponse APNSRespOk
    NTRegistered <- registerNtfToken a tkn NMPeriodic
    APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}, sendApnsResponse = sendApnsResponse'} <-
-      atomically $ readTBQueue apnsQ
+      atomically' $ readTBQueue apnsQ
    _ <- ntfData' .-> "verification"
    _ <- C.cbNonce <$> ntfData' .-> "nonce"
    liftIO $ sendApnsResponse' APNSRespOk
@@ -223,7 +224,7 @@ testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do
    let tkn = DeviceToken PPApnsTest "abcd"
    NTRegistered <- registerNtfToken a tkn NMPeriodic
    APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
-      atomically $ readTBQueue apnsQ
+      atomically' $ readTBQueue apnsQ
    verification <- ntfData .-> "verification"
    nonce <- C.cbNonce <$> ntfData .-> "nonce"
    liftIO $ sendApnsResponse APNSRespOk
@@ -231,7 +232,7 @@ testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do

    NTRegistered <- registerNtfToken a' tkn NMPeriodic
    APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}, sendApnsResponse = sendApnsResponse'} <-
-      atomically $ readTBQueue apnsQ
+      atomically' $ readTBQueue apnsQ
    verification' <- ntfData' .-> "verification"
    nonce' <- C.cbNonce <$> ntfData' .-> "nonce"
    liftIO $ sendApnsResponse' APNSRespOk
@@ -258,7 +259,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
  ntfData <- withNtfServer t . runRight $ do
    NTRegistered <- registerNtfToken a tkn NMPeriodic
    APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
-      atomically $ readTBQueue apnsQ
+      atomically' $ readTBQueue apnsQ
    liftIO $ sendApnsResponse APNSRespOk
    pure ntfData
  -- the new agent is created as otherwise when running the tests in CI the old agent was keeping the connection to the server
@@ -272,7 +273,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do
    nonce <- C.cbNonce <$> ntfData .-> "nonce"
    Left (NTF AUTH) <- tryE $ verifyNtfToken a' tkn nonce verification
    APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}, sendApnsResponse = sendApnsResponse'} <-
-      atomically $ readTBQueue apnsQ
+      atomically' $ readTBQueue apnsQ
    verification' <- ntfData' .-> "verification"
    nonce' <- C.cbNonce <$> ntfData' .-> "nonce"
    liftIO $ sendApnsResponse' APNSRespOk
@@ -295,7 +296,7 @@ testNtfTokenMultipleServers t APNSMockServer {apnsQ} = do
      -- register a new token, the agent picks a server and stores its choice
      NTRegistered <- registerNtfToken a tkn NMPeriodic
      APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
-        atomically $ readTBQueue apnsQ
+        atomically' $ readTBQueue apnsQ
      verification <- ntfData .-> "verification"
      nonce <- C.cbNonce <$> ntfData .-> "nonce"
      liftIO $ sendApnsResponse APNSRespOk
@@ -365,7 +366,7 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} alice@Agen
    let tkn = DeviceToken PPApnsTest "abcd"
    NTRegistered <- registerNtfToken alice tkn NMInstant
    APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse} <-
-      atomically $ readTBQueue apnsQ
+      atomically' $ readTBQueue apnsQ
    verification <- ntfData .-> "verification"
    vNonce <- C.cbNonce <$> ntfData .-> "nonce"
    liftIO $ sendApnsResponse APNSRespOk
@@ -450,7 +451,7 @@ registerTestToken a token mode apnsQ = do
  let tkn = DeviceToken PPApnsTest token
  NTRegistered <- registerNtfToken a tkn mode
  Just APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData'}, sendApnsResponse = sendApnsResponse'} <-
-    timeout 1000000 . atomically $ readTBQueue apnsQ
+    timeout 1000000 . atomically' $ readTBQueue apnsQ
  verification' <- ntfData' .-> "verification"
  nonce' <- C.cbNonce <$> ntfData' .-> "nonce"
  liftIO $ sendApnsResponse' APNSRespOk
@@ -764,7 +765,7 @@ testMessage_ apnsQ a aId b bId msg = do

 messageNotification :: HasCallStack => TBQueue APNSMockRequest -> ExceptT AgentErrorType IO (C.CbNonce, ByteString)
 messageNotification apnsQ = do
-  1000000 `timeout` atomically (readTBQueue apnsQ) >>= \case
+  1000000 `timeout` atomically' (readTBQueue apnsQ) >>= \case
    Nothing -> error "no notification"
    Just APNSMockRequest {notification = APNSNotification {aps = APNSMutableContent {}, notificationData = Just ntfData}, sendApnsResponse} -> do
      nonce <- C.cbNonce <$> ntfData .-> "nonce"
@@ -782,6 +783,6 @@ messageNotificationData c apnsQ = do

 noNotification :: TBQueue APNSMockRequest -> ExceptT AgentErrorType IO ()
 noNotification apnsQ = do
-  500000 `timeout` atomically (readTBQueue apnsQ) >>= \case
+  500000 `timeout` atomically' (readTBQueue apnsQ) >>= \case
    Nothing -> pure ()
    _ -> error "unexpected notification"
@@ -54,6 +54,7 @@ import qualified Simplex.Messaging.Protocol as SMP
 import System.Random
 import Test.Hspec
 import UnliftIO.Directory (removeFile)
+import Simplex.Messaging.Util (atomically')

 testDB :: String
 testDB = "tests/tmp/smp-agent.test.db"
@@ -88,7 +89,7 @@ removeStore db = do
  removeFile $ dbFilePath db
  where
    close :: SQLiteStore -> IO ()
-    close st = mapM_ DB.close =<< atomically (tryTakeTMVar $ dbConnection st)
+    close st = mapM_ DB.close =<< atomically' (tryTakeTMVar $ dbConnection st)

 storeTests :: Spec
 storeTests = do
@@ -16,6 +16,7 @@ import qualified Simplex.Messaging.Crypto as C
 import Simplex.Messaging.Protocol (SMPServer, pattern VersionSMPC)
 import Test.Hspec
 import UnliftIO
+import Simplex.Messaging.Util (atomically')

 tRcvQueuesTests :: Spec
 tRcvQueuesTests = do
@@ -28,7 +29,7 @@ tRcvQueuesTests = do
    it "getDelSessQueues" getDelSessQueuesTest

 checkDataInvariant :: RQ.TRcvQueues -> IO Bool
-checkDataInvariant trq = atomically $ do
+checkDataInvariant trq = atomically' $ do
  conns <- readTVar $ RQ.getConnections trq
  qs <- readTVar $ RQ.getRcvQueues trq
  -- three invariant checks
@@ -39,87 +40,87 @@ checkDataInvariant trq = atomically $ do

 hasConnTest :: IO ()
 hasConnTest = do
-  trq <- atomically RQ.empty
-  atomically $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c1") trq
+  trq <- atomically' RQ.empty
+  atomically' $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c1") trq
  checkDataInvariant trq `shouldReturn` True
-  atomically $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c2") trq
+  atomically' $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c2") trq
  checkDataInvariant trq `shouldReturn` True
-  atomically $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@beta" "c3") trq
+  atomically' $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@beta" "c3") trq
  checkDataInvariant trq `shouldReturn` True
-  atomically (RQ.hasConn "c1" trq) `shouldReturn` True
-  atomically (RQ.hasConn "c2" trq) `shouldReturn` True
-  atomically (RQ.hasConn "c3" trq) `shouldReturn` True
-  atomically (RQ.hasConn "nope" trq) `shouldReturn` False
+  atomically' (RQ.hasConn "c1" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "c2" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "c3" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "nope" trq) `shouldReturn` False

 hasConnTestBatch :: IO ()
 hasConnTestBatch = do
-  trq <- atomically RQ.empty
+  trq <- atomically' RQ.empty
  let qs = [dummyRQ 0 "smp://1234-w==@alpha" "c1", dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@beta" "c3"]
-  atomically $ RQ.batchAddQueues trq qs
+  atomically' $ RQ.batchAddQueues trq qs
  checkDataInvariant trq `shouldReturn` True
-  atomically (RQ.hasConn "c1" trq) `shouldReturn` True
-  atomically (RQ.hasConn "c2" trq) `shouldReturn` True
-  atomically (RQ.hasConn "c3" trq) `shouldReturn` True
-  atomically (RQ.hasConn "nope" trq) `shouldReturn` False
+  atomically' (RQ.hasConn "c1" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "c2" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "c3" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "nope" trq) `shouldReturn` False

 deleteConnTest :: IO ()
 deleteConnTest = do
-  trq <- atomically RQ.empty
-  atomically $ do
+  trq <- atomically' RQ.empty
+  atomically' $ do
    RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c1") trq
    RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c2") trq
    RQ.addQueue (dummyRQ 0 "smp://1234-w==@beta" "c3") trq
  checkDataInvariant trq `shouldReturn` True
-  atomically $ RQ.deleteConn "c1" trq
+  atomically' $ RQ.deleteConn "c1" trq
  checkDataInvariant trq `shouldReturn` True
-  atomically $ RQ.deleteConn "nope" trq
+  atomically' $ RQ.deleteConn "nope" trq
  checkDataInvariant trq `shouldReturn` True
  M.keys <$> readTVarIO (RQ.getConnections trq) `shouldReturn` ["c2", "c3"]

 getSessQueuesTest :: IO ()
 getSessQueuesTest = do
-  trq <- atomically RQ.empty
-  atomically $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c1") trq
+  trq <- atomically' RQ.empty
+  atomically' $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c1") trq
  checkDataInvariant trq `shouldReturn` True
-  atomically $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c2") trq
+  atomically' $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@alpha" "c2") trq
  checkDataInvariant trq `shouldReturn` True
-  atomically $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@beta" "c3") trq
+  atomically' $ RQ.addQueue (dummyRQ 0 "smp://1234-w==@beta" "c3") trq
  checkDataInvariant trq `shouldReturn` True
-  atomically $ RQ.addQueue (dummyRQ 1 "smp://1234-w==@beta" "c4") trq
+  atomically' $ RQ.addQueue (dummyRQ 1 "smp://1234-w==@beta" "c4") trq
  checkDataInvariant trq `shouldReturn` True
-  atomically (RQ.getSessQueues (0, "smp://1234-w==@alpha", Just "c1") trq) `shouldReturn` [dummyRQ 0 "smp://1234-w==@alpha" "c1"]
-  atomically (RQ.getSessQueues (1, "smp://1234-w==@alpha", Just "c1") trq) `shouldReturn` []
-  atomically (RQ.getSessQueues (0, "smp://1234-w==@alpha", Just "nope") trq) `shouldReturn` []
-  atomically (RQ.getSessQueues (0, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` [dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@alpha" "c1"]
+  atomically' (RQ.getSessQueues (0, "smp://1234-w==@alpha", Just "c1") trq) `shouldReturn` [dummyRQ 0 "smp://1234-w==@alpha" "c1"]
+  atomically' (RQ.getSessQueues (1, "smp://1234-w==@alpha", Just "c1") trq) `shouldReturn` []
+  atomically' (RQ.getSessQueues (0, "smp://1234-w==@alpha", Just "nope") trq) `shouldReturn` []
+  atomically' (RQ.getSessQueues (0, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` [dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@alpha" "c1"]

 getDelSessQueuesTest :: IO ()
 getDelSessQueuesTest = do
-  trq <- atomically RQ.empty
+  trq <- atomically' RQ.empty
  let qs =
        [ dummyRQ 0 "smp://1234-w==@alpha" "c1",
          dummyRQ 0 "smp://1234-w==@alpha" "c2",
          dummyRQ 0 "smp://1234-w==@beta" "c3",
          dummyRQ 1 "smp://1234-w==@beta" "c4"
        ]
-  atomically $ RQ.batchAddQueues trq qs
+  atomically' $ RQ.batchAddQueues trq qs
  checkDataInvariant trq `shouldReturn` True
  -- no user
-  atomically (RQ.getDelSessQueues (2, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` ([], [])
+  atomically' (RQ.getDelSessQueues (2, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` ([], [])
  checkDataInvariant trq `shouldReturn` True
  -- wrong user
-  atomically (RQ.getDelSessQueues (1, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` ([], [])
+  atomically' (RQ.getDelSessQueues (1, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` ([], [])
  checkDataInvariant trq `shouldReturn` True
  -- connections intact
-  atomically (RQ.hasConn "c1" trq) `shouldReturn` True
-  atomically (RQ.hasConn "c2" trq) `shouldReturn` True
-  atomically (RQ.getDelSessQueues (0, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` ([dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@alpha" "c1"], ["c1", "c2"])
+  atomically' (RQ.hasConn "c1" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "c2" trq) `shouldReturn` True
+  atomically' (RQ.getDelSessQueues (0, "smp://1234-w==@alpha", Nothing) trq) `shouldReturn` ([dummyRQ 0 "smp://1234-w==@alpha" "c2", dummyRQ 0 "smp://1234-w==@alpha" "c1"], ["c1", "c2"])
  checkDataInvariant trq `shouldReturn` True
  -- connections gone
-  atomically (RQ.hasConn "c1" trq) `shouldReturn` False
-  atomically (RQ.hasConn "c2" trq) `shouldReturn` False
+  atomically' (RQ.hasConn "c1" trq) `shouldReturn` False
+  atomically' (RQ.hasConn "c2" trq) `shouldReturn` False
  -- non-matched connections intact
-  atomically (RQ.hasConn "c3" trq) `shouldReturn` True
-  atomically (RQ.hasConn "c4" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "c3" trq) `shouldReturn` True
+  atomically' (RQ.hasConn "c4" trq) `shouldReturn` True

 dummyRQ :: UserId -> SMPServer -> ConnId -> RcvQueue
 dummyRQ userId server connId =
@@ -51,6 +51,7 @@ import UnliftIO.Async
 import UnliftIO.Concurrent
 import qualified UnliftIO.Exception as E
 import UnliftIO.STM
+import Simplex.Messaging.Util (atomically')

 testHost :: NonEmpty TransportHost
 testHost = "localhost"
@@ -223,7 +224,7 @@ getAPNSMockServer config@HTTP2ServerConfig {qSize} = do
  pure APNSMockServer {action, apnsQ, http2Server}
  where
    runAPNSMockServer apnsQ HTTP2Server {reqQ} = forever $ do
-      HTTP2Request {reqBody = HTTP2Body {bodyHead}, sendResponse} <- atomically $ readTBQueue reqQ
+      HTTP2Request {reqBody = HTTP2Body {bodyHead}, sendResponse} <- atomically' $ readTBQueue reqQ
      let sendApnsResponse = \case
            APNSRespOk -> sendResponse $ H.responseNoBody N.ok200 []
            APNSRespError status reason ->
@@ -45,6 +45,7 @@ import Simplex.Messaging.Protocol hiding (notification)
 import Simplex.Messaging.Transport
 import Test.Hspec
 import UnliftIO.STM
+import Simplex.Messaging.Util (atomically')

 ntfServerTests :: ATransport -> Spec
 ntfServerTests t = do
@@ -112,7 +113,7 @@ testNotificationSubscription (ATransport t) =
          -- register and verify token
          RespNtf "1" "" (NRTknId tId ntfDh) <- signSendRecvNtf nh tknKey ("1", "", TNEW $ NewNtfTkn tkn tknPub dhPub)
          APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData}, sendApnsResponse = send} <-
-            atomically $ readTBQueue apnsQ
+            atomically' $ readTBQueue apnsQ
          send APNSRespOk
          let dhSecret = C.dh' ntfDh dhPriv
              Right verification = ntfData .-> "verification"
@@ -131,7 +132,7 @@ testNotificationSubscription (ATransport t) =
          threadDelay 50000
          Resp "5" _ OK <- signSendRecv sh sKey ("5", sId, _SEND' "hello")
          -- receive notification
-          APNSMockRequest {notification, sendApnsResponse = send'} <- atomically $ readTBQueue apnsQ
+          APNSMockRequest {notification, sendApnsResponse = send'} <- atomically' $ readTBQueue apnsQ
          let APNSNotification {aps = APNSMutableContent {}, notificationData = Just ntfData'} = notification
              Right nonce' = C.cbNonce <$> ntfData' .-> "nonce"
              Right message = ntfData' .-> "message"
@@ -155,7 +156,7 @@ testNotificationSubscription (ATransport t) =
          RespNtf "7" tId' NROk <- signSendRecvNtf nh tknKey ("7", tId, TRPL tkn')
          tId `shouldBe` tId'
          APNSMockRequest {notification = APNSNotification {aps = APNSBackground _, notificationData = Just ntfData2}, sendApnsResponse = send2} <-
-            atomically $ readTBQueue apnsQ
+            atomically' $ readTBQueue apnsQ
          send2 APNSRespOk
          let Right verification2 = ntfData2 .-> "verification"
              Right nonce2 = C.cbNonce <$> ntfData2 .-> "nonce"
@@ -164,7 +165,7 @@ testNotificationSubscription (ATransport t) =
          RespNtf "8a" _ (NRTkn NTActive) <- signSendRecvNtf nh tknKey ("8a", tId, TCHK)
          -- send message
          Resp "9" _ OK <- signSendRecv sh sKey ("9", sId, _SEND' "hello 2")
-          APNSMockRequest {notification = notification3, sendApnsResponse = send3} <- atomically $ readTBQueue apnsQ
+          APNSMockRequest {notification = notification3, sendApnsResponse = send3} <- atomically' $ readTBQueue apnsQ
          let APNSNotification {aps = APNSMutableContent {}, notificationData = Just ntfData3} = notification3
              Right nonce3 = C.cbNonce <$> ntfData3 .-> "nonce"
              Right message3 = ntfData3 .-> "message"
@@ -19,6 +19,7 @@ import Simplex.RemoteControl.Types
 import Test.Hspec
 import UnliftIO
 import UnliftIO.Concurrent
+import Simplex.Messaging.Util (atomically')

 remoteControlTests :: Spec
 remoteControlTests = do
@@ -72,9 +73,9 @@ testNewPairing = do
    logNote "c 2"
    putMVar invVar (inv, hc)
    logNote "c 3"
-    Right (sessId, _tls, r') <- atomically $ takeTMVar r
+    Right (sessId, _tls, r') <- atomically' $ takeTMVar r
    logNote "c 4"
-    Right (_rcHostSession, _rcHelloBody, _hp') <- atomically $ takeTMVar r'
+    Right (_rcHostSession, _rcHelloBody, _hp') <- atomically' $ takeTMVar r'
    logNote "c 5"
    threadDelay 250000
    logNote "ctrl: ciao"
@@ -89,11 +90,11 @@ testNewPairing = do
    logNote "h 1"
    (rcCtrlClient, r) <- RC.connectRCCtrl drg inv Nothing (J.String "app")
    logNote "h 2"
-    Right (sessId', _tls, r') <- atomically $ takeTMVar r
+    Right (sessId', _tls, r') <- atomically' $ takeTMVar r
    logNote "h 3"
    liftIO $ RC.confirmCtrlSession rcCtrlClient True
    logNote "h 4"
-    Right (_rcCtrlSession, _rcCtrlPairing) <- atomically $ takeTMVar r'
+    Right (_rcCtrlSession, _rcCtrlPairing) <- atomically' $ takeTMVar r'
    logNote "h 5"
    threadDelay 250000
    logNote "ctrl: adios"
@@ -162,8 +163,8 @@ runCtrl :: TVar ChaChaDRG -> Bool -> RCHostPairing -> MVar RCSignedInvitation ->
 runCtrl drg multicast hp invVar = async . runRight $ do
  (_found, inv, hc, r) <- RC.connectRCHost drg hp (J.String "app") multicast Nothing Nothing
  putMVar invVar inv
-  Right (_sessId, _tls, r') <- atomically $ takeTMVar r
-  Right (_rcHostSession, _rcHelloBody, hp') <- atomically $ takeTMVar r'
+  Right (_sessId, _tls, r') <- atomically' $ takeTMVar r
+  Right (_rcHostSession, _rcHelloBody, hp') <- atomically' $ takeTMVar r'
  threadDelay 250000
  liftIO $ RC.cancelHostClient hc
  pure hp'
@@ -172,9 +173,9 @@ runHostURI :: TVar ChaChaDRG -> Maybe RCCtrlPairing -> RCSignedInvitation -> IO
 runHostURI drg cp_ signedInv = async . runRight $ do
  inv <- maybe (fail "bad invite") pure $ verifySignedInvitation signedInv
  (rcCtrlClient, r) <- RC.connectRCCtrl drg inv cp_ (J.String "app")
-  Right (_sessId', _tls, r') <- atomically $ takeTMVar r
+  Right (_sessId', _tls, r') <- atomically' $ takeTMVar r
  liftIO $ RC.confirmCtrlSession rcCtrlClient True
-  Right (_rcCtrlSession, cp') <- atomically $ takeTMVar r'
+  Right (_rcCtrlSession, cp') <- atomically' $ takeTMVar r'
  threadDelay 250000
  pure cp'

@@ -182,8 +183,8 @@ runHostMulticast :: TVar ChaChaDRG -> TMVar Int -> RCCtrlPairing -> IO (Async RC
 runHostMulticast drg subscribers cp = async . runRight $ do
  (pairing, inv) <- RC.discoverRCCtrl subscribers (cp :| [])
  (rcCtrlClient, r) <- RC.connectRCCtrl drg inv (Just pairing) (J.String "app")
-  Right (_sessId', _tls, r') <- atomically $ takeTMVar r
+  Right (_sessId', _tls, r') <- atomically' $ takeTMVar r
  liftIO $ RC.confirmCtrlSession rcCtrlClient True
-  Right (_rcCtrlSession, cp') <- atomically $ takeTMVar r'
+  Right (_rcCtrlSession, cp') <- atomically' $ takeTMVar r'
  threadDelay 250000
  pure cp'
@@ -25,6 +25,7 @@ import Simplex.Messaging.Server.Env.STM
 import Simplex.Messaging.Transport
 import Simplex.Messaging.Transport.Client
 import Simplex.Messaging.Transport.Server
+import Simplex.Messaging.Util (atomically')
 import Simplex.Messaging.Version (mkVersionRange)
 import System.Environment (lookupEnv)
 import System.Info (os)
@@ -123,7 +124,7 @@ withSmpServerConfigOn :: HasCallStack => ATransport -> ServerConfig -> ServiceNa
 withSmpServerConfigOn t cfg' port' =
  serverBracket
    (\started -> runSMPServerBlocking started cfg' {transports = [(port', t)]})
-    (pure ())
+    (threadDelay 10000)

 withSmpServerThreadOn :: HasCallStack => ATransport -> ServiceName -> (HasCallStack => ThreadId -> IO a) -> IO a
 withSmpServerThreadOn t = withSmpServerConfigOn t cfg
@@ -137,7 +138,7 @@ serverBracket process afterProcess f = do
    (\t -> waitFor started "start" >> f t >>= \r -> r <$ threadDelay 100000)
  where
    waitFor started s =
-      5_000_000 `timeout` atomically (takeTMVar started) >>= \case
+      5_000_000 `timeout` atomically' (takeTMVar started) >>= \case
        Nothing -> error $ "server did not " <> s
        _ -> pure ()

@@ -44,6 +44,7 @@ import System.TimeIt (timeItT)
 import System.Timeout
 import Test.HUnit
 import Test.Hspec
+import Simplex.Messaging.Util (atomically')

 serverTests :: ATransport -> Spec
 serverTests t@(ATransport t') = do
@@ -396,9 +397,9 @@ testGetCommand t =
    smpTest t $ \sh -> do
      queue <- newEmptyTMVarIO
      testSMPClient @c $ \rh ->
-        atomically . putTMVar queue =<< createAndSecureQueue rh sPub
+        atomically' . putTMVar queue =<< createAndSecureQueue rh sPub
      testSMPClient @c $ \rh -> do
-        (sId, rId, rKey, dhShared) <- atomically $ takeTMVar queue
+        (sId, rId, rKey, dhShared) <- atomically' $ takeTMVar queue
        let dec = decryptMsgV3 dhShared
        Resp "1" _ OK <- signSendRecv sh sKey ("1", sId, _SEND "hello")
        Resp "2" _ (Msg mId1 msg1) <- signSendRecv rh rKey ("2", rId, GET)
@@ -5,7 +5,7 @@

 module XFTPClient where

-import Control.Concurrent (ThreadId)
+import Control.Concurrent (ThreadId, threadDelay)
 import Data.String (fromString)
 import Network.Socket (ServiceName)
 import SMPClient (serverBracket)
@@ -53,7 +53,7 @@ withXFTPServerCfg :: HasCallStack => XFTPServerConfig -> (HasCallStack => Thread
 withXFTPServerCfg cfg =
  serverBracket
    (`runXFTPServerBlocking` cfg)
-    (pure ())
+    (threadDelay 10000)

 withXFTPServerThreadOn :: HasCallStack => (HasCallStack => ThreadId -> IO a) -> IO a
 withXFTPServerThreadOn = withXFTPServerCfg testXFTPServerConfig
@@ -33,6 +33,7 @@ import System.FilePath ((</>))
 import Test.Hspec
 import UnliftIO.STM
 import XFTPClient
+import Simplex.Messaging.Util (atomically')

 xftpServerTests :: Spec
 xftpServerTests =
@@ -185,7 +186,7 @@ testWrongChunkSize = xftpTest $ \c -> do
  g <- C.newRandom
  (sndKey, spKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
  (rcvKey, _rpKey) <- atomically $ C.generateAuthKeyPair C.SEd25519 g
-  B.writeFile testChunkPath =<< atomically (C.randomBytes (kb 96) g)
+  B.writeFile testChunkPath =<< atomically' (C.randomBytes (kb 96) g)
  digest <- LC.sha256Hash <$> LB.readFile testChunkPath
  let file = FileInfo {sndKey, size = kb 96, digest}
  runRight_ $
@@ -220,15 +221,15 @@ testInactiveClientExpiration :: Expectation
 testInactiveClientExpiration = withXFTPServerCfg testXFTPServerConfig {inactiveClientExpiration} $ \_ -> runRight_ $ do
  disconnected <- newEmptyTMVarIO
  g <- liftIO C.newRandom
-  c <- ExceptT $ getXFTPClient g (1, testXFTPServer, Nothing) testXFTPClientConfig (\_ -> atomically $ putTMVar disconnected ())
+  c <- ExceptT $ getXFTPClient g (1, testXFTPServer, Nothing) testXFTPClientConfig (\_ -> atomically' $ putTMVar disconnected ())
  pingXFTP c
  liftIO $ do
    threadDelay 100000
-    atomically (tryReadTMVar disconnected) `shouldReturn` Nothing
+    atomically' (tryReadTMVar disconnected) `shouldReturn` Nothing
  pingXFTP c
  liftIO $ do
    threadDelay 3000000
-    atomically (tryTakeTMVar disconnected) `shouldReturn` Just ()
+    atomically' (tryTakeTMVar disconnected) `shouldReturn` Just ()
  where
    inactiveClientExpiration = Just ExpirationConfig {ttl = 1, checkInterval = 1}
Author	SHA1	Message	Date
Alexander Bondarenko	157dd8782c	expect DOWN	2024-04-09 22:43:47 +03:00
Alexander Bondarenko	12cfd30d94	fix some tests	2024-04-09 21:44:23 +03:00
Alexander Bondarenko	e56b4b2b40	mark waitForWork	2024-04-09 20:26:32 +03:00
Alexander Bondarenko	aca371e547	WIP	2024-04-09 20:16:50 +03:00