diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 5ca1b4be1..d2a96a5ed 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -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) diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index c2a3cab57..1c6fa38f8 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -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} = diff --git a/src/Simplex/Messaging/Agent/Protocol.hs b/src/Simplex/Messaging/Agent/Protocol.hs index 635bb1aee..17ebb7d43 100644 --- a/src/Simplex/Messaging/Agent/Protocol.hs +++ b/src/Simplex/Messaging/Agent/Protocol.hs @@ -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 diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 76e8a79f6..924ba6712 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -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) diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index 199b1a36a..075b40f10 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -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