From 6e7089284e220c7fcfa7003063cf5d6b891a979b Mon Sep 17 00:00:00 2001 From: Efim Poberezkin <8711996+efim-poberezkin@users.noreply.github.com> Date: Tue, 18 Jan 2022 13:50:43 +0400 Subject: [PATCH] cease attempts to send HELLO after timeout (#298) --- src/Simplex/Messaging/Agent.hs | 23 +++++++++++++++------ src/Simplex/Messaging/Agent/Env/SQLite.hs | 22 +++++++++++++------- src/Simplex/Messaging/Agent/Protocol.hs | 4 ++++ src/Simplex/Messaging/Agent/Store.hs | 2 +- src/Simplex/Messaging/Agent/Store/SQLite.hs | 4 ++-- tests/AgentTests/FunctionalAPITests.hs | 15 +++++++++++++- 6 files changed, 53 insertions(+), 17 deletions(-) diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index a57a2e5c5..f79f1f68e 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -458,18 +458,27 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} connId sq = do withStore (\st -> E.try $ getPendingMsgData st connId msgId) >>= \case Left (e :: E.SomeException) -> notify $ MERR mId (INTERNAL $ show e) - Right (rq_, (msgType, msgBody)) -> + Right (rq_, (msgType, msgBody, internalTs)) -> withRetryInterval ri $ \loop -> tryError (sendAgentMessage c sq msgBody) >>= \case Left e -> do case e of SMP SMP.QUOTA -> loop SMP SMP.AUTH -> case msgType of - HELLO_ -> loop - REPLY_ -> notify (ERR e) >> delMsg msgId - A_MSG_ -> notify (MERR mId e) >> delMsg msgId - SMP (SMP.CMD _) -> notify (MERR mId e) >> delMsg msgId - SMP SMP.LARGE_MSG -> notify (MERR mId e) >> delMsg msgId + HELLO_ -> do + helloTimeout <- asks $ helloTimeout . config + currentTime <- liftIO getCurrentTime + if diffUTCTime currentTime internalTs > helloTimeout + then case rq_ of + -- party initiating connection + Just _ -> notifyDel msgId . ERR $ CONN NOT_AVAILABLE + -- party joining connection + _ -> notifyDel msgId . ERR $ CONN NOT_ACCEPTED + else loop + REPLY_ -> notifyDel msgId $ ERR e + A_MSG_ -> notifyDel msgId $ MERR mId e + SMP (SMP.CMD _) -> notifyDel msgId $ MERR mId e + SMP SMP.LARGE_MSG -> notifyDel msgId $ MERR mId e SMP {} -> notify (MERR mId e) >> loop _ -> loop Right () -> do @@ -491,6 +500,8 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} connId sq = do delMsg msgId = withStore $ \st -> deleteMsg st connId msgId notify :: ACommand 'Agent -> m () notify cmd = atomically $ writeTBQueue subQ ("", connId, cmd) + notifyDel :: InternalId -> ACommand 'Agent -> m () + notifyDel msgId cmd = notify cmd >> delMsg msgId ackMessage' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m () ackMessage' c connId msgId = do diff --git a/src/Simplex/Messaging/Agent/Env/SQLite.hs b/src/Simplex/Messaging/Agent/Env/SQLite.hs index 247072fbf..0e549d517 100644 --- a/src/Simplex/Messaging/Agent/Env/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Env/SQLite.hs @@ -4,11 +4,18 @@ {-# LANGUAGE NumericUnderscores #-} {-# OPTIONS_GHC -fno-warn-unticked-promoted-constructors #-} -module Simplex.Messaging.Agent.Env.SQLite where +module Simplex.Messaging.Agent.Env.SQLite + ( AgentConfig (..), + defaultAgentConfig, + Env (..), + newSMPAgentEnv, + ) +where import Control.Monad.IO.Unlift import Crypto.Random import Data.List.NonEmpty (NonEmpty) +import Data.Time.Clock (NominalDiffTime, nominalDay) import Network.Socket import Numeric.Natural import Simplex.Messaging.Agent.Protocol (SMPServer) @@ -30,14 +37,12 @@ data AgentConfig = AgentConfig dbPoolSize :: Int, smpCfg :: SMPClientConfig, reconnectInterval :: RetryInterval, + helloTimeout :: NominalDiffTime, caCertificateFile :: FilePath, privateKeyFile :: FilePath, certificateFile :: FilePath } -minute :: Int -minute = 60_000_000 - defaultAgentConfig :: AgentConfig defaultAgentConfig = AgentConfig @@ -51,16 +56,19 @@ defaultAgentConfig = smpCfg = smpDefaultConfig, reconnectInterval = RetryInterval - { initialInterval = 1_000_000, - increaseAfter = 10_000_000, - maxInterval = 10_000_000 + { initialInterval = second, + increaseAfter = 10 * second, + maxInterval = 10 * second }, + helloTimeout = 7 * nominalDay, -- CA certificate private key is not needed for initialization -- ! we do not generate these caCertificateFile = "/etc/opt/simplex-agent/ca.crt", privateKeyFile = "/etc/opt/simplex-agent/agent.key", certificateFile = "/etc/opt/simplex-agent/agent.crt" } + where + second = 1_000_000 data Env = Env { config :: AgentConfig, diff --git a/src/Simplex/Messaging/Agent/Protocol.hs b/src/Simplex/Messaging/Agent/Protocol.hs index b274c696d..24fc0fa50 100644 --- a/src/Simplex/Messaging/Agent/Protocol.hs +++ b/src/Simplex/Messaging/Agent/Protocol.hs @@ -663,6 +663,10 @@ data ConnectionErrorType DUPLICATE | -- | connection is simplex, but operation requires another queue SIMPLEX + | -- | connection not accepted on join HELLO after timeout + NOT_ACCEPTED + | -- | connection not available on reply HELLO after timeout + NOT_AVAILABLE deriving (Eq, Generic, Read, Show, Exception) -- | SMP server errors. diff --git a/src/Simplex/Messaging/Agent/Store.hs b/src/Simplex/Messaging/Agent/Store.hs index a664355f2..89afa2dba 100644 --- a/src/Simplex/Messaging/Agent/Store.hs +++ b/src/Simplex/Messaging/Agent/Store.hs @@ -62,7 +62,7 @@ class Monad m => MonadAgentStore s m where createRcvMsg :: s -> ConnId -> RcvMsgData -> m () updateSndIds :: s -> ConnId -> m (InternalId, InternalSndId, PrevSndMsgHash) createSndMsg :: s -> ConnId -> SndMsgData -> m () - getPendingMsgData :: s -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody)) + getPendingMsgData :: s -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody, InternalTs)) getPendingMsgs :: s -> ConnId -> m [InternalId] checkRcvMsg :: s -> ConnId -> InternalId -> m () deleteMsg :: s -> ConnId -> InternalId -> m () diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 7dc35bbb4..f215bbb75 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -440,7 +440,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto insertSndMsgDetails_ db connId sndMsgData updateHashSnd_ db connId sndMsgData - getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody)) + getPendingMsgData :: SQLiteStore -> ConnId -> InternalId -> m (Maybe RcvQueue, (AMsgType, MsgBody, InternalTs)) getPendingMsgData st connId msgId = liftIOEither . withTransaction st $ \db -> runExceptT $ do rq_ <- liftIO $ getRcvQueueByConnId_ db connId @@ -449,7 +449,7 @@ instance (MonadUnliftIO m, MonadError StoreError m) => MonadAgentStore SQLiteSto DB.query db [sql| - SELECT m.msg_type, m.msg_body + SELECT m.msg_type, m.msg_body, m.internal_ts FROM messages m JOIN snd_messages s ON s.conn_id = m.conn_id AND s.internal_id = m.internal_id WHERE m.conn_id = ? AND m.internal_id = ? diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 32bb6056c..116a38b55 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -12,7 +12,7 @@ import Control.Monad.IO.Unlift import SMPAgentClient import SMPClient (withSmpServer) import Simplex.Messaging.Agent -import Simplex.Messaging.Agent.Env.SQLite (dbFile) +import Simplex.Messaging.Agent.Env.SQLite (AgentConfig (..)) import Simplex.Messaging.Agent.Protocol import Simplex.Messaging.Protocol (ErrorType (..), MsgBody) import Simplex.Messaging.Transport (ATransport (..)) @@ -44,6 +44,8 @@ functionalAPITests t = do withSmpServer t testAsyncJoiningOfflineBeforeActivation it "should connect with both clients going offline" $ withSmpServer t testAsyncBothOffline + it "should notify after HELLO timeout" $ + withSmpServer t testAsyncHelloTimeout testAgentClient :: IO () testAgentClient = do @@ -146,6 +148,17 @@ testAsyncBothOffline = do exchangeGreetings alice' bobId bob' aliceId pure () +testAsyncHelloTimeout :: IO () +testAsyncHelloTimeout = do + alice <- getSMPAgentClient cfg + bob <- getSMPAgentClient cfg {dbFile = testDB2, helloTimeout = 1} + Right () <- runExceptT $ do + (_, cReq) <- createConnection alice SCMInvitation + disconnectAgentClient alice + aliceId <- joinConnection bob cReq "bob's connInfo" + get bob ##> ("", aliceId, ERR $ CONN NOT_ACCEPTED) + pure () + exchangeGreetings :: AgentClient -> ConnId -> AgentClient -> ConnId -> ExceptT AgentErrorType IO () exchangeGreetings alice bobId bob aliceId = do 4 <- sendMessage alice bobId "hello"