From 49d2f4ede8ed2541ebd88b391787c96faad35b75 Mon Sep 17 00:00:00 2001 From: sneak Date: Tue, 3 Nov 2026 16:36:19 +0000 Subject: [PATCH] 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 | 19 +++---- TODO.md | 8 +++ internal/routewatch/prefixhandler.go | 15 +++--- internal/streamer/streamer.go | 58 +++++--------------- internal/streamer/streamer_internal_test.go | 60 +++++++++++++++++++++ 5 files changed, 100 insertions(+), 60 deletions(-) diff --git a/README.md b/README.md index 0d81f41..19c3bbc 100644 --- a/README.md +++ b/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: diff --git a/TODO.md b/TODO.md index 20effb1..54beb9d 100644 --- a/TODO.md +++ b/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) diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index c7068c3..3de5114 100644 --- a/internal/routewatch/prefixhandler.go +++ b/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 diff --git a/internal/streamer/streamer.go b/internal/streamer/streamer.go index b90b8cc..1844068 100644 --- a/internal/streamer/streamer.go +++ b/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 -} diff --git a/internal/streamer/streamer_internal_test.go b/internal/streamer/streamer_internal_test.go index e124caa..7e0b23a 100644 --- a/internal/streamer/streamer_internal_test.go +++ b/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 {