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

README.md

@@ -235,7 +235,9 @@ What happens at each limit:
   out-of-memory error, and the handler logs it and drops that batch. The process
   keeps running.
 - Handler queues: each of the four handler queues holds at most 20,000 messages.
-  When a queue fills, the streamer drops messages instead of blocking.
+  When a queue fills, the streamer drops messages instead of blocking, except a
+  `STATE` message, which it holds, in feed order, until there is room: a few
+  hundred bytes each, about ten a second.
 - Handler batches: while the AS, peer or prefix handler writes a batch to the
   database, the updates that arrive collect in its next batch, which stops at
   ten times the batch size: 300,000 updates for the AS handler

internal/routewatch/peerhandler.go

@@ -1,7 +1,7 @@
 package routewatch
 
 // peerhandler.go provides batched peer tracking functionality for BGP route monitoring.
-// It tracks BGP peers from all incoming RIS messages and maintains peer state in the database.
+// It tracks BGP peers from incoming UPDATE messages and maintains peer state in the database.
 
 import (
 	"strconv"
@@ -15,9 +15,9 @@ import (
 
 const (
 	// peerHandlerQueueSize is the queue capacity for peer tracking operations,
-	// about 4 seconds of feed at peak. The streamer drops messages rather than
+	// about 4 seconds of feed at peak. The streamer drops rather than blocks
-	// blocks when a queue is full (a STATE message waits for room), so this
+	// when a queue is full, so this bounds memory. Batches still flush on a
-	// bounds memory. Batches still flush on a timer (peerBatchTimeout).
+	// timer (peerBatchTimeout).
 	peerHandlerQueueSize = 20000
 
 	// peerBatchSize is the number of peer updates to batch together
@@ -33,7 +33,7 @@ const (
 	peerBatchTimeout = 2 * time.Second
 )
 
-// PeerHandler tracks BGP peers from all message types using batched operations.
+// PeerHandler tracks BGP peers from UPDATE messages using batched operations.
 // It maintains a queue of peer updates and periodically flushes them to the database
 // in batches to improve performance. The handler deduplicates peer updates within
 // each batch, keeping only the most recent update for each peer IP address.
@@ -82,11 +82,11 @@ func NewPeerHandler(db database.Store, logger *logger.Logger) *PeerHandler {
 	return h
 }
 
-// WantsMessage returns true for all message types since peer information
+// WantsMessage returns true for UPDATE messages only. A peer whose session
-// is extracted from every RIS message regardless of type. This satisfies
+// never comes up sends STATE messages and nothing else, so a STATE message is
-// the MessageHandler interface.
+// not a sign of a peer that carries routes.
-func (h *PeerHandler) WantsMessage(_ string) bool {
+func (h *PeerHandler) WantsMessage(messageType string) bool {
-	return true
+	return messageType == ristypes.MessageTypeUpdate
 }
 
 // QueueCapacity returns the desired queue capacity for this handler.

internal/routewatch/prefixhandler.go

@@ -18,8 +18,8 @@ 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 waits for
+	// rather than blocks when a queue is full (a STATE message is held until
-	// room), so this bounds memory. Batches still flush on a timer
+	// there is room), so this bounds memory. Batches still flush on a timer
 	// (prefixBatchTimeout).
 	prefixHandlerQueueSize = 20000
 

internal/routewatch/prefixhandler_test.go

@@ -289,6 +289,21 @@ func TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer(t *testing.T) {
 	}
 }
 
+// TestStateMessagesGoToThePrefixHandlerNotThePeerHandler checks that STATE
+// messages go to the prefix handler, which acts on a down, and not to the peer
+// handler, which would record a peer whose session never comes up as active.
+func TestStateMessagesGoToThePrefixHandlerNotThePeerHandler(t *testing.T) {
+	if !(&PrefixHandler{}).WantsMessage(ristypes.MessageTypeState) {
+		t.Error("the prefix handler does not want STATE messages")
+	}
+	if (&PeerHandler{}).WantsMessage(ristypes.MessageTypeState) {
+		t.Error("the peer handler wants STATE messages")
+	}
+	if !(&PeerHandler{}).WantsMessage(ristypes.MessageTypeUpdate) {
+		t.Error("the peer handler does not want UPDATE messages")
+	}
+}
+
 // TestPrefixHandlerDeletesAPeersRoutesWhenItsSessionGoesDown runs
 // announcements from two peers through the prefix handler into a real
 // database, then a batch in which one of the peers announces a prefix and its

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, except STATE messages, which wait for room.
+	// 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
@@ -693,7 +697,7 @@ func (s *Streamer) stream(ctx context.Context) error {
 				continue
 			}
 
-			if !s.queueMessage(connCtx, info, &msg) {
+			if !s.queueMessage(info, &msg) {
 				atomic.AddUint64(&info.metrics.droppedCount, 1)
 				atomic.AddUint64(&s.totalDropped, 1)
 
@@ -726,28 +730,44 @@ func (s *Streamer) stream(ctx context.Context) error {
 	return errStreamEnded
 }
 
-// queueMessage puts msg on the handler's queue and reports whether it did. A
+// queueMessage puts msg on the handler's queue and reports false if it dropped
-// message is dropped while the queue is filling up (shouldDropForBackpressure)
+// 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, so a STATE message waits for room
+// that all of a peer's routes are gone. A STATE message skips the random drop,
-// instead. The wait ends if the idle timer drops the connection by cancelling
+// and one that finds the queue full is held in info.heldState. Held messages
-// connCtx, so the read loop never waits longer than its idle timeout.
+// go into the queue, oldest first, before anything else for the handler, as
-func (s *Streamer) queueMessage(connCtx context.Context, info *handlerInfo, msg *ristypes.RISMessage) bool {
+// 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 {
-		select {
+		if len(info.heldState) > 0 || !trySend(info.queue, msg) {
-		case info.queue <- msg:
+			info.heldState = append(info.heldState, msg)
+		}
+
 		return true
-		case <-connCtx.Done():
-			return false
-		}
 	}
 
-	if s.shouldDropForBackpressure(info) {
+	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 info.queue <- msg:
+	case queue <- msg:
 		return true
 	default:
 		return false

internal/streamer/streamer_test.go

@@ -154,94 +154,155 @@ func TestStreamHandsStateMessagesToHandlersThatWantThem(t *testing.T) {
 	}
 }
 
-// streamWithFullStateQueue runs stream, in a goroutine, against a streamer
+// TestHeldStateMessagesKeepTheirPlaceInTheQueue hands queueMessage two STATE
-// reading from url whose stateHandler's queue is full of STATE messages with
+// messages and two updates while the queue is full, making room in between.
-// no peer. It returns the queue and a channel that gets what stream returns.
+// Neither STATE message may be dropped: each must reach the queue after
-func streamWithFullStateQueue(url string, idleTimeout time.Duration) (
+// everything queued before it and before anything handed over after it. The
-	queue chan *ristypes.RISMessage, streamErr chan error) {
+// 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.idleTimeout = idleTimeout
 	s.RegisterHandler(stateHandler{})
-	queue = s.handlers[0].queue
+	queue := s.handlers[0].queue
 	for len(queue) < cap(queue) {
 		queue <- &ristypes.RISMessage{Type: ristypes.MessageTypeState}
 	}
 
-	streamErr = make(chan error, 1)
+	return s
-	go func() { streamErr <- s.stream(context.Background()) }()
-
-	return queue, streamErr
 }
 
-// TestStreamWaitsForRoomForAStateMessage has the server send one STATE message
+// TestOtherHandlersKeepReceivingWhileAStateMessageIsHeld has the server send a
-// whose state is down while the queue of the handler that wants it is full. A
+// STATE message whose state is down, then an UPDATE, while the queue of the
-// full queue is where the random drop always drops, and where a message that
+// handler that wants STATE messages is full. The STATE message must be held
-// cannot wait is dropped. The streamer must not drop it: it waits until the
+// for that handler, and the read loop must go on to queue the UPDATE for the
-// handler takes a message off its queue, then queues it.
+// handler that wants it.
-func TestStreamWaitsForRoomForAStateMessage(t *testing.T) {
+func TestOtherHandlersKeepReceivingWhileAStateMessageIsHeld(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)
+		_, _ = io.WriteString(w, stateDownLine+updateLine)
 	}))
 	defer srv.Close()
 
-	queue, streamErr := streamWithFullStateQueue(srv.URL, streamIdleTimeout)
+	s := newStreamerWithFullStateQueue(srv.URL)
+	s.RegisterHandler(updateHandler{})
 
-	select {
+	if err := s.stream(context.Background()); !errors.Is(err, errStreamEnded) {
-	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
+	held := s.handlers[0].heldState
-	for len(queue) > 0 {
+	if len(held) != 1 || held[0].Peer != stateDownPeer || held[0].State != ristypes.StateDown {
-		last = <-queue
+		t.Errorf("held %d STATE messages, want the down from %s alone", len(held), stateDownPeer)
 	}
-	if last.Peer != stateDownPeer || last.State != ristypes.StateDown {
+	if got := len(s.handlers[1].queue); got != 1 {
-		t.Errorf("last queued message has peer %q, state %q; want %q, %q",
+		t.Errorf("%d messages queued for the handler that wants UPDATE messages, want 1", got)
-			last.Peer, last.State, stateDownPeer, ristypes.StateDown)
 	}
 }
 
-// TestStreamStopsWaitingForRoomAtTheIdleTimeout has the server send one STATE
+// TestStopReturnsAtOnceWithStateMessagesHeld has the server send two STATE
-// message and then hold the response open, while the queue of the handler
+// messages and hold the response open while the queue of the handler that
-// that wants it stays full. The wait for room must not stall the read loop
+// wants them is full. Stop must return at once and end the stream.
-// past its idle timeout: the stream ends as when no line arrives.
+func TestStopReturnsAtOnceWithStateMessagesHeld(t *testing.T) {
-func TestStreamStopsWaitingForRoomAtTheIdleTimeout(t *testing.T) {
 	const (
-		idleTimeout = 50 * time.Millisecond
+		lines   = 2
 		maxWait = 5 * time.Second
 	)
 	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-		_, _ = io.WriteString(w, stateDownLine)
+		_, _ = io.WriteString(w, stateDownLine+stateDownLine)
 		w.(http.Flusher).Flush()
 		<-r.Context().Done()
 	}))
 	defer srv.Close()
 
-	_, streamErr := streamWithFullStateQueue(srv.URL, idleTimeout)
+	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 err == nil || errors.Is(err, errStreamEnded) {
+		if !errors.Is(err, context.Canceled) {
-			t.Errorf("stream returned %v, want the error of a stream dropped as idle", err)
+			t.Errorf("stream returned %v, want %v", err, context.Canceled)
 		}
 	case <-time.After(maxWait):
-		t.Fatalf("stream still waiting for room %s after the idle timeout of %s", maxWait, idleTimeout)
+		t.Fatalf("stream did not end within %s of Stop", maxWait)
 	}
 }