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

README.md

@@ -204,9 +204,9 @@ routewatch/
   capacity
 - **Graceful Shutdown**: on SIGTERM or SIGINT the streamer stops, the handlers
   write the batches they hold, the HTTP server stops and the database is
-  checkpointed and closed, all within 8 seconds, and the daemon exits with
+  checkpointed and closed. The daemon gives this 8 seconds and exits with status
-  status 0. What is not written by then is lost, and the handler logs how many
+  0; "Running under upaas" says what a stop loses, logs and leaves undone when
-  updates (see "Running under upaas").
+  the 8 seconds run out.
 - **Reconnection**: Any end of the stream other than shutdown, including RIS
   Live ending the response normally, marks the feed disconnected and reconnects
   with exponential backoff (5s-320s). The backoff resets after a stream that

internal/routewatch/app_integration_internal_test.go

@@ -671,8 +671,9 @@ func freePort(t *testing.T) int {
 // TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked starts the HTTP server
 // and has the AS handler start a write that the store holds until the test
 // ends, past Shutdown's deadline. Shutdown must return within its deadline
-// plus a small margin, with the HTTP server stopped. Before the fix Shutdown
+// plus a small margin, with the HTTP server stopped, and must neither
-// waited for the write.
+// checkpoint nor close the database: the real one would wait for the write.
+// Before the fix Shutdown waited for the write.
 func TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) {
 	t.Parallel()
 
@@ -683,9 +684,10 @@ func TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) {
 		margin = 2 * time.Second
 	)
 
-	store := newBlockingStore()
+	blocking := newBlockingStore()
-	defer store.releaseWrites()
+	defer blocking.releaseWrites()
 
+	store := &closeOrderStore{Store: blocking}
 	log := logger.New()
 	port := freePort(t)
 	cfg := &config.Config{Port: port}
@@ -706,7 +708,8 @@ func TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) {
 
 	rw.dbHandler = NewASHandler(store, log)
 	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
-	startBlockedWrite(t, store, rw.dbHandler, []*ristypes.RISMessage{{Path: fill}})
+	startBlockedWrite(t, blocking, rw.dbHandler,
+		[]*ristypes.RISMessage{{Path: fill}})
 
 	ctx, cancel := context.WithTimeout(t.Context(), deadline)
 	defer cancel()
@@ -736,12 +739,21 @@ func TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) {
 		t.Errorf("the HTTP server still accepts connections on %s after "+
 			"Shutdown returned", addr)
 	}
+
+	store.callsMu.Lock()
+	defer store.callsMu.Unlock()
+
+	if slices.Contains(store.calls, callCheckpoint) ||
+		slices.Contains(store.calls, callClose) {
+		t.Errorf("Shutdown made the calls %q past its deadline, want no %s "+
+			"and no %s", store.calls, callCheckpoint, callClose)
+	}
 }
 
-// closeOrderStore is a mockStore that records, in order, the AS writes,
+// closeOrderStore is a store that records, in order, the AS writes,
-// checkpoints and closes it is given.
+// checkpoints and closes it is given. It passes the AS writes on to Store.
 type closeOrderStore struct {
-	*mockStore
+	database.Store
 
 	callsMu sync.Mutex
 	calls   []string
@@ -757,7 +769,7 @@ const (
 func (s *closeOrderStore) GetOrCreateASNBatch(asns map[int]time.Time) error {
 	s.record(callWrite)
 
-	return s.mockStore.GetOrCreateASNBatch(asns)
+	return s.Store.GetOrCreateASNBatch(asns)
 }
 
 func (s *closeOrderStore) Checkpoint(_ context.Context) error {
@@ -788,7 +800,7 @@ func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
 
 	const deadline = 10 * time.Second
 
-	store := &closeOrderStore{mockStore: newMockStore()}
+	store := &closeOrderStore{Store: newMockStore()}
 	log := logger.New()
 	cfg := &config.Config{}
 	s := streamer.New(log, metrics.New())

internal/routewatch/asnfetcher_internal_test.go

@@ -155,3 +155,51 @@ func TestFetchAndUpdateMarksAnASWithoutReferral(t *testing.T) {
 			baseInterval)
 	}
 }
+
+// TestASNFetcherStopReturnsByItsDeadline starts a fetcher, has it look up an
+// AS whose WHOIS server does not answer until the test ends, and stops it with
+// a short deadline while that lookup is in progress. Stop must return all the
+// same instead of waiting for the lookup.
+func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
+	t.Parallel()
+
+	const stopDeadline = 100 * time.Millisecond
+
+	lookupStarted := make(chan struct{})
+	release := make(chan struct{})
+
+	f := NewASNFetcher(newMockStore(), logger.New().Logger)
+	f.whoisClient.DialContext = func(
+		context.Context, string, string,
+	) (net.Conn, error) {
+		close(lookupStarted)
+		<-release
+
+		return nil, syscall.ECONNREFUSED
+	}
+
+	f.Start(t.Context())
+	f.QueueImmediate(testASNA)
+
+	select {
+	case <-lookupStarted:
+	case <-time.After(handleMessageTimeout):
+		t.Fatal("the fetcher did not start the lookup")
+	}
+
+	ctx, cancel := context.WithTimeout(t.Context(), stopDeadline)
+	defer cancel()
+
+	if !returnsInTime(func() { f.Stop(ctx) }) {
+		t.Fatalf("Stop with a deadline of %s did not return while a lookup "+
+			"was in progress", stopDeadline)
+	}
+
+	close(release)
+
+	select {
+	case <-f.done:
+	case <-time.After(handleMessageTimeout):
+		t.Fatal("the fetcher goroutine did not return after the lookup ended")
+	}
+}

internal/routewatch/dbmaintainer_internal_test.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"log/slog"
 	"testing"
+	"time"
 
 	"git.eeqj.de/sneak/routewatch/internal/database"
 )
@@ -52,3 +53,68 @@ func TestIncompleteCheckpointNotLoggedAsCompleted(t *testing.T) {
 			m.checkpointCount)
 	}
 }
+
+// heldCheckpointStore is a store whose checkpoint closes started and then
+// waits until release is closed.
+type heldCheckpointStore struct {
+	*mockStore
+
+	started chan struct{}
+	release chan struct{}
+}
+
+// Checkpoint waits until release is closed.
+func (s heldCheckpointStore) Checkpoint(context.Context) error {
+	close(s.started)
+	<-s.release
+
+	return nil
+}
+
+// TestDBMaintainerStopReturnsByItsDeadline stops the maintainer with a short
+// deadline while its goroutine runs a checkpoint that the store holds until
+// the test ends. Stop must return all the same instead of waiting for the
+// checkpoint.
+func TestDBMaintainerStopReturnsByItsDeadline(t *testing.T) {
+	t.Parallel()
+
+	const stopDeadline = 100 * time.Millisecond
+
+	store := heldCheckpointStore{
+		mockStore: newMockStore(),
+		started:   make(chan struct{}),
+		release:   make(chan struct{}),
+	}
+	m := NewDBMaintainer(store, slog.Default())
+
+	// Start would run the first checkpoint only after checkpointInterval.
+	// This goroutine stands for the one Start runs and checkpoints at once.
+	go func() {
+		defer close(m.done)
+
+		m.runCheckpoint(t.Context())
+	}()
+
+	select {
+	case <-store.started:
+	case <-time.After(handleMessageTimeout):
+		t.Fatal("the maintainer did not start the checkpoint")
+	}
+
+	ctx, cancel := context.WithTimeout(t.Context(), stopDeadline)
+	defer cancel()
+
+	if !returnsInTime(func() { m.Stop(ctx) }) {
+		t.Fatalf("Stop with a deadline of %s did not return while a "+
+			"checkpoint was in progress", stopDeadline)
+	}
+
+	close(store.release)
+
+	select {
+	case <-m.done:
+	case <-time.After(handleMessageTimeout):
+		t.Fatal("the maintenance goroutine did not return after the " +
+			"checkpoint ended")
+	}
+}

internal/routewatch/prefixhandler_internal_test.go

@@ -796,6 +796,25 @@ func TestPrefixHandlerStopLogsAFailedDeleteWithoutTheDeadline(t *testing.T) {
 	}
 }
 
+// TestPrefixHandlerStopLogsNothingAfterADeleteThatSucceeds stops a prefix
+// handler, with no deadline, after a peer's session went down, into a database
+// whose delete of the peer's routes succeeds. Stop must log nothing: a peer
+// whose routes are deleted no longer counts as a peer down left undone.
+func TestPrefixHandlerStopLogsNothingAfterADeleteThatSucceeds(t *testing.T) {
+	t.Parallel()
+
+	var logs bytes.Buffer
+
+	h := NewPrefixHandler(newTestDatabase(t), newErrorLogger(&logs))
+	h.HandleMessage(sessionDown())
+	h.Stop(t.Context())
+
+	if logs.Len() != 0 {
+		t.Errorf("Stop logged %q after the routes of the peer whose session "+
+			"went down were deleted", logs.String())
+	}
+}
+
 // newTestDatabase opens a database in a new temporary directory. It is closed
 // when the test ends.
 func newTestDatabase(t *testing.T) *database.Database {

internal/server/handlers.go

@@ -675,10 +675,7 @@ func (s *Server) handleIPInfo() http.HandlerFunc {
 		}
 
 		ctx := r.Context()
-		response := IPLookupResponse{
+		response := IPLookupResponse{Query: target, Results: make([]*database.IPInfo, 0)}
-			Query:   target,
-			Results: make([]*database.IPInfo, 0),
-		}
 
 		// Collect all IPs to look up
 		ipsToLookup, resolveErrors := s.resolveLookupTargets(ctx, target)
@@ -708,11 +705,16 @@ func (s *Server) handleIPInfo() http.HandlerFunc {
 				return
 			}
 
-			if err != nil {
+			switch {
-				response.Errors = append(response.Errors,
+			case errors.Is(err, database.ErrNoRoute), errors.Is(err, database.ErrInvalidIP):
-					ip+": "+s.lookupErrorMessage(ip, err))
+				response.Errors = append(response.Errors, ip+": "+err.Error())
 
 				continue
+			case err != nil:
+				s.logger.Error("Failed to look up IP", "ip", ip, "error", err)
+				s.writeJSONError(w, http.StatusInternalServerError, internalErrorMessage)
+
+				return
 			}
 
 			// Do PTR lookup for this IP
@@ -751,19 +753,6 @@ func (s *Server) handleIPInfo() http.HandlerFunc {
 	}
 }
 
-// lookupErrorMessage returns what an IP lookup answer says about err, the
-// failed lookup of ip: err's own text when ip is not a valid address or has no
-// route, otherwise a fixed message, with err in the log.
-func (s *Server) lookupErrorMessage(ip string, err error) string {
-	if errors.Is(err, database.ErrNoRoute) || errors.Is(err, database.ErrInvalidIP) {
-		return err.Error()
-	}
-
-	s.logger.Error("Failed to look up IP", "ip", ip, "error", err)
-
-	return internalErrorMessage
-}
-
 // lookupTarget returns what an IP lookup request asks for: the ip or addr URL
 // parameter, or else the client's own address.
 func lookupTarget(r *http.Request) string {

internal/server/handlers_internal_test.go

@@ -806,8 +806,7 @@ func (failingDB) GetPrefixDetailsContext(
 // the health check with a store whose reads fail with the driver's error text,
 // and a prefix that does not parse, and checks that no answer carries the
 // driver's or the prefix parser's text. Before the fix the answers sent it to
-// the client (https://git.eeqj.de/sneak/routewatch/issues/108). The IP lookup
+// the client (https://git.eeqj.de/sneak/routewatch/issues/108).
-// answers 404, as it does whenever no address could be looked up.
 func TestJSONRoutesDoNotSendErrorDetails(t *testing.T) {
 	t.Parallel()
 
@@ -823,7 +822,7 @@ func TestJSONRoutesDoNotSendErrorDetails(t *testing.T) {
 		{statsPath, http.StatusInternalServerError},
 		{"/api/v1/as/64500", http.StatusInternalServerError},
 		{"/api/v1/prefix/192.0.2.0/24", http.StatusInternalServerError},
-		{"/api/v1/ip/192.0.2.1", http.StatusNotFound},
+		{"/api/v1/ip/192.0.2.1", http.StatusInternalServerError},
 		{"/api/v1/prefix/foo/24", http.StatusBadRequest},
 	}
 
@@ -846,6 +845,78 @@ func TestJSONRoutesDoNotSendErrorDetails(t *testing.T) {
 	}
 }
 
+// The addresses failingIPLookupDB fails the read of and finds a route for.
+const (
+	failingLookupIP = "192.0.2.1"
+	routedLookupIP  = "192.0.2.2"
+)
+
+// failingIPLookupDB embeds database.Store (left nil) and answers the IP lookup
+// with errDiskIO for failingLookupIP, a route for routedLookupIP, and no route
+// for any other address. Every other Store method is unused and would panic if
+// called.
+type failingIPLookupDB struct {
+	database.Store
+}
+
+func (failingIPLookupDB) GetIPInfoContext(
+	_ context.Context, ip string,
+) (*database.IPInfo, error) {
+	switch ip {
+	case failingLookupIP:
+		return nil, errDiskIO
+	case routedLookupIP:
+		return &database.IPInfo{IP: ip}, nil
+	default:
+		return nil, database.ErrNoRoute
+	}
+}
+
+// TestIPLookupAnswers500WhenDatabaseReadFails sends IP lookups through the
+// server's router with a store whose read of one address fails with the
+// driver's error. A lookup that asks for that address answers 500 with the
+// JSON error body, as the AS and prefix JSON routes do; an address with no
+// route still answers 404. Before the fix the failed read was reported as if
+// the address had no route: 404 for that address alone, and 200 beside a
+// routed address (https://git.eeqj.de/sneak/routewatch/issues/109).
+func TestIPLookupAnswers500WhenDatabaseReadFails(t *testing.T) {
+	t.Parallel()
+
+	s := New(failingIPLookupDB{}, streamer.New(logger.New(), metrics.New()),
+		logger.New(), &config.Config{})
+
+	tests := []struct {
+		path     string
+		wantCode int
+	}{
+		{
+			path:     "/api/v1/ip/" + failingLookupIP,
+			wantCode: http.StatusInternalServerError,
+		},
+		{
+			path:     "/api/v1/ip/" + failingLookupIP + "," + routedLookupIP,
+			wantCode: http.StatusInternalServerError,
+		},
+		{
+			path:     "/api/v1/ip/203.0.113.1",
+			wantCode: http.StatusNotFound,
+		},
+	}
+
+	for _, tt := range tests {
+		rec := httptest.NewRecorder()
+		s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
+			t.Context(), http.MethodGet, tt.path, nil))
+
+		if rec.Code != tt.wantCode {
+			t.Errorf("%s: status %d, want %d; body %s",
+				tt.path, rec.Code, tt.wantCode, rec.Body)
+		}
+
+		checkJSONErrorCode(t, tt.path, rec, tt.wantCode)
+	}
+}
+
 // settledGoroutineCount lets transient goroutines finish, then reports the
 // current count.
 func settledGoroutineCount() int {

internal/streamer/streamer_internal_test.go

@@ -1,9 +1,12 @@
 package streamer
 
 import (
+	"bytes"
 	"context"
+	"encoding/json"
 	"errors"
 	"io"
+	"log/slog"
 	"net/http"
 	"net/http/httptest"
 	"runtime"
@@ -416,7 +419,8 @@ func TestStopQueuesTheHeldStateMessages(t *testing.T) {
 // and a STATE message held for it, and a worker that stays busy until the test
 // ends, as one does whose handler waits for a database write. Stop must return
 // within its deadline plus a small margin instead of waiting for room on the
-// queue and for the worker.
+// queue and for the worker, and log as not handed over the held message and
+// every message on the queue.
 func TestStopReturnsByItsDeadline(t *testing.T) {
 	t.Parallel()
 
@@ -425,10 +429,15 @@ func TestStopReturnsByItsDeadline(t *testing.T) {
 		margin   = 2 * time.Second
 	)
 
+	var logs bytes.Buffer
+
 	s := newStreamerWithFullStateQueue("")
+	s.logger = &logger.Logger{Logger: slog.New(slog.NewJSONHandler(&logs,
+		&slog.HandlerOptions{Level: slog.LevelError}))}
 	s.handlers[0].heldState = []*ristypes.RISMessage{
 		{Type: ristypes.MessageTypeState},
 	}
+	wantNotHandedOver := 1 + len(s.handlers[0].queue)
 	// Stop cancels the stream Start would have started; there is none here.
 	s.cancel = func() {}
 
@@ -452,6 +461,20 @@ func TestStopReturnsByItsDeadline(t *testing.T) {
 	case <-time.After(deadline + margin):
 		t.Fatalf("Stop did not return within %s of its deadline", margin)
 	}
+
+	var logged struct {
+		Messages int `json:"messages"`
+	}
+
+	err := json.Unmarshal(logs.Bytes(), &logged)
+	if err != nil {
+		t.Fatalf("want the error Stop logs, got %q: %v", logs.String(), err)
+	}
+
+	if logged.Messages != wantNotHandedOver {
+		t.Errorf("Stop logged %d messages not handed over, want %d",
+			logged.Messages, wantNotHandedOver)
+	}
 }
 
 // TestStopBeforeMessageReachesQueues stops the streamer after the read loop