Changeset 409 in code for trunk

Timestamp:

Aug 20, 2020, 6:05:01 PM (5 years ago)

Author:

contact

Message:

Nuke in-memory ring buffer

Instead, always read chat history from logs. Unify the implicit chat
history (pushing history to clients) and explicit chat history
(via the CHATHISTORY command).

Instead of keeping track of ring buffer cursors for each client, use
message IDs.

If necessary, the ring buffer could be re-introduced behind a
common MessageStore interface (could be useful when on-disk logs are
disabled).

References: ​https://todo.sr.ht/~emersion/soju/80

Location:

trunk

Files:

• downstream.go (modified) (5 diffs)
• ring.go (deleted)
• server.go (modified) (2 diffs)
• upstream.go (modified) (7 diffs)
• user.go (modified) (1 diff)

trunk/downstream.go

@@ -851 +851 @@
                         delete(net.offlineClients, dc.clientName)
                 }
+
+                // Fast-forward history to last message
+                for target, history := range net.history {
+                        if ch, ok := net.channels[target]; ok && ch.Detached {
+                                continue
+                        }
+
+                        lastID, err := lastMsgID(net, target, time.Now())
+                        if err != nil {
+                                dc.logger.Printf("failed to get last message ID: %v", err)
+                                continue
+                        }
+                        history.clients[dc.clientName] = lastID
+                }
         })
 
@@ -857 +871 @@
 
 func (dc *downstreamConn) sendNetworkHistory(net *network) {
-        if dc.caps["draft/chathistory"] {
+        if dc.caps["draft/chathistory"] || dc.srv.LogPath == "" {
                 return
         }
@@ -865 +879 @@
                 }
 
-                seq, ok := history.clients[dc.clientName]
+                lastDelivered, ok := history.clients[dc.clientName]
                 if !ok {
                         continue
                 }
-                history.clients[dc.clientName] = history.ring.Cur()
-
-                consumer := history.ring.NewConsumer(seq)
+
+                limit := 4000
+                history, err := loadHistoryLatestID(dc.network, target, lastDelivered, limit)
+                if err != nil {
+                        dc.logger.Printf("failed to send implicit history for %q: %v", target, err)
+                        continue
+                }
 
                 batchRef := "history"
@@ -882 +900 @@
                 }
 
-                for {
-                        msg := consumer.Consume()
-                        if msg == nil {
-                                break
-                        }
-
+                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
@@ -901 +914 @@
 
                         if dc.caps["batch"] {
-                                msg = msg.Copy()
                                 msg.Tags["batch"] = irc.TagValue(batchRef)
                         }
-
                         dc.SendMessage(dc.marshalMessage(msg, net))
                 }

trunk/server.go

@@ -48 +48 @@
         Hostname       string
         Logger         Logger
-        RingCap        int
         HistoryLimit   int
         LogPath        string
@@ -65 +64 @@
         return &Server{
                 Logger:       log.New(log.Writer(), "", log.LstdFlags),
-                RingCap:      4096,
                 HistoryLimit: 1000,
                 users:        make(map[string]*user),

trunk/upstream.go

@@ -709 +709 @@
                                 ch.Members[newNick] = memberships
                                 uc.appendLog(ch.Name, msg)
-                                uc.appendHistory(ch.Name, msg)
                         }
                 }
@@ -819 +818 @@
 
                                 uc.appendLog(ch.Name, msg)
-                                uc.appendHistory(ch.Name, msg)
                         }
                 }
@@ -1612 +1610 @@
 }
 
-// TODO: handle moving logs when a network name changes, when support for this is added
 func (uc *upstreamConn) appendLog(entity string, msg *irc.Message) {
         if uc.srv.LogPath == "" {
@@ -1624 +1621 @@
         }
 
-        if _, err := ml.Append(msg); err != nil {
-                uc.logger.Printf("failed to log message: %v", err)
-        }
-}
-
-// appendHistory appends a message to the history. entity can be empty.
-func (uc *upstreamConn) appendHistory(entity string, msg *irc.Message) {
         detached := false
         if ch, ok := uc.network.channels[entity]; ok {
@@ -1638 +1628 @@
         history, ok := uc.network.history[entity]
         if !ok {
+                lastID, err := lastMsgID(uc.network, entity, time.Now())
+                if err != nil {
+                        uc.logger.Printf("failed to log message: failed to get last message ID: %v", err)
+                        return
+                }
+
                 history = &networkHistory{
-                        clients: make(map[string]uint64),
-                        ring:    NewRing(uc.srv.RingCap),
+                        clients: make(map[string]string),
                 }
                 uc.network.history[entity] = history
 
                 for clientName, _ := range uc.network.offlineClients {
-                        history.clients[clientName] = 0
+                        history.clients[clientName] = lastID
                 }
 
@@ -1652 +1647 @@
                         // clients too
                         uc.forEachDownstream(func(dc *downstreamConn) {
-                                history.clients[dc.clientName] = 0
-                        })
-                }
-        }
-
-        history.ring.Produce(msg)
+                                history.clients[dc.clientName] = lastID
+                        })
+                }
+        }
+
+        msgID, err := ml.Append(msg)
+        if err != nil {
+                uc.logger.Printf("failed to log message: %v", err)
+                return
+        }
 
         if !detached {
                 uc.forEachDownstream(func(dc *downstreamConn) {
-                        history.clients[dc.clientName] = history.ring.Cur()
-                })
-        }
-}
-
-// produce appends a message to the logs, adds it to the history and forwards
-// it to connected downstream connections.
+                        history.clients[dc.clientName] = msgID
+                })
+        }
+}
+
+// produce appends a message to the logs and forwards it to connected downstream
+// connections.
 //
 // If origin is not nil and origin doesn't support echo-message, the message is
@@ -1675 +1674 @@
                 uc.appendLog(target, msg)
         }
-
-        uc.appendHistory(target, msg)
 
         // Don't forward messages if it's a detached channel

trunk/user.go

@@ -52 +52 @@
 
 type networkHistory struct {
-        clients map[string]uint64 // indexed by client name
-        ring    *Ring             // can be nil if there are no offline clients
+        clients map[string]string // indexed by client name
 }