From bab6a1577b4b01d375239702aa6d16d78c45825d Mon Sep 17 00:00:00 2001 From: Evgeny Poberezkin <2769109+epoberezkin@users.noreply.github.com> Date: Sat, 10 Sep 2022 16:33:15 +0100 Subject: [PATCH] correlation IDs and command tags for async commands (#519) * agent protocol command tags * store/send async command correlation IDs * fix, refactor * delete failed command --- src/Simplex/Messaging/Agent.hs | 74 +++--- src/Simplex/Messaging/Agent/Protocol.hs | 240 ++++++++++++++---- src/Simplex/Messaging/Agent/Store/SQLite.hs | 32 +-- .../SQLite/Migrations/M20220905_commands.hs | 6 +- .../Store/SQLite/Migrations/agent_schema.sql | 6 +- tests/AgentTests/FunctionalAPITests.hs | 29 ++- 6 files changed, 269 insertions(+), 118 deletions(-) diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index af931dcc9..57d53e406 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -83,7 +83,7 @@ import Control.Monad.Reader import Crypto.Random (MonadRandom) import Data.Bifunctor (bimap, first, second) import Data.ByteString.Char8 (ByteString) -import Data.Composition ((.:), (.:.)) +import Data.Composition ((.:), (.:.), (.::)) import Data.Functor (($>)) import Data.List (deleteFirstsBy) import Data.List.NonEmpty (NonEmpty (..)) @@ -146,20 +146,20 @@ resumeAgentClient c = atomically $ writeTVar (active c) True type AgentErrorMonad m = (MonadUnliftIO m, MonadError AgentErrorType m) -- | Create SMP agent connection (NEW command) asynchronously, synchronous response is new connection id -createConnectionAsync :: forall m c. (AgentErrorMonad m, ConnectionModeI c) => AgentClient -> Bool -> SConnectionMode c -> m ConnId -createConnectionAsync c enableNtfs cMode = withAgentEnv c $ newConnAsync c enableNtfs cMode +createConnectionAsync :: forall m c. (AgentErrorMonad m, ConnectionModeI c) => AgentClient -> ACorrId -> Bool -> SConnectionMode c -> m ConnId +createConnectionAsync c corrId enableNtfs cMode = withAgentEnv c $ newConnAsync c corrId enableNtfs cMode -- | Join SMP agent connection (JOIN command) asynchronously, synchronous response is new connection id -joinConnectionAsync :: AgentErrorMonad m => AgentClient -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId -joinConnectionAsync c enableNtfs = withAgentEnv c .: joinConnAsync c enableNtfs +joinConnectionAsync :: AgentErrorMonad m => AgentClient -> ACorrId -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId +joinConnectionAsync c corrId enableNtfs = withAgentEnv c .: joinConnAsync c corrId enableNtfs -- | Allow connection to continue after CONF notification (LET command), no synchronous response -allowConnectionAsync :: AgentErrorMonad m => AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> m () -allowConnectionAsync c = withAgentEnv c .:. allowConnectionAsync' c +allowConnectionAsync :: AgentErrorMonad m => AgentClient -> ACorrId -> ConnId -> ConfirmationId -> ConnInfo -> m () +allowConnectionAsync c = withAgentEnv c .:: allowConnectionAsync' c -- | Acknowledge message (ACK command) asynchronously, no synchronous response -ackMessageAsync :: forall m. AgentErrorMonad m => AgentClient -> ConnId -> AgentMsgId -> m () -ackMessageAsync c = withAgentEnv c .: ackMessageAsync' c +ackMessageAsync :: forall m. AgentErrorMonad m => AgentClient -> ACorrId -> ConnId -> AgentMsgId -> m () +ackMessageAsync c = withAgentEnv c .:. ackMessageAsync' c -- | Create SMP agent connection (NEW command) createConnection :: AgentErrorMonad m => AgentClient -> Bool -> SConnectionMode c -> m (ConnId, ConnectionRequestUri c) @@ -315,17 +315,17 @@ processCommand c (connId, cmd) = case cmd of DEL -> deleteConnection' c connId $> (connId, OK) CHK -> (connId,) . STAT <$> getConnectionServers' c connId -newConnAsync :: forall m c. (AgentMonad m, ConnectionModeI c) => AgentClient -> Bool -> SConnectionMode c -> m ConnId -newConnAsync c enableNtfs cMode = do +newConnAsync :: forall m c. (AgentMonad m, ConnectionModeI c) => AgentClient -> ACorrId -> Bool -> SConnectionMode c -> m ConnId +newConnAsync c corrId enableNtfs cMode = do g <- asks idsDrg connAgentVersion <- asks $ maxVersion . smpAgentVRange . config let cData = ConnData {connId = "", connAgentVersion, enableNtfs, duplexHandshake = Nothing} -- connection mode is determined by the accepting agent connId <- withStore c $ \db -> createNewConn db g cData cMode - enqueueCommand c connId Nothing $ NEW enableNtfs (ACM cMode) + enqueueCommand c corrId connId Nothing $ NEW enableNtfs (ACM cMode) pure connId -joinConnAsync :: AgentMonad m => AgentClient -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId -joinConnAsync c enableNtfs cReqUri@(CRInvitationUri (ConnReqUriData _ agentVRange _) _) cInfo = do +joinConnAsync :: AgentMonad m => AgentClient -> ACorrId -> Bool -> ConnectionRequestUri c -> ConnInfo -> m ConnId +joinConnAsync c corrId enableNtfs cReqUri@(CRInvitationUri (ConnReqUriData _ agentVRange _) _) cInfo = do aVRange <- asks $ smpAgentVRange . config case agentVRange `compatibleVersion` aVRange of Just (Compatible connAgentVersion) -> do @@ -333,21 +333,21 @@ joinConnAsync c enableNtfs cReqUri@(CRInvitationUri (ConnReqUriData _ agentVRang let duplexHS = connAgentVersion /= 1 cData = ConnData {connId = "", connAgentVersion, enableNtfs, duplexHandshake = Just duplexHS} connId <- withStore c $ \db -> createNewConn db g cData SCMInvitation - enqueueCommand c connId Nothing $ JOIN enableNtfs (ACR sConnectionMode cReqUri) cInfo + enqueueCommand c corrId connId Nothing $ JOIN enableNtfs (ACR sConnectionMode cReqUri) cInfo pure connId _ -> throwError $ AGENT A_VERSION -joinConnAsync _c _enableNtfs (CRContactUri _) _cInfo = +joinConnAsync _c _corrId _enableNtfs (CRContactUri _) _cInfo = throwError $ CMD PROHIBITED -allowConnectionAsync' :: AgentMonad m => AgentClient -> ConnId -> ConfirmationId -> ConnInfo -> m () -allowConnectionAsync' c connId confId ownConnInfo = +allowConnectionAsync' :: AgentMonad m => AgentClient -> ACorrId -> ConnId -> ConfirmationId -> ConnInfo -> m () +allowConnectionAsync' c corrId connId confId ownConnInfo = withStore c (`getConn` connId) >>= \case SomeConn _ (RcvConnection _ RcvQueue {server}) -> - enqueueCommand c connId (Just server) $ LET confId ownConnInfo + enqueueCommand c corrId connId (Just server) $ LET confId ownConnInfo _ -> throwError $ CMD PROHIBITED -ackMessageAsync' :: forall m. AgentMonad m => AgentClient -> ConnId -> AgentMsgId -> m () -ackMessageAsync' c connId msgId = +ackMessageAsync' :: forall m. AgentMonad m => AgentClient -> ACorrId -> ConnId -> AgentMsgId -> m () +ackMessageAsync' c corrId connId msgId = withStore c (`getConn` connId) >>= \case SomeConn _ (DuplexConnection _ rq _) -> enqueueAck rq SomeConn _ (RcvConnection _ rq) -> enqueueAck rq @@ -357,7 +357,7 @@ ackMessageAsync' c connId msgId = where enqueueAck :: RcvQueue -> m () enqueueAck RcvQueue {server} = do - enqueueCommand c connId (Just server) $ ACK msgId + enqueueCommand c corrId connId (Just server) $ ACK msgId newConn :: AgentMonad m => AgentClient -> ConnId -> Bool -> Bool -> SConnectionMode c -> m (ConnId, ConnectionRequestUri c) newConn c connId asyncMode enableNtfs cMode = @@ -638,10 +638,10 @@ sendMessage' c connId msgFlags msg = -- / async command processing v v v -enqueueCommand :: forall m. AgentMonad m => AgentClient -> ConnId -> Maybe SMPServer -> ACommand 'Client -> m () -enqueueCommand c connId server aCommand = do +enqueueCommand :: forall m. AgentMonad m => AgentClient -> ACorrId -> ConnId -> Maybe SMPServer -> ACommand 'Client -> m () +enqueueCommand c corrId connId server aCommand = do resumeSrvCmds c server - commandId <- withStore c $ \db -> runExceptT . liftIO $ createCommand db connId server aCommand + commandId <- withStore' c $ \db -> createCommand db corrId connId server aCommand queuePendingCommands c server [commandId] resumeSrvCmds :: forall m. AgentMonad m => AgentClient -> Maybe SMPServer -> m () @@ -688,36 +688,38 @@ runCommandProcessing c@AgentClient {subQ} server = do cmdId <- atomically $ readTQueue cq atomically $ beginAgentOperation c AOSndNetwork E.try (withStore c $ \db -> getPendingCommand db cmdId) >>= \case - Left (e :: E.SomeException) -> notify "" $ ERR (INTERNAL $ show e) - Right (connId, ACmd _ cmd) -> processCmd ri connId cmdId cmd + Left (e :: E.SomeException) -> atomically $ writeTBQueue subQ ("", "", ERR . INTERNAL $ show e) + Right (corrId, connId, ACmd _ cmd) -> processCmd ri corrId connId cmdId cmd where - processCmd :: RetryInterval -> ConnId -> AsyncCmdId -> ACommand p -> m () - processCmd ri connId cmdId = \case + processCmd :: RetryInterval -> ACorrId -> ConnId -> AsyncCmdId -> ACommand p -> m () + processCmd ri corrId connId cmdId = \case NEW enableNtfs (ACM cMode) -> do usedSrvs <- newTVarIO ([] :: [SMPServer]) tryCommand . withNextSrv usedSrvs [] $ \srv -> do (_, cReq) <- newConnSrv c connId True enableNtfs cMode srv - notify connId $ INV (ACR cMode cReq) + notify $ INV (ACR cMode cReq) JOIN enableNtfs (ACR _ cReq@(CRInvitationUri ConnReqUriData {crSmpQueues = SMPQueueUri {queueAddress} :| _} _)) connInfo -> do let initUsed = [smpServer (queueAddress :: SMPQueueAddress)] usedSrvs <- newTVarIO initUsed - tryCommand . withNextSrv usedSrvs initUsed $ \srv -> + tryCommand . withNextSrv usedSrvs initUsed $ \srv -> do void $ joinConnSrv c connId True enableNtfs cReq connInfo srv - LET confId ownCInfo -> tryCommand $ allowConnection' c connId confId ownCInfo - ACK msgId -> tryCommand $ ackMessage' c connId msgId - cmd -> notify connId $ ERR $ INTERNAL $ "unsupported async command " <> show cmd + notify OK + LET confId ownCInfo -> tryCommand $ allowConnection' c connId confId ownCInfo >> notify OK + ACK msgId -> tryCommand $ ackMessage' c connId msgId >> notify OK + cmd -> notify $ ERR $ INTERNAL $ "unsupported async command " <> show (aCommandTag cmd) where tryCommand action = withRetryInterval ri $ \loop -> tryError action >>= \case Left e | temporaryAgentError e || e == BROKER HOST -> retryCommand loop - | otherwise -> notify connId $ ERR e + | otherwise -> notify (ERR e) >> withStore' c (`deleteCommand` cmdId) Right () -> withStore' c (`deleteCommand` cmdId) retryCommand loop = do -- end... is in a separate atomically because if begin... blocks, SUSPENDED won't be sent atomically $ endAgentOperation c AOSndNetwork atomically $ beginAgentOperation c AOSndNetwork loop + notify cmd = atomically $ writeTBQueue subQ (corrId, connId, cmd) withNextSrv :: TVar [SMPServer] -> [SMPServer] -> (SMPServer -> m ()) -> m () withNextSrv usedSrvs initUsed action = do used <- readTVarIO usedSrvs @@ -727,8 +729,6 @@ runCommandProcessing c@AgentClient {subQ} server = do let used' = if length used + 1 >= L.length srvs then initUsed else srv : used writeTVar usedSrvs used' action srv - notify :: ConnId -> ACommand 'Agent -> m () - notify connId cmd = atomically $ writeTBQueue subQ ("", connId, cmd) -- ^ ^ ^ async command processing / enqueueMessage :: forall m. AgentMonad m => AgentClient -> ConnData -> SndQueue -> MsgFlags -> AMessage -> m AgentMsgId diff --git a/src/Simplex/Messaging/Agent/Protocol.hs b/src/Simplex/Messaging/Agent/Protocol.hs index 9187381e5..7579f13bc 100644 --- a/src/Simplex/Messaging/Agent/Protocol.hs +++ b/src/Simplex/Messaging/Agent/Protocol.hs @@ -40,9 +40,13 @@ module Simplex.Messaging.Agent.Protocol -- * SMP agent protocol types ConnInfo, ACommand (..), + ACommandTag (..), + aCommandTag, ACmd (..), + ACmdTag (..), AParty (..), SAParty (..), + APartyI (..), MsgHash, MsgMeta (..), ConnectionStats (..), @@ -119,7 +123,6 @@ import qualified Data.Attoparsec.ByteString.Char8 as A import Data.ByteString.Base64 import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B -import Data.Composition ((.:), (.:.)) import Data.Functor (($>)) import Data.Int (Int64) import Data.Kind (Type) @@ -213,6 +216,12 @@ instance TestEquality SAParty where testEquality SClient SClient = Just Refl testEquality _ _ = Nothing +class APartyI (p :: AParty) where sAParty :: SAParty p + +instance APartyI Agent where sAParty = SAgent + +instance APartyI Client where sAParty = SClient + data ACmd = forall p. ACmd (SAParty p) (ACommand p) deriving instance Show ACmd @@ -255,6 +264,73 @@ deriving instance Eq (ACommand p) deriving instance Show (ACommand p) +data ACmdTag = forall p. APartyI p => ACmdTag (SAParty p) (ACommandTag p) + +data ACommandTag (p :: AParty) where + NEW_ :: ACommandTag Client + INV_ :: ACommandTag Agent + JOIN_ :: ACommandTag Client + CONF_ :: ACommandTag Agent + LET_ :: ACommandTag Client + REQ_ :: ACommandTag Agent + ACPT_ :: ACommandTag Client + RJCT_ :: ACommandTag Client + INFO_ :: ACommandTag Agent + CON_ :: ACommandTag Agent + SUB_ :: ACommandTag Client + END_ :: ACommandTag Agent + CONNECT_ :: ACommandTag Agent + DISCONNECT_ :: ACommandTag Agent + DOWN_ :: ACommandTag Agent + UP_ :: ACommandTag Agent + SEND_ :: ACommandTag Client + MID_ :: ACommandTag Agent + SENT_ :: ACommandTag Agent + MERR_ :: ACommandTag Agent + MSG_ :: ACommandTag Agent + ACK_ :: ACommandTag Client + OFF_ :: ACommandTag Client + DEL_ :: ACommandTag Client + CHK_ :: ACommandTag Client + STAT_ :: ACommandTag Agent + OK_ :: ACommandTag Agent + ERR_ :: ACommandTag Agent + SUSPENDED_ :: ACommandTag Agent + +deriving instance Show (ACommandTag p) + +aCommandTag :: ACommand p -> ACommandTag p +aCommandTag = \case + NEW {} -> NEW_ + INV _ -> INV_ + JOIN {} -> JOIN_ + CONF {} -> CONF_ + LET {} -> LET_ + REQ {} -> REQ_ + ACPT {} -> ACPT_ + RJCT _ -> RJCT_ + INFO _ -> INFO_ + CON -> CON_ + SUB -> SUB_ + END -> END_ + CONNECT {} -> CONNECT_ + DISCONNECT {} -> DISCONNECT_ + DOWN {} -> DOWN_ + UP {} -> UP_ + SEND {} -> SEND_ + MID _ -> MID_ + SENT _ -> SENT_ + MERR {} -> MERR_ + MSG {} -> MSG_ + ACK _ -> ACK_ + OFF -> OFF_ + DEL -> DEL_ + CHK -> CHK_ + STAT _ -> STAT_ + OK -> OK_ + ERR _ -> ERR_ + SUSPENDED -> SUSPENDED_ + data ConnectionStats = ConnectionStats { rcvServers :: [SMPServer], sndServers :: [SMPServer] @@ -930,58 +1006,121 @@ networkCommandP = commandP A.takeByteString dbCommandP :: Parser ACmd dbCommandP = commandP $ A.take =<< (A.decimal <* "\n") +instance Encoding ACmdTag where + smpEncode (ACmdTag _ cmd) = smpEncode cmd + smpP = + A.takeTill (== ' ') >>= \case + "NEW" -> pure $ ACmdTag SClient NEW_ + "INV" -> pure $ ACmdTag SAgent INV_ + "JOIN" -> pure $ ACmdTag SClient JOIN_ + "CONF" -> pure $ ACmdTag SAgent CONF_ + "LET" -> pure $ ACmdTag SClient LET_ + "REQ" -> pure $ ACmdTag SAgent REQ_ + "ACPT" -> pure $ ACmdTag SClient ACPT_ + "RJCT" -> pure $ ACmdTag SClient RJCT_ + "INFO" -> pure $ ACmdTag SAgent INFO_ + "CON" -> pure $ ACmdTag SAgent CON_ + "SUB" -> pure $ ACmdTag SClient SUB_ + "END" -> pure $ ACmdTag SAgent END_ + "CONNECT" -> pure $ ACmdTag SAgent CONNECT_ + "DISCONNECT" -> pure $ ACmdTag SAgent DISCONNECT_ + "DOWN" -> pure $ ACmdTag SAgent DOWN_ + "UP" -> pure $ ACmdTag SAgent UP_ + "SEND" -> pure $ ACmdTag SClient SEND_ + "MID" -> pure $ ACmdTag SAgent MID_ + "SENT" -> pure $ ACmdTag SAgent SENT_ + "MERR" -> pure $ ACmdTag SAgent MERR_ + "MSG" -> pure $ ACmdTag SAgent MSG_ + "ACK" -> pure $ ACmdTag SClient ACK_ + "OFF" -> pure $ ACmdTag SClient OFF_ + "DEL" -> pure $ ACmdTag SClient DEL_ + "CHK" -> pure $ ACmdTag SClient CHK_ + "STAT" -> pure $ ACmdTag SAgent STAT_ + "OK" -> pure $ ACmdTag SAgent OK_ + "ERR" -> pure $ ACmdTag SAgent ERR_ + "SUSPENDED" -> pure $ ACmdTag SAgent SUSPENDED_ + _ -> fail "bad ACmdTag" + +instance APartyI p => Encoding (ACommandTag p) where + smpEncode = \case + NEW_ -> "NEW" + INV_ -> "INV" + JOIN_ -> "JOIN" + CONF_ -> "CONF" + LET_ -> "LET" + REQ_ -> "REQ" + ACPT_ -> "ACPT" + RJCT_ -> "RJCT" + INFO_ -> "INFO" + CON_ -> "CON" + SUB_ -> "SUB" + END_ -> "END" + CONNECT_ -> "CONNECT" + DISCONNECT_ -> "DISCONNECT" + DOWN_ -> "DOWN" + UP_ -> "UP" + SEND_ -> "SEND" + MID_ -> "MID" + SENT_ -> "SENT" + MERR_ -> "MERR" + MSG_ -> "MSG" + ACK_ -> "ACK" + OFF_ -> "OFF" + DEL_ -> "DEL" + CHK_ -> "CHK" + STAT_ -> "STAT" + OK_ -> "OK" + ERR_ -> "ERR" + SUSPENDED_ -> "SUSPENDED" + smpP = (\(ACmdTag _ t) -> checkParty t) <$?> smpP + +checkParty :: forall t p p'. (APartyI p, APartyI p') => t p' -> Either String (t p) +checkParty x = case testEquality (sAParty @p) (sAParty @p') of + Just Refl -> Right x + Nothing -> Left "bad party" + -- | SMP agent command and response parser commandP :: Parser ByteString -> Parser ACmd -commandP parseByteString = - "NEW " *> newCmd - <|> "INV " *> invResp - <|> "JOIN " *> joinCmd - <|> "CONF " *> confMsg - <|> "LET " *> letCmd - <|> "REQ " *> reqMsg - <|> "ACPT " *> acptCmd - <|> "RJCT " *> rjctCmd - <|> "INFO " *> infoCmd - <|> "SUB" $> ACmd SClient SUB - <|> "END" $> ACmd SAgent END - <|> "CONNECT " *> connectResp - <|> "DISCONNECT " *> disconnectResp - <|> "DOWN " *> downResp - <|> "UP " *> upResp - <|> "SEND " *> sendCmd - <|> "MID " *> msgIdResp - <|> "SENT " *> sentResp - <|> "MERR " *> msgErrResp - <|> "MSG " *> message - <|> "ACK " *> ackCmd - <|> "OFF" $> ACmd SClient OFF - <|> "DEL" $> ACmd SClient DEL - <|> "CHK" $> ACmd SClient CHK - <|> "STAT " *> statResp - <|> "ERR " *> agentError - <|> "CON" $> ACmd SAgent CON - <|> "OK" $> ACmd SAgent OK +commandP binaryP = + smpP + >>= \case + ACmdTag SClient cmd -> + ACmd SClient <$> case cmd of + NEW_ -> s (NEW <$> strP_ <*> strP) + JOIN_ -> s (JOIN <$> strP_ <*> strP_ <*> binaryP) + LET_ -> s (LET <$> A.takeTill (== ' ') <* A.space <*> binaryP) + ACPT_ -> s (ACPT <$> A.takeTill (== ' ') <* A.space <*> binaryP) + RJCT_ -> s (RJCT <$> A.takeByteString) + SUB_ -> pure SUB + SEND_ -> s (SEND <$> smpP <* A.space <*> binaryP) + ACK_ -> s (ACK <$> A.decimal) + OFF_ -> pure OFF + DEL_ -> pure DEL + CHK_ -> pure CHK + ACmdTag SAgent cmd -> + ACmd SAgent <$> case cmd of + INV_ -> s (INV <$> strP) + CONF_ -> s (CONF <$> A.takeTill (== ' ') <* A.space <*> strListP <* A.space <*> binaryP) + REQ_ -> s (REQ <$> A.takeTill (== ' ') <* A.space <*> strP_ <*> binaryP) + INFO_ -> s (INFO <$> binaryP) + CON_ -> pure CON + END_ -> pure END + CONNECT_ -> s (CONNECT <$> strP_ <*> strP) + DISCONNECT_ -> s (DISCONNECT <$> strP_ <*> strP) + DOWN_ -> s (DOWN <$> strP_ <*> connections) + UP_ -> s (UP <$> strP_ <*> connections) + MID_ -> s (MID <$> A.decimal) + SENT_ -> s (SENT <$> A.decimal) + MERR_ -> s (MERR <$> A.decimal <* A.space <*> strP) + MSG_ -> s (MSG <$> msgMetaP <* A.space <*> smpP <* A.space <*> binaryP) + STAT_ -> s (STAT <$> strP) + OK_ -> pure OK + ERR_ -> s (ERR <$> strP) + SUSPENDED_ -> pure SUSPENDED where - newCmd = ACmd SClient .: NEW <$> strP_ <*> strP - invResp = ACmd SAgent . INV <$> strP - joinCmd = ACmd SClient .:. JOIN <$> strP_ <*> strP_ <*> parseByteString - confMsg = ACmd SAgent .:. CONF <$> A.takeTill (== ' ') <* A.space <*> strListP <* A.space <*> parseByteString - letCmd = ACmd SClient .: LET <$> A.takeTill (== ' ') <* A.space <*> parseByteString - reqMsg = ACmd SAgent .:. REQ <$> A.takeTill (== ' ') <* A.space <*> strP_ <*> parseByteString - acptCmd = ACmd SClient .: ACPT <$> A.takeTill (== ' ') <* A.space <*> parseByteString - rjctCmd = ACmd SClient . RJCT <$> A.takeByteString - infoCmd = ACmd SAgent . INFO <$> parseByteString - connectResp = ACmd SAgent .: CONNECT <$> strP_ <*> strP - disconnectResp = ACmd SAgent .: DISCONNECT <$> strP_ <*> strP - downResp = ACmd SAgent .: DOWN <$> strP_ <*> connections - upResp = ACmd SAgent .: UP <$> strP_ <*> connections - sendCmd = ACmd SClient .: SEND <$> smpP <* A.space <*> parseByteString - msgIdResp = ACmd SAgent . MID <$> A.decimal - sentResp = ACmd SAgent . SENT <$> A.decimal - msgErrResp = ACmd SAgent .: MERR <$> A.decimal <* A.space <*> strP - message = ACmd SAgent .:. MSG <$> msgMetaP <* A.space <*> smpP <* A.space <*> parseByteString - ackCmd = ACmd SClient . ACK <$> A.decimal - statResp = ACmd SAgent . STAT <$> strP + s :: Parser a -> Parser a + s p = A.space *> p + connections :: Parser [ConnId] connections = strP `A.sepBy'` A.char ',' msgMetaP = do integrity <- strP @@ -990,7 +1129,6 @@ commandP parseByteString = sndMsgId <- " S=" *> A.decimal pure MsgMeta {integrity, recipient, broker, sndMsgId} partyMeta idParser = (,) <$> idParser <* A.char ',' <*> tsISO8601P - agentError = ACmd SAgent . ERR <$> strP parseCommand :: ByteString -> Either AgentErrorType ACmd parseCommand = parse (commandP A.takeByteString) $ CMD SYNTAX diff --git a/src/Simplex/Messaging/Agent/Store/SQLite.hs b/src/Simplex/Messaging/Agent/Store/SQLite.hs index 4291b53d5..50e752c57 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite.hs @@ -695,19 +695,18 @@ updateRatchet db connId rc skipped = do forM_ (M.assocs mks) $ \(msgN, mk) -> DB.execute db "INSERT INTO skipped_messages (conn_id, header_key, msg_n, msg_key) VALUES (?, ?, ?, ?)" (connId, hk, msgN, mk) -createCommand :: DB.Connection -> ConnId -> Maybe SMPServer -> ACommand 'Client -> IO AsyncCmdId -createCommand db connId (Just (SMPServer host port _)) command = do +createCommand :: DB.Connection -> ACorrId -> ConnId -> Maybe SMPServer -> ACommand 'Client -> IO AsyncCmdId +createCommand db corrId connId srv cmd = do DB.execute db - "INSERT INTO commands (host, port, conn_id, command) VALUES (?, ?, ?, ?)" - (host, port, connId, serializeCommand command) - insertedRowId db -createCommand db connId Nothing command = do - DB.execute - db - "INSERT INTO commands (conn_id, command) VALUES (?, ?)" - (connId, command) + "INSERT INTO commands (host, port, corr_id, conn_id, command_tag, command) VALUES (?,?,?,?,?,?)" + (host_, port_, corrId, connId, aCommandTag cmd, cmd) insertedRowId db + where + (host_, port_) = + case srv of + Just (SMPServer host port _) -> (Just host, Just port) + _ -> (Nothing, Nothing) insertedRowId :: DB.Connection -> IO Int64 insertedRowId db = fromOnly . head <$> DB.query_ db "SELECT last_insert_rowid()" @@ -728,16 +727,13 @@ getPendingCommands db connId = do where srvCmdId (host, port, keyHash, cmdId) = (SMPServer <$> host <*> port <*> keyHash, cmdId) -getPendingCommand :: DB.Connection -> AsyncCmdId -> IO (Either StoreError (ConnId, ACmd)) +getPendingCommand :: DB.Connection -> AsyncCmdId -> IO (Either StoreError (ACorrId, ConnId, ACmd)) getPendingCommand db msgId = do - firstRow pendingCmd SECmdNotFound $ + firstRow id SECmdNotFound $ DB.query db - "SELECT conn_id, command FROM commands WHERE command_id = ?" + "SELECT corr_id, conn_id, command FROM commands WHERE command_id = ?" (Only msgId) - where - pendingCmd :: (ConnId, ACmd) -> (ConnId, ACmd) - pendingCmd (connId, commandStr) = (connId, commandStr) deleteCommand :: DB.Connection -> AsyncCmdId -> IO () deleteCommand db cmdId = @@ -1109,6 +1105,10 @@ instance ToField (ACommand p) where toField = toField . serializeCommand instance FromField ACmd where fromField = blobFieldParser dbCommandP +instance APartyI p => ToField (ACommandTag p) where toField = toField . smpEncode + +instance FromField ACmdTag where fromField = blobFieldParser smpP + listToEither :: e -> [a] -> Either e a listToEither _ (x : _) = Right x listToEither e _ = Left e diff --git a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20220905_commands.hs b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20220905_commands.hs index 0553969a6..57939a58a 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20220905_commands.hs +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/M20220905_commands.hs @@ -13,8 +13,10 @@ CREATE TABLE commands ( conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE, host TEXT, port TEXT, - command TEXT NOT NULL, - command_version INTEGER NOT NULL DEFAULT 1, + corr_id BLOB NOT NULL, + command_tag BLOB NOT NULL, + command BLOB NOT NULL, + agent_version INTEGER NOT NULL DEFAULT 1, FOREIGN KEY (host, port) REFERENCES servers ON DELETE RESTRICT ON UPDATE CASCADE ); 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 2f363b471..5787978bb 100644 --- a/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql +++ b/src/Simplex/Messaging/Agent/Store/SQLite/Migrations/agent_schema.sql @@ -199,8 +199,10 @@ CREATE TABLE commands( conn_id BLOB NOT NULL REFERENCES connections ON DELETE CASCADE, host TEXT, port TEXT, - command TEXT NOT NULL, - command_version INTEGER NOT NULL DEFAULT 1, + corr_id BLOB NOT NULL, + command_tag BLOB NOT NULL, + command BLOB NOT NULL, + agent_version INTEGER NOT NULL DEFAULT 1, 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 bdda18f50..cf42e9dc8 100644 --- a/tests/AgentTests/FunctionalAPITests.hs +++ b/tests/AgentTests/FunctionalAPITests.hs @@ -570,11 +570,15 @@ testAsyncCommands = do alice <- getSMPAgentClient agentCfg initAgentServers bob <- getSMPAgentClient agentCfg {dbFile = testDB2} initAgentServers Right () <- runExceptT $ do - bobId <- createConnectionAsync alice True SCMInvitation - ("", _, INV (ACR _ qInfo)) <- get alice - aliceId <- joinConnectionAsync bob True qInfo "bob's connInfo" + bobId <- createConnectionAsync alice "1" True SCMInvitation + ("1", bobId', INV (ACR _ qInfo)) <- get alice + liftIO $ bobId' `shouldBe` bobId + aliceId <- joinConnectionAsync bob "2" True qInfo "bob's connInfo" + ("2", aliceId', OK) <- get bob + liftIO $ aliceId' `shouldBe` aliceId ("", _, CONF confId _ "bob's connInfo") <- get alice - allowConnectionAsync alice bobId confId "alice's connInfo" + allowConnectionAsync alice "3" bobId confId "alice's connInfo" + ("3", _, OK) <- get alice get alice ##> ("", bobId, CON) get bob ##> ("", aliceId, INFO "alice's connInfo") get bob ##> ("", aliceId, CON) @@ -584,17 +588,22 @@ testAsyncCommands = do 2 <- msgId <$> sendMessage alice bobId SMP.noMsgFlags "how are you?" get alice ##> ("", bobId, SENT $ baseId + 2) get bob =##> \case ("", c, Msg "hello") -> c == aliceId; _ -> False - ackMessageAsync bob aliceId $ baseId + 1 + ackMessageAsync bob "4" aliceId $ baseId + 1 + ("4", _, OK) <- get bob get bob =##> \case ("", c, Msg "how are you?") -> c == aliceId; _ -> False - ackMessageAsync bob aliceId $ baseId + 2 + ackMessageAsync bob "5" aliceId $ baseId + 2 + ("5", _, OK) <- get bob 3 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "hello too" get bob ##> ("", aliceId, SENT $ baseId + 3) 4 <- msgId <$> sendMessage bob aliceId SMP.noMsgFlags "message 1" get bob ##> ("", aliceId, SENT $ baseId + 4) get alice =##> \case ("", c, Msg "hello too") -> c == bobId; _ -> False - ackMessageAsync alice bobId $ baseId + 3 + ackMessageAsync alice "6" bobId $ baseId + 3 + ("6", _, OK) <- get alice get alice =##> \case ("", c, Msg "message 1") -> c == bobId; _ -> False - ackMessageAsync alice bobId $ baseId + 4 + ackMessageAsync alice "7" bobId $ baseId + 4 + ("7", _, OK) <- get alice + pure () pure () where baseId = 3 @@ -603,14 +612,14 @@ testAsyncCommands = do testAsyncCommandsRestore :: ATransport -> IO () testAsyncCommandsRestore t = do alice <- getSMPAgentClient agentCfg initAgentServers - Right bobId <- runExceptT $ createConnectionAsync alice True SCMInvitation + Right bobId <- runExceptT $ createConnectionAsync alice "1" True SCMInvitation liftIO $ noMessages alice "alice doesn't receive INV because server is down" disconnectAgentClient alice alice' <- liftIO $ getSMPAgentClient agentCfg initAgentServers withSmpServerStoreLogOn t testPort $ \_ -> do Right () <- runExceptT $ do subscribeConnection alice' bobId - ("", _, INV _) <- get alice' + ("1", _, INV _) <- get alice' pure () pure ()