diff --git a/TODO.md b/TODO.md index e3c6751..748dc91 100644 --- a/TODO.md +++ b/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 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/blocklist/blocklist_internal_test.go b/internal/blocklist/blocklist_internal_test.go index 0ce9770..0c2c4ec 100644 --- a/internal/blocklist/blocklist_internal_test.go +++ b/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) } 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/server/handlers_internal_test.go b/internal/server/handlers_internal_test.go index d2f4654..6a6198c 100644 --- a/internal/server/handlers_internal_test.go +++ b/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. - 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() @@ -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 } diff --git a/internal/server/home_internal_test.go b/internal/server/home_internal_test.go index 2cfcf30..5ea27d0 100644 --- a/internal/server/home_internal_test.go +++ b/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()) 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..3191ae4 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,33 @@ 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. +type pipeTransport struct { + 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( - 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(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()) - 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 +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 }