Changeset 667 in code

Timestamp:

Nov 3, 2021, 5:21:05 PM (4 years ago)

Author:

contact

Message:

msgstore: add context to messageStore methods

This allows setting a hard timeout.

Location:

trunk

Files:

• downstream.go (modified) (2 diffs)
• msgstore.go (modified) (3 diffs)
• msgstore_fs.go (modified) (5 diffs)
• msgstore_memory.go (modified) (2 diffs)
• server.go (modified) (1 diff)

trunk/downstream.go

@@ -1329 +1329 @@
         ch := net.channels.Value(target)
 
+        ctx, cancel := context.WithTimeout(context.TODO(), messageStoreTimeout)
+        defer cancel()
+
         limit := 4000
         targetCM := net.casemap(target)
-        history, err := dc.user.msgStore.LoadLatestID(&net.Network, targetCM, msgID, limit)
+        history, err := dc.user.msgStore.LoadLatestID(ctx, &net.Network, targetCM, msgID, limit)
         if err != nil {
                 dc.logger.Printf("failed to send backlog for %q: %v", target, err)
@@ -2335 +2338 @@
                 eventPlayback := dc.caps["draft/event-playback"]
 
+                ctx, cancel := context.WithTimeout(context.TODO(), messageStoreTimeout)
+                defer cancel()
+
                 var history []*irc.Message
                 switch subcommand {
                 case "BEFORE":
-                        history, err = store.LoadBeforeTime(&network.Network, entity, bounds[0], time.Time{}, limit, eventPlayback)
+                        history, err = store.LoadBeforeTime(ctx, &network.Network, entity, bounds[0], time.Time{}, limit, eventPlayback)
                 case "AFTER":
-                        history, err = store.LoadAfterTime(&network.Network, entity, bounds[0], time.Now(), limit, eventPlayback)
+                        history, err = store.LoadAfterTime(ctx, &network.Network, entity, bounds[0], time.Now(), limit, eventPlayback)
                 case "BETWEEN":
                         if bounds[0].Before(bounds[1]) {
-                                history, err = store.LoadAfterTime(&network.Network, entity, bounds[0], bounds[1], limit, eventPlayback)
+                                history, err = store.LoadAfterTime(ctx, &network.Network, entity, bounds[0], bounds[1], limit, eventPlayback)
                         } else {
-                                history, err = store.LoadBeforeTime(&network.Network, entity, bounds[0], bounds[1], limit, eventPlayback)
+                                history, err = store.LoadBeforeTime(ctx, &network.Network, entity, bounds[0], bounds[1], limit, eventPlayback)
                         }
                 case "TARGETS":
                         // TODO: support TARGETS in multi-upstream mode
-                        targets, err := store.ListTargets(&network.Network, bounds[0], bounds[1], limit, eventPlayback)
+                        targets, err := store.ListTargets(ctx, &network.Network, bounds[0], bounds[1], limit, eventPlayback)
                         if err != nil {
                                 dc.logger.Printf("failed fetching targets for chathistory: %v", err)

trunk/msgstore.go

@@ -3 +3 @@
 import (
         "bytes"
+        "context"
         "encoding/base64"
         "fmt"
@@ -20 +21 @@
         // LoadLatestID queries the latest non-event messages for the given network,
         // entity and date, up to a count of limit messages, sorted from oldest to newest.
-        LoadLatestID(network *Network, entity, id string, limit int) ([]*irc.Message, error)
+        LoadLatestID(ctx context.Context, network *Network, entity, id string, limit int) ([]*irc.Message, error)
         Append(network *Network, entity string, msg *irc.Message) (id string, err error)
 }
@@ -38 +39 @@
         // both excluded. end may be before or after start.
         // If events is false, only PRIVMSG/NOTICE messages are considered.
-        ListTargets(network *Network, start, end time.Time, limit int, events bool) ([]chatHistoryTarget, error)
+        ListTargets(ctx context.Context, network *Network, start, end time.Time, limit int, events bool) ([]chatHistoryTarget, error)
         // LoadBeforeTime loads up to limit messages before start down to end. The
         // returned messages must be between and excluding the provided bounds.
         // end is before start.
         // If events is false, only PRIVMSG/NOTICE messages are considered.
-        LoadBeforeTime(network *Network, entity string, start, end time.Time, limit int, events bool) ([]*irc.Message, error)
+        LoadBeforeTime(ctx context.Context, network *Network, entity string, start, end time.Time, limit int, events bool) ([]*irc.Message, error)
         // LoadBeforeTime loads up to limit messages after start up to end. The
         // returned messages must be between and excluding the provided bounds.
         // end is after start.
         // If events is false, only PRIVMSG/NOTICE messages are considered.
-        LoadAfterTime(network *Network, entity string, start, end time.Time, limit int, events bool) ([]*irc.Message, error)
+        LoadAfterTime(ctx context.Context, network *Network, entity string, start, end time.Time, limit int, events bool) ([]*irc.Message, error)
 }
 

trunk/msgstore_fs.go

@@ -3 +3 @@
 import (
         "bufio"
+        "context"
         "fmt"
         "io"
@@ -477 +478 @@
 }
 
-func (ms *fsMessageStore) LoadBeforeTime(network *Network, entity string, start time.Time, end time.Time, limit int, events bool) ([]*irc.Message, error) {
+func (ms *fsMessageStore) LoadBeforeTime(ctx context.Context, network *Network, entity string, start time.Time, end time.Time, limit int, events bool) ([]*irc.Message, error) {
         start = start.In(time.Local)
         end = end.In(time.Local)
@@ -502 +503 @@
 }
 
-func (ms *fsMessageStore) LoadAfterTime(network *Network, entity string, start time.Time, end time.Time, limit int, events bool) ([]*irc.Message, error) {
+func (ms *fsMessageStore) LoadAfterTime(ctx context.Context, network *Network, entity string, start time.Time, end time.Time, limit int, events bool) ([]*irc.Message, error) {
         start = start.In(time.Local)
         end = end.In(time.Local)
@@ -526 +527 @@
 }
 
-func (ms *fsMessageStore) LoadLatestID(network *Network, entity, id string, limit int) ([]*irc.Message, error) {
+func (ms *fsMessageStore) LoadLatestID(ctx context.Context, network *Network, entity, id string, limit int) ([]*irc.Message, error) {
         var afterTime time.Time
         var afterOffset int64
@@ -570 +571 @@
 }
 
-func (ms *fsMessageStore) ListTargets(network *Network, start, end time.Time, limit int, events bool) ([]chatHistoryTarget, error) {
+func (ms *fsMessageStore) ListTargets(ctx context.Context, network *Network, start, end time.Time, limit int, events bool) ([]chatHistoryTarget, error) {
         start = start.In(time.Local)
         end = end.In(time.Local)

trunk/msgstore_memory.go

@@ -2 +2 @@
 
 import (
+        "context"
         "fmt"
         "time"
@@ -92 +93 @@
 }
 
-func (ms *memoryMessageStore) LoadLatestID(network *Network, entity, id string, limit int) ([]*irc.Message, error) {
+func (ms *memoryMessageStore) LoadLatestID(ctx context.Context, network *Network, entity, id string, limit int) ([]*irc.Message, error) {
         _, _, seq, err := parseMemoryMsgID(id)
         if err != nil {

trunk/server.go

@@ -26 +26 @@
 var upstreamMessageDelay = 2 * time.Second
 var upstreamMessageBurst = 10
+var messageStoreTimeout = 10 * time.Second
 
 type Logger interface {