From baf2c470658a675f383a0aa628f37b39337f41d9 Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin Date: Fri, 19 Jan 2024 17:21:25 +0000 Subject: [PATCH] agent: expire messages failed after quota exceeded after 7 days, and expire multiple messages at once (#973) * agent: expire messages failed after quota exceeded after 7 days, and expire multiple messages at once * fix, test * refactor * catch in loop --- src/Simplex/Messaging/Agent.hs | 29 +++-- src/Simplex/Messaging/Agent/Env/SQLite.hs | 9 +- src/Simplex/Messaging/Agent/Protocol.hs | 15 ++- src/Simplex/Messaging/Agent/Store/SQLite.hs | 15 +++ tests/AgentTests/FunctionalAPITests.hs | 122 ++++++++++++++++++-- tests/SMPAgentClient.hs | 8 +- 6 files changed, 170 insertions(+), 28 deletions(-) diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index bcec478fd..36dcee49d 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -1142,7 +1142,7 @@ submitPendingMsg c cData sq = do runSmpQueueMsgDelivery :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> (Worker, TMVar ()) -> m () runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, duplexHandshake} sq (Worker {doWork}, qLock) = do - ri <- asks $ messageRetryInterval . config + AgentConfig {messageRetryInterval = ri, messageTimeout, helloTimeout, quotaExceededTimeout} <- asks config forever $ do atomically $ endAgentOperation c AOSndNetwork waitForWork doWork @@ -1166,7 +1166,9 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl SMP SMP.QUOTA -> case msgType of AM_CONN_INFO -> connError msgId NOT_AVAILABLE AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE - _ -> retrySndMsg RISlow + _ -> do + expireTs <- addUTCTime (-quotaExceededTimeout) <$> liftIO getCurrentTime + if internalTs < expireTs then notifyDelMsgs msgId e expireTs else retrySndMsg RISlow SMP SMP.AUTH -> case msgType of AM_CONN_INFO -> connError msgId NOT_AVAILABLE AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE @@ -1175,8 +1177,11 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl -- in duplexHandshake mode (v2) HELLO is only sent once, without retrying, -- because the queue must be secured by the time the confirmation or the first HELLO is received | duplexHandshake == Just True -> connErr - | otherwise -> - ifM (msgExpired helloTimeout) connErr (retrySndMsg RIFast) + -- otherwise branch is not used in clients with v2+ of agent protocol (since June 2022) + -- TODO remove in v6 + | otherwise -> do + expireTs <- addUTCTime (-helloTimeout) <$> liftIO getCurrentTime + if internalTs < expireTs then connErr else retrySndMsg RIFast where connErr = case rq_ of -- party initiating connection @@ -1196,14 +1201,11 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl -- for other operations BROKER HOST is treated as a permanent error (e.g., when connecting to the server), -- the message sending would be retried | temporaryOrHostError e -> do - let timeoutSel = if msgType == AM_HELLO_ then helloTimeout else messageTimeout - ifM (msgExpired timeoutSel) (notifyDel msgId err) (retrySndMsg RIFast) + let msgTimeout = if msgType == AM_HELLO_ then helloTimeout else messageTimeout + expireTs <- addUTCTime (-msgTimeout) <$> liftIO getCurrentTime + if internalTs < expireTs then notifyDelMsgs msgId e expireTs else retrySndMsg RIFast | otherwise -> notifyDel msgId err where - msgExpired timeoutSel = do - msgTimeout <- asks $ timeoutSel . config - currentTime <- liftIO getCurrentTime - pure $ diffUTCTime currentTime internalTs > msgTimeout retrySndMsg riMode = do withStore' c $ \db -> updatePendingMsgRIState db connId msgId riState retrySndOp c $ loop riMode @@ -1279,6 +1281,13 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl when (isJust rq_) $ removeConfirmations db connId unless (duplexHandshake == Just True) . void $ enqueueMessage c cData sq SMP.noMsgFlags HELLO where + notifyDelMsgs :: InternalId -> AgentErrorType -> UTCTime -> m () + notifyDelMsgs msgId err expireTs = do + notifyDel msgId $ MERR (unId msgId) err + msgIds_ <- withStore' c $ \db -> getExpiredSndMessages db connId sq expireTs + forM_ (L.nonEmpty msgIds_) $ \msgIds -> do + notify $ MERRS (L.map unId msgIds) err + withStore' c $ \db -> forM_ msgIds $ \msgId' -> deleteSndMsgDelivery db connId sq msgId' False `catchAll_` pure () delMsg :: InternalId -> m () delMsg = delMsgKeep False delMsgKeep :: Bool -> InternalId -> m () diff --git a/src/Simplex/Messaging/Agent/Env/SQLite.hs b/src/Simplex/Messaging/Agent/Env/SQLite.hs index 73588a39d..02d172d0e 100644 --- a/src/Simplex/Messaging/Agent/Env/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Env/SQLite.hs @@ -92,6 +92,7 @@ data AgentConfig = AgentConfig messageRetryInterval :: RetryInterval2, messageTimeout :: NominalDiffTime, helloTimeout :: NominalDiffTime, + quotaExceededTimeout :: NominalDiffTime, initialCleanupDelay :: Int64, cleanupInterval :: Int64, cleanupStepInterval :: Int, @@ -135,13 +136,10 @@ defaultMessageRetryInterval = maxInterval = 60_000000 }, riSlow = - -- TODO: these timeouts can be increased in v5.0 once most clients are updated - -- to resume sending on QCONT messages. - -- After that local message expiration period should be also increased. RetryInterval - { initialInterval = 60_000000, + { initialInterval = 180_000000, -- 3 minutes increaseAfter = 60_000000, - maxInterval = 3600_000000 -- 1 hour + maxInterval = 3 * 3600_000000 -- 3 hours } } @@ -159,6 +157,7 @@ defaultAgentConfig = messageRetryInterval = defaultMessageRetryInterval, messageTimeout = 2 * nominalDay, helloTimeout = 2 * nominalDay, + quotaExceededTimeout = 7 * nominalDay, initialCleanupDelay = 30 * 1000000, -- 30 seconds cleanupInterval = 30 * 60 * 1000000, -- 30 minutes cleanupStepInterval = 200000, -- 200ms diff --git a/src/Simplex/Messaging/Agent/Protocol.hs b/src/Simplex/Messaging/Agent/Protocol.hs index 7a9e4929e..54777bb64 100644 --- a/src/Simplex/Messaging/Agent/Protocol.hs +++ b/src/Simplex/Messaging/Agent/Protocol.hs @@ -337,6 +337,7 @@ data ACommand (p :: AParty) (e :: AEntity) where MID :: AgentMsgId -> ACommand Agent AEConn SENT :: AgentMsgId -> ACommand Agent AEConn MERR :: AgentMsgId -> AgentErrorType -> ACommand Agent AEConn + MERRS :: NonEmpty AgentMsgId -> AgentErrorType -> ACommand Agent AEConn MSG :: MsgMeta -> MsgFlags -> MsgBody -> ACommand Agent AEConn MSGNTF :: SMPMsgMeta -> ACommand Agent AEConn ACK :: AgentMsgId -> Maybe MsgReceiptInfo -> ACommand Client AEConn @@ -398,6 +399,7 @@ data ACommandTag (p :: AParty) (e :: AEntity) where MID_ :: ACommandTag Agent AEConn SENT_ :: ACommandTag Agent AEConn MERR_ :: ACommandTag Agent AEConn + MERRS_ :: ACommandTag Agent AEConn MSG_ :: ACommandTag Agent AEConn MSGNTF_ :: ACommandTag Agent AEConn ACK_ :: ACommandTag Client AEConn @@ -452,6 +454,7 @@ aCommandTag = \case MID _ -> MID_ SENT _ -> SENT_ MERR {} -> MERR_ + MERRS {} -> MERRS_ MSG {} -> MSG_ MSGNTF {} -> MSGNTF_ ACK {} -> ACK_ @@ -1611,6 +1614,7 @@ instance StrEncoding ACmdTag where "MID" -> ct MID_ "SENT" -> ct SENT_ "MERR" -> ct MERR_ + "MERRS" -> ct MERRS_ "MSG" -> ct MSG_ "MSGNTF" -> ct MSGNTF_ "ACK" -> t ACK_ @@ -1667,6 +1671,7 @@ instance (APartyI p, AEntityI e) => StrEncoding (ACommandTag p e) where MID_ -> "MID" SENT_ -> "SENT" MERR_ -> "MERR" + MERRS_ -> "MERRS" MSG_ -> "MSG" MSGNTF_ -> "MSGNTF" ACK_ -> "ACK" @@ -1736,6 +1741,7 @@ commandP binaryP = MID_ -> s (MID <$> A.decimal) SENT_ -> s (SENT <$> A.decimal) MERR_ -> s (MERR <$> A.decimal <* A.space <*> strP) + MERRS_ -> s (MERRS <$> strP_ <*> strP) MSG_ -> s (MSG <$> strP <* A.space <*> smpP <* A.space <*> binaryP) MSGNTF_ -> s (MSGNTF <$> strP) RCVD_ -> s (RCVD <$> strP <* A.space <*> strP) @@ -1788,12 +1794,13 @@ serializeCommand = \case SWITCH dir phase srvs -> s (SWITCH_, dir, phase, srvs) RSYNC rrState cryptoErr cstats -> s (RSYNC_, rrState, cryptoErr, cstats) SEND msgFlags msgBody -> B.unwords [s SEND_, smpEncode msgFlags, serializeBinary msgBody] - MID mId -> s (MID_, Str $ bshow mId) - SENT mId -> s (SENT_, Str $ bshow mId) - MERR mId e -> s (MERR_, Str $ bshow mId, e) + MID mId -> s (MID_, mId) + SENT mId -> s (SENT_, mId) + MERR mId e -> s (MERR_, mId, e) + MERRS mIds e -> s (MERRS_, mIds, e) MSG msgMeta msgFlags msgBody -> B.unwords [s MSG_, s msgMeta, smpEncode msgFlags, serializeBinary msgBody] MSGNTF smpMsgMeta -> s (MSGNTF_, smpMsgMeta) - ACK mId rcptInfo_ -> s (ACK_, Str $ bshow mId) <> maybe "" (B.cons ' ' . serializeBinary) rcptInfo_ + ACK mId rcptInfo_ -> s (ACK_, mId) <> maybe "" (B.cons ' ' . serializeBinary) rcptInfo_ RCVD msgMeta rcpts -> s (RCVD_, msgMeta, rcpts) SWCH -> s SWCH_ OFF -> s OFF_ diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 18e8bf555..f84c68967 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -110,6 +110,7 @@ module Simplex.Messaging.Agent.Store.SQLite getPendingQueueMsg, updatePendingMsgRIState, deletePendingMsgs, + getExpiredSndMessages, setMsgUserAck, getRcvMsg, getLastMsg, @@ -1041,6 +1042,20 @@ deletePendingMsgs :: DB.Connection -> ConnId -> SndQueue -> IO () deletePendingMsgs db connId SndQueue {dbQueueId} = DB.execute db "DELETE FROM snd_message_deliveries WHERE conn_id = ? AND snd_queue_id = ?" (connId, dbQueueId) +getExpiredSndMessages :: DB.Connection -> ConnId -> SndQueue -> UTCTime -> IO [InternalId] +getExpiredSndMessages db connId SndQueue {dbQueueId} expireTs = + map fromOnly + <$> DB.query + db + [sql| + SELECT d.internal_id + FROM snd_message_deliveries d + JOIN messages m ON m.conn_id = d.conn_id AND m.internal_id = d.internal_id + WHERE d.conn_id = ? AND d.snd_queue_id = ? AND d.failed = 0 AND m.internal_ts < ? + ORDER BY d.internal_id ASC + |] + (connId, dbQueueId, expireTs) + setMsgUserAck :: DB.Connection -> ConnId -> InternalId -> IO (Either StoreError (RcvQueue, SMP.MsgId)) setMsgUserAck db connId agentMsgId = runExceptT $ do (dbRcvId, srvMsgId) <- diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index ed795aae7..94a994104 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -117,6 +117,9 @@ pGet c = do pattern Msg :: MsgBody -> ACommand 'Agent e pattern Msg msgBody <- MSG MsgMeta {integrity = MsgOk} _ msgBody +pattern MsgErr :: AgentMsgId -> MsgErrorType -> MsgBody -> ACommand 'Agent e +pattern MsgErr msgId err msgBody <- MSG MsgMeta {recipient = (msgId, _), integrity = MsgError err} _ msgBody + pattern Rcvd :: AgentMsgId -> ACommand 'Agent e pattern Rcvd agentMsgId <- RCVD MsgMeta {integrity = MsgOk} [MsgReceipt {agentMsgId, msgRcptStatus = MROk}] @@ -221,6 +224,11 @@ functionalAPITests t = do testDuplicateMessage t it "should report error via msg integrity on skipped messages" $ testSkippedMessages t + describe "message expiration" $ do + it "should expire one message" $ testExpireMessage t + it "should expire multiple messages" $ testExpireManyMessages t + it "should expire one message if quota is exceeded" $ testExpireMessageQuota t + it "should expire multiple messages if quota is exceeded" $ testExpireManyMessagesQuota t describe "Ratchet synchronization" $ do it "should report ratchet de-synchronization, synchronize ratchets" $ testRatchetSync t @@ -394,7 +402,7 @@ withAgentClients2 :: (AgentClient -> AgentClient -> IO ()) -> IO () withAgentClients2 = withAgentClientsCfg2 agentCfg agentCfg runAgentClientTest :: HasCallStack => AgentClient -> AgentClient -> AgentMsgId -> IO () -runAgentClientTest alice bob baseId = do +runAgentClientTest alice bob baseId = runRight_ $ do (bobId, qInfo) <- createConnection alice 1 True SCMInvitation Nothing SMSubscribe aliceId <- joinConnection bob 1 True qInfo "bob's connInfo" SMSubscribe @@ -455,7 +463,7 @@ testAgentClient3 = do ackMessage c aIdForC 5 Nothing runAgentClientContactTest :: HasCallStack => AgentClient -> AgentClient -> AgentMsgId -> IO () -runAgentClientContactTest alice bob baseId = do +runAgentClientContactTest alice bob baseId = runRight_ $ do (_, qInfo) <- createConnection alice 1 True SCMContact Nothing SMSubscribe aliceId <- joinConnection bob 1 True qInfo "bob's connInfo" SMSubscribe @@ -864,6 +872,102 @@ testSkippedMessages t = do disconnectAgentClient alice2 disconnectAgentClient bob2 +testExpireMessage :: HasCallStack => ATransport -> IO () +testExpireMessage t = do + a <- getSMPAgentClient' 1 agentCfg {messageTimeout = 1, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB + b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2 + (aId, bId) <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ makeConnection a b + nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False + nGet b =##> \case ("", "", DOWN _ [c]) -> c == aId; _ -> False + 4 <- runRight $ sendMessage a bId SMP.noMsgFlags "1" + threadDelay 1000000 + 5 <- runRight $ sendMessage a bId SMP.noMsgFlags "2" -- this won't expire + get a =##> \case ("", c, MERR 4 (BROKER _ e)) -> bId == c && (e == TIMEOUT || e == NETWORK); _ -> False + withSmpServerStoreLogOn t testPort $ \_ -> runRight_ $ do + withUP a bId $ \case ("", _, SENT 5) -> True; _ -> False + withUP b aId $ \case ("", _, MsgErr 4 (MsgSkipped 3 3) "2") -> True; _ -> False + ackMessage b aId 4 Nothing + +testExpireManyMessages :: HasCallStack => ATransport -> IO () +testExpireManyMessages t = do + a <- getSMPAgentClient' 1 agentCfg {messageTimeout = 1, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB + b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2 + (aId, bId) <- withSmpServerStoreLogOn t testPort $ \_ -> runRight $ makeConnection a b + runRight_ $ do + nGet a =##> \case ("", "", DOWN _ [c]) -> c == bId; _ -> False + nGet b =##> \case ("", "", DOWN _ [c]) -> c == aId; _ -> False + 4 <- sendMessage a bId SMP.noMsgFlags "1" + 5 <- sendMessage a bId SMP.noMsgFlags "2" + 6 <- sendMessage a bId SMP.noMsgFlags "3" + liftIO $ threadDelay 1000000 + 7 <- sendMessage a bId SMP.noMsgFlags "4" -- this won't expire + get a =##> \case ("", c, MERR 4 (BROKER _ e)) -> bId == c && (e == TIMEOUT || e == NETWORK); _ -> False + get a =##> \case ("", c, MERRS [5, 6] (BROKER _ e)) -> bId == c && (e == TIMEOUT || e == NETWORK); _ -> False + withSmpServerStoreLogOn t testPort $ \_ -> runRight_ $ do + withUP a bId $ \case ("", _, SENT 7) -> True; _ -> False + withUP b aId $ \case ("", _, MsgErr 4 (MsgSkipped 3 5) "4") -> True; _ -> False + ackMessage b aId 4 Nothing + +withUP :: AgentClient -> ConnId -> (AEntityTransmission 'AEConn -> Bool) -> ExceptT AgentErrorType IO () +withUP a bId p = + liftIO $ + getInAnyOrder + a + [ \case ("", "", APC SAENone (UP _ [c])) -> c == bId; _ -> False, + \case (corrId, c, APC SAEConn cmd) -> c == bId && p (corrId, c, cmd); _ -> False + ] + +testExpireMessageQuota :: HasCallStack => ATransport -> IO () +testExpireMessageQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testPort $ \_ -> do + a <- getSMPAgentClient' 1 agentCfg {quotaExceededTimeout = 1, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB + b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2 + (aId, bId) <- runRight $ do + (aId, bId) <- makeConnection a b + liftIO $ threadDelay 500000 + disconnectAgentClient b + 4 <- sendMessage a bId SMP.noMsgFlags "1" + get a ##> ("", bId, SENT 4) + 5 <- sendMessage a bId SMP.noMsgFlags "2" + liftIO $ threadDelay 1000000 + 6 <- sendMessage a bId SMP.noMsgFlags "3" -- this won't expire + get a =##> \case ("", c, MERR 5 (SMP QUOTA)) -> bId == c; _ -> False + pure (aId, bId) + b' <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2 + runRight_ $ do + subscribeConnection b' aId + get b' =##> \case ("", c, Msg "1") -> c == aId; _ -> False + ackMessage b' aId 4 Nothing + get a ##> ("", bId, SENT 6) + get b' =##> \case ("", c, MsgErr 6 (MsgSkipped 4 4) "3") -> c == aId; _ -> False + ackMessage b' aId 6 Nothing + +testExpireManyMessagesQuota :: HasCallStack => ATransport -> IO () +testExpireManyMessagesQuota t = withSmpServerConfigOn t cfg {msgQueueQuota = 1} testPort $ \_ -> do + a <- getSMPAgentClient' 1 agentCfg {quotaExceededTimeout = 1, messageRetryInterval = fastMessageRetryInterval} initAgentServers testDB + b <- getSMPAgentClient' 2 agentCfg initAgentServers testDB2 + (aId, bId) <- runRight $ do + (aId, bId) <- makeConnection a b + liftIO $ threadDelay 500000 + disconnectAgentClient b + 4 <- sendMessage a bId SMP.noMsgFlags "1" + get a ##> ("", bId, SENT 4) + 5 <- sendMessage a bId SMP.noMsgFlags "2" + 6 <- sendMessage a bId SMP.noMsgFlags "3" + 7 <- sendMessage a bId SMP.noMsgFlags "4" + liftIO $ threadDelay 1000000 + 8 <- sendMessage a bId SMP.noMsgFlags "5" -- this won't expire + get a =##> \case ("", c, MERR 5 (SMP QUOTA)) -> bId == c; _ -> False + get a =##> \case ("", c, MERRS [6, 7] (SMP QUOTA)) -> bId == c; _ -> False + pure (aId, bId) + b' <- getSMPAgentClient' 3 agentCfg initAgentServers testDB2 + runRight_ $ do + subscribeConnection b' aId + get b' =##> \case ("", c, Msg "1") -> c == aId; _ -> False + ackMessage b' aId 4 Nothing + get a ##> ("", bId, SENT 8) + get b' =##> \case ("", c, MsgErr 6 (MsgSkipped 4 6) "5") -> c == aId; _ -> False + ackMessage b' aId 6 Nothing + testRatchetSync :: HasCallStack => ATransport -> IO () testRatchetSync t = withAgentClients2 $ \alice bob -> withSmpServerStoreMsgLogOn t testPort $ \_ -> do @@ -1309,9 +1413,10 @@ testAsyncCommands = get alice ##> ("", bobId, SENT $ baseId + 2) get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False ackMessageAsync bob "4" aliceId (baseId + 1) Nothing - inAnyOrder (get bob) - [ \case {("4", _, OK) -> True; _ -> False}, - \case {("", c, Msg "how are you?") -> c == aliceId; _ -> False} + inAnyOrder + (get bob) + [ \case ("4", _, OK) -> True; _ -> False, + \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False ] ackMessageAsync bob "5" aliceId (baseId + 2) Nothing get bob =##> \case ("5", _, OK) -> True; _ -> False @@ -1321,9 +1426,10 @@ testAsyncCommands = get bob ##> ("", aliceId, SENT $ baseId + 4) get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False ackMessageAsync alice "6" bobId (baseId + 3) Nothing - inAnyOrder (get alice) - [ \case {("6", _, OK) -> True; _ -> False}, - \case {("", c, Msg "message 1") -> c == bobId; _ -> False} + inAnyOrder + (get alice) + [ \case ("6", _, OK) -> True; _ -> False, + \case ("", c, Msg "message 1") -> c == bobId; _ -> False ] ackMessageAsync alice "7" bobId (baseId + 4) Nothing get alice =##> \case ("7", _, OK) -> True; _ -> False diff --git a/tests/SMPAgentClient.hs b/tests/SMPAgentClient.hs index 48b0540c3..3a3c69f8c 100644 --- a/tests/SMPAgentClient.hs +++ b/tests/SMPAgentClient.hs @@ -206,7 +206,7 @@ agentCfg = -- database = testDB, smpCfg = defaultClientConfig {qSize = 1, defaultTransport = (testPort, transport @TLS), networkConfig}, ntfCfg = defaultClientConfig {qSize = 1, defaultTransport = (ntfTestPort, transport @TLS), networkConfig}, - reconnectInterval = defaultReconnectInterval {initialInterval = 50_000}, + reconnectInterval = fastRetryInterval, xftpNotifyErrsOnRetry = False, ntfWorkerDelay = 100, ntfSMPWorkerDelay = 100, @@ -217,6 +217,12 @@ agentCfg = where networkConfig = defaultNetworkConfig {tcpConnectTimeout = 3_000_000, tcpTimeout = 2_000_000} +fastRetryInterval :: RetryInterval +fastRetryInterval = defaultReconnectInterval {initialInterval = 50_000} + +fastMessageRetryInterval :: RetryInterval2 +fastMessageRetryInterval = RetryInterval2 {riFast = fastRetryInterval, riSlow = fastRetryInterval} + type AgentTestMonad m = (MonadUnliftIO m, MonadRandom m, MonadFail m) withSmpAgentThreadOn_ :: AgentTestMonad m => ATransport -> (ServiceName, ServiceName, FilePath) -> m () -> (ThreadId -> m a) -> m a