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

README.md

@@ -235,9 +235,7 @@ 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, except a
+  When a queue fills, the streamer drops messages instead of blocking.
-  `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 incoming UPDATE messages and maintains peer state in the database.
+// It tracks BGP peers from all incoming RIS 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 rather than blocks
+	// about 4 seconds of feed at peak. The streamer drops messages rather than
-	// when a queue is full, so this bounds memory. Batches still flush on a
+	// blocks when a queue is full (a STATE message waits for room), so this
-	// timer (peerBatchTimeout).
+	// bounds memory. Batches still flush on a 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 UPDATE messages using batched operations.
+// PeerHandler tracks BGP peers from all message types 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 UPDATE messages only. A peer whose session
+// WantsMessage returns true for all message types since peer information
-// never comes up sends STATE messages and nothing else, so a STATE message is
+// is extracted from every RIS message regardless of type. This satisfies
-// not a sign of a peer that carries routes.
+// the MessageHandler interface.
-func (h *PeerHandler) WantsMessage(messageType string) bool {
+func (h *PeerHandler) WantsMessage(_ string) bool {
-	return messageType == ristypes.MessageTypeUpdate
+	return true
 }
 
 // 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 is held until
+	// rather than blocks when a queue is full (a STATE message waits for
-	// there is room), so this bounds memory. Batches still flush on a timer
+	// room), so this bounds memory. Batches still flush on a timer
 	// (prefixBatchTimeout).
 	prefixHandlerQueueSize = 20000
 

internal/routewatch/prefixhandler_test.go

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

internal/streamer/streamer_test.go

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