Make the tests pass on a busy host (closes #240)

Tests that waited a few seconds for something to happen now wait 15 seconds.
The wait ends as soon as the condition holds, so a passing run takes no longer.

TestStreamNotDroppedWhileLinesArrive runs on a fake clock, so a stall on the
host cannot use up its idle timeout.

The checkpoint tests give the writing connection a 30-second busy timeout, so
their limit can be 15 seconds and a checkpoint that waits for a read still
fails them.

Model: opus-5-5

TODO.md

@@ -56,6 +56,9 @@ problem the run finds.
 
 # Completed Steps
 
+- 2026-11-10: the tests no longer fail when a busy host stalls for several
+  seconds. A wait for something that should happen is now 15 seconds. The test
+  that lines keep a stream up runs on a fake clock (closes #240)
 - 2026-11-05: the prefix page lists the prefix's origin changes, newest first.
   The routing table records each change, and the prefix handler writes them to a
   new `origin_changes` table. The database maintainer prunes each prefix to its

cmd/asinfo-gen/main_test.go

@@ -12,7 +12,7 @@ import (
 
 // parseTimeLimit is how long a test waits for parseCSV to return before it
 // fails.
-const parseTimeLimit = 2 * time.Second
+const parseTimeLimit = 15 * time.Second
 
 // TestParseCSVReadsTodaysLayout parses an as.csv with the country code column
 // that the source added after the description. It expects each AS with its

internal/blocklist/blocklist_internal_test.go

@@ -301,9 +301,13 @@ func TestRestartLoadsCopiesFromDisk(t *testing.T) {
 	second.Start(t.Context())
 	second.Stop(t.Context())
 
+	// The download takes its time after writing the file, and the restart
+	// reads the file's time. A busy host can put seconds between the two.
+	const maxUpdatedDifference = 15 * time.Second
+
 	listings := lookup(t, second, listedAddr)
 	if len(listings) != 1 ||
-		listings[0].Updated.Sub(downloaded).Abs() > time.Second {
+		listings[0].Updated.Sub(downloaded).Abs() > maxUpdatedDifference {
 		t.Errorf("after the restart, listings %+v, want one updated at %s",
 			listings, downloaded)
 	}

internal/database/database_internal_test.go

@@ -72,17 +72,34 @@ const (
 	// lockTestPrefix is the prefix the batch writes in the checkpoint tests
 	// store.
 	lockTestPrefix = "192.0.2.0/24"
+	// longBusyTimeoutMs is the busy timeout setLongBusyTimeout gives the
+	// writing connection. A checkpoint that waits for an open read takes this
+	// long.
+	longBusyTimeoutMs = 30000
 	// readerCheckpointLimit is how long a checkpoint, and a batch write
-	// waiting for it, may take while a reader is open: well under the busy
+	// waiting for it, may take while a reader is open: well under
-	// timeout, which a checkpoint that waits for the reader takes.
+	// longBusyTimeoutMs, with room for a stall on a busy host.
-	readerCheckpointLimit = time.Second
+	readerCheckpointLimit = 15 * time.Second
 	// truncateCheckpointLimit is how long a checkpoint of a WAL file over
 	// sqliteWALSizeLimitBytes may take to give up while a reader is open: its
-	// wait for the reader plus a second, well under the busy timeout.
+	// wait for the reader plus readerCheckpointLimit.
 	truncateCheckpointLimit = truncateBusyTimeoutMs*time.Millisecond +
-		time.Second
+		readerCheckpointLimit
 )
 
+// setLongBusyTimeout gives the writing connection of db longBusyTimeoutMs, so
+// that a checkpoint that waits for an open read takes far longer than a stall
+// on a busy host.
+func setLongBusyTimeout(ctx context.Context, t *testing.T, db *Database) {
+	t.Helper()
+
+	_, err := db.writer.ExecContext(ctx,
+		fmt.Sprintf("PRAGMA busy_timeout=%d", longBusyTimeoutMs))
+	if err != nil {
+		t.Fatalf("failed to set the busy timeout: %v", err)
+	}
+}
+
 // TestConnectionPoolPragmas holds several pooled connections open at once and
 // checks each one carries the per-connection settings from the DSN, plus the
 // process-wide heap limits, and sees the new file with auto_vacuum
@@ -579,6 +596,8 @@ func TestCheckpointDoesNotWaitForReader(t *testing.T) {
 	ctx := context.Background()
 	ts := time.Now().UTC()
 
+	setLongBusyTimeout(ctx, t, db)
+
 	err = db.UpdatePrefixesBatch(map[string]time.Time{"198.51.100.0/24": ts})
 	if err != nil {
 		t.Fatalf("failed to store the prefix the reader reads: %v", err)
@@ -690,6 +709,7 @@ func TestCheckpointEmptiesWALOnceNoReadIsOpen(t *testing.T) {
 	}
 
 	fillWALPastLimit(ctx, t, db)
+	setLongBusyTimeout(ctx, t, db)
 
 	started := time.Now()
 

internal/server/handlers_internal_test.go

@@ -1397,8 +1397,11 @@ func TestIPLookupKeepsRouteWhenPTRLookupDoesNotAnswer(t *testing.T) {
 	s := newServerWithTestRoutes(t)
 	s.resolver = &fakeResolver{ptrNeverAnswers: true}
 
-	// The request's time limit passes after the PTR lookup's.
+	// The request's time limit passes well after the PTR lookup's, so that a
-	ctx, cancel := context.WithTimeout(t.Context(), 2*ptrLookupTimeout)
+	// stall on a busy host does not use up the difference.
+	const requestTimeLimit = ptrLookupTimeout + 15*time.Second
+
+	ctx, cancel := context.WithTimeout(t.Context(), requestTimeLimit)
 	defer cancel()
 
 	rec := httptest.NewRecorder()
@@ -2226,9 +2229,13 @@ func settledGoroutineCount() int {
 	return prev
 }
 
-// waitForGoroutines waits until the goroutine count drops to target or below.
+// waitForGoroutines waits up to 15 seconds for the goroutine count to drop to
+// target or below.
 func waitForGoroutines(target int) bool {
-	for range 100 {
+	const maxWait = 15 * time.Second
+
+	deadline := time.Now().Add(maxWait)
+	for time.Now().Before(deadline) {
 		if runtime.NumGoroutine() <= target {
 			return true
 		}

internal/server/home_internal_test.go

@@ -279,7 +279,7 @@ func TestKeepHomeCardsFreshWorksThemOutEveryInterval(t *testing.T) {
 
 	const (
 		interval = 10 * time.Millisecond
-		maxWait  = 5 * time.Second
+		maxWait  = 15 * time.Second
 	)
 
 	table := emptyTable()
@@ -321,7 +321,7 @@ func TestKeepHomeCardsFreshWorksThemOutEveryInterval(t *testing.T) {
 func TestStartedServerWorksOutTheCards(t *testing.T) {
 	t.Parallel()
 
-	const maxWait = 5 * time.Second
+	const maxWait = 15 * time.Second
 
 	s := newServerWithTable(t, newCardsTable())
 

internal/server/server_internal_test.go

@@ -22,10 +22,10 @@ func TestStopClosesTheConnectionOfARequestStillRunning(t *testing.T) {
 		stopDeadline = 100 * time.Millisecond
 		// waitTimeout is how long the test waits for the request to reach
 		// the handler, and then for the client to see its connection closed.
-		waitTimeout = 2 * time.Second
+		waitTimeout = 15 * time.Second
 		// readHeaderTimeout bounds the test server's read of the request
 		// header, as every http.Server must.
-		readHeaderTimeout = time.Second
+		readHeaderTimeout = waitTimeout
 	)
 
 	release := make(chan struct{})

internal/streamer/streamer_internal_test.go

@@ -14,6 +14,7 @@ import (
 	"strings"
 	"sync/atomic"
 	"testing"
+	"testing/synctest"
 	"time"
 
 	"git.eeqj.de/sneak/routewatch/internal/logger"
@@ -398,7 +399,7 @@ func TestStopQueuesTheHeldStateMessages(t *testing.T) {
 	const (
 		held    = 2
 		lines   = held + 1
-		maxWait = 5 * time.Second
+		maxWait = 15 * time.Second
 	)
 
 	srv := httptest.NewServer(http.HandlerFunc(
@@ -478,15 +479,15 @@ func TestStopQueuesTheHeldStateMessages(t *testing.T) {
 // TestStopReturnsByItsDeadline stops a streamer whose handler has a full queue
 // 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
+// within its deadline plus a margin instead of waiting for room on the queue
-// queue and for the worker, and log as not handed over the held message and
+// and for the worker, and log as not handed over the held message and every
-// every message on the queue.
+// message on the queue.
 func TestStopReturnsByItsDeadline(t *testing.T) {
 	t.Parallel()
 
 	const (
 		deadline = 100 * time.Millisecond
-		margin   = 2 * time.Second
+		margin   = 15 * time.Second
 	)
 
 	var logs bytes.Buffer
@@ -610,10 +611,33 @@ func TestReconnectWhenServerNeverAnswers(t *testing.T) {
 	checkReconnect(t, neverAnswer, idleTimeout)
 }
 
-// TestStreamNotDroppedWhileLinesArrive has the server send lines well within
+// pipeTransport answers every request with a response whose body write fills.
-// the idle timeout for several times the timeout, then end the response. Each
+// The body fails when the request's context ends, as a connection's does.
-// line must start the idle timeout over, so the stream lasts until the server
+type pipeTransport struct {
-// ends it instead of being dropped as idle.
+	write func(io.Writer)
+}
+
+func (p pipeTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	body, w := io.Pipe()
+
+	context.AfterFunc(req.Context(), func() {
+		_ = w.CloseWithError(req.Context().Err())
+	})
+
+	go func() {
+		p.write(w)
+
+		_ = w.Close()
+	}()
+
+	return &http.Response{StatusCode: http.StatusOK, Body: body}, nil
+}
+
+// TestStreamNotDroppedWhileLinesArrive has a response send lines well within
+// the idle timeout for several times the timeout, then end. Each line must
+// start the idle timeout over, so the stream lasts until the response ends
+// instead of being dropped as idle. It runs in a synctest bubble, whose clock
+// is fake, so a stall on the host does not use up the timeout.
 func TestStreamNotDroppedWhileLinesArrive(t *testing.T) {
 	t.Parallel()
 
@@ -623,26 +647,28 @@ func TestStreamNotDroppedWhileLinesArrive(t *testing.T) {
 		sendFor      = 5 * idleTimeout
 	)
 
-	srv := httptest.NewServer(http.HandlerFunc(
+	// The streamer is made outside the bubble. Its metrics start a goroutine
-		func(w http.ResponseWriter, _ *http.Request) {
+	// that never returns, and a bubble waits for each of its goroutines to
+	// return.
+	s := New(logger.New(), metrics.New())
+	s.idleTimeout = idleTimeout
+	s.client = &http.Client{Transport: pipeTransport{
+		write: func(w io.Writer) {
 			deadline := time.Now().Add(sendFor)
 			for time.Now().Before(deadline) {
 				_, _ = io.WriteString(w, updateLine)
 
-				flush(w)
 				time.Sleep(lineInterval)
 			}
-		}))
+		},
-	defer srv.Close()
+	}}
 
-	s := New(logger.New(), metrics.New())
+	synctest.Test(t, func(t *testing.T) {
-	s.url = srv.URL
+		err := s.stream(context.Background())
-	s.idleTimeout = idleTimeout
+		if !errors.Is(err, errStreamEnded) {
-
+			t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
-	err := s.stream(context.Background())
+		}
-	if !errors.Is(err, errStreamEnded) {
+	})
-		t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
-	}
 }
 
 // TestStreamReadsLongLine has the server send one UPDATE line of about
@@ -706,7 +732,7 @@ func TestHandlerStatsReadWhileMessagesAreDropped(t *testing.T) {
 	const (
 		lines       = 5
 		wantDropped = lines - 2
-		maxWait     = 5 * time.Second
+		maxWait     = 15 * time.Second
 	)
 
 	srv := httptest.NewServer(http.HandlerFunc(
@@ -825,7 +851,7 @@ func checkReconnect(
 ) {
 	t.Helper()
 
-	const maxWait = 5 * time.Second
+	const maxWait = 15 * time.Second
 
 	tracker := metrics.New()
 	connectedAtReconnect := make(chan bool, 1)
@@ -896,9 +922,13 @@ func settledGoroutineCount() int {
 	return prev
 }
 
-// waitForGoroutines waits until the goroutine count drops to target or below.
+// waitForGoroutines waits up to 15 seconds for the goroutine count to drop to
+// target or below.
 func waitForGoroutines(target int) bool {
-	for range 100 {
+	const maxWait = 15 * time.Second
+
+	deadline := time.Now().Add(maxWait)
+	for time.Now().Before(deadline) {
 		if runtime.NumGoroutine() <= target {
 			return true
 		}