Drop an update only when its handler's queue is full (closes #229)

The streamer dropped a random share of a handler's updates once its
queue was half full. When the feed paused and then sent its backlog
while the routing table's Overview walked every prefix, the prefix
handler lost about 10,000 updates with its queue never full.

The streamer now drops an update only when the queue is full. The
prefix handler's queue holds 100,000 messages, about 100 MiB, so it
can hold such a backlog. The README's Memory and Feed handling
sections give the new sizes and the rule.

Model: opus-5-5

internal/routewatch/prefixhandler.go

@@ -15,12 +15,15 @@ import (
 )
 
 const (
-	// prefixHandlerQueueSize is the queue capacity for prefix tracking
-	// operations, about 4 seconds of feed at peak. The streamer drops updates
-	// rather than blocks when a queue is full (a STATE message is held until
-	// there is room), so this bounds memory. Batches still flush on a timer
-	// (prefixBatchTimeout).
-	prefixHandlerQueueSize = 20000
+	// prefixHandlerQueueSize is the most messages the handler's queue holds.
+	// While the routing table's Overview or PeersDown walks every prefix, it
+	// holds the table's lock about half of the time. The handler then applies
+	// updates more slowly. When the feed pauses for some seconds and then
+	// catches up, its backlog can arrive faster than that, and the queue
+	// holds the backlog. The streamer drops an update that finds the queue
+	// full. A message takes about 1 KiB, so a full queue takes about 100 MiB
+	// (README "Memory").
+	prefixHandlerQueueSize = 100000
 
 	// prefixBatchSize is the number of prefix updates to batch together
 	prefixBatchSize = 25000

internal/streamer/streamer.go

@@ -12,7 +12,6 @@ import (
 	"fmt"
 	"io"
 	"math"
-	"math/rand"
 	"net/http"
 	"sync"
 	"sync/atomic"
@@ -58,10 +57,6 @@ const (
 	// Well above an extended BGP message (RFC 8654) in hex plus its prefixes
 	maxLineSize           = bytesPerMB
 	maxConcurrentHandlers = 800 // Maximum number of concurrent message handlers
-
-	// Backpressure constants
-	backpressureThreshold = 0.5 // Start dropping at 50% queue utilization
-	backpressureSlope     = 2.0 // Slope for linear drop probability increase
 )
 
 // messageTypeNotification is the type of a BGP NOTIFICATION message, which ends
@@ -93,9 +88,8 @@ type MessageHandler interface {
 
 	// QueueCapacity returns the desired queue capacity for this handler.
 	// Handlers that process quickly can have larger queues to buffer bursts.
-	// When the queue fills up, messages will be dropped according to the
-	// backpressure algorithm, except STATE messages, which are held until
-	// there is room.
+	// While the queue is full, the streamer drops the handler's messages,
+	// except STATE messages, which are held until there is room.
 	QueueCapacity() int
 }
 
@@ -126,8 +120,8 @@ type handlerInfo struct {
 
 // Streamer manages a connection to the RIPE RIS Live streaming API for receiving
 // real-time BGP UPDATE messages. It handles automatic reconnection with exponential
-// backoff, dispatches messages to registered handlers via per-handler queues, and
-// implements backpressure to prevent queue overflow during high traffic periods.
+// backoff and dispatches messages to registered handlers via per-handler queues.
+// Its read loop never waits for room on a queue (see queueMessage).
 type Streamer struct {
 	logger *logger.Logger
 	client *http.Client
@@ -144,7 +138,6 @@ type Streamer struct {
 	workers      sync.WaitGroup // the handler workers; Stop waits for them
 	metrics      *metrics.Tracker
 	totalDropped atomic.Uint64   // Total dropped messages across all handlers
-	random       *rand.Rand      // Random number generator for backpressure drops
 	bgpPeers     map[string]bool // Track active BGP peers by peer IP
 	bgpPeersMu   sync.RWMutex    // Protects bgpPeers map
 }
@@ -168,8 +161,6 @@ func New(logger *logger.Logger, metrics *metrics.Tracker) *Streamer {
 		},
 		handlers: make([]*handlerInfo, 0),
 		metrics:  metrics,
-		//nolint:gosec // Non-cryptographic randomness is fine for backpressure
-		random:   rand.New(rand.NewSource(time.Now().UnixNano())),
 		bgpPeers: make(map[string]bool),
 	}
 }
@@ -408,7 +399,7 @@ func (s *Streamer) GetHandlerStats() []HandlerStats {
 }
 
 // GetDroppedMessages returns the total number of messages dropped across all
-// handlers due to queue overflow or backpressure. This counter is monotonically
+// handlers because their queue was full. This counter is monotonically
 // increasing and is safe to call concurrently.
 func (s *Streamer) GetDroppedMessages() uint64 {
 	return s.totalDropped.Load()
@@ -891,13 +882,13 @@ func (s *Streamer) dispatchMessage(
 // msg instead. It never waits for room, so a full queue does not hold up the
 // read loop.
 //
-// A message is dropped while the queue is filling up (shouldDropForBackpressure)
-// or full, except a STATE message: one whose state is down is the only notice
-// that all of a peer's routes are gone. A STATE message skips the random drop,
-// and one that finds the queue full is held in info.heldState. Held messages
-// go into the queue, oldest first, before anything else for the handler, as
-// soon as there is room, or when Stop is called; while any is still held, the
-// handler's other messages are dropped, so none of them overtakes a held down.
+// A message that finds the queue full is dropped, except a STATE message. A
+// STATE message whose state is down is the only notice that all of a peer's
+// routes are gone. A STATE message that finds the queue full is held in
+// info.heldState instead. Held messages go into the queue, oldest first, when a
+// later message for the handler finds room, or when Stop is called. While any
+// message is held, the handler's other messages are dropped, so none of them
+// overtakes a held STATE message.
 //
 // The held list stays small: a STATE message is a few hundred bytes and the
 // feed sends about ten a second, so a queue full for a minute holds about 600.
@@ -916,7 +907,7 @@ func (s *Streamer) queueMessage(
 		return true
 	}
 
-	if len(info.heldState) > 0 || s.shouldDropForBackpressure(info) {
+	if len(info.heldState) > 0 {
 		return false
 	}
 
@@ -932,26 +923,3 @@ func trySend(queue chan *ristypes.RISMessage, msg *ristypes.RISMessage) bool {
 		return false
 	}
 }
-
-// shouldDropForBackpressure determines if a message should be dropped based on
-// queue utilization
-func (s *Streamer) shouldDropForBackpressure(info *handlerInfo) bool {
-	// Calculate queue utilization
-	queueLen := len(info.queue)
-	queueCap := cap(info.queue)
-	utilization := float64(queueLen) / float64(queueCap)
-
-	// No drops below threshold
-	if utilization < backpressureThreshold {
-		return false
-	}
-
-	// Calculate drop probability (0.0 at threshold, 1.0 at 100% full)
-	dropProbability := (utilization - backpressureThreshold) * backpressureSlope
-	if dropProbability > 1.0 {
-		dropProbability = 1.0
-	}
-
-	// Random drop based on probability
-	return s.random.Float64() < dropProbability
-}

internal/streamer/streamer_internal_test.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"fmt"
 	"io"
 	"log/slog"
 	"net/http"
@@ -258,6 +259,65 @@ func TestHeldStateMessagesKeepTheirPlaceInTheQueue(t *testing.T) {
 	}
 }
 
+// burstQueueCapacity is the size of a burstHandler's queue.
+const burstQueueCapacity = 1000
+
+// burstHandler wants UPDATE messages and has a queue of burstQueueCapacity.
+// Tests read what it is handed from its queue, so HandleMessage does nothing.
+type burstHandler struct{}
+
+func (burstHandler) WantsMessage(messageType string) bool {
+	return messageType == ristypes.MessageTypeUpdate
+}
+
+func (burstHandler) QueueCapacity() int { return burstQueueCapacity }
+
+func (burstHandler) HandleMessage(*ristypes.RISMessage) {}
+
+// TestUpdatesDroppedOnlyWhenTheQueueIsFull has the server send one UPDATE more
+// than the handler's queue holds, each with its own timestamp. Nothing takes
+// messages off the queue meanwhile. The streamer must queue the updates in
+// feed order until the queue is full, and drop only the last one.
+func TestUpdatesDroppedOnlyWhenTheQueueIsFull(t *testing.T) {
+	t.Parallel()
+
+	const (
+		firstTimestamp = 1700000000
+		lineFormat     = `{"type":"ris_message","data":{"type":"UPDATE",` +
+			`"peer":"192.0.2.1","peer_asn":"64496","timestamp":%d}}` + "\n"
+	)
+
+	var lines strings.Builder
+
+	for i := range burstQueueCapacity + 1 {
+		fmt.Fprintf(&lines, lineFormat, firstTimestamp+i)
+	}
+
+	s := New(logger.New(), metrics.New())
+	s.RegisterHandler(burstHandler{})
+
+	streamLines(t, s, lines.String())
+
+	dropped := s.GetHandlerStats()[0].DroppedCount
+	if dropped != 1 {
+		t.Errorf("%d updates dropped, want only the one that found the queue "+
+			"full", dropped)
+	}
+
+	queue := s.handlers[0].queue
+	if len(queue) != burstQueueCapacity {
+		t.Fatalf("%d updates queued, want %d", len(queue), burstQueueCapacity)
+	}
+
+	for i := range burstQueueCapacity {
+		msg := <-queue
+		if msg.Timestamp != float64(firstTimestamp+i) {
+			t.Fatalf("update %d on the queue has timestamp %.0f, want %d",
+				i, msg.Timestamp, firstTimestamp+i)
+		}
+	}
+}
+
 // takeAll takes the messages off queue, as a handler's worker would, and sends
 // them on the returned channel once queue is closed.
 func takeAll(queue chan *ristypes.RISMessage) chan []*ristypes.RISMessage {