From b32259d048b491d7e1b102e31807899afdbd8c29 Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin Date: Thu, 21 Mar 2024 14:54:57 +0000 Subject: [PATCH] notify about critical error on agent crash (#1062) * notify about critical error on agent crash * waitUntilActive * disposeAgent * fix --- src/Simplex/Messaging/Agent.hs | 43 +++++-- src/Simplex/Messaging/Agent/Client.hs | 10 +- tests/AgentTests/FunctionalAPITests.hs | 160 ++++++++++++------------- tests/AgentTests/NotificationTests.hs | 54 ++++----- tests/XFTPAgent.hs | 34 +++--- 5 files changed, 164 insertions(+), 137 deletions(-) diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 2b2db1dd5..e9b05c5f0 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -41,6 +41,7 @@ module Simplex.Messaging.Agent getSMPAgentClient, getSMPAgentClient_, disconnectAgentClient, + disposeAgentClient, resumeAgentClient, withConnLock, withInvLock, @@ -122,7 +123,7 @@ import Control.Monad import Control.Monad.Except import Control.Monad.IO.Unlift (MonadUnliftIO) import Control.Monad.Reader -import Crypto.Random (ChaChaDRG, MonadRandom) +import Crypto.Random (ChaChaDRG) import qualified Data.Aeson as J import Data.Bifunctor (bimap, first, second) import Data.ByteString.Char8 (ByteString) @@ -179,28 +180,40 @@ import Simplex.Messaging.Version import Simplex.RemoteControl.Client import Simplex.RemoteControl.Invitation import Simplex.RemoteControl.Types +import System.Mem.Weak (deRefWeak) import UnliftIO.Async (race_) -import UnliftIO.Concurrent (forkFinally, forkIO, threadDelay) +import UnliftIO.Concurrent (forkFinally, forkIO, killThread, mkWeakThreadId, threadDelay) +import qualified UnliftIO.Exception as E import UnliftIO.STM -- import GHC.Conc (unsafeIOToSTM) -- | Creates an SMP agent client instance -getSMPAgentClient :: (MonadRandom m, MonadUnliftIO m) => AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> m AgentClient +getSMPAgentClient :: MonadIO m => AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> m AgentClient getSMPAgentClient = getSMPAgentClient_ 1 {-# INLINE getSMPAgentClient #-} -getSMPAgentClient_ :: (MonadRandom m, MonadUnliftIO m) => Int -> AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> m AgentClient +getSMPAgentClient_ :: MonadIO m => Int -> AgentConfig -> InitialAgentServers -> SQLiteStore -> Bool -> m AgentClient getSMPAgentClient_ clientId cfg initServers store backgroundMode = - liftIO (newSMPAgentEnv cfg store) >>= runReaderT runAgent + liftIO $ newSMPAgentEnv cfg store >>= runReaderT runAgent where runAgent = do - c <- getAgentClient clientId initServers - void $ runAgentThreads c `forkFinally` const (disconnectAgentClient c) + c@AgentClient {acThread} <- getAgentClient clientId initServers + t <- runAgentThreads c `forkFinally` const (disconnectAgentClient c) + atomically . writeTVar acThread . Just =<< mkWeakThreadId t pure c runAgentThreads c - | backgroundMode = subscriber c - | otherwise = raceAny_ [subscriber c, runNtfSupervisor c, cleanupManager c] + | backgroundMode = run c "subscriber" $ subscriber c + | otherwise = + raceAny_ + [ run c "subscriber" $ subscriber c, + run c "runNtfSupervisor" $ runNtfSupervisor c, + run c "cleanupManager" $ cleanupManager c + ] + run AgentClient {subQ, acThread} name a = + a `E.catchAny` \e -> whenM (isJust <$> readTVarIO acThread) $ do + logError $ "Agent thread " <> name <> " crashed: " <> tshow e + atomically $ writeTBQueue subQ ("", "", APC SAEConn $ ERR $ CRITICAL True $ show e) disconnectAgentClient :: MonadUnliftIO m => AgentClient -> m () disconnectAgentClient c@AgentClient {agentEnv = Env {ntfSupervisor = ns, xftpAgent = xa}} = do @@ -209,6 +222,14 @@ disconnectAgentClient c@AgentClient {agentEnv = Env {ntfSupervisor = ns, xftpAge closeXFTPAgent xa logConnection c False +-- only used in the tests +disposeAgentClient :: MonadUnliftIO m => AgentClient -> m () +disposeAgentClient c@AgentClient {acThread, agentEnv = Env {store}} = do + t_ <- atomically (swapTVar acThread Nothing) $>>= (liftIO . deRefWeak) + disconnectAgentClient c + mapM_ killThread t_ + liftIO $ closeSQLiteStore store + resumeAgentClient :: MonadIO m => AgentClient -> m () resumeAgentClient c = atomically $ writeTVar (active c) True @@ -1916,9 +1937,11 @@ cleanupManager c@AgentClient {subQ} = do where run :: forall e. AEntityI e => (AgentErrorType -> ACommand 'Agent e) -> ExceptT AgentErrorType m () -> m () run err a = do - void . runExceptT $ a `catchAgentError` (notify "" . err) + waitActive . runExceptT $ a `catchAgentError` (notify "" . err) 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) deleteConns = withLock (deleteLock c) "cleanupManager" $ do void $ withStore' c getDeletedConnIds >>= deleteDeletedConns c diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index ec5400b73..834026e6a 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -129,7 +129,7 @@ module Simplex.Messaging.Agent.Client where import Control.Applicative ((<|>)) -import Control.Concurrent (forkIO, threadDelay) +import Control.Concurrent (ThreadId, forkIO, threadDelay) import Control.Concurrent.Async (Async, uninterruptibleCancel) import Control.Concurrent.STM (retry, throwSTM) import Control.Exception (AsyncException (..)) @@ -225,6 +225,7 @@ import qualified Simplex.Messaging.TMap as TM import Simplex.Messaging.Transport.Client (TransportHost) import Simplex.Messaging.Util import Simplex.Messaging.Version +import System.Mem.Weak (Weak) import System.Random (randomR) import UnliftIO (mapConcurrently, timeout) import UnliftIO.Async (async) @@ -252,7 +253,8 @@ type NtfTransportSession = TransportSession NtfResponse type XFTPTransportSession = TransportSession FileResponse data AgentClient = AgentClient - { active :: TVar Bool, + { acThread :: TVar (Maybe (Weak ThreadId)), + active :: TVar Bool, rcvQ :: TBQueue (ATransmission 'Client), subQ :: TBQueue (ATransmission 'Agent), msgQ :: TBQueue (ServerTransmission SMPVersion BrokerMsg), @@ -395,6 +397,7 @@ data AgentStatsKey = AgentStatsKey newAgentClient :: Int -> InitialAgentServers -> Env -> STM AgentClient newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = do let qSize = tbqSize $ config agentEnv + acThread <- newTVar Nothing active <- newTVar True rcvQ <- newTBQueue qSize subQ <- newTBQueue qSize @@ -428,7 +431,8 @@ newAgentClient clientId InitialAgentServers {smp, ntf, xftp, netCfg} agentEnv = agentStats <- TM.empty return AgentClient - { active, + { acThread, + active, rcvQ, subQ, msgQ, diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index e17f44df3..1fe002cb0 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -473,8 +473,8 @@ withAgentClientsCfg2 aCfg bCfg runTest = do a <- getSMPAgentClient' 1 aCfg initAgentServers testDB b <- getSMPAgentClient' 2 bCfg initAgentServers testDB2 runTest a b - disconnectAgentClient a - disconnectAgentClient b + disposeAgentClient a + disposeAgentClient b withAgentClients2 :: (AgentClient -> AgentClient -> IO ()) -> IO () withAgentClients2 = withAgentClientsCfg2 agentCfg agentCfg @@ -664,7 +664,7 @@ testAsyncInitiatingOffline :: HasCallStack => IO () testAsyncInitiatingOffline = withAgentClients2 $ \alice bob -> runRight_ $ do (bobId, cReq) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe - disconnectAgentClient alice + disposeAgentClient alice aliceId <- joinConnection bob 1 True cReq "bob's connInfo" SMSubscribe alice' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB subscribeConnection alice' bobId @@ -680,7 +680,7 @@ testAsyncJoiningOfflineBeforeActivation = withAgentClients2 $ \alice bob -> runRight_ $ do (bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe aliceId <- joinConnection bob 1 True qInfo "bob's connInfo" SMSubscribe - disconnectAgentClient bob + disposeAgentClient bob ("", _, CONF confId _ "bob's connInfo") <- get alice allowConnection alice bobId confId "alice's connInfo" bob' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB2 @@ -694,9 +694,9 @@ testAsyncBothOffline :: HasCallStack => IO () testAsyncBothOffline = withAgentClients2 $ \alice bob -> runRight_ $ do (bobId, cReq) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe - disconnectAgentClient alice + disposeAgentClient alice aliceId <- joinConnection bob 1 True cReq "bob's connInfo" SMSubscribe - disconnectAgentClient bob + disposeAgentClient bob alice' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB subscribeConnection alice' bobId ("", _, CONF confId _ "bob's connInfo") <- get alice' @@ -754,7 +754,7 @@ testAllowConnectionClientRestart t = do pure () threadDelay 100000 -- give time to enqueue confirmation (enqueueConfirmation) - disconnectAgentClient alice + disposeAgentClient alice alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB @@ -769,8 +769,8 @@ testAllowConnectionClientRestart t = do get bob ##> ("", aliceId, CON) exchangeGreetingsMsgId 4 alice2 bobId bob aliceId - disconnectAgentClient alice2 - disconnectAgentClient bob + disposeAgentClient alice2 + disposeAgentClient bob testIncreaseConnAgentVersion :: HasCallStack => ATransport -> IO () testIncreaseConnAgentVersion t = do @@ -786,7 +786,7 @@ testIncreaseConnAgentVersion t = do -- version doesn't increase if incompatible - disconnectAgentClient alice + disposeAgentClient alice alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 3} initAgentServers testDB runRight_ $ do @@ -797,7 +797,7 @@ testIncreaseConnAgentVersion t = do -- version increases if compatible - disconnectAgentClient bob + disposeAgentClient bob bob2 <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 3} initAgentServers testDB2 runRight_ $ do @@ -808,7 +808,7 @@ testIncreaseConnAgentVersion t = do -- version doesn't decrease, even if incompatible - disconnectAgentClient alice2 + disposeAgentClient alice2 alice3 <- getSMPAgentClient' 5 agentCfg {smpAgentVRange = \_ -> mkVersionRange 2 2} initAgentServers testDB runRight_ $ do @@ -817,7 +817,7 @@ testIncreaseConnAgentVersion t = do checkVersion alice3 bobId 3 checkVersion bob2 aliceId 3 - disconnectAgentClient bob2 + disposeAgentClient bob2 bob3 <- getSMPAgentClient' 6 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 1} initAgentServers testDB2 runRight_ $ do @@ -825,8 +825,8 @@ testIncreaseConnAgentVersion t = do exchangeGreetingsMsgId_ PQEncOff 12 alice3 bobId bob3 aliceId checkVersion alice3 bobId 3 checkVersion bob3 aliceId 3 - disconnectAgentClient alice3 - disconnectAgentClient bob3 + disposeAgentClient alice3 + disposeAgentClient bob3 checkVersion :: AgentClient -> ConnId -> Word16 -> ExceptT AgentErrorType IO () checkVersion c connId v = do @@ -847,9 +847,9 @@ testIncreaseConnAgentVersionMaxCompatible t = do -- version increases to max compatible - disconnectAgentClient alice + disposeAgentClient alice alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 3} initAgentServers testDB - disconnectAgentClient bob + disposeAgentClient bob bob2 <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = supportedSMPAgentVRange} initAgentServers testDB2 runRight_ $ do @@ -858,8 +858,8 @@ testIncreaseConnAgentVersionMaxCompatible t = do exchangeGreetingsMsgId_ PQEncOff 6 alice2 bobId bob2 aliceId checkVersion alice2 bobId 3 checkVersion bob2 aliceId 3 - disconnectAgentClient alice2 - disconnectAgentClient bob2 + disposeAgentClient alice2 + disposeAgentClient bob2 testIncreaseConnAgentVersionStartDifferentVersion :: HasCallStack => ATransport -> IO () testIncreaseConnAgentVersionStartDifferentVersion t = do @@ -875,7 +875,7 @@ testIncreaseConnAgentVersionStartDifferentVersion t = do -- version increases to max compatible - disconnectAgentClient alice + disposeAgentClient alice alice2 <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = \_ -> mkVersionRange 1 3} initAgentServers testDB runRight_ $ do @@ -883,8 +883,8 @@ testIncreaseConnAgentVersionStartDifferentVersion t = do exchangeGreetingsMsgId_ PQEncOff 6 alice2 bobId bob aliceId checkVersion alice2 bobId 3 checkVersion bob aliceId 3 - disconnectAgentClient alice2 - disconnectAgentClient bob + disposeAgentClient alice2 + disposeAgentClient bob testDeliverClientRestart :: HasCallStack => ATransport -> IO () testDeliverClientRestart t = do @@ -902,7 +902,7 @@ testDeliverClientRestart t = do 6 <- runRight $ sendMessage bob aliceId SMP.noMsgFlags "hello" - disconnectAgentClient bob + disposeAgentClient bob bob2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2 @@ -914,8 +914,8 @@ testDeliverClientRestart t = do get bob2 ##> ("", aliceId, SENT 6) get alice =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False - disconnectAgentClient alice - disconnectAgentClient bob2 + disposeAgentClient alice + disposeAgentClient bob2 testDuplicateMessage :: HasCallStack => ATransport -> IO () testDuplicateMessage t = do @@ -927,7 +927,7 @@ testDuplicateMessage t = do 4 <- sendMessage alice bobId SMP.noMsgFlags "hello" get alice ##> ("", bobId, SENT 4) get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False - disconnectAgentClient bob + disposeAgentClient bob -- if the agent user did not send ACK, the message will be delivered again bob1 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2 @@ -948,8 +948,8 @@ testDuplicateMessage t = do threadDelay 200000 Left (BROKER _ NETWORK) <- runExceptT $ ackMessage bob1 aliceId 5 Nothing - disconnectAgentClient alice - disconnectAgentClient bob1 + disposeAgentClient alice + disposeAgentClient bob1 alice2 <- getSMPAgentClient' 4 agentCfg initAgentServers testDB bob2 <- getSMPAgentClient' 5 agentCfg initAgentServers testDB2 @@ -964,8 +964,8 @@ testDuplicateMessage t = do 6 <- sendMessage alice2 bobId SMP.noMsgFlags "hello 3" get alice2 ##> ("", bobId, SENT 6) get bob2 =##> \case ("", c, Msg "hello 3") -> c == aliceId; _ -> False - disconnectAgentClient alice2 - disconnectAgentClient bob2 + disposeAgentClient alice2 + disposeAgentClient bob2 testSkippedMessages :: HasCallStack => ATransport -> IO () testSkippedMessages t = do @@ -979,7 +979,7 @@ testSkippedMessages t = do get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False ackMessage bob aliceId 4 Nothing - disconnectAgentClient bob + disposeAgentClient bob runRight_ $ do 5 <- sendMessage alice bobId SMP.noMsgFlags "hello 2" @@ -994,7 +994,7 @@ testSkippedMessages t = do nGet alice =##> \case ("", "", DOWN _ [c]) -> c == bobId; _ -> False threadDelay 200000 - disconnectAgentClient alice + disposeAgentClient alice alice2 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB bob2 <- getSMPAgentClient' 4 agentCfg initAgentServers testDB2 @@ -1013,8 +1013,8 @@ testSkippedMessages t = do get alice2 ##> ("", bobId, SENT 9) get bob2 =##> \case ("", c, Msg "hello 6") -> c == aliceId; _ -> False ackMessage bob2 aliceId 6 Nothing - disconnectAgentClient alice2 - disconnectAgentClient bob2 + disposeAgentClient alice2 + disposeAgentClient bob2 testExpireMessage :: HasCallStack => ATransport -> IO () testExpireMessage t = do @@ -1068,7 +1068,7 @@ testExpireMessageQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testP (aId, bId) <- runRight $ do (aId, bId) <- makeConnection a b liftIO $ threadDelay 500000 - disconnectAgentClient b + disposeAgentClient b 4 <- sendMessage a bId SMP.noMsgFlags "1" get a ##> ("", bId, SENT 4) 5 <- sendMessage a bId SMP.noMsgFlags "2" @@ -1092,7 +1092,7 @@ testExpireManyMessagesQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} (aId, bId) <- runRight $ do (aId, bId) <- makeConnection a b liftIO $ threadDelay 500000 - disconnectAgentClient b + disposeAgentClient b 4 <- sendMessage a bId SMP.noMsgFlags "1" get a ##> ("", bId, SENT 4) 5 <- sendMessage a bId SMP.noMsgFlags "2" @@ -1151,7 +1151,7 @@ setupDesynchronizedRatchet alice bob = do get alice =##> \case ("", c, Msg "hello 4") -> c == bobId; _ -> False ackMessage alice bobId 7 Nothing - disconnectAgentClient bob + disposeAgentClient bob -- importing database backup after progressing ratchet de-synchronizes ratchet liftIO $ renameFile (testDB2 <> ".bak") testDB2 @@ -1224,7 +1224,7 @@ testRatchetSyncClientRestart t = do ("", "", DOWN _ _) <- nGet bob2 ConnectionStats {ratchetSyncState} <- runRight $ synchronizeRatchet bob2 aliceId PQSupportOn False liftIO $ ratchetSyncState `shouldBe` RSStarted - disconnectAgentClient bob2 + disposeAgentClient bob2 bob3 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2 withSmpServerStoreMsgLogOn t testPort $ \_ -> do runRight_ $ do @@ -1235,9 +1235,9 @@ testRatchetSyncClientRestart t = do get alice =##> ratchetSyncP bobId RSOk get bob3 =##> ratchetSyncP aliceId RSOk exchangeGreetingsMsgIds alice bobId 12 bob3 aliceId 9 - disconnectAgentClient alice - disconnectAgentClient bob - disconnectAgentClient bob3 + disposeAgentClient alice + disposeAgentClient bob + disposeAgentClient bob3 testRatchetSyncSuspendForeground :: HasCallStack => ATransport -> IO () testRatchetSyncSuspendForeground t = do @@ -1269,9 +1269,9 @@ testRatchetSyncSuspendForeground t = do get alice =##> ratchetSyncP bobId RSOk get bob2 =##> ratchetSyncP aliceId RSOk exchangeGreetingsMsgIds alice bobId 12 bob2 aliceId 9 - disconnectAgentClient alice - disconnectAgentClient bob - disconnectAgentClient bob2 + disposeAgentClient alice + disposeAgentClient bob + disposeAgentClient bob2 testRatchetSyncSimultaneous :: HasCallStack => ATransport -> IO () testRatchetSyncSimultaneous t = do @@ -1302,9 +1302,9 @@ testRatchetSyncSimultaneous t = do get alice =##> ratchetSyncP bobId RSOk get bob2 =##> ratchetSyncP aliceId RSOk exchangeGreetingsMsgIds alice bobId 12 bob2 aliceId 9 - disconnectAgentClient alice - disconnectAgentClient bob - disconnectAgentClient bob2 + disposeAgentClient alice + disposeAgentClient bob + disposeAgentClient bob2 testOnlyCreatePull :: IO () testOnlyCreatePull = withAgentClients2 $ \alice bob -> runRight_ $ do @@ -1370,7 +1370,7 @@ testInactiveNoSubs t = do 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) - disconnectAgentClient alice + disposeAgentClient alice testInactiveWithSubs :: ATransport -> IO () testInactiveWithSubs t = do @@ -1382,7 +1382,7 @@ testInactiveWithSubs t = do liftIO $ threadDelay 1200000 -- and after 2 sec of inactivity no DOWN is sent as we have a live subscription liftIO $ timeout 1200000 (get alice) `shouldReturn` Nothing - disconnectAgentClient alice + disposeAgentClient alice testActiveClientNotDisconnected :: ATransport -> IO () testActiveClientNotDisconnected t = do @@ -1393,7 +1393,7 @@ testActiveClientNotDisconnected t = do runRight_ $ do (connId, _cReq) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe keepSubscribing alice connId ts - disconnectAgentClient alice + disposeAgentClient alice where keepSubscribing :: AgentClient -> ConnId -> SystemTime -> ExceptT AgentErrorType IO () keepSubscribing alice connId ts = do @@ -1512,8 +1512,8 @@ testBatchedSubscriptions nCreate nDel t = do delete b aIds' deleteFail a bIds' deleteFail b aIds' - disconnectAgentClient a - disconnectAgentClient b + disposeAgentClient a + disposeAgentClient b where subscribe :: AgentClient -> [ConnId] -> ExceptT AgentErrorType IO () subscribe c cs = do @@ -1598,14 +1598,14 @@ testAsyncCommandsRestore t = do alice <- getSMPAgentClient' 1 agentCfg initAgentServers testDB bobId <- runRight $ createConnectionAsync alice 1 "1" True SCMInvitation (IKNoPQ PQSupportOn) SMSubscribe liftIO $ noMessages alice "alice doesn't receive INV because server is down" - disconnectAgentClient alice + disposeAgentClient alice alice' <- liftIO $ getSMPAgentClient' 2 agentCfg initAgentServers testDB withSmpServerStoreLogOn t testPort $ \_ -> do runRight_ $ do subscribeConnection alice' bobId get alice' =##> \case ("1", _, INV _) -> True; _ -> False pure () - disconnectAgentClient alice' + disposeAgentClient alice' testAcceptContactAsync :: IO () testAcceptContactAsync = @@ -1663,7 +1663,7 @@ testDeleteConnectionAsync t = do get a =##> \case ("", c, DEL_CONN) -> c `elem` connIds; _ -> False get a =##> \case ("", c, DEL_CONN) -> c `elem` connIds; _ -> False liftIO $ noMessages a "nothing else should be delivered to alice" - disconnectAgentClient a + disposeAgentClient a testWaitDeliveryNoPending :: ATransport -> IO () testWaitDeliveryNoPending t = do @@ -1692,8 +1692,8 @@ testWaitDeliveryNoPending t = do liftIO $ noMessages alice "nothing else should be delivered to alice" liftIO $ noMessages bob "nothing else should be delivered to bob" - disconnectAgentClient alice - disconnectAgentClient bob + disposeAgentClient alice + disposeAgentClient bob where baseId = 3 msgId = subtract baseId @@ -1749,8 +1749,8 @@ testWaitDelivery t = do liftIO $ noMessages alice "nothing else should be delivered to alice" liftIO $ noMessages bob "nothing else should be delivered to bob" - disconnectAgentClient alice - disconnectAgentClient bob + disposeAgentClient alice + disposeAgentClient bob where baseId = 3 msgId = subtract baseId @@ -1795,8 +1795,8 @@ testWaitDeliveryAUTHErr t = do liftIO $ noMessages alice "nothing else should be delivered to alice" liftIO $ noMessages bob "nothing else should be delivered to bob" - disconnectAgentClient alice - disconnectAgentClient bob + disposeAgentClient alice + disposeAgentClient bob where baseId = 3 msgId = subtract baseId @@ -1838,8 +1838,8 @@ testWaitDeliveryTimeout t = do liftIO $ noMessages alice "nothing else should be delivered to alice" liftIO $ noMessages bob "nothing else should be delivered to bob" - disconnectAgentClient alice - disconnectAgentClient bob + disposeAgentClient alice + disposeAgentClient bob where baseId = 3 msgId = subtract baseId @@ -1887,8 +1887,8 @@ testWaitDeliveryTimeout2 t = do liftIO $ noMessages alice "nothing else should be delivered to alice" liftIO $ noMessages bob "nothing else should be delivered to bob" - disconnectAgentClient alice - disconnectAgentClient bob + disposeAgentClient alice + disposeAgentClient bob where baseId = 3 msgId = subtract baseId @@ -1931,8 +1931,8 @@ testJoinConnectionAsyncReplyError t = do get b ##> ("", aId, INFO "alice's connInfo") get b ##> ("", aId, CON) exchangeGreetings a bId b aId - disconnectAgentClient a - disconnectAgentClient b + disposeAgentClient a + disposeAgentClient b testUsers :: IO () testUsers = @@ -1995,8 +1995,8 @@ testSwitchConnection servers = do exchangeGreetingsMsgId 4 a bId b aId testFullSwitch a bId b aId 10 testFullSwitch a bId b aId 16 - disconnectAgentClient a - disconnectAgentClient b + disposeAgentClient a + disposeAgentClient b testFullSwitch :: AgentClient -> ByteString -> AgentClient -> ByteString -> Int64 -> ExceptT AgentErrorType IO () testFullSwitch a bId b aId msgId = do @@ -2077,7 +2077,7 @@ testSwitchAsync servers = do withB = withAgent 2 agentCfg servers testDB2 withAgent :: Int -> AgentConfig -> InitialAgentServers -> FilePath -> (AgentClient -> IO a) -> IO a -withAgent clientId cfg' servers dbPath = bracket (getSMPAgentClient' clientId cfg' servers dbPath) disconnectAgentClient +withAgent clientId cfg' servers dbPath = bracket (getSMPAgentClient' clientId cfg' servers dbPath) disposeAgentClient sessionSubscribe :: (forall a. (AgentClient -> IO a) -> IO a) -> [ConnId] -> (AgentClient -> ExceptT AgentErrorType IO ()) -> IO () sessionSubscribe withC connIds a = @@ -2095,7 +2095,7 @@ testSwitchDelete servers = do runRight_ $ do (aId, bId) <- makeConnection a b exchangeGreetingsMsgId 4 a bId b aId - disconnectAgentClient b + disposeAgentClient b stats <- switchConnectionAsync a "" bId liftIO $ rcvSwchStatuses' stats `shouldMatchList` [Just RSSwitchStarted] phaseRcv a bId SPStarted [Just RSSendingQADD, Nothing] @@ -2104,8 +2104,8 @@ testSwitchDelete servers = do get a =##> \case ("", c, DEL_RCVQ _ _ Nothing) -> c == bId; _ -> False get a =##> \case ("", c, DEL_CONN) -> c == bId; _ -> False liftIO $ noMessages a "nothing else should be delivered to alice" - disconnectAgentClient a - disconnectAgentClient b + disposeAgentClient a + disposeAgentClient b testAbortSwitchStarted :: HasCallStack => InitialAgentServers -> IO () testAbortSwitchStarted servers = do @@ -2393,8 +2393,8 @@ testCreateQueueAuth srvVersion clnt1 clnt2 = do get b ##> ("", aId, CON) exchangeGreetings a bId b aId pure 2 - disconnectAgentClient a - disconnectAgentClient b + disposeAgentClient a + disposeAgentClient b pure r where getClient clientId (clntAuth, clntVersion) db = @@ -2458,8 +2458,8 @@ testDeliveryReceiptsVersion t = do liftIO $ noMessages b "no delivery receipt (unsupported version)" pure (aId, bId) - disconnectAgentClient a - disconnectAgentClient b + disposeAgentClient a + disposeAgentClient b a' <- getSMPAgentClient' 3 agentCfg {smpAgentVRange = supportedSMPAgentVRange} initAgentServers testDB b' <- getSMPAgentClient' 4 agentCfg {smpAgentVRange = supportedSMPAgentVRange} initAgentServers testDB2 @@ -2487,8 +2487,8 @@ testDeliveryReceiptsVersion t = do ackMessage b' aId 12 $ Just "" get a' =##> \case ("", c, Rcvd 12) -> c == bId; _ -> False ackMessage a' bId 13 Nothing - disconnectAgentClient a' - disconnectAgentClient b' + disposeAgentClient a' + disposeAgentClient b' testDeliveryReceiptsConcurrent :: HasCallStack => ATransport -> IO () testDeliveryReceiptsConcurrent t = @@ -2626,7 +2626,7 @@ testServerMultipleIdentities = exchangeGreetings alice bobId bob aliceId -- this saves queue with second server identity Left (BROKER _ NETWORK) <- runExceptT $ joinConnection bob 1 True secondIdentityCReq "bob's connInfo" SMSubscribe - disconnectAgentClient bob + disposeAgentClient bob bob' <- liftIO $ getSMPAgentClient' 3 agentCfg initAgentServers testDB2 subscribeConnection bob' aliceId exchangeGreetingsMsgId 6 alice bobId bob' aliceId diff --git a/tests/AgentTests/NotificationTests.hs b/tests/AgentTests/NotificationTests.hs index d8354efed..18fccbf52 100644 --- a/tests/AgentTests/NotificationTests.hs +++ b/tests/AgentTests/NotificationTests.hs @@ -179,7 +179,7 @@ testNotificationToken APNSMockServer {apnsQ} = do deleteNtfToken a tkn -- agent deleted this token Left (CMD PROHIBITED) <- tryE $ checkNtfToken a tkn - disconnectAgentClient a + disposeAgentClient a (.->) :: J.Value -> J.Key -> ExceptT AgentErrorType IO ByteString v .-> key = do @@ -211,7 +211,7 @@ testNtfTokenRepeatRegistration APNSMockServer {apnsQ} = do -- can still use the first verification code, it is the same after decryption verifyNtfToken a tkn nonce verification NTActive <- checkNtfToken a tkn - disconnectAgentClient a + disposeAgentClient a testNtfTokenSecondRegistration :: APNSMockServer -> IO () testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do @@ -247,8 +247,8 @@ testNtfTokenSecondRegistration APNSMockServer {apnsQ} = do Left (NTF AUTH) <- tryE $ checkNtfToken a tkn -- and the second is active NTActive <- checkNtfToken a' tkn - disconnectAgentClient a - disconnectAgentClient a' + disposeAgentClient a + disposeAgentClient a' testNtfTokenServerRestart :: ATransport -> APNSMockServer -> IO () testNtfTokenServerRestart t APNSMockServer {apnsQ} = do @@ -262,7 +262,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do 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 threadDelay 1000000 - disconnectAgentClient a + disposeAgentClient a a' <- getSMPAgentClient' 2 agentCfg initAgentServers testDB -- server stopped before token is verified, so now the attempt to verify it will return AUTH error but re-register token, -- so that repeat verification happens without restarting the clients, when notification arrives @@ -277,7 +277,7 @@ testNtfTokenServerRestart t APNSMockServer {apnsQ} = do liftIO $ sendApnsResponse' APNSRespOk verifyNtfToken a' tkn nonce' verification' NTActive <- checkNtfToken a' tkn - disconnectAgentClient a' + disposeAgentClient a' getTestNtfTokenPort :: (MonadUnliftIO m, MonadError AgentErrorType m) => AgentClient -> m String getTestNtfTokenPort a = @@ -319,7 +319,7 @@ testNtfTokenChangeServers t APNSMockServer {apnsQ} = NTActive <- checkNtfToken a tkn setNtfServers a [testNtfServer2] NTActive <- checkNtfToken a tkn -- still works on old server - disconnectAgentClient a + disposeAgentClient a pure tkn threadDelay 1000000 @@ -386,7 +386,7 @@ testNotificationSubscriptionExistingConnection APNSMockServer {apnsQ} alice@Agen runRight_ $ do (_, [SMPMsgMeta {msgFlags = MsgFlags True}]) <- getNotificationMessage aliceNtf nonce message pure () - disconnectAgentClient aliceNtf + disposeAgentClient aliceNtf runRight_ $ do get alice =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False @@ -517,8 +517,8 @@ testChangeNotificationsMode APNSMockServer {apnsQ} = do ackMessage alice bobId (baseId + 5) Nothing -- no notifications should follow noNotification apnsQ - disconnectAgentClient alice - disconnectAgentClient bob + disposeAgentClient alice + disposeAgentClient bob where baseId = 3 msgId = subtract baseId @@ -546,7 +546,7 @@ testChangeToken APNSMockServer {apnsQ} = do get alice =##> \case ("", c, Msg "hello") -> c == bobId; _ -> False ackMessage alice bobId (baseId + 1) Nothing pure (aliceId, bobId) - disconnectAgentClient alice + disposeAgentClient alice alice1 <- getSMPAgentClient' 3 agentCfg initAgentServers testDB runRight_ $ do @@ -562,8 +562,8 @@ testChangeToken APNSMockServer {apnsQ} = do ackMessage alice1 bobId (baseId + 2) Nothing -- no notifications should follow noNotification apnsQ - disconnectAgentClient alice1 - disconnectAgentClient bob + disposeAgentClient alice1 + disposeAgentClient bob where baseId = 3 msgId = subtract baseId @@ -593,8 +593,8 @@ testNotificationsStoreLog t APNSMockServer {apnsQ} = do void $ messageNotificationData alice apnsQ get alice =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False liftIO $ killThread threadId - disconnectAgentClient alice - disconnectAgentClient bob + disposeAgentClient alice + disposeAgentClient bob testNotificationsSMPRestart :: ATransport -> APNSMockServer -> IO () testNotificationsSMPRestart t APNSMockServer {apnsQ} = do @@ -625,8 +625,8 @@ testNotificationsSMPRestart t APNSMockServer {apnsQ} = do _ <- messageNotificationData alice apnsQ get alice =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False liftIO $ killThread threadId - disconnectAgentClient alice - disconnectAgentClient bob + disposeAgentClient alice + disposeAgentClient bob testNotificationsSMPRestartBatch :: Int -> ATransport -> APNSMockServer -> IO () testNotificationsSMPRestartBatch n t APNSMockServer {apnsQ} = do @@ -666,8 +666,8 @@ testNotificationsSMPRestartBatch n t APNSMockServer {apnsQ} = do get b ##> ("", aliceId, SENT msgId) _ <- messageNotificationData a apnsQ get a =##> \case ("", c, Msg "hello again") -> c == bobId; _ -> False - disconnectAgentClient a - disconnectAgentClient b + disposeAgentClient a + disposeAgentClient b where runServers :: ExceptT AgentErrorType IO a -> IO a runServers a = do @@ -697,8 +697,8 @@ testSwitchNotifications servers APNSMockServer {apnsQ} = do switchComplete a bId b aId liftIO $ threadDelay 500000 testMessage "hello again" - disconnectAgentClient a - disconnectAgentClient b + disposeAgentClient a + disposeAgentClient b testNotificationsOldToken :: APNSMockServer -> IO () testNotificationsOldToken APNSMockServer {apnsQ} = do @@ -721,9 +721,9 @@ testNotificationsOldToken APNSMockServer {apnsQ} = do (acId, caId) <- makeConnection a c let testMessageAC = testMessage_ apnsQ a acId c caId testMessageAC "greetings" - disconnectAgentClient a - disconnectAgentClient b - disconnectAgentClient c + disposeAgentClient a + disposeAgentClient b + disposeAgentClient c testNotificationsNewToken :: APNSMockServer -> ThreadId -> IO () testNotificationsNewToken APNSMockServer {apnsQ} oldNtf = do @@ -749,9 +749,9 @@ testNotificationsNewToken APNSMockServer {apnsQ} oldNtf = do (acId, caId) <- makeConnection a c let testMessageAC = testMessage_ apnsQ a acId c caId testMessageAC "greetings" - disconnectAgentClient a - disconnectAgentClient b - disconnectAgentClient c + disposeAgentClient a + disposeAgentClient b + disposeAgentClient c testMessage_ :: HasCallStack => TBQueue APNSMockRequest -> AgentClient -> ConnId -> AgentClient -> ConnId -> SMP.MsgBody -> ExceptT AgentErrorType IO () testMessage_ apnsQ a aId b bId msg = do diff --git a/tests/XFTPAgent.hs b/tests/XFTPAgent.hs index 999746858..59c668d66 100644 --- a/tests/XFTPAgent.hs +++ b/tests/XFTPAgent.hs @@ -24,7 +24,7 @@ import Simplex.FileTransfer.Description (FileDescription (..), FileDescriptionUR import Simplex.FileTransfer.Protocol (FileParty (..)) import Simplex.FileTransfer.Transport (XFTPErrorType (AUTH)) import Simplex.FileTransfer.Server.Env (XFTPServerConfig (..)) -import Simplex.Messaging.Agent (AgentClient, disconnectAgentClient, testProtocolServer, xftpDeleteRcvFile, xftpDeleteSndFileInternal, xftpDeleteSndFileRemote, xftpReceiveFile, xftpSendDescription, xftpSendFile, xftpStartWorkers) +import Simplex.Messaging.Agent (AgentClient, disposeAgentClient, testProtocolServer, xftpDeleteRcvFile, xftpDeleteSndFileInternal, xftpDeleteSndFileRemote, xftpReceiveFile, xftpSendDescription, xftpSendFile, xftpStartWorkers) import Simplex.Messaging.Agent.Client (ProtocolTestFailure (..), ProtocolTestStep (..)) import Simplex.Messaging.Agent.Protocol (ACommand (..), AgentErrorType (..), BrokerErrorType (..), RcvFileId, SndFileId, noAuthSrv) import qualified Simplex.Messaging.Crypto as C @@ -115,7 +115,7 @@ testXFTPAgentSendReceive = withXFTPServer $ do runRight_ $ do rfId <- testReceive rcp rfd originalFilePath xftpDeleteRcvFile rcp rfId - disconnectAgentClient rcp + disposeAgentClient rcp testXFTPAgentSendReceiveEncrypted :: HasCallStack => IO () testXFTPAgentSendReceiveEncrypted = withXFTPServer $ do @@ -139,7 +139,7 @@ testXFTPAgentSendReceiveEncrypted = withXFTPServer $ do runRight_ $ do rfId <- testReceiveCF rcp rfd cfArgs originalFilePath xftpDeleteRcvFile rcp rfId - disconnectAgentClient rcp + disposeAgentClient rcp testXFTPAgentSendReceiveRedirect :: HasCallStack => IO () testXFTPAgentSendReceiveRedirect = withXFTPServer $ do @@ -173,7 +173,7 @@ testXFTPAgentSendReceiveRedirect = withXFTPServer $ do case strDecode uri of Left err -> fail err Right ok -> ok `shouldBe` fileDescriptionURI vfdRedirect - disconnectAgentClient sndr + disposeAgentClient sndr --- recipient rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2 FileDescriptionURI {description} <- either fail pure $ strDecode uri @@ -190,7 +190,7 @@ testXFTPAgentSendReceiveRedirect = withXFTPServer $ do rfGet rcp >>= \case (_, _, RFDONE out) -> pure out r -> error $ "Expected RFDONE, got " <> show r - disconnectAgentClient rcp + disposeAgentClient rcp inBytes <- B.readFile filePathIn B.readFile out `shouldReturn` inBytes @@ -215,7 +215,7 @@ testXFTPAgentSendReceiveNoRedirect = withXFTPServer $ do case strDecode uri of Left err -> fail err Right ok -> ok `shouldBe` fileDescriptionURI vfdDirect - disconnectAgentClient sndr + disposeAgentClient sndr --- recipient rcp <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2 FileDescriptionURI {description} <- either fail pure $ strDecode uri @@ -230,7 +230,7 @@ testXFTPAgentSendReceiveNoRedirect = withXFTPServer $ do rfGet rcp >>= \case (_, _, RFDONE out) -> pure out r -> error $ "Expected RFDONE, got " <> show r - disconnectAgentClient rcp + disposeAgentClient rcp inBytes <- B.readFile filePathIn B.readFile out `shouldReturn` inBytes @@ -303,7 +303,7 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do rfId <- xftpReceiveFile rcp 1 rfd Nothing liftIO $ timeout 300000 (get rcp) `shouldReturn` Nothing -- wait for worker attempt pure rfId - disconnectAgentClient rcp + disposeAgentClient rcp [prefixDir] <- listDirectory recipientFiles let tmpPath = recipientFiles prefixDir "xftp.encrypted" @@ -315,7 +315,7 @@ testXFTPAgentReceiveRestore = withGlobalLogging logCfgNoLogs $ do runRight_ $ xftpStartWorkers rcp' (Just recipientFiles) ("", rfId', RFPROG _ _) <- rfGet rcp' liftIO $ rfId' `shouldBe` rfId - disconnectAgentClient rcp' + disposeAgentClient rcp' threadDelay 100000 @@ -351,7 +351,7 @@ testXFTPAgentReceiveCleanup = withGlobalLogging logCfgNoLogs $ do rfId <- xftpReceiveFile rcp 1 rfd Nothing liftIO $ timeout 300000 (get rcp) `shouldReturn` Nothing -- wait for worker attempt pure rfId - disconnectAgentClient rcp + disposeAgentClient rcp [prefixDir] <- listDirectory recipientFiles let tmpPath = recipientFiles prefixDir "xftp.encrypted" @@ -378,7 +378,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do sfId <- xftpSendFile sndr 1 (CF.plain filePath) 2 liftIO $ timeout 1000000 (get sndr) `shouldReturn` Nothing -- wait for worker to encrypt and attempt to create file pure sfId - disconnectAgentClient sndr + disposeAgentClient sndr dirEntries <- listDirectory senderFiles let prefixDir = fromJust $ find (isSuffixOf "_snd.xftp") dirEntries @@ -393,7 +393,7 @@ testXFTPAgentSendRestore = withGlobalLogging logCfgNoLogs $ do runRight_ $ xftpStartWorkers sndr' (Just senderFiles) ("", sfId', SFPROG _ _) <- sfGet sndr' liftIO $ sfId' `shouldBe` sfId - disconnectAgentClient sndr' + disposeAgentClient sndr' threadDelay 100000 @@ -430,7 +430,7 @@ testXFTPAgentSendCleanup = withGlobalLogging logCfgNoLogs $ do (_, _, SFPROG _ _) <- sfGet sndr pure () pure sfId - disconnectAgentClient sndr + disposeAgentClient sndr pure sfId dirEntries <- listDirectory senderFiles @@ -473,7 +473,7 @@ testXFTPAgentDelete = withGlobalLogging logCfgNoLogs $ xftpDeleteSndFileRemote sndr 1 sfId sndDescr Nothing <- liftIO $ 100000 `timeout` sfGet sndr pure () - disconnectAgentClient rcp1 + disposeAgentClient rcp1 threadDelay 1000000 length <$> listDirectory xftpServerFiles `shouldReturn` 0 @@ -499,8 +499,8 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do rcp1 <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2 runRight_ . void $ testReceive rcp1 rfd1 filePath - disconnectAgentClient rcp1 - disconnectAgentClient sndr + disposeAgentClient rcp1 + disposeAgentClient sndr pure (sfId, sndDescr, rfd2) -- delete file - should not succeed with server down @@ -509,7 +509,7 @@ testXFTPAgentDeleteRestore = withGlobalLogging logCfgNoLogs $ do xftpStartWorkers sndr (Just senderFiles) xftpDeleteSndFileRemote sndr 1 sfId sndDescr liftIO $ timeout 300000 (get sndr) `shouldReturn` Nothing -- wait for worker attempt - disconnectAgentClient sndr + disposeAgentClient sndr threadDelay 300000 length <$> listDirectory xftpServerFiles `shouldReturn` 6