mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-08-29 09:58:41 +00:00
errs
This commit is contained in:
spaced4ndy
@@ -1331,7 +1331,7 @@ submitPendingMsg c cData sq = do
|
||||
void $ getDeliveryWorker True c cData sq
|
||||
|
||||
runSmpQueueMsgDelivery :: AgentClient -> ConnData -> SndQueue -> (Worker, TMVar ()) -> AM ()
|
||||
runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork}, qLock) = do
|
||||
runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq@SndQueue {userId, server} (Worker {doWork}, qLock) = do
|
||||
AgentConfig {messageRetryInterval = ri, messageTimeout, helloTimeout, quotaExceededTimeout} <- asks config
|
||||
forever $ do
|
||||
atomically $ endAgentOperation c AOSndNetwork
|
||||
@@ -1354,36 +1354,40 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork
|
||||
Left e -> do
|
||||
let err = if msgType == AM_A_MSG_ then MERR mId e else ERR e
|
||||
case e of
|
||||
SMP _ SMP.QUOTA -> case msgType of
|
||||
AM_CONN_INFO -> connError msgId NOT_AVAILABLE
|
||||
AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
|
||||
_ -> do
|
||||
expireTs <- addUTCTime (-quotaExceededTimeout) <$> liftIO getCurrentTime
|
||||
if internalTs < expireTs
|
||||
then notifyDelMsgs msgId e expireTs
|
||||
else do
|
||||
notify $ MWARN (unId msgId) e
|
||||
retrySndMsg RISlow
|
||||
SMP _ SMP.AUTH -> case msgType of
|
||||
AM_CONN_INFO -> connError msgId NOT_AVAILABLE
|
||||
AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
|
||||
AM_RATCHET_INFO -> connError msgId NOT_AVAILABLE
|
||||
-- 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
|
||||
AM_HELLO_ -> case rq_ of
|
||||
-- party initiating connection
|
||||
Just _ -> connError msgId NOT_AVAILABLE
|
||||
-- party joining connection
|
||||
_ -> connError msgId NOT_ACCEPTED
|
||||
AM_REPLY_ -> notifyDel msgId err
|
||||
AM_A_MSG_ -> notifyDel msgId err
|
||||
AM_A_RCVD_ -> notifyDel msgId err
|
||||
AM_QCONT_ -> notifyDel msgId err
|
||||
AM_QADD_ -> qError msgId "QADD: AUTH"
|
||||
AM_QKEY_ -> qError msgId "QKEY: AUTH"
|
||||
AM_QUSE_ -> qError msgId "QUSE: AUTH"
|
||||
AM_QTEST_ -> qError msgId "QTEST: AUTH"
|
||||
AM_EREADY_ -> notifyDel msgId err
|
||||
SMP _ SMP.QUOTA -> do
|
||||
atomically $ incSMPServerStat c userId server sentQuotaErrs 1
|
||||
case msgType of
|
||||
AM_CONN_INFO -> connError msgId NOT_AVAILABLE
|
||||
AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
|
||||
_ -> do
|
||||
expireTs <- addUTCTime (-quotaExceededTimeout) <$> liftIO getCurrentTime
|
||||
if internalTs < expireTs
|
||||
then notifyDelMsgs msgId e expireTs
|
||||
else do
|
||||
notify $ MWARN (unId msgId) e
|
||||
retrySndMsg RISlow
|
||||
SMP _ SMP.AUTH -> do
|
||||
atomically $ incSMPServerStat c userId server sentAuthErrs 1
|
||||
case msgType of
|
||||
AM_CONN_INFO -> connError msgId NOT_AVAILABLE
|
||||
AM_CONN_INFO_REPLY -> connError msgId NOT_AVAILABLE
|
||||
AM_RATCHET_INFO -> connError msgId NOT_AVAILABLE
|
||||
-- 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
|
||||
AM_HELLO_ -> case rq_ of
|
||||
-- party initiating connection
|
||||
Just _ -> connError msgId NOT_AVAILABLE
|
||||
-- party joining connection
|
||||
_ -> connError msgId NOT_ACCEPTED
|
||||
AM_REPLY_ -> notifyDel msgId err
|
||||
AM_A_MSG_ -> notifyDel msgId err
|
||||
AM_A_RCVD_ -> notifyDel msgId err
|
||||
AM_QCONT_ -> notifyDel msgId err
|
||||
AM_QADD_ -> qError msgId "QADD: AUTH"
|
||||
AM_QKEY_ -> qError msgId "QKEY: AUTH"
|
||||
AM_QUSE_ -> qError msgId "QUSE: AUTH"
|
||||
AM_QTEST_ -> qError msgId "QTEST: AUTH"
|
||||
AM_EREADY_ -> notifyDel msgId err
|
||||
_
|
||||
-- for other operations BROKER HOST is treated as a permanent error (e.g., when connecting to the server),
|
||||
-- the message sending would be retried
|
||||
@@ -1395,7 +1399,8 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork
|
||||
else do
|
||||
when (serverHostError e) $ notify $ MWARN (unId msgId) e
|
||||
retrySndMsg RIFast
|
||||
| otherwise -> notifyDel msgId err
|
||||
| otherwise -> do
|
||||
notifyDel msgId err
|
||||
where
|
||||
retrySndMsg riMode = do
|
||||
withStore' c $ \db -> updatePendingMsgRIState db connId msgId riState
|
||||
@@ -1472,6 +1477,7 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} ConnData {connId} sq (Worker {doWork
|
||||
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 ()
|
||||
atomically $ incSMPServerStat c userId server sentExpiredErrs (length msgIds_ + 1)
|
||||
delMsg :: InternalId -> AM ()
|
||||
delMsg = delMsgKeep False
|
||||
delMsgKeep :: Bool -> InternalId -> AM ()
|
||||
|
||||
@@ -139,6 +139,7 @@ module Simplex.Messaging.Agent.Client
|
||||
getNextServer,
|
||||
withUserServers,
|
||||
withNextSrv,
|
||||
incSMPServerStat,
|
||||
AgentWorkersDetails (..),
|
||||
getAgentWorkersDetails,
|
||||
AgentWorkersSummary (..),
|
||||
|
||||
@@ -19,8 +19,9 @@ data AgentSMPServerStats = AgentSMPServerStats
|
||||
sentAuthErrs :: TVar Int, -- send AUTH errors
|
||||
sentQuotaErrs :: TVar Int, -- send QUOTA permanent errors (message expired)
|
||||
sentExpiredErrs :: TVar Int, -- send expired errors
|
||||
sentDirectErrs :: TVar Int, -- other direct send permanent errors (excluding above)
|
||||
sentProxyErrs :: TVar Int, -- other proxy send permanent errors (excluding above)
|
||||
sentOtherErrs :: TVar Int, -- other send permanent errors (excluding above)
|
||||
-- sentDirectErrs :: TVar Int, -- other direct send permanent errors (excluding above)
|
||||
-- sentProxyErrs :: TVar Int, -- other proxy send permanent errors (excluding above)
|
||||
recvMsgs :: TVar Int, -- total messages received
|
||||
recvDuplicates :: TVar Int, -- duplicate messages received
|
||||
recvCryptoErrs :: TVar Int, -- message decryption errors
|
||||
@@ -42,8 +43,9 @@ data AgentSMPServerStatsData = AgentSMPServerStatsData
|
||||
_sentAuthErrs :: Int,
|
||||
_sentQuotaErrs :: Int,
|
||||
_sentExpiredErrs :: Int,
|
||||
_sentDirectErrs :: Int,
|
||||
_sentProxyErrs :: Int,
|
||||
_sentOtherErrs :: Int,
|
||||
-- _sentDirectErrs :: Int,
|
||||
-- _sentProxyErrs :: Int,
|
||||
_recvMsgs :: Int,
|
||||
_recvDuplicates :: Int,
|
||||
_recvCryptoErrs :: Int,
|
||||
@@ -67,8 +69,9 @@ newAgentSMPServerStats = do
|
||||
sentAuthErrs <- newTVar 0
|
||||
sentQuotaErrs <- newTVar 0
|
||||
sentExpiredErrs <- newTVar 0
|
||||
sentDirectErrs <- newTVar 0
|
||||
sentProxyErrs <- newTVar 0
|
||||
sentOtherErrs <- newTVar 0
|
||||
-- sentDirectErrs <- newTVar 0
|
||||
-- sentProxyErrs <- newTVar 0
|
||||
recvMsgs <- newTVar 0
|
||||
recvDuplicates <- newTVar 0
|
||||
recvCryptoErrs <- newTVar 0
|
||||
@@ -89,8 +92,9 @@ newAgentSMPServerStats = do
|
||||
sentAuthErrs,
|
||||
sentQuotaErrs,
|
||||
sentExpiredErrs,
|
||||
sentDirectErrs,
|
||||
sentProxyErrs,
|
||||
sentOtherErrs,
|
||||
-- sentDirectErrs,
|
||||
-- sentProxyErrs,
|
||||
recvMsgs,
|
||||
recvDuplicates,
|
||||
recvCryptoErrs,
|
||||
@@ -113,8 +117,9 @@ newAgentSMPServerStats' s = do
|
||||
sentAuthErrs <- newTVar $ _sentDirect s
|
||||
sentQuotaErrs <- newTVar $ _sentDirect s
|
||||
sentExpiredErrs <- newTVar $ _sentDirect s
|
||||
sentDirectErrs <- newTVar $ _sentDirect s
|
||||
sentProxyErrs <- newTVar $ _sentDirect s
|
||||
sentOtherErrs <- newTVar $ _sentDirect s
|
||||
-- sentDirectErrs <- newTVar $ _sentDirect s
|
||||
-- sentProxyErrs <- newTVar $ _sentDirect s
|
||||
recvMsgs <- newTVar $ _sentDirect s
|
||||
recvDuplicates <- newTVar $ _sentDirect s
|
||||
recvCryptoErrs <- newTVar $ _sentDirect s
|
||||
@@ -135,8 +140,9 @@ newAgentSMPServerStats' s = do
|
||||
sentAuthErrs,
|
||||
sentQuotaErrs,
|
||||
sentExpiredErrs,
|
||||
sentDirectErrs,
|
||||
sentProxyErrs,
|
||||
sentOtherErrs,
|
||||
-- sentDirectErrs,
|
||||
-- sentProxyErrs,
|
||||
recvMsgs,
|
||||
recvDuplicates,
|
||||
recvCryptoErrs,
|
||||
@@ -159,8 +165,9 @@ getAgentSMPServerStats s = do
|
||||
_sentAuthErrs <- readTVar $ sentDirect s
|
||||
_sentQuotaErrs <- readTVar $ sentDirect s
|
||||
_sentExpiredErrs <- readTVar $ sentDirect s
|
||||
_sentDirectErrs <- readTVar $ sentDirect s
|
||||
_sentProxyErrs <- readTVar $ sentDirect s
|
||||
_sentOtherErrs <- readTVar $ sentDirect s
|
||||
-- _sentDirectErrs <- readTVar $ sentDirect s
|
||||
-- _sentProxyErrs <- readTVar $ sentDirect s
|
||||
_recvMsgs <- readTVar $ sentDirect s
|
||||
_recvDuplicates <- readTVar $ sentDirect s
|
||||
_recvCryptoErrs <- readTVar $ sentDirect s
|
||||
@@ -181,8 +188,9 @@ getAgentSMPServerStats s = do
|
||||
_sentAuthErrs,
|
||||
_sentQuotaErrs,
|
||||
_sentExpiredErrs,
|
||||
_sentDirectErrs,
|
||||
_sentProxyErrs,
|
||||
_sentOtherErrs,
|
||||
-- _sentDirectErrs,
|
||||
-- _sentProxyErrs,
|
||||
_recvMsgs,
|
||||
_recvDuplicates,
|
||||
_recvCryptoErrs,
|
||||
@@ -205,8 +213,9 @@ setAgentSMPServerStats s d = do
|
||||
writeTVar (sentAuthErrs s) $! _sentAuthErrs d
|
||||
writeTVar (sentQuotaErrs s) $! _sentQuotaErrs d
|
||||
writeTVar (sentExpiredErrs s) $! _sentExpiredErrs d
|
||||
writeTVar (sentDirectErrs s) $! _sentDirectErrs d
|
||||
writeTVar (sentProxyErrs s) $! _sentProxyErrs d
|
||||
writeTVar (sentOtherErrs s) $! _sentOtherErrs d
|
||||
-- writeTVar (sentDirectErrs s) $! _sentDirectErrs d
|
||||
-- writeTVar (sentProxyErrs s) $! _sentProxyErrs d
|
||||
writeTVar (recvMsgs s) $! _recvMsgs d
|
||||
writeTVar (recvDuplicates s) $! _recvDuplicates d
|
||||
writeTVar (recvCryptoErrs s) $! _recvCryptoErrs d
|
||||
|
||||
Reference in New Issue
Block a user