routewatch/internal/routewatch/batchwrite_internal_test.go

package routewatch

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"log/slog"
	"slices"
	"sync"
	"testing"
	"testing/synctest"
	"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"
)

const (
	// stepTimeout is how long a test waits for a step it expects, such as
	// HandleMessage returning, a write starting or Shutdown returning, before
	// it fails. Only a step that never comes should reach it. On a loaded
	// host, under the race detector, HandleMessage takes seconds to apply a
	// full prefix batch. The helpers below bound every wait with it, Stop
	// included. A test that hangs in one of them waits out at most two of
	// these bounds, its step's and Stop's, and then fails with its own
	// message. That leaves time before go test's 90-second timeout.
	stepTimeout = 15 * time.Second

	// stillWaitingAfter is how long a test watches a HandleMessage that must
	// wait. One that does not wait returns well within it.
	stillWaitingAfter = 100 * time.Millisecond
)

// TestFlushLoopWritesOnEveryTick runs flushLoop in a synctest bubble, whose
// clock is fake: time.Sleep there moves the clock on and returns at once. Each
// write takes a tenth of the interval. Two timed writes must start one interval
// apart. A full batch's write, which starts as the second timed write ends,
// must be followed by a timed write one interval after it began.
func TestFlushLoopWritesOnEveryTick(t *testing.T) {
	t.Parallel()

	synctest.Test(t, func(t *testing.T) {
		const (
			interval  = time.Second
			writeTime = 100 * time.Millisecond
		)

		writeStarts := make(chan time.Time)
		batchFullCh := make(chan struct{}, 1)
		stopCh := make(chan context.Context, 1)
		done := make(chan struct{})

		go func() {
			flushLoop(interval, batchFullCh, stopCh, done, func() {
				writeStarts <- time.Now()

				time.Sleep(writeTime)
			})
		}()

		first := <-writeStarts
		second := <-writeStarts

		if gap := second.Sub(first); gap != interval {
			t.Errorf("the second timed write started %s after the first, "+
				"want %s", gap, interval)
		}

		batchFullCh <- struct{}{}

		full := <-writeStarts
		timed := <-writeStarts

		if gap := timed.Sub(full); gap != interval {
			t.Errorf("the timed write after a full batch's started %s after "+
				"it, want %s", gap, interval)
		}

		stopCh <- t.Context()

		<-writeStarts // Stop's last write
		<-done
	})
}

// blockingStore is a mockStore whose batch writes for the AS, peer and prefix
// handlers and peering write wait until the test calls releaseWrites.
// writeStarted is closed when the first of them starts. It records what each
// batch write was given.
type blockingStore struct {
	*mockStore

	release      chan struct{}
	releaseOnce  sync.Once
	writeStarted chan struct{}
	startOnce    sync.Once

	mu       sync.Mutex
	prefixes map[string]bool
	asns     map[int]bool
	peers    map[string]bool
}

func newBlockingStore() *blockingStore {
	return &blockingStore{
		mockStore:    newMockStore(),
		release:      make(chan struct{}),
		writeStarted: make(chan struct{}),
		prefixes:     make(map[string]bool),
		asns:         make(map[int]bool),
		peers:        make(map[string]bool),
	}
}

func (s *blockingStore) UpdatePrefixesBatch(prefixes map[string]time.Time) error {
	s.waitForRelease()
	s.mu.Lock()
	defer s.mu.Unlock()

	for prefix := range prefixes {
		s.prefixes[prefix] = true
	}

	return nil
}

func (s *blockingStore) GetOrCreateASNBatch(asns map[int]database.SeenTimes) error {
	s.waitForRelease()
	s.mu.Lock()
	defer s.mu.Unlock()

	for asn := range asns {
		s.asns[asn] = true
	}

	return nil
}

func (s *blockingStore) UpdatePeerBatch(peers map[string]database.PeerUpdate) error {
	s.waitForRelease()
	s.mu.Lock()
	defer s.mu.Unlock()

	for peerIP := range peers {
		s.peers[peerIP] = true
	}

	return nil
}

func (s *blockingStore) RecordPeering(
	asA, asB int, seen database.SeenTimes,
) error {
	s.waitForRelease()

	return s.mockStore.RecordPeering(asA, asB, seen)
}

func (s *blockingStore) waitForRelease() {
	s.startOnce.Do(func() { close(s.writeStarted) })
	<-s.release
}

// releaseWrites lets the held writes, and all later ones, finish. A second call
// does nothing.
func (s *blockingStore) releaseWrites() {
	s.releaseOnce.Do(func() { close(s.release) })
}

// stopLog holds what a handler's Stop logs it did not write.
//
//nolint:tagliatelle // field names are the keys the handlers log
type stopLog struct {
	Msg          string `json:"msg"`
	Updates      int    `json:"updates"`
	TableChanges int    `json:"table_changes"`
	HeldMessages int    `json:"held_messages"`
	Paths        int    `json:"paths"`
	BatchEntries int    `json:"batch_entries"`
}

// newErrorLogger returns a logger that writes only errors, as JSON lines, to
// logs.
func newErrorLogger(logs io.Writer) *logger.Logger {
	return &logger.Logger{Logger: slog.New(slog.NewJSONHandler(logs,
		&slog.HandlerOptions{Level: slog.LevelError}))}
}

// lastLogged decodes the last line in logs, which is the one Stop logs: a
// handler logs nothing after it.
func lastLogged(t *testing.T, logs *bytes.Buffer) stopLog {
	t.Helper()

	lines := bytes.Split(bytes.TrimSpace(logs.Bytes()), []byte("\n"))

	var logged stopLog

	err := json.Unmarshal(lines[len(lines)-1], &logged)
	if err != nil {
		t.Fatalf("want the error Stop logs, got %q: %v", logs.String(), err)
	}

	return logged
}

// returnsInTime reports whether f returns within stepTimeout.
func returnsInTime(f func()) bool {
	done := make(chan struct{})

	go func() {
		f()
		close(done)
	}()

	select {
	case <-done:
		return true
	case <-time.After(stepTimeout):
		return false
	}
}

// stopInTime stops h, with no deadline, and fails the test if Stop does not
// return within stepTimeout. It only marks the test failed, so it can be
// deferred.
func stopInTime(t *testing.T, h interface{ Stop(ctx context.Context) }) {
	t.Helper()

	if !returnsInTime(func() { h.Stop(t.Context()) }) {
		t.Errorf("Stop did not return within %s", stepTimeout)
	}
}

// startBlockedWrite hands h the messages in fill and waits until h starts to
// write them to store. The prefix, AS and peer handlers start when fill fills
// their batch, and the peering handler on its timer. store holds the write
// until the test calls releaseWrites.
func startBlockedWrite(
	t *testing.T,
	store *blockingStore,
	h interface {
		HandleMessage(msg *ristypes.RISMessage)
	},
	fill []*ristypes.RISMessage,
) {
	t.Helper()

	if !returnsInTime(func() {
		for _, msg := range fill {
			h.HandleMessage(msg)
		}
	}) {
		t.Fatal("HandleMessage did not return after filling the batch: " +
			"it waits for the database write")
	}

	select {
	case <-store.writeStarted:
	case <-time.After(stepTimeout):
		t.Fatal("the handler did not start to write the messages")
	}
}

// handleWhileAWriteIsBlocked starts a blocked write with the messages in fill
// and hands h the message extra while store holds the write. HandleMessage must
// return each time. Then it lets the writes finish and stops h, which writes
// what it holds.
func handleWhileAWriteIsBlocked(
	t *testing.T,
	store *blockingStore,
	h interface {
		HandleMessage(msg *ristypes.RISMessage)
		Stop(ctx context.Context)
	},
	fill []*ristypes.RISMessage,
	extra *ristypes.RISMessage,
) {
	t.Helper()

	defer stopInTime(t, h)
	defer store.releaseWrites()

	startBlockedWrite(t, store, h, fill)

	if !returnsInTime(func() { h.HandleMessage(extra) }) {
		t.Fatal("HandleMessage did not return while a database write was blocked")
	}
}

// handleAtMaxBatchSize starts a blocked write with the messages in fill, then,
// while store holds the write, hands h the messages in toMax, which bring its
// next batch to its maximum size, and the message extra. HandleMessage for
// extra must wait until the test lets the write finish, and return after. Then
// it stops h, which writes what it holds. The tests lower h's maximum batch
// size to one batch, so toMax is one batch: the real maximum of ten batches is
// slow to reach under the race detector.
func handleAtMaxBatchSize(
	t *testing.T,
	store *blockingStore,
	h interface {
		HandleMessage(msg *ristypes.RISMessage)
		Stop(ctx context.Context)
	},
	fill, toMax []*ristypes.RISMessage,
	extra *ristypes.RISMessage,
) {
	t.Helper()

	defer stopInTime(t, h)
	defer store.releaseWrites()

	startBlockedWrite(t, store, h, fill)

	if !returnsInTime(func() {
		for _, msg := range toMax {
			h.HandleMessage(msg)
		}
	}) {
		t.Fatal("HandleMessage did not return while the batch grew to its " +
			"maximum size")
	}

	returned := make(chan struct{})

	go func() {
		h.HandleMessage(extra)
		close(returned)
	}()

	select {
	case <-returned:
		t.Fatal("HandleMessage returned while the batch was at its maximum " +
			"size and a write was blocked")
	case <-time.After(stillWaitingAfter):
	}

	store.releaseWrites()

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

// stopWhileAWriteIsBlocked starts a blocked write with the messages in fill,
// hands h the messages in extra, and stops h with a deadline that passes while
// store holds the write. Stop must return all the same. Then it lets the write
// finish and waits until done, closed when h's flush goroutine returns. Past
// the deadline that goroutine skips its last write, so extra is not written,
// and Stop logs as not written both extra and what the held write took.
func stopWhileAWriteIsBlocked(
	t *testing.T,
	store *blockingStore,
	h interface {
		HandleMessage(msg *ristypes.RISMessage)
		Stop(ctx context.Context)
	},
	done <-chan struct{},
	fill []*ristypes.RISMessage,
	extra ...*ristypes.RISMessage,
) {
	t.Helper()

	const stopDeadline = 100 * time.Millisecond

	defer store.releaseWrites()

	startBlockedWrite(t, store, h, fill)

	if !returnsInTime(func() {
		for _, msg := range extra {
			h.HandleMessage(msg)
		}
	}) {
		t.Fatal("HandleMessage did not return while a database write was blocked")
	}

	ctx, cancel := context.WithTimeout(t.Context(), stopDeadline)
	defer cancel()

	if !returnsInTime(func() { h.Stop(ctx) }) {
		t.Fatalf("Stop with a deadline of %s did not return while a write "+
			"was blocked", stopDeadline)
	}

	store.releaseWrites()

	select {
	case <-done:
	case <-time.After(stepTimeout):
		t.Fatal("the flush goroutine did not return after the write finished")
	}
}

// TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline also checks the
// routing table's changes: the flush whose prefix write is held wrote those of
// the fill, so Stop must log the one of the extra prefix as not written. The
// flush timer must not fire. A flush it started before the fill would be the
// one held, and it would not have written the fill's change.
func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
	t.Parallel()

	const extraPrefix = "198.51.100.0/24"

	var logs bytes.Buffer

	store := newBlockingStore()
	h := newPrefixHandler(store, newErrorLogger(&logs),
		routetable.New(routetable.Params{}), longerThanAnyTest,
		maxTableChangesPerWrite)
	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
	stopWhileAWriteIsBlocked(t, store, h, h.done,
		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
		prefixAnnouncement([]string{extraPrefix}))

	store.mu.Lock()
	defer store.mu.Unlock()

	if store.prefixes[extraPrefix] {
		t.Errorf("%s was written after the stop deadline", extraPrefix)
	}

	logged := lastLogged(t, &logs)
	if logged.Updates != prefixBatchSize+1 || logged.TableChanges != 1 {
		t.Errorf("Stop logged %d updates and %d table changes not written, "+
			"want %d and 1", logged.Updates, logged.TableChanges,
			prefixBatchSize+1)
	}
}

// TestPrefixHandlerSkipsAFullBatchAfterTheStopDeadline hands the prefix handler
// a full batch while a write is held, so the batch's wake-up of the flush
// goroutine waits. The stop deadline passes meanwhile. When the write ends, the
// flush goroutine must stop instead of writing the batch.
func TestPrefixHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) {
	t.Parallel()

	const extraPrefix = "198.51.100.0/24"

	store := newBlockingStore()
	h := NewPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}))
	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
	full := slices.Repeat([]string{extraPrefix}, prefixBatchSize)
	stopWhileAWriteIsBlocked(t, store, h, h.done,
		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
		prefixAnnouncement(full))

	store.mu.Lock()
	defer store.mu.Unlock()

	if store.prefixes[extraPrefix] {
		t.Errorf("%s was written after the stop deadline", extraPrefix)
	}
}

// tableWriteBlockingStore is a blockingStore whose writes of the routing
// table's changes also wait until the test calls releaseWrites.
type tableWriteBlockingStore struct {
	*blockingStore
}

func (s tableWriteBlockingStore) WriteRouteChanges(
	changes []routetable.Change, newPeers map[string]uint,
) error {
	s.waitForRelease()

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

// TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress holds the write
// of the routing table's changes that a full batch starts, then hands the
// prefix handler one more prefix and stops it with a deadline that passes
// meanwhile. Stop must log as not written both the change the held write took
// and the one since.
func TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress(t *testing.T) {
	t.Parallel()

	const extraPrefix = "198.51.100.0/24"

	var logs bytes.Buffer

	blocking := newBlockingStore()
	h := NewPrefixHandler(tableWriteBlockingStore{blockingStore: blocking},
		newErrorLogger(&logs), routetable.New(routetable.Params{}))
	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
	stopWhileAWriteIsBlocked(t, blocking, h, h.done,
		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
		prefixAnnouncement([]string{extraPrefix}))

	if logged := lastLogged(t, &logs); logged.TableChanges != 2 {
		t.Errorf("Stop logged %d table changes not written, want 2",
			logged.TableChanges)
	}
}

// TestASHandlerSkipsItsLastWriteAfterTheStopDeadline also checks what Stop
// logs: the asnBatchSize entries of the held write and the entry for testASNB,
// handled since, are not written.
func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
	t.Parallel()

	const wantEntries = asnBatchSize + 1

	var logs bytes.Buffer

	store := newBlockingStore()
	h := newASHandler(store, newErrorLogger(&logs), longerThanAnyTest)
	stopWhileAWriteIsBlocked(t, store, h, h.done,
		[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}},
		&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})

	store.mu.Lock()
	defer store.mu.Unlock()

	if store.asns[testASNB] {
		t.Errorf("AS%d was written after the stop deadline", testASNB)
	}

	if logged := lastLogged(t, &logs); logged.BatchEntries != wantEntries {
		t.Errorf("Stop logged %d batch entries not written, want %d",
			logged.BatchEntries, wantEntries)
	}
}

// TestASHandlerSkipsAFullBatchAfterTheStopDeadline hands the AS handler a full
// batch while a write is held, so the batch's wake-up of the flush goroutine
// waits. The full batch holds the fill's ASes again, and testASNB. The stop
// deadline passes meanwhile. When the write ends, the flush goroutine must stop
// instead of writing the batch.
func TestASHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) {
	t.Parallel()

	store := newBlockingStore()
	h := newASHandler(store, logger.New(), longerThanAnyTest)
	fill := &ristypes.RISMessage{Path: pathOfASes(asnBatchSize)}
	stopWhileAWriteIsBlocked(t, store, h, h.done,
		[]*ristypes.RISMessage{fill},
		fill,
		&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})

	store.mu.Lock()
	defer store.mu.Unlock()

	if store.asns[testASNB] {
		t.Errorf("AS%d was written after the stop deadline", testASNB)
	}
}

func TestPeerHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
	t.Parallel()

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

	var logs bytes.Buffer

	store := newBlockingStore()
	h := newPeerHandler(store, newErrorLogger(&logs), longerThanAnyTest)
	stopWhileAWriteIsBlocked(t, store, h, h.done,
		slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize),
		&ristypes.RISMessage{Peer: otherPeerIP})

	store.mu.Lock()
	defer store.mu.Unlock()

	if store.peers[otherPeerIP] {
		t.Errorf("peer %s was written after the stop deadline", otherPeerIP)
	}

	if logged := lastLogged(t, &logs); logged.Updates != peerBatchSize+1 {
		t.Errorf("Stop logged %d updates not written, want %d",
			logged.Updates, peerBatchSize+1)
	}
}

// TestPeerHandlerSkipsAFullBatchAfterTheStopDeadline hands the peer handler a
// full batch while a write is held, so the batch's wake-up of the flush
// goroutine waits. The stop deadline passes meanwhile. When the write ends, the
// flush goroutine must stop instead of writing the batch.
func TestPeerHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) {
	t.Parallel()

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

	store := newBlockingStore()
	h := newPeerHandler(store, logger.New(), longerThanAnyTest)
	stopWhileAWriteIsBlocked(t, store, h, h.done,
		slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize),
		slices.Repeat([]*ristypes.RISMessage{{Peer: otherPeerIP}},
			peerBatchSize)...)

	store.mu.Lock()
	defer store.mu.Unlock()

	if store.peers[otherPeerIP] {
		t.Errorf("peer %s was written after the stop deadline", otherPeerIP)
	}
}

// TestPeeringHandlerStopCountsTheWriteInProgress hands a running peering
// handler one AS path and stops it with a deadline that passes while store
// holds Stop's last write, of that path's peering. Stop must log that path as
// not written.
func TestPeeringHandlerStopCountsTheWriteInProgress(t *testing.T) {
	t.Parallel()

	const stopDeadline = 100 * time.Millisecond

	var logs bytes.Buffer

	store := newBlockingStore()
	defer store.releaseWrites()

	h := NewPeeringHandler(store, newErrorLogger(&logs))
	handle(h, time.Now().UTC(), testASNA, testASNB)

	ctx, cancel := context.WithTimeout(t.Context(), stopDeadline)
	defer cancel()

	if !returnsInTime(func() { h.Stop(ctx) }) {
		t.Fatalf("Stop with a deadline of %s did not return while a write "+
			"was blocked", stopDeadline)
	}

	store.releaseWrites()

	select {
	case <-h.done:
	case <-time.After(stepTimeout):
		t.Fatal("the processing goroutine did not return after the write " +
			"finished")
	}

	if logged := lastLogged(t, &logs); logged.Paths != 1 {
		t.Errorf("Stop logged %d paths not written, want 1", logged.Paths)
	}
}

// TestPeeringHandlerStartsNoWriteAfterTheStopDeadline has the peering handler's
// timer start a write that store holds, and hands the handler one more AS path
// meanwhile. The stop deadline passes, and a tick of the timer waits. When the
// write ends, the processing goroutine must stop instead of writing the path.
func TestPeeringHandlerStartsNoWriteAfterTheStopDeadline(t *testing.T) {
	t.Parallel()

	// The timer fires many times while Stop waits for its deadline.
	const processInterval = 10 * time.Millisecond

	store := newBlockingStore()
	h := newPeeringHandler(store, logger.New(), processInterval, maxTrackedPaths)
	stopWhileAWriteIsBlocked(t, store, h, h.done,
		[]*ristypes.RISMessage{
			{Path: ristypes.ASPath{Sequence: []int{testASNA, testASNB}}},
		},
		&ristypes.RISMessage{
			Path: ristypes.ASPath{Sequence: []int{testASNC, testASND}},
		})

	store.mockStore.mu.Lock()
	defer store.mockStore.mu.Unlock()

	if store.Peerings[fmt.Sprintf("%d_%d", testASNC, testASND)] {
		t.Errorf("the peering of AS%d and AS%d was written after the stop "+
			"deadline", testASNC, testASND)
	}
}

// prefixAnnouncement returns a message in which testPeerIP announces prefixes
// with origin AS testASNA.
func prefixAnnouncement(prefixes []string) *ristypes.RISMessage {
	return &ristypes.RISMessage{
		Peer: testPeerIP,
		Path: ristypes.ASPath{Sequence: []int{testASNA}},
		Announcements: []ristypes.RISAnnouncement{
			{NextHop: testPeerIP, Prefixes: prefixes},
		},
	}
}

func TestPrefixHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
	t.Parallel()

	const extraPrefix = "198.51.100.0/24"

	store := newBlockingStore()
	h := NewPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}))
	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
	handleWhileAWriteIsBlocked(t, store, h,
		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
		prefixAnnouncement([]string{extraPrefix}))

	store.mu.Lock()
	defer store.mu.Unlock()

	if !store.prefixes[extraPrefix] {
		t.Errorf("Stop did not write %s, handled while a write was blocked",
			extraPrefix)
	}
}

func TestASHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
	t.Parallel()

	store := newBlockingStore()
	h := newASHandler(store, logger.New(), longerThanAnyTest)
	handleWhileAWriteIsBlocked(t, store, h,
		[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}},
		&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})

	store.mu.Lock()
	defer store.mu.Unlock()

	if !store.asns[testASNB] {
		t.Errorf("Stop did not write AS%d, handled while a write was blocked",
			testASNB)
	}
}

func TestPeerHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
	t.Parallel()

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

	store := newBlockingStore()
	h := newPeerHandler(store, logger.New(), longerThanAnyTest)
	handleWhileAWriteIsBlocked(t, store, h,
		slices.Repeat([]*ristypes.RISMessage{{Peer: testPeerIP}}, peerBatchSize),
		&ristypes.RISMessage{Peer: otherPeerIP})

	store.mu.Lock()
	defer store.mu.Unlock()

	if !store.peers[otherPeerIP] {
		t.Errorf("Stop did not write peer %s, handled while a write was blocked",
			otherPeerIP)
	}
}

func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
	t.Parallel()

	const extraPrefix = "198.51.100.0/24"

	store := newBlockingStore()
	h := newPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}),
		longerThanAnyTest, maxTableChangesPerWrite)
	h.batchMaxSize = prefixBatchSize
	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
	toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
	handleAtMaxBatchSize(t, store, h,
		[]*ristypes.RISMessage{prefixAnnouncement(fill)},
		[]*ristypes.RISMessage{prefixAnnouncement(toMax)},
		prefixAnnouncement([]string{extraPrefix}))

	store.mu.Lock()
	defer store.mu.Unlock()

	if !store.prefixes[extraPrefix] {
		t.Errorf("Stop did not write %s, handled after the batch reached its "+
			"maximum size", extraPrefix)
	}
}

// TestASHandlerNeverWaitsForAWrite starts a write that the store holds. Then it
// hands the AS handler twenty messages whose paths each hold one AS
// asnBatchSize times. HandleMessage must return each time, and the batch must
// hold one entry, for that AS.
func TestASHandlerNeverWaitsForAWrite(t *testing.T) {
	t.Parallel()

	const messages = 20

	store := newBlockingStore()
	h := newASHandler(store, logger.New(), longerThanAnyTest)

	defer stopInTime(t, h)
	defer store.releaseWrites()

	startBlockedWrite(t, store, h,
		[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}})

	oneAS := &ristypes.RISMessage{
		Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNB}, asnBatchSize)},
	}

	if !returnsInTime(func() {
		for range messages {
			h.HandleMessage(oneAS)
		}
	}) {
		t.Fatal("HandleMessage did not return while a database write was blocked")
	}

	h.mu.Lock()
	defer h.mu.Unlock()

	if len(h.batch) != 1 {
		t.Errorf("the batch holds %d entries for one AS, want 1", len(h.batch))
	}
}

// TestPeerHandlerWaitsAtTheMaximumBatchSize must not have its flush timer fire.
// A timer flush during the fill would take only part of it. The batch would
// then reach its maximum size before the test's last message, and the test
// would fail.
func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
	t.Parallel()

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

	store := newBlockingStore()
	h := newPeerHandler(store, logger.New(), longerThanAnyTest)
	h.batchMaxSize = peerBatchSize
	message := []*ristypes.RISMessage{{Peer: testPeerIP}}
	handleAtMaxBatchSize(t, store, h,
		slices.Repeat(message, peerBatchSize),
		slices.Repeat(message, peerBatchSize),
		&ristypes.RISMessage{Peer: otherPeerIP})

	store.mu.Lock()
	defer store.mu.Unlock()

	if !store.peers[otherPeerIP] {
		t.Errorf("Stop did not write peer %s, handled after the batch reached "+
			"its maximum size", otherPeerIP)
	}
}

// TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull hands each handler, built
// without its flush goroutine so that nothing takes a wake-up, one update short
// of a full batch, and a second one exactly a full batch. Only the full batch
// leaves a wake-up in batchFullCh.
func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
	t.Parallel()

	tests := []struct {
		name      string
		batchSize int
		// handle hands a new handler n updates and returns its batchFullCh.
		handle func(n int) chan struct{}
	}{
		{
			name:      "prefix",
			batchSize: prefixBatchSize,
			handle: func(n int) chan struct{} {
				h := &PrefixHandler{
					batchMaxSize: prefixBatchMaxSize,
					batchFullCh:  make(chan struct{}, 1),
					table:        routetable.New(routetable.Params{}),
				}
				h.HandleMessage(prefixAnnouncement(
					slices.Repeat([]string{testV4Prefix}, n)))

				return h.batchFullCh
			},
		},
		{
			name:      "AS",
			batchSize: asnBatchSize,
			handle: func(n int) chan struct{} {
				h := &ASHandler{
					batch:       make(map[int]database.SeenTimes),
					batchFullCh: make(chan struct{}, 1),
				}
				// Each of the path's n ASes is one entry of the batch.
				h.HandleMessage(&ristypes.RISMessage{Path: pathOfASes(n)})

				return h.batchFullCh
			},
		},
		{
			name:      "peer",
			batchSize: peerBatchSize,
			handle: func(n int) chan struct{} {
				h := &PeerHandler{
					batchMaxSize: peerBatchMaxSize,
					batchFullCh:  make(chan struct{}, 1),
				}
				for range n {
					h.HandleMessage(&ristypes.RISMessage{Peer: testPeerIP})
				}

				return h.batchFullCh
			},
		},
	}

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

			got := len(tt.handle(tt.batchSize - 1))
			if got != 0 {
				t.Errorf("one update short of a full batch left %d wake-ups, "+
					"want none", got)
			}

			got = len(tt.handle(tt.batchSize))
			if got != 1 {
				t.Errorf("a full batch left %d wake-ups, want 1", got)
			}
		})
	}
}