Author SHA1 Message Date
sneak 5051aff4dd Resolve PTR names over DNS-over-HTTPS and cache them for the record TTL (closes #238)
check / check (push) Successful in 1m34s
`internal/ptr` asks Google, Cloudflare, AdGuard and Quad9 in turn, 2 seconds in all per address. It keeps answers in memory and writes them through to `cache.db`, opened by the new `internal/cachedb`. A reserved address is not sent to the resolvers.

Judgement calls:
- Each resolver gets an equal share of the time left, so a hung resolver does not use up the 2 seconds.
- The negative TTL is the smaller of the SOA minimum and the SOA's own TTL (RFC 2308).
- Names that are not host names are dropped, as `net.Resolver.LookupAddr` does.

Model: opus-5-5
2026-11-10 11:37:33 +00:00
14 changed files with 207 additions and 514 deletions

No files matched your search

-2
View File
@@ -498,8 +498,6 @@ How RouteWatch keeps the answers:
whose SOA record has a TTL or minimum field of 0. whose SOA record has a TTL or minimum field of 0.
- A failed lookup is not kept, so the next lookup of the address asks again. - A failed lookup is not kept, so the next lookup of the address asks again.
- Every 10 minutes it deletes expired answers from memory and from `cache.db`. - Every 10 minutes it deletes expired answers from memory and from `cache.db`.
- It keeps answers for at most 250,000 addresses, and keeps no answer for a new
address while it holds that many.
## Configuration ## Configuration
-3
View File
@@ -56,9 +56,6 @@ problem the run finds.
- 2026-11-10: the IP lookup finds PTR names over DNS-over-HTTPS and caches them - 2026-11-10: the IP lookup finds PTR names over DNS-over-HTTPS and caches them
for the record's TTL, in memory and in `cache.db` (closes for the record's TTL, in memory and in `cache.db` (closes
https://git.eeqj.de/sneak/routewatch/issues/238). https://git.eeqj.de/sneak/routewatch/issues/238).
- 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, - 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 NordVPN, Private Internet Access and IVPN, and X4BNet's community list. Each
result has `flags.is_vpn` and `vpn_providers` (closes result has `flags.is_vpn` and `vpn_providers` (closes
+1 -1
View File
@@ -12,7 +12,7 @@ import (
// parseTimeLimit is how long a test waits for parseCSV to return before it // parseTimeLimit is how long a test waits for parseCSV to return before it
// fails. // fails.
const parseTimeLimit = 15 * time.Second const parseTimeLimit = 2 * time.Second
// TestParseCSVReadsTodaysLayout parses an as.csv with the country code column // TestParseCSVReadsTodaysLayout parses an as.csv with the country code column
// that the source added after the description. It expects each AS with its // that the source added after the description. It expects each AS with its
+5 -25
View File
@@ -72,34 +72,17 @@ const (
// lockTestPrefix is the prefix the batch writes in the checkpoint tests // lockTestPrefix is the prefix the batch writes in the checkpoint tests
// store. // store.
lockTestPrefix = "192.0.2.0/24" 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 // readerCheckpointLimit is how long a checkpoint, and a batch write
// waiting for it, may take while a reader is open: well under // waiting for it, may take while a reader is open: well under the busy
// longBusyTimeoutMs, with room for a stall on a busy host. // timeout, which a checkpoint that waits for the reader takes.
readerCheckpointLimit = 15 * time.Second readerCheckpointLimit = time.Second
// truncateCheckpointLimit is how long a checkpoint of a WAL file over // truncateCheckpointLimit is how long a checkpoint of a WAL file over
// sqliteWALSizeLimitBytes may take to give up while a reader is open: its // sqliteWALSizeLimitBytes may take to give up while a reader is open: its
// wait for the reader plus readerCheckpointLimit. // wait for the reader plus a second, well under the busy timeout.
truncateCheckpointLimit = truncateBusyTimeoutMs*time.Millisecond + truncateCheckpointLimit = truncateBusyTimeoutMs*time.Millisecond +
readerCheckpointLimit time.Second
) )
// 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 // TestConnectionPoolPragmas holds several pooled connections open at once and
// checks each one carries the per-connection settings from the DSN, plus the // 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 // process-wide heap limits, and sees the new file with auto_vacuum
@@ -596,8 +579,6 @@ func TestCheckpointDoesNotWaitForReader(t *testing.T) {
ctx := context.Background() ctx := context.Background()
ts := time.Now().UTC() ts := time.Now().UTC()
setLongBusyTimeout(ctx, t, db)
err = db.UpdatePrefixesBatch(map[string]time.Time{"198.51.100.0/24": ts}) err = db.UpdatePrefixesBatch(map[string]time.Time{"198.51.100.0/24": ts})
if err != nil { if err != nil {
t.Fatalf("failed to store the prefix the reader reads: %v", err) t.Fatalf("failed to store the prefix the reader reads: %v", err)
@@ -709,7 +690,6 @@ func TestCheckpointEmptiesWALOnceNoReadIsOpen(t *testing.T) {
} }
fillWALPastLimit(ctx, t, db) fillWALPastLimit(ctx, t, db)
setLongBusyTimeout(ctx, t, db)
started := time.Now() started := time.Now()
+1 -5
View File
@@ -646,13 +646,9 @@ func TestRestartLoadsCopiesFromDisk(t *testing.T) {
second.Start(t.Context()) second.Start(t.Context())
second.Stop(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) listings := lookup(t, second, listedAddr)
if len(listings) != 1 || if len(listings) != 1 ||
listings[0].Updated.Sub(downloaded).Abs() > maxUpdatedDifference { listings[0].Updated.Sub(downloaded).Abs() > time.Second {
t.Errorf("after the restart, listings %+v, want one updated at %s", t.Errorf("after the restart, listings %+v, want one updated at %s",
listings, downloaded) listings, downloaded)
} }
+20 -33
View File
@@ -1,6 +1,6 @@
// Package ptr looks up the PTR names of addresses over DNS-over-HTTPS. It // Package ptr looks up the PTR names of addresses over DNS-over-HTTPS. It
// keeps answers in memory for as long as DNS allows, and in cache.db so that // keeps each answer in memory for as long as DNS allows, and in cache.db so
// the answers survive a restart. // that the answers survive a restart.
package ptr package ptr
import ( import (
@@ -29,10 +29,6 @@ const (
// nameSeparator separates the names in a row of the hostnames table. // nameSeparator separates the names in a row of the hostnames table.
nameSeparator = " " nameSeparator = " "
// maxAddresses limits how many addresses the cache holds, so that a public
// lookup cannot grow memory and cache.db without bound.
maxAddresses = 250_000
) )
// Params is what New needs. // Params is what New needs.
@@ -56,16 +52,15 @@ type entry struct {
// Cache answers PTR lookups from memory, and asks the resolvers for an address // Cache answers PTR lookups from memory, and asks the resolvers for an address
// it holds no current answer for. Start loads it from cache.db. // it holds no current answer for. Start loads it from cache.db.
type Cache struct { type Cache struct {
db *sql.DB db *sql.DB
resolvers []string resolvers []string
timeLimit time.Duration timeLimit time.Duration
maxEntries int logger *slog.Logger
logger *slog.Logger client *http.Client
client *http.Client now func() time.Time
now func() time.Time
// entries holds the kept answers by address, at most maxEntries of them. // entries holds every answer by address. pending holds those that
// pending holds those that cache.db does not have yet. // cache.db does not have yet.
mu sync.RWMutex mu sync.RWMutex
entries map[string]entry entries map[string]entry
pending map[string]entry pending map[string]entry
@@ -78,16 +73,15 @@ type Cache struct {
// New returns an empty Cache. Start loads it. // New returns an empty Cache. Start loads it.
func New(params Params) *Cache { func New(params Params) *Cache {
return &Cache{ return &Cache{
db: params.DB, db: params.DB,
resolvers: params.Resolvers, resolvers: params.Resolvers,
timeLimit: LookupTimeout, timeLimit: LookupTimeout,
maxEntries: maxAddresses, logger: params.Logger.With("component", "ptr"),
logger: params.Logger.With("component", "ptr"), client: &http.Client{},
client: &http.Client{}, now: time.Now,
now: time.Now, entries: make(map[string]entry),
entries: make(map[string]entry), pending: make(map[string]entry),
pending: make(map[string]entry), done: make(chan struct{}),
done: make(chan struct{}),
} }
} }
@@ -174,18 +168,11 @@ func (c *Cache) current(address string) (entry, bool) {
return found, true return found, true
} }
// store keeps an answer in memory and queues it for cache.db. Once entries // store keeps an answer in memory and queues it for cache.db.
// holds maxEntries addresses it keeps no new address. It still replaces the
// answer of an address already held.
func (c *Cache) store(address string, answer entry) { func (c *Cache) store(address string, answer entry) {
c.mu.Lock() c.mu.Lock()
defer c.mu.Unlock() defer c.mu.Unlock()
_, held := c.entries[address]
if !held && len(c.entries) >= c.maxEntries {
return
}
c.entries[address] = answer c.entries[address] = answer
c.pending[address] = answer c.pending[address] = answer
} }
-69
View File
@@ -880,72 +880,3 @@ func TestFlushKeepsAnswersItCouldNotWrite(t *testing.T) {
t.Errorf("%d rows after the second write, want 1", got) t.Errorf("%d rows after the second write, want 1", got)
} }
} }
// TestNewAddressIsNotKeptAtTheLimit fills a cache that holds one address. A
// lookup of a second address still returns its answer, and the cache keeps
// nothing for it until the sweep has deleted the first address's expired
// answer.
func TestNewAddressIsNotKeptAtTheLimit(t *testing.T) {
t.Parallel()
db := openTestDB(t, t.TempDir())
resolver := newFakeResolver(t, ptrReply(300, testName))
cache, clock := newTestCache(t, db, resolver)
cache.maxEntries = 1
lookupOK(t, cache, testAddress)
names := lookupOK(t, cache, otherAddress)
if !slices.Equal(names, []string{testName}) {
t.Errorf("names %v at the limit, want %s", names, testName)
}
_, inEntries := cache.entries[otherAddress]
_, inPending := cache.pending[otherAddress]
if inEntries || inPending {
t.Errorf("a new address was kept at the limit: in entries %t, in pending %t",
inEntries, inPending)
}
cache.flush(t.Context())
if got := countRows(t, db); got != 1 {
t.Errorf("%d rows at the limit, want 1", got)
}
clock.advance(300 * time.Second)
cache.sweep(t.Context())
lookupOK(t, cache, otherAddress)
_, inEntries = cache.entries[otherAddress]
if !inEntries {
t.Error("a new address was not kept after the sweep made room")
}
}
// TestHeldAddressIsReplacedAtTheLimit lets the answer of the only address a
// cache holds expire. The new answer replaces it in memory and is queued for
// cache.db.
func TestHeldAddressIsReplacedAtTheLimit(t *testing.T) {
t.Parallel()
resolver := newFakeResolver(t, ptrReply(300, testName))
cache, clock := newTestCache(t, openTestDB(t, t.TempDir()), resolver)
cache.maxEntries = 1
lookupOK(t, cache, testAddress)
clock.advance(300 * time.Second)
resolver.setReply(ptrReply(300, otherName))
lookupOK(t, cache, testAddress)
want := []string{otherName}
if got := cache.entries[testAddress].names; !slices.Equal(got, want) {
t.Errorf("entries holds %v, want %v", got, want)
}
if got := cache.pending[testAddress].names; !slices.Equal(got, want) {
t.Errorf("pending holds %v, want %v", got, want)
}
}
@@ -8,8 +8,6 @@ import (
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"net/netip" "net/netip"
"os"
"path/filepath"
"slices" "slices"
"strconv" "strconv"
"strings" "strings"
@@ -785,74 +783,3 @@ func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
t.Errorf("Shutdown made the calls %q, want %q", store.calls, want) t.Errorf("Shutdown made the calls %q, want %q", store.calls, want)
} }
} }
// TestPTRCacheAnswersTheIPLookup starts the PTR cache and the HTTP server as
// Run does, and looks up a reserved address, which asks no resolver. Without
// the PTR cache the server has no PTR resolver, and every IP lookup answers
// 500.
func TestPTRCacheAnswersTheIPLookup(t *testing.T) {
t.Parallel()
store := newMockStore()
log := logger.New()
port := freePort(t)
cfg := &config.Config{Port: port, StateDir: t.TempDir()}
s := streamer.New(log, metrics.New())
// A started server reads its routing table in the background, so it
// needs one.
rw := New(Dependencies{
DB: store,
Streamer: s,
Server: server.New(store, routetable.New(routetable.Params{}), s, log,
cfg),
Logger: log,
Config: cfg,
})
err := rw.startPTRCache(t.Context())
if err != nil {
t.Fatalf("failed to start the PTR cache: %v", err)
}
defer rw.stopPTRCache(t.Context())
err = rw.server.Start(t.Context())
if err != nil {
t.Fatalf("failed to start the HTTP server: %v", err)
}
defer func() { _ = rw.server.Stop(t.Context()) }()
req, err := http.NewRequestWithContext(t.Context(), http.MethodGet,
"http://"+net.JoinHostPort("127.0.0.1", strconv.Itoa(port))+
"/api/v1/ip/10.0.0.1", nil)
if err != nil {
t.Fatalf("failed to build the request: %v", err)
}
resp, err := http.DefaultClient.Do(req)
if err != nil {
t.Fatalf("the IP lookup failed: %v", err)
}
defer func() { _ = resp.Body.Close() }()
body, err := io.ReadAll(resp.Body)
if err != nil {
t.Fatalf("failed to read the answer: %v", err)
}
if resp.StatusCode != http.StatusOK {
t.Fatalf("status %d, want %d; body %s", resp.StatusCode, http.StatusOK, body)
}
if !strings.Contains(string(body), `"ptr":[]`) {
t.Errorf("answer %s, want an empty ptr", body)
}
_, err = os.Stat(filepath.Join(cfg.StateDir, "cache.db"))
if err != nil {
t.Errorf("cache.db is not in the state directory: %v", err)
}
}
+67 -113
View File
@@ -17,7 +17,6 @@ import (
"strings" "strings"
"sync/atomic" "sync/atomic"
"testing" "testing"
"testing/synctest"
"time" "time"
"git.eeqj.de/sneak/routewatch/internal/config" "git.eeqj.de/sneak/routewatch/internal/config"
@@ -175,9 +174,6 @@ type tableStats struct {
// carried 203.0.113.0/24, the two with kept routes withdrew it, so its origin // 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 // keeps no route. The oldest and newest routes are those of the other two
// prefixes. // 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) { func TestStatsHandlersAnswerFromMemory(t *testing.T) {
t.Parallel() t.Parallel()
@@ -223,11 +219,7 @@ func TestStatsHandlersAnswerFromMemory(t *testing.T) {
for name, handler := range handlers { for name, handler := range handlers {
rec := httptest.NewRecorder() 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 { if rec.Code != http.StatusOK {
t.Errorf("%s: status %d, want %d; body %s", t.Errorf("%s: status %d, want %d; body %s",
@@ -262,20 +254,15 @@ func TestStatsHandlersAnswerFromMemory(t *testing.T) {
// TestStatsAnswerCarriesChurnPerMinute checks that the stream object of the // TestStatsAnswerCarriesChurnPerMinute checks that the stream object of the
// /api/v1/stats answer holds route_churn_per_min as a whole number, and no // /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 // longer route_churn_per_sec, the moving average it replaces
// (https://git.eeqj.de/sneak/routewatch/issues/160). The request runs in a // (https://git.eeqj.de/sneak/routewatch/issues/160).
// 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) { func TestStatsAnswerCarriesChurnPerMinute(t *testing.T) {
t.Parallel() t.Parallel()
s := newServerWithTestRoutes(t) s := newServerWithTestRoutes(t)
rec := httptest.NewRecorder() rec := httptest.NewRecorder()
s.handleStats()(rec,
synctest.Test(t, func(t *testing.T) { httptest.NewRequestWithContext(t.Context(), http.MethodGet, statsPath, nil))
s.handleStats()(rec,
httptest.NewRequestWithContext(t.Context(), http.MethodGet, statsPath, nil))
})
if rec.Code != http.StatusOK { if rec.Code != http.StatusOK {
t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body) t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body)
@@ -323,26 +310,18 @@ func (heldBackHandler) HeldBackCount() int { return heldBackMess
// TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack registers a handler // TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack registers a handler
// that holds back messages. Its figures in /api/v1/stats, which the status // that holds back messages. Its figures in /api/v1/stats, which the status
// page shows, must give how many. The request runs in a synctest bubble, whose // page shows, must give how many.
// clock is fake, so a stall on the host does not use up the handler's 4-second
// limit.
func TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack(t *testing.T) { func TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack(t *testing.T) {
t.Parallel() t.Parallel()
s := newServerWithTestRoutes(t) s := newServerWithTestRoutes(t)
s.streamer.RegisterHandler(heldBackHandler{}) s.streamer.RegisterHandler(heldBackHandler{})
var answer []byte
synctest.Test(t, func(t *testing.T) {
answer = get(t, s, statsPath).Body.Bytes()
})
var body struct { var body struct {
Data StatsResponse `json:"data"` Data StatsResponse `json:"data"`
} }
err := json.Unmarshal(answer, &body) err := json.Unmarshal(get(t, s, statsPath).Body.Bytes(), &body)
if err != nil { if err != nil {
t.Fatalf("decoding the answer: %v", err) t.Fatalf("decoding the answer: %v", err)
} }
@@ -386,9 +365,7 @@ type unitStats struct {
// names. They used to be strings with the unit in them, such as "1h18m34s" or // 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 // "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 // 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. The requests run in a // same number of bytes as its gc.heap_alloc_bytes.
// 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) { func TestStatsAnswersGiveUnitsInFieldNames(t *testing.T) {
t.Parallel() t.Parallel()
@@ -398,11 +375,7 @@ func TestStatsAnswersGiveUnitsInFieldNames(t *testing.T) {
s.SetASNFetcher(fixedIntervalASNFetcher{interval: fetchInterval}) s.SetASNFetcher(fixedIntervalASNFetcher{interval: fetchInterval})
for _, path := range []string{statusJSONPath, statsPath} { for _, path := range []string{statusJSONPath, statsPath} {
var answer []byte answer := get(t, s, path).Body.Bytes()
synctest.Test(t, func(t *testing.T) {
answer = get(t, s, path).Body.Bytes()
})
var body struct { var body struct {
Data unitStats `json:"data"` Data unitStats `json:"data"`
@@ -443,89 +416,82 @@ func TestStatsAnswersGiveUnitsInFieldNames(t *testing.T) {
// connection_duration_seconds of /api/v1/stats is null while the stream is not // 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 // connected, and a whole number of seconds once it is. It used to be the
// string "disconnected", or a Go duration such as "1h18m34s" // string "disconnected", or a Go duration such as "1h18m34s"
// (https://git.eeqj.de/sneak/routewatch/issues/199). The test runs in a // (https://git.eeqj.de/sneak/routewatch/issues/199).
// 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) { func TestStatsAnswerGivesConnectionDurationInSeconds(t *testing.T) {
t.Parallel() t.Parallel()
s := newServerWithTestRoutes(t) s := newServerWithTestRoutes(t)
synctest.Test(t, func(t *testing.T) { connectionDuration := func() json.RawMessage {
connectionDuration := func() json.RawMessage { answer := get(t, s, statsPath).Body.Bytes()
answer := get(t, s, statsPath).Body.Bytes()
var body struct { var body struct {
Data map[string]json.RawMessage `json:"data"` 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
} }
disconnected := connectionDuration() err := json.Unmarshal(answer, &body)
if string(disconnected) != "null" { if err != nil {
t.Errorf("disconnected: connection_duration_seconds = %s, want null", t.Fatalf("decoding the answer: %v; answer %s", err, answer)
disconnected)
} }
s.streamer.GetMetricsTracker().SetConnected(true) duration, ok := body.Data["connection_duration_seconds"]
if !ok {
var seconds int64 t.Fatalf("no connection_duration_seconds: %s", answer)
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 // TestStreamDurationsCountSeconds checks that the uptime and the connection
// duration of a stream that connected 90 seconds ago are both 90. A stream // 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 // that has never connected has an uptime of 0, and one that is not connected
// has no connection duration. The test runs in a synctest bubble, whose clock // has no connection duration.
// is fake, so a stall on the host does not add to the 90 seconds.
func TestStreamDurationsCountSeconds(t *testing.T) { func TestStreamDurationsCountSeconds(t *testing.T) {
t.Parallel() t.Parallel()
synctest.Test(t, func(t *testing.T) { const connectedFor = 90 * time.Second
const connectedFor = 90 * time.Second
connectedSince := time.Now().Add(-connectedFor) connectedSince := time.Now().Add(-connectedFor)
want := int64(connectedFor / time.Second) want := int64(connectedFor / time.Second)
if got := uptimeSeconds(connectedSince); got != want { if got := uptimeSeconds(connectedSince); got != want {
t.Errorf("uptimeSeconds = %d, want %d", got, want) t.Errorf("uptimeSeconds = %d, want %d", got, want)
} }
if got := uptimeSeconds(time.Time{}); got != 0 { if got := uptimeSeconds(time.Time{}); got != 0 {
t.Errorf("uptimeSeconds before any connection = %d, want 0", got) t.Errorf("uptimeSeconds before any connection = %d, want 0", got)
} }
switch got := connectionDurationSeconds(true, connectedSince); { switch got := connectionDurationSeconds(true, connectedSince); {
case got == nil: case got == nil:
t.Errorf("connectionDurationSeconds = nil, want %d", want) t.Errorf("connectionDurationSeconds = nil, want %d", want)
case *got != want: case *got != want:
t.Errorf("connectionDurationSeconds = %d, want %d", *got, want) t.Errorf("connectionDurationSeconds = %d, want %d", *got, want)
} }
if got := connectionDurationSeconds(false, connectedSince); got != nil { if got := connectionDurationSeconds(false, connectedSince); got != nil {
t.Errorf("connectionDurationSeconds while not connected = %d, want nil", t.Errorf("connectionDurationSeconds while not connected = %d, want nil",
*got) *got)
} }
})
} }
// newTestTable returns a routing table that holds a route for testPrefixV4, // newTestTable returns a routing table that holds a route for testPrefixV4,
@@ -1172,9 +1138,7 @@ func (failingDB) GetPrefixFirstSeen(
// with the driver's error text. It also sends a prefix that does not parse. // 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 // No answer may carry the driver's or the prefix parser's text. Before the fix
// the answers sent it to the client // the answers sent it to the client
// (https://git.eeqj.de/sneak/routewatch/issues/108). The requests run in a // (https://git.eeqj.de/sneak/routewatch/issues/108).
// 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) { func TestJSONRoutesDoNotSendErrorDetails(t *testing.T) {
t.Parallel() t.Parallel()
@@ -1195,11 +1159,8 @@ func TestJSONRoutesDoNotSendErrorDetails(t *testing.T) {
for _, tt := range tests { for _, tt := range tests {
rec := httptest.NewRecorder() rec := httptest.NewRecorder()
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
synctest.Test(t, func(t *testing.T) { t.Context(), http.MethodGet, tt.path, nil))
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
t.Context(), http.MethodGet, tt.path, nil))
})
if rec.Code != tt.wantCode { if rec.Code != tt.wantCode {
t.Errorf("%s: status %d, want %d; body %s", t.Errorf("%s: status %d, want %d; body %s",
@@ -1453,11 +1414,8 @@ func TestIPLookupKeepsRouteWhenPTRLookupDoesNotAnswer(t *testing.T) {
s := newServerWithTestRoutes(t) s := newServerWithTestRoutes(t)
useFakeResolver(s, &fakeResolver{ptrNeverAnswers: true}) useFakeResolver(s, &fakeResolver{ptrNeverAnswers: true})
// The request's time limit passes well after the PTR lookup's, so that a // The request's time limit passes after the PTR lookup's.
// stall on a busy host does not use up the difference. ctx, cancel := context.WithTimeout(t.Context(), 2*ptr.LookupTimeout)
const requestTimeLimit = ptr.LookupTimeout + 15*time.Second
ctx, cancel := context.WithTimeout(t.Context(), requestTimeLimit)
defer cancel() defer cancel()
rec := httptest.NewRecorder() rec := httptest.NewRecorder()
@@ -2456,13 +2414,9 @@ func settledGoroutineCount() int {
return prev return prev
} }
// waitForGoroutines waits up to 15 seconds for the goroutine count to drop to // waitForGoroutines waits until the goroutine count drops to target or below.
// target or below.
func waitForGoroutines(target int) bool { func waitForGoroutines(target int) bool {
const maxWait = 15 * time.Second for range 100 {
deadline := time.Now().Add(maxWait)
for time.Now().Before(deadline) {
if runtime.NumGoroutine() <= target { if runtime.NumGoroutine() <= target {
return true return true
} }
+4 -8
View File
@@ -76,9 +76,7 @@ func asLink(asn int) string {
} }
// TestHomePageCardsShowTheTableFigures fills a routing table and the asns // TestHomePageCardsShowTheTableFigures fills a routing table and the asns
// table, works out the cards, and checks that each card shows its figure. The // table, works out the cards, and checks that each card shows its figure.
// 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) { func TestHomePageCardsShowTheTableFigures(t *testing.T) {
t.Parallel() t.Parallel()
@@ -97,9 +95,7 @@ func TestHomePageCardsShowTheTableFigures(t *testing.T) {
t.Fatalf("GetOrCreateASNBatch: %v", err) t.Fatalf("GetOrCreateASNBatch: %v", err)
} }
synctest.Test(t, func(t *testing.T) { s.refreshHomeCards(t.Context())
s.refreshHomeCards(t.Context())
})
page := get(t, s, "/").Body.String() page := get(t, s, "/").Body.String()
@@ -283,7 +279,7 @@ func TestKeepHomeCardsFreshWorksThemOutEveryInterval(t *testing.T) {
const ( const (
interval = 10 * time.Millisecond interval = 10 * time.Millisecond
maxWait = 15 * time.Second maxWait = 5 * time.Second
) )
table := emptyTable() table := emptyTable()
@@ -325,7 +321,7 @@ func TestKeepHomeCardsFreshWorksThemOutEveryInterval(t *testing.T) {
func TestStartedServerWorksOutTheCards(t *testing.T) { func TestStartedServerWorksOutTheCards(t *testing.T) {
t.Parallel() t.Parallel()
const maxWait = 15 * time.Second const maxWait = 5 * time.Second
s := newServerWithTable(t, newCardsTable()) s := newServerWithTable(t, newCardsTable())
+32 -39
View File
@@ -14,7 +14,6 @@ import (
"regexp" "regexp"
"strings" "strings"
"testing" "testing"
"testing/synctest"
"time" "time"
"git.eeqj.de/sneak/routewatch/internal/config" "git.eeqj.de/sneak/routewatch/internal/config"
@@ -164,10 +163,6 @@ func TestJSONAnswersPutNoUnitInAString(t *testing.T) {
// router and the middleware write. Before the fix a panic in a handler, and a // 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 // method /status.json or the health check does not take, answered with an
// empty body (https://git.eeqj.de/sneak/routewatch/issues/201). // 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) { func TestJSONAnswersCarryMeta(t *testing.T) {
t.Parallel() t.Parallel()
@@ -186,10 +181,6 @@ func TestJSONAnswersCarryMeta(t *testing.T) {
s.router.Get(panicsPath, JSONValidationMiddleware(http.HandlerFunc( s.router.Get(panicsPath, JSONValidationMiddleware(http.HandlerFunc(
func(http.ResponseWriter, *http.Request) { panic(errDiskIO) })).ServeHTTP) func(http.ResponseWriter, *http.Request) { panic(errDiskIO) })).ServeHTTP)
failing := New(failingDB{}, newTestTable(),
streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{})
useFakeResolver(failing, &fakeResolver{})
tests := []struct { tests := []struct {
method string method string
path string path string
@@ -225,40 +216,42 @@ func TestJSONAnswersCarryMeta(t *testing.T) {
{http.MethodGet, panicsPath, http.StatusInternalServerError}, {http.MethodGet, panicsPath, http.StatusInternalServerError},
} }
synctest.Test(t, func(t *testing.T) { for _, tt := range tests {
for _, tt := range tests { checkAnswerCarriesMeta(t, s, httptest.NewRequestWithContext(
checkAnswerCarriesMeta(t, s, httptest.NewRequestWithContext( t.Context(), tt.method, tt.path, nil), tt.wantCode)
t.Context(), tt.method, tt.path, nil), tt.wantCode) }
}
// A request whose time limit has already passed answers 504. The // A request whose time limit has already passed answers 504. The handlers
// handlers write the answers of the first five routes, and the JSON // write the answers of the first five routes, and the JSON check on the
// check on the route writes the last. // route writes the last.
pastTimeLimit, cancel := context.WithDeadline(t.Context(), time.Now()) pastTimeLimit, cancel := context.WithDeadline(t.Context(), time.Now())
defer cancel() defer cancel()
for _, path := range []string{ for _, path := range []string{
statusJSONPath, statsPath, "/ip/" + addrInTestPrefixV4, statusJSONPath, statsPath, "/ip/" + addrInTestPrefixV4,
"/api/v1/ip/" + addrInTestPrefixV4, asJSONPath, writesNothingPath, "/api/v1/ip/" + addrInTestPrefixV4, asJSONPath, writesNothingPath,
} { } {
checkAnswerCarriesMeta(t, s, httptest.NewRequestWithContext( checkAnswerCarriesMeta(t, s, httptest.NewRequestWithContext(
pastTimeLimit, http.MethodGet, path, nil), http.StatusGatewayTimeout) pastTimeLimit, http.MethodGet, path, nil), http.StatusGatewayTimeout)
} }
// With a store whose reads fail, the health check answers 503, and // With a store whose reads fail, the health check answers 503, and the
// the handlers of the other routes 500. // handlers of the other routes 500.
failing := New(failingDB{}, newTestTable(),
streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{})
useFakeResolver(failing, &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,
} {
checkAnswerCarriesMeta(t, failing, httptest.NewRequestWithContext( checkAnswerCarriesMeta(t, failing, httptest.NewRequestWithContext(
t.Context(), http.MethodGet, healthCheckPath, nil), t.Context(), http.MethodGet, path, nil), http.StatusInternalServerError)
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 // checkAnswerCarriesMeta serves req through the server's router, and checks
+2 -11
View File
@@ -5,7 +5,6 @@ import (
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"testing" "testing"
"testing/synctest"
) )
// TestRoutesAnswerHEAD sends HEAD for a page and JSON routes through the // TestRoutesAnswerHEAD sends HEAD for a page and JSON routes through the
@@ -13,23 +12,15 @@ import (
// lookup of the caller's own address, are in the list because chi's GetHead // 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 // does not reach them. Before the fix every route answered HEAD with 405
// (https://git.eeqj.de/sneak/routewatch/issues/111). // (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) { func TestRoutesAnswerHEAD(t *testing.T) {
t.Parallel() t.Parallel()
s := newServerWithTestRoutes(t) s := newServerWithTestRoutes(t)
s.resolver = &fakeResolver{}
for _, path := range []string{"/", statusJSONPath, statsPath, "/ip", "/ip/"} { for _, path := range []string{"/", statusJSONPath, statsPath, "/ip", "/ip/"} {
rec := httptest.NewRecorder() rec := httptest.NewRecorder()
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
synctest.Test(t, func(t *testing.T) { t.Context(), http.MethodHead, path, nil))
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
t.Context(), http.MethodHead, path, nil))
})
if rec.Code != http.StatusOK { if rec.Code != http.StatusOK {
t.Errorf("HEAD %s: status %d, want %d; body %s", t.Errorf("HEAD %s: status %d, want %d; body %s",
+2 -2
View File
@@ -22,10 +22,10 @@ func TestStopClosesTheConnectionOfARequestStillRunning(t *testing.T) {
stopDeadline = 100 * time.Millisecond stopDeadline = 100 * time.Millisecond
// waitTimeout is how long the test waits for the request to reach // waitTimeout is how long the test waits for the request to reach
// the handler, and then for the client to see its connection closed. // the handler, and then for the client to see its connection closed.
waitTimeout = 15 * time.Second waitTimeout = 2 * time.Second
// readHeaderTimeout bounds the test server's read of the request // readHeaderTimeout bounds the test server's read of the request
// header, as every http.Server must. // header, as every http.Server must.
readHeaderTimeout = waitTimeout readHeaderTimeout = time.Second
) )
release := make(chan struct{}) release := make(chan struct{})
+73 -130
View File
@@ -14,7 +14,6 @@ import (
"strings" "strings"
"sync/atomic" "sync/atomic"
"testing" "testing"
"testing/synctest"
"time" "time"
"git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/logger"
@@ -399,7 +398,7 @@ func TestStopQueuesTheHeldStateMessages(t *testing.T) {
const ( const (
held = 2 held = 2
lines = held + 1 lines = held + 1
maxWait = 15 * time.Second maxWait = 5 * time.Second
) )
srv := httptest.NewServer(http.HandlerFunc( srv := httptest.NewServer(http.HandlerFunc(
@@ -479,15 +478,15 @@ func TestStopQueuesTheHeldStateMessages(t *testing.T) {
// TestStopReturnsByItsDeadline stops a streamer whose handler has a full queue // 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 // 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 // ends, as one does whose handler waits for a database write. Stop must return
// within its deadline plus a margin instead of waiting for room on the queue // within its deadline plus a small margin instead of waiting for room on the
// and for the worker, and log as not handed over the held message and every // queue and for the worker, and log as not handed over the held message and
// message on the queue. // every message on the queue.
func TestStopReturnsByItsDeadline(t *testing.T) { func TestStopReturnsByItsDeadline(t *testing.T) {
t.Parallel() t.Parallel()
const ( const (
deadline = 100 * time.Millisecond deadline = 100 * time.Millisecond
margin = 15 * time.Second margin = 2 * time.Second
) )
var logs bytes.Buffer var logs bytes.Buffer
@@ -578,7 +577,7 @@ func TestStopBeforeMessageReachesQueues(t *testing.T) {
func TestReconnectWhenStreamEnds(t *testing.T) { func TestReconnectWhenStreamEnds(t *testing.T) {
t.Parallel() t.Parallel()
endResponse := pipeTransport{write: func(context.Context, io.Writer) {}} endResponse := func(http.ResponseWriter, *http.Request) {}
checkReconnect(t, endResponse, streamIdleTimeout) checkReconnect(t, endResponse, streamIdleTimeout)
} }
@@ -590,8 +589,9 @@ func TestReconnectWhenStreamGoesIdle(t *testing.T) {
const idleTimeout = 50 * time.Millisecond const idleTimeout = 50 * time.Millisecond
holdResponse := pipeTransport{ holdResponse := func(w http.ResponseWriter, r *http.Request) {
write: func(ctx context.Context, _ io.Writer) { <-ctx.Done() }, flush(w)
<-r.Context().Done()
} }
checkReconnect(t, holdResponse, idleTimeout) checkReconnect(t, holdResponse, idleTimeout)
} }
@@ -604,51 +604,16 @@ func TestReconnectWhenServerNeverAnswers(t *testing.T) {
const idleTimeout = 50 * time.Millisecond const idleTimeout = 50 * time.Millisecond
neverAnswer := roundTripFunc( neverAnswer := func(_ http.ResponseWriter, r *http.Request) {
func(req *http.Request) (*http.Response, error) { <-r.Context().Done()
<-req.Context().Done() }
return nil, req.Context().Err()
},
)
checkReconnect(t, neverAnswer, idleTimeout) checkReconnect(t, neverAnswer, idleTimeout)
} }
// pipeTransport answers every request with a response whose body write fills. // TestStreamNotDroppedWhileLinesArrive has the server send lines well within
// The body fails when the request's context ends, as a connection's does. // the idle timeout for several times the timeout, then end the response. Each
// write gets that context, so that it can end when the request does. // line must start the idle timeout over, so the stream lasts until the server
type pipeTransport struct { // ends it instead of being dropped as idle.
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) { func TestStreamNotDroppedWhileLinesArrive(t *testing.T) {
t.Parallel() t.Parallel()
@@ -658,35 +623,26 @@ func TestStreamNotDroppedWhileLinesArrive(t *testing.T) {
sendFor = 5 * idleTimeout sendFor = 5 * idleTimeout
) )
// The streamer is made outside the bubble. Its metrics start a goroutine srv := httptest.NewServer(http.HandlerFunc(
// that never returns, and a bubble waits for each of its goroutines to func(w http.ResponseWriter, _ *http.Request) {
// 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) deadline := time.Now().Add(sendFor)
for time.Now().Before(deadline) { for time.Now().Before(deadline) {
_, _ = io.WriteString(w, updateLine) _, _ = io.WriteString(w, updateLine)
// The wait ends with the stream. A bubble's clock stops when flush(w)
// the test function returns, and a goroutine then still time.Sleep(lineInterval)
// asleep panics the test binary.
select {
case <-ctx.Done():
return
case <-time.After(lineInterval):
}
} }
}, }))
}} defer srv.Close()
synctest.Test(t, func(t *testing.T) { s := New(logger.New(), metrics.New())
err := s.stream(context.Background()) s.url = srv.URL
if !errors.Is(err, errStreamEnded) { s.idleTimeout = idleTimeout
t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
} 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 // TestStreamReadsLongLine has the server send one UPDATE line of about
@@ -750,7 +706,7 @@ func TestHandlerStatsReadWhileMessagesAreDropped(t *testing.T) {
const ( const (
lines = 5 lines = 5
wantDropped = lines - 2 wantDropped = lines - 2
maxWait = 15 * time.Second maxWait = 5 * time.Second
) )
srv := httptest.NewServer(http.HandlerFunc( srv := httptest.NewServer(http.HandlerFunc(
@@ -864,70 +820,61 @@ func streamLines(t *testing.T, s *Streamer, lines string) {
// line and then stays open. The streamer must come back for a second // 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 // response, must not be marked connected when it does, and must handle the
// UPDATE in it. // 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( func checkReconnect(
t *testing.T, firstResponse http.RoundTripper, idleTimeout time.Duration, t *testing.T, firstResponse http.HandlerFunc, idleTimeout time.Duration,
) { ) {
t.Helper() t.Helper()
const maxWait = 15 * time.Second const maxWait = 5 * 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() tracker := metrics.New()
connectedAtReconnect := make(chan bool, 1)
synctest.Test(t, func(t *testing.T) { var requests atomic.Int32
connectedAtReconnect := make(chan bool, 1)
laterResponse := pipeTransport{
write: func(ctx context.Context, w io.Writer) {
_, _ = io.WriteString(w, updateLine)
<-ctx.Done() srv := httptest.NewServer(http.HandlerFunc(
}, func(w http.ResponseWriter, r *http.Request) {
} if requests.Add(1) == 1 {
firstResponse(w, r)
var requests atomic.Int32 return
}
s := New(logger.New(), tracker) select {
s.client = &http.Client{Transport: roundTripFunc( case connectedAtReconnect <- tracker.IsConnected():
func(req *http.Request) (*http.Response, error) { default:
if requests.Add(1) == 1 { }
return firstResponse.RoundTrip(req)
}
select { _, _ = io.WriteString(w, updateLine)
case connectedAtReconnect <- tracker.IsConnected():
default:
}
return laterResponse.RoundTrip(req) flush(w)
}, <-r.Context().Done()
)} }))
s.idleTimeout = idleTimeout defer srv.Close()
s.minBackoff = time.Millisecond
handler := updateHandler{handled: make(chan struct{}, 1)}
s.RegisterHandler(handler)
err := s.Start(t.Context()) s := New(logger.New(), tracker)
if err != nil { s.url = srv.URL
t.Fatalf("Start returned error: %v", err) s.idleTimeout = idleTimeout
} s.minBackoff = time.Millisecond
defer s.Stop(t.Context()) handler := updateHandler{handled: make(chan struct{}, 1)}
s.RegisterHandler(handler)
select { err := s.Start(t.Context())
case <-handler.handled: if err != nil {
case <-time.After(maxWait): t.Fatalf("Start returned error: %v", err)
t.Fatalf("no message from a second response handled within %s", maxWait) }
} defer s.Stop(t.Context())
if <-connectedAtReconnect { select {
t.Error("streamer was still marked connected when it came back " + case <-handler.handled:
"for a second response") 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")
}
} }
// settledGoroutineCount lets transient goroutines finish, then reports the // settledGoroutineCount lets transient goroutines finish, then reports the
@@ -949,13 +896,9 @@ func settledGoroutineCount() int {
return prev return prev
} }
// waitForGoroutines waits up to 15 seconds for the goroutine count to drop to // waitForGoroutines waits until the goroutine count drops to target or below.
// target or below.
func waitForGoroutines(target int) bool { func waitForGoroutines(target int) bool {
const maxWait = 15 * time.Second for range 100 {
deadline := time.Now().Add(maxWait)
for time.Now().Before(deadline) {
if runtime.NumGoroutine() <= target { if runtime.NumGoroutine() <= target {
return true return true
} }