diff --git a/simplexmq.cabal b/simplexmq.cabal index eb205f1a9..3a3ef9c86 100644 --- a/simplexmq.cabal +++ b/simplexmq.cabal @@ -82,6 +82,7 @@ library Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230401_snd_files Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230510_files_pending_replicas_indexes Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230516_encrypted_rcv_message_hashes + Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230531_switch_status Simplex.Messaging.Agent.TAsyncs Simplex.Messaging.Agent.TRcvQueues Simplex.Messaging.Client diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index f6181bb92..dcada9b9b 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -63,6 +63,7 @@ module Simplex.Messaging.Agent sendMessage, ackMessage, switchConnection, + stopConnectionSwitch, suspendConnection, deleteConnection, deleteConnections, @@ -110,7 +111,7 @@ import Data.Composition ((.:), (.:.), (.::)) import Data.Foldable (foldl') import Data.Functor (($>)) import Data.List (find) -import Data.List.NonEmpty (NonEmpty (..), (<|)) +import Data.List.NonEmpty (NonEmpty (..)) import qualified Data.List.NonEmpty as L import Data.Map.Strict (Map) import qualified Data.Map.Strict as M @@ -268,6 +269,10 @@ ackMessage c = withAgentEnv c .: ackMessage' c switchConnection :: AgentErrorMonad m => AgentClient -> ConnId -> m ConnectionStats switchConnection c = withAgentEnv c . switchConnection' c +-- | Stop switching connection to the new receive queue +stopConnectionSwitch :: AgentErrorMonad m => AgentClient -> ConnId -> m ConnectionStats +stopConnectionSwitch c = withAgentEnv c . stopConnectionSwitch' c + -- | Suspend SMP agent connection (OFF command) suspendConnection :: AgentErrorMonad m => AgentClient -> ConnId -> m () suspendConnection c = withAgentEnv c . suspendConnection' c @@ -526,9 +531,14 @@ deleteConnectionsAsync_ onSuccess c connIds = case connIds of -- | Add connection to the new receive queue switchConnectionAsync' :: AgentMonad m => AgentClient -> ACorrId -> ConnId -> m () switchConnectionAsync' c corrId connId = - withStore c (`getConn` connId) >>= \case - SomeConn _ DuplexConnection {} -> enqueueCommand c corrId connId Nothing $ AClientCommand $ APC SAEConn SWCH - _ -> throwError $ CMD PROHIBITED + withConnLock c connId "switchConnectionAsync" $ + withStore c (`getConn` connId) >>= \case + SomeConn _ (DuplexConnection _ rqs@(rq :| _rqs) _) + | isJust (switchingRQ rqs) -> throwError $ CMD PROHIBITED + | otherwise -> do + void $ withStore' c $ \db -> setRcvSwitchStatus db rq $ Just RSSwitchStarted + enqueueCommand c corrId connId Nothing $ AClientCommand $ APC SAEConn SWCH + _ -> throwError $ CMD PROHIBITED newConn :: AgentMonad m => AgentClient -> UserId -> ConnId -> Bool -> SConnectionMode c -> Maybe CRClientData -> m (ConnId, ConnectionRequestUri c) newConn c userId connId enableNtfs cMode clientData = @@ -867,7 +877,12 @@ runCommandProcessing c@AgentClient {subQ} server_ = do notify OK LET confId ownCInfo -> withServer' . tryCommand $ allowConnection' c connId confId ownCInfo >> notify OK ACK msgId -> withServer' . tryCommand $ ackMessage' c connId msgId >> notify OK - SWCH -> noServer $ tryCommand $ switchConnection' c connId >>= notify . SWITCH QDRcv SPStarted + SWCH -> + noServer . tryCommand . withConnLock c connId "switchConnection" $ + withStore c (`getConn` connId) >>= \case + SomeConn _ conn@(DuplexConnection _ (replaced :| _rqs) _) -> + switchDuplexConnection c conn replaced >>= notify . SWITCH QDRcv SPStarted + _ -> throwError $ CMD PROHIBITED DEL -> withServer' . tryCommand $ deleteConnection' c connId >> notify OK _ -> notify $ ERR $ INTERNAL $ "unsupported async command " <> show (aCommandTag cmd) AInternalCommand cmd -> case cmd of @@ -887,13 +902,24 @@ runCommandProcessing c@AgentClient {subQ} server_ = do enqueueMessage c cData sq SMP.MsgFlags {notification = True} HELLO -- ICDeleteConn is no longer used, but it can be present in old client databases ICDeleteConn -> withStore' c (`deleteCommand` cmdId) + ICDeleteRcvQueue rId -> withServer $ \srv -> tryWithLock "ICDeleteRcvQueue" $ do + rq <- withStore c (\db -> getDeletedRcvQueue db connId srv rId) + deleteQueue c rq + withStore' c (`deleteConnRcvQueue` rq) ICQSecure rId senderKey -> withServer $ \srv -> tryWithLock "ICQSecure" . withDuplexConn $ \(DuplexConnection cData rqs sqs) -> case find (sameQueue (srv, rId)) rqs of - Just rq'@RcvQueue {server, sndId, status} -> when (status == Confirmed) $ do - secureQueue c rq' senderKey - withStore' c $ \db -> setRcvQueueStatus db rq' Secured - void . enqueueMessages c cData sqs SMP.noMsgFlags $ QUSE [((server, sndId), True)] + Just rq'@RcvQueue {server, sndId, status, dbReplaceQueueId = Just replaceQId} -> + case find ((replaceQId ==) . dbQId) rqs of + Just rq1 -> when (status == Confirmed) $ do + secureQueue c rq' senderKey + withStore' c $ \db -> setRcvQueueStatus db rq' Secured + void . enqueueMessages c cData sqs SMP.noMsgFlags $ QUSE [((server, sndId), True)] + rq1' <- withStore' c $ \db -> setRcvSwitchStatus db rq1 $ Just RSSendingQUSE + let rqs' = updatedQs rq1' rqs + conn' = DuplexConnection cData rqs' sqs + notify . SWITCH QDRcv SPSecured $ connectionStats conn' + _ -> internalErr "ICQSecure: no switching queue found" _ -> internalErr "ICQSecure: queue address not found in connection" ICQDelete rId -> do withServer $ \srv -> tryWithLock "ICQDelete" . withDuplexConn $ \(DuplexConnection cData rqs sqs) -> do @@ -902,13 +928,20 @@ runCommandProcessing c@AgentClient {subQ} server_ = do Just (rq'@RcvQueue {primary}, rq'' : rqs') | primary -> internalErr "ICQDelete: cannot delete primary rcv queue" | otherwise -> do - deleteQueue c rq' - withStore' c $ \db -> deleteConnRcvQueue db rq' - when (enableNtfs cData) $ do - ns <- asks ntfSupervisor - atomically $ sendNtfSubCommand ns (connId, NSCCreate) - let conn' = DuplexConnection cData (rq'' :| rqs') sqs - notify $ SWITCH QDRcv SPCompleted $ connectionStats conn' + checkRQSwchStatus rq' RSReceivedMessage + tryError (deleteQueue c rq') >>= \case + Right () -> finalizeSwitch + Left e + | temporaryOrHostError e -> throwError e + | otherwise -> finalizeSwitch >> throwError e + where + finalizeSwitch = do + withStore' c $ \db -> deleteConnRcvQueue db rq' + when (enableNtfs cData) $ do + ns <- asks ntfSupervisor + atomically $ sendNtfSubCommand ns (connId, NSCCreate) + let conn' = DuplexConnection cData (rq'' :| rqs') sqs + notify $ SWITCH QDRcv SPCompleted $ connectionStats conn' _ -> internalErr "ICQDelete: cannot delete the only queue in connection" where ack srv rId srvMsgId = do @@ -1112,9 +1145,11 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl AM_A_MSG_ -> notify $ SENT mId AM_QCONT_ -> pure () AM_QADD_ -> pure () - AM_QKEY_ -> pure () + AM_QKEY_ -> do + SomeConn _ conn <- withStore c (`getConn` connId) + notify . SWITCH QDSnd SPConfirmed $ connectionStats conn AM_QUSE_ -> pure () - AM_QTEST_ -> do + AM_QTEST_ -> withConnLock c connId "runSmpQueueMsgDelivery AM_QTEST_" $ do withStore' c $ \db -> setSndQueueStatus db sq Active SomeConn _ conn <- withStore c (`getConn` connId) case conn of @@ -1125,9 +1160,10 @@ runSmpQueueMsgDelivery c@AgentClient {subQ} cData@ConnData {userId, connId, dupl -- this is the same queue where this loop delivers messages to but with updated state Just SndQueue {dbReplaceQueueId = Just replacedId, primary} -> -- second part of this condition is a sanity check because dbReplaceQueueId cannot point to the same queue, see switchConnection' - case removeQP (\sq'@SndQueue {dbQueueId} -> dbQueueId == replacedId && not (sameQueue addr sq')) sqs of + case removeQP (\sq' -> dbQId sq' == replacedId && not (sameQueue addr sq')) sqs of Nothing -> internalErr msgId "sent QTEST: queue not found in connection" Just (sq', sq'' : sqs') -> do + checkSQSwchStatus sq' SSSendingQTEST -- remove the delivery from the map to stop the thread when the delivery loop is complete atomically $ TM.delete (qAddress sq') $ smpQueueMsgQueues c withStore' c $ \db -> do @@ -1178,21 +1214,66 @@ ackMessage' c connId msgId = withConnLock c connId "ackMessage" $ do withStore' c $ \db -> deleteMsg db connId mId switchConnection' :: AgentMonad m => AgentClient -> ConnId -> m ConnectionStats -switchConnection' c connId = withConnLock c connId "switchConnection" $ do - SomeConn _ conn <- withStore c (`getConn` connId) - case conn of - DuplexConnection cData@ConnData {userId} rqs@(rq@RcvQueue {server, dbQueueId, sndId} :| rqs_) sqs -> do - clientVRange <- asks $ smpClientVRange . config - -- try to get the server that is different from all queues, or at least from the primary rcv queue - srvAuth@(ProtoServerWithAuth srv _) <- getNextServer c userId $ map qServer (L.toList rqs) <> map qServer (L.toList sqs) - srv' <- if srv == server then getNextServer c userId [server] else pure srvAuth - (q, qUri) <- newRcvQueue c userId connId srv' clientVRange - let rq' = (q :: RcvQueue) {primary = True, dbReplaceQueueId = Just dbQueueId} - void . withStore c $ \db -> addConnRcvQueue db connId rq' - addSubscription c rq' - void . enqueueMessages c cData sqs SMP.noMsgFlags $ QADD [(qUri, Just (server, sndId))] - pure . connectionStats $ DuplexConnection cData (rq <| rq' :| rqs_) sqs - _ -> throwError $ CMD PROHIBITED +switchConnection' c connId = + withConnLock c connId "switchConnection" $ + withStore c (`getConn` connId) >>= \case + SomeConn _ conn@(DuplexConnection _ rqs@(rq :| _rqs) _) + | isJust (switchingRQ rqs) -> throwError $ CMD PROHIBITED + | otherwise -> do + rq' <- withStore' c $ \db -> setRcvSwitchStatus db rq $ Just RSSwitchStarted + switchDuplexConnection c conn rq' + _ -> throwError $ CMD PROHIBITED + +switchDuplexConnection :: AgentMonad m => AgentClient -> Connection 'CDuplex -> RcvQueue -> m ConnectionStats +switchDuplexConnection c (DuplexConnection cData@ConnData {connId, userId} rqs sqs) rq@RcvQueue {server, dbQueueId, sndId} = do + checkRQSwchStatus rq RSSwitchStarted + clientVRange <- asks $ smpClientVRange . config + -- try to get the server that is different from all queues, or at least from the primary rcv queue + srvAuth@(ProtoServerWithAuth srv _) <- getNextServer c userId $ map qServer (L.toList rqs) <> map qServer (L.toList sqs) + srv' <- if srv == server then getNextServer c userId [server] else pure srvAuth + (q, qUri) <- newRcvQueue c userId connId srv' clientVRange + let rq' = (q :: RcvQueue) {primary = True, dbReplaceQueueId = Just dbQueueId} + void . withStore c $ \db -> addConnRcvQueue db connId rq' + addSubscription c rq' + void . enqueueMessages c cData sqs SMP.noMsgFlags $ QADD [(qUri, Just (server, sndId))] + rq1 <- withStore' c $ \db -> setRcvSwitchStatus db rq $ Just RSSendingQADD + let rqs' = updatedQs rq1 rqs <> [rq'] + pure . connectionStats $ DuplexConnection cData rqs' sqs + +stopConnectionSwitch' :: AgentMonad m => AgentClient -> ConnId -> m ConnectionStats +stopConnectionSwitch' c connId = + withConnLock c connId "stopConnectionSwitch" $ + withStore c (`getConn` connId) >>= \case + SomeConn _ (DuplexConnection cData rqs sqs) -> case switchingRQ rqs of + Just rq + | canStopRcvSwitch rq -> do + -- multiple queues to which the connections switches were possible when repeating switch was allowed + let (delRqs, keepRqs) = L.partition ((Just (dbQId rq) ==) . dbReplaceQId) rqs + case L.nonEmpty keepRqs of + Just rqs' -> do + rq' <- withStore' c $ \db -> do + mapM_ (setRcvQueueDeleted db) delRqs + setRcvSwitchStatus db rq Nothing + forM_ delRqs $ \RcvQueue {server, rcvId} -> enqueueCommand c "" connId (Just server) $ AInternalCommand $ ICDeleteRcvQueue rcvId + let rqs'' = updatedQs rq' rqs' + conn' = DuplexConnection cData rqs'' sqs + pure $ connectionStats conn' + _ -> throwError $ INTERNAL "won't delete all rcv queues in connection" + | otherwise -> throwError $ CMD PROHIBITED + _ -> throwError $ CMD PROHIBITED + _ -> throwError $ CMD PROHIBITED + +canStopRcvSwitch :: RcvQueue -> Bool +canStopRcvSwitch = maybe False canStop . rcvSwchStatus + where + canStop = \case + RSSwitchStarted -> True + RSSendingQADD -> True + -- if switch is in RSSendingQUSE, a race condition with sender deleting the original queue is possible + RSSendingQUSE -> False + -- if switch is in RSReceivedMessage status, stopping switch (deleting new queue) + -- will break the connection because the sender would have original queue deleted + RSReceivedMessage -> False ackQueueMessage :: AgentMonad m => AgentClient -> RcvQueue -> SMP.MsgId -> m () ackQueueMessage c rq srvMsgId = @@ -1331,11 +1412,11 @@ getConnectionRatchetAdHash' c connId = do connectionStats :: Connection c -> ConnectionStats connectionStats = \case - RcvConnection _ rq -> ConnectionStats {rcvServers = [qServer rq], sndServers = []} - SndConnection _ sq -> ConnectionStats {rcvServers = [], sndServers = [qServer sq]} - DuplexConnection _ rqs sqs -> ConnectionStats {rcvServers = map qServer $ L.toList rqs, sndServers = map qServer $ L.toList sqs} - ContactConnection _ rq -> ConnectionStats {rcvServers = [qServer rq], sndServers = []} - NewConnection _ -> ConnectionStats {rcvServers = [], sndServers = []} + RcvConnection _ rq -> ConnectionStats {rcvQueuesInfo = [rcvQueueInfo rq], sndQueuesInfo = []} + SndConnection _ sq -> ConnectionStats {rcvQueuesInfo = [], sndQueuesInfo = [sndQueueInfo sq]} + DuplexConnection _ rqs sqs -> ConnectionStats {rcvQueuesInfo = map rcvQueueInfo $ L.toList rqs, sndQueuesInfo = map sndQueueInfo $ L.toList sqs} + ContactConnection _ rq -> ConnectionStats {rcvQueuesInfo = [rcvQueueInfo rq], sndQueuesInfo = []} + NewConnection _ -> ConnectionStats {rcvQueuesInfo = [], sndQueuesInfo = []} -- | Change servers to be used for creating new queues, in Reader monad setProtocolServers' :: forall p m. (ProtocolTypeI p, UserProtocol p, AgentMonad m) => AgentClient -> UserId -> NonEmpty (ProtoServerWithAuth p) -> m () @@ -1698,8 +1779,10 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s case (conn, dbReplaceQueueId) of (DuplexConnection _ rqs _, Just replacedId) -> do when primary . withStore' c $ \db -> setRcvQueuePrimary db connId rq - case find (\RcvQueue {dbQueueId} -> dbQueueId == replacedId) rqs of - Just RcvQueue {server, rcvId} -> do + case find ((replacedId ==) . dbQId) rqs of + Just rq'@RcvQueue {server, rcvId} -> do + checkRQSwchStatus rq' RSSendingQUSE + void $ withStore' c $ \db -> setRcvSwitchStatus db rq' $ Just RSReceivedMessage enqueueCommand c "" connId (Just server) $ AInternalCommand $ ICQDelete rcvId _ -> notify . ERR . AGENT $ A_QUEUE "replaced RcvQueue not found in connection" _ -> pure () @@ -1901,24 +1984,33 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s -- processed by queue sender qAddMsg :: NonEmpty (SMPQueueUri, Maybe SndQAddr) -> Connection 'CDuplex -> m () qAddMsg ((_, Nothing) :| _) _ = qError "adding queue without switching is not supported" - qAddMsg ((qUri, Just addr) :| _) (DuplexConnection _ rqs sqs@(sq :| sqs_)) = do + qAddMsg ((qUri, Just addr) :| _) (DuplexConnection _ rqs sqs) = do clientVRange <- asks $ smpClientVRange . config case qUri `compatibleVersion` clientVRange of Just qInfo@(Compatible sqInfo@SMPQueueInfo {queueAddress}) -> case (findQ (qAddress sqInfo) sqs, findQ addr sqs) of (Just _, _) -> qError "QADD: queue address is already used in connection" - (_, Just _replaced@SndQueue {dbQueueId}) -> do - sq_@SndQueue {sndPublicKey, e2ePubKey} <- newSndQueue userId connId qInfo - let sq' = (sq_ :: SndQueue) {primary = True, dbReplaceQueueId = Just dbQueueId} - void . withStore c $ \db -> addConnSndQueue db connId sq' - case (sndPublicKey, e2ePubKey) of - (Just sndPubKey, Just dhPublicKey) -> do - logServer "<--" c srv rId $ "MSG " <> logSecret (senderId queueAddress) - let sqInfo' = (sqInfo :: SMPQueueInfo) {queueAddress = queueAddress {dhPublicKey}} - void . enqueueMessages c cData sqs SMP.noMsgFlags $ QKEY [(sqInfo', sndPubKey)] - let conn' = DuplexConnection cData rqs (sq <| sq' :| sqs_) - notify . SWITCH QDSnd SPStarted $ connectionStats conn' - _ -> qError "absent sender keys" + (_, Just sq@SndQueue {dbQueueId}) -> do + let (delSqs, keepSqs) = L.partition ((Just dbQueueId ==) . dbReplaceQId) sqs + case L.nonEmpty keepSqs of + Just sqs' -> do + -- move inside case? + withStore' c $ \db -> mapM_ (deleteConnSndQueue db connId) delSqs + sq_@SndQueue {sndPublicKey, e2ePubKey} <- newSndQueue userId connId qInfo + let sq'' = (sq_ :: SndQueue) {primary = True, dbQueueId, dbReplaceQueueId = Just dbQueueId} + dbId <- withStore c $ \db -> addConnSndQueue db connId sq'' + let sq2 = (sq'' :: SndQueue) {dbQueueId = dbId} + case (sndPublicKey, e2ePubKey) of + (Just sndPubKey, Just dhPublicKey) -> do + logServer "<--" c srv rId $ "MSG " <> logSecret (senderId queueAddress) + let sqInfo' = (sqInfo :: SMPQueueInfo) {queueAddress = queueAddress {dhPublicKey}} + void . enqueueMessages c cData sqs SMP.noMsgFlags $ QKEY [(sqInfo', sndPubKey)] + sq1 <- withStore' c $ \db -> setSndSwitchStatus db sq $ Just SSSendingQKEY + let sqs'' = updatedQs sq1 sqs' <> [sq2] + conn' = DuplexConnection cData rqs sqs'' + notify . SWITCH QDSnd SPStarted $ connectionStats conn' + _ -> qError "absent sender keys" + _ -> qError "QADD: won't delete all snd queues in connection" _ -> qError "QADD: replaced queue address is not found in connection" _ -> throwError $ AGENT A_VERSION @@ -1930,6 +2022,7 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s case findRQ (smpServer, senderId) rqs of Just rq'@RcvQueue {rcvId, e2ePrivKey = dhPrivKey, smpClientVersion = cVer, status = status'} | status' == New || status' == Confirmed -> do + checkRQSwchStatus rq RSSendingQADD logServer "<--" c srv rId $ "MSG " <> logSecret senderId let dhSecret = C.dh' dhPublicKey dhPrivKey withStore' c $ \db -> setRcvQueueConfirmedE2E db rq' dhSecret $ min cVer cVer' @@ -1944,16 +2037,23 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s -- mark queue as Secured and to start sending messages to it qUseMsg :: NonEmpty ((SMPServer, SMP.SenderId), Bool) -> Connection 'CDuplex -> m () -- NOTE: does not yet support the change of the primary status during the rotation - qUseMsg ((addr, _primary) :| _) (DuplexConnection _ _ sqs) = + qUseMsg ((addr, _primary) :| _) (DuplexConnection _ rqs sqs) = case findQ addr sqs of - Just sq' -> do - logServer "<--" c srv rId $ "MSG " <> logSecret (snd addr) - withStore' c $ \db -> setSndQueueStatus db sq' Secured - let sq'' = (sq' :: SndQueue) {status = Secured} - -- sending QTEST to the new queue only, the old one will be removed if sent successfully - void $ enqueueMessages c cData [sq''] SMP.noMsgFlags $ QTEST [addr] - notify . SWITCH QDSnd SPConfirmed $ connectionStats conn - _ -> qError "QUSE: queue address not found in connection" + Just sq'@SndQueue {dbReplaceQueueId = Just replaceQId} -> do + case find ((replaceQId ==) . dbQId) sqs of + Just sq1 -> do + checkSQSwchStatus sq1 SSSendingQKEY + logServer "<--" c srv rId $ "MSG " <> logSecret (snd addr) + withStore' c $ \db -> setSndQueueStatus db sq' Secured + let sq'' = (sq' :: SndQueue) {status = Secured} + -- sending QTEST to the new queue only, the old one will be removed if sent successfully + void $ enqueueMessages c cData [sq''] SMP.noMsgFlags $ QTEST [addr] + sq1' <- withStore' c $ \db -> setSndSwitchStatus db sq1 $ Just SSSendingQTEST + let sqs' = updatedQs sq1' sqs + conn' = DuplexConnection cData rqs sqs' + notify . SWITCH QDSnd SPSecured $ connectionStats conn' + _ -> qError "QUSE: switching SndQueue not found in connection" + _ -> qError "QUSE: switched queue address not found in connection" qError :: String -> m () qError = throwError . AGENT . A_QUEUE @@ -1979,6 +2079,21 @@ processSMPTransmission c@AgentClient {smpClients, subQ} (tSess@(_, srv, _), v, s | internalPrevMsgHash /= receivedPrevMsgHash = MsgError MsgBadHash | otherwise = MsgError MsgDuplicate -- this case is not possible +checkRQSwchStatus :: AgentMonad m => RcvQueue -> RcvSwitchStatus -> m () +checkRQSwchStatus rq@RcvQueue {rcvSwchStatus} expected = + unless (rcvSwchStatus == Just expected) $ switchStatusError rq expected rcvSwchStatus + +checkSQSwchStatus :: AgentMonad m => SndQueue -> SndSwitchStatus -> m () +checkSQSwchStatus sq@SndQueue {sndSwchStatus} expected = + unless (sndSwchStatus == Just expected) $ switchStatusError sq expected sndSwchStatus + +switchStatusError :: (SMPQueueRec q, AgentMonad m, Show a) => q -> a -> Maybe a -> m () +switchStatusError q expected actual = + throwError . INTERNAL $ + ("unexpected switch status, queueId=" <> show (queueId q)) + <> (", expected=" <> show expected) + <> (", actual=" <> show actual) + connectReplyQueues :: AgentMonad m => AgentClient -> ConnData -> ConnInfo -> NonEmpty SMPQueueInfo -> m () connectReplyQueues c cData@ConnData {userId, connId} ownConnInfo (qInfo :| _) = do clientVRange <- asks $ smpClientVRange . config @@ -2063,5 +2178,6 @@ newSndQueue userId connId (Compatible (SMPQueueInfo smpClientVersion SMPQueueAdd dbQueueId = 0, primary = True, dbReplaceQueueId = Nothing, + sndSwchStatus = Nothing, smpClientVersion } diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index 9a135b6c0..895f33943 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -841,6 +841,7 @@ newRcvQueue c userId connId (ProtoServerWithAuth srv auth) vRange = do dbQueueId = 0, primary = True, dbReplaceQueueId = Nothing, + rcvSwchStatus = Nothing, smpClientVersion = maxVersion vRange, clientNtfCreds = Nothing, deleteErrors = 0 diff --git a/src/Simplex/Messaging/Agent/Protocol.hs b/src/Simplex/Messaging/Agent/Protocol.hs index 89c30eae0..0d260c150 100644 --- a/src/Simplex/Messaging/Agent/Protocol.hs +++ b/src/Simplex/Messaging/Agent/Protocol.hs @@ -55,8 +55,12 @@ module Simplex.Messaging.Agent.Protocol AEntityI (..), MsgHash, MsgMeta (..), + RcvQueueInfo (..), + SndQueueInfo (..), ConnectionStats (..), SwitchPhase (..), + RcvSwitchStatus (..), + SndSwitchStatus (..), QueueDirection (..), SMPConfirmation (..), AgentMsgEnvelope (..), @@ -473,18 +477,20 @@ instance ToJSON QueueDirection where instance FromJSON QueueDirection where parseJSON = strParseJSON "QueueDirection" -data SwitchPhase = SPStarted | SPConfirmed | SPCompleted +data SwitchPhase = SPStarted | SPConfirmed | SPSecured | SPCompleted deriving (Eq, Show) instance StrEncoding SwitchPhase where strEncode = \case SPStarted -> "started" SPConfirmed -> "confirmed" + SPSecured -> "secured" SPCompleted -> "completed" strP = A.takeTill (== ' ') >>= \case "started" -> pure SPStarted "confirmed" -> pure SPConfirmed + "secured" -> pure SPSecured "completed" -> pure SPCompleted _ -> fail "bad SwitchPhase" @@ -495,19 +501,109 @@ instance ToJSON SwitchPhase where instance FromJSON SwitchPhase where parseJSON = strParseJSON "SwitchPhase" +data RcvSwitchStatus + = RSSwitchStarted + | RSSendingQADD + | RSSendingQUSE + | RSReceivedMessage + deriving (Eq, Show) + +instance StrEncoding RcvSwitchStatus where + strEncode = \case + RSSwitchStarted -> "switch_started" + RSSendingQADD -> "sending_qadd" + RSSendingQUSE -> "sending_quse" + RSReceivedMessage -> "received_message" + strP = + A.takeTill (== ' ') >>= \case + "switch_started" -> pure RSSwitchStarted + "sending_qadd" -> pure RSSendingQADD + "sending_quse" -> pure RSSendingQUSE + "received_message" -> pure RSReceivedMessage + _ -> fail "bad RcvSwitchStatus" + +instance ToField RcvSwitchStatus where toField = toField . strEncode + +instance FromField RcvSwitchStatus where fromField = blobFieldDecoder $ parseAll strP + +instance ToJSON RcvSwitchStatus where + toEncoding = strToJEncoding + toJSON = strToJSON + +instance FromJSON RcvSwitchStatus where + parseJSON = strParseJSON "RcvSwitchStatus" + +data SndSwitchStatus + = SSSendingQKEY + | SSSendingQTEST + deriving (Eq, Show) + +instance StrEncoding SndSwitchStatus where + strEncode = \case + SSSendingQKEY -> "sending_qkey" + SSSendingQTEST -> "sending_qtest" + strP = + A.takeTill (== ' ') >>= \case + "sending_qkey" -> pure SSSendingQKEY + "sending_qtest" -> pure SSSendingQTEST + _ -> fail "bad SndSwitchStatus" + +instance ToField SndSwitchStatus where toField = toField . strEncode + +instance FromField SndSwitchStatus where fromField = blobFieldDecoder $ parseAll strP + +instance ToJSON SndSwitchStatus where + toEncoding = strToJEncoding + toJSON = strToJSON + +instance FromJSON SndSwitchStatus where + parseJSON = strParseJSON "SndSwitchStatus" + +data RcvQueueInfo = RcvQueueInfo + { rcvServer :: SMPServer, + rcvSwitchStatus :: Maybe RcvSwitchStatus + } + deriving (Eq, Show, Generic) + +instance ToJSON RcvQueueInfo where toEncoding = J.genericToEncoding J.defaultOptions {J.omitNothingFields = True} + +instance StrEncoding RcvQueueInfo where + strEncode RcvQueueInfo {rcvServer, rcvSwitchStatus} = + "srv=" <> strEncode rcvServer <> maybe "" (\switch -> ";switch=" <> strEncode switch) rcvSwitchStatus + strP = do + rcvServer <- "srv=" *> strP + rcvSwitchStatus <- optional $ ";switch=" *> strP + pure RcvQueueInfo {rcvServer, rcvSwitchStatus} + +data SndQueueInfo = SndQueueInfo + { sndServer :: SMPServer, + sndSwitchStatus :: Maybe SndSwitchStatus + } + deriving (Eq, Show, Generic) + +instance ToJSON SndQueueInfo where toEncoding = J.genericToEncoding J.defaultOptions {J.omitNothingFields = True} + +instance StrEncoding SndQueueInfo where + strEncode SndQueueInfo {sndServer, sndSwitchStatus} = + "srv=" <> strEncode sndServer <> maybe "" (\switch -> ";switch=" <> strEncode switch) sndSwitchStatus + strP = do + sndServer <- "srv=" *> strP + sndSwitchStatus <- optional $ ";switch=" *> strP + pure SndQueueInfo {sndServer, sndSwitchStatus} + data ConnectionStats = ConnectionStats - { rcvServers :: [SMPServer], - sndServers :: [SMPServer] + { rcvQueuesInfo :: [RcvQueueInfo], + sndQueuesInfo :: [SndQueueInfo] } deriving (Eq, Show, Generic) instance StrEncoding ConnectionStats where - strEncode ConnectionStats {rcvServers, sndServers} = - "rcv=" <> strEncodeList rcvServers <> " snd=" <> strEncodeList sndServers + strEncode ConnectionStats {rcvQueuesInfo, sndQueuesInfo} = + "rcv=" <> strEncodeList rcvQueuesInfo <> " snd=" <> strEncodeList sndQueuesInfo strP = do - rcvServers <- "rcv=" *> strListP - sndServers <- " snd=" *> strListP - pure ConnectionStats {rcvServers, sndServers} + rcvQueuesInfo <- "rcv=" *> strListP + sndQueuesInfo <- " snd=" *> strListP + pure ConnectionStats {rcvQueuesInfo, sndQueuesInfo} instance ToJSON ConnectionStats where toEncoding = J.genericToEncoding J.defaultOptions diff --git a/src/Simplex/Messaging/Agent/Store.hs b/src/Simplex/Messaging/Agent/Store.hs index 59f5c06b0..8e83a1451 100644 --- a/src/Simplex/Messaging/Agent/Store.hs +++ b/src/Simplex/Messaging/Agent/Store.hs @@ -21,6 +21,7 @@ import Data.Kind (Type) import Data.List (find) import Data.List.NonEmpty (NonEmpty) import qualified Data.List.NonEmpty as L +import Data.Maybe (isJust) import Data.Time (UTCTime) import Data.Type.Equality import Simplex.Messaging.Agent.Protocol @@ -66,12 +67,13 @@ data RcvQueue = RcvQueue sndId :: SMP.SenderId, -- | queue status status :: QueueStatus, - -- | database queue ID (within connection), can be Nothing for old queues + -- | database queue ID (within connection) dbQueueId :: Int64, -- | True for a primary or a next primary queue of the connection (next if dbReplaceQueueId is set) primary :: Bool, -- | database queue ID to replace, Nothing if this queue is not replacing another, `Just Nothing` is used for replacing old queues dbReplaceQueueId :: Maybe Int64, + rcvSwchStatus :: Maybe RcvSwitchStatus, -- | SMP client version smpClientVersion :: Version, -- | credentials used in context of notifications @@ -80,6 +82,10 @@ data RcvQueue = RcvQueue } deriving (Eq, Show) +rcvQueueInfo :: RcvQueue -> RcvQueueInfo +rcvQueueInfo RcvQueue {server, rcvSwchStatus} = + RcvQueueInfo {rcvServer = server, rcvSwitchStatus = rcvSwchStatus} + data ClientNtfCreds = ClientNtfCreds { -- | key pair to be used by the notification server to sign transmissions ntfPublicKey :: NtfPublicVerifyKey, @@ -107,17 +113,22 @@ data SndQueue = SndQueue e2eDhSecret :: C.DhSecretX25519, -- | queue status status :: QueueStatus, - -- | database queue ID (within connection), can be Nothing for old queues + -- | database queue ID (within connection) dbQueueId :: Int64, -- | True for a primary or a next primary queue of the connection (next if dbReplaceQueueId is set) primary :: Bool, -- | ID of the queue this one is replacing dbReplaceQueueId :: Maybe Int64, + sndSwchStatus :: Maybe SndSwitchStatus, -- | SMP client version smpClientVersion :: Version } deriving (Eq, Show) +sndQueueInfo :: SndQueue -> SndQueueInfo +sndQueueInfo SndQueue {server, sndSwchStatus} = + SndQueueInfo {sndServer = server, sndSwitchStatus = sndSwchStatus} + instance SMPQueue RcvQueue where qServer RcvQueue {server} = server {-# INLINE qServer #-} @@ -155,10 +166,20 @@ findRQ :: (SMPServer, SMP.SenderId) -> NonEmpty RcvQueue -> Maybe RcvQueue findRQ sAddr = find $ sameQAddress sAddr . sndAddress {-# INLINE findRQ #-} +switchingRQ :: NonEmpty RcvQueue -> Maybe RcvQueue +switchingRQ = find $ isJust . rcvSwchStatus +{-# INLINE switchingRQ #-} + +updatedQs :: SMPQueueRec q => q -> NonEmpty q -> NonEmpty q +updatedQs q = L.map $ \q' -> if dbQId q == dbQId q' then q else q' +{-# INLINE updatedQs #-} + class SMPQueue q => SMPQueueRec q where qUserId :: q -> UserId qConnId :: q -> ConnId queueId :: q -> QueueId + dbQId :: q -> Int64 + dbReplaceQId :: q -> Maybe Int64 instance SMPQueueRec RcvQueue where qUserId = userId @@ -167,6 +188,10 @@ instance SMPQueueRec RcvQueue where {-# INLINE qConnId #-} queueId = rcvId {-# INLINE queueId #-} + dbQId = dbQueueId + {-# INLINE dbQId #-} + dbReplaceQId = dbReplaceQueueId + {-# INLINE dbReplaceQId #-} instance SMPQueueRec SndQueue where qUserId = userId @@ -175,6 +200,10 @@ instance SMPQueueRec SndQueue where {-# INLINE qConnId #-} queueId = sndId {-# INLINE queueId #-} + dbQId = dbQueueId + {-# INLINE dbQId #-} + dbReplaceQId = dbReplaceQueueId + {-# INLINE dbReplaceQId #-} -- * Connection types @@ -308,6 +337,7 @@ data InternalCommand | ICAllowSecure SMP.RecipientId SMP.SndPublicVerifyKey | ICDuplexSecure SMP.RecipientId SMP.SndPublicVerifyKey | ICDeleteConn + | ICDeleteRcvQueue SMP.RecipientId | ICQSecure SMP.RecipientId SMP.SndPublicVerifyKey | ICQDelete SMP.RecipientId @@ -317,6 +347,7 @@ data InternalCommandTag | ICAllowSecure_ | ICDuplexSecure_ | ICDeleteConn_ + | ICDeleteRcvQueue_ | ICQSecure_ | ICQDelete_ deriving (Show) @@ -328,6 +359,7 @@ instance StrEncoding InternalCommand where ICAllowSecure rId sndKey -> strEncode (ICAllowSecure_, rId, sndKey) ICDuplexSecure rId sndKey -> strEncode (ICDuplexSecure_, rId, sndKey) ICDeleteConn -> strEncode ICDeleteConn_ + ICDeleteRcvQueue rId -> strEncode (ICDeleteRcvQueue_, rId) ICQSecure rId senderKey -> strEncode (ICQSecure_, rId, senderKey) ICQDelete rId -> strEncode (ICQDelete_, rId) strP = @@ -337,6 +369,7 @@ instance StrEncoding InternalCommand where ICAllowSecure_ -> ICAllowSecure <$> _strP <*> _strP ICDuplexSecure_ -> ICDuplexSecure <$> _strP <*> _strP ICDeleteConn_ -> pure ICDeleteConn + ICDeleteRcvQueue_ -> ICDeleteRcvQueue <$> _strP ICQSecure_ -> ICQSecure <$> _strP <*> _strP ICQDelete_ -> ICQDelete <$> _strP @@ -347,6 +380,7 @@ instance StrEncoding InternalCommandTag where ICAllowSecure_ -> "ALLOW_SECURE" ICDuplexSecure_ -> "DUPLEX_SECURE" ICDeleteConn_ -> "DELETE_CONN" + ICDeleteRcvQueue_ -> "DELETE_RCV_QUEUE" ICQSecure_ -> "QSECURE" ICQDelete_ -> "QDELETE" strP = @@ -356,6 +390,7 @@ instance StrEncoding InternalCommandTag where "ALLOW_SECURE" -> pure ICAllowSecure_ "DUPLEX_SECURE" -> pure ICDuplexSecure_ "DELETE_CONN" -> pure ICDeleteConn_ + "DELETE_RCV_QUEUE" -> pure ICDeleteRcvQueue_ "QSECURE" -> pure ICQSecure_ "QDELETE" -> pure ICQDelete_ _ -> fail "bad InternalCommandTag" @@ -372,6 +407,7 @@ internalCmdTag = \case ICAllowSecure {} -> ICAllowSecure_ ICDuplexSecure {} -> ICDuplexSecure_ ICDeleteConn -> ICDeleteConn_ + ICDeleteRcvQueue {} -> ICDeleteRcvQueue_ ICQSecure {} -> ICQSecure_ ICQDelete _ -> ICQDelete_ diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 5e9df8562..aacb7d54b 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -54,14 +54,19 @@ module Simplex.Messaging.Agent.Store.SQLite setConnDeleted, getDeletedConnIds, getRcvConn, + getRcvQueueById, + getSndQueueById, deleteConn, upgradeRcvConnToDuplex, upgradeSndConnToDuplex, addConnRcvQueue, addConnSndQueue, setRcvQueueStatus, + setRcvSwitchStatus, + setRcvQueueDeleted, setRcvQueueConfirmedE2E, setSndQueueStatus, + setSndSwitchStatus, setRcvQueuePrimary, setSndQueuePrimary, deleteConnRcvQueue, @@ -69,6 +74,7 @@ module Simplex.Messaging.Agent.Store.SQLite deleteConnSndQueue, getPrimaryRcvQueue, getRcvQueue, + getDeletedRcvQueue, setRcvQueueNtfCreds, -- Confirmations createConfirmation, @@ -561,7 +567,7 @@ getRcvConn :: DB.Connection -> SMPServer -> SMP.RecipientId -> IO (Either StoreE getRcvConn db ProtocolServer {host, port} rcvId = runExceptT $ do rq@RcvQueue {connId} <- ExceptT . firstRow toRcvQueue SEConnNotFound $ - DB.query db (rcvQueueQuery <> " WHERE q.host = ? AND q.port = ? AND q.rcv_id = ?") (host, port, rcvId) + DB.query db (rcvQueueQuery <> " WHERE q.host = ? AND q.port = ? AND q.rcv_id = ? AND q.deleted = 0") (host, port, rcvId) (rq,) <$> ExceptT (getConn db connId) deleteConn :: DB.Connection -> ConnId -> IO () @@ -620,6 +626,29 @@ setRcvQueueStatus db RcvQueue {rcvId, server = ProtocolServer {host, port}} stat |] [":status" := status, ":host" := host, ":port" := port, ":rcv_id" := rcvId] +setRcvSwitchStatus :: DB.Connection -> RcvQueue -> Maybe RcvSwitchStatus -> IO RcvQueue +setRcvSwitchStatus db rq@RcvQueue {rcvId, server = ProtocolServer {host, port}} rcvSwchStatus = do + DB.execute + db + [sql| + UPDATE rcv_queues + SET switch_status = ? + WHERE host = ? AND port = ? AND rcv_id = ? + |] + (rcvSwchStatus, host, port, rcvId) + pure rq {rcvSwchStatus} + +setRcvQueueDeleted :: DB.Connection -> RcvQueue -> IO () +setRcvQueueDeleted db RcvQueue {rcvId, server = ProtocolServer {host, port}} = do + DB.execute + db + [sql| + UPDATE rcv_queues + SET deleted = 1 + WHERE host = ? AND port = ? AND rcv_id = ? + |] + (host, port, rcvId) + setRcvQueueConfirmedE2E :: DB.Connection -> RcvQueue -> C.DhSecretX25519 -> Version -> IO () setRcvQueueConfirmedE2E db RcvQueue {rcvId, server = ProtocolServer {host, port}} e2eDhSecret smpClientVersion = DB.executeNamed @@ -651,6 +680,18 @@ setSndQueueStatus db SndQueue {sndId, server = ProtocolServer {host, port}} stat |] [":status" := status, ":host" := host, ":port" := port, ":snd_id" := sndId] +setSndSwitchStatus :: DB.Connection -> SndQueue -> Maybe SndSwitchStatus -> IO SndQueue +setSndSwitchStatus db sq@SndQueue {sndId, server = ProtocolServer {host, port}} sndSwchStatus = do + DB.execute + db + [sql| + UPDATE snd_queues + SET switch_status = ? + WHERE host = ? AND port = ? AND snd_id = ? + |] + (sndSwchStatus, host, port, sndId) + pure sq {sndSwchStatus} + setRcvQueuePrimary :: DB.Connection -> ConnId -> RcvQueue -> IO () setRcvQueuePrimary db connId RcvQueue {dbQueueId} = do DB.execute db "UPDATE rcv_queues SET rcv_primary = ? WHERE conn_id = ?" (False, connId) @@ -687,7 +728,12 @@ getPrimaryRcvQueue db connId = getRcvQueue :: DB.Connection -> ConnId -> SMPServer -> SMP.RecipientId -> IO (Either StoreError RcvQueue) getRcvQueue db connId (SMPServer host port _) rcvId = firstRow toRcvQueue SEConnNotFound $ - DB.query db (rcvQueueQuery <> "WHERE q.conn_id = ? AND q.host = ? AND q.port = ? AND q.rcv_id = ?") (connId, host, port, rcvId) + DB.query db (rcvQueueQuery <> "WHERE q.conn_id = ? AND q.host = ? AND q.port = ? AND q.rcv_id = ? AND q.deleted = 0") (connId, host, port, rcvId) + +getDeletedRcvQueue :: DB.Connection -> ConnId -> SMPServer -> SMP.RecipientId -> IO (Either StoreError RcvQueue) +getDeletedRcvQueue db connId (SMPServer host port _) rcvId = + firstRow toRcvQueue SEConnNotFound $ + DB.query db (rcvQueueQuery <> "WHERE q.conn_id = ? AND q.host = ? AND q.port = ? AND q.rcv_id = ? AND q.deleted = 1") (connId, host, port, rcvId) setRcvQueueNtfCreds :: DB.Connection -> ConnId -> Maybe ClientNtfCreds -> IO () setRcvQueueNtfCreds db connId clientNtfCreds = @@ -918,7 +964,7 @@ setMsgUserAck db connId agentMsgId = runExceptT $ do (dbRcvId, srvMsgId) <- ExceptT . firstRow id SEMsgNotFound $ DB.query db "SELECT rcv_queue_id, broker_id FROM rcv_messages WHERE conn_id = ? AND internal_id = ?" (connId, agentMsgId) - rq <- ExceptT $ getRcvQueueById_ db connId dbRcvId + rq <- ExceptT $ getRcvQueueById db connId dbRcvId pure (rq, srvMsgId) getLastMsg :: DB.Connection -> ConnId -> SMP.MsgId -> IO (Maybe RcvMsg) @@ -1375,7 +1421,7 @@ getNtfRcvQueue db SMPQueueNtf {smpServer = (SMPServer host port _), notifierId} [sql| SELECT conn_id, rcv_ntf_dh_secret FROM rcv_queues - WHERE host = ? AND port = ? AND ntf_id = ? + WHERE host = ? AND port = ? AND ntf_id = ? AND deleted = 0 |] (host, port, notifierId) where @@ -1611,7 +1657,7 @@ getDeletedConnIds db = map fromOnly <$> DB.query db "SELECT conn_id FROM connect getRcvQueuesByConnId_ :: DB.Connection -> ConnId -> IO (Maybe (NonEmpty RcvQueue)) getRcvQueuesByConnId_ db connId = L.nonEmpty . sortBy primaryFirst . map toRcvQueue - <$> DB.query db (rcvQueueQuery <> "WHERE q.conn_id = ?") (Only connId) + <$> DB.query db (rcvQueueQuery <> "WHERE q.conn_id = ? AND q.deleted = 0") (Only connId) where primaryFirst RcvQueue {primary = p, dbReplaceQueueId = i} RcvQueue {primary = p', dbReplaceQueueId = i'} = -- the current primary queue is ordered first, the next primary - second @@ -1622,7 +1668,7 @@ rcvQueueQuery = [sql| SELECT c.user_id, COALESCE(q.server_key_hash, s.key_hash), q.conn_id, q.host, q.port, q.rcv_id, q.rcv_private_key, q.rcv_dh_secret, q.e2e_priv_key, q.e2e_dh_secret, q.snd_id, q.status, - q.rcv_queue_id, q.rcv_primary, q.replace_rcv_queue_id, q.smp_client_version, q.delete_errors, + q.rcv_queue_id, q.rcv_primary, q.replace_rcv_queue_id, q.switch_status, q.smp_client_version, q.delete_errors, q.ntf_public_key, q.ntf_private_key, q.ntf_id, q.rcv_ntf_dh_secret FROM rcv_queues q JOIN servers s ON q.host = s.host AND q.port = s.port @@ -1631,44 +1677,62 @@ rcvQueueQuery = toRcvQueue :: (UserId, C.KeyHash, ConnId, NonEmpty TransportHost, ServiceName, SMP.RecipientId, SMP.RcvPrivateSignKey, SMP.RcvDhSecret, C.PrivateKeyX25519, Maybe C.DhSecretX25519, SMP.SenderId, QueueStatus) - :. (Int64, Bool, Maybe Int64, Maybe Version, Int) + :. (Int64, Bool, Maybe Int64, Maybe RcvSwitchStatus, Maybe Version, Int) :. (Maybe SMP.NtfPublicVerifyKey, Maybe SMP.NtfPrivateSignKey, Maybe SMP.NotifierId, Maybe RcvNtfDhSecret) -> RcvQueue -toRcvQueue ((userId, keyHash, connId, host, port, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status) :. (dbQueueId, primary, dbReplaceQueueId, smpClientVersion_, deleteErrors) :. (ntfPublicKey_, ntfPrivateKey_, notifierId_, rcvNtfDhSecret_)) = +toRcvQueue ((userId, keyHash, connId, host, port, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status) :. (dbQueueId, primary, dbReplaceQueueId, rcvSwchStatus, smpClientVersion_, deleteErrors) :. (ntfPublicKey_, ntfPrivateKey_, notifierId_, rcvNtfDhSecret_)) = let server = SMPServer host port keyHash smpClientVersion = fromMaybe 1 smpClientVersion_ clientNtfCreds = case (ntfPublicKey_, ntfPrivateKey_, notifierId_, rcvNtfDhSecret_) of (Just ntfPublicKey, Just ntfPrivateKey, Just notifierId, Just rcvNtfDhSecret) -> Just $ ClientNtfCreds {ntfPublicKey, ntfPrivateKey, notifierId, rcvNtfDhSecret} _ -> Nothing - in RcvQueue {userId, connId, server, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status, dbQueueId, primary, dbReplaceQueueId, smpClientVersion, clientNtfCreds, deleteErrors} + in RcvQueue {userId, connId, server, rcvId, rcvPrivateKey, rcvDhSecret, e2ePrivKey, e2eDhSecret, sndId, status, dbQueueId, primary, dbReplaceQueueId, rcvSwchStatus, smpClientVersion, clientNtfCreds, deleteErrors} -getRcvQueueById_ :: DB.Connection -> ConnId -> Int64 -> IO (Either StoreError RcvQueue) -getRcvQueueById_ db connId dbRcvId = +getRcvQueueById :: DB.Connection -> ConnId -> Int64 -> IO (Either StoreError RcvQueue) +getRcvQueueById db connId dbRcvId = firstRow toRcvQueue SEConnNotFound $ - DB.query db (rcvQueueQuery <> " WHERE q.conn_id = ? AND q.rcv_queue_id = ?") (connId, dbRcvId) + DB.query db (rcvQueueQuery <> " WHERE q.conn_id = ? AND q.rcv_queue_id = ? AND q.deleted = 0") (connId, dbRcvId) -- | returns all connection queues, the first queue is the primary one getSndQueuesByConnId_ :: DB.Connection -> ConnId -> IO (Maybe (NonEmpty SndQueue)) getSndQueuesByConnId_ dbConn connId = - L.nonEmpty . sortBy primaryFirst . map sndQueue - <$> DB.query - dbConn - [sql| - SELECT c.user_id, COALESCE(q.server_key_hash, s.key_hash), q.host, q.port, q.snd_id, q.snd_public_key, q.snd_private_key, q.e2e_pub_key, q.e2e_dh_secret, q.status, q.snd_queue_id, q.snd_primary, q.replace_snd_queue_id, q.smp_client_version - FROM snd_queues q - JOIN servers s ON q.host = s.host AND q.port = s.port - JOIN connections c ON q.conn_id = c.conn_id - WHERE q.conn_id = ?; - |] - (Only connId) + L.nonEmpty . sortBy primaryFirst . map toSndQueue + <$> DB.query dbConn (sndQueueQuery <> "WHERE q.conn_id = ?") (Only connId) where - sndQueue ((userId, keyHash, host, port, sndId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status) :. (dbQueueId, primary, dbReplaceQueueId, smpClientVersion)) = - let server = SMPServer host port keyHash - in SndQueue {userId, connId, server, sndId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status, dbQueueId, primary, dbReplaceQueueId, smpClientVersion} primaryFirst SndQueue {primary = p, dbReplaceQueueId = i} SndQueue {primary = p', dbReplaceQueueId = i'} = -- the current primary queue is ordered first, the next primary - second compare (Down p) (Down p') <> compare i i' +sndQueueQuery :: Query +sndQueueQuery = + [sql| + SELECT + c.user_id, COALESCE(q.server_key_hash, s.key_hash), q.conn_id, q.host, q.port, q.snd_id, + q.snd_public_key, q.snd_private_key, q.e2e_pub_key, q.e2e_dh_secret, q.status, + q.snd_queue_id, q.snd_primary, q.replace_snd_queue_id, q.switch_status, q.smp_client_version + FROM snd_queues q + JOIN servers s ON q.host = s.host AND q.port = s.port + JOIN connections c ON q.conn_id = c.conn_id + |] + +toSndQueue :: + (UserId, C.KeyHash, ConnId, NonEmpty TransportHost, ServiceName, SenderId) + :. (Maybe C.APublicVerifyKey, SndPrivateSignKey, Maybe C.PublicKeyX25519, C.DhSecretX25519, QueueStatus) + :. (Int64, Bool, Maybe Int64, Maybe SndSwitchStatus, Version) -> + SndQueue +toSndQueue + ( (userId, keyHash, connId, host, port, sndId) + :. (sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status) + :. (dbQueueId, primary, dbReplaceQueueId, sndSwchStatus, smpClientVersion) + ) = + let server = SMPServer host port keyHash + in SndQueue {userId, connId, server, sndId, sndPublicKey, sndPrivateKey, e2ePubKey, e2eDhSecret, status, dbQueueId, primary, dbReplaceQueueId, sndSwchStatus, smpClientVersion} + +getSndQueueById :: DB.Connection -> ConnId -> Int64 -> IO (Either StoreError SndQueue) +getSndQueueById db connId dbSndId = + firstRow toSndQueue SEConnNotFound $ + DB.query db (sndQueueQuery <> " WHERE q.conn_id = ? AND q.snd_queue_id = ?") (connId, dbSndId) + -- * updateRcvIds helpers retrieveLastIdsAndHashRcv_ :: DB.Connection -> ConnId -> IO (InternalId, InternalRcvId, PrevExternalSndId, PrevRcvMsgHash) diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs index 07d7b8486..4e03c4e1a 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations.hs @@ -60,6 +60,7 @@ import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230320_retry_state import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230401_snd_files import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230510_files_pending_replicas_indexes import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230516_encrypted_rcv_message_hashes +import Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230531_switch_status import Simplex.Messaging.Encoding.String import Simplex.Messaging.Parsers (dropPrefix, sumTypeJSON) import Simplex.Messaging.Transport.Client (TransportHost) @@ -86,7 +87,8 @@ schemaMigrations = ("m20230320_retry_state", m20230320_retry_state, Just down_m20230320_retry_state), ("m20230401_snd_files", m20230401_snd_files, Just down_m20230401_snd_files), ("m20230510_files_pending_replicas_indexes", m20230510_files_pending_replicas_indexes, Just down_m20230510_files_pending_replicas_indexes), - ("m20230516_encrypted_rcv_message_hashes", m20230516_encrypted_rcv_message_hashes, Just down_m20230516_encrypted_rcv_message_hashes) + ("m20230516_encrypted_rcv_message_hashes", m20230516_encrypted_rcv_message_hashes, Just down_m20230516_encrypted_rcv_message_hashes), + ("m20230531_switch_status", m20230531_switch_status, Just down_m20230531_switch_status) ] -- | The list of migrations in ascending order by date diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20230531_switch_status.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20230531_switch_status.hs new file mode 100644 index 000000000..7211e3ab1 --- /dev/null +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20230531_switch_status.hs @@ -0,0 +1,22 @@ +{-# LANGUAGE QuasiQuotes #-} + +module Simplex.Messaging.Agent.Store.SQLite.Migrations.M20230531_switch_status where + +import Database.SQLite.Simple (Query) +import Database.SQLite.Simple.QQ (sql) + +m20230531_switch_status :: Query +m20230531_switch_status = + [sql| +ALTER TABLE rcv_queues ADD COLUMN switch_status TEXT; +ALTER TABLE rcv_queues ADD COLUMN deleted INTEGER NOT NULL DEFAULT 0; +ALTER TABLE snd_queues ADD COLUMN switch_status TEXT; +|] + +down_m20230531_switch_status :: Query +down_m20230531_switch_status = + [sql| +ALTER TABLE snd_queues DROP COLUMN switch_status; +ALTER TABLE rcv_queues DROP COLUMN deleted; +ALTER TABLE rcv_queues DROP COLUMN switch_status; +|] diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql index 922a1e912..c391e484c 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql @@ -50,6 +50,8 @@ CREATE TABLE rcv_queues( replace_rcv_queue_id INTEGER NULL, delete_errors INTEGER DEFAULT 0 CHECK(delete_errors NOT NULL), server_key_hash BLOB, + switch_status TEXT, + deleted INTEGER NOT NULL DEFAULT 0, PRIMARY KEY(host, port, rcv_id), FOREIGN KEY(host, port) REFERENCES servers ON DELETE RESTRICT ON UPDATE CASCADE, @@ -71,6 +73,7 @@ CREATE TABLE snd_queues( snd_primary INTEGER CHECK(snd_primary NOT NULL), replace_snd_queue_id INTEGER NULL, server_key_hash BLOB, + switch_status TEXT, PRIMARY KEY(host, port, snd_id), FOREIGN KEY(host, port) REFERENCES servers ON DELETE RESTRICT ON UPDATE CASCADE diff --git a/tests/AgentTests/FunctionalAPITests.hs b/tests/AgentTests/FunctionalAPITests.hs index fbf697b3b..de66c1d6d 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -125,6 +125,18 @@ runRight action = Right x -> pure x Left e -> error $ "Unexpected error: " <> show e +getInAnyOrder :: HasCallStack => AgentClient -> [AEntityTransmission 'AEConn -> Bool] -> Expectation +getInAnyOrder _ [] = pure () +getInAnyOrder c rs = do + r <- get c + let rest = filter (not . expected r) rs + if length rest < length rs + then getInAnyOrder c rest + else error $ "unexpected event: " <> show r + where + expected :: AEntityTransmission 'AEConn -> (AEntityTransmission 'AEConn -> Bool) -> Bool + expected r rp = rp r + functionalAPITests :: ATransport -> Spec functionalAPITests t = do describe "Establishing duplex connection" $ do @@ -191,13 +203,23 @@ functionalAPITests t = do testUsersNoServer t it "should connect two users and switch session mode" $ withSmpServer t testTwoUsers - describe "Queue rotation" $ do + describe "Connection switch" $ do describe "should switch delivery to the new queue" $ testServerMatrix2 t testSwitchConnection describe "should switch to new queue asynchronously" $ testServerMatrix2 t testSwitchAsync - describe "should delete connection during rotation" $ + describe "should delete connection during switch" $ testServerMatrix2 t testSwitchDelete + describe "should stop switch in Started phase" $ + testServerMatrix2 t testStopSwitchStarted + describe "should stop switch in Started phase, reinitiate immediately" $ + testServerMatrix2 t testStopSwitchStartedReinitiate + describe "should prohibit to stop switch in Secured phase" $ + testServerMatrix2 t testCannotStopSwitchSecured + describe "should switch two connections simultaneously" $ + testServerMatrix2 t testSwitch2Connections + describe "should switch two connections simultaneously, stop one" $ + testServerMatrix2 t testSwitch2ConnectionsStop1 describe "SMP basic auth" $ do describe "with server auth" $ do -- allow NEW | server auth, v | clnt1 auth, v | clnt2 auth, v | 2 - success, 1 - JOIN fail, 0 - NEW fail @@ -961,28 +983,48 @@ testSwitchConnection servers = do runRight_ $ do (aId, bId) <- makeConnection a b exchangeGreetingsMsgId 4 a bId b aId - switchConnectionAsync a "" bId - switchComplete a bId b aId - exchangeGreetingsMsgId 10 a bId b aId + testFullSwitch a bId b aId 10 + testFullSwitch a bId b aId 16 + +testFullSwitch :: AgentClient -> ByteString -> AgentClient -> ByteString -> Int64 -> ExceptT AgentErrorType IO () +testFullSwitch a bId b aId msgId = do + switchConnectionAsync a "" bId + switchComplete a bId b aId + exchangeGreetingsMsgId msgId a bId b aId switchComplete :: AgentClient -> ByteString -> AgentClient -> ByteString -> ExceptT AgentErrorType IO () switchComplete a bId b aId = do - phase a bId QDRcv SPStarted - phase b aId QDSnd SPStarted - phase a bId QDRcv SPConfirmed - phase b aId QDSnd SPConfirmed - phase b aId QDSnd SPCompleted - phase a bId QDRcv SPCompleted + phaseRcv a bId SPStarted [Just RSSendingQADD, Nothing] + phaseSnd b aId SPStarted [Just SSSendingQKEY, Nothing] + phaseSnd b aId SPConfirmed [Just SSSendingQKEY, Nothing] + phaseRcv a bId SPConfirmed [Just RSSendingQADD, Nothing] + phaseRcv a bId SPSecured [Just RSSendingQUSE, Nothing] + phaseSnd b aId SPSecured [Just SSSendingQTEST, Nothing] + phaseSnd b aId SPCompleted [Nothing] + phaseRcv a bId SPCompleted [Nothing] -phase :: AgentClient -> ByteString -> QueueDirection -> SwitchPhase -> ExceptT AgentErrorType IO () -phase c connId d p = +phaseRcv :: AgentClient -> ByteString -> SwitchPhase -> [Maybe RcvSwitchStatus] -> ExceptT AgentErrorType IO () +phaseRcv c connId p swchStatuses = phase c connId QDRcv p (\stats -> rcvSwchStatuses' stats `shouldMatchList` swchStatuses) + +rcvSwchStatuses' :: ConnectionStats -> [Maybe RcvSwitchStatus] +rcvSwchStatuses' ConnectionStats {rcvQueuesInfo} = map (\RcvQueueInfo {rcvSwitchStatus} -> rcvSwitchStatus) rcvQueuesInfo + +phaseSnd :: AgentClient -> ByteString -> SwitchPhase -> [Maybe SndSwitchStatus] -> ExceptT AgentErrorType IO () +phaseSnd c connId p swchStatuses = phase c connId QDSnd p (\stats -> sndSwchStatuses' stats `shouldMatchList` swchStatuses) + +sndSwchStatuses' :: ConnectionStats -> [Maybe SndSwitchStatus] +sndSwchStatuses' ConnectionStats {sndQueuesInfo} = map (\SndQueueInfo {sndSwitchStatus} -> sndSwitchStatus) sndQueuesInfo + +phase :: AgentClient -> ByteString -> QueueDirection -> SwitchPhase -> (ConnectionStats -> Expectation) -> ExceptT AgentErrorType IO () +phase c connId d p statsExpectation = get c >>= \(_, connId', msg) -> do liftIO $ connId `shouldBe` connId' case msg of - SWITCH d' p' _ -> liftIO $ do + SWITCH d' p' stats -> liftIO $ do d `shouldBe` d' p `shouldBe` p' - ERR (AGENT A_DUPLICATE) -> phase c connId d p + statsExpectation stats + ERR (AGENT A_DUPLICATE) -> phase c connId d p statsExpectation r -> do liftIO . putStrLn $ "expected: " <> show p <> ", received: " <> show r SWITCH {} <- pure r @@ -994,33 +1036,43 @@ testSwitchAsync servers = do (aId, bId) <- makeConnection a b exchangeGreetingsMsgId 4 a bId b aId pure (aId, bId) - let withA' = session withA bId - withB' = session withB aId + let withA' = sessionSubscribe withA [bId] + withB' = sessionSubscribe withB [aId] withA' $ \a -> do switchConnectionAsync a "" bId - phase a bId QDRcv SPStarted - withB' $ \b -> phase b aId QDSnd SPStarted - withA' $ \a -> phase a bId QDRcv SPConfirmed + phaseRcv a bId SPStarted [Just RSSendingQADD, Nothing] withB' $ \b -> do - phase b aId QDSnd SPConfirmed - phase b aId QDSnd SPCompleted - withA' $ \a -> phase a bId QDRcv SPCompleted + phaseSnd b aId SPStarted [Just SSSendingQKEY, Nothing] + phaseSnd b aId SPConfirmed [Just SSSendingQKEY, Nothing] + withA' $ \a -> do + phaseRcv a bId SPConfirmed [Just RSSendingQADD, Nothing] + phaseRcv a bId SPSecured [Just RSSendingQUSE, Nothing] + withB' $ \b -> do + phaseSnd b aId SPSecured [Just SSSendingQTEST, Nothing] + phaseSnd b aId SPCompleted [Nothing] + withA' $ \a -> phaseRcv a bId SPCompleted [Nothing] withA $ \a -> withB $ \b -> runRight_ $ do subscribeConnection a bId subscribeConnection b aId exchangeGreetingsMsgId 10 a bId b aId + testFullSwitch a bId b aId 16 where - withAgent :: AgentConfig -> FilePath -> (AgentClient -> IO a) -> IO a - withAgent cfg' dbPath = bracket (getSMPAgentClient' cfg' servers dbPath) disconnectAgentClient - session :: (forall a. (AgentClient -> IO a) -> IO a) -> ConnId -> (AgentClient -> ExceptT AgentErrorType IO ()) -> IO () - session withC connId a = - withC $ \c -> runRight_ $ do - subscribeConnection c connId - r <- a c - liftIO $ threadDelay 500000 - pure r - withA = withAgent agentCfg testDB - withB = withAgent agentCfg {initialClientId = 1} testDB2 + withA :: (AgentClient -> IO a) -> IO a + withA = withAgent agentCfg servers testDB + withB :: (AgentClient -> IO a) -> IO a + withB = withAgent agentCfg {initialClientId = 1} servers testDB2 + +withAgent :: AgentConfig -> InitialAgentServers -> FilePath -> (AgentClient -> IO a) -> IO a +withAgent cfg' servers dbPath = bracket (getSMPAgentClient' cfg' servers dbPath) disconnectAgentClient + +sessionSubscribe :: (forall a. (AgentClient -> IO a) -> IO a) -> [ConnId] -> (AgentClient -> ExceptT AgentErrorType IO ()) -> IO () +sessionSubscribe withC connIds a = + withC $ \c -> runRight_ $ do + void $ subscribeConnections c connIds + r <- a c + liftIO $ threadDelay 500000 + liftIO $ noMessages c "nothing else should be delivered" + pure r testSwitchDelete :: InitialAgentServers -> IO () testSwitchDelete servers = do @@ -1031,13 +1083,269 @@ testSwitchDelete servers = do exchangeGreetingsMsgId 4 a bId b aId disconnectAgentClient b switchConnectionAsync a "" bId - phase a bId QDRcv SPStarted + phaseRcv a bId SPStarted [Just RSSendingQADD, Nothing] deleteConnectionAsync a bId get a =##> \case ("", c, DEL_RCVQ _ _ Nothing) -> c == bId; _ -> False get a =##> \case ("", c, DEL_RCVQ _ _ Nothing) -> c == bId; _ -> False get a =##> \case ("", c, DEL_CONN) -> c == bId; _ -> False liftIO $ noMessages a "nothing else should be delivered to alice" +testStopSwitchStarted :: HasCallStack => InitialAgentServers -> IO () +testStopSwitchStarted servers = do + (aId, bId) <- withA $ \a -> withB $ \b -> runRight $ do + (aId, bId) <- makeConnection a b + exchangeGreetingsMsgId 4 a bId b aId + pure (aId, bId) + let withA' = sessionSubscribe withA [bId] + withB' = sessionSubscribe withB [aId] + withA' $ \a -> do + switchConnectionAsync a "" bId + phaseRcv a bId SPStarted [Just RSSendingQADD, Nothing] + -- repeat switch is prohibited + Left Agent.CMD {cmdErr = PROHIBITED} <- runExceptT $ switchConnectionAsync a "" bId + -- stop current switch + stats <- stopConnectionSwitch a bId + liftIO $ rcvSwchStatuses' stats `shouldMatchList` [Nothing] + withB' $ \b -> do + phaseSnd b aId SPStarted [Just SSSendingQKEY, Nothing] + phaseSnd b aId SPConfirmed [Just SSSendingQKEY, Nothing] + withA' $ \a -> do + get a ##> ("", bId, ERR (AGENT {agentErr = A_QUEUE {queueErr = "QKEY: queue address not found in connection"}})) + -- repeat switch + switchConnectionAsync a "" bId + phaseRcv a bId SPStarted [Just RSSendingQADD, Nothing] + withA $ \a -> withB $ \b -> runRight_ $ do + subscribeConnection a bId + subscribeConnection b aId + + phaseSnd b aId SPStarted [Just SSSendingQKEY, Nothing] + phaseSnd b aId SPConfirmed [Just SSSendingQKEY, Nothing] + + phaseRcv a bId SPConfirmed [Just RSSendingQADD, Nothing] + phaseRcv a bId SPSecured [Just RSSendingQUSE, Nothing] + + phaseSnd b aId SPSecured [Just SSSendingQTEST, Nothing] + phaseSnd b aId SPCompleted [Nothing] + + phaseRcv a bId SPCompleted [Nothing] + + exchangeGreetingsMsgId 12 a bId b aId + + testFullSwitch a bId b aId 18 + where + withA :: (AgentClient -> IO a) -> IO a + withA = withAgent agentCfg servers testDB + withB :: (AgentClient -> IO a) -> IO a + withB = withAgent agentCfg {initialClientId = 1} servers testDB2 + +testStopSwitchStartedReinitiate :: HasCallStack => InitialAgentServers -> IO () +testStopSwitchStartedReinitiate servers = do + (aId, bId) <- withA $ \a -> withB $ \b -> runRight $ do + (aId, bId) <- makeConnection a b + exchangeGreetingsMsgId 4 a bId b aId + pure (aId, bId) + let withA' = sessionSubscribe withA [bId] + withB' = sessionSubscribe withB [aId] + withA' $ \a -> do + switchConnectionAsync a "" bId + phaseRcv a bId SPStarted [Just RSSendingQADD, Nothing] + -- stop current switch + stats <- stopConnectionSwitch a bId + liftIO $ rcvSwchStatuses' stats `shouldMatchList` [Nothing] + -- repeat switch + switchConnectionAsync a "" bId + phaseRcv a bId SPStarted [Just RSSendingQADD, Nothing] + withB' $ \b -> do + phaseSnd b aId SPStarted [Just SSSendingQKEY, Nothing] + liftIO . getInAnyOrder b $ + [ switchPhaseSndP aId SPStarted [Just SSSendingQKEY, Nothing], + switchPhaseSndP aId SPConfirmed [Just SSSendingQKEY, Nothing] + ] + phaseSnd b aId SPConfirmed [Just SSSendingQKEY, Nothing] + withA $ \a -> withB $ \b -> runRight_ $ do + subscribeConnection a bId + subscribeConnection b aId + + liftIO . getInAnyOrder a $ + [ errQueueNotFoundP bId, + switchPhaseRcvP bId SPConfirmed [Just RSSendingQADD, Nothing] + ] + + phaseRcv a bId SPSecured [Just RSSendingQUSE, Nothing] + + phaseSnd b aId SPSecured [Just SSSendingQTEST, Nothing] + phaseSnd b aId SPCompleted [Nothing] + + phaseRcv a bId SPCompleted [Nothing] + + exchangeGreetingsMsgId 12 a bId b aId + + testFullSwitch a bId b aId 18 + where + withA :: (AgentClient -> IO a) -> IO a + withA = withAgent agentCfg servers testDB + withB :: (AgentClient -> IO a) -> IO a + withB = withAgent agentCfg {initialClientId = 1} servers testDB2 + +switchPhaseRcvP :: ConnId -> SwitchPhase -> [Maybe RcvSwitchStatus] -> AEntityTransmission 'AEConn -> Bool +switchPhaseRcvP cId sphase swchStatuses = switchPhaseP cId QDRcv sphase (\stats -> rcvSwchStatuses' stats == swchStatuses) + +switchPhaseSndP :: ConnId -> SwitchPhase -> [Maybe SndSwitchStatus] -> AEntityTransmission 'AEConn -> Bool +switchPhaseSndP cId sphase swchStatuses = switchPhaseP cId QDSnd sphase (\stats -> sndSwchStatuses' stats == swchStatuses) + +switchPhaseP :: ConnId -> QueueDirection -> SwitchPhase -> (ConnectionStats -> Bool) -> AEntityTransmission 'AEConn -> Bool +switchPhaseP cId qd sphase statsP = \case + (_, cId', SWITCH qd' sphase' stats) -> cId' == cId && qd' == qd && sphase' == sphase && statsP stats + _ -> False + +errQueueNotFoundP :: ConnId -> AEntityTransmission 'AEConn -> Bool +errQueueNotFoundP cId = \case + (_, cId', ERR AGENT {agentErr = A_QUEUE {queueErr = "QKEY: queue address not found in connection"}}) -> cId' == cId + _ -> False + +testCannotStopSwitchSecured :: HasCallStack => InitialAgentServers -> IO () +testCannotStopSwitchSecured servers = do + (aId, bId) <- withA $ \a -> withB $ \b -> runRight $ do + (aId, bId) <- makeConnection a b + exchangeGreetingsMsgId 4 a bId b aId + pure (aId, bId) + let withA' = sessionSubscribe withA [bId] + withB' = sessionSubscribe withB [aId] + withA' $ \a -> do + switchConnectionAsync a "" bId + phaseRcv a bId SPStarted [Just RSSendingQADD, Nothing] + withB' $ \b -> do + phaseSnd b aId SPStarted [Just SSSendingQKEY, Nothing] + phaseSnd b aId SPConfirmed [Just SSSendingQKEY, Nothing] + withA' $ \a -> do + phaseRcv a bId SPConfirmed [Just RSSendingQADD, Nothing] + phaseRcv a bId SPSecured [Just RSSendingQUSE, Nothing] + Left Agent.CMD {cmdErr = PROHIBITED} <- runExceptT $ stopConnectionSwitch a bId + pure () + withA $ \a -> withB $ \b -> runRight_ $ do + subscribeConnection a bId + subscribeConnection b aId + + phaseSnd b aId SPSecured [Just SSSendingQTEST, Nothing] + phaseSnd b aId SPCompleted [Nothing] + + phaseRcv a bId SPCompleted [Nothing] + + exchangeGreetingsMsgId 10 a bId b aId + + testFullSwitch a bId b aId 16 + where + withA :: (AgentClient -> IO a) -> IO a + withA = withAgent agentCfg servers testDB + withB :: (AgentClient -> IO a) -> IO a + withB = withAgent agentCfg {initialClientId = 1} servers testDB2 + +testSwitch2Connections :: HasCallStack => InitialAgentServers -> IO () +testSwitch2Connections servers = do + (aId1, bId1, aId2, bId2) <- withA $ \a -> withB $ \b -> runRight $ do + (aId1, bId1) <- makeConnection a b + exchangeGreetingsMsgId 4 a bId1 b aId1 + (aId2, bId2) <- makeConnection a b + exchangeGreetingsMsgId 4 a bId2 b aId2 + pure (aId1, bId1, aId2, bId2) + withA $ \a -> runRight_ $ do + void $ subscribeConnections a [bId1, bId2] + switchConnectionAsync a "" bId1 + phaseRcv a bId1 SPStarted [Just RSSendingQADD, Nothing] + switchConnectionAsync a "" bId2 + phaseRcv a bId2 SPStarted [Just RSSendingQADD, Nothing] + withA $ \a -> withB $ \b -> runRight_ $ do + void $ subscribeConnections a [bId1, bId2] + void $ subscribeConnections b [aId1, aId2] + + liftIO . getInAnyOrder b $ + [ switchPhaseSndP aId1 SPStarted [Just SSSendingQKEY, Nothing], + switchPhaseSndP aId1 SPConfirmed [Just SSSendingQKEY, Nothing], + switchPhaseSndP aId2 SPStarted [Just SSSendingQKEY, Nothing], + switchPhaseSndP aId2 SPConfirmed [Just SSSendingQKEY, Nothing] + ] + + liftIO . getInAnyOrder a $ + [ switchPhaseRcvP bId1 SPConfirmed [Just RSSendingQADD, Nothing], + switchPhaseRcvP bId1 SPSecured [Just RSSendingQUSE, Nothing], + switchPhaseRcvP bId2 SPConfirmed [Just RSSendingQADD, Nothing], + switchPhaseRcvP bId2 SPSecured [Just RSSendingQUSE, Nothing] + ] + + liftIO . getInAnyOrder b $ + [ switchPhaseSndP aId1 SPSecured [Just SSSendingQTEST, Nothing], + switchPhaseSndP aId1 SPCompleted [Nothing], + switchPhaseSndP aId2 SPSecured [Just SSSendingQTEST, Nothing], + switchPhaseSndP aId2 SPCompleted [Nothing] + ] + + liftIO . getInAnyOrder a $ + [ switchPhaseRcvP bId1 SPCompleted [Nothing], + switchPhaseRcvP bId2 SPCompleted [Nothing] + ] + + exchangeGreetingsMsgId 10 a bId1 b aId1 + exchangeGreetingsMsgId 10 a bId2 b aId2 + + testFullSwitch a bId1 b aId1 16 + testFullSwitch a bId2 b aId2 16 + where + withA :: (AgentClient -> IO a) -> IO a + withA = withAgent agentCfg servers testDB + withB :: (AgentClient -> IO a) -> IO a + withB = withAgent agentCfg {initialClientId = 1} servers testDB2 + +testSwitch2ConnectionsStop1 :: HasCallStack => InitialAgentServers -> IO () +testSwitch2ConnectionsStop1 servers = do + (aId1, bId1, aId2, bId2) <- withA $ \a -> withB $ \b -> runRight $ do + (aId1, bId1) <- makeConnection a b + exchangeGreetingsMsgId 4 a bId1 b aId1 + (aId2, bId2) <- makeConnection a b + exchangeGreetingsMsgId 4 a bId2 b aId2 + pure (aId1, bId1, aId2, bId2) + let withA' = sessionSubscribe withA [bId1, bId2] + withB' = sessionSubscribe withB [aId1, aId2] + withA' $ \a -> do + switchConnectionAsync a "" bId1 + phaseRcv a bId1 SPStarted [Just RSSendingQADD, Nothing] + switchConnectionAsync a "" bId2 + phaseRcv a bId2 SPStarted [Just RSSendingQADD, Nothing] + -- stop switch of second connection + stats <- stopConnectionSwitch a bId2 + liftIO $ rcvSwchStatuses' stats `shouldMatchList` [Nothing] + withB' $ \b -> do + liftIO . getInAnyOrder b $ + [ switchPhaseSndP aId1 SPStarted [Just SSSendingQKEY, Nothing], + switchPhaseSndP aId1 SPConfirmed [Just SSSendingQKEY, Nothing], + switchPhaseSndP aId2 SPStarted [Just SSSendingQKEY, Nothing], + switchPhaseSndP aId2 SPConfirmed [Just SSSendingQKEY, Nothing] + ] + withA' $ \a -> do + liftIO . getInAnyOrder a $ + [ switchPhaseRcvP bId1 SPConfirmed [Just RSSendingQADD, Nothing], + switchPhaseRcvP bId1 SPSecured [Just RSSendingQUSE, Nothing], + errQueueNotFoundP bId2 + ] + withA $ \a -> withB $ \b -> runRight_ $ do + void $ subscribeConnections a [bId1, bId2] + void $ subscribeConnections b [aId1, aId2] + + phaseSnd b aId1 SPSecured [Just SSSendingQTEST, Nothing] + phaseSnd b aId1 SPCompleted [Nothing] + + phaseRcv a bId1 SPCompleted [Nothing] + + exchangeGreetingsMsgId 10 a bId1 b aId1 + exchangeGreetingsMsgId 8 a bId2 b aId2 + + testFullSwitch a bId1 b aId1 16 + testFullSwitch a bId2 b aId2 14 + where + withA :: (AgentClient -> IO a) -> IO a + withA = withAgent agentCfg servers testDB + withB :: (AgentClient -> IO a) -> IO a + withB = withAgent agentCfg {initialClientId = 1} servers testDB2 + testCreateQueueAuth :: (Maybe BasicAuth, Version) -> (Maybe BasicAuth, Version) -> IO Int testCreateQueueAuth clnt1 clnt2 = do a <- getClient clnt1 diff --git a/tests/AgentTests/SQLiteTests.hs b/tests/AgentTests/SQLiteTests.hs index 43c95e0b8..89b54face 100644 --- a/tests/AgentTests/SQLiteTests.hs +++ b/tests/AgentTests/SQLiteTests.hs @@ -167,6 +167,7 @@ rcvQueue1 = dbQueueId = 1, primary = True, dbReplaceQueueId = Nothing, + rcvSwchStatus = Nothing, smpClientVersion = 1, clientNtfCreds = Nothing, deleteErrors = 0 @@ -187,6 +188,7 @@ sndQueue1 = dbQueueId = 1, primary = True, dbReplaceQueueId = Nothing, + sndSwchStatus = Nothing, smpClientVersion = 1 } @@ -324,6 +326,7 @@ testUpgradeRcvConnToDuplex = e2eDhSecret = testDhSecret, status = New, dbQueueId = 1, + sndSwchStatus = Nothing, primary = True, dbReplaceQueueId = Nothing, smpClientVersion = 1 @@ -352,6 +355,7 @@ testUpgradeSndConnToDuplex = sndId = "4567", status = New, dbQueueId = 1, + rcvSwchStatus = Nothing, primary = True, dbReplaceQueueId = Nothing, smpClientVersion = 1,