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
-announcements and withdrawals. A `STATE` message says that the BGP session
-between a collector and a peer changed state, for example went down.
+that handler's queue. An `UPDATE` message carries announcements and withdrawals.
+A `STATE` message says that the BGP session between a collector and a peer
+changed state, for example went down.
 
-When a queue is more than half full, the streamer drops a share of the messages
-for that handler. The share rises from none at half full to all at full. A
-`STATE` message is never dropped. One that finds the queue full is held until
-there is room, and the handler's other messages are dropped until the held ones
-are in.
+The prefix handler's queue holds 100,000 messages. Each of the other handlers'
+queues holds 20,000. A message that finds its handler's queue full is dropped
+for that handler. A `STATE` message is never dropped. One that finds the queue
+full is held until there is room, and the handler's other messages are dropped
+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
-	// 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 {