Changeset 428 in code

Timestamp:

Nov 24, 2020, 1:13:24 PM (5 years ago)

Author:

contact

Message:

Implement delivery receipts via PING messages

This patch implements basic message delivery receipts via PING and PONG.

When a PRIVMSG or NOTICE message is sent, a PING message with a token is
also sent. The history cursor isn't immediately advanced, instead the
bouncer will wait for a PONG message before doing so.

Self-messages trigger a PING for simplicity's sake. We can't immediately
advance the history cursor in this case, because a prior message might
still have an outstanding PING.

Future work may include optimizations such as removing the need to send
a PING after a self-message, or groupping multiple PING messages
together.

Closes: ​https://todo.sr.ht/~emersion/soju/11

Location:

trunk

Files:

• downstream.go (modified) (4 diffs)
• upstream.go (modified) (5 diffs)

trunk/downstream.go

@@ -288 +288 @@
 
         dc.conn.SendMessage(msg)
+}
+
+// sendMessageWithID sends an outgoing message with the specified internal ID.
+func (dc *downstreamConn) sendMessageWithID(msg *irc.Message, id string) {
+        dc.SendMessage(msg)
+
+        if id == "" || !dc.messageSupportsHistory(msg) {
+                return
+        }
+
+        dc.sendPing(id)
+}
+
+// advanceMessageWithID advances history to the specified message ID without
+// sending a message. This is useful e.g. for self-messages when echo-message
+// isn't enabled.
+func (dc *downstreamConn) advanceMessageWithID(msg *irc.Message, id string) {
+        if id == "" || !dc.messageSupportsHistory(msg) {
+                return
+        }
+
+        dc.sendPing(id)
+}
+
+// ackMsgID acknowledges that a message has been received.
+func (dc *downstreamConn) ackMsgID(id string) {
+        netName, entity, _, _, err := parseMsgID(id)
+        if err != nil {
+                dc.logger.Printf("failed to ACK message ID %q: %v", id, err)
+                return
+        }
+
+        network := dc.user.getNetwork(netName)
+        if network == nil {
+                return
+        }
+
+        history, ok := network.history[entity]
+        if !ok {
+                return
+        }
+
+        history.clients[dc.clientName] = id
+}
+
+func (dc *downstreamConn) sendPing(msgID string) {
+        token := "soju-msgid-" + base64.RawURLEncoding.EncodeToString([]byte(msgID))
+        dc.SendMessage(&irc.Message{
+                Command: "PING",
+                Params:  []string{token},
+        })
+}
+
+func (dc *downstreamConn) handlePong(token string) {
+        if !strings.HasPrefix(token, "soju-msgid-") {
+                dc.logger.Printf("received unrecognized PONG token %q", token)
+                return
+        }
+        token = strings.TrimPrefix(token, "soju-msgid-")
+        b, err := base64.RawURLEncoding.DecodeString(token)
+        if err != nil {
+                dc.logger.Printf("received malformed PONG token: %v", err)
+                return
+        }
+        msgID := string(b)
+
+        dc.ackMsgID(msgID)
 }
 
@@ -876 +943 @@
 }
 
+// messageSupportsHistory checks whether the provided message can be sent as
+// part of an history batch.
+func (dc *downstreamConn) messageSupportsHistory(msg *irc.Message) bool {
+        // Don't replay all messages, because that would mess up client
+        // state. For instance we just sent the list of users, sending
+        // PART messages for one of these users would be incorrect.
+        // TODO: add support for draft/event-playback
+        switch msg.Command {
+        case "PRIVMSG", "NOTICE":
+                return true
+        }
+        return false
+}
+
 func (dc *downstreamConn) sendNetworkHistory(net *network) {
         if dc.caps["draft/chathistory"] || dc.user.msgStore == nil {
@@ -907 +988 @@
 
                 for _, msg := range history {
-                        // Don't replay all messages, because that would mess up client
-                        // state. For instance we just sent the list of users, sending
-                        // PART messages for one of these users would be incorrect.
-                        ignore := true
-                        switch msg.Command {
-                        case "PRIVMSG", "NOTICE":
-                                ignore = false
-                        }
-                        if ignore {
+                        if !dc.messageSupportsHistory(msg) {
                                 continue
                         }
@@ -984 +1057 @@
                 })
                 return nil
+        case "PONG":
+                if len(msg.Params) == 0 {
+                        return newNeedMoreParamsError(msg.Command)
+                }
+                token := msg.Params[len(msg.Params)-1]
+                dc.handlePong(token)
         case "USER":
                 return ircError{&irc.Message{

trunk/upstream.go

@@ -1608 +1608 @@
 }
 
-func (uc *upstreamConn) appendLog(entity string, msg *irc.Message) {
+// appendLog appends a message to the log file.
+//
+// The internal message ID is returned. If the message isn't recorded in the
+// log file, an empty string is returned.
+func (uc *upstreamConn) appendLog(entity string, msg *irc.Message) (msgID string) {
         if uc.user.msgStore == nil {
-                return
+                return ""
         }
 
@@ -1623 +1627 @@
                 if err != nil {
                         uc.logger.Printf("failed to log message: failed to get last message ID: %v", err)
-                        return
+                        return ""
                 }
 
@@ -1647 +1651 @@
         if err != nil {
                 uc.logger.Printf("failed to log message: %v", err)
-                return
-        }
-
-        if !detached && msgID != "" {
-                uc.forEachDownstream(func(dc *downstreamConn) {
-                        history.clients[dc.clientName] = msgID
-                })
-        }
+                return ""
+        }
+
+        return msgID
 }
 
@@ -1663 +1663 @@
 // forwarded to all connections except origin.
 func (uc *upstreamConn) produce(target string, msg *irc.Message, origin *downstreamConn) {
+        var msgID string
         if target != "" {
-                uc.appendLog(target, msg)
+                msgID = uc.appendLog(target, msg)
         }
 
@@ -1674 +1675 @@
         uc.forEachDownstream(func(dc *downstreamConn) {
                 if dc != origin || dc.caps["echo-message"] {
-                        dc.SendMessage(dc.marshalMessage(msg, uc.network))
+                        dc.sendMessageWithID(dc.marshalMessage(msg, uc.network), msgID)
+                } else {
+                        dc.advanceMessageWithID(msg, msgID)
                 }
         })