routewatch/internal/routewatch/prefixhandler_internal_test.go

package routewatch

import (
	"bytes"
	"context"
	"net/netip"
	"testing"
	"time"

	"git.eeqj.de/sneak/routewatch/internal/config"
	"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"
)

const testPeerIP = "2001:db8:ffff::1"

// An IPv4 and an IPv6 prefix the tests announce.
const (
	testV4Prefix = "192.0.2.0/24"
	testV6Prefix = "2001:db8:1::/48"
)

// secondUpdateAfter is how long after the first update the tests of first_seen
// and last_seen send the second, in the same batch.
const secondUpdateAfter = 5 * time.Second

// TestPrefixHandlerStoresPrefixesTheIPLookupFinds runs announcements and
// withdrawals through the prefix handler into a real database. The prefixes
// are written the way the feed sends them: IPv6 announcements compressed, IPv6
// withdrawals uncompressed. The IP lookup must find each announced prefix in
// the routing table, and its first-seen time in the prefix tables. The
// withdrawal must remove the peer's route and store no second form of the
// prefix.
func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) {
	t.Parallel()

	// The two prefixes the announcement names.
	const wantPrefixes = 2

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

	ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
	h.HandleMessage(&ristypes.RISMessage{
		Peer:            testPeerIP,
		Path:            ristypes.ASPath{Sequence: []int{testASNA, testASNB}},
		ParsedTimestamp: ts,
		Announcements: []ristypes.RISAnnouncement{{
			NextHop:  testPeerIP,
			Prefixes: []string{testV6Prefix, testV4Prefix},
		}},
	})
	h.flushBatch()

	for ip, want := range map[string]string{
		"2001:db8:1::1": testV6Prefix,
		"192.0.2.1":     testV4Prefix,
	} {
		summary, found := h.table.Lookup(netip.MustParseAddr(ip))
		if !found || summary.Prefix.String() != want ||
			summary.OriginASN != testASNB {
			t.Errorf("Lookup(%s) after announcement = %+v, %t; want %s AS%d",
				ip, summary, found, want, testASNB)
		}

		firstSeen, err := db.GetPrefixFirstSeen(t.Context(),
			netip.MustParsePrefix(want))
		if err != nil || !firstSeen.Equal(ts) {
			t.Errorf("GetPrefixFirstSeen(%s) = %v, %v; want %v",
				want, firstSeen, err, ts)
		}
	}

	h.HandleMessage(&ristypes.RISMessage{
		Peer:            testPeerIP,
		ParsedTimestamp: ts.Add(time.Minute),
		Withdrawals:     []string{"2001:db8:1:0:0:0:0:0/48", testV4Prefix},
	})
	h.flushBatch()

	for _, ip := range []string{"2001:db8:1::1", "192.0.2.1"} {
		summary, found := h.table.Lookup(netip.MustParseAddr(ip))
		if found {
			t.Errorf("Lookup(%s) after withdrawal = %+v, want none", ip, summary)
		}
	}

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

	if stats.Prefixes != wantPrefixes {
		t.Errorf("the prefix tables hold %d prefixes after the withdrawal, "+
			"want %d", stats.Prefixes, wantPrefixes)
	}
}

// TestPrefixHandlerStoresANewPrefixFirstSeenAtItsEarliestUpdate announces a
// prefix the database does not hold yet twice in one batch, the second time 5
// seconds after the first, through the prefix handler into a real database. The
// prefix is stored first seen at the time of the first announcement, whether
// the two come from two peers or from one.
func TestPrefixHandlerStoresANewPrefixFirstSeenAtItsEarliestUpdate(
	t *testing.T,
) {
	t.Parallel()

	const otherPeerIP = "2001:db8:ffff::2"

	tests := []struct {
		name                  string
		firstPeer, secondPeer string
	}{
		{name: "two peers", firstPeer: testPeerIP, secondPeer: otherPeerIP},
		{name: "one peer", firstPeer: testPeerIP, secondPeer: testPeerIP},
	}

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

			first := peerAnnouncement(tt.firstPeer, testV4Prefix)
			second := peerAnnouncement(tt.secondPeer, testV4Prefix)
			second.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter)
			db := handleBatches(t, [][]*ristypes.RISMessage{{first, second}})

			firstSeen, err := db.GetPrefixFirstSeen(t.Context(),
				netip.MustParsePrefix(testV4Prefix))
			if err != nil {
				t.Fatalf("GetPrefixFirstSeen(%s): %v", testV4Prefix, err)
			}

			if !firstSeen.Equal(first.ParsedTimestamp) {
				t.Errorf("first_seen of %s = %s, want %s", testV4Prefix,
					firstSeen, first.ParsedTimestamp)
			}
		})
	}
}

// TestPrefixHandlerDoesNotStoreAPrefixThatDoesNotParse runs an announcement of
// "not-a-prefix" and a withdrawal of "also/bad" through the prefix handler into
// a real database. Neither is stored in the prefix tables: the prefix count in
// the stats stays 0, both as kept in memory and as counted from the tables when
// the database is opened again.
func TestPrefixHandlerDoesNotStoreAPrefixThatDoesNotParse(t *testing.T) {
	t.Parallel()

	cfg := &config.Config{StateDir: t.TempDir()}

	db, err := database.New(cfg, logger.New())
	if err != nil {
		t.Fatalf("failed to create database: %v", err)
	}

	h := prefixHandlerWithoutTimer(db)

	withdrawal := &ristypes.RISMessage{
		Type:            ristypes.MessageTypeUpdate,
		Peer:            testPeerIP,
		ParsedTimestamp: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC),
		Withdrawals:     []string{"also/bad"},
	}
	h.HandleMessage(peerAnnouncement(testPeerIP, "not-a-prefix"))
	h.HandleMessage(withdrawal)
	h.flushBatch()

	checkNoPrefixes := func(when string) {
		t.Helper()

		stats, err := db.GetStatsContext(context.Background())
		if err != nil {
			t.Fatalf("GetStatsContext %s: %v", when, err)
		}

		if stats.Prefixes != 0 {
			t.Errorf("prefix count %s = %d, want 0", when, stats.Prefixes)
		}
	}

	checkNoPrefixes("after the flush")

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

	db, err = database.New(cfg, logger.New())
	if err != nil {
		t.Fatalf("failed to open the database again: %v", err)
	}

	t.Cleanup(func() { _ = db.Close() })

	checkNoPrefixes("after opening the database again")
}

// 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) {
	t.Parallel()

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

// TestPrefixHandlerStopLogsNothingWhenItWroteEverything stops a prefix
// handler, with no deadline, after a peer announced a prefix and its session
// went down. Its last flush writes the routing table's changes and the prefix,
// so Stop must log nothing.
func TestPrefixHandlerStopLogsNothingWhenItWroteEverything(t *testing.T) {
	t.Parallel()

	var logs bytes.Buffer

	h := NewPrefixHandler(newTestDatabase(t), newErrorLogger(&logs),
		routetable.New(routetable.Params{}))
	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
	h.HandleMessage(sessionDown())
	h.Stop(t.Context())

	if logs.Len() != 0 {
		t.Errorf("Stop logged %q after it wrote everything", logs.String())
	}
}

// TestPrefixHandlerStopLogsTheMessagesHeldBack stops a prefix handler with a
// deadline that has passed, while a peer's PeersDown is held up and one message
// of the peer is held back. Stop must log that message as not applied.
func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
	t.Parallel()

	var logs bytes.Buffer

	h := newPrefixHandler(newMockStore(), newErrorLogger(&logs),
		routetable.New(routetable.Params{}), longerThanAnyTest,
		maxTableChangesPerWrite)

	release := holdUpPeersDown(h)
	defer close(release)

	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
	h.HandleMessage(sessionDown())
	h.HandleMessage(peerAnnouncement(testPeerIP, testV6Prefix))

	ctx, cancel := context.WithCancel(t.Context())
	cancel()
	h.Stop(ctx)

	logged := lastLogged(t, &logs)
	if logged.HeldMessages != 1 {
		t.Errorf("Stop logged %d held messages not applied, want 1",
			logged.HeldMessages)
	}
}

// TestPrefixHandlerStopWaitsForTheMessagesHeldBack stops a prefix handler
// while a peer's PeersDown is held up and an announcement of the peer is held
// back. Stop must wait. Once the PeersDown is let go and Stop returns, the
// announcement must be in the database.
func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) {
	t.Parallel()

	db := newTestDatabase(t)
	h := newPrefixHandler(db, logger.New(), routetable.New(routetable.Params{}),
		longerThanAnyTest, maxTableChangesPerWrite)

	release := holdUpPeersDown(h)
	h.HandleMessage(sessionDown())
	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))

	stopped := make(chan struct{})

	go func() {
		h.Stop(t.Context())
		close(stopped)
	}()

	select {
	case <-stopped:
		t.Error("Stop returned while a PeersDown was held up")
	case <-time.After(stillWaitingAfter):
	}

	close(release)

	select {
	case <-stopped:
	case <-time.After(stepTimeout):
		t.Fatalf("Stop did not return within %s of the PeersDown's release",
			stepTimeout)
	}

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

	checkTableOrigins(t, read, map[string]map[int][]string{
		testV4Prefix: {testASNB: {testPeerIP}},
	})
}

// longerThanAnyTest is a timer interval that no test lasts. A test gives it to
// newPrefixHandler, newASHandler, newPeerHandler or newPeeringHandler when the
// handler must write only when its batch is full, when the test says, or on
// Stop.
const longerThanAnyTest = time.Hour

// prefixHandlerWithoutTimer returns a prefix handler that writes to store and
// applies the updates to an empty routing table. It is built without
// NewPrefixHandler's flush goroutine, so the test flushes itself.
func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
	table := routetable.New(routetable.Params{})

	return &PrefixHandler{
		db:                    store,
		logger:                logger.New(),
		table:                 table,
		batchMaxSize:          prefixBatchMaxSize,
		tableChangesPerWrite:  maxTableChangesPerWrite,
		originChangesPerWrite: maxOriginChangesPerWrite,
		heldBack:              make(map[string][]*ristypes.RISMessage),
		tablePeersDown:        table.PeersDown,
	}
}

// holdUpPeersDown makes each PeersDown of h wait until the returned channel is
// closed, as another peer's long PeersDown would hold it up. The test calls it
// before it hands h a message.
func holdUpPeersDown(h *PrefixHandler) chan struct{} {
	release := make(chan struct{})
	h.tablePeersDown = func(peers []string) {
		<-release
		h.table.PeersDown(peers)
	}

	return release
}

// newTestDatabase opens a database in a new temporary directory. It is closed
// when the test ends.
func newTestDatabase(t *testing.T) *database.Database {
	t.Helper()

	db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
	if err != nil {
		t.Fatalf("failed to create database: %v", err)
	}

	t.Cleanup(func() { _ = db.Close() })

	return db
}

// handleBatches runs each batch of messages through a prefix handler into a
// new database, flushing after each batch, and returns the database.
func handleBatches(
	t *testing.T, batches [][]*ristypes.RISMessage,
) *database.Database {
	t.Helper()

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

	for _, batch := range batches {
		for _, msg := range batch {
			h.HandleMessage(msg)
		}

		h.flushBatch()
	}

	return db
}

// peerAnnouncement returns an UPDATE message in which peer announces prefixes
// with origin AS testASNB.
func peerAnnouncement(peer string, prefixes ...string) *ristypes.RISMessage {
	return &ristypes.RISMessage{
		Type:            ristypes.MessageTypeUpdate,
		Peer:            peer,
		Path:            ristypes.ASPath{Sequence: []int{testASNA, testASNB}},
		ParsedTimestamp: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC),
		Announcements: []ristypes.RISAnnouncement{
			{NextHop: peer, Prefixes: prefixes},
		},
	}
}

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