mirror of
https://github.com/simplex-chat/simplexmq.git
synced 2026-10-12 02:22:45 +00:00
ntf: agent to notfiy MSGNTF on successful ack (#1355)
* ntf: agent to notfiy MSGNTF on successful ack * move msgntf * release lock earlier * fix tests * do not release get lock * fix * diff --------- Co-authored-by: Evgeny Poberezkin <[email protected]>
This commit is contained in:
spaced4ndy
co-authored by
Evgeny Poberezkin
Evgeny Poberezkin
parent
0705d05078
commit
49935473ae
@@ -1681,10 +1681,14 @@ synchronizeRatchet' c connId pqSupport' force = withConnLock c connId "synchroni
|
||||
_ -> throwE $ CMD PROHIBITED "synchronizeRatchet: not duplex"
|
||||
|
||||
ackQueueMessage :: AgentClient -> RcvQueue -> SMP.MsgId -> AM ()
|
||||
ackQueueMessage c rq@RcvQueue {userId, server} srvMsgId = do
|
||||
ackQueueMessage c rq@RcvQueue {userId, connId, server} srvMsgId = do
|
||||
atomically $ incSMPServerStat c userId server ackAttempts
|
||||
tryAgentError (sendAck c rq srvMsgId) >>= \case
|
||||
Right _ -> atomically $ incSMPServerStat c userId server ackMsgs
|
||||
Right _ -> do
|
||||
atomically $ incSMPServerStat c userId server ackMsgs
|
||||
whenM (liftIO $ hasGetLock c rq) $ do
|
||||
brokerTs_ <- (Just <$> withStore c (\db -> getRcvMsgBrokerTs db connId srvMsgId)) `catchAgentError` \_ -> pure Nothing
|
||||
atomically $ writeTBQueue (subQ c) ("", connId, AEvt SAEConn $ MSGNTF srvMsgId brokerTs_)
|
||||
Left (SMP _ SMP.NO_MSG) -> atomically $ incSMPServerStat c userId server ackNoMsgErrs
|
||||
Left e -> do
|
||||
unless (temporaryOrHostError e) $ atomically $ incSMPServerStat c userId server ackOtherErrs
|
||||
@@ -2285,12 +2289,9 @@ processSMPTransmissions c@AgentClient {subQ} (tSess@(userId, srv, _), _v, sessId
|
||||
atomically $ incSMPServerStat c userId srv recvMsgs
|
||||
void . handleNotifyAck $ do
|
||||
msg' <- decryptSMPMessage rq msg
|
||||
ack' <- handleNotifyAck $ case msg' of
|
||||
handleNotifyAck $ case msg' of
|
||||
SMP.ClientRcvMsgBody {msgTs = srvTs, msgFlags, msgBody} -> processClientMsg srvTs msgFlags msgBody
|
||||
SMP.ClientRcvMsgQuota {} -> queueDrained >> ack
|
||||
whenM (liftIO $ hasGetLock c rq) $
|
||||
notify (MSGNTF $ SMP.rcvMessageMeta srvMsgId msg')
|
||||
pure ack'
|
||||
where
|
||||
queueDrained = case conn of
|
||||
DuplexConnection _ _ sqs -> void $ enqueueMessages c cData sqs SMP.noMsgFlags $ A_QCONT (sndAddress rq)
|
||||
|
||||
@@ -1646,10 +1646,9 @@ disableQueuesNtfs = sendTSessionBatches "NDEL" snd disableQueues_
|
||||
queueCreds (_, RcvQueue {rcvPrivateKey, rcvId}) = (rcvPrivateKey, rcvId)
|
||||
|
||||
sendAck :: AgentClient -> RcvQueue -> MsgId -> AM ()
|
||||
sendAck c rq@RcvQueue {rcvId, rcvPrivateKey} msgId = do
|
||||
withSMPClient c rq ("ACK:" <> logSecret' msgId) $ \smp ->
|
||||
ackSMPMessage smp rcvPrivateKey rcvId msgId
|
||||
atomically $ releaseGetLock c rq
|
||||
sendAck c rq@RcvQueue {rcvId, rcvPrivateKey} msgId =
|
||||
withSMPClient c rq ("ACK:" <> logSecret' msgId) (\smp -> ackSMPMessage smp rcvPrivateKey rcvId msgId)
|
||||
`agentFinally` atomically (releaseGetLock c rq)
|
||||
|
||||
hasGetLock :: AgentClient -> RcvQueue -> IO Bool
|
||||
hasGetLock c RcvQueue {server, rcvId} =
|
||||
|
||||
@@ -198,7 +198,6 @@ import Simplex.Messaging.Protocol
|
||||
NMsgMeta,
|
||||
ProtocolServer (..),
|
||||
SMPClientVersion,
|
||||
SMPMsgMeta,
|
||||
SMPServer,
|
||||
SMPServerWithAuth,
|
||||
SndPublicAuthKey,
|
||||
@@ -356,7 +355,7 @@ data AEvent (e :: AEntity) where
|
||||
MERR :: AgentMsgId -> AgentErrorType -> AEvent AEConn
|
||||
MERRS :: NonEmpty AgentMsgId -> AgentErrorType -> AEvent AEConn
|
||||
MSG :: MsgMeta -> MsgFlags -> MsgBody -> AEvent AEConn
|
||||
MSGNTF :: SMPMsgMeta -> AEvent AEConn
|
||||
MSGNTF :: MsgId -> Maybe UTCTime -> AEvent AEConn
|
||||
RCVD :: MsgMeta -> NonEmpty MsgReceipt -> AEvent AEConn
|
||||
QCONT :: AEvent AEConn
|
||||
DEL_RCVQ :: SMPServer -> SMP.RecipientId -> Maybe AgentErrorType -> AEvent AEConn
|
||||
|
||||
@@ -122,6 +122,7 @@ module Simplex.Messaging.Agent.Store.SQLite
|
||||
getRcvMsg,
|
||||
getLastMsg,
|
||||
checkRcvMsgHashExists,
|
||||
getRcvMsgBrokerTs,
|
||||
deleteMsg,
|
||||
deleteDeliveredSndMsg,
|
||||
deleteSndMsgDelivery,
|
||||
@@ -1185,6 +1186,11 @@ checkRcvMsgHashExists db connId hash = do
|
||||
(connId, hash)
|
||||
)
|
||||
|
||||
getRcvMsgBrokerTs :: DB.Connection -> ConnId -> SMP.MsgId -> IO (Either StoreError BrokerTs)
|
||||
getRcvMsgBrokerTs db connId msgId =
|
||||
firstRow fromOnly SEMsgNotFound $
|
||||
DB.query db "SELECT broker_ts FROM rcv_messages WHERE conn_id = ? AND broker_id = ?" (connId, msgId)
|
||||
|
||||
deleteMsg :: DB.Connection -> ConnId -> InternalId -> IO ()
|
||||
deleteMsg db connId msgId =
|
||||
DB.execute db "DELETE FROM messages WHERE conn_id = ? AND internal_id = ?;" (connId, msgId)
|
||||
|
||||
@@ -1681,34 +1681,45 @@ testOnlyCreatePullSlowHandshake = withAgentClientsCfg2 agentProxyCfgV8 agentProx
|
||||
(aliceId, sqSecured) <- joinConnection bob 1 True qInfo "bob's connInfo" SMOnlyCreate
|
||||
liftIO $ sqSecured `shouldBe` False
|
||||
Just ("", _, CONF confId _ "bob's connInfo") <- getMsg alice bobId $ timeout 5_000000 $ get alice
|
||||
getMSGNTF alice bobId
|
||||
allowConnection alice bobId confId "alice's connInfo"
|
||||
liftIO $ threadDelay 1_000000
|
||||
getMsg bob aliceId $
|
||||
get bob ##> ("", aliceId, INFO "alice's connInfo")
|
||||
getMSGNTF bob aliceId
|
||||
liftIO $ threadDelay 1_000000
|
||||
getMsg alice bobId $ pure ()
|
||||
get alice ##> ("", bobId, CON)
|
||||
inAnyOrder
|
||||
(get alice)
|
||||
[ \case ("", c, CON) -> c == bobId; _ -> False,
|
||||
\case ("", c, MSGNTF {}) -> c == bobId; _ -> False
|
||||
]
|
||||
getMsg bob aliceId $
|
||||
get bob ##> ("", aliceId, CON)
|
||||
getMSGNTF bob aliceId
|
||||
-- exchange messages
|
||||
4 <- sendMessage alice bobId SMP.noMsgFlags "hello"
|
||||
get alice ##> ("", bobId, SENT 4)
|
||||
getMsg bob aliceId $
|
||||
get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
|
||||
ackMessage bob aliceId 4 Nothing
|
||||
getMSGNTF bob aliceId
|
||||
5 <- sendMessage bob aliceId SMP.noMsgFlags "hello too"
|
||||
get bob ##> ("", aliceId, SENT 5)
|
||||
getMsg alice bobId $
|
||||
get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False
|
||||
ackMessage alice bobId 5 Nothing
|
||||
getMSGNTF alice bobId
|
||||
|
||||
getMsg :: AgentClient -> ConnId -> ExceptT AgentErrorType IO a -> ExceptT AgentErrorType IO a
|
||||
getMsg c cId action = do
|
||||
liftIO $ noMessages c "nothing should be delivered before GET"
|
||||
Just _ <- getConnectionMessage c cId
|
||||
r <- action
|
||||
get c =##> \case ("", cId', MSGNTF _) -> cId == cId'; _ -> False
|
||||
pure r
|
||||
action
|
||||
|
||||
getMSGNTF :: AgentClient -> ConnId -> ExceptT AgentErrorType IO ()
|
||||
getMSGNTF c cId =
|
||||
get c =##> \case ("", c', MSGNTF {}) -> c' == cId; _ -> False
|
||||
|
||||
testOnlyCreatePull :: IO ()
|
||||
testOnlyCreatePull = withAgentClients2 $ \alice bob -> runRight_ $ do
|
||||
@@ -1716,11 +1727,13 @@ testOnlyCreatePull = withAgentClients2 $ \alice bob -> runRight_ $ do
|
||||
(aliceId, sqSecured) <- joinConnection bob 1 True qInfo "bob's connInfo" SMOnlyCreate
|
||||
liftIO $ sqSecured `shouldBe` True
|
||||
Just ("", _, CONF confId _ "bob's connInfo") <- getMsg alice bobId $ timeout 5_000000 $ get alice
|
||||
getMSGNTF alice bobId
|
||||
allowConnection alice bobId confId "alice's connInfo"
|
||||
liftIO $ threadDelay 1_000000
|
||||
getMsg bob aliceId $ do
|
||||
get bob ##> ("", aliceId, INFO "alice's connInfo")
|
||||
get bob ##> ("", aliceId, CON)
|
||||
getMSGNTF bob aliceId
|
||||
liftIO $ threadDelay 1_000000
|
||||
get alice ##> ("", bobId, CON) -- sent to initiating party after sending confirmation
|
||||
-- exchange messages
|
||||
@@ -1729,11 +1742,13 @@ testOnlyCreatePull = withAgentClients2 $ \alice bob -> runRight_ $ do
|
||||
getMsg bob aliceId $
|
||||
get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False
|
||||
ackMessage bob aliceId 2 Nothing
|
||||
getMSGNTF bob aliceId
|
||||
3 <- sendMessage bob aliceId SMP.noMsgFlags "hello too"
|
||||
get bob ##> ("", aliceId, SENT 3)
|
||||
getMsg alice bobId $
|
||||
get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False
|
||||
ackMessage alice bobId 3 Nothing
|
||||
getMSGNTF alice bobId
|
||||
|
||||
makeConnection :: AgentClient -> AgentClient -> ExceptT AgentErrorType IO (ConnId, ConnId)
|
||||
makeConnection = makeConnection_ PQSupportOn True
|
||||
|
||||
Reference in New Issue
Block a user