1 Commits
Author SHA1 Message Date
sneak d79b4169b6 Give every JSON answer @meta, errors included (closes #201)
check / check (push) Canceled after 0s
A panic in a JSON route's handler answered 500 with an empty body,
because the router's recoverer sits outside the middleware that adds
`@meta`. `JSONValidationMiddleware` now wraps the handler in the same
recoverer, and answers the panic with the JSON error body.

A method that `/status.json` or the health check does not take got
chi's empty 405. Each such method now goes to the JSON 405 that the
routes under `/api/v1` and `/ip` give.

The home page's IP lookup card showed only the answer's `data`. It now
shows the whole answer, `@meta` included.

A table test checks `@meta` on every JSON route, in its successes and
its errors.

Model: opus-5-5
2026-10-27 19:51:41 +00:00
20 changed files with 338 additions and 840 deletions
+28 -53
View File
@@ -9,38 +9,19 @@ table in real time. It works like this:
## Rationale ## Rationale
RouteWatch exists to serve our other services as an internal API for information RIPE RIS Live streams the BGP updates that RIPE's route collectors receive, as
about IP addresses. they happen, from the moment a client connects. It reports changes, not the
routing table they add up to, so it cannot say which AS an IP address is routed
to now or which routes a prefix has. RouteWatch applies the feed's updates to a
routing table that it keeps in memory, with a copy in its own SQLite database.
It answers IP-to-AS lookups and prefix pages from that table.
RouteWatch follows BGP routing table updates from RIPE RIS Live. RIS Live The feed sends no table when a client connects. So the routing table holds only
streams the updates that RIPE's route collectors receive, as they happen, from the routes announced since RouteWatch first started. An IP-to-AS lookup finds
the moment a client connects. It reports changes, not the routing table they add nothing for an address whose routes have not been announced since then. The
up to. RouteWatch applies those changes to a routing table that it keeps in routing table also misses updates RouteWatch drops when it falls behind, and
memory, with a copy in its own SQLite database. From that table, RouteWatch sees updates sent while it is stopped. So a lookup or a prefix page can show a route
directly: the feed has withdrawn.
- which AS each IP address is routed to
- which announced prefix holds each IP address, and how big that prefix is
RouteWatch adds details of each AS from WHOIS, the public directory of the
internet registries. The details include the organisation that holds the AS.
RouteWatch keeps them in its database and looks them up again once they are 30
days old. An IP lookup answers from those stored details and never waits for
WHOIS.
So our own services can ask RouteWatch about an IP address, instead of each
working it out for itself.
RIS Live sends no routing table when a client connects. So the routing table
holds only the routes announced since RouteWatch first started. An IP lookup
finds no route for an address until a prefix that holds it is announced. The
routing table also misses these updates:
- those sent while RouteWatch is stopped
- those that RouteWatch drops when it falls behind
A lookup or a prefix page can therefore show a route that has since been
withdrawn.
## Features ## Features
@@ -252,7 +233,7 @@ originates no prefix and has no record in the database or in the AS list built
into RouteWatch. AS 0 always answers 404, because it means no known origin. The into RouteWatch. AS 0 always answers 404, because it means no known origin. The
AS name search never lists it. A prefix with no live route answers 404 too. The AS name search never lists it. A prefix with no live route answers 404 too. The
HTML pages give these errors as plain text. Under `/api/v1` and `/ip/`, a path HTML pages give these errors as plain text. Under `/api/v1` and `/ip/`, a path
that matches no route answers 404 with the message `Not found`. A request with no route matches answers 404 with the message `Not found`. A request with
`POST`, `PUT`, `PATCH`, `DELETE`, `OPTIONS`, `TRACE` or `CONNECT` answers 405 `POST`, `PUT`, `PATCH`, `DELETE`, `OPTIONS`, `TRACE` or `CONNECT` answers 405
with `Method not allowed` on these routes: with `Method not allowed` on these routes:
@@ -260,9 +241,8 @@ with `Method not allowed` on these routes:
- `/status.json` - `/status.json`
- `/.well-known/healthcheck.json` - `/.well-known/healthcheck.json`
A 500 answer says only `Internal server error`. The error for a host name that A 500 answer says only `Internal server error`, and a host name that does not
does not resolve says only `could not resolve host name`. The detail goes to the resolve only `could not resolve host name`; the detail goes to the log.
log.
Every answer carries `Strict-Transport-Security`, `Content-Security-Policy`, Every answer carries `Strict-Transport-Security`, `Content-Security-Policy`,
`X-Frame-Options`, `X-Content-Type-Options`, `Referrer-Policy` and `X-Frame-Options`, `X-Content-Type-Options`, `Referrer-Policy` and
@@ -429,12 +409,10 @@ the routing table, and everything else from the database.
- Parses BGP UPDATE messages from JSON stream - Parses BGP UPDATE messages from JSON stream
- Dispatches to registered handlers based on message type - Dispatches to registered handlers based on message type
- Each handler has its own queue with backpressure handling - Each handler has its own queue with backpressure handling
- Handlers batch writes for efficiency. The AS, peer and prefix handlers - Handlers batch writes for efficiency: the AS, peer and prefix handlers
write a batch when it reaches a size, or after a time: write 30,000, 10,000 and 25,000 updates at a time, or after 2, 2 and 1
- the AS handler at 30,000 ASes, or after 2 seconds; seconds; the peering handler writes the peerings it has collected every 30
- the peer handler at 10,000 updates, or after 2 seconds; seconds
- the prefix handler at 25,000 updates, or after 1 second.
- The peering handler writes the peerings it has collected every 30 seconds
4. **Handler Details** 4. **Handler Details**
- **ASHandler**: Tracks all ASNs seen in AS paths - **ASHandler**: Tracks all ASNs seen in AS paths
@@ -634,19 +612,16 @@ What happens at each limit:
`STATE` message, which it holds, in feed order, until there is room: a few `STATE` message, which it holds, in feed order, until there is room: a few
hundred bytes each, about ten a second. hundred bytes each, about ten a second.
- Handler batches: while the AS, peer or prefix handler writes a batch to the - Handler batches: while the AS, peer or prefix handler writes a batch to the
database, the updates that arrive collect in its next batch. database, the updates that arrive collect in its next batch. That batch stops
- The AS handler's batch holds one entry per AS, with the earliest and the at ten times the batch size:
latest time of its updates. About 80,000 ASes appear in the feed, so the - 300,000 updates for the AS handler (`asnBatchMaxSize`, about 10 MiB);
batch takes at most about 10 MiB, however fast the feed runs. - 100,000 for the peer handler (`peerBatchMaxSize`, about 10 MiB);
- The peer handler's batch stops at 100,000 updates (`peerBatchMaxSize`), - 250,000 for the prefix handler (`prefixBatchMaxSize`, about 15 MiB).
about 10 MiB.
- The prefix handler's batch stops at 250,000 updates
(`prefixBatchMaxSize`), about 15 MiB.
At that maximum the peer or prefix handler takes nothing more off its queue At that size the handler takes nothing more off its queue until the write
until the write ends. The queue then fills, and the streamer drops messages ends. The queue then fills, and the streamer drops messages as above. With
as above. With the batches being written, the three handlers' batches take the batches being written, the three handlers' batches take at most about
at most about 0.1 GiB of the Go heap. 0.1 GiB of the Go heap.
With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds
with the goroutine count and Go memory figures. with the goroutine count and Go memory figures.
+6 -32
View File
@@ -35,39 +35,13 @@ tested.
# Completed Steps # Completed Steps
- 2026-11-02: RouteWatch pauses an RIR's or NIC.br's WHOIS server for an hour
after the server refuses a query. A refused or reset connection is a refusal.
So is an answer with an `%ERROR:201:` or `%ERROR:202:` line. Lookups at the
other servers go on. An AS whose RIR is paused is looked up after the pause.
IANA is never paused, because a pause of IANA would stop every lookup. A
refusal by IANA is an ordinary failure. NIC.br is also paused for an hour
after it says RouteWatch has passed NIC.br's query limit. Before, the fetcher
asked a server again within seconds of a refusal (closes #218)
- 2026-11-02: the AS handler never waits for a database write. Its batch holds
one entry per AS, with the earliest and the latest time of its updates. So the
batch no longer grows with the feed's rate. It used to hold one entry for each
AS of each path and to wait once it held 300,000. In a burst of the feed its
queue then filled, and the streamer dropped messages for it (closes #217)
- 2026-10-27: the tests of `internal/routewatch` no longer fail on a loaded
host. Five of them gave `HandleMessage` two seconds to apply a full prefix
batch, which under the race detector can take longer. A test now waits up to
15 seconds for each step it expects. The tests of the handlers' limits give
the handlers small ones, so the package no longer runs past its 90-second
limit (closes #213, #215)
- 2026-10-27: the server works out the home page's cards every minute until it
stops. It used to work them out only once, at startup: the end of the context
that `Start` is given stopped the refreshes (closes #211)
- 2026-10-27: README "Rationale" says why RouteWatch exists: to answer our other
services' questions about IP addresses, from its own routing table and the
WHOIS details it keeps. It still says what RIS Live does not give (closes
#204)
- 2026-10-27: every JSON answer carries `@meta`. A panic on a JSON route answers - 2026-10-27: every JSON answer carries `@meta`. A panic on a JSON route answers
500 with the JSON error body. `/status.json` and the health check answer 405 500 with the JSON error body. A method that `/status.json` or the health check
with the JSON error body to any method they do not take. The 500 and the 405 does not take answers 405 with it, as under `/api/v1`. Both used to answer
used to have an empty body. The home page's IP lookup shows the whole JSON with an empty body. The home page's IP lookup shows the whole JSON answer,
answer, `@meta` included. It used to show only the answer's `data`, or only `@meta` included. It used to show only the answer's `data`, or only the
the message of an error answer. A test checks `@meta` on every JSON route, in message of an error answer. A test checks `@meta` on every JSON route, in its
its successes and its errors (closes #201) successes and its errors (closes #201)
- 2026-10-27: every value in a JSON answer that has a unit is a number, with the - 2026-10-27: every value in a JSON answer that has a unit is a number, with the
unit at the end of its field name. `@meta.execution_time` is now unit at the end of its field name. `@meta.execution_time` is now
`execution_time_ms`. The stats' `uptime`, `connection_duration` and `execution_time_ms`. The stats' `uptime`, `connection_duration` and
@@ -509,6 +509,8 @@ func announcementFeed(prefixes []string) string {
func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) { func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
t.Parallel() t.Parallel()
const maxWait = 10 * time.Second
prefixes := []string{"192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"} prefixes := []string{"192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"}
feed := announcementFeed(prefixes) feed := announcementFeed(prefixes)
@@ -554,9 +556,9 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
for range cap(linesSeen) { for range cap(linesSeen) {
select { select {
case <-linesSeen: case <-linesSeen:
case <-time.After(stepTimeout): case <-time.After(maxWait):
t.Fatalf("the streamer did not read %d lines within %s", t.Fatalf("the streamer did not read %d lines within %s",
cap(linesSeen), stepTimeout) cap(linesSeen), maxWait)
} }
} }
@@ -569,8 +571,8 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
select { select {
case <-shutDown: case <-shutDown:
case <-time.After(stepTimeout): case <-time.After(maxWait):
t.Fatalf("Shutdown did not return within %s", stepTimeout) t.Fatalf("Shutdown did not return within %s", maxWait)
} }
store.mu.Lock() store.mu.Lock()
@@ -607,16 +609,21 @@ func freePort(t *testing.T) int {
return addr.Port return addr.Port
} }
// TestShutdownDoesNotWaitForABlockedWrite starts the HTTP server and has the // TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked starts the HTTP server
// AS handler start a write that the store holds until the test ends, past // and has the AS handler start a write that the store holds until the test
// Shutdown's deadline. Shutdown must return without waiting for the write, with // ends, past Shutdown's deadline. Shutdown must return within its deadline
// the HTTP server stopped, and must neither checkpoint nor close the database: // plus a small margin, with the HTTP server stopped, and must neither
// the real one would wait for the write. // checkpoint nor close the database: the real one would wait for the write.
// Before the fix Shutdown waited for the write. // Before the fix Shutdown waited for the write.
func TestShutdownDoesNotWaitForABlockedWrite(t *testing.T) { func TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) {
t.Parallel() t.Parallel()
const deadline = 100 * time.Millisecond const (
deadline = 100 * time.Millisecond
// margin is how long Shutdown may take after its deadline, to stop
// the HTTP server.
margin = 2 * time.Second
)
blocking := newBlockingStore() blocking := newBlockingStore()
defer blocking.releaseWrites() defer blocking.releaseWrites()
@@ -644,8 +651,9 @@ func TestShutdownDoesNotWaitForABlockedWrite(t *testing.T) {
} }
rw.dbHandler = NewASHandler(store, log) rw.dbHandler = NewASHandler(store, log)
fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
startBlockedWrite(t, blocking, rw.dbHandler, startBlockedWrite(t, blocking, rw.dbHandler,
[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}}) []*ristypes.RISMessage{{Path: fill}})
ctx, cancel := context.WithTimeout(t.Context(), deadline) ctx, cancel := context.WithTimeout(t.Context(), deadline)
defer cancel() defer cancel()
@@ -659,9 +667,9 @@ func TestShutdownDoesNotWaitForABlockedWrite(t *testing.T) {
select { select {
case <-shutDown: case <-shutDown:
case <-time.After(deadline + stepTimeout): case <-time.After(deadline + margin):
t.Fatalf("Shutdown did not return within %s of its deadline while "+ t.Fatalf("Shutdown did not return within %s of its deadline while "+
"a write was blocked", stepTimeout) "a write was blocked", margin)
} }
var dialer net.Dialer var dialer net.Dialer
@@ -736,6 +744,8 @@ func (s *closeOrderStore) record(call string) {
func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) { func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
t.Parallel() t.Parallel()
const deadline = 10 * time.Second
store := &closeOrderStore{Store: newMockStore()} store := &closeOrderStore{Store: newMockStore()}
log := logger.New() log := logger.New()
cfg := &config.Config{} cfg := &config.Config{}
@@ -753,7 +763,7 @@ func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
Path: ristypes.ASPath{Sequence: []int{testASNA}}, Path: ristypes.ASPath{Sequence: []int{testASNA}},
}) })
ctx, cancel := context.WithTimeout(t.Context(), stepTimeout) ctx, cancel := context.WithTimeout(t.Context(), deadline)
defer cancel() defer cancel()
rw.Shutdown(ctx) rw.Shutdown(ctx)
+85 -40
View File
@@ -18,34 +18,39 @@ const (
// (asnBatchTimeout), so a queue smaller than asnBatchSize is fine. // (asnBatchTimeout), so a queue smaller than asnBatchSize is fine.
asHandlerQueueSize = 20000 asHandlerQueueSize = 20000
// asnBatchSize is the number of ASes in a full batch, which wakes the // asnBatchSize is the number of ASN operations to batch together
// flush goroutine to write it.
asnBatchSize = 30000 asnBatchSize = 30000
// asnBatchMaxSize is the most ASN operations the batch holds while the
// flush goroutine writes the previous one: ten batches, about 10 MiB
// (README "Memory"). At this size HandleMessage waits until the flush
// goroutine takes the batch.
asnBatchMaxSize = 300000
// asnBatchTimeout is the maximum time to wait before flushing a batch // asnBatchTimeout is the maximum time to wait before flushing a batch
// DO NOT reduce this timeout - larger batches are more efficient // DO NOT reduce this timeout - larger batches are more efficient
asnBatchTimeout = 2 * time.Second asnBatchTimeout = 2 * time.Second
) )
// ASHandler records every AS in the AS paths of BGP UPDATE messages. Its batch // ASHandler processes Autonomous System Number (ASN) information extracted from
// holds the earliest and the latest time of each AS's updates. The flush // BGP UPDATE messages. It uses batched database operations to efficiently store
// goroutine writes the batch to the database when it holds asnBatchSize ASes, // ASN data, collecting operations into batches that are flushed either when the
// and every asnBatchTimeout. // batch reaches a size threshold or after a timeout period.
type ASHandler struct { type ASHandler struct {
db database.Store db database.Store
logger *logger.Logger logger *logger.Logger
// Batching. mu guards batch and writing only. The flush goroutine takes // Batching. mu guards batch and writing only: the flush goroutine takes
// the batch under mu and writes it after releasing mu. So HandleMessage // the batch under mu and writes it after releasing mu, so HandleMessage
// never waits for a database write. // waits for a database write only while batch is at asnBatchMaxSize.
mu sync.Mutex mu sync.Mutex
// batch holds the earliest and the latest time of the updates for each AS batch []asnOp
// since the flush goroutine last took it. It holds one entry per AS, // writing is how many operations the flush goroutine took from batch
// however many updates arrive. // and has not finished writing.
batch map[int]database.SeenTimes
// writing is how many ASes the flush goroutine took from batch and has
// not finished writing.
writing int writing int
// batchTaken is signalled when the flush goroutine takes batch; its L is
// &mu.
batchTaken sync.Cond
batchFullCh chan struct{} // wakes the flush goroutine when batch is full batchFullCh chan struct{} // wakes the flush goroutine when batch is full
// stopCh hands Stop's context to the flush goroutine. It holds one, so // stopCh hands Stop's context to the flush goroutine. It holds one, so
// Stop does not wait for a write in progress to hand it over. // Stop does not wait for a write in progress to hand it over.
@@ -53,10 +58,16 @@ type ASHandler struct {
done chan struct{} // closed when the flush goroutine returns done chan struct{} // closed when the flush goroutine returns
} }
type asnOp struct {
number int
timestamp time.Time
}
// NewASHandler creates and returns a new ASHandler instance. It initializes // NewASHandler creates and returns a new ASHandler instance. It initializes
// the batching system and starts a background goroutine that periodically // the batching system and starts a background goroutine that periodically
// writes the collected ASes to the database. The caller must call Stop when // flushes accumulated ASN operations to the database. The caller must call
// finished, to write the pending ASes and end the background goroutine. // Stop when finished to ensure all pending operations are flushed and the
// background goroutine is terminated.
func NewASHandler(db database.Store, logger *logger.Logger) *ASHandler { func NewASHandler(db database.Store, logger *logger.Logger) *ASHandler {
return newASHandler(db, logger, asnBatchTimeout) return newASHandler(db, logger, asnBatchTimeout)
} }
@@ -70,11 +81,12 @@ func newASHandler(
h := &ASHandler{ h := &ASHandler{
db: db, db: db,
logger: logger, logger: logger,
batch: make(map[int]database.SeenTimes), batch: make([]asnOp, 0, asnBatchSize),
batchFullCh: make(chan struct{}, 1), batchFullCh: make(chan struct{}, 1),
stopCh: make(chan context.Context, 1), stopCh: make(chan context.Context, 1),
done: make(chan struct{}), done: make(chan struct{}),
} }
h.batchTaken.L = &h.mu
// Start the flush timer goroutine // Start the flush timer goroutine
go flushLoop(batchTimeout, h.batchFullCh, h.stopCh, h.done, h.flushBatch) go flushLoop(batchTimeout, h.batchFullCh, h.stopCh, h.done, h.flushBatch)
@@ -98,25 +110,48 @@ func (h *ASHandler) QueueCapacity() int {
return asHandlerQueueSize return asHandlerQueueSize
} }
// HandleMessage adds the time of msg to the batch entry of each AS in its AS // HandleMessage processes a RIS Live BGP message by extracting all ASNs from
// path, the origin AS and the members of an AS set included. When the batch // the AS path and queuing them for batch insertion into the database. The
// holds asnBatchSize ASes, HandleMessage wakes the flush goroutine to write it. // origin ASN and all transit ASNs are recorded with their associated
// timestamps. When the batch reaches the configured size threshold, the flush
// goroutine is woken to write it.
func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) { func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
// Use the pre-parsed timestamp
timestamp := msg.ParsedTimestamp
originASN := msg.Path.Origin()
h.mu.Lock() h.mu.Lock()
defer h.mu.Unlock() defer h.mu.Unlock()
// The origin AS is the last AS of the sequence, so the loop adds it too. // While the batch is at its maximum size, wait until the flush goroutine
// takes it. Meanwhile the handler takes nothing off its queue, which
// fills, and the streamer drops messages for it.
for len(h.batch) >= asnBatchMaxSize {
h.batchTaken.Wait()
}
// Queue origin ASN operation
if originASN > 0 {
h.batch = append(h.batch, asnOp{
number: originASN,
timestamp: timestamp,
})
}
// Also track all ASNs in the path, the members of an AS set included
for _, asn := range slices.Concat(msg.Path.Sequence, msg.Path.Set) { for _, asn := range slices.Concat(msg.Path.Sequence, msg.Path.Set) {
if asn > 0 { if asn > 0 {
seen := h.batch[asn] h.batch = append(h.batch, asnOp{
seen.Add(msg.ParsedTimestamp) number: asn,
h.batch[asn] = seen timestamp: timestamp,
})
} }
} }
// A full batch wakes the flush goroutine. If it is writing, the wake-up // A full batch wakes the flush goroutine. If it is writing, the wake-up
// waits in batchFullCh and the batch keeps taking updates until the write // waits in batchFullCh and the batch keeps growing until the write ends
// ends. // or the batch reaches asnBatchMaxSize.
if len(h.batch) >= asnBatchSize { if len(h.batch) >= asnBatchSize {
select { select {
case h.batchFullCh <- struct{}{}: case h.batchFullCh <- struct{}{}:
@@ -125,11 +160,10 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
} }
} }
// Stop stops the flush goroutine and waits for it until ctx ends. The flush // Stop stops the flush goroutine, which writes the operations the handler holds
// goroutine writes the batch before it returns, unless ctx has ended. A write // before it returns, and waits for it until ctx ends. Once ctx has ended the
// still running when the process exits is lost. Stop logs how many entries are // flush goroutine skips that last write, and a write still running is lost
// not written, in the batch being written and in the next batch. An AS in both // when the process exits, so Stop logs how many operations are not written.
// batches counts twice.
func (h *ASHandler) Stop(ctx context.Context) { func (h *ASHandler) Stop(ctx context.Context) {
h.stopCh <- ctx h.stopCh <- ctx
@@ -144,14 +178,14 @@ func (h *ASHandler) Stop(ctx context.Context) {
if unwritten > 0 { if unwritten > 0 {
h.logger.Error("Stop deadline passed before the AS handler wrote "+ h.logger.Error("Stop deadline passed before the AS handler wrote "+
"its batches", "batch_entries", unwritten) "its updates", "updates", unwritten)
} }
} }
// takeBatch returns the pending ASes and leaves an empty batch in their place // takeBatch returns the pending operations and leaves an empty batch in their
// for HandleMessage to fill. They count in writing until flushBatch has written // place for HandleMessage to fill. They count in writing until flushBatch has
// them. // written them.
func (h *ASHandler) takeBatch() map[int]database.SeenTimes { func (h *ASHandler) takeBatch() []asnOp {
h.mu.Lock() h.mu.Lock()
defer h.mu.Unlock() defer h.mu.Unlock()
@@ -161,7 +195,8 @@ func (h *ASHandler) takeBatch() map[int]database.SeenTimes {
batch := h.batch batch := h.batch
h.writing = len(batch) h.writing = len(batch)
h.batch = make(map[int]database.SeenTimes) h.batch = make([]asnOp, 0, asnBatchSize)
h.batchTaken.Broadcast()
return batch return batch
} }
@@ -174,11 +209,21 @@ func (h *ASHandler) flushBatch() {
return return
} }
// The earliest and the latest time the batch saw each AS. The batch holds
// several updates for an AS, from many peers, so a new AS is stored first
// seen at the earliest of them and last seen at the latest.
asns := make(map[int]database.SeenTimes)
for _, op := range batch {
seen := asns[op.number]
seen.Add(op.timestamp)
asns[op.number] = seen
}
// Process all ASNs in a single batch transaction // Process all ASNs in a single batch transaction
err := h.db.GetOrCreateASNBatch(batch) err := h.db.GetOrCreateASNBatch(asns)
if err != nil { if err != nil {
h.logger.Error("Failed to process ASN batch", h.logger.Error("Failed to process ASN batch",
"error", err, "count", len(batch)) "error", err, "count", len(asns))
} }
h.mu.Lock() h.mu.Lock()
+7 -24
View File
@@ -5,43 +5,28 @@ import (
"slices" "slices"
"testing" "testing"
"git.eeqj.de/sneak/routewatch/internal/database"
"git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/logger"
"git.eeqj.de/sneak/routewatch/internal/ristypes"
) )
// pathOfASes returns an AS path of n different ASes: AS 1 to AS n. With n up to
// asnBatchSize, none of them is testASNA to testASND.
func pathOfASes(n int) ristypes.ASPath {
sequence := make([]int, n)
for i := range sequence {
sequence[i] = i + 1
}
return ristypes.ASPath{Sequence: sequence}
}
// TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS handles two updates // TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS handles two updates
// whose path carries an AS the database does not hold yet, in one batch into a // whose path carries an AS the database does not hold yet, the second 5
// real database. The second is 5 seconds after the first, and the handler gets // seconds after the first, in one batch into a real database. The AS is stored
// it first. The AS is stored first seen at the time of the first update and // first seen at the time of the first update and last seen at the time of the
// last seen at the time of the second. // second.
func TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS(t *testing.T) { func TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS(t *testing.T) {
t.Parallel() t.Parallel()
db := newTestDatabase(t) db := newTestDatabase(t)
// Built without NewASHandler's flush goroutine, so the test flushes itself. // Built without NewASHandler's flush goroutine, so the test flushes itself.
h := &ASHandler{ h := &ASHandler{db: db, logger: logger.New()}
db: db, logger: logger.New(), batch: make(map[int]database.SeenTimes),
}
first := peerAnnouncement(testPeerIP, testV4Prefix) first := peerAnnouncement(testPeerIP, testV4Prefix)
second := peerAnnouncement(testPeerIP, testV4Prefix) second := peerAnnouncement(testPeerIP, testV4Prefix)
second.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter) second.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter)
h.HandleMessage(second)
h.HandleMessage(first) h.HandleMessage(first)
h.HandleMessage(second)
h.flushBatch() h.flushBatch()
asn, _, err := db.GetAS(t.Context(), testASNA) asn, _, err := db.GetAS(t.Context(), testASNA)
@@ -69,9 +54,7 @@ func TestASHandlerRecordsNothingFromAPathWithASZero(t *testing.T) {
store := newMockStore() store := newMockStore()
// Built without NewASHandler's flush goroutine, so the test flushes itself. // Built without NewASHandler's flush goroutine, so the test flushes itself.
h := &ASHandler{ h := &ASHandler{db: store, logger: logger.New()}
db: store, logger: logger.New(), batch: make(map[int]database.SeenTimes),
}
msg := peerAnnouncement(testPeerIP, testV4Prefix) msg := peerAnnouncement(testPeerIP, testV4Prefix)
msg.Path = decodePath(t, testASNA, 0, testASNB) msg.Path = decodePath(t, testASNA, 0, testASNB)
-14
View File
@@ -287,13 +287,6 @@ func (f *ASNFetcher) fetchNextStale(ctx context.Context) {
// so the AS stays stale. An AS IANA refers to no RIR is stored with no // so the AS stays stale. An AS IANA refers to no RIR is stored with no
// WHOIS data, so it is not looked up again until it is stale; that counts // WHOIS data, so it is not looked up again until it is stale; that counts
// as neither a success nor a failure. // as neither a success nor a failure.
//
// The WHOIS client pauses an RIR's or NIC.br's server for whois.PauseTime
// after the server refuses a query. Lookups at the other servers go on. IANA
// is never paused, so a refusal by IANA is an ordinary failure. A lookup
// whose RIR is paused fails with whois.ErrPaused. Such a lookup leaves the
// AS stale, and counts as neither a success nor a failure. The lookups that
// QueueImmediate asks for also run through fetchAndUpdate.
func (f *ASNFetcher) fetchAndUpdate(ctx context.Context, asn int) bool { func (f *ASNFetcher) fetchAndUpdate(ctx context.Context, asn int) bool {
ctx, cancel := context.WithTimeout(ctx, time.Minute) ctx, cancel := context.WithTimeout(ctx, time.Minute)
defer cancel() defer cancel()
@@ -314,13 +307,6 @@ func (f *ASNFetcher) fetchAndUpdate(ctx context.Context, asn int) bool {
return false return false
} }
if errors.Is(err, whois.ErrPaused) {
f.logger.Info("Skipped WHOIS lookup, the RIR's server is paused",
"asn", asn, "error", err)
return false
}
if err != nil { if err != nil {
f.logger.Error("WHOIS lookup failed", "asn", asn, "error", err) f.logger.Error("WHOIS lookup failed", "asn", asn, "error", err)
f.recordFailure() f.recordFailure()
+4 -108
View File
@@ -13,7 +13,6 @@ import (
"git.eeqj.de/sneak/routewatch/internal/database" "git.eeqj.de/sneak/routewatch/internal/database"
"git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/logger"
"git.eeqj.de/sneak/routewatch/internal/whois"
) )
// The WHOIS servers a lookup asks: IANA first, then the RIR IANA refers it // The WHOIS servers a lookup asks: IANA first, then the RIR IANA refers it
@@ -132,109 +131,6 @@ func TestFetchAndUpdateKeepsTheRowWhenTheLookupFails(t *testing.T) {
} }
} }
// TestFetchAndUpdateWaitsOutAPause looks up an AS at RIPE while RIPE refuses
// the query. The test sets the WHOIS client's clock. The next lookup comes a
// second before whois.PauseTime has passed. It sends RIPE nothing and leaves
// the AS never looked up. It counts as neither a success nor a failure. A
// lookup at whois.PauseTime asks RIPE again.
func TestFetchAndUpdateWaitsOutAPause(t *testing.T) {
t.Parallel()
f, db := newFetcherWithWHOISAnswers(t, map[string]string{
ianaWHOISServer: ianaRefersToRIPE,
ripeWHOISServer: ripeAccessDenied,
})
ripeAsks := 0
dial := f.whoisClient.DialContext
f.whoisClient.DialContext = func(
ctx context.Context, network, address string,
) (net.Conn, error) {
if address == ripeWHOISServer {
ripeAsks++
}
return dial(ctx, network, address)
}
start := time.Date(2026, 11, 2, 0, 0, 0, 0, time.UTC)
now := start
f.whoisClient.Now = func() time.Time { return now }
f.fetchAndUpdate(t.Context(), testASNA)
now = start.Add(whois.PauseTime - time.Second)
f.fetchAndUpdate(t.Context(), testASNA)
if ripeAsks != 1 {
t.Errorf("RIPE was asked %d times during its pause, want 1", ripeAsks)
}
stats, err := db.GetWHOISStats(t.Context(), whoisStaleThreshold)
if err != nil {
t.Fatalf("GetWHOISStats: %v", err)
}
if stats.NeverFetched != 1 {
t.Errorf("never looked up = %d, want 1", stats.NeverFetched)
}
got := f.GetStats()
if got.ConsecutiveFails != 1 || got.ErrorsLastHour != 1 ||
got.SuccessesLastHour != 0 {
t.Errorf("%d failures in a row, %d errors, %d successes; want 1, 1, 0",
got.ConsecutiveFails, got.ErrorsLastHour, got.SuccessesLastHour)
}
now = start.Add(whois.PauseTime)
f.fetchAndUpdate(t.Context(), testASNA)
if ripeAsks != 2 {
t.Errorf("RIPE was asked %d times after its pause, want 2", ripeAsks)
}
}
// TestFetchAndUpdateKeepsAskingIANA looks up an AS while IANA's server
// refuses the connection. The test sets the WHOIS client's clock. The next
// lookup comes a second before whois.PauseTime has passed. IANA is never
// paused, so that lookup asks IANA again. Each refusal counts as a failure.
func TestFetchAndUpdateKeepsAskingIANA(t *testing.T) {
t.Parallel()
// With no answers, every server refuses the connection.
f, _ := newFetcherWithWHOISAnswers(t, map[string]string{})
ianaAsks := 0
dial := f.whoisClient.DialContext
f.whoisClient.DialContext = func(
ctx context.Context, network, address string,
) (net.Conn, error) {
if address == ianaWHOISServer {
ianaAsks++
}
return dial(ctx, network, address)
}
start := time.Date(2026, 11, 2, 0, 0, 0, 0, time.UTC)
now := start
f.whoisClient.Now = func() time.Time { return now }
f.fetchAndUpdate(t.Context(), testASNA)
now = start.Add(whois.PauseTime - time.Second)
f.fetchAndUpdate(t.Context(), testASNA)
got := f.GetStats()
if ianaAsks != 2 || got.ConsecutiveFails != 2 {
t.Errorf("IANA was asked %d times, with %d failures in a row; "+
"want 2 and 2", ianaAsks, got.ConsecutiveFails)
}
}
// TestRecordStepsTheInterval has the fetcher fail from its base interval of // TestRecordStepsTheInterval has the fetcher fail from its base interval of
// 15 seconds, then succeed. Each failure makes the interval 1.5 times as // 15 seconds, then succeed. Each failure makes the interval 1.5 times as
// long, up to one minute, and each success halves it, down to one second. // long, up to one minute, and each success halves it, down to one second.
@@ -328,7 +224,7 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
select { select {
case <-lookupStarted: case <-lookupStarted:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatal("the fetcher did not start the lookup") t.Fatal("the fetcher did not start the lookup")
} }
@@ -344,22 +240,22 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
select { select {
case <-f.done: case <-f.done:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatal("the fetcher goroutine did not return after the lookup ended") t.Fatal("the fetcher goroutine did not return after the lookup ended")
} }
} }
// TestRecordKeepsOnlyTheStatsWindow records a success and a failure every // TestRecordKeepsOnlyTheStatsWindow records a success and a failure every
// second, by a clock the test sets, for three times statsWindow, and never // second, by a clock the test sets, far longer than statsWindow, and never
// asks for the stats. The fetcher must keep only the times within the last // asks for the stats. The fetcher must keep only the times within the last
// statsWindow. // statsWindow.
func TestRecordKeepsOnlyTheStatsWindow(t *testing.T) { func TestRecordKeepsOnlyTheStatsWindow(t *testing.T) {
t.Parallel() t.Parallel()
const ( const (
records = 100_000
step = time.Second step = time.Second
want = int(statsWindow / step) want = int(statsWindow / step)
records = 3 * want
) )
clock := time.Now() clock := time.Now()
+52 -107
View File
@@ -19,15 +19,9 @@ import (
) )
const ( const (
// stepTimeout is how long a test waits for a step it expects, such as // handleMessageTimeout is how long a test waits for HandleMessage to
// HandleMessage returning, a write starting or Shutdown returning, before // return, or for a write to start, before it fails.
// it fails. Only a step that never comes should reach it. On a loaded handleMessageTimeout = 2 * time.Second
// host, under the race detector, HandleMessage takes seconds to apply a
// full prefix batch. The helpers below bound every wait with it, Stop
// included. A test that hangs in one of them waits out at most two of
// these bounds, its step's and Stop's, and then fails with its own
// message. That leaves time before go test's 90-second timeout.
stepTimeout = 15 * time.Second
// stillWaitingAfter is how long a test watches a HandleMessage that must // stillWaitingAfter is how long a test watches a HandleMessage that must
// wait. One that does not wait returns well within it. // wait. One that does not wait returns well within it.
@@ -178,7 +172,6 @@ type stopLog struct {
Updates int `json:"updates"` Updates int `json:"updates"`
TableChanges int `json:"table_changes"` TableChanges int `json:"table_changes"`
Paths int `json:"paths"` Paths int `json:"paths"`
BatchEntries int `json:"batch_entries"`
} }
// newErrorLogger returns a logger that writes only errors, as JSON lines, to // newErrorLogger returns a logger that writes only errors, as JSON lines, to
@@ -205,7 +198,7 @@ func lastLogged(t *testing.T, logs *bytes.Buffer) stopLog {
return logged return logged
} }
// returnsInTime reports whether f returns within stepTimeout. // returnsInTime reports whether f returns within handleMessageTimeout.
func returnsInTime(f func()) bool { func returnsInTime(f func()) bool {
done := make(chan struct{}) done := make(chan struct{})
@@ -217,22 +210,11 @@ func returnsInTime(f func()) bool {
select { select {
case <-done: case <-done:
return true return true
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
return false return false
} }
} }
// stopInTime stops h, with no deadline, and fails the test if Stop does not
// return within stepTimeout. It only marks the test failed, so it can be
// deferred.
func stopInTime(t *testing.T, h interface{ Stop(ctx context.Context) }) {
t.Helper()
if !returnsInTime(func() { h.Stop(t.Context()) }) {
t.Errorf("Stop did not return within %s", stepTimeout)
}
}
// startBlockedWrite hands h the messages in fill and waits until h starts to // startBlockedWrite hands h the messages in fill and waits until h starts to
// write them to store. The prefix, AS and peer handlers start when fill fills // write them to store. The prefix, AS and peer handlers start when fill fills
// their batch, and the peering handler on its timer. store holds the write // their batch, and the peering handler on its timer. store holds the write
@@ -258,7 +240,7 @@ func startBlockedWrite(
select { select {
case <-store.writeStarted: case <-store.writeStarted:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatal("the handler did not start to write the messages") t.Fatal("the handler did not start to write the messages")
} }
} }
@@ -279,7 +261,7 @@ func handleWhileAWriteIsBlocked(
) { ) {
t.Helper() t.Helper()
defer stopInTime(t, h) defer h.Stop(t.Context())
defer store.releaseWrites() defer store.releaseWrites()
startBlockedWrite(t, store, h, fill) startBlockedWrite(t, store, h, fill)
@@ -293,9 +275,7 @@ func handleWhileAWriteIsBlocked(
// while store holds the write, hands h the messages in toMax, which bring its // while store holds the write, hands h the messages in toMax, which bring its
// next batch to its maximum size, and the message extra. HandleMessage for // next batch to its maximum size, and the message extra. HandleMessage for
// extra must wait until the test lets the write finish, and return after. Then // extra must wait until the test lets the write finish, and return after. Then
// it stops h, which writes what it holds. The tests lower h's maximum batch // it stops h, which writes what it holds.
// size to one batch, so toMax is one batch: the real maximum of ten batches is
// slow to reach under the race detector.
func handleAtMaxBatchSize( func handleAtMaxBatchSize(
t *testing.T, t *testing.T,
store *blockingStore, store *blockingStore,
@@ -308,19 +288,14 @@ func handleAtMaxBatchSize(
) { ) {
t.Helper() t.Helper()
defer stopInTime(t, h) defer h.Stop(t.Context())
defer store.releaseWrites() defer store.releaseWrites()
startBlockedWrite(t, store, h, fill) startBlockedWrite(t, store, h, fill)
if !returnsInTime(func() {
for _, msg := range toMax { for _, msg := range toMax {
h.HandleMessage(msg) h.HandleMessage(msg)
} }
}) {
t.Fatal("HandleMessage did not return while the batch grew to its " +
"maximum size")
}
returned := make(chan struct{}) returned := make(chan struct{})
@@ -340,7 +315,7 @@ func handleAtMaxBatchSize(
select { select {
case <-returned: case <-returned:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatal("HandleMessage did not return after the database write finished") t.Fatal("HandleMessage did not return after the database write finished")
} }
} }
@@ -370,13 +345,9 @@ func stopWhileAWriteIsBlocked(
startBlockedWrite(t, store, h, fill) startBlockedWrite(t, store, h, fill)
if !returnsInTime(func() {
for _, msg := range extra { for _, msg := range extra {
h.HandleMessage(msg) h.HandleMessage(msg)
} }
}) {
t.Fatal("HandleMessage did not return while a database write was blocked")
}
ctx, cancel := context.WithTimeout(t.Context(), stopDeadline) ctx, cancel := context.WithTimeout(t.Context(), stopDeadline)
defer cancel() defer cancel()
@@ -390,7 +361,7 @@ func stopWhileAWriteIsBlocked(
select { select {
case <-done: case <-done:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatal("the flush goroutine did not return after the write finished") t.Fatal("the flush goroutine did not return after the write finished")
} }
} }
@@ -409,8 +380,7 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
store := newBlockingStore() store := newBlockingStore()
h := newPrefixHandler(store, newErrorLogger(&logs), h := newPrefixHandler(store, newErrorLogger(&logs),
routetable.New(routetable.Params{}), longerThanAnyTest, routetable.New(routetable.Params{}), longerThanAnyTest)
maxTableChangesPerWrite)
fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize) fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
stopWhileAWriteIsBlocked(t, store, h, h.done, stopWhileAWriteIsBlocked(t, store, h, h.done,
[]*ristypes.RISMessage{prefixAnnouncement(fill)}, []*ristypes.RISMessage{prefixAnnouncement(fill)},
@@ -496,20 +466,20 @@ func TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress(t *testing.T)
} }
} }
// TestASHandlerSkipsItsLastWriteAfterTheStopDeadline also checks what Stop
// logs: the asnBatchSize entries of the held write and the entry for testASNB,
// handled since, are not written.
func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) { func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
t.Parallel() t.Parallel()
const wantEntries = asnBatchSize + 1 // Each message gives an update for its origin AS and one for each AS
// in its path: fill asnBatchSize+1, extra 2.
const wantUpdates = asnBatchSize + 1 + 2
var logs bytes.Buffer var logs bytes.Buffer
store := newBlockingStore() store := newBlockingStore()
h := newASHandler(store, newErrorLogger(&logs), longerThanAnyTest) h := newASHandler(store, newErrorLogger(&logs), longerThanAnyTest)
fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
stopWhileAWriteIsBlocked(t, store, h, h.done, stopWhileAWriteIsBlocked(t, store, h, h.done,
[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}}, []*ristypes.RISMessage{{Path: fill}},
&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}}) &ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})
store.mu.Lock() store.mu.Lock()
@@ -519,27 +489,26 @@ func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
t.Errorf("AS%d was written after the stop deadline", testASNB) t.Errorf("AS%d was written after the stop deadline", testASNB)
} }
if logged := lastLogged(t, &logs); logged.BatchEntries != wantEntries { if logged := lastLogged(t, &logs); logged.Updates != wantUpdates {
t.Errorf("Stop logged %d batch entries not written, want %d", t.Errorf("Stop logged %d updates not written, want %d",
logged.BatchEntries, wantEntries) logged.Updates, wantUpdates)
} }
} }
// TestASHandlerSkipsAFullBatchAfterTheStopDeadline hands the AS handler a full // TestASHandlerSkipsAFullBatchAfterTheStopDeadline hands the AS handler a full
// batch while a write is held, so the batch's wake-up of the flush goroutine // batch while a write is held, so the batch's wake-up of the flush goroutine
// waits. The full batch holds the fill's ASes again, and testASNB. The stop // waits. The stop deadline passes meanwhile. When the write ends, the flush
// deadline passes meanwhile. When the write ends, the flush goroutine must stop // goroutine must stop instead of writing the batch.
// instead of writing the batch.
func TestASHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) { func TestASHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) {
t.Parallel() t.Parallel()
store := newBlockingStore() store := newBlockingStore()
h := newASHandler(store, logger.New(), longerThanAnyTest) h := newASHandler(store, logger.New(), longerThanAnyTest)
fill := &ristypes.RISMessage{Path: pathOfASes(asnBatchSize)} fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
full := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNB}, asnBatchSize)}
stopWhileAWriteIsBlocked(t, store, h, h.done, stopWhileAWriteIsBlocked(t, store, h, h.done,
[]*ristypes.RISMessage{fill}, []*ristypes.RISMessage{{Path: fill}},
fill, &ristypes.RISMessage{Path: full})
&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})
store.mu.Lock() store.mu.Lock()
defer store.mu.Unlock() defer store.mu.Unlock()
@@ -628,7 +597,7 @@ func TestPeeringHandlerStopCountsTheWriteInProgress(t *testing.T) {
select { select {
case <-h.done: case <-h.done:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatal("the processing goroutine did not return after the write " + t.Fatal("the processing goroutine did not return after the write " +
"finished") "finished")
} }
@@ -649,7 +618,7 @@ func TestPeeringHandlerStartsNoWriteAfterTheStopDeadline(t *testing.T) {
const processInterval = 10 * time.Millisecond const processInterval = 10 * time.Millisecond
store := newBlockingStore() store := newBlockingStore()
h := newPeeringHandler(store, logger.New(), processInterval, maxTrackedPaths) h := newPeeringHandler(store, logger.New(), processInterval)
stopWhileAWriteIsBlocked(t, store, h, h.done, stopWhileAWriteIsBlocked(t, store, h, h.done,
[]*ristypes.RISMessage{ []*ristypes.RISMessage{
{Path: ristypes.ASPath{Sequence: []int{testASNA, testASNB}}}, {Path: ristypes.ASPath{Sequence: []int{testASNA, testASNB}}},
@@ -705,8 +674,9 @@ func TestASHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
store := newBlockingStore() store := newBlockingStore()
h := newASHandler(store, logger.New(), longerThanAnyTest) h := newASHandler(store, logger.New(), longerThanAnyTest)
fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
handleWhileAWriteIsBlocked(t, store, h, handleWhileAWriteIsBlocked(t, store, h,
[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}}, []*ristypes.RISMessage{{Path: fill}},
&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}}) &ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})
store.mu.Lock() store.mu.Lock()
@@ -744,11 +714,9 @@ func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
const extraPrefix = "198.51.100.0/24" const extraPrefix = "198.51.100.0/24"
store := newBlockingStore() store := newBlockingStore()
h := newPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}), h := NewPrefixHandler(store, logger.New(), routetable.New(routetable.Params{}))
longerThanAnyTest, maxTableChangesPerWrite)
h.batchMaxSize = prefixBatchSize
fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize) fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchSize) toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchMaxSize)
handleAtMaxBatchSize(t, store, h, handleAtMaxBatchSize(t, store, h,
[]*ristypes.RISMessage{prefixAnnouncement(fill)}, []*ristypes.RISMessage{prefixAnnouncement(fill)},
[]*ristypes.RISMessage{prefixAnnouncement(toMax)}, []*ristypes.RISMessage{prefixAnnouncement(toMax)},
@@ -763,48 +731,31 @@ func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
} }
} }
// TestASHandlerNeverWaitsForAWrite starts a write that the store holds. Then it func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
// hands the AS handler twenty messages whose paths each hold one AS
// asnBatchSize times. HandleMessage must return each time, and the batch must
// hold one entry, for that AS.
func TestASHandlerNeverWaitsForAWrite(t *testing.T) {
t.Parallel() t.Parallel()
const messages = 20
store := newBlockingStore() store := newBlockingStore()
h := newASHandler(store, logger.New(), longerThanAnyTest) h := newASHandler(store, logger.New(), longerThanAnyTest)
fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchMaxSize)}
handleAtMaxBatchSize(t, store, h,
[]*ristypes.RISMessage{{Path: fill}},
[]*ristypes.RISMessage{{Path: toMax}},
&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})
defer stopInTime(t, h) store.mu.Lock()
defer store.releaseWrites() defer store.mu.Unlock()
startBlockedWrite(t, store, h, if !store.asns[testASNB] {
[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}}) t.Errorf("Stop did not write AS%d, handled after the batch reached its "+
"maximum size", testASNB)
oneAS := &ristypes.RISMessage{
Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNB}, asnBatchSize)},
}
if !returnsInTime(func() {
for range messages {
h.HandleMessage(oneAS)
}
}) {
t.Fatal("HandleMessage did not return while a database write was blocked")
}
h.mu.Lock()
defer h.mu.Unlock()
if len(h.batch) != 1 {
t.Errorf("the batch holds %d entries for one AS, want 1", len(h.batch))
} }
} }
// TestPeerHandlerWaitsAtTheMaximumBatchSize must not have its flush timer fire. // TestPeerHandlerWaitsAtTheMaximumBatchSize must not have its flush timer fire.
// A timer flush during the fill would take only part of it. The batch would // A timer flush during the fill would take only part of it. The batch would
// then reach its maximum size before the test's last message, and the test // then reach its maximum size before the test's last message, and the test
// would fail. // would hang.
func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) { func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
t.Parallel() t.Parallel()
@@ -812,11 +763,10 @@ func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
store := newBlockingStore() store := newBlockingStore()
h := newPeerHandler(store, logger.New(), longerThanAnyTest) h := newPeerHandler(store, logger.New(), longerThanAnyTest)
h.batchMaxSize = peerBatchSize
message := []*ristypes.RISMessage{{Peer: testPeerIP}} message := []*ristypes.RISMessage{{Peer: testPeerIP}}
handleAtMaxBatchSize(t, store, h, handleAtMaxBatchSize(t, store, h,
slices.Repeat(message, peerBatchSize), slices.Repeat(message, peerBatchSize),
slices.Repeat(message, peerBatchSize), slices.Repeat(message, peerBatchMaxSize),
&ristypes.RISMessage{Peer: otherPeerIP}) &ristypes.RISMessage{Peer: otherPeerIP})
store.mu.Lock() store.mu.Lock()
@@ -846,7 +796,6 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
batchSize: prefixBatchSize, batchSize: prefixBatchSize,
handle: func(n int) chan struct{} { handle: func(n int) chan struct{} {
h := &PrefixHandler{ h := &PrefixHandler{
batchMaxSize: prefixBatchMaxSize,
batchFullCh: make(chan struct{}, 1), batchFullCh: make(chan struct{}, 1),
table: routetable.New(routetable.Params{}), table: routetable.New(routetable.Params{}),
} }
@@ -860,12 +809,11 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
name: "AS", name: "AS",
batchSize: asnBatchSize, batchSize: asnBatchSize,
handle: func(n int) chan struct{} { handle: func(n int) chan struct{} {
h := &ASHandler{ h := &ASHandler{batchFullCh: make(chan struct{}, 1)}
batch: make(map[int]database.SeenTimes), // The origin AS and each AS in the path are one update each.
batchFullCh: make(chan struct{}, 1), h.HandleMessage(&ristypes.RISMessage{
} Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, n-1)},
// Each of the path's n ASes is one entry of the batch. })
h.HandleMessage(&ristypes.RISMessage{Path: pathOfASes(n)})
return h.batchFullCh return h.batchFullCh
}, },
@@ -874,10 +822,7 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
name: "peer", name: "peer",
batchSize: peerBatchSize, batchSize: peerBatchSize,
handle: func(n int) chan struct{} { handle: func(n int) chan struct{} {
h := &PeerHandler{ h := &PeerHandler{batchFullCh: make(chan struct{}, 1)}
batchMaxSize: peerBatchMaxSize,
batchFullCh: make(chan struct{}, 1),
}
for range n { for range n {
h.HandleMessage(&ristypes.RISMessage{Peer: testPeerIP}) h.HandleMessage(&ristypes.RISMessage{Peer: testPeerIP})
} }
@@ -97,7 +97,7 @@ func TestDBMaintainerStopReturnsByItsDeadline(t *testing.T) {
select { select {
case <-store.started: case <-store.started:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatal("the maintainer did not start the checkpoint") t.Fatal("the maintainer did not start the checkpoint")
} }
@@ -113,7 +113,7 @@ func TestDBMaintainerStopReturnsByItsDeadline(t *testing.T) {
select { select {
case <-m.done: case <-m.done:
case <-time.After(stepTimeout): case <-time.After(handleMessageTimeout):
t.Fatal("the maintenance goroutine did not return after the " + t.Fatal("the maintenance goroutine did not return after the " +
"checkpoint ended") "checkpoint ended")
} }
+3 -8
View File
@@ -43,14 +43,10 @@ type PeerHandler struct {
db database.Store db database.Store
logger *logger.Logger logger *logger.Logger
// batchMaxSize is peerBatchMaxSize. Only HandleMessage reads it, so a test
// may lower it before it hands the handler a message.
batchMaxSize int
// Batching. mu guards peerBatch and writing only: the flush goroutine // Batching. mu guards peerBatch and writing only: the flush goroutine
// takes the batch under mu and writes it after releasing mu, so // takes the batch under mu and writes it after releasing mu, so
// HandleMessage waits for a database write only while peerBatch is at // HandleMessage waits for a database write only while peerBatch is at
// batchMaxSize. // peerBatchMaxSize.
mu sync.Mutex mu sync.Mutex
peerBatch []peerUpdate peerBatch []peerUpdate
// writing is how many updates the flush goroutine took from peerBatch // writing is how many updates the flush goroutine took from peerBatch
@@ -90,7 +86,6 @@ func newPeerHandler(
h := &PeerHandler{ h := &PeerHandler{
db: db, db: db,
logger: logger, logger: logger,
batchMaxSize: peerBatchMaxSize,
peerBatch: make([]peerUpdate, 0, peerBatchSize), peerBatch: make([]peerUpdate, 0, peerBatchSize),
batchFullCh: make(chan struct{}, 1), batchFullCh: make(chan struct{}, 1),
stopCh: make(chan context.Context, 1), stopCh: make(chan context.Context, 1),
@@ -140,7 +135,7 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
// While the batch is at its maximum size, wait until the flush goroutine // While the batch is at its maximum size, wait until the flush goroutine
// takes it. Meanwhile the handler takes nothing off its queue, which // takes it. Meanwhile the handler takes nothing off its queue, which
// fills, and the streamer drops messages for it. // fills, and the streamer drops messages for it.
for len(h.peerBatch) >= h.batchMaxSize { for len(h.peerBatch) >= peerBatchMaxSize {
h.batchTaken.Wait() h.batchTaken.Wait()
} }
@@ -154,7 +149,7 @@ func (h *PeerHandler) HandleMessage(msg *ristypes.RISMessage) {
// A full batch wakes the flush goroutine. If it is writing, the wake-up // A full batch wakes the flush goroutine. If it is writing, the wake-up
// waits in batchFullCh and the batch keeps growing until the write ends // waits in batchFullCh and the batch keeps growing until the write ends
// or the batch reaches batchMaxSize. // or the batch reaches peerBatchMaxSize.
if len(h.peerBatch) >= peerBatchSize { if len(h.peerBatch) >= peerBatchSize {
select { select {
case h.batchFullCh <- struct{}{}: case h.batchFullCh <- struct{}{}:
@@ -35,7 +35,7 @@ func TestPeerHandlerWritesTheEarliestAndLatestTimesOfANewPeer(t *testing.T) {
// Built without NewPeerHandler's flush goroutine, so the test flushes // Built without NewPeerHandler's flush goroutine, so the test flushes
// itself. // itself.
h := &PeerHandler{db: store, logger: logger.New(), batchMaxSize: peerBatchMaxSize} h := &PeerHandler{db: store, logger: logger.New()}
first := peerAnnouncement(testPeerIP, testV4Prefix) first := peerAnnouncement(testPeerIP, testV4Prefix)
second := peerAnnouncement(testPeerIP, testV4Prefix) second := peerAnnouncement(testPeerIP, testV4Prefix)
+4 -11
View File
@@ -42,10 +42,6 @@ type PeeringHandler struct {
db database.Store db database.Store
logger *logger.Logger logger *logger.Logger
// maxPaths is the most paths asPaths holds: maxTrackedPaths, except in
// tests that give a smaller one.
maxPaths int
// In-memory AS path tracking: each path, JSON-encoded, with the earliest // In-memory AS path tracking: each path, JSON-encoded, with the earliest
// and the latest time an update carried it. // and the latest time an update carried it.
mu sync.Mutex mu sync.Mutex
@@ -66,22 +62,19 @@ type PeeringHandler struct {
// periodically processes accumulated AS paths into peering records. The // periodically processes accumulated AS paths into peering records. The
// handler begins processing immediately upon creation. // handler begins processing immediately upon creation.
func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler { func NewPeeringHandler(db database.Store, logger *logger.Logger) *PeeringHandler {
return newPeeringHandler(db, logger, peeringProcessInterval, maxTrackedPaths) return newPeeringHandler(db, logger, peeringProcessInterval)
} }
// newPeeringHandler is NewPeeringHandler with the processing timer's interval, // newPeeringHandler is NewPeeringHandler with the processing timer's interval,
// processInterval, in place of peeringProcessInterval, and the most paths it // processInterval, in place of peeringProcessInterval. A test gives a short
// holds, maxPaths, in place of maxTrackedPaths. A test gives a short interval to // one to have the timer start a write.
// have the timer start a write, and a small maxPaths to fill the map quickly.
func newPeeringHandler( func newPeeringHandler(
db database.Store, logger *logger.Logger, processInterval time.Duration, db database.Store, logger *logger.Logger, processInterval time.Duration,
maxPaths int,
) *PeeringHandler { ) *PeeringHandler {
h := &PeeringHandler{ h := &PeeringHandler{
db: db, db: db,
logger: logger, logger: logger,
asPaths: make(map[string]database.SeenTimes), asPaths: make(map[string]database.SeenTimes),
maxPaths: maxPaths,
stopCh: make(chan context.Context, 1), stopCh: make(chan context.Context, 1),
done: make(chan struct{}), done: make(chan struct{}),
} }
@@ -138,7 +131,7 @@ func (h *PeeringHandler) HandleMessage(msg *ristypes.RISMessage) {
defer h.mu.Unlock() defer h.mu.Unlock()
seen, exists := h.asPaths[key] seen, exists := h.asPaths[key]
if !exists && len(h.asPaths) >= h.maxPaths { if !exists && len(h.asPaths) >= maxTrackedPaths {
// Map is full; drop this new path and count it. // Map is full; drop this new path and count it.
h.droppedPaths++ h.droppedPaths++
@@ -5,6 +5,7 @@ import (
"fmt" "fmt"
"maps" "maps"
"slices" "slices"
"strconv"
"sync" "sync"
"testing" "testing"
"time" "time"
@@ -19,10 +20,6 @@ const (
testASNB = 64501 testASNB = 64501
testASNC = 64502 testASNC = 64502
testASND = 64503 testASND = 64503
// testMaxTrackedPaths is a limit of the path map that a test fills
// quickly, in place of maxTrackedPaths.
testMaxTrackedPaths = 10
) )
// recordingStore wraps mockStore to count every RecordPeering call, so a // recordingStore wraps mockStore to count every RecordPeering call, so a
@@ -58,7 +55,6 @@ func newTestHandler(db database.Store) *PeeringHandler {
db: db, db: db,
logger: logger.New(), logger: logger.New(),
asPaths: make(map[string]database.SeenTimes), asPaths: make(map[string]database.SeenTimes),
maxPaths: maxTrackedPaths,
} }
} }
@@ -248,28 +244,25 @@ func TestPeeringHandlerStopWritesTheRemainingPaths(t *testing.T) {
// TestPeeringHandlerCapDropsAndCounts verifies that a full map drops new paths // TestPeeringHandlerCapDropsAndCounts verifies that a full map drops new paths
// and counts them, while a path already tracked is refreshed rather than // and counts them, while a path already tracked is refreshed rather than
// dropped. The handler's limit is small, and its timer does not fire. // dropped.
func TestPeeringHandlerCapDropsAndCounts(t *testing.T) { func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
t.Parallel() t.Parallel()
store := &recordingStore{mockStore: newMockStore()} store := &recordingStore{mockStore: newMockStore()}
h := newPeeringHandler(store, logger.New(), longerThanAnyTest, h := newTestHandler(store)
testMaxTrackedPaths)
defer stopInTime(t, h)
now := time.Now().UTC() now := time.Now().UTC()
// Fill the map to exactly its limit, including one real path key so the // Fill the map to exactly maxTrackedPaths, including one real path key so
// "already tracked" branch can be exercised. The other paths are of one // the "already tracked" branch can be exercised. The filler keys are never
// AS, so they carry no peering. // processed in this test, so their contents do not matter.
existing := pathKey(t, testASNA, testASNB) existing := pathKey(t, testASNA, testASNB)
h.mu.Lock() h.mu.Lock()
h.asPaths[existing] = database.SeenTimes{First: now, Last: now} h.asPaths[existing] = database.SeenTimes{First: now, Last: now}
for asn := 1; len(h.asPaths) < testMaxTrackedPaths; asn++ { for i := 0; len(h.asPaths) < maxTrackedPaths; i++ {
h.asPaths[pathKey(t, asn)] = database.SeenTimes{First: now, Last: now} h.asPaths[strconv.Itoa(i)] = database.SeenTimes{First: now, Last: now}
} }
h.mu.Unlock() h.mu.Unlock()
@@ -277,9 +270,8 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
handle(h, now.Add(time.Second), testASNA, testASNC) handle(h, now.Add(time.Second), testASNA, testASNC)
tracked, dropped := snapshot(h) tracked, dropped := snapshot(h)
if tracked != testMaxTrackedPaths { if tracked != maxTrackedPaths {
t.Fatalf("want map size %d after drop, got %d", testMaxTrackedPaths, t.Fatalf("want map size %d after drop, got %d", maxTrackedPaths, tracked)
tracked)
} }
if dropped != 1 { if dropped != 1 {
@@ -291,9 +283,8 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
handle(h, refreshed, testASNA, testASNB) handle(h, refreshed, testASNA, testASNB)
tracked, dropped = snapshot(h) tracked, dropped = snapshot(h)
if tracked != testMaxTrackedPaths { if tracked != maxTrackedPaths {
t.Fatalf("want map size %d after refresh, got %d", testMaxTrackedPaths, t.Fatalf("want map size %d after refresh, got %d", maxTrackedPaths, tracked)
tracked)
} }
if dropped != 1 { if dropped != 1 {
@@ -312,21 +303,17 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
// TestPeeringHandlerRunResetsTheDroppedCount fills the map, has the handler // TestPeeringHandlerRunResetsTheDroppedCount fills the map, has the handler
// drop one path, and runs the processing. The count of dropped paths is the // drop one path, and runs the processing. The count of dropped paths is the
// run's own, so it is 0 again after the run. The handler's limit is small, and // run's own, so it is 0 again after the run.
// its timer does not fire.
func TestPeeringHandlerRunResetsTheDroppedCount(t *testing.T) { func TestPeeringHandlerRunResetsTheDroppedCount(t *testing.T) {
t.Parallel() t.Parallel()
h := newPeeringHandler(newMockStore(), logger.New(), longerThanAnyTest, h := newTestHandler(newMockStore())
testMaxTrackedPaths)
defer stopInTime(t, h)
now := time.Now().UTC() now := time.Now().UTC()
// Paths of one AS carry no peering, so the run records nothing. // Paths of one AS carry no peering, so the run records nothing.
h.mu.Lock() h.mu.Lock()
for asn := 1; len(h.asPaths) < testMaxTrackedPaths; asn++ { for asn := 1; len(h.asPaths) < maxTrackedPaths; asn++ {
h.asPaths[pathKey(t, asn)] = database.SeenTimes{First: now, Last: now} h.asPaths[pathKey(t, asn)] = database.SeenTimes{First: now, Last: now}
} }
h.mu.Unlock() h.mu.Unlock()
+12 -24
View File
@@ -60,17 +60,10 @@ type PrefixHandler struct {
metrics *metrics.Tracker metrics *metrics.Tracker
table *routetable.Table table *routetable.Table
// batchMaxSize is prefixBatchMaxSize. Only HandleMessage reads it, so a
// test may lower it before it hands the handler a message.
batchMaxSize int
// tableChangesPerWrite is maxTableChangesPerWrite, except in tests that
// give a smaller one.
tableChangesPerWrite int
// Batching. mu guards batch, writing and writingTableChanges only: the // Batching. mu guards batch, writing and writingTableChanges only: the
// flush goroutine takes them under mu and writes them after releasing // flush goroutine takes them under mu and writes them after releasing
// mu, so HandleMessage waits for a database write only while batch is at // mu, so HandleMessage waits for a database write only while batch is at
// batchMaxSize. // prefixBatchMaxSize.
mu sync.Mutex mu sync.Mutex
batch []seenPrefix batch []seenPrefix
// writing is how many updates the flush goroutine took and has not // writing is how many updates the flush goroutine took and has not
@@ -126,27 +119,21 @@ func (u prefixUpdate) announcesRoute() bool {
func NewPrefixHandler( func NewPrefixHandler(
db database.Store, logger *logger.Logger, table *routetable.Table, db database.Store, logger *logger.Logger, table *routetable.Table,
) *PrefixHandler { ) *PrefixHandler {
return newPrefixHandler(db, logger, table, prefixBatchTimeout, return newPrefixHandler(db, logger, table, prefixBatchTimeout)
maxTableChangesPerWrite)
} }
// newPrefixHandler is NewPrefixHandler with the flush timer's interval, // newPrefixHandler is NewPrefixHandler with the flush timer's interval,
// batchTimeout, in place of prefixBatchTimeout, and tableChangesPerWrite in // batchTimeout, in place of prefixBatchTimeout. A test that the timer must not
// place of maxTableChangesPerWrite. A test that the timer must not disturb // disturb gives an interval longer than the test runs.
// gives an interval longer than the test runs. A test of writes split over
// several flushes gives a small tableChangesPerWrite, so that it exceeds it
// with few changes.
func newPrefixHandler( func newPrefixHandler(
db database.Store, logger *logger.Logger, table *routetable.Table, db database.Store, logger *logger.Logger, table *routetable.Table,
batchTimeout time.Duration, tableChangesPerWrite int, batchTimeout time.Duration,
) *PrefixHandler { ) *PrefixHandler {
h := &PrefixHandler{ h := &PrefixHandler{
db: db, db: db,
logger: logger, logger: logger,
table: table, table: table,
batch: make([]seenPrefix, 0, prefixBatchSize), batch: make([]seenPrefix, 0, prefixBatchSize),
batchMaxSize: prefixBatchMaxSize,
tableChangesPerWrite: tableChangesPerWrite,
batchFullCh: make(chan struct{}, 1), batchFullCh: make(chan struct{}, 1),
stopCh: make(chan context.Context, 1), stopCh: make(chan context.Context, 1),
done: make(chan struct{}), done: make(chan struct{}),
@@ -204,7 +191,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
// While the batch is at its maximum size, wait until the flush goroutine // While the batch is at its maximum size, wait until the flush goroutine
// takes it. Meanwhile the handler takes nothing off its queue, which // takes it. Meanwhile the handler takes nothing off its queue, which
// fills, and the streamer drops messages for it. // fills, and the streamer drops messages for it.
for len(h.batch) >= h.batchMaxSize { for len(h.batch) >= prefixBatchMaxSize {
h.batchTaken.Wait() h.batchTaken.Wait()
} }
@@ -252,7 +239,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
// A full batch wakes the flush goroutine. If it is writing, the wake-up // A full batch wakes the flush goroutine. If it is writing, the wake-up
// waits in batchFullCh and the batch keeps growing until the write ends // waits in batchFullCh and the batch keeps growing until the write ends
// or the batch reaches batchMaxSize. // or the batch reaches prefixBatchMaxSize.
if len(h.batch) >= prefixBatchSize { if len(h.batch) >= prefixBatchSize {
select { select {
case h.batchFullCh <- struct{}{}: case h.batchFullCh <- struct{}{}:
@@ -263,7 +250,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
// Stop stops the flush goroutine, which makes one last flush before it returns, // Stop stops the flush goroutine, which makes one last flush before it returns,
// and waits for it until ctx ends. That flush writes at most // and waits for it until ctx ends. That flush writes at most
// tableChangesPerWrite of the routing table's changes, and the prefixes of // maxTableChangesPerWrite of the routing table's changes, and the prefixes of
// the updates the handler holds. Once ctx has ended the flush goroutine skips // the updates the handler holds. Once ctx has ended the flush goroutine skips
// that last flush, and a write still running is lost when the process exits. A // that last flush, and a write still running is lost when the process exits. A
// failed write of the table's changes is not retried either. Stop logs how many // failed write of the table's changes is not retried either. Stop logs how many
@@ -391,13 +378,14 @@ func (h *PrefixHandler) applyToTable(update prefixUpdate) {
}) })
} }
// writeTableChanges writes at most tableChangesPerWrite changes of the routing // writeTableChanges writes at most maxTableChangesPerWrite changes of the
// table in memory, and every peer the table numbered since the last write. // routing table in memory, and every peer the table numbered since the last
// write.
// When the write fails, they go back to the table, so that a later write writes // When the write fails, they go back to the table, so that a later write writes
// them again, each pair with its content then. // them again, each pair with its content then.
func (h *PrefixHandler) writeTableChanges() { func (h *PrefixHandler) writeTableChanges() {
h.mu.Lock() h.mu.Lock()
changes, newPeers := h.table.TakeChanges(h.tableChangesPerWrite) changes, newPeers := h.table.TakeChanges(maxTableChangesPerWrite)
h.writingTableChanges = len(changes) h.writingTableChanges = len(changes)
h.mu.Unlock() h.mu.Unlock()
@@ -237,10 +237,9 @@ func TestPrefixHandlerStopLogsNothingWhenItWroteEverything(t *testing.T) {
} }
} }
// longerThanAnyTest is a timer interval that no test lasts. A test gives it to // longerThanAnyTest is a flush timer interval that no test lasts. A test whose
// newPrefixHandler, newASHandler, newPeerHandler or newPeeringHandler when the // prefix, AS or peer handler must flush only when its batch is full and on Stop
// handler must write only when its batch is full, when the test says, or on // gives it to newPrefixHandler, newASHandler or newPeerHandler.
// Stop.
const longerThanAnyTest = time.Hour const longerThanAnyTest = time.Hour
// prefixHandlerWithoutTimer returns a prefix handler that writes to store and // prefixHandlerWithoutTimer returns a prefix handler that writes to store and
@@ -251,8 +250,6 @@ func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
db: store, db: store,
logger: logger.New(), logger: logger.New(),
table: routetable.New(routetable.Params{}), table: routetable.New(routetable.Params{}),
batchMaxSize: prefixBatchMaxSize,
tableChangesPerWrite: maxTableChangesPerWrite,
} }
} }
@@ -30,13 +30,12 @@ const (
// firstManyPrefix is the first prefix announceManyPrefixes announces. // firstManyPrefix is the first prefix announceManyPrefixes announces.
firstManyPrefix = "10.0.0.0/32" firstManyPrefix = "10.0.0.0/32"
// testMaxTableChangesPerWrite is a limit of one write of the routing
// table's changes that a test exceeds quickly, in place of
// maxTableChangesPerWrite.
testMaxTableChangesPerWrite = 10
) )
// tableWriteTimeout is how long a test waits for the flush timer to write the
// routing table's changes: a few ticks of it.
const tableWriteTimeout = 5 * prefixBatchTimeout
// tableCase is one test of the routing table in memory: the messages handed to // tableCase is one test of the routing table in memory: the messages handed to
// the prefix handler, and for each prefix the peers that carry it with each // the prefix handler, and for each prefix the peers that carry it with each
// origin AS after them. // origin AS after them.
@@ -277,9 +276,9 @@ func TestPrefixHandlerWritesTheTableChangesEachSecond(t *testing.T) {
t.Fatalf("wrote %+v, want the pair of %s and AS%d with "+ t.Fatalf("wrote %+v, want the pair of %s and AS%d with "+
"removed %t", changes, prefix, testASNB, step.wantRemoved) "removed %t", changes, prefix, testASNB, step.wantRemoved)
} }
case <-time.After(stepTimeout): case <-time.After(tableWriteTimeout):
t.Fatalf("the routing table's change was not written within %s", t.Fatalf("the routing table's change was not written within %s",
stepTimeout) tableWriteTimeout)
} }
} }
} }
@@ -376,32 +375,25 @@ func (s *tableWriteSizesStore) WriteRouteChanges(
// takes, then flushes three times. The first write must hold as many pairs as // takes, then flushes three times. The first write must hold as many pairs as
// one write takes, and the peer. The second must hold the other pairs, and the // one write takes, and the peer. The second must hold the other pairs, and the
// third flush must write nothing. The database must then hold the table in // third flush must write nothing. The database must then hold the table in
// memory. The handler's limit is small, and its timer does not fire. // memory.
func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) { func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
t.Parallel() t.Parallel()
const ( const (
pairs = testMaxTableChangesPerWrite * 3 / 2 pairs = maxTableChangesPerWrite * 3 / 2
flushes = 3 flushes = 3
) )
db := newTestDatabase(t) db := newTestDatabase(t)
store := &tableWriteSizesStore{Database: db} store := &tableWriteSizesStore{Database: db}
table := routetable.New(routetable.Params{}) h := prefixHandlerWithoutTimer(store)
announceManyPrefixes(table, pairs) announceManyPrefixes(h.table, pairs)
h := newPrefixHandler(store, logger.New(), table, longerThanAnyTest,
testMaxTableChangesPerWrite)
defer stopInTime(t, h)
for range flushes { for range flushes {
h.flushBatch() h.flushBatch()
} }
wantChanges := []int{ wantChanges := []int{maxTableChangesPerWrite, pairs - maxTableChangesPerWrite}
testMaxTableChangesPerWrite, pairs - testMaxTableChangesPerWrite,
}
wantNewPeers := []int{1, 0} wantNewPeers := []int{1, 0}
if !slices.Equal(store.changes, wantChanges) || if !slices.Equal(store.changes, wantChanges) ||
@@ -423,11 +415,11 @@ func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
// takes. Its last flush writes one write's worth, so Stop must log the pairs // takes. Its last flush writes one write's worth, so Stop must log the pairs
// that no write took, and must not say that the stop deadline passed. The // that no write took, and must not say that the stop deadline passed. The
// flush timer must not fire, or its writes could leave no pairs for Stop to // flush timer must not fire, or its writes could leave no pairs for Stop to
// log. The handler's limit of one write is small. // log.
func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) { func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
t.Parallel() t.Parallel()
const pairs = testMaxTableChangesPerWrite + 1 const pairs = maxTableChangesPerWrite + 1
var logs bytes.Buffer var logs bytes.Buffer
@@ -435,8 +427,7 @@ func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
table := routetable.New(routetable.Params{}) table := routetable.New(routetable.Params{})
announceManyPrefixes(table, pairs) announceManyPrefixes(table, pairs)
h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest, h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest)
testMaxTableChangesPerWrite)
h.Stop(t.Context()) h.Stop(t.Context())
written := 0 written := 0
-61
View File
@@ -6,7 +6,6 @@ import (
"strconv" "strconv"
"strings" "strings"
"testing" "testing"
"testing/synctest"
"time" "time"
"git.eeqj.de/sneak/routewatch/internal/database" "git.eeqj.de/sneak/routewatch/internal/database"
@@ -342,66 +341,6 @@ func TestStartedServerWorksOutTheCards(t *testing.T) {
} }
} }
// TestStartedServerWorksOutTheCardsUntilStop starts a server with a context
// that ends as soon as Start returns. It runs in a synctest bubble, whose clock
// is fake: time.Sleep there moves the clock on and returns at once. The server
// must work the cards out again a minute later, and no more after Stop. Each
// time it works them out, it keeps a new copy. Before the fix the end of
// Start's context stopped the goroutine that works them out.
func TestStartedServerWorksOutTheCardsUntilStop(t *testing.T) {
t.Parallel()
// The server is made outside the bubble. The streamer's metrics start a
// goroutine that never returns, and a bubble waits for each of its
// goroutines to return.
s := newServerWithTable(t, newCardsTable())
synctest.Test(t, func(t *testing.T) {
ctx, cancel := context.WithCancel(t.Context())
err := s.Start(ctx)
if err != nil {
t.Fatalf("Start: %v", err)
}
cancel()
// The bubble's clock moves only while each of its goroutines waits
// on the bubble, and the goroutine that serves HTTP waits on the
// network. The test sends no request, so it closes the HTTP server.
_ = s.srv.Close()
synctest.Wait()
started := s.cards.Load()
if started == nil {
t.Fatal("the server did not work out the cards when it started")
}
time.Sleep(homeCardsRefreshInterval)
synctest.Wait()
if s.cards.Load() == started {
t.Error("the server did not work out the cards again a minute " +
"after its start context ended")
}
err = s.Stop(t.Context())
if err != nil {
t.Fatalf("Stop: %v", err)
}
stopped := s.cards.Load()
time.Sleep(homeCardsRefreshInterval)
synctest.Wait()
if s.cards.Load() != stopped {
t.Error("the server worked out the cards again after Stop")
}
})
}
// waitUntil fails the test unless done reports true within maxWait. // waitUntil fails the test unless done reports true within maxWait.
func waitUntil(t *testing.T, maxWait time.Duration, done func() bool) { func waitUntil(t *testing.T, maxWait time.Duration, done func() bool) {
t.Helper() t.Helper()
+4 -27
View File
@@ -82,12 +82,6 @@ type Server struct {
// cards is the copy of the home page's cards that the pages serve, nil // cards is the copy of the home page's cards that the pages serve, nil
// until the server first works them out // until the server first works them out
cards atomic.Pointer[homeCards] cards atomic.Pointer[homeCards]
// stopCards ends the goroutine that works out the cards, and
// cardsStopped is closed once it has returned. Both are nil until Start
// starts the goroutine.
stopCards context.CancelFunc
cardsStopped chan struct{}
} }
// New creates a new HTTP server that reads routes from table, the routing // New creates a new HTTP server that reads routes from table, the routing
@@ -117,8 +111,7 @@ func New(
// Start listens on the configured port and serves HTTP in the background. // Start listens on the configured port and serves HTTP in the background.
// It returns an error if it cannot listen, for example because the port is // It returns an error if it cannot listen, for example because the port is
// already in use. ctx bounds opening the port, not serving. Once listening, // already in use. ctx bounds opening the port, not serving. Once listening,
// the server works out the home page's cards every minute until Stop: the end // the server works out the home page's cards every minute until ctx ends.
// of ctx does not stop it.
func (s *Server) Start(ctx context.Context) error { func (s *Server) Start(ctx context.Context) error {
const ( const (
network = "tcp" network = "tcp"
@@ -155,22 +148,16 @@ func (s *Server) Start(ctx context.Context) error {
} }
}() }()
ctx, s.stopCards = context.WithCancel(context.WithoutCancel(ctx))
s.cardsStopped = make(chan struct{})
go func() { go func() {
defer close(s.cardsStopped)
s.keepHomeCardsFresh(ctx, homeCardsRefreshInterval) s.keepHomeCardsFresh(ctx, homeCardsRefreshInterval)
}() }()
return nil return nil
} }
// Stop stops the goroutine that works out the home page's cards, and waits // Stop stops the HTTP server. It waits, until ctx ends, for the requests in
// until it has returned or ctx ends. It then stops the HTTP server. It waits, // progress to finish, and then closes the connections of those still running
// until ctx ends, for the requests in progress to finish, and then closes the // and returns ctx's error.
// connections of those still running and returns ctx's error.
func (s *Server) Stop(ctx context.Context) error { func (s *Server) Stop(ctx context.Context) error {
if s.srv == nil { if s.srv == nil {
return nil return nil
@@ -178,16 +165,6 @@ func (s *Server) Stop(ctx context.Context) error {
s.logger.Info("Stopping HTTP server") s.logger.Info("Stopping HTTP server")
// Start returns before it starts the goroutine when it cannot listen.
if s.stopCards != nil {
s.stopCards()
select {
case <-s.cardsStopped:
case <-ctx.Done():
}
}
err := s.srv.Shutdown(ctx) err := s.srv.Shutdown(ctx)
if err != nil { if err != nil {
return errors.Join(err, s.srv.Close()) return errors.Join(err, s.srv.Close())
+36 -82
View File
@@ -12,7 +12,6 @@ import (
"regexp" "regexp"
"strings" "strings"
"sync" "sync"
"syscall"
"time" "time"
"unicode/utf8" "unicode/utf8"
@@ -71,12 +70,12 @@ const nicbrHeader = "% Copyright (c) Nic.br"
// queries to stop the data being taken in bulk. // queries to stop the data being taken in bulk.
const nicbrLimitReached = "Query rate limit exceeded" const nicbrLimitReached = "Query rate limit exceeded"
// PauseTime is how long the client sends an RIR's or NIC.br's server no // NICbrSkipTime is how long the client stops asking NIC.br after NIC.br
// query after that server refuses a query. queryRegistry says what a // answers that the client is over its query limit. NIC.br publishes
// refusal is. RIPE refuses an address for good when the address keeps // neither the limit nor how long it lasts. An hour outlasts a limit
// passing RIPE's limit (see noPersonalFlag). An hour outlasts a limit // counted per minute or per hour. The ASes looked up meanwhile get
// counted per minute or per hour. // NIC.br's fields when they are next refreshed.
const PauseTime = time.Hour const NICbrSkipTime = time.Hour
// RIR identifiers. // RIR identifiers.
const ( const (
@@ -126,10 +125,6 @@ var (
errAnswerTooLarge = errors.New("WHOIS answer reached the size cap") errAnswerTooLarge = errors.New("WHOIS answer reached the size cap")
) )
// ErrPaused is returned when a lookup needs a server that refused a query
// less than PauseTime ago. The client sends that server nothing meanwhile.
var ErrPaused = errors.New("WHOIS server paused after refusing a query")
// ASNInfo contains parsed WHOIS information for an ASN. // ASNInfo contains parsed WHOIS information for an ASN.
type ASNInfo struct { type ASNInfo struct {
ASN int ASN int
@@ -161,12 +156,12 @@ type Client struct {
logger *slog.Logger logger *slog.Logger
// mu guards pausedUntil. // mu guards nicbrSkipUntil.
mu sync.Mutex mu sync.Mutex
// pausedUntil maps a server, given as host:port, to the time the client // nicbrSkipUntil is when the client next asks NIC.br, after NIC.br
// may ask it again. // answered that the client is over its query limit.
pausedUntil map[string]time.Time nicbrSkipUntil time.Time
} }
// NewClient creates a new WHOIS client that logs to logger. // NewClient creates a new WHOIS client that logs to logger.
@@ -177,29 +172,23 @@ func NewClient(logger *slog.Logger) *Client {
DialContext: dialer.DialContext, DialContext: dialer.DialContext,
Now: time.Now, Now: time.Now,
logger: logger, logger: logger,
pausedUntil: make(map[string]time.Time),
} }
} }
// LookupASN asks IANA which RIR holds the given ASN, asks that RIR, and // LookupASN asks IANA which RIR holds the given ASN, asks that RIR, and
// returns what the RIR's answer says. When LACNIC's answer is NIC.br's, it // returns what the RIR's answer says. When LACNIC's answer is NIC.br's, it
// returns what NIC.br's own server says instead, unless that query fails or // returns what NIC.br's own server says instead, unless that query fails or
// NIC.br is paused, as askNICbr says. // is skipped, as askNICbr says.
// //
// It returns ErrNoReferral when IANA refers the AS to no RIR. Its error // It returns ErrNoReferral when IANA refers the AS to no RIR, and another
// wraps ErrPaused when the RIR is paused. It returns another error in each // error when the query to IANA or to the RIR fails, when the RIR refuses
// of these cases: // it, when IANA's answer has neither a referral nor the AS's block, when
// // IANA refers the AS to a server none of the five RIRs run, or when the
// - the query to IANA or to the RIR fails // answer carries none of ASNInfo's fields.
// - the RIR refuses the query
// - IANA's answer has neither a referral nor the AS's block
// - IANA refers the AS to a server none of the five RIRs run
// - the answer carries none of ASNInfo's fields
func (c *Client) LookupASN(ctx context.Context, asn int) (*ASNInfo, error) { func (c *Client) LookupASN(ctx context.Context, asn int) (*ASNInfo, error) {
// Query IANA first to find the authoritative RIR // Query IANA first to find the authoritative RIR
query := fmt.Sprintf("AS%d", asn) query := fmt.Sprintf("AS%d", asn)
// IANA is never paused, because a pause of IANA would stop every lookup.
ianaResp, err := c.query(ctx, whoisServerIANA, query) ianaResp, err := c.query(ctx, whoisServerIANA, query)
if err != nil { if err != nil {
return nil, fmt.Errorf("IANA query failed: %w", err) return nil, fmt.Errorf("IANA query failed: %w", err)
@@ -222,7 +211,7 @@ func (c *Client) LookupASN(ctx context.Context, asn int) (*ASNInfo, error) {
} }
// Query the authoritative RIR // Query the authoritative RIR
rirResp, err := c.queryRegistry(ctx, whoisServer, rirQuery(rir, query)) rirResp, err := c.queryRIR(ctx, whoisServer, rirQuery(rir, query))
if err != nil { if err != nil {
return nil, fmt.Errorf("RIR query failed: %w", err) return nil, fmt.Errorf("RIR query failed: %w", err)
} }
@@ -252,19 +241,22 @@ func (c *Client) LookupASN(ctx context.Context, asn int) (*ASNInfo, error) {
// askNICbr returns lacnicInfo, what LACNIC's answer says, and logs a // askNICbr returns lacnicInfo, what LACNIC's answer says, and logs a
// warning when the query fails, when NIC.br sends the reduced answer for a // warning when the query fails, when NIC.br sends the reduced answer for a
// client over its limit, or when NIC.br's answer carries none of ASNInfo's // client over its limit, or when NIC.br's answer carries none of ASNInfo's
// fields. The reduced answer pauses NIC.br, as a refusal does. While NIC.br // fields. After the reduced answer, it returns lacnicInfo without asking
// is paused, askNICbr returns lacnicInfo without a warning. The ASes looked // NIC.br until NICbrSkipTime has passed.
// up meanwhile get NIC.br's fields when they are next refreshed.
func (c *Client) askNICbr( func (c *Client) askNICbr(
ctx context.Context, asQuery string, lacnicInfo *ASNInfo, ctx context.Context, asQuery string, lacnicInfo *ASNInfo,
) *ASNInfo { ) *ASNInfo {
asn := lacnicInfo.ASN asn := lacnicInfo.ASN
answer, err := c.queryRegistry(ctx, whoisServerNICbr, asQuery) c.mu.Lock()
if errors.Is(err, ErrPaused) { skipping := c.Now().Before(c.nicbrSkipUntil)
c.mu.Unlock()
if skipping {
return lacnicInfo return lacnicInfo
} }
answer, err := c.query(ctx, whoisServerNICbr, asQuery)
if err != nil { if err != nil {
c.logger.Warn("NIC.br query failed, keeping LACNIC's fields", c.logger.Warn("NIC.br query failed, keeping LACNIC's fields",
"asn", asn, "error", err) "asn", asn, "error", err)
@@ -273,11 +265,13 @@ func (c *Client) askNICbr(
} }
if strings.Contains(answer, nicbrLimitReached) { if strings.Contains(answer, nicbrLimitReached) {
c.pause(whoisServerNICbr) c.mu.Lock()
c.nicbrSkipUntil = c.Now().Add(NICbrSkipTime)
c.mu.Unlock()
c.logger.Warn("NIC.br answered that the client is over its query "+ c.logger.Warn("NIC.br answered that the client is over its query "+
"limit, keeping LACNIC's fields and pausing NIC.br", "limit, keeping LACNIC's fields and skipping NIC.br",
"asn", asn, "pause_for", PauseTime) "asn", asn, "skip_for", NICbrSkipTime)
return lacnicInfo return lacnicInfo
} }
@@ -315,65 +309,25 @@ func rirQuery(rir, asQuery string) string {
} }
} }
// queryRegistry sends query to the server of an RIR or of NIC.br, and // queryRIR sends query to an RIR's server and returns the answer. An answer
// returns the answer. The client pauses the server for PauseTime when the // holding an error line with which the server refuses the query is an
// server refuses the query. Each of these is a refusal: // error, and the error holds that line.
// func (c *Client) queryRIR(
// - a refused connection
// - a reset connection
// - an %ERROR:201: or %ERROR:202: line in the answer
//
// A refusal is returned as an error. While the server is paused,
// queryRegistry sends it nothing and returns an error that wraps ErrPaused.
func (c *Client) queryRegistry(
ctx context.Context, server, query string, ctx context.Context, server, query string,
) (string, error) { ) (string, error) {
c.mu.Lock()
pausedUntil := c.pausedUntil[server]
c.mu.Unlock()
if c.Now().Before(pausedUntil) {
return "", fmt.Errorf("%w: %s until %s", ErrPaused, server,
pausedUntil.UTC().Format(time.RFC3339))
}
answer, err := c.query(ctx, server, query) answer, err := c.query(ctx, server, query)
if err == nil {
err = checkRefusal(answer)
}
if errors.Is(err, errRefused) || errors.Is(err, syscall.ECONNREFUSED) ||
errors.Is(err, syscall.ECONNRESET) {
c.pause(server)
}
if err != nil { if err != nil {
return "", err return "", err
} }
return answer, nil
}
// checkRefusal returns an error holding the first line of answer that
// starts with accessDeniedError or accessLimitError. It returns nil when
// answer has no such line.
func checkRefusal(answer string) error {
for line := range strings.SplitSeq(answer, "\n") { for line := range strings.SplitSeq(answer, "\n") {
if strings.HasPrefix(line, accessDeniedError) || if strings.HasPrefix(line, accessDeniedError) ||
strings.HasPrefix(line, accessLimitError) { strings.HasPrefix(line, accessLimitError) {
return fmt.Errorf("%w: %s", errRefused, line) return "", fmt.Errorf("%w: %s", errRefused, line)
} }
} }
return nil return answer, nil
}
// pause stops queryRegistry from querying server until PauseTime has passed.
func (c *Client) pause(server string) {
c.mu.Lock()
defer c.mu.Unlock()
c.pausedUntil[server] = c.Now().Add(PauseTime)
} }
// query performs a raw WHOIS query to the specified server and returns the // query performs a raw WHOIS query to the specified server and returns the
+5 -142
View File
@@ -718,10 +718,9 @@ func TestLookupASNKeepsLACNICsAnswerWhenNICbrFails(t *testing.T) {
// TestLookupASNSkipsNICbrAfterItsLimit looks up an AS that NIC.br // TestLookupASNSkipsNICbrAfterItsLimit looks up an AS that NIC.br
// registers three times with one client, while NIC.br answers that the // registers three times with one client, while NIC.br answers that the
// client is over its query limit. The test sets the client's clock. The // client is over its query limit. The first lookup asks NIC.br. The
// first lookup asks NIC.br. The next comes a second before PauseTime has // second, a second before NICbrSkipTime has passed, does not. The third,
// passed, and does not ask NIC.br. The last comes at PauseTime and asks // once NICbrSkipTime has passed, asks again. Each keeps LACNIC's fields.
// NIC.br again. Each keeps LACNIC's fields.
func TestLookupASNSkipsNICbrAfterItsLimit(t *testing.T) { func TestLookupASNSkipsNICbrAfterItsLimit(t *testing.T) {
t.Parallel() t.Parallel()
@@ -741,8 +740,8 @@ func TestLookupASNSkipsNICbrAfterItsLimit(t *testing.T) {
wantAsks int wantAsks int
}{ }{
{after: 0, wantAsks: 1}, {after: 0, wantAsks: 1},
{after: whois.PauseTime - time.Second, wantAsks: 1}, {after: whois.NICbrSkipTime - time.Second, wantAsks: 1},
{after: whois.PauseTime, wantAsks: 2}, {after: whois.NICbrSkipTime, wantAsks: 2},
} { } {
now = start.Add(step.after) now = start.Add(step.after)
@@ -786,142 +785,6 @@ func TestLookupASNReportsARefusal(t *testing.T) {
} }
} }
// resetServer starts a TCP server on the loopback address and returns the
// server's address. The server reads one query on each connection, then
// resets the connection.
func resetServer(t *testing.T) string {
t.Helper()
var listenConfig net.ListenConfig
listener, err := listenConfig.Listen(t.Context(), "tcp", "127.0.0.1:0")
if err != nil {
t.Fatalf("listen: %v", err)
}
t.Cleanup(func() { _ = listener.Close() })
go func() {
for {
conn, err := listener.Accept()
if err != nil {
return
}
_, _ = bufio.NewReader(conn).ReadString('\n')
// Closing a connection whose linger time is zero resets it.
if tcpConn, ok := conn.(*net.TCPConn); ok {
_ = tcpConn.SetLinger(0)
}
_ = conn.Close()
}
}()
return listener.Addr().String()
}
// newRIPEClient returns a client whose fake servers send answers. The
// client adds one to *ripeAsks each time it connects to RIPE's server. When
// reset is true, RIPE's server resets the connection instead of answering.
func newRIPEClient(
t *testing.T, answers map[string]string, reset bool, ripeAsks *int,
) *whois.Client {
t.Helper()
dial := fakeServers(t, answers)
resetAddress := resetServer(t)
client := whois.NewClient(slog.New(slog.DiscardHandler))
client.DialContext = func(
ctx context.Context, network, address string,
) (net.Conn, error) {
if address == ripeServer {
*ripeAsks++
}
if address == ripeServer && reset {
var dialer net.Dialer
return dialer.DialContext(ctx, network, resetAddress)
}
return dial(ctx, network, address)
}
return client
}
// TestLookupASNPausesAServerThatRefuses looks up an AS at RIPE while RIPE
// refuses the query, in each of the ways a server refuses. The test sets
// the client's clock. A lookup a second before PauseTime has passed fails
// with ErrPaused and asks RIPE nothing. A lookup of an AS at APNIC succeeds
// during the pause. A lookup at PauseTime asks RIPE again.
func TestLookupASNPausesAServerThatRefuses(t *testing.T) {
t.Parallel()
for name, test := range map[string]struct {
// ripeAnswer is RIPE's answer. With none, RIPE's server refuses
// the connection, or resets it when reset is true.
ripeAnswer string
reset bool
}{
"access denied": {ripeAnswer: ripeRefusal},
"limit reached": {ripeAnswer: ripeLimitReached + "\n"},
"refused connection": {},
"reset connection": {reset: true},
} {
t.Run(name, func(t *testing.T) {
t.Parallel()
answers := map[string]string{
ianaQuery: ianaReferral,
apnicIANAQuery: "refer: whois.apnic.net\n",
apnicQuery: apnicRecord,
}
if test.ripeAnswer != "" {
answers[ripeQuery] = test.ripeAnswer
}
ripeAsks := 0
client := newRIPEClient(t, answers, test.reset, &ripeAsks)
start := time.Date(2026, 11, 2, 0, 0, 0, 0, time.UTC)
now := start
client.Now = func() time.Time { return now }
_, err := client.LookupASN(t.Context(), testASN)
if err == nil || errors.Is(err, whois.ErrPaused) {
t.Fatalf("LookupASN while RIPE refuses: error %v, want a "+
"refusal", err)
}
now = start.Add(whois.PauseTime - time.Second)
_, err = client.LookupASN(t.Context(), testASN)
if !errors.Is(err, whois.ErrPaused) || ripeAsks != 1 {
t.Errorf("LookupASN during the pause: error %v after %d "+
"queries to RIPE, want ErrPaused after 1", err, ripeAsks)
}
_, err = client.LookupASN(t.Context(), apnicASN)
if err != nil {
t.Errorf("LookupASN of an AS at APNIC during RIPE's "+
"pause: %v", err)
}
now = start.Add(whois.PauseTime)
_, _ = client.LookupASN(t.Context(), testASN)
if ripeAsks != 2 {
t.Errorf("RIPE was asked %d times after the pause, want 2",
ripeAsks)
}
})
}
}
// TestLookupASNFailsWhenTheRIRGivesNoAnswer looks up an AS IANA refers to // TestLookupASNFailsWhenTheRIRGivesNoAnswer looks up an AS IANA refers to
// RIPE when RIPE's server refuses the connection, and when it answers only // RIPE when RIPE's server refuses the connection, and when it answers only
// with an error. Neither is a lookup that found the AS. // with an error. Neither is a lookup that found the AS.