routewatch/internal/routewatch/prefixhandler_originchanges_internal_test.go

package routewatch

import (
	"bytes"
	"encoding/json"
	"errors"
	"log/slog"
	"net/netip"
	"slices"
	"testing"

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

// TestPrefixHandlerWritesOriginChanges hands a running prefix handler an
// announcement and flushes. The flush must write the origin change to the
// database.
func TestPrefixHandlerWritesOriginChanges(t *testing.T) {
	t.Parallel()

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

	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
	h.flushBatch()

	got, err := db.GetOriginChanges(t.Context(),
		netip.MustParsePrefix(testV4Prefix), 10)
	if err != nil {
		t.Fatalf("GetOriginChanges: %v", err)
	}

	if len(got) != 1 || !got[0].Announced || got[0].OriginASN != testASNB {
		t.Errorf("GetOriginChanges = %+v, want one announced row for AS%d",
			got, testASNB)
	}
}

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

// originChangesFailsStore is a real database whose first write of the origin
// changes fails, writing nothing. writes counts the writes asked for.
type originChangesFailsStore struct {
	*database.Database

	writes int
}

func (s *originChangesFailsStore) WriteOriginChanges(
	changes []routetable.OriginChange,
) error {
	s.writes++
	if s.writes == 1 {
		return errTestOriginChangeWrite
	}

	return s.Database.WriteOriginChanges(changes)
}

// TestPrefixHandlerWritesAFailedOriginChangeWriteAgain runs two flushes into a
// database whose first write of the origin changes fails. The second write must
// hold the change of the first.
func TestPrefixHandlerWritesAFailedOriginChangeWriteAgain(t *testing.T) {
	t.Parallel()

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

	h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
	h.flushBatch()
	h.flushBatch()

	got, err := db.GetOriginChanges(t.Context(),
		netip.MustParsePrefix(testV4Prefix), 10)
	if err != nil {
		t.Fatalf("GetOriginChanges: %v", err)
	}

	if len(got) != 1 || !got[0].Announced || got[0].OriginASN != testASNB {
		t.Errorf("GetOriginChanges = %+v, want one announced row for AS%d",
			got, testASNB)
	}
}

// originChangesWriteSizesStore is a real database that records how many
// changes each write of the origin changes holds.
type originChangesWriteSizesStore struct {
	*database.Database

	counts []int
}

func (s *originChangesWriteSizesStore) WriteOriginChanges(
	changes []routetable.OriginChange,
) error {
	s.counts = append(s.counts, len(changes))

	return s.Database.WriteOriginChanges(changes)
}

// TestPrefixHandlerWritesManyOriginChangesOverSeveralFlushes has more origin
// changes than one write takes, then flushes three times. The first write must
// hold one write's worth, the second the rest, and the third nothing. The
// handler's limit is small.
func TestPrefixHandlerWritesManyOriginChangesOverSeveralFlushes(t *testing.T) {
	t.Parallel()

	const perWrite = 10

	db := newTestDatabase(t)
	store := &originChangesWriteSizesStore{Database: db}
	h := prefixHandlerWithoutTimer(store)
	h.originChangesPerWrite = perWrite

	announceManyPrefixes(h.table, perWrite*3/2)

	h.flushBatch()
	h.flushBatch()
	h.flushBatch()

	want := []int{perWrite, perWrite*3/2 - perWrite}
	if !slices.Equal(store.counts, want) {
		t.Errorf("WriteOriginChanges held %v changes, want %v", store.counts, want)
	}
}

// TestPrefixHandlerLogsTheDroppedOriginChanges fills the routing table with
// one origin change more than it keeps, then flushes. The warning the flush
// logs must carry the dropped count.
func TestPrefixHandlerLogsTheDroppedOriginChanges(t *testing.T) {
	t.Parallel()

	var logs bytes.Buffer

	db := newTestDatabase(t)
	h := prefixHandlerWithoutTimer(db)
	h.logger = &logger.Logger{
		Logger: slog.New(slog.NewJSONHandler(&logs,
			&slog.HandlerOptions{Level: slog.LevelWarn})),
	}

	announceManyPrefixes(h.table, routetable.MaxWaitingOriginChanges+1)

	h.flushBatch()

	var warning struct {
		Msg     string `json:"msg"`
		Dropped *int   `json:"dropped"`
	}

	found := false

	for line := range bytes.SplitSeq(bytes.TrimSpace(logs.Bytes()), []byte("\n")) {
		warning.Dropped = nil

		err := json.Unmarshal(line, &warning)
		if err != nil {
			t.Fatalf("want a JSON log line, got %q: %v", line, err)
		}

		if warning.Msg != "Origin changes dropped" {
			continue
		}

		found = true

		if warning.Dropped == nil || *warning.Dropped != 1 {
			t.Errorf("the warning's dropped = %v, want 1", warning.Dropped)
		}
	}

	if !found {
		t.Errorf("no warning logged the dropped origin changes")
	}
}