mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-10-23 15:12:26 +00:00
cease attempts to send HELLO after timeout (#298)
This commit is contained in:
1 parent
40efdf97de
commit
6e7089284e
6 files changed
+53
-17
No files matched your search
@@ -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
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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 ()
|
||||
|
||||
@@ -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 = ?
|
||||
|
||||
@@ -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"
|
||||
|
||||
Reference in new issue
Block a user