4 Commits
Author SHA1 Message Date
clawbot 38b43549b9 Make the routewatch tests pass on a loaded host (closes #213)
check / check (push) Waiting to run
Five prefix handler tests gave HandleMessage two seconds to apply a full
batch of 25,000 prefixes, which under the race detector on a loaded host
takes longer. A test now waits up to stepTimeout, 15 seconds, for each step
it expects.

The slowest tests filled production limits, such as the peering handler's
500,000 paths. The peering handler now takes its path limit, and the prefix
handler its limit on one write of the table's changes, as they take their
timer interval. Those tests give small limits, and the tests of the maximum
batch size lower it to one batch. This also fixes
#215.

The test helpers bound every wait, Stop included, so a test that hangs in
one fails with its own message.

Model: opus-5-5
2026-10-28 01:44:47 +01:00
clawbot 5ea1bb17fe Keep working out the home page cards after startup (closes #211)
check / check (push) Canceled after 0s
Server.Start ran the loop that works out the home page's cards on the
context it was given. fx ends that context once start-up is done, so the
cards were worked out once and never again. The loop now runs on a
context that only Stop ends, as the other background parts do. Stop
waits for the loop to return, until its own context ends, before it
stops the HTTP server.

A new test starts a server in a synctest bubble and ends its context at
once. On the bubble's fake clock, the cards must be worked out again a
minute later, and not after Stop.

Model: opus-5-5
2026-10-27 22:55:17 +01:00
clawbot 6d4f3e3419 Say in README why RouteWatch exists (closes #204)
check / check (push) Canceled after 0s
The Rationale section now gives the owner's reason for the program: it
is an internal API that our other services ask about IP addresses. It
sees which AS routes each address and which announced prefix holds it
from its own routing table. WHOIS details of each AS, kept in its
database, add to that.

What RIS Live does and does not give stays, shortened. Planned geojs
lookups are left out, because the README describes only what is built.

Model: opus-5-5
2026-10-27 21:34:08 +01:00
clawbot 47aed8761e Give every JSON answer @meta, errors included (closes #201)
check / check (push) Canceled after 0s
A panic in a JSON route's handler answered 500 with an empty body,
because the router's recoverer sits outside the middleware that adds
`@meta`. `JSONValidationMiddleware` now wraps the handler in the same
recoverer, and answers the panic with the JSON error body.

`/status.json` and the health check answered chi's empty 405 to any
method they do not take. They now answer with the same JSON 405 as the
routes under `/api/v1` and `/ip`.

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

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

Model: opus-5-5
2026-10-27 21:20:07 +01:00
17 changed files with 370 additions and 168 deletions
+35 -15
View File
@@ -9,19 +9,38 @@ table in real time. It works like this:
## Rationale ## Rationale
RIPE RIS Live streams the BGP updates that RIPE's route collectors receive, as RouteWatch exists to serve our other services as an internal API for information
they happen, from the moment a client connects. It reports changes, not the about IP addresses.
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.
The feed sends no table when a client connects. So the routing table holds only RouteWatch follows BGP routing table updates from RIPE RIS Live. RIS Live
the routes announced since RouteWatch first started. An IP-to-AS lookup finds streams the updates that RIPE's route collectors receive, as they happen, from
nothing for an address whose routes have not been announced since then. The the moment a client connects. It reports changes, not the routing table they add
routing table also misses updates RouteWatch drops when it falls behind, and up to. RouteWatch applies those changes to a routing table that it keeps in
updates sent while it is stopped. So a lookup or a prefix page can show a route memory, with a copy in its own SQLite database. From that table, RouteWatch sees
the feed has withdrawn. directly:
- 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
@@ -233,7 +252,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
no route matches answers 404 with the message `Not found`. A request with that matches no route 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:
@@ -241,8 +260,9 @@ 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`, and a host name that does not A 500 answer says only `Internal server error`. The error for a host name that
resolve only `could not resolve host name`; the detail goes to the log. does not resolve says only `could not resolve host name`. The detail goes to the
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
+19 -6
View File
@@ -35,13 +35,26 @@ tested.
# Completed Steps # Completed Steps
- 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. A method that `/status.json` or the health check 500 with the JSON error body. `/status.json` and the health check answer 405
does not take answers 405 with it, as under `/api/v1`. Both used to answer with the JSON error body to any method they do not take. The 500 and the 405
with an empty body. The home page's IP lookup shows the whole JSON answer, used to have an empty body. The home page's IP lookup shows the whole JSON
`@meta` included. It used to show only the answer's `data`, or only the answer, `@meta` included. It used to show only the answer's `data`, or only
message of an error answer. A test checks `@meta` on every JSON route, in its the message of an error answer. A test checks `@meta` on every JSON route, in
successes and its errors (closes #201) its 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,8 +509,6 @@ 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)
@@ -556,9 +554,9 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
for range cap(linesSeen) { for range cap(linesSeen) {
select { select {
case <-linesSeen: case <-linesSeen:
case <-time.After(maxWait): case <-time.After(stepTimeout):
t.Fatalf("the streamer did not read %d lines within %s", t.Fatalf("the streamer did not read %d lines within %s",
cap(linesSeen), maxWait) cap(linesSeen), stepTimeout)
} }
} }
@@ -571,8 +569,8 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
select { select {
case <-shutDown: case <-shutDown:
case <-time.After(maxWait): case <-time.After(stepTimeout):
t.Fatalf("Shutdown did not return within %s", maxWait) t.Fatalf("Shutdown did not return within %s", stepTimeout)
} }
store.mu.Lock() store.mu.Lock()
@@ -609,21 +607,16 @@ func freePort(t *testing.T) int {
return addr.Port return addr.Port
} }
// TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked starts the HTTP server // TestShutdownDoesNotWaitForABlockedWrite starts the HTTP server and has the
// and has the AS handler start a write that the store holds until the test // AS handler start a write that the store holds until the test ends, past
// ends, past Shutdown's deadline. Shutdown must return within its deadline // Shutdown's deadline. Shutdown must return without waiting for the write, with
// plus a small margin, with the HTTP server stopped, and must neither // the HTTP server stopped, and must neither checkpoint nor close the database:
// checkpoint nor close the database: the real one would wait for the write. // 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 TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) { func TestShutdownDoesNotWaitForABlockedWrite(t *testing.T) {
t.Parallel() t.Parallel()
const ( const deadline = 100 * time.Millisecond
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()
@@ -667,9 +660,9 @@ func TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) {
select { select {
case <-shutDown: case <-shutDown:
case <-time.After(deadline + margin): case <-time.After(deadline + stepTimeout):
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", margin) "a write was blocked", stepTimeout)
} }
var dialer net.Dialer var dialer net.Dialer
@@ -744,8 +737,6 @@ 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{}
@@ -763,7 +754,7 @@ func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
Path: ristypes.ASPath{Sequence: []int{testASNA}}, Path: ristypes.ASPath{Sequence: []int{testASNA}},
}) })
ctx, cancel := context.WithTimeout(t.Context(), deadline) ctx, cancel := context.WithTimeout(t.Context(), stepTimeout)
defer cancel() defer cancel()
rw.Shutdown(ctx) rw.Shutdown(ctx)
+14 -9
View File
@@ -40,9 +40,13 @@ type ASHandler struct {
db database.Store db database.Store
logger *logger.Logger logger *logger.Logger
// batchMaxSize is asnBatchMaxSize. Only HandleMessage reads it, so a test
// may lower it before it hands the handler a message.
batchMaxSize int
// 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
// waits for a database write only while batch is at asnBatchMaxSize. // waits for a database write only while batch is at batchMaxSize.
mu sync.Mutex mu sync.Mutex
batch []asnOp batch []asnOp
// writing is how many operations the flush goroutine took from batch // writing is how many operations the flush goroutine took from batch
@@ -79,12 +83,13 @@ func newASHandler(
db database.Store, logger *logger.Logger, batchTimeout time.Duration, db database.Store, logger *logger.Logger, batchTimeout time.Duration,
) *ASHandler { ) *ASHandler {
h := &ASHandler{ h := &ASHandler{
db: db, db: db,
logger: logger, logger: logger,
batch: make([]asnOp, 0, asnBatchSize), batchMaxSize: asnBatchMaxSize,
batchFullCh: make(chan struct{}, 1), batch: make([]asnOp, 0, asnBatchSize),
stopCh: make(chan context.Context, 1), batchFullCh: make(chan struct{}, 1),
done: make(chan struct{}), stopCh: make(chan context.Context, 1),
done: make(chan struct{}),
} }
h.batchTaken.L = &h.mu h.batchTaken.L = &h.mu
@@ -127,7 +132,7 @@ func (h *ASHandler) 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) >= asnBatchMaxSize { for len(h.batch) >= h.batchMaxSize {
h.batchTaken.Wait() h.batchTaken.Wait()
} }
@@ -151,7 +156,7 @@ func (h *ASHandler) 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 asnBatchMaxSize. // or the batch reaches batchMaxSize.
if len(h.batch) >= asnBatchSize { if len(h.batch) >= asnBatchSize {
select { select {
case h.batchFullCh <- struct{}{}: case h.batchFullCh <- struct{}{}:
@@ -19,7 +19,7 @@ func TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS(t *testing.T) {
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{db: db, logger: logger.New()} h := &ASHandler{db: db, logger: logger.New(), batchMaxSize: asnBatchMaxSize}
first := peerAnnouncement(testPeerIP, testV4Prefix) first := peerAnnouncement(testPeerIP, testV4Prefix)
second := peerAnnouncement(testPeerIP, testV4Prefix) second := peerAnnouncement(testPeerIP, testV4Prefix)
@@ -54,7 +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{db: store, logger: logger.New()} h := &ASHandler{db: store, logger: logger.New(), batchMaxSize: asnBatchMaxSize}
msg := peerAnnouncement(testPeerIP, testV4Prefix) msg := peerAnnouncement(testPeerIP, testV4Prefix)
msg.Path = decodePath(t, testASNA, 0, testASNB) msg.Path = decodePath(t, testASNA, 0, testASNB)
@@ -224,7 +224,7 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
select { select {
case <-lookupStarted: case <-lookupStarted:
case <-time.After(handleMessageTimeout): case <-time.After(stepTimeout):
t.Fatal("the fetcher did not start the lookup") t.Fatal("the fetcher did not start the lookup")
} }
@@ -240,22 +240,22 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
select { select {
case <-f.done: case <-f.done:
case <-time.After(handleMessageTimeout): case <-time.After(stepTimeout):
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, far longer than statsWindow, and never // second, by a clock the test sets, for three times 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()
+67 -27
View File
@@ -19,9 +19,15 @@ import (
) )
const ( const (
// handleMessageTimeout is how long a test waits for HandleMessage to // stepTimeout is how long a test waits for a step it expects, such as
// return, or for a write to start, before it fails. // HandleMessage returning, a write starting or Shutdown returning, before
handleMessageTimeout = 2 * time.Second // it fails. Only a step that never comes should reach it. On a loaded
// 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.
@@ -198,7 +204,7 @@ func lastLogged(t *testing.T, logs *bytes.Buffer) stopLog {
return logged return logged
} }
// returnsInTime reports whether f returns within handleMessageTimeout. // returnsInTime reports whether f returns within stepTimeout.
func returnsInTime(f func()) bool { func returnsInTime(f func()) bool {
done := make(chan struct{}) done := make(chan struct{})
@@ -210,11 +216,22 @@ func returnsInTime(f func()) bool {
select { select {
case <-done: case <-done:
return true return true
case <-time.After(handleMessageTimeout): case <-time.After(stepTimeout):
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
@@ -240,7 +257,7 @@ func startBlockedWrite(
select { select {
case <-store.writeStarted: case <-store.writeStarted:
case <-time.After(handleMessageTimeout): case <-time.After(stepTimeout):
t.Fatal("the handler did not start to write the messages") t.Fatal("the handler did not start to write the messages")
} }
} }
@@ -261,7 +278,7 @@ func handleWhileAWriteIsBlocked(
) { ) {
t.Helper() t.Helper()
defer h.Stop(t.Context()) defer stopInTime(t, h)
defer store.releaseWrites() defer store.releaseWrites()
startBlockedWrite(t, store, h, fill) startBlockedWrite(t, store, h, fill)
@@ -275,7 +292,9 @@ 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. // it stops h, which writes what it holds. The tests lower h's maximum batch
// 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,
@@ -288,13 +307,18 @@ func handleAtMaxBatchSize(
) { ) {
t.Helper() t.Helper()
defer h.Stop(t.Context()) defer stopInTime(t, h)
defer store.releaseWrites() defer store.releaseWrites()
startBlockedWrite(t, store, h, fill) startBlockedWrite(t, store, h, fill)
for _, msg := range toMax { if !returnsInTime(func() {
h.HandleMessage(msg) for _, msg := range toMax {
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{})
@@ -315,7 +339,7 @@ func handleAtMaxBatchSize(
select { select {
case <-returned: case <-returned:
case <-time.After(handleMessageTimeout): case <-time.After(stepTimeout):
t.Fatal("HandleMessage did not return after the database write finished") t.Fatal("HandleMessage did not return after the database write finished")
} }
} }
@@ -345,8 +369,12 @@ func stopWhileAWriteIsBlocked(
startBlockedWrite(t, store, h, fill) startBlockedWrite(t, store, h, fill)
for _, msg := range extra { if !returnsInTime(func() {
h.HandleMessage(msg) for _, msg := range extra {
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)
@@ -361,7 +389,7 @@ func stopWhileAWriteIsBlocked(
select { select {
case <-done: case <-done:
case <-time.After(handleMessageTimeout): case <-time.After(stepTimeout):
t.Fatal("the flush goroutine did not return after the write finished") t.Fatal("the flush goroutine did not return after the write finished")
} }
} }
@@ -380,7 +408,8 @@ 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)},
@@ -597,7 +626,7 @@ func TestPeeringHandlerStopCountsTheWriteInProgress(t *testing.T) {
select { select {
case <-h.done: case <-h.done:
case <-time.After(handleMessageTimeout): case <-time.After(stepTimeout):
t.Fatal("the processing goroutine did not return after the write " + t.Fatal("the processing goroutine did not return after the write " +
"finished") "finished")
} }
@@ -618,7 +647,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) h := newPeeringHandler(store, logger.New(), processInterval, maxTrackedPaths)
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}}},
@@ -714,9 +743,11 @@ 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}, prefixBatchMaxSize) toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
handleAtMaxBatchSize(t, store, h, handleAtMaxBatchSize(t, store, h,
[]*ristypes.RISMessage{prefixAnnouncement(fill)}, []*ristypes.RISMessage{prefixAnnouncement(fill)},
[]*ristypes.RISMessage{prefixAnnouncement(toMax)}, []*ristypes.RISMessage{prefixAnnouncement(toMax)},
@@ -736,8 +767,9 @@ func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
store := newBlockingStore() store := newBlockingStore()
h := newASHandler(store, logger.New(), longerThanAnyTest) h := newASHandler(store, logger.New(), longerThanAnyTest)
h.batchMaxSize = asnBatchSize
fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)} fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchMaxSize)} toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
handleAtMaxBatchSize(t, store, h, handleAtMaxBatchSize(t, store, h,
[]*ristypes.RISMessage{{Path: fill}}, []*ristypes.RISMessage{{Path: fill}},
[]*ristypes.RISMessage{{Path: toMax}}, []*ristypes.RISMessage{{Path: toMax}},
@@ -755,7 +787,7 @@ func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
// 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 hang. // would fail.
func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) { func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
t.Parallel() t.Parallel()
@@ -763,10 +795,11 @@ 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, peerBatchMaxSize), slices.Repeat(message, peerBatchSize),
&ristypes.RISMessage{Peer: otherPeerIP}) &ristypes.RISMessage{Peer: otherPeerIP})
store.mu.Lock() store.mu.Lock()
@@ -796,8 +829,9 @@ 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{
batchFullCh: make(chan struct{}, 1), batchMaxSize: prefixBatchMaxSize,
table: routetable.New(routetable.Params{}), batchFullCh: make(chan struct{}, 1),
table: routetable.New(routetable.Params{}),
} }
h.HandleMessage(prefixAnnouncement( h.HandleMessage(prefixAnnouncement(
slices.Repeat([]string{testV4Prefix}, n))) slices.Repeat([]string{testV4Prefix}, n)))
@@ -809,7 +843,10 @@ 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{batchFullCh: make(chan struct{}, 1)} h := &ASHandler{
batchMaxSize: asnBatchMaxSize,
batchFullCh: make(chan struct{}, 1),
}
// The origin AS and each AS in the path are one update each. // The origin AS and each AS in the path are one update each.
h.HandleMessage(&ristypes.RISMessage{ h.HandleMessage(&ristypes.RISMessage{
Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, n-1)}, Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, n-1)},
@@ -822,7 +859,10 @@ 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{batchFullCh: make(chan struct{}, 1)} h := &PeerHandler{
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(handleMessageTimeout): case <-time.After(stepTimeout):
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(handleMessageTimeout): case <-time.After(stepTimeout):
t.Fatal("the maintenance goroutine did not return after the " + t.Fatal("the maintenance goroutine did not return after the " +
"checkpoint ended") "checkpoint ended")
} }
+14 -9
View File
@@ -43,10 +43,14 @@ 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
// peerBatchMaxSize. // batchMaxSize.
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
@@ -84,12 +88,13 @@ func newPeerHandler(
db database.Store, logger *logger.Logger, batchTimeout time.Duration, db database.Store, logger *logger.Logger, batchTimeout time.Duration,
) *PeerHandler { ) *PeerHandler {
h := &PeerHandler{ h := &PeerHandler{
db: db, db: db,
logger: logger, logger: logger,
peerBatch: make([]peerUpdate, 0, peerBatchSize), batchMaxSize: peerBatchMaxSize,
batchFullCh: make(chan struct{}, 1), peerBatch: make([]peerUpdate, 0, peerBatchSize),
stopCh: make(chan context.Context, 1), batchFullCh: make(chan struct{}, 1),
done: make(chan struct{}), stopCh: make(chan context.Context, 1),
done: make(chan struct{}),
} }
h.batchTaken.L = &h.mu h.batchTaken.L = &h.mu
@@ -135,7 +140,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) >= peerBatchMaxSize { for len(h.peerBatch) >= h.batchMaxSize {
h.batchTaken.Wait() h.batchTaken.Wait()
} }
@@ -149,7 +154,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 peerBatchMaxSize. // or the batch reaches batchMaxSize.
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()} h := &PeerHandler{db: store, logger: logger.New(), batchMaxSize: peerBatchMaxSize}
first := peerAnnouncement(testPeerIP, testV4Prefix) first := peerAnnouncement(testPeerIP, testV4Prefix)
second := peerAnnouncement(testPeerIP, testV4Prefix) second := peerAnnouncement(testPeerIP, testV4Prefix)
+16 -9
View File
@@ -42,6 +42,10 @@ 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
@@ -62,21 +66,24 @@ 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) return newPeeringHandler(db, logger, peeringProcessInterval, maxTrackedPaths)
} }
// newPeeringHandler is NewPeeringHandler with the processing timer's interval, // newPeeringHandler is NewPeeringHandler with the processing timer's interval,
// processInterval, in place of peeringProcessInterval. A test gives a short // processInterval, in place of peeringProcessInterval, and the most paths it
// one to have the timer start a write. // holds, maxPaths, in place of maxTrackedPaths. A test gives a short interval to
// 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),
stopCh: make(chan context.Context, 1), maxPaths: maxPaths,
done: make(chan struct{}), stopCh: make(chan context.Context, 1),
done: make(chan struct{}),
} }
// Start the periodic processing goroutine // Start the periodic processing goroutine
@@ -131,7 +138,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) >= maxTrackedPaths { if !exists && len(h.asPaths) >= h.maxPaths {
// 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,7 +5,6 @@ import (
"fmt" "fmt"
"maps" "maps"
"slices" "slices"
"strconv"
"sync" "sync"
"testing" "testing"
"time" "time"
@@ -20,6 +19,10 @@ 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
@@ -52,9 +55,10 @@ func (r *recordingStore) callCount() int {
// processing goroutine, so tests drive processing explicitly. // processing goroutine, so tests drive processing explicitly.
func newTestHandler(db database.Store) *PeeringHandler { func newTestHandler(db database.Store) *PeeringHandler {
return &PeeringHandler{ return &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,
} }
} }
@@ -244,25 +248,28 @@ 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. // dropped. The handler's limit is small, and its timer does not fire.
func TestPeeringHandlerCapDropsAndCounts(t *testing.T) { func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
t.Parallel() t.Parallel()
store := &recordingStore{mockStore: newMockStore()} store := &recordingStore{mockStore: newMockStore()}
h := newTestHandler(store) h := newPeeringHandler(store, logger.New(), longerThanAnyTest,
testMaxTrackedPaths)
defer stopInTime(t, h)
now := time.Now().UTC() now := time.Now().UTC()
// Fill the map to exactly maxTrackedPaths, including one real path key so // Fill the map to exactly its limit, including one real path key so the
// the "already tracked" branch can be exercised. The filler keys are never // "already tracked" branch can be exercised. The other paths are of one
// processed in this test, so their contents do not matter. // AS, so they carry no peering.
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 i := 0; len(h.asPaths) < maxTrackedPaths; i++ { for asn := 1; len(h.asPaths) < testMaxTrackedPaths; asn++ {
h.asPaths[strconv.Itoa(i)] = database.SeenTimes{First: now, Last: now} h.asPaths[pathKey(t, asn)] = database.SeenTimes{First: now, Last: now}
} }
h.mu.Unlock() h.mu.Unlock()
@@ -270,8 +277,9 @@ 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 != maxTrackedPaths { if tracked != testMaxTrackedPaths {
t.Fatalf("want map size %d after drop, got %d", maxTrackedPaths, tracked) t.Fatalf("want map size %d after drop, got %d", testMaxTrackedPaths,
tracked)
} }
if dropped != 1 { if dropped != 1 {
@@ -283,8 +291,9 @@ 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 != maxTrackedPaths { if tracked != testMaxTrackedPaths {
t.Fatalf("want map size %d after refresh, got %d", maxTrackedPaths, tracked) t.Fatalf("want map size %d after refresh, got %d", testMaxTrackedPaths,
tracked)
} }
if dropped != 1 { if dropped != 1 {
@@ -303,17 +312,21 @@ 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. // run's own, so it is 0 again after the run. The handler's limit is small, and
// its timer does not fire.
func TestPeeringHandlerRunResetsTheDroppedCount(t *testing.T) { func TestPeeringHandlerRunResetsTheDroppedCount(t *testing.T) {
t.Parallel() t.Parallel()
h := newTestHandler(newMockStore()) h := newPeeringHandler(newMockStore(), logger.New(), longerThanAnyTest,
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) < maxTrackedPaths; asn++ { for asn := 1; len(h.asPaths) < testMaxTrackedPaths; 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()
+31 -19
View File
@@ -60,10 +60,17 @@ 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
// prefixBatchMaxSize. // batchMaxSize.
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
@@ -119,24 +126,30 @@ 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. A test that the timer must not // batchTimeout, in place of prefixBatchTimeout, and tableChangesPerWrite in
// disturb gives an interval longer than the test runs. // place of maxTableChangesPerWrite. A test that the timer must not disturb
// 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, batchTimeout time.Duration, tableChangesPerWrite int,
) *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),
batchFullCh: make(chan struct{}, 1), batchMaxSize: prefixBatchMaxSize,
stopCh: make(chan context.Context, 1), tableChangesPerWrite: tableChangesPerWrite,
done: make(chan struct{}), batchFullCh: make(chan struct{}, 1),
stopCh: make(chan context.Context, 1),
done: make(chan struct{}),
} }
h.batchTaken.L = &h.mu h.batchTaken.L = &h.mu
@@ -191,7 +204,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) >= prefixBatchMaxSize { for len(h.batch) >= h.batchMaxSize {
h.batchTaken.Wait() h.batchTaken.Wait()
} }
@@ -239,7 +252,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 prefixBatchMaxSize. // or the batch reaches batchMaxSize.
if len(h.batch) >= prefixBatchSize { if len(h.batch) >= prefixBatchSize {
select { select {
case h.batchFullCh <- struct{}{}: case h.batchFullCh <- struct{}{}:
@@ -250,7 +263,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
// maxTableChangesPerWrite of the routing table's changes, and the prefixes of // tableChangesPerWrite 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
@@ -378,14 +391,13 @@ func (h *PrefixHandler) applyToTable(update prefixUpdate) {
}) })
} }
// writeTableChanges writes at most maxTableChangesPerWrite changes of the // writeTableChanges writes at most tableChangesPerWrite changes of the routing
// routing table in memory, and every peer the table numbered since the last // table in memory, and every peer the table numbered since the last write.
// 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(maxTableChangesPerWrite) changes, newPeers := h.table.TakeChanges(h.tableChangesPerWrite)
h.writingTableChanges = len(changes) h.writingTableChanges = len(changes)
h.mu.Unlock() h.mu.Unlock()
@@ -237,9 +237,10 @@ func TestPrefixHandlerStopLogsNothingWhenItWroteEverything(t *testing.T) {
} }
} }
// longerThanAnyTest is a flush timer interval that no test lasts. A test whose // longerThanAnyTest is a timer interval that no test lasts. A test gives it to
// prefix, AS or peer handler must flush only when its batch is full and on Stop // newPrefixHandler, newASHandler, newPeerHandler or newPeeringHandler when the
// gives it to newPrefixHandler, newASHandler or newPeerHandler. // handler must write only when its batch is full, when the test says, or on
// 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
@@ -247,9 +248,11 @@ const longerThanAnyTest = time.Hour
// NewPrefixHandler's flush goroutine, so the test flushes itself. // NewPrefixHandler's flush goroutine, so the test flushes itself.
func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler { func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
return &PrefixHandler{ return &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,11 +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"
)
// tableWriteTimeout is how long a test waits for the flush timer to write the // testMaxTableChangesPerWrite is a limit of one write of the routing
// routing table's changes: a few ticks of it. // table's changes that a test exceeds quickly, in place of
const tableWriteTimeout = 5 * prefixBatchTimeout // maxTableChangesPerWrite.
testMaxTableChangesPerWrite = 10
)
// 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
@@ -276,9 +277,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(tableWriteTimeout): case <-time.After(stepTimeout):
t.Fatalf("the routing table's change was not written within %s", t.Fatalf("the routing table's change was not written within %s",
tableWriteTimeout) stepTimeout)
} }
} }
} }
@@ -375,25 +376,32 @@ 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. // memory. The handler's limit is small, and its timer does not fire.
func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) { func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
t.Parallel() t.Parallel()
const ( const (
pairs = maxTableChangesPerWrite * 3 / 2 pairs = testMaxTableChangesPerWrite * 3 / 2
flushes = 3 flushes = 3
) )
db := newTestDatabase(t) db := newTestDatabase(t)
store := &tableWriteSizesStore{Database: db} store := &tableWriteSizesStore{Database: db}
h := prefixHandlerWithoutTimer(store) table := routetable.New(routetable.Params{})
announceManyPrefixes(h.table, pairs) announceManyPrefixes(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{maxTableChangesPerWrite, pairs - maxTableChangesPerWrite} wantChanges := []int{
testMaxTableChangesPerWrite, pairs - testMaxTableChangesPerWrite,
}
wantNewPeers := []int{1, 0} wantNewPeers := []int{1, 0}
if !slices.Equal(store.changes, wantChanges) || if !slices.Equal(store.changes, wantChanges) ||
@@ -415,11 +423,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. // log. The handler's limit of one write is small.
func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) { func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
t.Parallel() t.Parallel()
const pairs = maxTableChangesPerWrite + 1 const pairs = testMaxTableChangesPerWrite + 1
var logs bytes.Buffer var logs bytes.Buffer
@@ -427,7 +435,8 @@ func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
table := routetable.New(routetable.Params{}) table := routetable.New(routetable.Params{})
announceManyPrefixes(table, pairs) announceManyPrefixes(table, pairs)
h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest) h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest,
testMaxTableChangesPerWrite)
h.Stop(t.Context()) h.Stop(t.Context())
written := 0 written := 0
+61
View File
@@ -6,6 +6,7 @@ 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"
@@ -341,6 +342,66 @@ 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()
+27 -4
View File
@@ -82,6 +82,12 @@ 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
@@ -111,7 +117,8 @@ 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 ctx ends. // the server works out the home page's cards every minute until Stop: the end
// 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"
@@ -148,16 +155,22 @@ 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 HTTP server. It waits, until ctx ends, for the requests in // Stop stops the goroutine that works out the home page's cards, and waits
// progress to finish, and then closes the connections of those still running // until it has returned or ctx ends. It then stops the HTTP server. It waits,
// and returns ctx's error. // until ctx ends, for the requests in progress to finish, and then closes the
// 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
@@ -165,6 +178,16 @@ 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())