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 handlers that want them and
never drops one for backpressure: it waits for room in a full queue until
the idle timer drops the connection. When the prefix handler gets one whose
state is down, it 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 the peer sends 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,7 @@ 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 wait for room.
 	QueueCapacity() int
 }
 
@@ -629,7 +629,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 +668,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 +693,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(connCtx, 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 +726,34 @@ func (s *Streamer) stream(ctx context.Context) error {
 	return errStreamEnded
 }
 
+// queueMessage puts msg on the handler's queue and reports whether it did. 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, so a STATE message waits for room
+// instead. The wait ends if the idle timer drops the connection by cancelling
+// connCtx, so the read loop never waits longer than its idle timeout.
+func (s *Streamer) queueMessage(connCtx context.Context, info *handlerInfo, msg *ristypes.RISMessage) bool {
+	if msg.Type == ristypes.MessageTypeState {
+		select {
+		case info.queue <- msg:
+			return true
+		case <-connCtx.Done():
+			return false
+		}
+	}
+
+	if s.shouldDropForBackpressure(info) {
+		return false
+	}
+
+	select {
+	case info.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,143 @@ 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")
+	}
+}
+
+// streamWithFullStateQueue runs stream, in a goroutine, against a streamer
+// reading from url whose stateHandler's queue is full of STATE messages with
+// no peer. It returns the queue and a channel that gets what stream returns.
+func streamWithFullStateQueue(url string, idleTimeout time.Duration) (
+	queue chan *ristypes.RISMessage, streamErr chan error) {
+	s := New(logger.New(), metrics.New())
+	s.url = url
+	s.idleTimeout = idleTimeout
+	s.RegisterHandler(stateHandler{})
+	queue = s.handlers[0].queue
+	for len(queue) < cap(queue) {
+		queue <- &ristypes.RISMessage{Type: ristypes.MessageTypeState}
+	}
+
+	streamErr = make(chan error, 1)
+	go func() { streamErr <- s.stream(context.Background()) }()
+
+	return queue, streamErr
+}
+
+// TestStreamWaitsForRoomForAStateMessage has the server send one STATE message
+// whose state is down while the queue of the handler that wants it is full. A
+// full queue is where the random drop always drops, and where a message that
+// cannot wait is dropped. The streamer must not drop it: it waits until the
+// handler takes a message off its queue, then queues it.
+func TestStreamWaitsForRoomForAStateMessage(t *testing.T) {
+	const (
+		stillWaitingAfter = 100 * time.Millisecond
+		maxWait           = 5 * time.Second
+	)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		_, _ = io.WriteString(w, stateDownLine)
+	}))
+	defer srv.Close()
+
+	queue, streamErr := streamWithFullStateQueue(srv.URL, streamIdleTimeout)
+
+	select {
+	case err := <-streamErr:
+		t.Fatalf("stream returned %v while the queue was full: it dropped the STATE message", err)
+	case <-time.After(stillWaitingAfter):
+	}
+
+	<-queue // the handler takes a message, which makes room
+	select {
+	case err := <-streamErr:
+		if !errors.Is(err, errStreamEnded) {
+			t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
+		}
+	case <-time.After(maxWait):
+		t.Fatalf("stream did not queue the STATE message within %s of there being room", maxWait)
+	}
+
+	var last *ristypes.RISMessage
+	for len(queue) > 0 {
+		last = <-queue
+	}
+	if last.Peer != stateDownPeer || last.State != ristypes.StateDown {
+		t.Errorf("last queued message has peer %q, state %q; want %q, %q",
+			last.Peer, last.State, stateDownPeer, ristypes.StateDown)
+	}
+}
+
+// TestStreamStopsWaitingForRoomAtTheIdleTimeout has the server send one STATE
+// message and then hold the response open, while the queue of the handler
+// that wants it stays full. The wait for room must not stall the read loop
+// past its idle timeout: the stream ends as when no line arrives.
+func TestStreamStopsWaitingForRoomAtTheIdleTimeout(t *testing.T) {
+	const (
+		idleTimeout = 50 * time.Millisecond
+		maxWait     = 5 * time.Second
+	)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = io.WriteString(w, stateDownLine)
+		w.(http.Flusher).Flush()
+		<-r.Context().Done()
+	}))
+	defer srv.Close()
+
+	_, streamErr := streamWithFullStateQueue(srv.URL, idleTimeout)
+
+	select {
+	case err := <-streamErr:
+		if err == nil || errors.Is(err, errStreamEnded) {
+			t.Errorf("stream returned %v, want the error of a stream dropped as idle", err)
+		}
+	case <-time.After(maxWait):
+		t.Fatalf("stream still waiting for room %s after the idle timeout of %s", maxWait, idleTimeout)
+	}
+}
+
 // 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