diff --git a/docs/logging.md b/rfcs/2021-01-20-logging.md similarity index 100% rename from docs/logging.md rename to rfcs/2021-01-20-logging.md diff --git a/rfcs/2021-01-26-messages.md b/rfcs/2021-01-26-messages.md new file mode 100644 index 000000000..71db02408 --- /dev/null +++ b/rfcs/2021-01-26-messages.md @@ -0,0 +1,33 @@ +# SMP Agent: message management + +The proposal is to change the way SMP agent manages the messages from the SMP servers. + +Currently, it acts just as a proxy, when it does not receive any further messages from the server until the client ACKs them. The downside with this approach is that the agent can't work autonomously from the client, and the client always has to be active to retrieve the messages from the server. So while the agent offers a convenient, higher abstraction level when it comes to establishing connections, it does not offer any advantage over message management than using SMP server directly. There is a benefit to have an agent as a local, mailbox-like layer that downloads all messages for all connections and makes them available for further processing on the application layer. + +## Phase 1. Managed agent MVP + +These should be sufficient for terminal chat prototype + +- Agent only maintains active subscriptions to the connections that the clients explicitly requested, only during client sessions, using `SUB` command. +- Agent would automatically send `ACK` commands to SMP servers and store them in the database. +- When the client subscribes to the connection (`SUB` - subscribe and deliver all messages received from the server) the response to `SUB` command is `OK`. +- Receiving agent should maintain its own numbering in addition to the sending agent numbering, in case the sending agent message number decreases. +- When there is an active subscription, Agent forwards messages to the connected clients in the order of their arrival, without any confirmations from the clients. It is client's responsibility to track that they didn't miss any messages and request re-delivery if needed. + +## Phase 2. Subscription management and message access + +- When the client subscribes to the connection (`SUB` - subscribe and deliver all missed messages) the response to `SUB` command is `NOMSG` or `MID n` where n is the ID of the last received message (1-based). +- `SUB SKIP` - subscribe but only deliver the new messages +- `SUB OFF` - turn off subscription +- Messages received by the agent during the period when there was no active client subscription are not sent to the client automatically - client has to request them one by one using `MGET n` command (it can also be used to request re-delivery of old messages or for any other direct access scenario). +- Clients can delete received messages using `MDEL n` command, where `n` is a sequential agent-assigned message ID - same that was used in `MSG` delivery. +- Clients can confirm the receipt of the messages to the sender using `MACK n` command +- Client can also request the last message ID with `MLAST` command - with the same responses. + +## Phase 3. Autonomous agent + +- Agent maintains active subscriptions to all connections that the clients created, whether there are active sessions with the clients or not. +- `SUB ON` command would be used to receive messages from the agent, and while it is active the agent would maintain an active TCP connection(s) with the server(s) to retrieve the messages as soon as they arrive. +- In the absence of the client connection the agent would periodically check and retrieve messages from the servers, possibly up to a quota, from all active connections. If the servers do not reply, the agent could have some retry logic with exponential back-off eventually marking connections (and/or servers) as unresponsive. + +This design allows to have a single agent running in the background with multiple client applications connecting to it. It provides better message delivery guarantees, assuming transient nature of SMP server storage. \ No newline at end of file diff --git a/src/Simplex/Messaging/Agent.hs b/src/Simplex/Messaging/Agent.hs index 82d229dd9..3dc902285 100644 --- a/src/Simplex/Messaging/Agent.hs +++ b/src/Simplex/Messaging/Agent.hs @@ -19,6 +19,7 @@ import Data.ByteString.Char8 (ByteString) import qualified Data.ByteString.Char8 as B import qualified Data.Text as T import Data.Text.Encoding +import Data.Time.Clock import Simplex.Messaging.Agent.Client import Simplex.Messaging.Agent.Env.SQLite import Simplex.Messaging.Agent.Store @@ -111,7 +112,6 @@ processCommand c@AgentClient {sndQ} (corrId, connAlias, cmd) = JOIN smpQueueInfo replyMode -> joinConnection smpQueueInfo replyMode SUB -> subscribeConnection SEND msgBody -> sendMessage msgBody - ACK aMsgId -> ackMessage aMsgId OFF -> suspendConnection DEL -> deleteConnection where @@ -159,17 +159,6 @@ processCommand c@AgentClient {sndQ} (corrId, connAlias, cmd) = -- TODO respond $ SENT aMsgId respond OK - ackMessage :: AgentMsgId -> m () - ackMessage _aMsgId = - withStore (`getConn` connAlias) >>= \case - SomeConn _ (DuplexConnection _ rq _) -> ackMsg rq - SomeConn _ (ReceiveConnection _ rq) -> ackMsg rq - -- TODO possibly there should be a separate error type trying to send the message to the connection without ReceiveQueue - -- NOT_READY ? - _ -> throwError PROHIBITED - where - ackMsg rq = sendAck c rq >> respond OK - suspendConnection :: m () suspendConnection = withStore (`getConn` connAlias) >>= \case @@ -213,7 +202,7 @@ processSMPTransmission :: forall m. AgentMonad m => AgentClient -> SMPServerTran processSMPTransmission c@AgentClient {sndQ} (srv, rId, cmd) = do (connAlias, rq@ReceiveQueue {decryptKey, status}) <- withStore $ \st -> getReceiveQueue st srv rId case cmd of - SMP.MSG _ srvTs msgBody -> do + SMP.MSG srvMsgId srvTs msgBody -> do -- TODO deduplicate with previously received agentMsg <- liftEither . parseSMPMessage =<< decryptMessage decryptKey msgBody case agentMsg of @@ -233,7 +222,7 @@ processSMPTransmission c@AgentClient {sndQ} (srv, rId, cmd) = do s -> -- TODO maybe send notification to the user liftIO . putStrLn $ "unexpected SMP confirmation, queue status " <> show s - SMPMessage {agentMessage, agentMsgId, agentTimestamp} -> + SMPMessage {agentMessage, senderMsgId, senderTimestamp} -> case agentMessage of HELLO _verifyKey _ -> do logServer "<--" c srv rId "MSG " @@ -251,7 +240,16 @@ processSMPTransmission c@AgentClient {sndQ} (srv, rId, cmd) = do A_MSG body -> do logServer "<--" c srv rId "MSG " -- TODO check message status - notify connAlias $ MSG agentMsgId agentTimestamp srvTs MsgOk body + recipientTs <- liftIO getCurrentTime + notify connAlias $ + MSG + { m_status = MsgOk, + m_recipient = (0, recipientTs), + m_sender = (senderMsgId, senderTimestamp), + m_broker = (srvMsgId, srvTs), + m_body = body + } + sendAck c rq return () SMP.END -> do removeSubscription c connAlias diff --git a/src/Simplex/Messaging/Agent/Client.hs b/src/Simplex/Messaging/Agent/Client.hs index 8eadb7a08..2eab90cce 100644 --- a/src/Simplex/Messaging/Agent/Client.hs +++ b/src/Simplex/Messaging/Agent/Client.hs @@ -270,12 +270,12 @@ sendAgentMessage c SendQueue {server, sndId, sndPrivateKey, encryptKey} agentMsg mkAgentMessage :: MonadUnliftIO m => PrivateKey -> AMessage -> m ByteString mkAgentMessage _encKey agentMessage = do - agentTimestamp <- liftIO getCurrentTime + senderTimestamp <- liftIO getCurrentTime let msg = serializeSMPMessage SMPMessage - { agentMsgId = 0, - agentTimestamp, + { senderMsgId = 0, + senderTimestamp, previousMsgHash = "1234", -- TODO hash of the previous message agentMessage } diff --git a/src/Simplex/Messaging/Agent/Transmission.hs b/src/Simplex/Messaging/Agent/Transmission.hs index 64f4fcc04..ea117ea76 100644 --- a/src/Simplex/Messaging/Agent/Transmission.hs +++ b/src/Simplex/Messaging/Agent/Transmission.hs @@ -79,8 +79,15 @@ data ACommand (p :: AParty) where -- QST :: QueueDirection -> ACommand Client -- STAT :: QueueDirection -> Maybe QueueStatus -> Maybe SubMode -> ACommand Agent SEND :: MsgBody -> ACommand Client - MSG :: AgentMsgId -> UTCTime -> UTCTime -> MsgStatus -> MsgBody -> ACommand Agent - ACK :: AgentMsgId -> ACommand Client + MSG :: + { m_recipient :: (AgentMsgId, UTCTime), + m_broker :: (ST.MsgId, UTCTime), + m_sender :: (AgentMsgId, UTCTime), + m_status :: MsgStatus, + m_body :: MsgBody + } -> + ACommand Agent + -- ACK :: AgentMsgId -> ACommand Client -- RCVD :: AgentMsgId -> ACommand Agent OFF :: ACommand Client DEL :: ACommand Client @@ -96,8 +103,8 @@ type Message = ByteString data SMPMessage = SMPConfirmation PublicKey | SMPMessage - { agentMsgId :: Integer, - agentTimestamp :: UTCTime, + { senderMsgId :: Integer, + senderTimestamp :: UTCTime, previousMsgHash :: ByteString, agentMessage :: AMessage } @@ -134,8 +141,8 @@ tsISO8601P = maybe (fail "timestamp") pure . parseISO8601 . B.unpack =<< A.takeT serializeSMPMessage :: SMPMessage -> ByteString serializeSMPMessage = \case SMPConfirmation sKey -> smpMessage ("KEY " <> encode sKey) "" "" - SMPMessage {agentMsgId, agentTimestamp, previousMsgHash, agentMessage} -> - let header = messageHeader agentMsgId agentTimestamp previousMsgHash + SMPMessage {senderMsgId, senderTimestamp, previousMsgHash, agentMessage} -> + let header = messageHeader senderMsgId senderTimestamp previousMsgHash body = serializeAgentMessage agentMessage in smpMessage "" header body where @@ -286,7 +293,6 @@ parseCommandP = <|> "END" $> ACmd SAgent END <|> "SEND " *> sendCmd <|> "MSG " *> message - <|> "ACK " *> acknowledge <|> "OFF" $> ACmd SClient OFF <|> "DEL" $> ACmd SClient DEL <|> "ERR " *> agentError @@ -297,10 +303,13 @@ parseCommandP = invResp = ACmd SAgent . INV <$> smpQueueInfoP joinCmd = ACmd SClient <$> (JOIN <$> smpQueueInfoP <*> replyMode) sendCmd = ACmd SClient <$> (SEND <$> A.takeByteString) - message = - let sp = A.space; msgId = A.decimal <* sp; ts = tsISO8601P <* sp; body = A.takeByteString - in ACmd SAgent <$> (MSG <$> msgId <*> ts <*> ts <*> status <* sp <*> body) - acknowledge = ACmd SClient <$> (ACK <$> A.decimal) + message = do + m_status <- status <* A.space + m_recipient <- "R=" *> partyMeta A.decimal + m_broker <- "B=" *> partyMeta base64P + m_sender <- "S=" *> partyMeta A.decimal + m_body <- A.takeByteString + return $ ACmd SAgent MSG {m_recipient, m_broker, m_sender, m_status, m_body} -- TODO other error types agentError = ACmd SAgent . ERR <$> ("SMP " *> smpErrorType) smpErrorType = "AUTH" $> SMP ST.AUTH @@ -308,6 +317,7 @@ parseCommandP = " NO_REPLY" $> ReplyOff <|> A.space *> (ReplyVia <$> smpServerP) <|> pure ReplyOn + partyMeta idParser = (,) <$> idParser <* "," <*> tsISO8601P <* A.space status = "OK" $> MsgOk <|> "ERR " *> (MsgError <$> msgErrorType) msgErrorType = "ID " *> (MsgBadId <$> A.decimal) @@ -325,9 +335,15 @@ serializeCommand = \case SUB -> "SUB" END -> "END" SEND msgBody -> "SEND " <> serializeMsg msgBody - MSG aMsgId aTs ts st body -> - B.unwords ["MSG", B.pack $ show aMsgId, B.pack $ formatISO8601Millis aTs, B.pack $ formatISO8601Millis ts, msgStatus st, serializeMsg body] - ACK aMsgId -> "ACK " <> B.pack (show aMsgId) + MSG {m_recipient = (rmId, rTs), m_broker = (bmId, bTs), m_sender = (smId, sTs), m_status, m_body} -> + B.unwords + [ "MSG", + msgStatus m_status, + "R=" <> bshow rmId <> "," <> showTs rTs, + "B=" <> encode bmId <> "," <> showTs bTs, + "S=" <> bshow smId <> "," <> showTs sTs, + serializeMsg m_body + ] OFF -> "OFF" DEL -> "DEL" CON -> "CON" @@ -339,6 +355,8 @@ serializeCommand = \case ReplyOff -> " NO_REPLY" ReplyVia srv -> " " <> serializeServer srv ReplyOn -> "" + showTs :: UTCTime -> ByteString + showTs = B.pack . formatISO8601Millis msgStatus :: MsgStatus -> ByteString msgStatus = \case MsgOk -> "OK" diff --git a/src/Simplex/Messaging/Util.hs b/src/Simplex/Messaging/Util.hs index e8e21672c..f79c38949 100644 --- a/src/Simplex/Messaging/Util.hs +++ b/src/Simplex/Messaging/Util.hs @@ -9,6 +9,8 @@ import Control.Monad.IO.Unlift import UnliftIO.Async import UnliftIO.Exception (Exception) import qualified UnliftIO.Exception as E +import Data.ByteString.Char8 (ByteString) +import qualified Data.ByteString.Char8 as B newtype InternalException e = InternalException {unInternalException :: e} deriving (Eq, Show) @@ -32,3 +34,6 @@ infixl 4 <$$> (<$$>) :: (Functor f, Functor g) => (a -> b) -> f (g a) -> f (g b) (<$$>) = fmap . fmap + +bshow :: Show a => a -> ByteString +bshow = B.pack . show \ No newline at end of file diff --git a/tests/AgentTests.hs b/tests/AgentTests.hs index 0e03643f1..74049b258 100644 --- a/tests/AgentTests.hs +++ b/tests/AgentTests.hs @@ -2,6 +2,7 @@ {-# LANGUAGE DataKinds #-} {-# LANGUAGE GADTs #-} {-# LANGUAGE LambdaCase #-} +{-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE PatternSynonyms #-} @@ -73,7 +74,7 @@ h #:# err = tryGet `shouldReturn` () _ -> return () pattern Msg :: MsgBody -> Either AgentErrorType (ACommand 'Agent) -pattern Msg msg <- Right (MSG _ _ _ _ msg) +pattern Msg m_body <- Right MSG {m_body} testDuplexConnection :: Handle -> Handle -> IO () testDuplexConnection alice bob = do @@ -84,14 +85,10 @@ testDuplexConnection alice bob = do alice #: ("2", "bob", "SEND :hello") #> ("2", "bob", OK) alice #: ("3", "bob", "SEND :how are you?") #> ("3", "bob", OK) bob <#= \case ("", "alice", Msg "hello") -> True; _ -> False - bob #: ("12", "alice", "ACK 0") #> ("12", "alice", OK) bob <#= \case ("", "alice", Msg "how are you?") -> True; _ -> False - bob #: ("13", "alice", "ACK 0") #> ("13", "alice", OK) bob #: ("14", "alice", "SEND 9\nhello too") #> ("14", "alice", OK) alice <#= \case ("", "bob", Msg "hello too") -> True; _ -> False - alice #: ("4", "bob", "ACK 0") #> ("4", "bob", OK) bob #: ("15", "alice", "SEND 9\nmessage 1") #> ("15", "alice", OK) - bob #: ("16", "alice", "SEND 9\nmessage 2") #> ("16", "alice", OK) alice <#= \case ("", "bob", Msg "message 1") -> True; _ -> False alice #: ("5", "bob", "OFF") #> ("5", "bob", OK) bob #: ("17", "alice", "SEND 9\nmessage 3") #> ("17", "alice", ERR (SMP AUTH)) @@ -107,12 +104,9 @@ testSubscription alice1 alice2 bob = do bob #: ("13", "alice", "SEND 11\nhello again") #> ("13", "alice", OK) alice1 <# ("", "bob", CON) alice1 <#= \case ("", "bob", Msg "hello") -> True; _ -> False - alice1 #: ("2", "bob", "ACK 0") #> ("2", "bob", OK) alice1 <#= \case ("", "bob", Msg "hello again") -> True; _ -> False alice2 #: ("21", "bob", "SUB") #> ("21", "bob", OK) - alice2 <#= \case ("", "bob", Msg "hello again") -> True; _ -> False alice1 <# ("", "bob", END) - alice2 #: ("22", "bob", "ACK 0") #> ("22", "bob", OK) bob #: ("14", "alice", "SEND 2\nhi") #> ("14", "alice", OK) alice2 <#= \case ("", "bob", Msg "hi") -> True; _ -> False alice1 #:# "nothing else should be delivered to alice1"