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/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.
+// It 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: one
+// 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. 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.
 //
 // 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
-}