Remove a peer's live routes when its session goes down (closes #56)

RIS sends no withdrawals for the routes of a peer whose BGP session goes
down, so those routes stayed in the live tables for good.

The streamer now hands STATE messages to the prefix handler and never drops
one for backpressure: one that finds the queue full is held, and queued in
feed order ahead of anything else for that handler once there is room. On a
down, the prefix handler drops the peer's updates still in its batch and
notes the peer; the next flush deletes all of the peer's live routes, IPv4
and IPv6, before writing the batch, so updates sent after the down are kept.
The live route tables' unique key now leads with peer_ip, so it also finds a
peer's routes.

Model: opus-5-5

internal/streamer/streamer.go

@@ -78,7 +78,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.
+	// backpressure algorithm, except STATE messages, which are held until
+	// there is room.
 	QueueCapacity() int
 }
 
@@ -102,6 +103,9 @@ type handlerInfo struct {
 	handler MessageHandler
 	queue   chan *ristypes.RISMessage
 	metrics handlerMetrics
+	// heldState holds, in feed order, the STATE messages that found queue
+	// full (see queueMessage). Only the read loop uses it.
+	heldState []*ristypes.RISMessage
 }
 
 // Streamer manages a connection to the RIPE RIS Live streaming API for receiving
@@ -629,7 +633,7 @@ func (s *Streamer) stream(ctx context.Context) error {
 
 		// Process based on message type
 		switch msg.Type {
-		case "UPDATE":
+		case ristypes.MessageTypeUpdate:
 			// Process BGP UPDATE messages
 			// Will be dispatched to handlers
 		case "RIS_PEER_STATE":
@@ -668,9 +672,9 @@ func (s *Streamer) stream(ctx context.Context) error {
 			)
 
 			continue
-		case "STATE":
-			// Peer state changes - silently ignore
-			continue
+		case ristypes.MessageTypeState:
+			// The BGP session with the peer changed state, for example went
+			// down. Dispatched to the handlers that want it.
 		default:
 			s.logger.Warn("Unknown message type, skipping",
 				"type", msg.Type,
@@ -693,30 +697,20 @@ func (s *Streamer) stream(ctx context.Context) error {
 				continue
 			}
 
-			// Check if we should drop due to backpressure
-			if s.shouldDropForBackpressure(info) {
+			if !s.queueMessage(info, &msg) {
 				atomic.AddUint64(&info.metrics.droppedCount, 1)
 				atomic.AddUint64(&s.totalDropped, 1)
 
 				continue
 			}
 
-			// Try to queue the message
-			select {
-			case info.queue <- &msg:
-				// Message queued successfully
-				// Update high water mark if needed
-				queueLen := len(info.queue)
-				info.metrics.mu.Lock()
-				if queueLen > info.metrics.queueHighWaterMark {
-					info.metrics.queueHighWaterMark = queueLen
-				}
-				info.metrics.mu.Unlock()
-			default:
-				// Queue is full, drop the message
-				atomic.AddUint64(&info.metrics.droppedCount, 1)
-				atomic.AddUint64(&s.totalDropped, 1)
+			// Update high water mark if needed
+			queueLen := len(info.queue)
+			info.metrics.mu.Lock()
+			if queueLen > info.metrics.queueHighWaterMark {
+				info.metrics.queueHighWaterMark = queueLen
 			}
+			info.metrics.mu.Unlock()
 		}
 		s.mu.RUnlock()
 	}
@@ -736,6 +730,50 @@ func (s *Streamer) stream(ctx context.Context) error {
 	return errStreamEnded
 }
 
+// queueMessage puts msg on the handler's queue and reports false if it dropped
+// msg instead. It never waits for room, so a full queue holds up neither the
+// read loop nor Stop.
+//
+// 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; 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.
+func (s *Streamer) queueMessage(info *handlerInfo, msg *ristypes.RISMessage) bool {
+	for len(info.heldState) > 0 && trySend(info.queue, info.heldState[0]) {
+		info.heldState = info.heldState[1:]
+	}
+
+	if msg.Type == ristypes.MessageTypeState {
+		if len(info.heldState) > 0 || !trySend(info.queue, msg) {
+			info.heldState = append(info.heldState, msg)
+		}
+
+		return true
+	}
+
+	if len(info.heldState) > 0 || s.shouldDropForBackpressure(info) {
+		return false
+	}
+
+	return trySend(info.queue, msg)
+}
+
+// trySend puts msg on queue if there is room and reports whether it did.
+func trySend(queue chan *ristypes.RISMessage, msg *ristypes.RISMessage) bool {
+	select {
+	case queue <- msg:
+		return true
+	default:
+		return false
+	}
+}
+
 // shouldDropForBackpressure determines if a message should be dropped based on queue utilization
 func (s *Streamer) shouldDropForBackpressure(info *handlerInfo) bool {
 	// Calculate queue utilization

internal/streamer/streamer_test.go

@@ -108,6 +108,204 @@ func (h updateHandler) HandleMessage(*ristypes.RISMessage) {
 	}
 }
 
+// stateDownLine is the STATE message RIS Live sends when the collector's BGP
+// session with the peer stateDownPeer goes down.
+const (
+	stateDownPeer = "192.0.2.1"
+	stateDownLine = `{"type":"ris_message","data":{"type":"STATE","peer":"` + stateDownPeer + `",` +
+		`"peer_asn":"64496","timestamp":1700000000,"state":"down"}}` + "\n"
+)
+
+// stateHandler wants STATE messages only. Tests read what it is handed from
+// its queue, so HandleMessage does nothing.
+type stateHandler struct{}
+
+func (stateHandler) WantsMessage(messageType string) bool {
+	return messageType == ristypes.MessageTypeState
+}
+func (stateHandler) QueueCapacity() int                 { return 10 }
+func (stateHandler) HandleMessage(*ristypes.RISMessage) {}
+
+// TestStreamHandsStateMessagesToHandlersThatWantThem has the server send one
+// STATE message whose state is down. The streamer must queue it, with its
+// peer and state, for a handler that wants STATE messages.
+func TestStreamHandsStateMessagesToHandlersThatWantThem(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, stateDownLine)
+	}))
+	defer srv.Close()
+
+	s := New(logger.New(), metrics.New())
+	s.url = srv.URL
+	s.RegisterHandler(stateHandler{})
+
+	if err := s.stream(context.Background()); !errors.Is(err, errStreamEnded) {
+		t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
+	}
+
+	select {
+	case msg := <-s.handlers[0].queue:
+		if msg.Type != ristypes.MessageTypeState || msg.Peer != stateDownPeer || msg.State != ristypes.StateDown {
+			t.Errorf("queued message has type %q, peer %q, state %q; want %q, %q, %q",
+				msg.Type, msg.Peer, msg.State, ristypes.MessageTypeState, stateDownPeer, ristypes.StateDown)
+		}
+	default:
+		t.Fatal("the STATE message was not queued for the handler that wants it")
+	}
+}
+
+// TestHeldStateMessagesKeepTheirPlaceInTheQueue hands queueMessage two STATE
+// messages and two updates while the queue is full, making room in between.
+// Neither STATE message may be dropped: each must reach the queue after
+// everything queued before it and before anything handed over after it. The
+// update handed over while a STATE message is held is dropped.
+func TestHeldStateMessagesKeepTheirPlaceInTheQueue(t *testing.T) {
+	s := New(logger.New(), metrics.New())
+	s.RegisterHandler(stateHandler{})
+	info := s.handlers[0]
+
+	before := &ristypes.RISMessage{Type: ristypes.MessageTypeUpdate}
+	for len(info.queue) < cap(info.queue) {
+		info.queue <- before
+	}
+	firstDown := &ristypes.RISMessage{Type: ristypes.MessageTypeState, State: ristypes.StateDown}
+	secondDown := &ristypes.RISMessage{Type: ristypes.MessageTypeState, State: ristypes.StateDown}
+	updateWhileHeld := &ristypes.RISMessage{Type: ristypes.MessageTypeUpdate}
+	after := &ristypes.RISMessage{Type: ristypes.MessageTypeUpdate}
+
+	if !s.queueMessage(info, firstDown) {
+		t.Fatal("the first STATE message was dropped on a full queue")
+	}
+	<-info.queue // the handler makes room for one message
+	if !s.queueMessage(info, secondDown) {
+		t.Fatal("the second STATE message was dropped")
+	}
+	if s.queueMessage(info, updateWhileHeld) {
+		t.Fatal("an update was queued while a STATE message was held")
+	}
+
+	// The handler takes every message queued before the first STATE message.
+	for range cap(info.queue) - 1 {
+		if <-info.queue != before {
+			t.Fatal("a message overtook those queued before the first STATE message")
+		}
+	}
+	if !s.queueMessage(info, after) {
+		t.Fatal("an update was dropped on a queue with room")
+	}
+
+	if <-info.queue != firstDown {
+		t.Fatal("the first STATE message did not come next")
+	}
+	if <-info.queue != secondDown {
+		t.Fatal("the second STATE message did not come after the first")
+	}
+	if <-info.queue != after {
+		t.Fatal("the update handed over after the STATE messages did not come after them")
+	}
+	if len(info.queue) != 0 {
+		t.Errorf("%d messages left on the queue, want none", len(info.queue))
+	}
+}
+
+// newStreamerWithFullStateQueue returns a streamer reading from url with one
+// handler, a stateHandler whose queue is full.
+func newStreamerWithFullStateQueue(url string) *Streamer {
+	s := New(logger.New(), metrics.New())
+	s.url = url
+	s.RegisterHandler(stateHandler{})
+	queue := s.handlers[0].queue
+	for len(queue) < cap(queue) {
+		queue <- &ristypes.RISMessage{Type: ristypes.MessageTypeState}
+	}
+
+	return s
+}
+
+// TestOtherHandlersKeepReceivingWhileAStateMessageIsHeld has the server send a
+// STATE message whose state is down, then an UPDATE, while the queue of the
+// handler that wants STATE messages is full. The STATE message must be held
+// for that handler, and the read loop must go on to queue the UPDATE for the
+// handler that wants it.
+func TestOtherHandlersKeepReceivingWhileAStateMessageIsHeld(t *testing.T) {
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, stateDownLine+updateLine)
+	}))
+	defer srv.Close()
+
+	s := newStreamerWithFullStateQueue(srv.URL)
+	s.RegisterHandler(updateHandler{})
+
+	if err := s.stream(context.Background()); !errors.Is(err, errStreamEnded) {
+		t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
+	}
+
+	held := s.handlers[0].heldState
+	if len(held) != 1 || held[0].Peer != stateDownPeer || held[0].State != ristypes.StateDown {
+		t.Errorf("held %d STATE messages, want the down from %s alone", len(held), stateDownPeer)
+	}
+	if got := len(s.handlers[1].queue); got != 1 {
+		t.Errorf("%d messages queued for the handler that wants UPDATE messages, want 1", got)
+	}
+}
+
+// TestStopReturnsAtOnceWithStateMessagesHeld has the server send two STATE
+// messages and hold the response open while the queue of the handler that
+// wants them is full. Stop must return at once and end the stream.
+func TestStopReturnsAtOnceWithStateMessagesHeld(t *testing.T) {
+	const (
+		lines   = 2
+		maxWait = 5 * time.Second
+	)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.WriteString(w, stateDownLine+stateDownLine)
+		w.(http.Flusher).Flush()
+		<-r.Context().Done()
+	}))
+	defer srv.Close()
+
+	s := newStreamerWithFullStateQueue(srv.URL)
+	// The raw handler sees each line before the read loop hands it to the
+	// queues, so once it has seen the second line, the first is held.
+	linesSeen := make(chan struct{}, lines)
+	s.RegisterRawHandler(func(string) { linesSeen <- struct{}{} })
+
+	// As in TestStopBeforeMessageReachesQueues, setting cancel as Start does
+	// lets Stop cancel the stream run here.
+	ctx, cancel := context.WithCancel(context.Background())
+	s.cancel = cancel
+	streamErr := make(chan error, 1)
+	go func() { streamErr <- s.stream(ctx) }()
+
+	for range lines {
+		select {
+		case <-linesSeen:
+		case <-time.After(maxWait):
+			t.Fatalf("the read loop did not read %d lines within %s", lines, maxWait)
+		}
+	}
+
+	stopped := make(chan struct{})
+	go func() {
+		s.Stop()
+		close(stopped)
+	}()
+	select {
+	case <-stopped:
+	case <-time.After(maxWait):
+		t.Fatalf("Stop did not return within %s while a STATE message was held", maxWait)
+	}
+
+	select {
+	case err := <-streamErr:
+		if !errors.Is(err, context.Canceled) {
+			t.Errorf("stream returned %v, want %v", err, context.Canceled)
+		}
+	case <-time.After(maxWait):
+		t.Fatalf("stream did not end within %s of Stop", maxWait)
+	}
+}
+
 // TestStopBeforeMessageReachesQueues stops the streamer after the read loop
 // has checked for cancellation but before it hands the message to the handler
 // queues. That is the gap a stop from another goroutine can land in, and it