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 reconnect tests, 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 reconnect tests also use a fake transport, so
a stall cannot reorder their requests.

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
@@ -577,7 +578,7 @@ func TestStopBeforeMessageReachesQueues(t *testing.T) {
 func TestReconnectWhenStreamEnds(t *testing.T) {
 	t.Parallel()
 
-	endResponse := func(http.ResponseWriter, *http.Request) {}
+	endResponse := pipeTransport{write: func(context.Context, io.Writer) {}}
 	checkReconnect(t, endResponse, streamIdleTimeout)
 }
 
@@ -589,9 +590,8 @@ func TestReconnectWhenStreamGoesIdle(t *testing.T) {
 
 	const idleTimeout = 50 * time.Millisecond
 
-	holdResponse := func(w http.ResponseWriter, r *http.Request) {
-		flush(w)
-		<-r.Context().Done()
+	holdResponse := pipeTransport{
+		write: func(ctx context.Context, _ io.Writer) { <-ctx.Done() },
 	}
 	checkReconnect(t, holdResponse, idleTimeout)
 }
@@ -604,16 +604,51 @@ func TestReconnectWhenServerNeverAnswers(t *testing.T) {
 
 	const idleTimeout = 50 * time.Millisecond
 
-	neverAnswer := func(_ http.ResponseWriter, r *http.Request) {
-		<-r.Context().Done()
-	}
+	neverAnswer := roundTripFunc(
+		func(req *http.Request) (*http.Response, error) {
+			<-req.Context().Done()
+
+			return nil, req.Context().Err()
+		},
+	)
 	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
+}
+
+// roundTripFunc makes a function an http.RoundTripper.
+type roundTripFunc func(*http.Request) (*http.Response, error)
+
+func (f roundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
+	return f(req)
+}
+
+// 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 +658,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 +750,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(
@@ -820,61 +864,70 @@ func streamLines(t *testing.T, s *Streamer, lines string) {
 // line and then stays open. The streamer must come back for a second
 // response, must not be marked connected when it does, and must handle the
 // UPDATE in it.
+//
+// It runs in a synctest bubble, whose clock is fake, so a stall on the host
+// does not use up the idle timeout. The fake server gets the requests in the
+// order the streamer makes them, because no real connection carries them.
 func checkReconnect(
-	t *testing.T, firstResponse http.HandlerFunc, idleTimeout time.Duration,
+	t *testing.T, firstResponse http.RoundTripper, idleTimeout time.Duration,
 ) {
 	t.Helper()
 
-	const maxWait = 5 * time.Second
+	const maxWait = 15 * time.Second
 
+	// The tracker is made outside the bubble. It starts a goroutine that never
+	// returns, and a bubble waits for each of its goroutines to return.
 	tracker := metrics.New()
-	connectedAtReconnect := make(chan bool, 1)
 
-	var requests atomic.Int32
+	synctest.Test(t, func(t *testing.T) {
+		connectedAtReconnect := make(chan bool, 1)
+		laterResponse := pipeTransport{
+			write: func(ctx context.Context, w io.Writer) {
+				_, _ = io.WriteString(w, updateLine)
 
-	srv := httptest.NewServer(http.HandlerFunc(
-		func(w http.ResponseWriter, r *http.Request) {
-			if requests.Add(1) == 1 {
-				firstResponse(w, r)
+				<-ctx.Done()
+			},
+		}
 
-				return
-			}
+		var requests atomic.Int32
 
-			select {
-			case connectedAtReconnect <- tracker.IsConnected():
-			default:
-			}
+		s := New(logger.New(), tracker)
+		s.client = &http.Client{Transport: roundTripFunc(
+			func(req *http.Request) (*http.Response, error) {
+				if requests.Add(1) == 1 {
+					return firstResponse.RoundTrip(req)
+				}
 
-			_, _ = io.WriteString(w, updateLine)
+				select {
+				case connectedAtReconnect <- tracker.IsConnected():
+				default:
+				}
 
-			flush(w)
-			<-r.Context().Done()
-		}))
-	defer srv.Close()
+				return laterResponse.RoundTrip(req)
+			},
+		)}
+		s.idleTimeout = idleTimeout
+		s.minBackoff = time.Millisecond
+		handler := updateHandler{handled: make(chan struct{}, 1)}
+		s.RegisterHandler(handler)
 
-	s := New(logger.New(), tracker)
-	s.url = srv.URL
-	s.idleTimeout = idleTimeout
-	s.minBackoff = time.Millisecond
-	handler := updateHandler{handled: make(chan struct{}, 1)}
-	s.RegisterHandler(handler)
+		err := s.Start(t.Context())
+		if err != nil {
+			t.Fatalf("Start returned error: %v", err)
+		}
+		defer s.Stop(t.Context())
 
-	err := s.Start(t.Context())
-	if err != nil {
-		t.Fatalf("Start returned error: %v", err)
-	}
-	defer s.Stop(t.Context())
+		select {
+		case <-handler.handled:
+		case <-time.After(maxWait):
+			t.Fatalf("no message from a second response handled within %s", maxWait)
+		}
 
-	select {
-	case <-handler.handled:
-	case <-time.After(maxWait):
-		t.Fatalf("no message from a second response handled within %s", maxWait)
-	}
-
-	if <-connectedAtReconnect {
-		t.Error("streamer was still marked connected when it came back " +
-			"for a second response")
-	}
+		if <-connectedAtReconnect {
+			t.Error("streamer was still marked connected when it came back " +
+				"for a second response")
+		}
+	})
 }
 
 // settledGoroutineCount lets transient goroutines finish, then reports the
@@ -896,9 +949,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
 		}