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, the tests whose requests pass a 2- or
4-second limit in the server, and TestStreamDurationsCountSeconds run on a fake
clock, so a stall on the host cannot change their result.

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

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
-// queue and for the worker, and log as not handed over the held message and
-// every message on the queue.
+// within its deadline plus a margin instead of waiting for room on the 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()
 
 	const (
 		deadline = 100 * time.Millisecond
-		margin   = 2 * time.Second
+		margin   = 15 * time.Second
 	)
 
 	var logs bytes.Buffer
@@ -610,10 +611,34 @@ func TestReconnectWhenServerNeverAnswers(t *testing.T) {
 	checkReconnect(t, neverAnswer, idleTimeout)
 }
 
-// TestStreamNotDroppedWhileLinesArrive has the server send lines well within
-// the idle timeout for several times the timeout, then end the response. Each
-// line must start the idle timeout over, so the stream lasts until the server
-// ends it instead of being dropped as idle.
+// pipeTransport answers every request with a response whose body write fills.
+// The body fails when the request's context ends, as a connection's does.
+// write gets that context, so that it can end when the request does.
+type pipeTransport struct {
+	write func(context.Context, 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(ctx context.Context) {
+		p.write(ctx, w)
+
+		_ = w.Close()
+	}(req.Context())
+
+	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 +648,35 @@ func TestStreamNotDroppedWhileLinesArrive(t *testing.T) {
 		sendFor      = 5 * idleTimeout
 	)
 
-	srv := httptest.NewServer(http.HandlerFunc(
-		func(w http.ResponseWriter, _ *http.Request) {
+	// The streamer is made outside the bubble. Its metrics start a goroutine
+	// 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(ctx context.Context, w io.Writer) {
 			deadline := time.Now().Add(sendFor)
 			for time.Now().Before(deadline) {
 				_, _ = io.WriteString(w, updateLine)
 
-				flush(w)
-				time.Sleep(lineInterval)
+				// The wait ends with the stream. A bubble's clock stops when
+				// the test function returns, and a goroutine then still
+				// asleep panics the test binary.
+				select {
+				case <-ctx.Done():
+					return
+				case <-time.After(lineInterval):
+				}
 			}
-		}))
-	defer srv.Close()
+		},
+	}}
 
-	s := New(logger.New(), metrics.New())
-	s.url = srv.URL
-	s.idleTimeout = idleTimeout
-
-	err := s.stream(context.Background())
-	if !errors.Is(err, errStreamEnded) {
-		t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
-	}
+	synctest.Test(t, func(t *testing.T) {
+		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 +740,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 +859,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 +930,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
 		}