message management (#32)

* message management rfc

* update message management rfc

* message management ideas (WIP)

* message management updated

* messages RFC

* update agent MSG constructor to include recipient/broker/sender message IDs and timestamps

* remove agent command ACK - agent automatically acknowledges server messages

* correct messages doc

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.

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 <HELLO>"
@@ -251,7 +240,16 @@ processSMPTransmission c@AgentClient {sndQ} (srv, rId, cmd) = do
             A_MSG body -> do
               logServer "<--" c srv rId "MSG <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

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
             }

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"

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

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"