routewatch/internal/routewatch/prefixhandler_table_internal_test.go

package routewatch

import (
	"bytes"
	"encoding/json"
	"errors"
	"log/slog"
	"net/netip"
	"reflect"
	"runtime"
	"slices"
	"strings"
	"sync"
	"testing"
	"time"

	"git.eeqj.de/sneak/routewatch/internal/database"
	"git.eeqj.de/sneak/routewatch/internal/logger"
	"git.eeqj.de/sneak/routewatch/internal/ristypes"
	"git.eeqj.de/sneak/routewatch/internal/routetable"
)

// More peers and prefixes for the tests of the routing table in memory.
const (
	secondPeerIP = "2001:db8:ffff::2"
	thirdPeerIP  = "2001:db8:ffff::3"
	fourthPeerIP = "192.0.2.4"
	fifthPeerIP  = "192.0.2.5"

	otherV4Prefix = "198.51.100.0/24"
	setV4Prefix   = "203.0.113.0/24"

	// firstManyPrefix is the first prefix announceManyPrefixes announces.
	firstManyPrefix = "10.0.0.0/32"

	// testMaxTableChangesPerWrite is a limit of one write of the routing
	// table's changes that a test exceeds quickly, in place of
	// maxTableChangesPerWrite.
	testMaxTableChangesPerWrite = 10
)

// tableCase is one test of the routing table in memory: the messages handed to
// the prefix handler, and for each prefix the peers that carry it with each
// origin AS after them.
type tableCase struct {
	name     string
	messages []*ristypes.RISMessage
	want     map[string]map[int][]string
}

// TestPrefixHandlerAppliesUpdatesToTheRoutingTable hands the prefix handler
// announcements, withdrawals and a peer's session going down.
func TestPrefixHandlerAppliesUpdatesToTheRoutingTable(t *testing.T) {
	t.Parallel()

	announceAndWithdraw := peerAnnouncement(testPeerIP, testV4Prefix)
	announceAndWithdraw.Withdrawals = []string{testV4Prefix}

	checkTableCases(t, []tableCase{
		{
			name:     "an announcement",
			messages: []*ristypes.RISMessage{peerAnnouncement(testPeerIP, testV4Prefix)},
			want:     map[string]map[int][]string{testV4Prefix: {testASNB: {testPeerIP}}},
		},
		{
			name: "a withdrawal",
			messages: []*ristypes.RISMessage{
				peerAnnouncement(testPeerIP, testV4Prefix),
				peerWithdrawal(testPeerIP, testV4Prefix),
			},
			want: map[string]map[int][]string{testV4Prefix: {}},
		},
		{
			// RFC 4271 section 4.3: the announcement counts, not the withdrawal.
			name:     "a message that announces and withdraws the prefix",
			messages: []*ristypes.RISMessage{announceAndWithdraw},
			want:     map[string]map[int][]string{testV4Prefix: {testASNB: {testPeerIP}}},
		},
		{
			name: "a peer's session going down",
			messages: []*ristypes.RISMessage{
				peerAnnouncement(testPeerIP, testV4Prefix, testV6Prefix),
				peerAnnouncement(secondPeerIP, testV4Prefix),
				sessionDown(),
			},
			want: map[string]map[int][]string{
				testV4Prefix: {testASNB: {secondPeerIP}},
				testV6Prefix: {},
			},
		},
		{
			name: "an announcement after the peer's session went down",
			messages: []*ristypes.RISMessage{
				peerAnnouncement(testPeerIP, testV4Prefix),
				sessionDown(),
				peerAnnouncement(testPeerIP, otherV4Prefix),
			},
			want: map[string]map[int][]string{
				testV4Prefix:  {},
				otherV4Prefix: {testASNB: {testPeerIP}},
			},
		},
	})
}

// TestPrefixHandlerTakesMessagesWhileAPeerGoesDown holds up the PeerDown of a
// peer whose session went down, as another peer's long PeerDown would. The
// handler must go on taking messages. It applies another peer's announcement
// at once. It holds back these messages of the down peer, in this order:
//   - an announcement;
//   - its session going down again;
//   - a second announcement.
//
// Once the PeerDowns have returned, the down peer must carry only the prefix
// of the second announcement.
func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
	t.Parallel()

	h := prefixHandlerWithoutTimer(newMockStore())
	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))

	h.peerDownMu.Lock()

	handled := returnsInTime(func() {
		for _, msg := range []*ristypes.RISMessage{
			sessionDown(),
			peerAnnouncement(secondPeerIP, otherV4Prefix),
			peerAnnouncement(testPeerIP, testV6Prefix),
			sessionDown(),
			peerAnnouncement(testPeerIP, setV4Prefix),
		} {
			h.HandleMessage(msg)
		}
	})
	if !handled {
		t.Fatalf("the handler did not take the messages after the session "+
			"went down within %s", stepTimeout)
	}

	checkTableOrigins(t, h.table, map[string]map[int][]string{
		testV4Prefix:  {testASNB: {testPeerIP}},
		otherV4Prefix: {testASNB: {secondPeerIP}},
		testV6Prefix:  {},
		setV4Prefix:   {},
	})

	h.peerDownMu.Unlock()

	if !returnsInTime(h.peerDowns.Wait) {
		t.Fatalf("the PeerDowns did not return within %s", stepTimeout)
	}

	checkTableOrigins(t, h.table, map[string]map[int][]string{
		testV4Prefix:  {},
		otherV4Prefix: {testASNB: {secondPeerIP}},
		testV6Prefix:  {},
		setV4Prefix:   {testASNB: {testPeerIP}},
	})
}

// TestPrefixHandlerAppliesMessagesWhilePeerDownRuns gives a peer 50,000
// prefixes and hands the prefix handler that peer's session going down. The
// table's PeerDown then visits the prefixes in five batches. Once PeerDown has
// begun, the handler is handed 1,000 announcements of another peer. The
// handler must apply them all before PeerDown ends, while the down peer still
// carries some of its prefixes.
func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) {
	t.Parallel()

	const (
		// downPeerPrefixes is five times the 10,000 prefixes PeerDown visits
		// in one hold of the table's lock.
		downPeerPrefixes = 50000
		// otherPeerMessages is far more messages than the handler applies
		// while PeerDown runs, if PeerDown takes the lock back at once after
		// each batch. The handler then applies only a few for each batch.
		otherPeerMessages = 1000
		// firstOtherAddr is the address of the other peer's first prefix.
		firstOtherAddr = "2001:db8:2::"
	)

	h := prefixHandlerWithoutTimer(newMockStore())
	announceManyPrefixes(h.table, downPeerPrefixes)

	h.HandleMessage(sessionDown())

	// Wait until PeerDown has begun: the down peer has left some of its
	// prefixes. The down peer's prefixes are IPv4 and the other peer's are
	// IPv6, so the counts tell them apart.
	for h.table.Counts().IPv4Routes == downPeerPrefixes {
		runtime.Gosched()
	}

	// Each announcement names the /128 of the next address.
	addr := netip.MustParseAddr(firstOtherAddr)
	for range otherPeerMessages {
		h.HandleMessage(peerAnnouncement(secondPeerIP,
			netip.PrefixFrom(addr, addr.BitLen()).String()))

		addr = addr.Next()
	}

	counts := h.table.Counts()
	if counts.IPv6Routes != otherPeerMessages {
		t.Errorf("the other peer carries %d prefixes, want %d",
			counts.IPv6Routes, otherPeerMessages)
	}

	if counts.IPv4Routes == 0 {
		t.Errorf("PeerDown ended before the handler applied the other peer's "+
			"%d announcements", otherPeerMessages)
	}

	if !returnsInTime(h.peerDowns.Wait) {
		t.Fatalf("the PeerDown did not return within %s", stepTimeout)
	}
}

// TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes holds back many
// messages of a peer whose session went down. Each names a prefix that does
// not parse, so applying it logs an error. Each log write takes a
// millisecond. While the handler applies the held-back messages, it must take
// another peer's announcement.
func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) {
	t.Parallel()

	const (
		// heldMessages take the handler at least a second to apply.
		heldMessages = 1000
		badPrefix    = "not a prefix"
	)

	logs := &slowWriter{firstWrite: make(chan struct{})}
	h := newPrefixHandler(newMockStore(), newErrorLogger(logs),
		routetable.New(routetable.Params{}), longerThanAnyTest,
		maxTableChangesPerWrite)

	defer stopInTime(t, h)

	h.peerDownMu.Lock()
	h.HandleMessage(sessionDown())

	for range heldMessages {
		h.HandleMessage(peerAnnouncement(testPeerIP, badPrefix))
	}

	h.peerDownMu.Unlock()

	select {
	case <-logs.firstWrite:
	case <-time.After(stepTimeout):
		t.Fatal("the handler did not start to apply the held-back messages")
	}

	if !returnsInTime(func() {
		h.HandleMessage(peerAnnouncement(secondPeerIP, otherV4Prefix))
	}) {
		t.Fatalf("HandleMessage did not return within %s", stepTimeout)
	}

	h.mu.Lock()
	left := len(h.heldBack[testPeerIP])
	h.mu.Unlock()

	if left == 0 {
		t.Error("the handler took another peer's announcement only once it " +
			"had applied every held-back message")
	}
}

// TestPrefixHandlerWaitsAtTheMaximumHeldBack holds up the PeerDown of a peer
// whose session went down. It hands the handler maxHeldBackMessages messages
// of that peer. HandleMessage must then wait, even for another peer's
// message, until the PeerDown returns and a held-back message is applied.
func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
	t.Parallel()

	h := newPrefixHandler(newMockStore(), logger.New(),
		routetable.New(routetable.Params{}), longerThanAnyTest,
		maxTableChangesPerWrite)
	defer stopInTime(t, h)

	h.peerDownMu.Lock()
	h.HandleMessage(sessionDown())

	held := peerAnnouncement(testPeerIP, testV6Prefix)
	for range maxHeldBackMessages {
		h.HandleMessage(held)
	}

	returned := make(chan struct{})

	go func() {
		h.HandleMessage(peerAnnouncement(secondPeerIP, otherV4Prefix))
		close(returned)
	}()

	select {
	case <-returned:
		t.Error("HandleMessage returned while maxHeldBackMessages messages " +
			"were held back")
	case <-time.After(stillWaitingAfter):
	}

	h.peerDownMu.Unlock()

	select {
	case <-returned:
	case <-time.After(stepTimeout):
		t.Fatal("HandleMessage did not return after the PeerDown returned")
	}
}

// slowWriter takes a millisecond over each write and drops what it is given.
// It closes firstWrite at the first write.
type slowWriter struct {
	firstWrite chan struct{}
	once       sync.Once
}

func (w *slowWriter) Write(p []byte) (int, error) {
	w.once.Do(func() { close(w.firstWrite) })
	time.Sleep(time.Millisecond)

	return len(p), nil
}

// TestPrefixHandlerAppliesAPathWithoutAnOriginToTheRoutingTable hands the
// prefix handler announcements whose AS path gives no origin. One whose path
// is empty or holds AS 0 removes the peer's route, as a withdrawal does; one
// whose path ends in an AS set of several members is kept with origin 0.
func TestPrefixHandlerAppliesAPathWithoutAnOriginToTheRoutingTable(t *testing.T) {
	t.Parallel()

	announceWithPath := func(path ristypes.ASPath) *ristypes.RISMessage {
		msg := peerAnnouncement(testPeerIP, testV4Prefix)
		msg.Path = path

		return msg
	}

	checkTableCases(t, []tableCase{
		{
			name: "a path whose last AS is 0",
			messages: []*ristypes.RISMessage{
				peerAnnouncement(testPeerIP, testV4Prefix),
				announceWithPath(ristypes.ASPath{Sequence: []int{testASNA, 0}}),
			},
			want: map[string]map[int][]string{testV4Prefix: {}},
		},
		{
			name: "an empty path",
			messages: []*ristypes.RISMessage{
				peerAnnouncement(testPeerIP, testV4Prefix),
				announceWithPath(ristypes.ASPath{}),
			},
			want: map[string]map[int][]string{testV4Prefix: {}},
		},
		{
			name: "a path with AS 0 before its end",
			messages: []*ristypes.RISMessage{
				peerAnnouncement(testPeerIP, testV4Prefix),
				announceWithPath(decodePath(t, testASNA, 0, testASNB)),
			},
			want: map[string]map[int][]string{testV4Prefix: {}},
		},
		{
			name: "a path that ends in an AS set",
			messages: []*ristypes.RISMessage{
				announceWithPath(decodePath(t, testASNA, []int{testASNC, testASND})),
			},
			want: map[string]map[int][]string{testV4Prefix: {0: {testPeerIP}}},
		},
	})
}

// TestPrefixHandlerKeepsThePathOfAPathEndingInASet announces a prefix with AS
// paths that end in an AS set. Any member of a set of several may be the
// origin, so the routing table keeps the route with origin 0 (none); a set of
// one member counts as that AS. The kept AS path lists the set's members after
// the rest of the path.
func TestPrefixHandlerKeepsThePathOfAPathEndingInASet(t *testing.T) {
	t.Parallel()

	tests := []struct {
		name       string
		path       []any
		wantOrigin int
		wantPath   []int
	}{
		{
			name:     "a set of several members",
			path:     []any{testASNA, testASNB, []int{testASNC, testASND}},
			wantPath: []int{testASNA, testASNB, testASNC, testASND},
		},
		{
			name:       "a set of one member",
			path:       []any{testASNA, testASNB, []int{testASNC}},
			wantOrigin: testASNC,
			wantPath:   []int{testASNA, testASNB, testASNC},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel()

			msg := peerAnnouncement(testPeerIP, testV4Prefix)
			msg.Path = decodePath(t, tt.path...)

			h := prefixHandlerWithoutTimer(newMockStore())
			h.HandleMessage(msg)

			origins := h.table.Origins(netip.MustParsePrefix(testV4Prefix))
			if len(origins) != 1 || origins[0].ASN != tt.wantOrigin ||
				len(origins[0].Routes) != 1 ||
				!slices.Equal(origins[0].Routes[0].ASPath, tt.wantPath) {
				t.Errorf("Origins(%s) = %+v, want one route with origin AS%d "+
					"and AS path %v", testV4Prefix, origins, tt.wantOrigin,
					tt.wantPath)
			}
		})
	}
}

// checkTableCases hands the messages of each case to a new prefix handler. It
// checks the handler's routing table once the handler's PeerDowns have
// returned, with no flush.
func checkTableCases(t *testing.T, tests []tableCase) {
	t.Helper()

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			t.Parallel()

			h := prefixHandlerWithoutTimer(newMockStore())
			for _, msg := range tt.messages {
				h.HandleMessage(msg)
			}

			h.peerDowns.Wait()
			checkTableOrigins(t, h.table, tt.want)
		})
	}
}

// checkTableOrigins checks, for each prefix in want, the peers that carry it
// in table with each origin AS.
func checkTableOrigins(
	t *testing.T, table *routetable.Table, want map[string]map[int][]string,
) {
	t.Helper()

	for prefix, wantOrigins := range want {
		got := tableOrigins(table, prefix)
		if !reflect.DeepEqual(got, wantOrigins) {
			t.Errorf("peers of %s by origin AS = %v, want %v",
				prefix, got, wantOrigins)
		}
	}
}

// tableWritesStore is a real database that hands each write of the routing
// table's changes to written once it is made.
type tableWritesStore struct {
	*database.Database

	written chan []routetable.Change
}

func (s tableWritesStore) WriteRouteChanges(
	changes []routetable.Change, newPeers map[string]uint,
) error {
	err := s.Database.WriteRouteChanges(changes, newPeers)
	s.written <- changes

	return err
}

// TestPrefixHandlerWritesTheTableChangesEachSecond hands a running prefix
// handler an announcement and then its withdrawal, each far short of a full
// batch. The flush timer must write the change of each to the database.
func TestPrefixHandlerWritesTheTableChangesEachSecond(t *testing.T) {
	t.Parallel()

	// The two writes the test waits for, and room for one more.
	const writesBuffered = 3

	store := tableWritesStore{
		Database: newTestDatabase(t),
		written:  make(chan []routetable.Change, writesBuffered),
	}
	h := NewPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}))

	defer h.Stop(t.Context())

	prefix := netip.MustParsePrefix(testV4Prefix)

	for _, step := range []struct {
		msg         *ristypes.RISMessage
		wantRemoved bool
	}{
		{msg: peerAnnouncement(testPeerIP, testV4Prefix)},
		{msg: peerWithdrawal(testPeerIP, testV4Prefix), wantRemoved: true},
	} {
		h.HandleMessage(step.msg)

		select {
		case changes := <-store.written:
			if len(changes) != 1 || changes[0].Prefix != prefix ||
				changes[0].OriginASN != testASNB ||
				changes[0].Removed != step.wantRemoved {
				t.Fatalf("wrote %+v, want the pair of %s and AS%d with "+
					"removed %t", changes, prefix, testASNB, step.wantRemoved)
			}
		case <-time.After(stepTimeout):
			t.Fatalf("the routing table's change was not written within %s",
				stepTimeout)
		}
	}
}

// errTestTableWrite is the error a test store's failing write of the routing
// table's changes returns.
var errTestTableWrite = errors.New("test: the write of the table's changes failed")

// firstTableWriteFailsStore is a real database whose first write of the
// routing table's changes fails, writing nothing. writes counts the writes
// asked for.
type firstTableWriteFailsStore struct {
	*database.Database

	writes int
}

func (s *firstTableWriteFailsStore) WriteRouteChanges(
	changes []routetable.Change, newPeers map[string]uint,
) error {
	s.writes++
	if s.writes == 1 {
		return errTestTableWrite
	}

	return s.Database.WriteRouteChanges(changes, newPeers)
}

// TestPrefixHandlerWritesAFailedTableWriteAgain runs two flushes into a
// database whose first write of the routing table's changes fails. Before the
// first, two new peers announce a prefix each; before the second, one of them
// announces its prefix again with another path to the same origin AS. The
// second write must hold the changes and new peers of the first, with the later
// path, so the database holds the table that is in memory.
func TestPrefixHandlerWritesAFailedTableWriteAgain(t *testing.T) {
	t.Parallel()

	db := newTestDatabase(t)
	store := &firstTableWriteFailsStore{Database: db}
	h := prefixHandlerWithoutTimer(store)

	first := peerAnnouncement(testPeerIP, testV4Prefix)
	h.HandleMessage(first)
	h.HandleMessage(peerAnnouncement(secondPeerIP, otherV4Prefix))
	h.flushBatch()

	laterPath := []int{testASNC, testASNB}
	later := peerAnnouncement(testPeerIP, testV4Prefix)
	later.Path = ristypes.ASPath{Sequence: laterPath}
	later.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter)
	h.HandleMessage(later)
	h.flushBatch()

	if store.writes != 2 {
		t.Fatalf("the store was asked for %d writes of the table's changes, "+
			"want 2", store.writes)
	}

	read, err := db.ReadRouteTable(t.Context())
	if err != nil {
		t.Fatalf("ReadRouteTable: %v", err)
	}

	checkSameTable(t, read, h.table, testV4Prefix, otherV4Prefix)

	origins := read.Origins(netip.MustParsePrefix(testV4Prefix))
	if len(origins) != 1 || len(origins[0].Routes) != 1 ||
		!slices.Equal(origins[0].Routes[0].ASPath, laterPath) {
		t.Errorf("Origins(%s) = %+v, want one route with the path %v",
			testV4Prefix, origins, laterPath)
	}
}

// tableWriteSizesStore is a real database that records how many changes and
// how many new peers each write of the routing table's changes holds.
type tableWriteSizesStore struct {
	*database.Database

	changes  []int
	newPeers []int
}

func (s *tableWriteSizesStore) WriteRouteChanges(
	changes []routetable.Change, newPeers map[string]uint,
) error {
	s.changes = append(s.changes, len(changes))
	s.newPeers = append(s.newPeers, len(newPeers))

	return s.Database.WriteRouteChanges(changes, newPeers)
}

// TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes has one new peer
// change one and a half times as many prefix and origin AS pairs as one write
// takes, then flushes three times. The first write must hold as many pairs as
// one write takes, and the peer. The second must hold the other pairs, and the
// third flush must write nothing. The database must then hold the table in
// memory. The handler's limit is small, and its timer does not fire.
func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
	t.Parallel()

	const (
		pairs   = testMaxTableChangesPerWrite * 3 / 2
		flushes = 3
	)

	db := newTestDatabase(t)
	store := &tableWriteSizesStore{Database: db}
	table := routetable.New(routetable.Params{})
	announceManyPrefixes(table, pairs)

	h := newPrefixHandler(store, logger.New(), table, longerThanAnyTest,
		testMaxTableChangesPerWrite)

	defer stopInTime(t, h)

	for range flushes {
		h.flushBatch()
	}

	wantChanges := []int{
		testMaxTableChangesPerWrite, pairs - testMaxTableChangesPerWrite,
	}
	wantNewPeers := []int{1, 0}

	if !slices.Equal(store.changes, wantChanges) ||
		!slices.Equal(store.newPeers, wantNewPeers) {
		t.Errorf("the writes held %v pairs and %v new peers, want %v and %v",
			store.changes, store.newPeers, wantChanges, wantNewPeers)
	}

	read, err := db.ReadRouteTable(t.Context())
	if err != nil {
		t.Fatalf("ReadRouteTable: %v", err)
	}

	checkSameTable(t, read, h.table, firstManyPrefix)
}

// TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten stops a prefix handler,
// with no deadline, whose routing table holds more changed pairs than one write
// takes. Its last flush writes one write's worth, so Stop must log the pairs
// that no write took, and must not say that the stop deadline passed. The
// flush timer must not fire, or its writes could leave no pairs for Stop to
// log. The handler's limit of one write is small.
func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
	t.Parallel()

	const pairs = testMaxTableChangesPerWrite + 1

	var logs bytes.Buffer

	store := &tableWriteSizesStore{Database: newTestDatabase(t)}
	table := routetable.New(routetable.Params{})
	announceManyPrefixes(table, pairs)

	h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest,
		testMaxTableChangesPerWrite)
	h.Stop(t.Context())

	written := 0
	for _, n := range store.changes {
		written += n
	}

	logged := lastLogged(t, &logs)
	if strings.Contains(logged.Msg, "deadline") {
		t.Errorf("Stop logged %q although its deadline had not passed",
			logged.Msg)
	}

	if logged.TableChanges != pairs-written {
		t.Errorf("Stop logged %d table changes not written, want %d",
			logged.TableChanges, pairs-written)
	}
}

// readLog holds what readRouteTable logs.
//
//nolint:tagliatelle // field names are the keys readRouteTable logs
type readLog struct {
	Msg        string `json:"msg"`
	Pairs      int    `json:"pairs"`
	Peers      int    `json:"peers"`
	DurationMs *int64 `json:"duration_ms"`
}

// TestARestartReadsTheRoutingTableBack runs announcements, a withdrawal and a
// peer's session going down through a prefix handler into a database, then
// reads the table back as startup does. The table read must answer as the one
// in memory, and the log must give its pairs and peers. A new handler on the
// table read then takes later updates and writes them as it stops: the
// database must then hold what the first table holds after the same updates.
func TestARestartReadsTheRoutingTableBack(t *testing.T) {
	t.Parallel()

	prefixes := []string{testV4Prefix, testV6Prefix, otherV4Prefix, setV4Prefix}

	db := newTestDatabase(t)
	h := prefixHandlerWithoutTimer(db)

	viaOtherOrigin := peerAnnouncement(thirdPeerIP, testV4Prefix)
	viaOtherOrigin.Path = ristypes.ASPath{Sequence: []int{testASNA, testASNC}}
	endsInSet := peerAnnouncement(secondPeerIP, setV4Prefix)
	endsInSet.Path = decodePath(t, testASNA, []int{testASNC, testASND})

	for _, msg := range []*ristypes.RISMessage{
		peerAnnouncement(testPeerIP, testV4Prefix, testV6Prefix),
		peerAnnouncement(secondPeerIP, testV4Prefix),
		viaOtherOrigin,
		endsInSet,
		peerAnnouncement(thirdPeerIP, otherV4Prefix),
		peerWithdrawal(thirdPeerIP, otherV4Prefix),
		peerAnnouncement(fourthPeerIP, otherV4Prefix),
		sessionDownOf(fourthPeerIP),
	} {
		h.HandleMessage(msg)
	}

	h.peerDowns.Wait()
	h.flushBatch()

	var logs bytes.Buffer

	table, err := readRouteTable(db,
		&logger.Logger{Logger: slog.New(slog.NewJSONHandler(&logs, nil))})
	if err != nil {
		t.Fatalf("readRouteTable: %v", err)
	}

	checkSameTable(t, table, h.table, prefixes...)

	// testV4Prefix with two origin ASes, testV6Prefix and setV4Prefix; four
	// peers, the last of which carries nothing.
	const wantPairs, wantPeers = 4, 4

	var logged readLog

	err = json.Unmarshal(bytes.TrimSpace(logs.Bytes()), &logged)
	if err != nil || logged.Pairs != wantPairs || logged.Peers != wantPeers ||
		logged.DurationMs == nil {
		t.Errorf("readRouteTable logged %q, want %d pairs, %d peers and the "+
			"duration", logs.String(), wantPairs, wantPeers)
	}

	restarted := NewPrefixHandler(db, logger.New(), table)

	for _, msg := range []*ristypes.RISMessage{
		peerWithdrawal(testPeerIP, testV6Prefix),
		sessionDownOf(secondPeerIP),
		peerAnnouncement(fourthPeerIP, otherV4Prefix),
		peerAnnouncement(fifthPeerIP, setV4Prefix),
	} {
		// The first handler only applies the updates to its table.
		h.HandleMessage(msg)
		restarted.HandleMessage(msg)
	}

	h.peerDowns.Wait()
	restarted.Stop(t.Context())

	read, err := db.ReadRouteTable(t.Context())
	if err != nil {
		t.Fatalf("ReadRouteTable after the restart: %v", err)
	}

	checkSameTable(t, read, h.table, prefixes...)
}

// peerWithdrawal returns an UPDATE message in which peer withdraws prefixes.
func peerWithdrawal(peer string, prefixes ...string) *ristypes.RISMessage {
	return &ristypes.RISMessage{
		Type:            ristypes.MessageTypeUpdate,
		Peer:            peer,
		ParsedTimestamp: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC),
		Withdrawals:     prefixes,
	}
}

// announceManyPrefixes has testPeerIP announce n prefixes in table, with the
// path of peerAnnouncement: the /32 of each address from firstManyPrefix on.
func announceManyPrefixes(table *routetable.Table, n int) {
	addr := netip.MustParsePrefix(firstManyPrefix).Addr()
	route := routetable.Route{
		Peer:   testPeerIP,
		ASPath: []int{testASNA, testASNB},
		Time:   time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC),
	}

	for range n {
		table.Announce(netip.PrefixFrom(addr, addr.BitLen()), testASNB, route)
		addr = addr.Next()
	}
}

// sessionDownOf returns the STATE message that says the BGP session of peer
// went down.
func sessionDownOf(peer string) *ristypes.RISMessage {
	return &ristypes.RISMessage{
		Type:  ristypes.MessageTypeState,
		Peer:  peer,
		State: ristypes.StateDown,
	}
}

// tableOrigins returns, for each origin AS of prefix in table, the peers that
// carry the prefix with it.
func tableOrigins(table *routetable.Table, prefix string) map[int][]string {
	origins := make(map[int][]string)
	for _, o := range table.Origins(netip.MustParsePrefix(prefix)) {
		origins[o.ASN] = o.Peers
	}

	return origins
}

// checkSameTable checks that got answers as want does for each of prefixes, and
// that the two tables have the same counts.
func checkSameTable(
	t *testing.T, got, want *routetable.Table, prefixes ...string,
) {
	t.Helper()

	for _, text := range prefixes {
		prefix := netip.MustParsePrefix(text)

		gotOrigins, wantOrigins := got.Origins(prefix), want.Origins(prefix)
		if !reflect.DeepEqual(gotOrigins, wantOrigins) {
			t.Errorf("Origins(%s) = %+v, want %+v", text, gotOrigins, wantOrigins)
		}
	}

	gotCounts, wantCounts := got.Counts(), want.Counts()
	if !reflect.DeepEqual(gotCounts, wantCounts) {
		t.Errorf("Counts() = %+v, want %+v", gotCounts, wantCounts)
	}
}