Compare commits
1
Commits
next
..
d79b4169b6
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d79b4169b6 |
@@ -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.
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|||||||
@@ -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")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -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)
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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()
|
||||||
|
|||||||
@@ -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
@@ -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
|
||||||
|
|||||||
@@ -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.
|
||||||
|
|||||||
Reference in New Issue
Block a user