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..830ae6a 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. +// 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 -} 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 {