Bound each shutdown stage by the stop deadline (closes #100)

Shutdown passes its context to every stage: the streamer, the four
handlers, the WHOIS fetcher, the database maintainer and the HTTP server
each stop waiting when it ends. A handler then skips its last write and
logs how many updates it did not write, and the HTTP server closes the
connections still open. The database is checkpointed and closed last,
and left open once the deadline has passed, since a write may still
hold it. The stop hook gives Shutdown 8 seconds and fx 9, inside the 10
seconds upaas and docker stop allow, so a stop signal exits 0.

Model: opus-5-5

internal/routewatch/batchwrite_internal_test.go

@@ -1,6 +1,10 @@
 package routewatch
 
 import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"log/slog"
 	"slices"
 	"sync"
 	"testing"
@@ -22,8 +26,9 @@ const (
 )
 
 // blockingStore is a mockStore whose batch writes for the AS, peer and prefix
-// handlers wait until the test calls releaseWrites. writeStarted is closed when
-// the first of them starts. It records what each write was given.
+// handlers, delete of a peer's routes 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
 
@@ -85,6 +90,18 @@ func (s *blockingStore) UpdatePeerBatch(peers map[string]database.PeerUpdate) er
 	return nil
 }
 
+func (s *blockingStore) DeleteLiveRoutesOfPeer(peerIP string) (int, error) {
+	s.waitForRelease()
+
+	return s.mockStore.DeleteLiveRoutesOfPeer(peerIP)
+}
+
+func (s *blockingStore) RecordPeering(asA, asB int, ts time.Time) error {
+	s.waitForRelease()
+
+	return s.mockStore.RecordPeering(asA, asB, ts)
+}
+
 func (s *blockingStore) waitForRelease() {
 	s.startOnce.Do(func() { close(s.writeStarted) })
 	<-s.release
@@ -96,6 +113,40 @@ 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"`
+	PeersDown int    `json:"peers_down"`
+	Paths     int    `json:"paths"`
+}
+
+// newErrorLogger returns a logger that writes only errors, as JSON lines, to
+// logs.
+func newErrorLogger(logs *bytes.Buffer) *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 handleMessageTimeout.
 func returnsInTime(f func()) bool {
 	done := make(chan struct{})
@@ -151,14 +202,14 @@ func handleWhileAWriteIsBlocked(
 	store *blockingStore,
 	h interface {
 		HandleMessage(msg *ristypes.RISMessage)
-		Stop()
+		Stop(ctx context.Context)
 	},
 	fill []*ristypes.RISMessage,
 	extra *ristypes.RISMessage,
 ) {
 	t.Helper()
 
-	defer h.Stop()
+	defer h.Stop(t.Context())
 	defer store.releaseWrites()
 
 	startBlockedWrite(t, store, h, fill)
@@ -178,14 +229,14 @@ func handleAtMaxBatchSize(
 	store *blockingStore,
 	h interface {
 		HandleMessage(msg *ristypes.RISMessage)
-		Stop()
+		Stop(ctx context.Context)
 	},
 	fill, toMax []*ristypes.RISMessage,
 	extra *ristypes.RISMessage,
 ) {
 	t.Helper()
 
-	defer h.Stop()
+	defer h.Stop(t.Context())
 	defer store.releaseWrites()
 
 	startBlockedWrite(t, store, h, fill)
@@ -217,6 +268,173 @@ func handleAtMaxBatchSize(
 	}
 }
 
+// stopWhileAWriteIsBlocked starts a blocked write with the messages in fill,
+// hands h the message 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)
+	h.HandleMessage(extra)
+
+	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(handleMessageTimeout):
+		t.Fatal("the flush goroutine did not return after the write finished")
+	}
+}
+
+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))
+	fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
+	// The peer's session goes down first, so the write held is the delete of
+	// its routes, and Stop must log that peer down too.
+	stopWhileAWriteIsBlocked(t, store, h, h.done,
+		[]*ristypes.RISMessage{sessionDown(), 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.PeersDown != 1 {
+		t.Errorf("Stop logged %d updates and %d peer downs not written, "+
+			"want %d and 1", logged.Updates, logged.PeersDown, prefixBatchSize+1)
+	}
+}
+
+func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
+	t.Parallel()
+
+	// Each message gives an update for its origin AS and one for each AS
+	// in its path: fill asnBatchSize+1, extra 2.
+	const wantUpdates = asnBatchSize + 1 + 2
+
+	var logs bytes.Buffer
+
+	store := newBlockingStore()
+	h := NewASHandler(store, newErrorLogger(&logs))
+	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
+	stopWhileAWriteIsBlocked(t, store, h, h.done,
+		[]*ristypes.RISMessage{{Path: 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)
+	}
+
+	if logged := lastLogged(t, &logs); logged.Updates != wantUpdates {
+		t.Errorf("Stop logged %d updates not written, want %d",
+			logged.Updates, wantUpdates)
+	}
+}
+
+func TestPeerHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
+	t.Parallel()
+
+	const otherPeerIP = "2001:db8:ffff::2"
+
+	var logs bytes.Buffer
+
+	store := newBlockingStore()
+	h := NewPeerHandler(store, newErrorLogger(&logs))
+	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)
+	}
+}
+
+// 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(handleMessageTimeout):
+		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)
+	}
+}
+
 // prefixAnnouncement returns a message in which testPeerIP announces prefixes
 // with origin AS testASNA.
 func prefixAnnouncement(prefixes []string) *ristypes.RISMessage {