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

README.md

@@ -543,7 +543,8 @@ What happens at each limit:
   changes again in a later write. Everything else in the failed write is
   dropped. The process keeps running.
 - The handler queues: the handlers are the four parts that take the feed's
-  messages (see Design). Each has a queue of at most 20,000 messages. When a
+  messages (see Design). The prefix handler's queue holds at most 100,000
+  messages, about 100 MiB. Each of the other three holds at most 20,000. When a
   queue fills, the streamer drops messages instead of waiting. The streamer is
   the part that reads the feed. The exception is a `STATE` message, which it
   holds, in feed order, until there is room. A `STATE` message says that a
@@ -703,15 +704,15 @@ carries are removed too, until they are announced again.
 
 The streamer reads the stream line by line. Each line is one message as JSON. It
 passes an `UPDATE` or `STATE` message to each handler that wants it, through
-that handler's queue of 20,000 messages. An `UPDATE` message carries
+that handler's queue. An `UPDATE` message carries announcements and withdrawals.
-announcements and withdrawals. A `STATE` message says that the BGP session
+A `STATE` message says that the BGP session between a collector and a peer
-between a collector and a peer changed state, for example went down.
+changed state, for example went down.
 
-When a queue is more than half full, the streamer drops a share of the messages
+The prefix handler's queue holds 100,000 messages. Each of the other handlers'
-for that handler. The share rises from none at half full to all at full. A
+queues holds 20,000. A message that finds its handler's queue full is dropped
-`STATE` message is never dropped. One that finds the queue full is held until
+for that handler. A `STATE` message is never dropped. One that finds the queue
-there is room, and the handler's other messages are dropped until the held ones
+full is held until there is room, and the handler's other messages are dropped
-are in.
+until the held ones are in.
 
 The AS, peer and prefix handlers write to the database in batches. A batch is
 written when it is full, or when its time is up:

TODO.md

@@ -57,6 +57,14 @@ problem the run finds.
 
 # Completed Steps
 
+- 2026-11-03: the streamer drops an update for a handler only when that
+  handler's queue is full. It used to drop a random share of the updates once
+  the queue was half full. The prefix handler's queue now holds 100,000
+  messages, about 100 MiB. It used to hold 20,000. While the routing table's
+  `Overview` walks every prefix, the prefix handler applies updates more slowly.
+  In a run on the live feed, the feed paused during such a walk and then sent
+  its backlog. The streamer then dropped about 10,000 of the prefix handler's
+  updates (closes #229)
 - 2026-11-02: `README.md` is rewritten from scratch, sentence by sentence,
   against the rules of https://git.eeqj.de/sneak/routewatch/issues/157. Every
   statement was checked against the code on `next` (closes #157)

internal/routewatch/prefixhandler.go

@@ -15,12 +15,15 @@ import (
 )
 
 const (
-	// prefixHandlerQueueSize is the queue capacity for prefix tracking
+	// prefixHandlerQueueSize is the most messages the handler's queue holds.
-	// operations, about 4 seconds of feed at peak. The streamer drops updates
+	// While the routing table's Overview or PeersDown walks every prefix, it
-	// rather than blocks when a queue is full (a STATE message is held until
+	// holds the table's lock about half of the time. The handler then applies
-	// there is room), so this bounds memory. Batches still flush on a timer
+	// updates more slowly. When the feed pauses for some seconds and then
-	// (prefixBatchTimeout).
+	// catches up, its backlog can arrive faster than that, and the queue
-	prefixHandlerQueueSize = 20000
+	// 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
+	// While the queue is full, the streamer drops the handler's messages,
-	// backpressure algorithm, except STATE messages, which are held until
+	// except STATE messages, which are held until there is room.
-	// 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
+// backoff and dispatches messages to registered handlers via per-handler queues.
-// implements backpressure to prevent queue overflow during high traffic periods.
+// 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)
+// A message that finds the queue full is dropped, except a STATE message. A
-// or full, except a STATE message: one whose state is down is the only notice
+// STATE message whose state is down is the only notice that all of a peer's
-// that all of a peer's routes are gone. A STATE message skips the random drop,
+// routes are gone. A STATE message that finds the queue full is held in
-// and one that finds the queue full is held in info.heldState. Held messages
+// info.heldState instead. Held messages go into the queue, oldest first, when a
-// go into the queue, oldest first, before anything else for the handler, as
+// later message for the handler finds room, or when Stop is called. While any
-// soon as there is room, or when Stop is called; while any is still held, the
+// message is held, the handler's other messages are dropped, so none of them
-// handler's other messages are dropped, so none of them overtakes a held down.
+// 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 {