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.
- 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`.
- It keeps answers for at most 250,000 addresses, and keeps no answer for a new
address while it holds that many.
## 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
for the record's TTL, in memory and in `cache.db` (closes
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,
NordVPN, Private Internet Access and IVPN, and X4BNet's community list. Each
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
// fails.
const parseTimeLimit = 15 * time.Second
const parseTimeLimit = 2 * 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
+5 -25
View File
@@ -72,34 +72,17 @@ 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
// longBusyTimeoutMs, with room for a stall on a busy host.
readerCheckpointLimit = 15 * time.Second
// 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
// 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 readerCheckpointLimit.
// wait for the reader plus a second, well under the busy timeout.
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
// 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
@@ -596,8 +579,6 @@ 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)
@@ -709,7 +690,6 @@ func TestCheckpointEmptiesWALOnceNoReadIsOpen(t *testing.T) {
}
fillWALPastLimit(ctx, t, db)
setLongBusyTimeout(ctx, t, db)
started := time.Now()
+1 -5
View File
@@ -646,13 +646,9 @@ 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() > maxUpdatedDifference {
listings[0].Updated.Sub(downloaded).Abs() > time.Second {
t.Errorf("after the restart, listings %+v, want one updated at %s",
listings, downloaded)
}
+20 -33
View File
@@ -1,6 +1,6 @@
// 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
// the answers survive a restart.
// keeps each answer in memory for as long as DNS allows, and in cache.db so
// that the answers survive a restart.
package ptr
import (
@@ -29,10 +29,6 @@ const (
// nameSeparator separates the names in a row of the hostnames table.
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.
@@ -56,16 +52,15 @@ type entry struct {
// 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.
type Cache struct {
db *sql.DB
resolvers []string
timeLimit time.Duration
maxEntries int
logger *slog.Logger
client *http.Client
now func() time.Time
db *sql.DB
resolvers []string
timeLimit time.Duration
logger *slog.Logger
client *http.Client
now func() time.Time
// entries holds the kept answers by address, at most maxEntries of them.
// pending holds those that cache.db does not have yet.
// entries holds every answer by address. pending holds those that
// cache.db does not have yet.
mu sync.RWMutex
entries map[string]entry
pending map[string]entry
@@ -78,16 +73,15 @@ type Cache struct {
// New returns an empty Cache. Start loads it.
func New(params Params) *Cache {
return &Cache{
db: params.DB,
resolvers: params.Resolvers,
timeLimit: LookupTimeout,
maxEntries: maxAddresses,
logger: params.Logger.With("component", "ptr"),
client: &http.Client{},
now: time.Now,
entries: make(map[string]entry),
pending: make(map[string]entry),
done: make(chan struct{}),
db: params.DB,
resolvers: params.Resolvers,
timeLimit: LookupTimeout,
logger: params.Logger.With("component", "ptr"),
client: &http.Client{},
now: time.Now,
entries: make(map[string]entry),
pending: make(map[string]entry),
done: make(chan struct{}),
}
}
@@ -174,18 +168,11 @@ func (c *Cache) current(address string) (entry, bool) {
return found, true
}
// store keeps an answer in memory and queues it for cache.db. Once entries
// holds maxEntries addresses it keeps no new address. It still replaces the
// answer of an address already held.
// store keeps an answer in memory and queues it for cache.db.
func (c *Cache) store(address string, answer entry) {
c.mu.Lock()
defer c.mu.Unlock()
_, held := c.entries[address]
if !held && len(c.entries) >= c.maxEntries {
return
}
c.entries[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)
}
}
// 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/httptest"
"net/netip"
"os"
"path/filepath"
"slices"
"strconv"
"strings"
@@ -785,74 +783,3 @@ func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
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"
"sync/atomic"
"testing"
"testing/synctest"
"time"
"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
// 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()
@@ -223,11 +219,7 @@ func TestStatsHandlersAnswerFromMemory(t *testing.T) {
for name, handler := range handlers {
rec := httptest.NewRecorder()
synctest.Test(t, func(t *testing.T) {
handler(rec,
httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil))
})
handler(rec, httptest.NewRequestWithContext(t.Context(), http.MethodGet, "/", nil))
if rec.Code != http.StatusOK {
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
// /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). 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.
// (https://git.eeqj.de/sneak/routewatch/issues/160).
func TestStatsAnswerCarriesChurnPerMinute(t *testing.T) {
t.Parallel()
s := newServerWithTestRoutes(t)
rec := httptest.NewRecorder()
synctest.Test(t, func(t *testing.T) {
s.handleStats()(rec,
httptest.NewRequestWithContext(t.Context(), http.MethodGet, statsPath, nil))
})
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)
@@ -323,26 +310,18 @@ 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. 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.
// page shows, must give how many.
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(answer, &body)
err := json.Unmarshal(get(t, s, statsPath).Body.Bytes(), &body)
if err != nil {
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
// "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. 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.
// same number of bytes as its gc.heap_alloc_bytes.
func TestStatsAnswersGiveUnitsInFieldNames(t *testing.T) {
t.Parallel()
@@ -398,11 +375,7 @@ func TestStatsAnswersGiveUnitsInFieldNames(t *testing.T) {
s.SetASNFetcher(fixedIntervalASNFetcher{interval: fetchInterval})
for _, path := range []string{statusJSONPath, statsPath} {
var answer []byte
synctest.Test(t, func(t *testing.T) {
answer = get(t, s, path).Body.Bytes()
})
answer := get(t, s, path).Body.Bytes()
var body struct {
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
// 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). 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.
// (https://git.eeqj.de/sneak/routewatch/issues/199).
func TestStatsAnswerGivesConnectionDurationInSeconds(t *testing.T) {
t.Parallel()
s := newServerWithTestRoutes(t)
synctest.Test(t, func(t *testing.T) {
connectionDuration := func() json.RawMessage {
answer := get(t, s, statsPath).Body.Bytes()
connectionDuration := func() json.RawMessage {
answer := get(t, s, statsPath).Body.Bytes()
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
var body struct {
Data map[string]json.RawMessage `json:"data"`
}
disconnected := connectionDuration()
if string(disconnected) != "null" {
t.Errorf("disconnected: connection_duration_seconds = %s, want null",
disconnected)
err := json.Unmarshal(answer, &body)
if err != nil {
t.Fatalf("decoding the answer: %v; answer %s", err, 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)
duration, ok := body.Data["connection_duration_seconds"]
if !ok {
t.Fatalf("no connection_duration_seconds: %s", answer)
}
})
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. The test runs in a synctest bubble, whose clock
// is fake, so a stall on the host does not add to the 90 seconds.
// has no connection duration.
func TestStreamDurationsCountSeconds(t *testing.T) {
t.Parallel()
synctest.Test(t, func(t *testing.T) {
const connectedFor = 90 * time.Second
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,
@@ -1172,9 +1138,7 @@ 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). 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.
// (https://git.eeqj.de/sneak/routewatch/issues/108).
func TestJSONRoutesDoNotSendErrorDetails(t *testing.T) {
t.Parallel()
@@ -1195,11 +1159,8 @@ func TestJSONRoutesDoNotSendErrorDetails(t *testing.T) {
for _, tt := range tests {
rec := httptest.NewRecorder()
synctest.Test(t, func(t *testing.T) {
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
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 {
t.Errorf("%s: status %d, want %d; body %s",
@@ -1453,11 +1414,8 @@ func TestIPLookupKeepsRouteWhenPTRLookupDoesNotAnswer(t *testing.T) {
s := newServerWithTestRoutes(t)
useFakeResolver(s, &fakeResolver{ptrNeverAnswers: true})
// 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 = ptr.LookupTimeout + 15*time.Second
ctx, cancel := context.WithTimeout(t.Context(), requestTimeLimit)
// The request's time limit passes after the PTR lookup's.
ctx, cancel := context.WithTimeout(t.Context(), 2*ptr.LookupTimeout)
defer cancel()
rec := httptest.NewRecorder()
@@ -2456,13 +2414,9 @@ func settledGoroutineCount() int {
return prev
}
// waitForGoroutines waits up to 15 seconds for the goroutine count to drop to
// target or below.
// waitForGoroutines waits until the goroutine count drops to target or below.
func waitForGoroutines(target int) bool {
const maxWait = 15 * time.Second
deadline := time.Now().Add(maxWait)
for time.Now().Before(deadline) {
for range 100 {
if runtime.NumGoroutine() <= target {
return true
}
+4 -8
View File
@@ -76,9 +76,7 @@ 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. 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.
// table, works out the cards, and checks that each card shows its figure.
func TestHomePageCardsShowTheTableFigures(t *testing.T) {
t.Parallel()
@@ -97,9 +95,7 @@ func TestHomePageCardsShowTheTableFigures(t *testing.T) {
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()
@@ -283,7 +279,7 @@ func TestKeepHomeCardsFreshWorksThemOutEveryInterval(t *testing.T) {
const (
interval = 10 * time.Millisecond
maxWait = 15 * time.Second
maxWait = 5 * time.Second
)
table := emptyTable()
@@ -325,7 +321,7 @@ func TestKeepHomeCardsFreshWorksThemOutEveryInterval(t *testing.T) {
func TestStartedServerWorksOutTheCards(t *testing.T) {
t.Parallel()
const maxWait = 15 * time.Second
const maxWait = 5 * time.Second
s := newServerWithTable(t, newCardsTable())
+32 -39
View File
@@ -14,7 +14,6 @@ import (
"regexp"
"strings"
"testing"
"testing/synctest"
"time"
"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
// 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()
@@ -186,10 +181,6 @@ 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{})
useFakeResolver(failing, &fakeResolver{})
tests := []struct {
method string
path string
@@ -225,40 +216,42 @@ func TestJSONAnswersCarryMeta(t *testing.T) {
{http.MethodGet, panicsPath, http.StatusInternalServerError},
}
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)
}
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.
// 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{})
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(
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)
}
})
t.Context(), http.MethodGet, path, nil), http.StatusInternalServerError)
}
}
// checkAnswerCarriesMeta serves req through the server's router, and checks
+2 -11
View File
@@ -5,7 +5,6 @@ import (
"net/http"
"net/http/httptest"
"testing"
"testing/synctest"
)
// 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
// 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()
synctest.Test(t, func(t *testing.T) {
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
t.Context(), http.MethodHead, path, nil))
})
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",
+2 -2
View File
@@ -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 = 15 * time.Second
waitTimeout = 2 * time.Second
// readHeaderTimeout bounds the test server's read of the request
// header, as every http.Server must.
readHeaderTimeout = waitTimeout
readHeaderTimeout = time.Second
)
release := make(chan struct{})
+73 -130
View File
@@ -14,7 +14,6 @@ import (
"strings"
"sync/atomic"
"testing"
"testing/synctest"
"time"
"git.eeqj.de/sneak/routewatch/internal/logger"
@@ -399,7 +398,7 @@ func TestStopQueuesTheHeldStateMessages(t *testing.T) {
const (
held = 2
lines = held + 1
maxWait = 15 * time.Second
maxWait = 5 * time.Second
)
srv := httptest.NewServer(http.HandlerFunc(
@@ -479,15 +478,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 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 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.
func TestStopReturnsByItsDeadline(t *testing.T) {
t.Parallel()
const (
deadline = 100 * time.Millisecond
margin = 15 * time.Second
margin = 2 * time.Second
)
var logs bytes.Buffer
@@ -578,7 +577,7 @@ func TestStopBeforeMessageReachesQueues(t *testing.T) {
func TestReconnectWhenStreamEnds(t *testing.T) {
t.Parallel()
endResponse := pipeTransport{write: func(context.Context, io.Writer) {}}
endResponse := func(http.ResponseWriter, *http.Request) {}
checkReconnect(t, endResponse, streamIdleTimeout)
}
@@ -590,8 +589,9 @@ func TestReconnectWhenStreamGoesIdle(t *testing.T) {
const idleTimeout = 50 * time.Millisecond
holdResponse := pipeTransport{
write: func(ctx context.Context, _ io.Writer) { <-ctx.Done() },
holdResponse := func(w http.ResponseWriter, r *http.Request) {
flush(w)
<-r.Context().Done()
}
checkReconnect(t, holdResponse, idleTimeout)
}
@@ -604,51 +604,16 @@ func TestReconnectWhenServerNeverAnswers(t *testing.T) {
const idleTimeout = 50 * time.Millisecond
neverAnswer := roundTripFunc(
func(req *http.Request) (*http.Response, error) {
<-req.Context().Done()
return nil, req.Context().Err()
},
)
neverAnswer := func(_ http.ResponseWriter, r *http.Request) {
<-r.Context().Done()
}
checkReconnect(t, neverAnswer, idleTimeout)
}
// 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.
// 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.
func TestStreamNotDroppedWhileLinesArrive(t *testing.T) {
t.Parallel()
@@ -658,35 +623,26 @@ func TestStreamNotDroppedWhileLinesArrive(t *testing.T) {
sendFor = 5 * idleTimeout
)
// 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) {
srv := httptest.NewServer(http.HandlerFunc(
func(w http.ResponseWriter, _ *http.Request) {
deadline := time.Now().Add(sendFor)
for time.Now().Before(deadline) {
_, _ = io.WriteString(w, updateLine)
// 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):
}
flush(w)
time.Sleep(lineInterval)
}
},
}}
}))
defer srv.Close()
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)
}
})
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)
}
}
// TestStreamReadsLongLine has the server send one UPDATE line of about
@@ -750,7 +706,7 @@ func TestHandlerStatsReadWhileMessagesAreDropped(t *testing.T) {
const (
lines = 5
wantDropped = lines - 2
maxWait = 15 * time.Second
maxWait = 5 * time.Second
)
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
// 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.RoundTripper, idleTimeout time.Duration,
t *testing.T, firstResponse http.HandlerFunc, idleTimeout time.Duration,
) {
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()
connectedAtReconnect := make(chan bool, 1)
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)
var requests atomic.Int32
<-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)
s.client = &http.Client{Transport: roundTripFunc(
func(req *http.Request) (*http.Response, error) {
if requests.Add(1) == 1 {
return firstResponse.RoundTrip(req)
}
select {
case connectedAtReconnect <- tracker.IsConnected():
default:
}
select {
case connectedAtReconnect <- tracker.IsConnected():
default:
}
_, _ = io.WriteString(w, updateLine)
return laterResponse.RoundTrip(req)
},
)}
s.idleTimeout = idleTimeout
s.minBackoff = time.Millisecond
handler := updateHandler{handled: make(chan struct{}, 1)}
s.RegisterHandler(handler)
flush(w)
<-r.Context().Done()
}))
defer srv.Close()
err := s.Start(t.Context())
if err != nil {
t.Fatalf("Start returned error: %v", err)
}
defer s.Stop(t.Context())
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)
select {
case <-handler.handled:
case <-time.After(maxWait):
t.Fatalf("no message from a second response handled within %s", maxWait)
}
err := s.Start(t.Context())
if err != nil {
t.Fatalf("Start returned error: %v", err)
}
defer s.Stop(t.Context())
if <-connectedAtReconnect {
t.Error("streamer was still marked connected when it came back " +
"for a second response")
}
})
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")
}
}
// settledGoroutineCount lets transient goroutines finish, then reports the
@@ -949,13 +896,9 @@ func settledGoroutineCount() int {
return prev
}
// waitForGoroutines waits up to 15 seconds for the goroutine count to drop to
// target or below.
// waitForGoroutines waits until the goroutine count drops to target or below.
func waitForGoroutines(target int) bool {
const maxWait = 15 * time.Second
deadline := time.Now().Add(maxWait)
for time.Now().Before(deadline) {
for range 100 {
if runtime.NumGoroutine() <= target {
return true
}