From 368fca99556f74a0b42bfe8f2c409ae11e0ed1e4 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, 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 --- 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 | 180 ++++++++++------- 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 | 203 +++++++++++++------- 10 files changed, 337 insertions(+), 187 deletions(-) diff --git a/TODO.md b/TODO.md index 7e90229..ad7e8c7 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: the IP lookup checks the public VPN server lists of Mullvad, NordVPN, Private Internet Access and IVPN, and X4BNet's community list. Each result has `flags.is_vpn` and `vpn_providers` (closes 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 82850ea..663ea8b 100644 --- a/internal/iplist/iplist_internal_test.go +++ b/internal/iplist/iplist_internal_test.go @@ -646,9 +646,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 2b3c07f..c8f1906 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,82 +442,89 @@ 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 // duration of a stream that connected 90 seconds ago are both 90. A stream // that has never connected has an uptime of 0, and one that is not connected -// has no connection duration. +// has no connection duration. The test runs in a synctest bubble, whose clock +// is fake, so a stall on the host does not add to the 90 seconds. func TestStreamDurationsCountSeconds(t *testing.T) { t.Parallel() - const connectedFor = 90 * time.Second + synctest.Test(t, func(t *testing.T) { + const connectedFor = 90 * time.Second - connectedSince := time.Now().Add(-connectedFor) - want := int64(connectedFor / time.Second) + connectedSince := time.Now().Add(-connectedFor) + want := int64(connectedFor / time.Second) - if got := uptimeSeconds(connectedSince); got != want { - t.Errorf("uptimeSeconds = %d, want %d", got, want) - } + if got := uptimeSeconds(connectedSince); got != want { + t.Errorf("uptimeSeconds = %d, want %d", got, want) + } - if got := uptimeSeconds(time.Time{}); got != 0 { - t.Errorf("uptimeSeconds before any connection = %d, want 0", got) - } + if got := uptimeSeconds(time.Time{}); got != 0 { + t.Errorf("uptimeSeconds before any connection = %d, want 0", got) + } - switch got := connectionDurationSeconds(true, connectedSince); { - case got == nil: - t.Errorf("connectionDurationSeconds = nil, want %d", want) - case *got != want: - t.Errorf("connectionDurationSeconds = %d, want %d", *got, want) - } + switch got := connectionDurationSeconds(true, connectedSince); { + case got == nil: + t.Errorf("connectionDurationSeconds = nil, want %d", want) + case *got != want: + t.Errorf("connectionDurationSeconds = %d, want %d", *got, want) + } - if got := connectionDurationSeconds(false, connectedSince); got != nil { - t.Errorf("connectionDurationSeconds while not connected = %d, want nil", - *got) - } + if got := connectionDurationSeconds(false, connectedSince); got != nil { + t.Errorf("connectionDurationSeconds while not connected = %d, want nil", + *got) + } + }) } // newTestTable returns a routing table that holds a route for testPrefixV4, @@ -1134,7 +1168,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 +1191,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 +1436,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() @@ -2367,9 +2409,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..c7a88ac 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 @@ -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 }