routewatch/internal/routewatch/prefixhandler_table_internal_test.go

package routewatch

import (
	"bytes"
	"encoding/json"
	"errors"
	"log/slog"
	"net/netip"
	"reflect"
	"slices"
	"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"
)

// tableWriteTimeout is how long a test waits for the flush timer to write the
// routing table's changes: a few ticks of it.
const tableWriteTimeout = 5 * prefixBatchTimeout

// 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: {},
			},
		},
	})
}

// TestPrefixHandlerAppliesAPathWithoutAnOriginToTheRoutingTable hands the
// prefix handler announcements whose AS path gives no origin. One whose path
// 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: "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}}},
		},
	})
}

// checkTableCases hands the messages of each case to a new prefix handler and
// checks its routing table right after, 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)
			}

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

// 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(tableWriteTimeout):
			t.Fatalf("the routing table's change was not written within %s",
				tableWriteTimeout)
		}
	}
}

// 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.
func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
	t.Parallel()

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

	db := newTestDatabase(t)
	store := &tableWriteSizesStore{Database: db}
	h := prefixHandlerWithoutTimer(store)
	announceManyPrefixes(h.table, pairs)

	for range flushes {
		h.flushBatch()
	}

	wantChanges := []int{maxTableChangesPerWrite, pairs - maxTableChangesPerWrite}
	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 two
// writes take. Its last flush writes one write's worth, so Stop must log the
// pairs that no write took.
func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
	t.Parallel()

	// More than two writes take, so that pairs are left even when the flush
	// timer writes once before Stop.
	const pairs = 2*maxTableChangesPerWrite + 1

	var logs bytes.Buffer

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

	h := NewPrefixHandler(store, newErrorLogger(&logs), table)
	h.Stop(t.Context())

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

	if logged := lastLogged(t, &logs); 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.flushBatch()

	var logs bytes.Buffer

	rw := &RouteWatch{
		db:     db,
		logger: &logger.Logger{Logger: slog.New(slog.NewJSONHandler(&logs, nil))},
	}

	table, err := rw.readRouteTable(t.Context())
	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)
	}

	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)
	}
}