From cf30c7460ce09e98687e2e6b2b99e392efe6c342 Mon Sep 17 00:00:00 2001 From: sneak Date: Tue, 10 Nov 2026 09:25:56 +0000 Subject: [PATCH] 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, and the tests whose requests pass a 2- or 4-second limit in the server, run on a fake clock, so a stall on the host cannot use up the limit. 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 | 3 + cmd/asinfo-gen/main_test.go | 2 +- internal/database/database_internal_test.go | 30 ++++- internal/iplist/iplist_internal_test.go | 6 +- internal/server/handlers_internal_test.go | 137 +++++++++++++------- internal/server/home_internal_test.go | 12 +- internal/server/middleware_internal_test.go | 71 +++++----- internal/server/routes_internal_test.go | 13 +- internal/server/server_internal_test.go | 4 +- internal/streamer/streamer_internal_test.go | 92 +++++++++---- 10 files changed, 249 insertions(+), 121 deletions(-) diff --git a/TODO.md b/TODO.md index 39f3c2e..f9544c2 100644 --- a/TODO.md +++ b/TODO.md @@ -53,6 +53,9 @@ problem the run finds. # Completed Steps +- 2026-11-10: the tests no longer fail when a busy host stalls for several + seconds. Most such waits are now 15 seconds. The test of the stream's idle + timeout runs on a fake clock (closes #240) - 2026-11-10: each result of the IP lookup has a `flags` object with `is_tor` and `is_bogon`. The public IP lists are named `ip_lists` in the answer, `internal/iplist` in the code and `iplists` in the state directory. A list diff --git a/cmd/asinfo-gen/main_test.go b/cmd/asinfo-gen/main_test.go index f629989..e11ab05 100644 --- a/cmd/asinfo-gen/main_test.go +++ b/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 diff --git a/internal/database/database_internal_test.go b/internal/database/database_internal_test.go index f767034..ecb7c68 100644 --- a/internal/database/database_internal_test.go +++ b/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 - // timeout, which a checkpoint that waits for the reader takes. - readerCheckpointLimit = time.Second + // waiting for it, may take while a reader is open: well under + // longBusyTimeoutMs, with room for a stall on a busy host. + 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() diff --git a/internal/iplist/iplist_internal_test.go b/internal/iplist/iplist_internal_test.go index 6e626f7..89b9a4c 100644 --- a/internal/iplist/iplist_internal_test.go +++ b/internal/iplist/iplist_internal_test.go @@ -380,9 +380,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) } diff --git a/internal/server/handlers_internal_test.go b/internal/server/handlers_internal_test.go index b3ceea2..b21490f 100644 --- a/internal/server/handlers_internal_test.go +++ b/internal/server/handlers_internal_test.go @@ -17,6 +17,7 @@ import ( "strings" "sync/atomic" "testing" + "testing/synctest" "time" "git.eeqj.de/sneak/routewatch/internal/config" @@ -173,6 +174,9 @@ type tableStats struct { // carried 203.0.113.0/24, the two with kept routes withdrew it, so its origin // keeps no route. The oldest and newest routes are those of the other two // prefixes. +// +// The requests run in a synctest bubble, whose clock is fake, so a stall on +// the host does not use up the handlers' 4-second limit. func TestStatsHandlersAnswerFromMemory(t *testing.T) { t.Parallel() @@ -218,7 +222,11 @@ func TestStatsHandlersAnswerFromMemory(t *testing.T) { for name, handler := range handlers { rec := httptest.NewRecorder() - handler(rec, httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil)) + + synctest.Test(t, func(t *testing.T) { + handler(rec, + httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil)) + }) if rec.Code != http.StatusOK { t.Errorf("%s: status %d, want %d; body %s", @@ -253,15 +261,20 @@ func TestStatsHandlersAnswerFromMemory(t *testing.T) { // TestStatsAnswerCarriesChurnPerMinute checks that the stream object of the // /api/v1/stats answer holds route_churn_per_min as a whole number, and no // longer route_churn_per_sec, the moving average it replaces -// (https://git.eeqj.de/sneak/routewatch/issues/160). +// (https://git.eeqj.de/sneak/routewatch/issues/160). The request runs in a +// synctest bubble, whose clock is fake, so a stall on the host does not use up +// the handler's 4-second limit. func TestStatsAnswerCarriesChurnPerMinute(t *testing.T) { t.Parallel() s := newServerWithTestRoutes(t) rec := httptest.NewRecorder() - s.handleStats()(rec, - httptest.NewRequestWithContext(t.Context(), http.MethodGet, statsPath, nil)) + + synctest.Test(t, func(t *testing.T) { + s.handleStats()(rec, + httptest.NewRequestWithContext(t.Context(), http.MethodGet, statsPath, nil)) + }) if rec.Code != http.StatusOK { t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body) @@ -309,18 +322,26 @@ func (heldBackHandler) HeldBackCount() int { return heldBackMess // TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack registers a handler // that holds back messages. Its figures in /api/v1/stats, which the status -// page shows, must give how many. +// page shows, must give how many. The request runs in a synctest bubble, whose +// clock is fake, so a stall on the host does not use up the handler's 4-second +// limit. func TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack(t *testing.T) { t.Parallel() s := newServerWithTestRoutes(t) s.streamer.RegisterHandler(heldBackHandler{}) + var answer []byte + + synctest.Test(t, func(t *testing.T) { + answer = get(t, s, statsPath).Body.Bytes() + }) + var body struct { Data StatsResponse `json:"data"` } - err := json.Unmarshal(get(t, s, statsPath).Body.Bytes(), &body) + err := json.Unmarshal(answer, &body) if err != nil { t.Fatalf("decoding the answer: %v", err) } @@ -364,7 +385,9 @@ type unitStats struct { // names. They used to be strings with the unit in them, such as "1h18m34s" or // "1.1 GB" (https://git.eeqj.de/sneak/routewatch/issues/199). The stream has // never connected, so the uptime is 0. The memory use in /api/v1/stats is the -// same number of bytes as its gc.heap_alloc_bytes. +// same number of bytes as its gc.heap_alloc_bytes. The requests run in a +// synctest bubble, whose clock is fake, so a stall on the host does not use up +// the handlers' 4-second limit. func TestStatsAnswersGiveUnitsInFieldNames(t *testing.T) { t.Parallel() @@ -374,7 +397,11 @@ func TestStatsAnswersGiveUnitsInFieldNames(t *testing.T) { s.SetASNFetcher(fixedIntervalASNFetcher{interval: fetchInterval}) for _, path := range []string{statusJSONPath, statsPath} { - answer := get(t, s, path).Body.Bytes() + var answer []byte + + synctest.Test(t, func(t *testing.T) { + answer = get(t, s, path).Body.Bytes() + }) var body struct { Data unitStats `json:"data"` @@ -415,49 +442,53 @@ func TestStatsAnswersGiveUnitsInFieldNames(t *testing.T) { // connection_duration_seconds of /api/v1/stats is null while the stream is not // connected, and a whole number of seconds once it is. It used to be the // string "disconnected", or a Go duration such as "1h18m34s" -// (https://git.eeqj.de/sneak/routewatch/issues/199). +// (https://git.eeqj.de/sneak/routewatch/issues/199). The test runs in a +// synctest bubble, whose clock is fake, so a stall on the host does not use up +// the handler's 4-second limit. func TestStatsAnswerGivesConnectionDurationInSeconds(t *testing.T) { t.Parallel() s := newServerWithTestRoutes(t) - connectionDuration := func() json.RawMessage { - answer := get(t, s, statsPath).Body.Bytes() + synctest.Test(t, func(t *testing.T) { + connectionDuration := func() json.RawMessage { + answer := get(t, s, statsPath).Body.Bytes() - var body struct { - Data map[string]json.RawMessage `json:"data"` + var body struct { + Data map[string]json.RawMessage `json:"data"` + } + + err := json.Unmarshal(answer, &body) + if err != nil { + t.Fatalf("decoding the answer: %v; answer %s", err, answer) + } + + duration, ok := body.Data["connection_duration_seconds"] + if !ok { + t.Fatalf("no connection_duration_seconds: %s", answer) + } + + return duration } - err := json.Unmarshal(answer, &body) - if err != nil { - t.Fatalf("decoding the answer: %v; answer %s", err, answer) + disconnected := connectionDuration() + if string(disconnected) != "null" { + t.Errorf("disconnected: connection_duration_seconds = %s, want null", + disconnected) } - duration, ok := body.Data["connection_duration_seconds"] - if !ok { - t.Fatalf("no connection_duration_seconds: %s", answer) + s.streamer.GetMetricsTracker().SetConnected(true) + + var seconds int64 + + connected := connectionDuration() + + err := json.Unmarshal(connected, &seconds) + if err != nil || seconds != 0 { + t.Errorf("just connected: connection_duration_seconds = %s, want 0", + connected) } - - return duration - } - - disconnected := connectionDuration() - if string(disconnected) != "null" { - t.Errorf("disconnected: connection_duration_seconds = %s, want null", - disconnected) - } - - s.streamer.GetMetricsTracker().SetConnected(true) - - var seconds int64 - - connected := connectionDuration() - - err := json.Unmarshal(connected, &seconds) - if err != nil || seconds != 0 { - t.Errorf("just connected: connection_duration_seconds = %s, want 0", - connected) - } + }) } // TestStreamDurationsCountSeconds checks that the uptime and the connection @@ -1134,7 +1165,9 @@ func (failingDB) GetPrefixFirstSeen( // with the driver's error text. It also sends a prefix that does not parse. // No answer may carry 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). +// (https://git.eeqj.de/sneak/routewatch/issues/108). The requests run in a +// synctest bubble, whose clock is fake, so a stall on the host does not use up +// the stats handlers' 4-second limit. func TestJSONRoutesDoNotSendErrorDetails(t *testing.T) { t.Parallel() @@ -1155,8 +1188,11 @@ func TestJSONRoutesDoNotSendErrorDetails(t *testing.T) { for _, tt := range tests { rec := httptest.NewRecorder() - s.router.ServeHTTP(rec, httptest.NewRequestWithContext( - t.Context(), http.MethodGet, tt.path, nil)) + + synctest.Test(t, func(t *testing.T) { + 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", @@ -1397,8 +1433,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. - ctx, cancel := context.WithTimeout(t.Context(), 2*ptrLookupTimeout) + // The request's time limit passes well after the PTR lookup's, so that a + // 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() @@ -2287,9 +2326,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 } diff --git a/internal/server/home_internal_test.go b/internal/server/home_internal_test.go index 2cfcf30..40f1b01 100644 --- a/internal/server/home_internal_test.go +++ b/internal/server/home_internal_test.go @@ -76,7 +76,9 @@ func asLink(asn int) string { } // TestHomePageCardsShowTheTableFigures fills a routing table and the asns -// table, works out the cards, and checks that each card shows its figure. +// table, works out the cards, and checks that each card shows its figure. The +// cards are worked out in a synctest bubble, whose clock is fake, so a stall on +// the host does not use up the 4-second limit on their database read. func TestHomePageCardsShowTheTableFigures(t *testing.T) { t.Parallel() @@ -95,7 +97,9 @@ func TestHomePageCardsShowTheTableFigures(t *testing.T) { t.Fatalf("GetOrCreateASNBatch: %v", err) } - s.refreshHomeCards(t.Context()) + synctest.Test(t, func(t *testing.T) { + s.refreshHomeCards(t.Context()) + }) page := get(t, s, "/").Body.String() @@ -279,7 +283,7 @@ func TestKeepHomeCardsFreshWorksThemOutEveryInterval(t *testing.T) { const ( interval = 10 * time.Millisecond - maxWait = 5 * time.Second + maxWait = 15 * time.Second ) table := emptyTable() @@ -321,7 +325,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()) diff --git a/internal/server/middleware_internal_test.go b/internal/server/middleware_internal_test.go index 57f82ad..8cb0c82 100644 --- a/internal/server/middleware_internal_test.go +++ b/internal/server/middleware_internal_test.go @@ -14,6 +14,7 @@ import ( "regexp" "strings" "testing" + "testing/synctest" "time" "git.eeqj.de/sneak/routewatch/internal/config" @@ -163,6 +164,10 @@ func TestJSONAnswersPutNoUnitInAString(t *testing.T) { // router and the middleware write. Before the fix a panic in a handler, and a // method /status.json or the health check does not take, answered with an // empty body (https://git.eeqj.de/sneak/routewatch/issues/201). +// +// The requests run in a synctest bubble, whose clock is fake, so a stall on +// the host does not use up the 2-second limit of the health check or the +// 4-second limit of the stats handlers. func TestJSONAnswersCarryMeta(t *testing.T) { t.Parallel() @@ -181,6 +186,10 @@ func TestJSONAnswersCarryMeta(t *testing.T) { s.router.Get(panicsPath, JSONValidationMiddleware(http.HandlerFunc( func(http.ResponseWriter, *http.Request) { panic(errDiskIO) })).ServeHTTP) + failing := New(failingDB{}, newTestTable(), + streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{}) + failing.resolver = &fakeResolver{} + tests := []struct { method string path string @@ -216,42 +225,40 @@ func TestJSONAnswersCarryMeta(t *testing.T) { {http.MethodGet, panicsPath, http.StatusInternalServerError}, } - for _, tt := range tests { - checkAnswerCarriesMeta(t, s, httptest.NewRequestWithContext( - t.Context(), tt.method, tt.path, nil), tt.wantCode) - } + synctest.Test(t, func(t *testing.T) { + for _, tt := range tests { + checkAnswerCarriesMeta(t, s, httptest.NewRequestWithContext( + t.Context(), tt.method, tt.path, nil), tt.wantCode) + } - // A request whose time limit has already passed answers 504. The handlers - // write the answers of the first five routes, and the JSON check on the - // route writes the last. - pastTimeLimit, cancel := context.WithDeadline(t.Context(), time.Now()) - defer cancel() + // A request whose time limit has already passed answers 504. The + // handlers write the answers of the first five routes, and the JSON + // check on the route writes the last. + pastTimeLimit, cancel := context.WithDeadline(t.Context(), time.Now()) + defer cancel() - for _, path := range []string{ - statusJSONPath, statsPath, "/ip/" + addrInTestPrefixV4, - "/api/v1/ip/" + addrInTestPrefixV4, asJSONPath, writesNothingPath, - } { - checkAnswerCarriesMeta(t, s, httptest.NewRequestWithContext( - pastTimeLimit, http.MethodGet, path, nil), http.StatusGatewayTimeout) - } + for _, path := range []string{ + statusJSONPath, statsPath, "/ip/" + addrInTestPrefixV4, + "/api/v1/ip/" + addrInTestPrefixV4, asJSONPath, writesNothingPath, + } { + checkAnswerCarriesMeta(t, s, httptest.NewRequestWithContext( + pastTimeLimit, http.MethodGet, path, nil), http.StatusGatewayTimeout) + } - // With a store whose reads fail, the health check answers 503, and the - // handlers of the other routes 500. - failing := New(failingDB{}, newTestTable(), - streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{}) - failing.resolver = &fakeResolver{} - - checkAnswerCarriesMeta(t, failing, httptest.NewRequestWithContext( - t.Context(), http.MethodGet, healthCheckPath, nil), - http.StatusServiceUnavailable) - - for _, path := range []string{ - statusJSONPath, statsPath, asJSONPath, - "/api/v1/ip/" + addrInTestPrefixV4, "/ip/" + addrInTestPrefixV4, - } { + // With a store whose reads fail, the health check answers 503, and + // the handlers of the other routes 500. checkAnswerCarriesMeta(t, failing, httptest.NewRequestWithContext( - t.Context(), http.MethodGet, path, nil), http.StatusInternalServerError) - } + t.Context(), http.MethodGet, healthCheckPath, nil), + http.StatusServiceUnavailable) + + for _, path := range []string{ + statusJSONPath, statsPath, asJSONPath, + "/api/v1/ip/" + addrInTestPrefixV4, "/ip/" + addrInTestPrefixV4, + } { + checkAnswerCarriesMeta(t, failing, httptest.NewRequestWithContext( + t.Context(), http.MethodGet, path, nil), http.StatusInternalServerError) + } + }) } // checkAnswerCarriesMeta serves req through the server's router, and checks diff --git a/internal/server/routes_internal_test.go b/internal/server/routes_internal_test.go index 01f75ee..118a156 100644 --- a/internal/server/routes_internal_test.go +++ b/internal/server/routes_internal_test.go @@ -5,6 +5,7 @@ import ( "net/http" "net/http/httptest" "testing" + "testing/synctest" ) // TestRoutesAnswerHEAD sends HEAD for a page and JSON routes through the @@ -12,15 +13,23 @@ import ( // lookup of the caller's own address, are in the list because chi's GetHead // does not reach them. Before the fix every route answered HEAD with 405 // (https://git.eeqj.de/sneak/routewatch/issues/111). +// +// The requests run in a synctest bubble, whose clock is fake, so a stall on +// the host does not use up the stats handlers' 4-second limit. The resolver +// is fake because no time limit would cut off a real DNS lookup in a bubble. func TestRoutesAnswerHEAD(t *testing.T) { t.Parallel() s := newServerWithTestRoutes(t) + s.resolver = &fakeResolver{} for _, path := range []string{"/", statusJSONPath, statsPath, "/ip", "/ip/"} { rec := httptest.NewRecorder() - s.router.ServeHTTP(rec, httptest.NewRequestWithContext( - t.Context(), http.MethodHead, path, nil)) + + synctest.Test(t, func(t *testing.T) { + s.router.ServeHTTP(rec, httptest.NewRequestWithContext( + t.Context(), http.MethodHead, path, nil)) + }) if rec.Code != http.StatusOK { t.Errorf("HEAD %s: status %d, want %d; body %s", diff --git a/internal/server/server_internal_test.go b/internal/server/server_internal_test.go index a5f89bb..1123d48 100644 --- a/internal/server/server_internal_test.go +++ b/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{}) diff --git a/internal/streamer/streamer_internal_test.go b/internal/streamer/streamer_internal_test.go index 7e0b23a..80d4804 100644 --- a/internal/streamer/streamer_internal_test.go +++ b/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 }