sneak/routewatch
1
0
Fork 0
Code Issues 5 Pull Requests 4 Actions 9 Packages Projects Releases Wiki Activity

Compare commits

base: sneak/routewatch:edc222f1bb9a54973c493668ede5993fb44e908a
Branches Tags
sneak/routewatch:main sneak/routewatch:issue-225 sneak/routewatch:next sneak/routewatch:issue-157-todo sneak/routewatch:issue-157-readme sneak/routewatch:prod sneak/routewatch:issue-27-cached-stats
..
compare: sneak/routewatch:issue-157-todo
Branches Tags
sneak/routewatch:issue-225 sneak/routewatch:next sneak/routewatch:issue-157-todo sneak/routewatch:issue-157-readme sneak/routewatch:prod sneak/routewatch:main sneak/routewatch:issue-27-cached-stats
9 Commits
edc222f1bb .. issue-157-todo
Author SHA1 Message Date
clawbot 9e7e87b68d Rewrite TODO.md sentence by sentence (part of #157)
check / check (push) Waiting to run
Details 
Every sentence of TODO.md is rewritten from scratch against the rules
of #157: about 30 words at
most, one idea each, subject and verb early, identifiers in backticks.
Status and Next Step describe the repo as the tracker shows it now.
Every Completed Steps entry keeps its date, its facts and its issue
reference, checked against the commit it names. The Workflow section
keeps its meaning. The entry for this rewrite is added.

Model: fable-5-1
 2026-11-02 03:55:01 +00:00
clawbot d66c59ef07 Lengthen the WHOIS fetcher's wait after a skipped lookup (closes #222)
check / check (push) Canceled after 0s
Details 
A lookup whose RIR is paused is skipped. The skip left the fetcher's wait
as it was. Each lookup asks IANA before it finds its RIR paused. So while
most stale ASes belonged to a paused RIR, the fetcher could ask IANA every
second.

Like a failure, a skipped lookup now makes the wait 1.5 times as long, up
to one minute. The skip is still not counted as a failure, so the status
page's failure figures leave it out.

Model: opus-5-5
 2026-11-02 03:59:47 +01:00
clawbot 26d939ad72 Keep taking messages while a peer's session goes down (closes #216)
check / check (push) Canceled after 0s
Details 
The table's PeerDown visits every prefix under the write lock, which
takes over a second on a full table. The prefix handler called it
inline, so its queue filled and the streamer dropped updates.

PeerDown now holds the lock for 10,000 prefixes at a time, then lets
the lock go for as long again. The handler runs PeerDown on a separate
goroutine, one peer at a time. The handler holds back the down peer's
later messages until that PeerDown returns. Then the handler applies
them in order, one at a time, and takes new messages in between. It
holds back at most 20,000 messages. Stop waits for all of this before
its last flush.

Model: opus-5-5
 2026-11-02 03:27:10 +01:00
clawbot df15e2b2cd Pause a registry's WHOIS server after it refuses a query (closes #218)
check / check (push) Canceled after 0s
Details 
The fetcher kept one interval for every registry, so a registry that
refused a query was asked again within seconds. RIPE blocked the test
host's address that way.

The WHOIS client now pauses an RIR's or NIC.br's server for an hour
after the server refuses a query. A refused or reset connection is a
refusal, and so is an %ERROR:201: or %ERROR:202: line. Lookups at the
other servers go on. A lookup whose RIR is paused leaves the AS stale,
and counts as neither a success nor a failure. IANA is never paused,
because a pause of IANA would stop every lookup. NIC.br's hour-long
skip after its over-limit answer is now the same pause.

Model: opus-5-5
 2026-11-02 02:59:57 +01:00
clawbot 6b004dfbc3 Keep one entry per AS in the AS handler's batch (closes #217)
check / check (push) Canceled after 0s
Details 
HandleMessage added one entry to the batch for every AS of every path,
and waited for the write in progress once the batch held 300,000. In a
burst of the feed its queue then filled, and the streamer dropped
messages.

The batch is now a map from AS number to the earliest and the latest
time of its updates. It holds one entry per AS, so it no longer grows
with the feed's rate. HandleMessage never waits for a write.
asnBatchSize now counts ASes. Stop logs the entries of the batch being
written and of the next batch under "batch_entries". An AS in both
counts twice.

Model: opus-5-5
 2026-11-02 02:10:17 +01:00
clawbot 38b43549b9 Make the routewatch tests pass on a loaded host (closes #213)
check / check (push) Canceled after 0s
Details 
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
Details 
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
Details 
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
Details 
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
23 changed files with 1795 additions and 630 deletions
Expand all files Collapse all files
README.md
+60 -28
View File
@@ -9,19 +9,38 @@ table in real time. It works like this:
## Rationale
RIPE RIS Live streams the BGP updates that RIPE's route collectors receive, as
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 exists to serve our other services as an internal API for information
about IP addresses.
The feed sends no table when a client connects. So the routing table holds only
the routes announced since RouteWatch first started. An IP-to-AS lookup finds
nothing for an address whose routes have not been announced since then. The
routing table also misses updates RouteWatch drops when it falls behind, and
updates sent while it is stopped. So a lookup or a prefix page can show a route
the feed has withdrawn.
RouteWatch follows BGP routing table updates from RIPE RIS Live. RIS Live
streams the updates that RIPE's route collectors receive, as they happen, from
the moment a client connects. It reports changes, not the routing table they add
up to. RouteWatch applies those changes to a routing table that it keeps in
memory, with a copy in its own SQLite database. From that table, RouteWatch sees
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
@@ -410,10 +429,12 @@ the routing table, and everything else from the database.
- Parses BGP UPDATE messages from JSON stream
- Dispatches to registered handlers based on message type
- Each handler has its own queue with backpressure handling
- Handlers batch writes for efficiency: the AS, peer and prefix handlers
write 30,000, 10,000 and 25,000 updates at a time, or after 2, 2 and 1
seconds; the peering handler writes the peerings it has collected every 30
seconds
- Handlers batch writes for efficiency. The AS, peer and prefix handlers
write a batch when it reaches a size, or after a time:
- the AS handler at 30,000 ASes, or after 2 seconds;
- the peer handler at 10,000 updates, or after 2 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**
- **ASHandler**: Tracks all ASNs seen in AS paths
@@ -423,7 +444,9 @@ the routing table, and everything else from the database.
- An announcement gives the peer a route to the prefix.
- A withdrawal removes the peer's route. So does an announcement whose
AS path is empty or holds AS 0 anywhere.
- A peer whose BGP session goes down leaves every prefix.
- A peer whose BGP session goes down leaves every prefix. The handler
holds back the peer's later updates until the peer has left every
prefix, then applies them in order.
Every second it writes the changed prefix and origin AS pairs to the
`routes` table. A write takes at most 20,000 pairs, and the rest follow
@@ -559,8 +582,8 @@ Budget:
- Go heap: a 2.5 GiB soft limit (`GOMEMLIMIT=2560MiB`, set in the image).
- The routing table takes about 700 bytes per prefix and origin AS. A full
table has about 1.4 million of these pairs, so it takes about 1 GB.
- The rest of the heap takes up to about 0.5 GiB. Most of it is the
handlers' batches and queues, described below.
- The rest of the heap takes up to about 0.5 GiB. Most of it is what the
handlers hold, described below.
- The remaining 1 GiB is room for the garbage collector. With much less, it
would run almost all the time.
- SQLite: 656 MiB of page cache.
@@ -613,16 +636,25 @@ What happens at each limit:
`STATE` message, which it holds, in feed order, until there is room: a few
hundred bytes each, about ten a second.
- Handler batches: while the AS, peer or prefix handler writes a batch to the
database, the updates that arrive collect in its next batch. That batch stops
at ten times the batch size:
- 300,000 updates for the AS handler (`asnBatchMaxSize`, about 10 MiB);
- 100,000 for the peer handler (`peerBatchMaxSize`, about 10 MiB);
- 250,000 for the prefix handler (`prefixBatchMaxSize`, about 15 MiB).
database, the updates that arrive collect in its next batch.
- The AS handler's batch holds one entry per AS, with the earliest and the
latest time of its updates. About 80,000 ASes appear in the feed, so the
batch takes at most about 10 MiB, however fast the feed runs.
- The peer handler's batch stops at 100,000 updates (`peerBatchMaxSize`),
about 10 MiB.
- The prefix handler's batch stops at 250,000 updates
(`prefixBatchMaxSize`), about 15 MiB.
At that size the handler takes nothing more off its queue until the write
ends. The queue then fills, and the streamer drops messages as above. With
the batches being written, the three handlers' batches take at most about
0.1 GiB of the Go heap.
At that maximum the peer or prefix handler takes nothing more off its queue
until the write ends. The queue then fills, and the streamer drops messages
as above. With the batches being written, the three handlers' batches take
at most about 0.1 GiB of the Go heap.
- Held-back messages: the prefix handler holds back a peer's messages after the
peer's session goes down. It applies them once the peer has left every prefix.
It holds at most 20,000 such messages (`maxHeldBackMessages`), about 20 MiB.
At that number it takes nothing more off its queue until it has applied one of
them.
With `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds
with the goroutine count and Go memory figures.
TODO.md
+334 -243
View File
@@ -10,130 +10,206 @@
# Status
pre-1.0. No git tags. `main` last took `next` on 2026-10-06, through
https://git.eeqj.de/sneak/routewatch/pulls/6. The linter and the tests are
phases of the Docker build, and the Gitea workflow runs `script/cibuild` (the
format check of the Go code and the markdown, both phases, then the image) on
every push. The image sets memory ceilings for a 5 GiB container (README
"Memory") and the README says how to run it under upaas (README "Running under
upaas"). A 25-hour run of `a9b6860` on the live feed peaked at about 0.6 GiB
(https://git.eeqj.de/sneak/routewatch/issues/3). A run of `825afa2` on the live
feed found the prefix handler dropping about a quarter of its updates because
its route writes were slower than the feed
(https://git.eeqj.de/sneak/routewatch/issues/96); the cheaper writes since then
are measured by a benchmark, not yet on the live feed.
RouteWatch is pre-1.0, with no git tags. `main` last took `next` on 2026-10-27,
through https://git.eeqj.de/sneak/routewatch/pulls/166.
https://git.eeqj.de/sneak/routewatch/pulls/202 carries the commits since then
into `main`. It is labelled `merge-ready` and waits for sneak.
The linter and the tests are phases of the Docker build. The Gitea workflow runs
`script/cibuild` on every push. That script runs the two phases and checks the
formatting of the Go code and the markdown. Then it builds the image.
The image sets memory ceilings for a 5 GiB container (README "Memory"). README
"Running under upaas" says how to run it.
A five-day run of `1c65656` on the live feed, from 27 October to 1 November,
found three problems:
- The prefix handler dropped updates while a peer's session went down
(https://git.eeqj.de/sneak/routewatch/issues/216).
- The AS handler dropped updates when the feed burst
(https://git.eeqj.de/sneak/routewatch/issues/217).
- The WHOIS fetcher kept asking a registry that had refused it
(https://git.eeqj.de/sneak/routewatch/issues/218).
All three are fixed on `next`. So is
https://git.eeqj.de/sneak/routewatch/issues/222, which the third fix brought up.
A 24-hour run of `26d939a` on the live feed is checking whether any handler
still drops updates.
The README rewrite, https://git.eeqj.de/sneak/routewatch/pulls/210, has failed
three reviews. It is paused until sneak has reviewed it. It is part of
https://git.eeqj.de/sneak/routewatch/issues/157.
Two questions wait for sneak: which history the prefix page keeps
(https://git.eeqj.de/sneak/routewatch/issues/161), and which fields the IP
lookup answer gains (https://git.eeqj.de/sneak/routewatch/issues/203).
The order of the project's remaining work is kept on
https://git.eeqj.de/sneak/routewatch/issues/86.
# Next Step
Run `next` on the live feed, as README "Running under upaas" describes, until
the routing table stops growing. Post its size and memory figures on
https://git.eeqj.de/sneak/routewatch/issues/122. The same run makes the final
check https://git.eeqj.de/sneak/routewatch/issues/96 names: whether the prefix
handler still drops updates after the first hour. If it does, a new issue
follows with the numbers. Then an audit of the code at the commit that run
tested.
Post the outcome of the 24-hour run of `26d939a` on
https://git.eeqj.de/sneak/routewatch/issues/216: whether any handler still drops
updates, and the process's peak memory against 5 GiB. File an issue for each
problem the run finds.
# Completed Steps
- 2026-10-27: every JSON answer carries `@meta`. A panic on a JSON route answers
- 2026-11-02: `TODO.md` is rewritten from scratch, sentence by sentence, against
the rules of https://git.eeqj.de/sneak/routewatch/issues/157. Every Completed
Steps entry keeps its date, its facts and its issue. Status and Next Step
describe the repo as the tracker shows it (part of #157)
- 2026-11-02: A skipped WHOIS lookup now makes the fetcher wait longer before
the next one, as a failed lookup does. A lookup is skipped when its RIR is
paused. An RIR is one of the five regional Internet registries. The longest
wait is one minute. The status page does not count a skip as a failure. A skip
used to leave the wait as it was. Each lookup asks IANA which RIR holds the AS
before it finds that RIR paused. So while most stale ASes belonged to a paused
RIR, the fetcher could ask IANA every second (closes #222)
- 2026-11-02: A peer's session going down no longer stops the prefix handler.
The handler runs the routing table's `PeerDown` on a goroutine of its own, one
peer at a time. Meanwhile it goes on taking messages. `PeerDown` holds the
table's lock for 10,000 prefixes at a time, then lets the lock go for as long
again. The handler holds back the down peer's later messages until that peer's
`PeerDown` returns. Then the handler applies them in order, one at a time, and
takes new messages in between. It holds back at most 20,000 messages. At that
number, `HandleMessage` waits until one of them is applied. When `Stop` runs
past its deadline, it logs how many held-back messages were not applied
(closes #216)
- 2026-11-02: RouteWatch pauses an RIR's or NIC.br's WHOIS server for an hour
after the server refuses a query. NIC.br is Brazil's registry. 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. While NIC.br is paused, a Brazilian AS
gets the details from LACNIC's answer. 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 its
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. At 300,000 entries it waited for the write in progress. 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. Under the race detector that can take longer. A test now waits up to 15
seconds for each step it expects. The tests of the handlers' limits now give
the handlers small limits. 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
given to `Start` stopped the refreshes (closes #211)
- 2026-10-27: README "Rationale" says why RouteWatch exists. RouteWatch answers
our other services' questions about IP addresses, from its own routing table
and the WHOIS details it keeps. The section 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
500 with the JSON error body. `/status.json` and the health check answer 405
with the JSON error body to any method they do not take. The 500 and the 405
used to have an empty body. The home page's IP lookup shows the whole JSON
answer, `@meta` included. It used to show only the answer's `data`, or only
the message of an error answer. A test checks `@meta` on every JSON route, in
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
unit at the end of its field name. `@meta.execution_time` is now
`execution_time_ms`. The stats' `uptime`, `connection_duration` and
`memory_usage` are now `uptime_seconds`, `connection_duration_seconds` and
`memory_usage_bytes`. `whois_stats.current_interval` is now
`current_interval_seconds`. They used to be strings such as "12 ms",
"1h18m34s" or "1.1 GB". `connection_duration_seconds` is null while the feed
is not connected. `API_POLICIES.md` is vendored from `sneak/prompts`, and
`REPO_POLICIES.md` re-vendored (closes #199)
- 2026-10-27: every result of the IP lookup has the same fields. A field without
a value is null. It used to be left out. An address with no live route has
every route and AS field set to null, and `ptr` is an empty list when there
are no PTR names. New fields give the number of origin ASes of the netblock,
in `num_origins`, and which they are, in `origin_asns` (closes #200)
- 2026-10-27: with `TRUSTED_PROXIES` unset, the server trusts reverse proxies at
- 2026-10-27: Every value with a unit in a JSON answer is a number. The unit is
at the end of the field name. `@meta.execution_time` is now
`execution_time_ms`. In the stats, `uptime` is now `uptime_seconds`,
`connection_duration` is now `connection_duration_seconds`, and `memory_usage`
is now `memory_usage_bytes`. `whois_stats.current_interval` is now
`whois_stats.current_interval_seconds`. They used to be strings such as "12
ms", "1h18m34s" or "1.1 GB". `connection_duration_seconds` is null while the
feed is not connected. `API_POLICIES.md` is vendored from `sneak/prompts`, and
`REPO_POLICIES.md` is re-vendored (closes #199)
- 2026-10-27: Every result of the IP lookup has the same fields. A field without
a value is null. It used to be left out. An address with no route in the
routing table has every route and AS field set to null. `ptr` is an empty list
when there are no PTR names. The new field `num_origins` gives the number of
origin ASes of the netblock, and `origin_asns` gives which they are (closes
#200)
- 2026-10-27: With `TRUSTED_PROXIES` unset, the server trusts reverse proxies at
the private addresses of RFC 1918. `/ip/`, the home page's `curl` card and the
request log then give the client's address from `X-Forwarded-For`. They used
to give the proxy's address. A list in `TRUSTED_PROXIES` replaces the default,
and `TRUSTED_PROXIES` set but empty trusts no proxy (closes #198)
- 2026-10-27: the home page has cards with figures of the routing table: the
newest and oldest kept route, the newest and oldest updated AS, the AS with
the most prefixes, the ASes with a single prefix, and fifteen well-known
networks. Two cards count the bogon announcements and the prefixes with more
than one origin AS, and link to the new pages `/bogons` and
`/multiple-origins` that list them. A new package, `internal/bogon`, tells
which prefixes and AS numbers are reserved. The server works the figures out
once a minute (closes #154)
- 2026-10-27: the per-peer route tables are gone, with the code that wrote,
to give the proxy's address. A list in `TRUSTED_PROXIES` replaces the default.
`TRUSTED_PROXIES` set but empty trusts no proxy (closes #198)
- 2026-10-27: The home page has cards with figures of the routing table. One
card shows the newest and the oldest kept route. One shows the newest and the
oldest updated AS. One shows the AS with the most prefixes. One shows the ASes
with a single prefix. One shows fifteen well-known networks. Two cards count
the bogon announcements and the prefixes with more than one origin AS. They
link to the new pages `/bogons` and `/multiple-origins`, which list them. A
new package, `internal/bogon`, tells which prefixes and AS numbers are
reserved. The server works the figures out once a minute (closes #154)
- 2026-10-27: The per-peer route tables are gone, with the code that wrote,
counted and read them. The prefix handler applies each update to the routing
table in memory and writes the table's changes every second. It also stores
each new prefix with its first-seen time. The image's memory limits are set
for a full routing table of about 1 GB: `GOMEMLIMIT` is 2.5 GiB, the writing
connection's page cache 512 MiB, and SQLite's heap limits 768 MiB and 1 GiB
(closes #122)
- 2026-10-27: the pages, the API and the stats read every route from the routing
table in memory. It writes the table's changes every second. It also stores
each new prefix with its first-seen time. The image's memory limits fit a full
routing table of about 1 GB. `GOMEMLIMIT` is 2.5 GiB. The writing connection's
page cache is 512 MiB. SQLite's soft and hard heap limits are 768 MiB and 1
GiB (closes #122)
- 2026-10-27: The pages, the API and the stats read every route from the routing
table in memory. For a prefix with several origin ASes, the IP lookup gives
the one most peers carry. Its `last_seen` is the newest kept route's time. The
prefix page and API list every peer that carries the prefix, with an AS path,
next hop and time only for the kept routes. The stats count the origins whose
kept routes were all withdrawn, in `origins_without_kept_route` (part of #122)
- 2026-10-27: the prefix, AS and peer handlers make a timed write on every tick
of their timers, every 1, 2 and 2 seconds. They used to skip every other tick.
A full batch's write restarts the timer, so the next timed write comes one
interval after that write began (closes #192)
- 2026-10-27: once the stop deadline has passed, the AS, peer and peering
handlers start no write either. A full batch or a tick of the timer that
waited while a write ran past the deadline could still start one. The flush
goroutines of the prefix, AS and peer handlers now run the same code (closes
#189)
- 2026-10-27: once the stop deadline has passed, the prefix handler starts no
flush. A batch that filled while a write ran past the deadline could still be
written after it (closes #188)
- 2026-10-27: the prefix handler applies each update to the routing table in
memory, and writes the table's changes to the database every second. A write
takes at most 20,000 changed pairs, and the rest follow in the next writes. A
failed write is written again later. Startup reads the table before the
streamer connects. The pages still read the per-peer route tables (part of
prefix page and API list every peer that carries the prefix. Only the kept
routes show an AS path, a next hop and a time. The stats count the origins
whose kept routes were all withdrawn, in `origins_without_kept_route` (part of
#122)
- 2026-10-27: the database keeps a copy of the whole routing table in a new
- 2026-10-27: The prefix, AS and peer handlers make a timed write on every tick
of their timers. The prefix handler's timer ticks every second. The other two
tick every two seconds. They used to skip every other tick. A full batch's
write restarts the timer. So the next timed write comes one interval after
that write began (closes #192)
- 2026-10-27: Once the stop deadline has passed, the AS, peer and peering
handlers start no write either. Before, a full batch or a timer tick could
wait while a write ran past the deadline. Either could then start one more
write. The flush goroutines of the prefix, AS and peer handlers now run the
same code (closes #189)
- 2026-10-27: Once the stop deadline has passed, the prefix handler starts no
flush. Before, a batch that filled during a late write could still be written
after the deadline (closes #188)
- 2026-10-27: The prefix handler applies each update to the routing table in
memory. It writes the table's changes to the database every second. A write
takes at most 20,000 changed pairs of prefix and origin AS. The rest follow in
the next writes. A failed write is written again later. Startup reads the
table before the streamer connects. The pages still read the per-peer route
tables (part of #122)
- 2026-10-27: The database keeps a copy of the whole routing table in a new
`routes` table, one row per prefix and origin AS. The peers' numbers are in a
new `route_peers` table. One function writes the pairs that changed in memory,
and another reads the whole table back into memory at startup. Nothing uses
them yet (part of #122)
- 2026-10-27: a new package, `internal/routetable`, holds the whole routing
new `route_peers` table. One function writes the pairs that changed in memory.
Another reads the whole table back into memory at startup. Nothing uses them
yet (part of #122)
- 2026-10-27: A new package, `internal/routetable`, holds the whole routing
table in memory. For each prefix and origin AS it keeps the peers that carry
it, one bit per peer, and the two newest of their routes. Nothing uses it yet
(part of #122)
- 2026-10-27: each result of the IP lookup lists the public blocklists the
it, one bit per peer. It also keeps the two newest of their routes. Nothing
uses it yet (part of #122)
- 2026-10-27: Each result of the IP lookup lists the public blocklists the
address is on, in `blocklists`. The lists are DShield, Feodo Tracker, two
FireHOL lists, Spamhaus DROP and the Tor exit list. RouteWatch downloads each
list on its own schedule and answers from copies in memory. The copies are
also kept in the state directory's `blocklists` directory. IP lookup answers
carry `Cache-Control`. An address with no live route now has a result too,
without route fields, so the IP lookup no longer answers 404 (closes #158)
- 2026-10-27: the home page has an IP information API card. It shows the `curl`
command that looks up the visitor's own address at `/ip/{addr}`, and that
command's live answer (closes #156)
- 2026-10-27: on the AS page, a click on the heading of the IPv4 or the IPv6
FireHOL lists, two Spamhaus DROP lists and the Tor exit list. RouteWatch
downloads each list on its own schedule and answers from copies in memory. The
copies are also kept in the state directory's `blocklists` directory. IP
lookup answers carry `Cache-Control`. An address with no route now has a
result too, without route fields. So the IP lookup no longer answers 404
(closes #158)
- 2026-10-27: The home page has an IP information API card. It shows the `curl`
command that looks up the visitor's own address at `/ip/{addr}`. It also shows
that command's live answer (closes #156)
- 2026-10-27: On the AS page, a click on the heading of the IPv4 or the IPv6
prefix list folds or unfolds the list. Both lists start open (closes #155)
- 2026-10-27: the AS page lists its IPv4 and its IPv6 prefixes by mask length,
shortest first, and prefixes of one mask length by address. They used to be
listed by address (closes #153)
- 2026-10-27: on `/` and `/status`, a spinner beside the status turns one frame
each time the stats show more messages than the time before. A page that has
not seen more messages for three refreshes shows "stale" and the time since
then instead, in red (closes #152)
- 2026-10-27: the WHOIS details of a Brazilian AS keep their accented letters.
LACNIC's answer for such an AS comes from NIC.br, Brazil's registry, with each
accented letter replaced. The details now come from NIC.br's own server,
`whois.registro.br`. When that query fails, the details from LACNIC's answer
are kept and a warning is logged. After NIC.br answers that RouteWatch is over
its query limit, RouteWatch stops asking NIC.br for an hour (closes #174)
- 2026-10-27: The AS page lists its IPv4 and its IPv6 prefixes by mask length,
shortest first. Prefixes of one mask length are listed by address. They used
to be sorted by address first, then by mask length (closes #153)
- 2026-10-27: The connection status on `/` and `/status` gets a spinner. It
turns one frame each time the stats show more messages than the time before.
After three refreshes with no new messages, the page shows "stale" in red
instead. Next to it is the time since the last fresh answer (closes #152)
- 2026-10-27: The WHOIS details of a Brazilian AS keep their accented letters.
LACNIC's answer for such an AS is NIC.br's answer with each accented letter
replaced. NIC.br is Brazil's registry. The details now come from NIC.br's own
server, `whois.registro.br`. When that query fails, the details from LACNIC's
answer are kept and a warning is logged. RouteWatch stops asking NIC.br for an
hour after NIC.br says RouteWatch is over its query limit (closes #174)
- 2026-10-27: `make asupdate` stops with an error when the download of `as.csv`
breaks off. The generator used to retry the failed read forever. A record that
does not parse is still logged and skipped (closes #177)
@@ -142,182 +218,197 @@ tested.
one AS. A failed download used to replace the list with an empty file. The
generator reads `as.csv` from `ipverse/as-metadata`, the source's new name
(closes #173)
- 2026-10-27: the WHOIS details of an AS at RIPE, APNIC or AFRINIC come only
- 2026-10-27: The WHOIS details of an AS at RIPE, APNIC or AFRINIC come only
from its `aut-num` object and the organisation object it names. RIPE and APNIC
start their answer with the `as-block` object of the AS's block. Its dates
were stored as the AS's (closes #168)
- 2026-10-27: AS names keep their accented letters. The AS list built into
RouteWatch is rebuilt from today's `as.csv`, which added a country code
column. That file no longer lists AS 0 or the documentation AS numbers. A
WHOIS answer that is not valid UTF-8 is decoded as Latin-1, which RIPE and
APNIC send, before it is parsed and stored (closes #165)
- 2026-10-27: the Age column on the AS and prefix pages reads in words, such as
"5 minutes ago" or "2 weeks ago", instead of Go durations such as "360h0m0s
ago". An age under a minute reads in seconds, and one over a month in months
or years instead of a date (closes #163)
- 2026-10-26: a write to a stored AS, peering or peer keeps the earlier of the
two first-seen times and the later of the two last-seen times. An old update
from a collector catching up on a backlog no longer moves last seen back
(closes #162)
- 2026-10-26: the status page shows "Route Churn/min", the number of prefix
announcements and withdrawals, from all peers, in the last full clock minute.
`/api/v1/stats` reports it as `route_churn_per_min` in `stream`, which
WHOIS answer that is not valid UTF-8 is decoded as Latin-1 before it is parsed
and stored. RIPE and APNIC send Latin-1 (closes #165)
- 2026-10-27: The Age column on the AS and prefix pages reads in words, such as
"5 minutes ago" or "2 weeks ago". It used to print Go durations such as
"360h0m0s ago". An age under a minute reads "now" or in seconds. An age over a
month reads in months or years instead of a date (closes #163)
- 2026-10-26: A write to a stored AS, peering or peer keeps the earlier of the
two first-seen times. It keeps the later of the two last-seen times. An old
update from a collector catching up on a backlog no longer moves last seen
back (closes #162)
- 2026-10-26: The status page shows "Route Churn/min". That is the number of
prefix announcements and withdrawals from all peers in the last full clock
minute. `/api/v1/stats` reports it as `route_churn_per_min` in `stream`. It
replaces the moving average `route_churn_per_sec`. The messages and updates
per second on `/` and `/status` are whole numbers (closes #160)
- 2026-10-26: the WHOIS fetcher backs off more slowly. A failure makes its wait
- 2026-10-26: The WHOIS fetcher backs off more slowly. A failure makes its wait
1.5 times as long, up to one minute instead of five. Its queries to RIPE,
APNIC and AFRINIC leave out person and role objects, which RIPE counts against
a daily limit per address. A refused query is logged with the RIR's error line
(closes #159)
- 2026-10-26: the home page's AS name search lists the ASes whose handle or
description holds the text, as links to their pages, from the new
`/api/v1/search/as`; the search ignores case. When the database has no record
of an AS, its page takes the handle and description from the AS list built
into RouteWatch and still lists the AS's prefixes (closes #151)
- 2026-10-19: README "HTTP Endpoints" lists every route with what it answers,
"Database Schema" matches `schema.sql`, and the rest of the README matches the
code; this file brought up to date (closes #72)
- 2026-10-19: route writes are cheaper: the live route and prefix tables lose
their random UUID keys and the prefix tables their duplicate index, a route
update sets only the columns that change, and every write goes through one
APNIC and AFRINIC leave out person and role objects. RIPE counts those objects
against a daily limit per address. A refused query is logged with the RIR's
error line (closes #159)
- 2026-10-26: The home page's AS name search works. It lists the ASes whose
handle or description holds the text, as links to their pages. The new
`/api/v1/search/as` answers it. The search ignores case. When the database has
no record of an AS, its page takes the handle and description from the AS list
built into RouteWatch. It still lists the AS's prefixes (closes #151)
- 2026-10-19: README "HTTP Endpoints" lists every route with what it answers.
README "Database Schema" matches `schema.sql`. The rest of the README matches
the code. This file is brought up to date (closes #72)
- 2026-10-19: Route writes are cheaper. The live route and prefix tables lose
their random UUID keys. The prefix tables lose their duplicate index. A route
update sets only the columns that change. Every write goes through one
connection with a page cache of 1 GiB (README "Memory"). A benchmark writes
the prefix handler's largest batch faster than the feed sends it (closes #96)
- 2026-10-19: the WAL checkpoint, the incremental vacuum and ANALYZE take the
database lock every batch write holds, so a vacuum no longer fails with
"database is locked" and an ANALYZE no longer makes a batch write lose its
updates; ANALYZE runs as `PRAGMA optimize` with an analysis limit (closes #94)
- 2026-10-19: shutdown stops the streamer first, and each handler writes
everything left on its queue before it stops, so the updates waiting there are
- 2026-10-19: The checkpoint of the WAL (SQLite's write-ahead log), the
incremental vacuum and `ANALYZE` take the database lock that every batch write
holds. So a vacuum no longer fails with "database is locked". An `ANALYZE` no
longer makes a batch write lose its updates. `ANALYZE` runs as
`PRAGMA optimize` with an analysis limit (closes #94)
- 2026-10-19: Shutdown stops the streamer first. Each handler then writes
everything left on its queue before it stops. So the updates waiting there are
no longer lost (closes #69)
- 2026-10-19: a path that ends in an AS set of several members has origin AS 0,
no known origin, instead of the set's last member, and the set adds no
peerings; the pages show such a route as having no known origin (closes #75)
- 2026-10-19: A path that ends in an AS set of several members has origin AS 0,
which means no known origin. It used to have the set's last member as its
origin. The set adds no peerings. The pages show such a route as having no
known origin (closes #75)
- 2026-10-19: `bgp_peer_count` drops to 0 when the last peer goes down (closes
#92)
- 2026-10-19: `bgp_peer_count` counts every peer the feed has heard from, not
only those whose session opened since startup, and drops a peer whose session
goes down (closes #74)
- 2026-10-19: the shared files are re-vendored from `sneak/prompts` at
`0653579`: golangci-lint v2.14.0 and its configuration, lint and test as
`Dockerfile` phases that keep Go's and golangci-lint's caches, markdown
formatted with prettier, agent guidance in `AGENTS.md`; the code is fixed for
what the new linter configuration finds (closes #52)
- 2026-10-19: the streamer reads feed lines up to 1 MiB, so a line over 64 KiB
no longer ends the connection (closes #76)
- 2026-10-19: a peer's live routes are removed when its BGP session goes down;
`STATE` messages are never dropped for backpressure, and one that finds its
- 2026-10-19: `bgp_peer_count` counts every peer the feed has heard from. It
used to count only the peers whose session opened since startup. It drops a
peer whose session goes down (closes #74)
- 2026-10-19: The shared files are re-vendored from `sneak/prompts` at
`0653579`. The linter is `golangci-lint` v2.14.0 with its configuration. Lint
and test are `Dockerfile` phases that keep Go's and `golangci-lint`'s caches.
`make fmt` also formats the markdown with `prettier`. The agent guidance moves
to `AGENTS.md`. The code is fixed for what the new linter configuration finds
(closes #52)
- 2026-10-19: The streamer reads feed lines up to 1 MiB. A line over 64 KiB used
to end the connection (closes #76)
- 2026-10-19: A peer's live routes are removed when its BGP session goes down.
`STATE` messages are never dropped when a queue is full. One that finds its
queue full waits for room ahead of the handler's other messages (closes #56)
- 2026-10-19: the prefix-length pages pick their sample with a bounded number of
index lookups instead of reading every route of that length, and list each
- 2026-10-19: The prefix-length pages pick their sample with a bounded number of
index lookups. They used to read every route of that length. They list each
prefix once (closes #66)
- 2026-10-19: a request past its time limit answers 504 instead of 408, and
- 2026-10-19: A request past its time limit answers 504 instead of 408.
`/api/v1/stats` answers 504 instead of 500 when its read runs out of time
(closes #70)
- 2026-10-19: the AS, peer and prefix handlers write each batch outside the lock
their message handling needs, so messages keep coming off their queues during
a write; meanwhile the next batch grows to at most ten batch sizes (closes
- 2026-10-19: The AS, peer and prefix handlers write each batch outside the lock
their message handling needs. So messages keep coming off their queues during
a write. Meanwhile the next batch grows to at most ten batch sizes (closes
#65)
- 2026-10-19: a live route is identified by prefix and peer, so an announcement
with a new origin AS replaces the peer's route and a withdrawal removes it
- 2026-10-19: A live route is identified by prefix and peer. So an announcement
with a new origin AS replaces the peer's route. A withdrawal removes it
whatever its origin AS (closes #55)
- 2026-10-14: the streamer reconnects when RIS Live ends the stream or sends
nothing for 60 seconds, and the health check reports the feed disconnected
- 2026-10-14: The streamer reconnects when RIS Live ends the stream or sends
nothing for 60 seconds. The health check reports the feed disconnected
meanwhile (closes #53)
- 2026-10-14: the prefix page takes the names of the origin ASes from the prefix
query instead of reading every route of each origin AS (closes #67)
- 2026-10-14: the daemon exits with status 1 when its web server cannot listen
on its port, instead of running on without it (closes #73)
- 2026-10-14: the prefix pages parse the prefix before looking it up: one that
does not parse answers 400, and any valid way of writing a prefix finds its
routes (closes #71)
- 2026-10-14: the IP lookup ignores routes shorter than /8 for IPv4 and /16 for
IPv6, so a default route is no longer named as an address's owner (closes #57)
- 2026-10-14: a prefix batch keeps each peer's last update for a prefix, instead
of one update per prefix across all peers (closes #54)
- 2026-10-14: The prefix page takes the names of the origin ASes from the prefix
query. It used to read every route of each origin AS (closes #67)
- 2026-10-14: The daemon exits with status 1 when its web server cannot listen
on its port. It used to run on without a web server (closes #73)
- 2026-10-14: The prefix pages parse the prefix before looking it up. One that
does not parse answers 400. Any valid way of writing a prefix finds its routes
(closes #71)
- 2026-10-14: The IP lookup ignores routes shorter than /8 for IPv4 and /16 for
IPv6. So a default route is no longer named as an address's owner (closes #57)
- 2026-10-14: A prefix batch keeps each peer's last update for a prefix. It used
to keep one update per prefix across all peers (closes #54)
- 2026-10-14: `reconnect_count` in `/api/v1/stats` counts every reconnect
(closes #68)
- 2026-10-03: `/api/v1/stats` serves the prefix distribution from memory, seeded
at startup and adjusted on every live-route write, so a request no longer
reads every live route (closes #30)
- 2026-10-03: a new database is created with `auto_vacuum` set to incremental,
through the connection string so it is set before the file is first written,
and each periodic incremental vacuum now returns up to 1000 free pages; the
late `PRAGMA auto_vacuum` in `Initialize`, which SQLite ignored, is gone
(closes #43)
- 2026-10-03: looking up an IP address no longer reads every IPv6 route: both
- 2026-10-03: `/api/v1/stats` serves the prefix distribution from memory. The
distribution is seeded at startup and adjusted on every live-route write. So a
request no longer reads every live route (closes #30)
- 2026-10-03: A new database is created with `auto_vacuum` set to incremental.
The setting goes in the connection string, so it is set before the file is
first written. Each periodic incremental vacuum now returns up to 1000 free
pages. The late `PRAGMA auto_vacuum` in `Initialize`, which SQLite ignored, is
gone (closes #43)
- 2026-10-03: Looking up an IP address no longer reads every IPv6 route. Both
families find the most specific live route with at most 33 or 129 lookups on
the prefix index, and the IPv4 range columns are gone. Prefixes from the feed
are stored in one text form, so an IPv6 withdrawal, which the feed sends
the prefix index. The IPv4 range columns are gone. Prefixes from the feed are
stored in one text form. So an IPv6 withdrawal, which the feed sends
uncompressed, now removes its route (closes #48)
- 2026-10-02: a plain `docker build .` stamps the commit's tag or short commit
(`git describe --tags --always`) into the page footer instead of `unknown`:
`.dockerignore` sends `.git` without `.git/config`, a `VERSION` build argument
takes precedence, and `make build` stamps the same value (closes #46)
- 2026-09-29: the entrypoint creates the data directory if it is missing and
stops the start if a step fails; README "Running under upaas" no longer asks
- 2026-10-02: A plain `docker build .` stamps the commit's tag or short commit
into the page footer. It used to stamp `unknown`. The value comes from
`git describe --tags --always`. `.dockerignore` sends `.git` without
`.git/config`. A `VERSION` build argument takes precedence. `make build`
stamps the same value (closes #46)
- 2026-09-29: The entrypoint creates the data directory if it is missing. It
stops the start if a step fails. README "Running under upaas" no longer asks
for the host directory to be created first (closes #42)
- 2026-09-29: `.dockerignore` keeps `.git`, local build output, local databases
and `.env` out of the Docker build context, and so out of the source archive
in the image (closes #39)
- 2026-09-29: README first line names the MIT license and the author; the
License section now says MIT and links `LICENSE`, and an Author section was
added; this file brought up to date (closes #38)
- 2026-09-29: MIT `LICENSE` (closes #1)
- 2026-09-28: stopping the daemon while the feed is flowing no longer panics
with "send on closed channel": the read loop checks for a stop just before
handing a message to the handler queues, and a second `Stop` no longer closes
the queues again (closes #34)
and `.env` out of the Docker build context. So they stay out of the source
archive in the image (closes #39)
- 2026-09-29: The README's first line names the MIT licence and the author. The
License section says MIT and links `LICENSE`. An Author section is added. This
file is brought up to date (closes #38)
- 2026-09-29: `LICENSE` is added, with the MIT licence (closes #1)
- 2026-09-28: Stopping the daemon while the feed is flowing no longer panics
with "send on closed channel". The read loop checks for a stop just before
handing a message to the handler queues. A second `Stop` no longer closes the
queues again (closes #34)
- 2026-09-28: `docker stop` no longer kills the daemon 2 seconds after the stop
signal: the entrypoint switches to the `routewatch` user with `setpriv`
instead of `runuser`, so the daemon receives the signal itself and gets the
signal. The entrypoint switches to the `routewatch` user with `setpriv`
instead of `runuser`. So the daemon receives the signal itself. It gets the
whole wait `docker stop` allows, up to its own 60-second limit (closes #33)
- 2026-09-28: ready to run under upaas: a set but invalid `PORT`,
`XDG_DATA_HOME` or `MALLOC_ARENA_MAX` stops the start, the health check
follows `PORT`, README "Running under upaas" section (closes #31)
- 2026-09-22: realtime in-memory database statistics: counts seeded at startup
and adjusted on every write, oldest/newest route timestamps via index-end
lookups; `/api/v1/stats` no longer scans the tables (closes #27)
- 2026-09-21: batch writes take the write lock when their transaction begins
(`_txlock=immediate`), so they wait out a WAL checkpoint instead of failing
- 2026-09-28: RouteWatch is ready to run under upaas. A set but invalid `PORT`,
`XDG_DATA_HOME` or `MALLOC_ARENA_MAX` stops the start. The health check
follows `PORT`. README gains a "Running under upaas" section (closes #31)
- 2026-09-22: The database statistics are kept in memory. The counts are seeded
at startup and adjusted on every write. The oldest and newest route times come
from a lookup at each end of an index. `/api/v1/stats` no longer scans the
tables (closes #27)
- 2026-09-21: Batch writes take the write lock when their transaction begins
(`_txlock=immediate`). So they wait out a WAL checkpoint instead of failing
with "database is locked" (closes #25)
- 2026-09-21: `MALLOC_ARENA_MAX=2` in the image caps glibc malloc arenas, so
memory outside the Go runtime no longer grows with the core count (closes #23)
- 2026-09-21: `GOMEMLIMIT=1536MiB` in the image; README Memory section with the
- 2026-09-21: `MALLOC_ARENA_MAX=2` in the image caps the number of glibc malloc
arenas. So memory outside the Go runtime no longer grows with the core count
(closes #23)
- 2026-09-21: The image sets `GOMEMLIMIT=1536MiB`. README "Memory" gives the
memory budget and the 5 GiB container limit (closes #13)
- 2026-09-21: the four handler queues hold at most 20,000 messages each, down
- 2026-09-21: The four handler queues hold at most 20,000 messages each, down
from 100,000 (closes #11)
- 2026-09-21: two goroutine leaks fixed: the stats handlers after a timeout and
the streamer's tickers on every reconnect (closes #12)
- 2026-09-21: parsed RIS messages no longer keep the unused `Community` and
- 2026-09-21: Two goroutine leaks are fixed. The stats handlers leaked after a
timeout. The streamer's tickers leaked on every reconnect (closes #12)
- 2026-09-21: Parsed RIS messages no longer keep the unused `Community` and
`Raw` fields (closes #9)
- 2026-09-21: `.editorconfig`, and a Gitea workflow that runs `script/cibuild`
- 2026-09-21: `.editorconfig` is added. A Gitea workflow runs `script/cibuild`
on every push (closes #14)
- 2026-09-21: the peering handler's AS-path map holds at most 500,000 paths and
is swapped for an empty one every 30 seconds instead of copied (closes #10)
- 2026-09-21: SQLite memory bounded across the whole connection pool: a 64 MiB
page cache on each connection, 1 GiB soft and 1.5 GiB hard heap limits (closes
#8)
- 2026-09-21: the Docker build runs the format check, the linter and the tests,
linting in a separate stage on a golangci-lint image pinned by digest (closes
#5)
- 2026-09-21: `make test` skips the live-network feed test (`-short`), so
- 2026-09-21: The peering handler's AS-path map holds at most 500,000 paths.
Every 30 seconds it is swapped for an empty one instead of copied (closes #10)
- 2026-09-21: SQLite's memory is bounded across the whole connection pool. Each
connection has a 64 MiB page cache. The heap limits are 1 GiB soft and 1.5 GiB
hard (closes #8)
- 2026-09-21: The Docker build runs the format check, the linter and the tests.
Linting runs in a separate stage on a `golangci-lint` image pinned by digest
(closes #5)
- 2026-09-21: `make test` skips the live feed test, with `-short`. So
`make check` no longer depends on the network (closes #2)
- 2026-07-07 Adopted scripts-to-rule-them-all: `script/` entrypoints, Makefile
shims, README Entrypoints section
- 2026-02-22: repo policy compliance: required policy files, .gitignore update,
Makefile fmt-check/check/docker/hooks targets, gofmt pass
(repo-policies-compliance, unmerged)
- 2026-01-01: WAL checkpointing: periodic (5s), on startup with logging,
TRUNCATE mode; fixed slow queries
- 2025-12-31: status page: navbar, home page with search, oldest/newest route
timestamps, NULL handling in WHOIS stats; container runs as routewatch user
with proper state dir
- 2025-12-30: structured HTTP request logging, increased timeouts, CIDR prefix
URL routing fixes, status page metrics and footer
- 2025-12-29: Dockerfile multi-stage build with source archive; SQLite
incremental vacuum for non-blocking space reclamation
- 2025-07-27: initial RouteWatch BGP stream monitor
- 2026-07-07: The scripts-to-rule-them-all layout is adopted. `script/` holds
the entrypoints. The `Makefile` targets `check`, `test`, `lint`, `fmt` and
`fmt-check` call them. So do `bootstrap`, `setup`, `docker` and `hooks`.
README gains an "Entrypoints" section
- 2026-02-22: The repo is brought in line with the repo policies. The required
policy files are added. `.gitignore` is updated. The `Makefile` gains
`fmt-check`, `check`, `docker` and `hooks` targets. The Go code is formatted
with `gofmt`. The work was on the branch `repo-policies-compliance`, unmerged
when this entry was written
- 2026-01-01: The WAL is checkpointed every 5 seconds, and at startup with a log
line. Checkpoints run in `TRUNCATE` mode. This fixed slow queries
- 2025-12-31: The pages get a navigation bar and a home page with a search. The
status page shows the oldest and newest route times. The WHOIS stats query
handles `NULL`. The container runs as the `routewatch` user with the right
state directory
- 2025-12-30: HTTP requests are logged as structured log lines. The HTTP
timeouts are longer. Prefix URLs with a slash in them route correctly. The
status page gets more metrics and a footer
- 2025-12-29: The `Dockerfile` builds in stages and puts a source archive in the
image. SQLite reclaims space with incremental vacuum, which does not block
- 2025-07-27: The first commit: RouteWatch, a BGP stream monitor
# Future Steps
None.
- Finish https://git.eeqj.de/sneak/routewatch/pulls/210 as sneak's review
directs. It merges once a review passes. Of the two PRs for
https://git.eeqj.de/sneak/routewatch/issues/157, the one that merges last
closes the issue.
internal/routetable/overview.go
-4
View File
@@ -10,10 +10,6 @@ import (
"git.eeqj.de/sneak/routewatch/internal/ristypes"
)
// walkBatch is how many prefixes the walk over the whole table reads in one
// hold of the read lock. The updates waiting for the lock get it in between.
const walkBatch = 10000
// Overview holds figures about the whole table. Working them out takes a walk
// over every prefix, so a caller asks for them now and then and keeps them.
type Overview struct {
internal/routetable/routetable.go
+28 -2
View File
@@ -27,6 +27,11 @@ const (
// kept routes in order: each of its nodes holds up to 63 times.
keptTimesDegree = 32
// walkBatch is how many prefixes a walk over the whole table visits in
// one hold of the lock. Overview and PeerDown each walk the whole table.
// Callers waiting for the lock get it between batches.
walkBatch = 10000
ipv4Bits = 32
ipv6Bits = 128
@@ -201,8 +206,13 @@ func (t *Table) Withdraw(prefix netip.Prefix, peer string) {
}
// PeerDown applies a peer's session going down: the peer leaves every peer
// set, and every route of it is dropped. When the peer carries any prefix, it
// visits every prefix in the table.
// set, and every route of the peer is dropped. When the peer carries any
// prefix, PeerDown visits every prefix in the table, in batches of walkBatch.
// After each batch, it lets the table's lock go for as long as it held the
// lock for the batch. During a walk over many batches, reads and updates can
// then take the lock about half of the time. An announcement by the peer
// applied during the walk may be dropped too. So the caller applies the peer's
// later updates only once PeerDown has returned.
func (t *Table) PeerDown(peer string) {
t.mu.Lock()
defer t.mu.Unlock()
@@ -212,6 +222,9 @@ func (t *Table) PeerDown(peer string) {
return
}
visited := 0
batchStart := time.Now()
for _, f := range []*family{&t.ipv4, &t.ipv6} {
for _, prefixes := range f.byLength {
for prefix, origins := range prefixes {
@@ -219,6 +232,19 @@ func (t *Table) PeerDown(peer string) {
if o != nil {
t.leave(prefix, o, number)
}
// Go allows the map to change before the next step of
// the range over it.
visited++
if visited%walkBatch == 0 {
held := time.Since(batchStart)
t.mu.Unlock()
time.Sleep(held)
t.mu.Lock()
batchStart = time.Now()
}
}
}
}
internal/routetable/routetable_internal_test.go
+69
View File
@@ -0,0 +1,69 @@
package routetable
import (
"net/netip"
"runtime"
"testing"
"time"
)
// TestPeerDownLetsReadsIn reads the table's counts again and again while a peer
// that carries more than two batches of prefixes goes down. PeerDown lets the
// lock go after every walkBatch prefixes, so a read must get in while the peer
// still carries some of its prefixes but no longer all.
func TestPeerDownLetsReadsIn(t *testing.T) {
t.Parallel()
const (
asn = 1000
peer = "192.0.2.1"
// prefixCount is more prefixes than two of walkBatch.
prefixCount = 2*walkBatch + 1
)
routeTime := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
table := New(Params{})
addr := netip.MustParseAddr("1.0.0.0")
for range prefixCount {
prefix := netip.PrefixFrom(addr, addr.BitLen())
table.Announce(prefix, asn, Route{Peer: peer, Time: routeTime})
addr = addr.Next()
}
// Hold a read lock until PeerDown waits for the lock, so that the first
// read below starts once PeerDown holds it. TryRLock fails once a writer
// waits.
table.mu.RLock()
peerDownDone := make(chan struct{})
go func() {
defer close(peerDownDone)
table.PeerDown(peer)
}()
for table.mu.TryRLock() {
table.mu.RUnlock()
runtime.Gosched()
}
table.mu.RUnlock()
for {
routes := table.Counts().IPv4Routes
if routes > 0 && routes < prefixCount {
break
}
select {
case <-peerDownDone:
t.Fatal("no read got in while PeerDown ran")
default:
}
}
<-peerDownDone
}
internal/routewatch/app_integration_internal_test.go
+15 -25
View File
@@ -509,8 +509,6 @@ func announcementFeed(prefixes []string) string {
func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
t.Parallel()
const maxWait = 10 * time.Second
prefixes := []string{"192.0.2.0/24", "198.51.100.0/24", "203.0.113.0/24"}
feed := announcementFeed(prefixes)
@@ -556,9 +554,9 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
for range cap(linesSeen) {
select {
case <-linesSeen:
case <-time.After(maxWait):
case <-time.After(stepTimeout):
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 {
case <-shutDown:
case <-time.After(maxWait):
t.Fatalf("Shutdown did not return within %s", maxWait)
case <-time.After(stepTimeout):
t.Fatalf("Shutdown did not return within %s", stepTimeout)
}
store.mu.Lock()
@@ -609,21 +607,16 @@ func freePort(t *testing.T) int {
return addr.Port
}
// TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked starts the HTTP server
// and has the AS handler start a write that the store holds until the test
// ends, past Shutdown's deadline. Shutdown must return within its deadline
// plus a small margin, with the HTTP server stopped, and must neither
// checkpoint nor close the database: the real one would wait for the write.
// TestShutdownDoesNotWaitForABlockedWrite starts the HTTP server and has the
// AS handler start a write that the store holds until the test ends, past
// Shutdown's deadline. Shutdown must return without waiting for the write, with
// the HTTP server stopped, and must neither checkpoint nor close the database:
// the real one would wait for the write.
// Before the fix Shutdown waited for the write.
func TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) {
func TestShutdownDoesNotWaitForABlockedWrite(t *testing.T) {
t.Parallel()
const (
deadline = 100 * time.Millisecond
// margin is how long Shutdown may take after its deadline, to stop
// the HTTP server.
margin = 2 * time.Second
)
const deadline = 100 * time.Millisecond
blocking := newBlockingStore()
defer blocking.releaseWrites()
@@ -651,9 +644,8 @@ func TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) {
}
rw.dbHandler = NewASHandler(store, log)
fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
startBlockedWrite(t, blocking, rw.dbHandler,
[]*ristypes.RISMessage{{Path: fill}})
[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}})
ctx, cancel := context.WithTimeout(t.Context(), deadline)
defer cancel()
@@ -667,9 +659,9 @@ func TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) {
select {
case <-shutDown:
case <-time.After(deadline + margin):
case <-time.After(deadline + stepTimeout):
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
@@ -744,8 +736,6 @@ func (s *closeOrderStore) record(call string) {
func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
t.Parallel()
const deadline = 10 * time.Second
store := &closeOrderStore{Store: newMockStore()}
log := logger.New()
cfg := &config.Config{}
@@ -763,7 +753,7 @@ func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
Path: ristypes.ASPath{Sequence: []int{testASNA}},
})
ctx, cancel := context.WithTimeout(t.Context(), deadline)
ctx, cancel := context.WithTimeout(t.Context(), stepTimeout)
defer cancel()
rw.Shutdown(ctx)
internal/routewatch/ashandler.go
+42 -87
View File
@@ -18,39 +18,34 @@ const (
// (asnBatchTimeout), so a queue smaller than asnBatchSize is fine.
asHandlerQueueSize = 20000
// asnBatchSize is the number of ASN operations to batch together
// asnBatchSize is the number of ASes in a full batch, which wakes the
// flush goroutine to write it.
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
// DO NOT reduce this timeout - larger batches are more efficient
asnBatchTimeout = 2 * time.Second
)
// ASHandler processes Autonomous System Number (ASN) information extracted from
// BGP UPDATE messages. It uses batched database operations to efficiently store
// ASN data, collecting operations into batches that are flushed either when the
// batch reaches a size threshold or after a timeout period.
// ASHandler records every AS in the AS paths of BGP UPDATE messages. Its batch
// holds the earliest and the latest time of each AS's updates. The flush
// goroutine writes the batch to the database when it holds asnBatchSize ASes,
// and every asnBatchTimeout.
type ASHandler struct {
db database.Store
logger *logger.Logger
// Batching. mu guards batch and writing only: the flush goroutine takes
// the batch under mu and writes it after releasing mu, so HandleMessage
// waits for a database write only while batch is at asnBatchMaxSize.
mu sync.Mutex
batch []asnOp
// writing is how many operations the flush goroutine took from batch
// and has not finished writing.
writing int
// batchTaken is signalled when the flush goroutine takes batch; its L is
// &mu.
batchTaken sync.Cond
// Batching. mu guards batch and writing only. The flush goroutine takes
// the batch under mu and writes it after releasing mu. So HandleMessage
// never waits for a database write.
mu sync.Mutex
// batch holds the earliest and the latest time of the updates for each AS
// since the flush goroutine last took it. It holds one entry per AS,
// however many updates arrive.
batch map[int]database.SeenTimes
// writing is how many ASes the flush goroutine took from batch and has
// not finished writing.
writing int
batchFullCh chan struct{} // wakes the flush goroutine when batch is full
// 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.
@@ -58,16 +53,10 @@ type ASHandler struct {
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
// the batching system and starts a background goroutine that periodically
// flushes accumulated ASN operations to the database. The caller must call
// Stop when finished to ensure all pending operations are flushed and the
// background goroutine is terminated.
// writes the collected ASes to the database. The caller must call Stop when
// finished, to write the pending ASes and end the background goroutine.
func NewASHandler(db database.Store, logger *logger.Logger) *ASHandler {
return newASHandler(db, logger, asnBatchTimeout)
}
@@ -81,12 +70,11 @@ func newASHandler(
h := &ASHandler{
db: db,
logger: logger,
batch: make([]asnOp, 0, asnBatchSize),
batch: make(map[int]database.SeenTimes),
batchFullCh: make(chan struct{}, 1),
stopCh: make(chan context.Context, 1),
done: make(chan struct{}),
}
h.batchTaken.L = &h.mu
// Start the flush timer goroutine
go flushLoop(batchTimeout, h.batchFullCh, h.stopCh, h.done, h.flushBatch)
@@ -110,48 +98,25 @@ func (h *ASHandler) QueueCapacity() int {
return asHandlerQueueSize
}
// HandleMessage processes a RIS Live BGP message by extracting all ASNs from
// the AS path and queuing them for batch insertion into the database. The
// 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.
// HandleMessage adds the time of msg to the batch entry of each AS in its AS
// path, the origin AS and the members of an AS set included. When the batch
// holds asnBatchSize ASes, HandleMessage wakes the flush goroutine to write it.
func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
// Use the pre-parsed timestamp
timestamp := msg.ParsedTimestamp
originASN := msg.Path.Origin()
h.mu.Lock()
defer h.mu.Unlock()
// 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
// The origin AS is the last AS of the sequence, so the loop adds it too.
for _, asn := range slices.Concat(msg.Path.Sequence, msg.Path.Set) {
if asn > 0 {
h.batch = append(h.batch, asnOp{
number: asn,
timestamp: timestamp,
})
seen := h.batch[asn]
seen.Add(msg.ParsedTimestamp)
h.batch[asn] = seen
}
}
// 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
// or the batch reaches asnBatchMaxSize.
// waits in batchFullCh and the batch keeps taking updates until the write
// ends.
if len(h.batch) >= asnBatchSize {
select {
case h.batchFullCh <- struct{}{}:
@@ -160,10 +125,11 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) {
}
}
// Stop stops the flush goroutine, which writes the operations the handler holds
// before it returns, and waits for it until ctx ends. Once ctx has ended the
// flush goroutine skips that last write, and a write still running is lost
// when the process exits, so Stop logs how many operations are not written.
// Stop stops the flush goroutine and waits for it until ctx ends. The flush
// goroutine writes the batch before it returns, unless ctx has ended. A write
// still running when the process exits is lost. Stop logs how many entries are
// not written, in the batch being written and in the next batch. An AS in both
// batches counts twice.
func (h *ASHandler) Stop(ctx context.Context) {
h.stopCh <- ctx
@@ -178,14 +144,14 @@ func (h *ASHandler) Stop(ctx context.Context) {
if unwritten > 0 {
h.logger.Error("Stop deadline passed before the AS handler wrote "+
"its updates", "updates", unwritten)
"its batches", "batch_entries", unwritten)
}
}
// takeBatch returns the pending operations and leaves an empty batch in their
// place for HandleMessage to fill. They count in writing until flushBatch has
// written them.
func (h *ASHandler) takeBatch() []asnOp {
// takeBatch returns the pending ASes and leaves an empty batch in their place
// for HandleMessage to fill. They count in writing until flushBatch has written
// them.
func (h *ASHandler) takeBatch() map[int]database.SeenTimes {
h.mu.Lock()
defer h.mu.Unlock()
@@ -195,8 +161,7 @@ func (h *ASHandler) takeBatch() []asnOp {
batch := h.batch
h.writing = len(batch)
h.batch = make([]asnOp, 0, asnBatchSize)
h.batchTaken.Broadcast()
h.batch = make(map[int]database.SeenTimes)
return batch
}
@@ -209,21 +174,11 @@ func (h *ASHandler) flushBatch() {
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
err := h.db.GetOrCreateASNBatch(asns)
err := h.db.GetOrCreateASNBatch(batch)
if err != nil {
h.logger.Error("Failed to process ASN batch",
"error", err, "count", len(asns))
"error", err, "count", len(batch))
}
h.mu.Lock()
internal/routewatch/ashandler_internal_test.go
+24 -7
View File
@@ -5,28 +5,43 @@ import (
"slices"
"testing"
"git.eeqj.de/sneak/routewatch/internal/database"
"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
// whose path carries an AS the database does not hold yet, the second 5
// seconds after the first, in one batch into a real database. The AS is stored
// first seen at the time of the first update and last seen at the time of the
// second.
// whose path carries an AS the database does not hold yet, in one batch into a
// real database. The second is 5 seconds after the first, and the handler gets
// it first. The AS is stored first seen at the time of the first update and
// last seen at the time of the second.
func TestASHandlerStoresTheEarliestAndLatestTimesOfANewAS(t *testing.T) {
t.Parallel()
db := newTestDatabase(t)
// 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(), batch: make(map[int]database.SeenTimes),
}
first := peerAnnouncement(testPeerIP, testV4Prefix)
second := peerAnnouncement(testPeerIP, testV4Prefix)
second.ParsedTimestamp = first.ParsedTimestamp.Add(secondUpdateAfter)
h.HandleMessage(first)
h.HandleMessage(second)
h.HandleMessage(first)
h.flushBatch()
asn, _, err := db.GetAS(t.Context(), testASNA)
@@ -54,7 +69,9 @@ func TestASHandlerRecordsNothingFromAPathWithASZero(t *testing.T) {
store := newMockStore()
// 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(), batch: make(map[int]database.SeenTimes),
}
msg := peerAnnouncement(testPeerIP, testV4Prefix)
msg.Path = decodePath(t, testASNA, 0, testASNB)
internal/routewatch/asnfetcher.go
+31 -5
View File
@@ -187,25 +187,33 @@ func (f *ASNFetcher) recordSuccess() {
f.statsMu.Unlock()
}
// recordFailure increases the interval on failed fetch using exponential backoff.
func (f *ASNFetcher) recordFailure() {
// lengthenInterval makes the interval 1.5 times as long, up to maxInterval.
// A failed lookup and a skipped one both lengthen it.
func (f *ASNFetcher) lengthenInterval() {
f.intervalMu.Lock()
f.consecutiveFails++
defer f.intervalMu.Unlock()
// Exponential backoff: multiply by 1.5, capped at max
newInterval := min(
f.currentInterval*backoffNumerator/backoffDenominator, maxInterval)
if newInterval != f.currentInterval {
f.logger.Debug("Increased fetch interval due to failure",
f.logger.Debug("Increased fetch interval",
"old_interval", f.currentInterval,
"new_interval", newInterval,
"consecutive_failures", f.consecutiveFails,
)
f.currentInterval = newInterval
}
}
// recordFailure counts a failed fetch and lengthens the interval.
func (f *ASNFetcher) recordFailure() {
f.intervalMu.Lock()
f.consecutiveFails++
f.intervalMu.Unlock()
f.lengthenInterval()
// Record error time for stats, dropping those older than statsWindow
f.statsMu.Lock()
now := f.now()
@@ -287,6 +295,16 @@ func (f *ASNFetcher) fetchNextStale(ctx context.Context) {
// 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
// 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 and is skipped. A skipped
// lookup leaves the AS stale and is not counted as a failure. Like a
// failure, a skipped lookup lengthens the interval. Each lookup asks IANA
// before it finds its RIR paused, so with a one-second interval the fetcher
// would ask IANA every second. The lookups that QueueImmediate asks for also
// run through fetchAndUpdate.
func (f *ASNFetcher) fetchAndUpdate(ctx context.Context, asn int) bool {
ctx, cancel := context.WithTimeout(ctx, time.Minute)
defer cancel()
@@ -307,6 +325,14 @@ func (f *ASNFetcher) fetchAndUpdate(ctx context.Context, asn int) bool {
return false
}
if errors.Is(err, whois.ErrPaused) {
f.logger.Info("Skipped WHOIS lookup, the RIR's server is paused",
"asn", asn, "error", err)
f.lengthenInterval()
return false
}
if err != nil {
f.logger.Error("WHOIS lookup failed", "asn", asn, "error", err)
f.recordFailure()
internal/routewatch/asnfetcher_internal_test.go
+150 -4
View File
@@ -13,6 +13,7 @@ import (
"git.eeqj.de/sneak/routewatch/internal/database"
"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
@@ -131,6 +132,151 @@ 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)
}
}
// TestFetchAndUpdateBacksOffDuringAPause looks up an AS at RIPE while RIPE
// refuses the query. The refusal pauses RIPE for an hour, so every later
// lookup in the test is skipped. The test sets the WHOIS client's clock and
// looks the AS up again each time the interval has passed. Like a failure,
// each skip must make the interval 1.5 times as long, up to one minute. The
// skips must not count as failures.
func TestFetchAndUpdateBacksOffDuringAPause(t *testing.T) {
t.Parallel()
f, _ := newFetcherWithWHOISAnswers(t, map[string]string{
ianaWHOISServer: ianaRefersToRIPE,
ripeWHOISServer: ripeAccessDenied,
})
now := time.Date(2026, 11, 2, 0, 0, 0, 0, time.UTC)
f.whoisClient.Now = func() time.Time { return now }
f.fetchAndUpdate(t.Context(), testASNA)
for i, want := range []time.Duration{
33750 * time.Millisecond,
50625 * time.Millisecond,
time.Minute,
time.Minute,
} {
now = now.Add(f.getInterval())
f.fetchAndUpdate(t.Context(), testASNA)
if got := f.getInterval(); got != want {
t.Errorf("interval after %d skipped lookups = %v, want %v",
i+1, got, want)
}
}
got := f.GetStats()
if got.ConsecutiveFails != 1 || got.ErrorsLastHour != 1 {
t.Errorf("%d failures in a row and %d errors, want 1 and 1",
got.ConsecutiveFails, got.ErrorsLastHour)
}
}
// 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
// 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.
@@ -224,7 +370,7 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
select {
case <-lookupStarted:
case <-time.After(handleMessageTimeout):
case <-time.After(stepTimeout):
t.Fatal("the fetcher did not start the lookup")
}
@@ -240,22 +386,22 @@ func TestASNFetcherStopReturnsByItsDeadline(t *testing.T) {
select {
case <-f.done:
case <-time.After(handleMessageTimeout):
case <-time.After(stepTimeout):
t.Fatal("the fetcher goroutine did not return after the lookup ended")
}
}
// 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
// statsWindow.
func TestRecordKeepsOnlyTheStatsWindow(t *testing.T) {
t.Parallel()
const (
records = 100_000
step = time.Second
want = int(statsWindow / step)
records = 3 * want
)
clock := time.Now()
internal/routewatch/batchwrite_internal_test.go
+116 -59
View File
@@ -5,6 +5,7 @@ import (
"context"
"encoding/json"
"fmt"
"io"
"log/slog"
"slices"
"sync"
@@ -19,9 +20,15 @@ import (
)
const (
// handleMessageTimeout is how long a test waits for HandleMessage to
// return, or for a write to start, before it fails.
handleMessageTimeout = 2 * time.Second
// stepTimeout is how long a test waits for a step it expects, such as
// HandleMessage returning, a write starting or Shutdown returning, before
// 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
// wait. One that does not wait returns well within it.
@@ -171,12 +178,14 @@ type stopLog struct {
Msg string `json:"msg"`
Updates int `json:"updates"`
TableChanges int `json:"table_changes"`
HeldMessages int `json:"held_messages"`
Paths int `json:"paths"`
BatchEntries int `json:"batch_entries"`
}
// newErrorLogger returns a logger that writes only errors, as JSON lines, to
// logs.
func newErrorLogger(logs *bytes.Buffer) *logger.Logger {
func newErrorLogger(logs io.Writer) *logger.Logger {
return &logger.Logger{Logger: slog.New(slog.NewJSONHandler(logs,
&slog.HandlerOptions{Level: slog.LevelError}))}
}
@@ -198,7 +207,7 @@ func lastLogged(t *testing.T, logs *bytes.Buffer) stopLog {
return logged
}
// returnsInTime reports whether f returns within handleMessageTimeout.
// returnsInTime reports whether f returns within stepTimeout.
func returnsInTime(f func()) bool {
done := make(chan struct{})
@@ -210,11 +219,22 @@ func returnsInTime(f func()) bool {
select {
case <-done:
return true
case <-time.After(handleMessageTimeout):
case <-time.After(stepTimeout):
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
// 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
@@ -240,7 +260,7 @@ func startBlockedWrite(
select {
case <-store.writeStarted:
case <-time.After(handleMessageTimeout):
case <-time.After(stepTimeout):
t.Fatal("the handler did not start to write the messages")
}
}
@@ -261,7 +281,7 @@ func handleWhileAWriteIsBlocked(
) {
t.Helper()
defer h.Stop(t.Context())
defer stopInTime(t, h)
defer store.releaseWrites()
startBlockedWrite(t, store, h, fill)
@@ -275,7 +295,9 @@ func handleWhileAWriteIsBlocked(
// 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
// 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(
t *testing.T,
store *blockingStore,
@@ -288,13 +310,18 @@ func handleAtMaxBatchSize(
) {
t.Helper()
defer h.Stop(t.Context())
defer stopInTime(t, h)
defer store.releaseWrites()
startBlockedWrite(t, store, h, fill)
for _, msg := range toMax {
h.HandleMessage(msg)
if !returnsInTime(func() {
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{})
@@ -315,7 +342,7 @@ func handleAtMaxBatchSize(
select {
case <-returned:
case <-time.After(handleMessageTimeout):
case <-time.After(stepTimeout):
t.Fatal("HandleMessage did not return after the database write finished")
}
}
@@ -345,8 +372,12 @@ func stopWhileAWriteIsBlocked(
startBlockedWrite(t, store, h, fill)
for _, msg := range extra {
h.HandleMessage(msg)
if !returnsInTime(func() {
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)
@@ -361,7 +392,7 @@ func stopWhileAWriteIsBlocked(
select {
case <-done:
case <-time.After(handleMessageTimeout):
case <-time.After(stepTimeout):
t.Fatal("the flush goroutine did not return after the write finished")
}
}
@@ -380,7 +411,8 @@ func TestPrefixHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
store := newBlockingStore()
h := newPrefixHandler(store, newErrorLogger(&logs),
routetable.New(routetable.Params{}), longerThanAnyTest)
routetable.New(routetable.Params{}), longerThanAnyTest,
maxTableChangesPerWrite)
fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
stopWhileAWriteIsBlocked(t, store, h, h.done,
[]*ristypes.RISMessage{prefixAnnouncement(fill)},
@@ -466,20 +498,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) {
t.Parallel()
// 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
const wantEntries = asnBatchSize + 1
var logs bytes.Buffer
store := newBlockingStore()
h := newASHandler(store, newErrorLogger(&logs), longerThanAnyTest)
fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
stopWhileAWriteIsBlocked(t, store, h, h.done,
[]*ristypes.RISMessage{{Path: fill}},
[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}},
&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})
store.mu.Lock()
@@ -489,26 +521,27 @@ func TestASHandlerSkipsItsLastWriteAfterTheStopDeadline(t *testing.T) {
t.Errorf("AS%d was written after the stop deadline", testASNB)
}
if logged := lastLogged(t, &logs); logged.Updates != wantUpdates {
t.Errorf("Stop logged %d updates not written, want %d",
logged.Updates, wantUpdates)
if logged := lastLogged(t, &logs); logged.BatchEntries != wantEntries {
t.Errorf("Stop logged %d batch entries not written, want %d",
logged.BatchEntries, wantEntries)
}
}
// TestASHandlerSkipsAFullBatchAfterTheStopDeadline hands the AS handler a full
// batch while a write is held, so the batch's wake-up of the flush goroutine
// waits. The stop deadline passes meanwhile. When the write ends, the flush
// goroutine must stop instead of writing the batch.
// waits. The full batch holds the fill's ASes again, and testASNB. The stop
// deadline passes meanwhile. When the write ends, the flush goroutine must stop
// instead of writing the batch.
func TestASHandlerSkipsAFullBatchAfterTheStopDeadline(t *testing.T) {
t.Parallel()
store := newBlockingStore()
h := newASHandler(store, logger.New(), longerThanAnyTest)
fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
full := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNB}, asnBatchSize)}
fill := &ristypes.RISMessage{Path: pathOfASes(asnBatchSize)}
stopWhileAWriteIsBlocked(t, store, h, h.done,
[]*ristypes.RISMessage{{Path: fill}},
&ristypes.RISMessage{Path: full})
[]*ristypes.RISMessage{fill},
fill,
&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})
store.mu.Lock()
defer store.mu.Unlock()
@@ -597,7 +630,7 @@ func TestPeeringHandlerStopCountsTheWriteInProgress(t *testing.T) {
select {
case <-h.done:
case <-time.After(handleMessageTimeout):
case <-time.After(stepTimeout):
t.Fatal("the processing goroutine did not return after the write " +
"finished")
}
@@ -618,7 +651,7 @@ func TestPeeringHandlerStartsNoWriteAfterTheStopDeadline(t *testing.T) {
const processInterval = 10 * time.Millisecond
store := newBlockingStore()
h := newPeeringHandler(store, logger.New(), processInterval)
h := newPeeringHandler(store, logger.New(), processInterval, maxTrackedPaths)
stopWhileAWriteIsBlocked(t, store, h, h.done,
[]*ristypes.RISMessage{
{Path: ristypes.ASPath{Sequence: []int{testASNA, testASNB}}},
@@ -674,9 +707,8 @@ func TestASHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) {
store := newBlockingStore()
h := newASHandler(store, logger.New(), longerThanAnyTest)
fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
handleWhileAWriteIsBlocked(t, store, h,
[]*ristypes.RISMessage{{Path: fill}},
[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}},
&ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}})
store.mu.Lock()
@@ -714,9 +746,11 @@ func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
const extraPrefix = "198.51.100.0/24"
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)
toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchMaxSize)
toMax := slices.Repeat([]string{testV4Prefix}, prefixBatchSize)
handleAtMaxBatchSize(t, store, h,
[]*ristypes.RISMessage{prefixAnnouncement(fill)},
[]*ristypes.RISMessage{prefixAnnouncement(toMax)},
@@ -731,31 +765,48 @@ func TestPrefixHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
}
}
func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
// TestASHandlerNeverWaitsForAWrite starts a write that the store holds. Then it
// 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()
const messages = 20
store := newBlockingStore()
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}}})
store.mu.Lock()
defer store.mu.Unlock()
defer stopInTime(t, h)
defer store.releaseWrites()
if !store.asns[testASNB] {
t.Errorf("Stop did not write AS%d, handled after the batch reached its "+
"maximum size", testASNB)
startBlockedWrite(t, store, h,
[]*ristypes.RISMessage{{Path: pathOfASes(asnBatchSize)}})
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.
// 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
// would hang.
// would fail.
func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
t.Parallel()
@@ -763,10 +814,11 @@ func TestPeerHandlerWaitsAtTheMaximumBatchSize(t *testing.T) {
store := newBlockingStore()
h := newPeerHandler(store, logger.New(), longerThanAnyTest)
h.batchMaxSize = peerBatchSize
message := []*ristypes.RISMessage{{Peer: testPeerIP}}
handleAtMaxBatchSize(t, store, h,
slices.Repeat(message, peerBatchSize),
slices.Repeat(message, peerBatchMaxSize),
slices.Repeat(message, peerBatchSize),
&ristypes.RISMessage{Peer: otherPeerIP})
store.mu.Lock()
@@ -796,8 +848,9 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
batchSize: prefixBatchSize,
handle: func(n int) chan struct{} {
h := &PrefixHandler{
batchFullCh: make(chan struct{}, 1),
table: routetable.New(routetable.Params{}),
batchMaxSize: prefixBatchMaxSize,
batchFullCh: make(chan struct{}, 1),
table: routetable.New(routetable.Params{}),
}
h.HandleMessage(prefixAnnouncement(
slices.Repeat([]string{testV4Prefix}, n)))
@@ -809,11 +862,12 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
name: "AS",
batchSize: asnBatchSize,
handle: func(n int) chan struct{} {
h := &ASHandler{batchFullCh: make(chan struct{}, 1)}
// The origin AS and each AS in the path are one update each.
h.HandleMessage(&ristypes.RISMessage{
Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, n-1)},
})
h := &ASHandler{
batch: make(map[int]database.SeenTimes),
batchFullCh: make(chan struct{}, 1),
}
// Each of the path's n ASes is one entry of the batch.
h.HandleMessage(&ristypes.RISMessage{Path: pathOfASes(n)})
return h.batchFullCh
},
@@ -822,7 +876,10 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) {
name: "peer",
batchSize: peerBatchSize,
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 {
h.HandleMessage(&ristypes.RISMessage{Peer: testPeerIP})
}
internal/routewatch/dbmaintainer_internal_test.go
+2 -2
View File
@@ -97,7 +97,7 @@ func TestDBMaintainerStopReturnsByItsDeadline(t *testing.T) {
select {
case <-store.started:
case <-time.After(handleMessageTimeout):
case <-time.After(stepTimeout):
t.Fatal("the maintainer did not start the checkpoint")
}
@@ -113,7 +113,7 @@ func TestDBMaintainerStopReturnsByItsDeadline(t *testing.T) {
select {
case <-m.done:
case <-time.After(handleMessageTimeout):
case <-time.After(stepTimeout):
t.Fatal("the maintenance goroutine did not return after the " +
"checkpoint ended")
}
internal/routewatch/peerhandler.go
+14 -9
View File
@@ -43,10 +43,14 @@ type PeerHandler struct {
db database.Store
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
// takes the batch under mu and writes it after releasing mu, so
// HandleMessage waits for a database write only while peerBatch is at
// peerBatchMaxSize.
// batchMaxSize.
mu sync.Mutex
peerBatch []peerUpdate
// 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,
) *PeerHandler {
h := &PeerHandler{
db: db,
logger: logger,
peerBatch: make([]peerUpdate, 0, peerBatchSize),
batchFullCh: make(chan struct{}, 1),
stopCh: make(chan context.Context, 1),
done: make(chan struct{}),
db: db,
logger: logger,
batchMaxSize: peerBatchMaxSize,
peerBatch: make([]peerUpdate, 0, peerBatchSize),
batchFullCh: make(chan struct{}, 1),
stopCh: make(chan context.Context, 1),
done: make(chan struct{}),
}
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
// takes it. Meanwhile the handler takes nothing off its queue, which
// fills, and the streamer drops messages for it.
for len(h.peerBatch) >= peerBatchMaxSize {
for len(h.peerBatch) >= h.batchMaxSize {
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
// 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 {
select {
case h.batchFullCh <- struct{}{}:
internal/routewatch/peerhandler_internal_test.go
+1 -1
View File
@@ -35,7 +35,7 @@ func TestPeerHandlerWritesTheEarliestAndLatestTimesOfANewPeer(t *testing.T) {
// Built without NewPeerHandler's flush goroutine, so the test flushes
// itself.
h := &PeerHandler{db: store, logger: logger.New()}
h := &PeerHandler{db: store, logger: logger.New(), batchMaxSize: peerBatchMaxSize}
first := peerAnnouncement(testPeerIP, testV4Prefix)
second := peerAnnouncement(testPeerIP, testV4Prefix)
internal/routewatch/peeringhandler.go
+16 -9
View File
@@ -42,6 +42,10 @@ type PeeringHandler struct {
db database.Store
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
// and the latest time an update carried it.
mu sync.Mutex
@@ -62,21 +66,24 @@ type PeeringHandler struct {
// periodically processes accumulated AS paths into peering records. The
// handler begins processing immediately upon creation.
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,
// processInterval, in place of peeringProcessInterval. A test gives a short
// one to have the timer start a write.
// processInterval, in place of peeringProcessInterval, and the most paths it
// 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(
db database.Store, logger *logger.Logger, processInterval time.Duration,
maxPaths int,
) *PeeringHandler {
h := &PeeringHandler{
db: db,
logger: logger,
asPaths: make(map[string]database.SeenTimes),
stopCh: make(chan context.Context, 1),
done: make(chan struct{}),
db: db,
logger: logger,
asPaths: make(map[string]database.SeenTimes),
maxPaths: maxPaths,
stopCh: make(chan context.Context, 1),
done: make(chan struct{}),
}
// Start the periodic processing goroutine
@@ -131,7 +138,7 @@ func (h *PeeringHandler) HandleMessage(msg *ristypes.RISMessage) {
defer h.mu.Unlock()
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.
h.droppedPaths++
internal/routewatch/peeringhandler_internal_test.go
+31 -18
View File
@@ -5,7 +5,6 @@ import (
"fmt"
"maps"
"slices"
"strconv"
"sync"
"testing"
"time"
@@ -20,6 +19,10 @@ const (
testASNB = 64501
testASNC = 64502
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
@@ -52,9 +55,10 @@ func (r *recordingStore) callCount() int {
// processing goroutine, so tests drive processing explicitly.
func newTestHandler(db database.Store) *PeeringHandler {
return &PeeringHandler{
db: db,
logger: logger.New(),
asPaths: make(map[string]database.SeenTimes),
db: db,
logger: logger.New(),
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
// 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) {
t.Parallel()
store := &recordingStore{mockStore: newMockStore()}
h := newTestHandler(store)
h := newPeeringHandler(store, logger.New(), longerThanAnyTest,
testMaxTrackedPaths)
defer stopInTime(t, h)
now := time.Now().UTC()
// Fill the map to exactly maxTrackedPaths, including one real path key so
// the "already tracked" branch can be exercised. The filler keys are never
// processed in this test, so their contents do not matter.
// Fill the map to exactly its limit, including one real path key so the
// "already tracked" branch can be exercised. The other paths are of one
// AS, so they carry no peering.
existing := pathKey(t, testASNA, testASNB)
h.mu.Lock()
h.asPaths[existing] = database.SeenTimes{First: now, Last: now}
for i := 0; len(h.asPaths) < maxTrackedPaths; i++ {
h.asPaths[strconv.Itoa(i)] = database.SeenTimes{First: now, Last: now}
for asn := 1; len(h.asPaths) < testMaxTrackedPaths; asn++ {
h.asPaths[pathKey(t, asn)] = database.SeenTimes{First: now, Last: now}
}
h.mu.Unlock()
@@ -270,8 +277,9 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
handle(h, now.Add(time.Second), testASNA, testASNC)
tracked, dropped := snapshot(h)
if tracked != maxTrackedPaths {
t.Fatalf("want map size %d after drop, got %d", maxTrackedPaths, tracked)
if tracked != testMaxTrackedPaths {
t.Fatalf("want map size %d after drop, got %d", testMaxTrackedPaths,
tracked)
}
if dropped != 1 {
@@ -283,8 +291,9 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
handle(h, refreshed, testASNA, testASNB)
tracked, dropped = snapshot(h)
if tracked != maxTrackedPaths {
t.Fatalf("want map size %d after refresh, got %d", maxTrackedPaths, tracked)
if tracked != testMaxTrackedPaths {
t.Fatalf("want map size %d after refresh, got %d", testMaxTrackedPaths,
tracked)
}
if dropped != 1 {
@@ -303,17 +312,21 @@ func TestPeeringHandlerCapDropsAndCounts(t *testing.T) {
// TestPeeringHandlerRunResetsTheDroppedCount fills the map, has the handler
// 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) {
t.Parallel()
h := newTestHandler(newMockStore())
h := newPeeringHandler(newMockStore(), logger.New(), longerThanAnyTest,
testMaxTrackedPaths)
defer stopInTime(t, h)
now := time.Now().UTC()
// Paths of one AS carry no peering, so the run records nothing.
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.mu.Unlock()
internal/routewatch/prefixhandler.go
+193 -53
View File
@@ -44,6 +44,12 @@ const (
// most one pair. So the million pairs that a full-table peer's session
// going down changes are written in about a minute.
maxTableChangesPerWrite = 20000
// maxHeldBackMessages is the most messages the handler holds back from
// peers whose PeerDown has not returned. A message takes about 1 KiB, so
// that many take about 20 MiB (README "Memory"). At this number
// HandleMessage waits until one of them is applied.
maxHeldBackMessages = 20000
)
// PrefixHandler keeps the routing table in memory current from the feed. It
@@ -52,6 +58,15 @@ const (
// does an announcement whose AS path is empty or holds AS 0 anywhere. A peer
// whose BGP session goes down leaves every prefix.
//
// The table's PeerDown applies a session going down. It visits every prefix,
// which takes seconds on a full table. So the handler runs PeerDown on
// a separate goroutine, one peer at a time, and goes on taking messages
// meanwhile. PeerDown lets the table's lock go about half of the time, so the
// handler applies other peers' messages while PeerDown runs. The handler holds
// back the down peer's later messages until PeerDown returns. Then the handler
// applies them in order, one at a time, and takes new messages in between. It
// holds back at most maxHeldBackMessages messages.
//
// Each flush writes the table's changes to the database. It also stores each
// prefix the updates name with the time it was first seen.
type PrefixHandler struct {
@@ -60,10 +75,17 @@ type PrefixHandler struct {
metrics *metrics.Tracker
table *routetable.Table
// Batching. mu guards batch, writing and writingTableChanges only: the
// 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: the
// flush goroutine takes them under mu and writes them after releasing
// mu, so HandleMessage waits for a database write only while batch is at
// prefixBatchMaxSize.
// batchMaxSize.
mu sync.Mutex
batch []seenPrefix
// writing is how many updates the flush goroutine took and has not
@@ -80,6 +102,21 @@ type PrefixHandler struct {
// Stop does not wait for a write in progress to hand it over.
stopCh chan context.Context
done chan struct{} // closed when the flush goroutine returns
// Peers whose session went down. peerDownMu lets one PeerDown run at a
// time, so the handler's own updates wait for the table's lock behind
// at most one. heldBack, under mu, holds each such peer's later
// messages, in order, until its PeerDown has returned and they are
// applied. A peer with no such messages yet has an empty entry.
// heldCount, under mu, is how many messages heldBack holds.
// heldTaken is signalled each time a held-back message is taken out to
// be applied; its L is &mu. peerDowns counts the peerDown goroutines
// that have not returned.
peerDownMu sync.Mutex
heldBack map[string][]*ristypes.RISMessage
heldCount int
heldTaken sync.Cond
peerDowns sync.WaitGroup
}
// seenPrefix is what the batch keeps of an update: the prefix it names and its
@@ -119,26 +156,34 @@ func (u prefixUpdate) announcesRoute() bool {
func NewPrefixHandler(
db database.Store, logger *logger.Logger, table *routetable.Table,
) *PrefixHandler {
return newPrefixHandler(db, logger, table, prefixBatchTimeout)
return newPrefixHandler(db, logger, table, prefixBatchTimeout,
maxTableChangesPerWrite)
}
// newPrefixHandler is NewPrefixHandler with the flush timer's interval,
// batchTimeout, in place of prefixBatchTimeout. A test that the timer must not
// disturb gives an interval longer than the test runs.
// batchTimeout, in place of prefixBatchTimeout, and tableChangesPerWrite in
// 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(
db database.Store, logger *logger.Logger, table *routetable.Table,
batchTimeout time.Duration,
batchTimeout time.Duration, tableChangesPerWrite int,
) *PrefixHandler {
h := &PrefixHandler{
db: db,
logger: logger,
table: table,
batch: make([]seenPrefix, 0, prefixBatchSize),
batchFullCh: make(chan struct{}, 1),
stopCh: make(chan context.Context, 1),
done: make(chan struct{}),
db: db,
logger: logger,
table: table,
batch: make([]seenPrefix, 0, prefixBatchSize),
batchMaxSize: prefixBatchMaxSize,
tableChangesPerWrite: tableChangesPerWrite,
batchFullCh: make(chan struct{}, 1),
stopCh: make(chan context.Context, 1),
done: make(chan struct{}),
heldBack: make(map[string][]*ristypes.RISMessage),
}
h.batchTaken.L = &h.mu
h.heldTaken.L = &h.mu
// Start the flush timer goroutine
go flushLoop(batchTimeout, h.batchFullCh, h.stopCh, h.done, h.flushBatch)
@@ -166,11 +211,91 @@ func (h *PrefixHandler) QueueCapacity() int {
return prefixHandlerQueueSize
}
// HandleMessage processes a message to track prefix information
// HandleMessage processes a message to track prefix information. A message
// from a peer whose PeerDown has not returned is held back until it has.
func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
h.mu.Lock()
defer h.mu.Unlock()
// While maxHeldBackMessages messages are held back, wait until one of them
// is taken out to be applied. Meanwhile the handler takes nothing off its
// queue, as at batchMaxSize.
for h.heldCount >= maxHeldBackMessages {
h.heldTaken.Wait()
}
held, goingDown := h.heldBack[msg.Peer]
if goingDown {
h.heldBack[msg.Peer] = append(held, msg)
h.heldCount++
return
}
h.applyMessage(msg)
}
// Stop waits for every PeerDown the handler started and for the messages held
// back meanwhile to be applied. It gives up this wait once ctx ends. Then Stop
// stops the flush goroutine and waits for it until ctx ends. The flush
// goroutine makes one last flush before it returns. That flush writes at most
// tableChangesPerWrite of the routing table's changes, and the prefixes of 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
// failed write of the table's changes is not retried either. Stop logs how
// many updates and table changes are not written. It also logs how many
// held-back messages are not applied. The caller calls Stop once its last
// HandleMessage has returned.
func (h *PrefixHandler) Stop(ctx context.Context) {
peerDownsDone := make(chan struct{})
go func() {
h.peerDowns.Wait()
close(peerDownsDone)
}()
select {
case <-peerDownsDone:
case <-ctx.Done():
}
h.stopCh <- ctx
select {
case <-h.done:
case <-ctx.Done():
}
h.mu.Lock()
unwritten := h.writing + len(h.batch)
tableChanges := h.writingTableChanges + h.table.ChangeCount()
heldMessages := h.heldCount
h.mu.Unlock()
if unwritten == 0 && tableChanges == 0 && heldMessages == 0 {
return
}
if ctx.Err() != nil {
h.logger.Error("Stop deadline passed before the prefix handler wrote "+
"its updates", "updates", unwritten, "table_changes", tableChanges,
"held_messages", heldMessages)
return
}
h.logger.Error("The prefix handler stopped before it wrote everything",
"updates", unwritten, "table_changes", tableChanges,
"held_messages", heldMessages)
}
// applyMessage applies msg. The caller holds mu. A peer's session going down
// starts the peer's PeerDown. An update is applied to the routing table, and
// its prefixes join the batch.
func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) {
if msg.Type == ristypes.MessageTypeState {
if msg.State == ristypes.StateDown {
h.table.PeerDown(msg.Peer)
h.startPeerDown(msg.Peer)
}
return
@@ -185,13 +310,10 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
// the path after the rest of it.
path := slices.Concat(msg.Path.Sequence, msg.Path.Set)
h.mu.Lock()
defer h.mu.Unlock()
// 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) >= prefixBatchMaxSize {
for len(h.batch) >= h.batchMaxSize {
h.batchTaken.Wait()
}
@@ -239,7 +361,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
// 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
// or the batch reaches prefixBatchMaxSize.
// or the batch reaches batchMaxSize.
if len(h.batch) >= prefixBatchSize {
select {
case h.batchFullCh <- struct{}{}:
@@ -248,39 +370,58 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
}
}
// 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
// maxTableChangesPerWrite of the routing table's changes, and the prefixes of
// 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
// failed write of the table's changes is not retried either. Stop logs how many
// updates and how many of the table's changes are not written.
func (h *PrefixHandler) Stop(ctx context.Context) {
h.stopCh <- ctx
select {
case <-h.done:
case <-ctx.Done():
// startPeerDown starts peerDown for peer on a new goroutine. The caller holds
// mu. The peer's messages are held back until peerDown has applied them.
// Messages held back for an earlier PeerDown of the peer, and not yet applied,
// stay held back.
func (h *PrefixHandler) startPeerDown(peer string) {
_, goingDown := h.heldBack[peer]
if !goingDown {
h.heldBack[peer] = []*ristypes.RISMessage{}
}
h.mu.Lock()
unwritten := h.writing + len(h.batch)
tableChanges := h.writingTableChanges + h.table.ChangeCount()
h.mu.Unlock()
h.peerDowns.Add(1)
if unwritten == 0 && tableChanges == 0 {
return
go h.peerDown(peer)
}
// peerDown runs the table's PeerDown for peer, once no other PeerDown is
// running. Then peerDown applies the messages held back from the peer, in
// order. It takes mu for one message at a time, so HandleMessage takes new
// messages in between. The peer's new messages join the end of the held-back
// ones. A held-back message that says the peer's session went down again
// starts a new peerDown, which applies the rest.
func (h *PrefixHandler) peerDown(peer string) {
defer h.peerDowns.Done()
h.peerDownMu.Lock()
h.table.PeerDown(peer)
h.peerDownMu.Unlock()
for {
h.mu.Lock()
held := h.heldBack[peer]
if len(held) == 0 {
delete(h.heldBack, peer)
h.mu.Unlock()
return
}
msg := held[0]
held[0] = nil // so that the message is freed once applied
h.heldBack[peer] = held[1:]
h.heldCount--
h.heldTaken.Broadcast()
h.applyMessage(msg)
h.mu.Unlock()
if msg.Type == ristypes.MessageTypeState &&
msg.State == ristypes.StateDown {
return
}
}
if ctx.Err() != nil {
h.logger.Error("Stop deadline passed before the prefix handler wrote "+
"its updates", "updates", unwritten, "table_changes", tableChanges)
return
}
h.logger.Error("The prefix handler stopped before it wrote everything",
"updates", unwritten, "table_changes", tableChanges)
}
// takeBatch returns the pending updates and leaves an empty batch in their
@@ -378,14 +519,13 @@ func (h *PrefixHandler) applyToTable(update prefixUpdate) {
})
}
// writeTableChanges writes at most maxTableChangesPerWrite changes of the
// routing table in memory, and every peer the table numbered since the last
// write.
// writeTableChanges writes at most tableChangesPerWrite changes of the 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
// them again, each pair with its content then.
func (h *PrefixHandler) writeTableChanges() {
h.mu.Lock()
changes, newPeers := h.table.TakeChanges(maxTableChangesPerWrite)
changes, newPeers := h.table.TakeChanges(h.tableChangesPerWrite)
h.writingTableChanges = len(changes)
h.mu.Unlock()
internal/routewatch/prefixhandler_internal_test.go
+87 -6
View File
@@ -237,9 +237,87 @@ func TestPrefixHandlerStopLogsNothingWhenItWroteEverything(t *testing.T) {
}
}
// longerThanAnyTest is a flush timer interval that no test lasts. A test whose
// prefix, AS or peer handler must flush only when its batch is full and on Stop
// gives it to newPrefixHandler, newASHandler or newPeerHandler.
// TestPrefixHandlerStopLogsTheMessagesHeldBack stops a prefix handler with a
// deadline that has passed, while a peer's PeerDown is held up and one message
// of the peer is held back. Stop must log that message as not applied.
func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
t.Parallel()
var logs bytes.Buffer
h := newPrefixHandler(newMockStore(), newErrorLogger(&logs),
routetable.New(routetable.Params{}), longerThanAnyTest,
maxTableChangesPerWrite)
h.peerDownMu.Lock()
defer h.peerDownMu.Unlock()
h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
h.HandleMessage(sessionDown())
h.HandleMessage(peerAnnouncement(testPeerIP, testV6Prefix))
ctx, cancel := context.WithCancel(t.Context())
cancel()
h.Stop(ctx)
logged := lastLogged(t, &logs)
if logged.HeldMessages != 1 {
t.Errorf("Stop logged %d held messages not applied, want 1",
logged.HeldMessages)
}
}
// TestPrefixHandlerStopWaitsForTheMessagesHeldBack stops a prefix handler
// while a peer's PeerDown is held up and an announcement of the peer is held
// back. Stop must wait. Once the PeerDown is let go and Stop returns, the
// announcement must be in the database.
func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) {
t.Parallel()
db := newTestDatabase(t)
h := newPrefixHandler(db, logger.New(), routetable.New(routetable.Params{}),
longerThanAnyTest, maxTableChangesPerWrite)
h.peerDownMu.Lock()
h.HandleMessage(sessionDown())
h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
stopped := make(chan struct{})
go func() {
h.Stop(t.Context())
close(stopped)
}()
select {
case <-stopped:
t.Error("Stop returned while a PeerDown was held up")
case <-time.After(stillWaitingAfter):
}
h.peerDownMu.Unlock()
select {
case <-stopped:
case <-time.After(stepTimeout):
t.Fatalf("Stop did not return within %s of the PeerDown's release",
stepTimeout)
}
read, err := db.ReadRouteTable(t.Context())
if err != nil {
t.Fatalf("ReadRouteTable: %v", err)
}
checkTableOrigins(t, read, map[string]map[int][]string{
testV4Prefix: {testASNB: {testPeerIP}},
})
}
// longerThanAnyTest is a timer interval that no test lasts. A test gives it to
// newPrefixHandler, newASHandler, newPeerHandler or newPeeringHandler when the
// handler must write only when its batch is full, when the test says, or on
// Stop.
const longerThanAnyTest = time.Hour
// prefixHandlerWithoutTimer returns a prefix handler that writes to store and
@@ -247,9 +325,12 @@ const longerThanAnyTest = time.Hour
// NewPrefixHandler's flush goroutine, so the test flushes itself.
func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
return &PrefixHandler{
db: store,
logger: logger.New(),
table: routetable.New(routetable.Params{}),
db: store,
logger: logger.New(),
table: routetable.New(routetable.Params{}),
batchMaxSize: prefixBatchMaxSize,
tableChangesPerWrite: maxTableChangesPerWrite,
heldBack: make(map[string][]*ristypes.RISMessage),
}
}
internal/routewatch/prefixhandler_table_internal_test.go
+270 -23
View File
@@ -7,8 +7,10 @@ import (
"log/slog"
"net/netip"
"reflect"
"runtime"
"slices"
"strings"
"sync"
"testing"
"time"
@@ -30,11 +32,12 @@ const (
// firstManyPrefix is the first prefix announceManyPrefixes announces.
firstManyPrefix = "10.0.0.0/32"
)
// 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
// testMaxTableChangesPerWrite is a limit of one write of the routing
// table's changes that a test exceeds quickly, in place of
// maxTableChangesPerWrite.
testMaxTableChangesPerWrite = 10
)
// 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
@@ -100,6 +103,228 @@ func TestPrefixHandlerAppliesUpdatesToTheRoutingTable(t *testing.T) {
})
}
// TestPrefixHandlerTakesMessagesWhileAPeerGoesDown holds up the PeerDown of a
// peer whose session went down, as another peer's long PeerDown would. The
// handler must go on taking messages. It applies another peer's announcement
// at once. It holds back these messages of the down peer, in this order:
// - an announcement;
// - its session going down again;
// - a second announcement.
//
// Once the PeerDowns have returned, the down peer must carry only the prefix
// of the second announcement.
func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
t.Parallel()
h := prefixHandlerWithoutTimer(newMockStore())
h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
h.peerDownMu.Lock()
handled := returnsInTime(func() {
for _, msg := range []*ristypes.RISMessage{
sessionDown(),
peerAnnouncement(secondPeerIP, otherV4Prefix),
peerAnnouncement(testPeerIP, testV6Prefix),
sessionDown(),
peerAnnouncement(testPeerIP, setV4Prefix),
} {
h.HandleMessage(msg)
}
})
if !handled {
t.Fatalf("the handler did not take the messages after the session "+
"went down within %s", stepTimeout)
}
checkTableOrigins(t, h.table, map[string]map[int][]string{
testV4Prefix: {testASNB: {testPeerIP}},
otherV4Prefix: {testASNB: {secondPeerIP}},
testV6Prefix: {},
setV4Prefix: {},
})
h.peerDownMu.Unlock()
if !returnsInTime(h.peerDowns.Wait) {
t.Fatalf("the PeerDowns did not return within %s", stepTimeout)
}
checkTableOrigins(t, h.table, map[string]map[int][]string{
testV4Prefix: {},
otherV4Prefix: {testASNB: {secondPeerIP}},
testV6Prefix: {},
setV4Prefix: {testASNB: {testPeerIP}},
})
}
// TestPrefixHandlerAppliesMessagesWhilePeerDownRuns gives a peer 50,000
// prefixes and hands the prefix handler that peer's session going down. The
// table's PeerDown then visits the prefixes in five batches. Once PeerDown has
// begun, the handler is handed 1,000 announcements of another peer. The
// handler must apply them all before PeerDown ends, while the down peer still
// carries some of its prefixes.
func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) {
t.Parallel()
const (
// downPeerPrefixes is five times the 10,000 prefixes PeerDown visits
// in one hold of the table's lock.
downPeerPrefixes = 50000
// otherPeerMessages is far more messages than the handler applies
// while PeerDown runs, if PeerDown takes the lock back at once after
// each batch. The handler then applies only a few for each batch.
otherPeerMessages = 1000
// firstOtherAddr is the address of the other peer's first prefix.
firstOtherAddr = "2001:db8:2::"
)
h := prefixHandlerWithoutTimer(newMockStore())
announceManyPrefixes(h.table, downPeerPrefixes)
h.HandleMessage(sessionDown())
// Wait until PeerDown has begun: the down peer has left some of its
// prefixes. The down peer's prefixes are IPv4 and the other peer's are
// IPv6, so the counts tell them apart.
for h.table.Counts().IPv4Routes == downPeerPrefixes {
runtime.Gosched()
}
// Each announcement names the /128 of the next address.
addr := netip.MustParseAddr(firstOtherAddr)
for range otherPeerMessages {
h.HandleMessage(peerAnnouncement(secondPeerIP,
netip.PrefixFrom(addr, addr.BitLen()).String()))
addr = addr.Next()
}
counts := h.table.Counts()
if counts.IPv6Routes != otherPeerMessages {
t.Errorf("the other peer carries %d prefixes, want %d",
counts.IPv6Routes, otherPeerMessages)
}
if counts.IPv4Routes == 0 {
t.Errorf("PeerDown ended before the handler applied the other peer's "+
"%d announcements", otherPeerMessages)
}
if !returnsInTime(h.peerDowns.Wait) {
t.Fatalf("the PeerDown did not return within %s", stepTimeout)
}
}
// TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes holds back many
// messages of a peer whose session went down. Each names a prefix that does
// not parse, so applying it logs an error. Each log write takes a
// millisecond. While the handler applies the held-back messages, it must take
// another peer's announcement.
func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) {
t.Parallel()
const (
// heldMessages take the handler at least a second to apply.
heldMessages = 1000
badPrefix = "not a prefix"
)
logs := &slowWriter{firstWrite: make(chan struct{})}
h := newPrefixHandler(newMockStore(), newErrorLogger(logs),
routetable.New(routetable.Params{}), longerThanAnyTest,
maxTableChangesPerWrite)
defer stopInTime(t, h)
h.peerDownMu.Lock()
h.HandleMessage(sessionDown())
for range heldMessages {
h.HandleMessage(peerAnnouncement(testPeerIP, badPrefix))
}
h.peerDownMu.Unlock()
select {
case <-logs.firstWrite:
case <-time.After(stepTimeout):
t.Fatal("the handler did not start to apply the held-back messages")
}
if !returnsInTime(func() {
h.HandleMessage(peerAnnouncement(secondPeerIP, otherV4Prefix))
}) {
t.Fatalf("HandleMessage did not return within %s", stepTimeout)
}
h.mu.Lock()
left := len(h.heldBack[testPeerIP])
h.mu.Unlock()
if left == 0 {
t.Error("the handler took another peer's announcement only once it " +
"had applied every held-back message")
}
}
// TestPrefixHandlerWaitsAtTheMaximumHeldBack holds up the PeerDown of a peer
// whose session went down. It hands the handler maxHeldBackMessages messages
// of that peer. HandleMessage must then wait, even for another peer's
// message, until the PeerDown returns and a held-back message is applied.
func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
t.Parallel()
h := newPrefixHandler(newMockStore(), logger.New(),
routetable.New(routetable.Params{}), longerThanAnyTest,
maxTableChangesPerWrite)
defer stopInTime(t, h)
h.peerDownMu.Lock()
h.HandleMessage(sessionDown())
held := peerAnnouncement(testPeerIP, testV6Prefix)
for range maxHeldBackMessages {
h.HandleMessage(held)
}
returned := make(chan struct{})
go func() {
h.HandleMessage(peerAnnouncement(secondPeerIP, otherV4Prefix))
close(returned)
}()
select {
case <-returned:
t.Error("HandleMessage returned while maxHeldBackMessages messages " +
"were held back")
case <-time.After(stillWaitingAfter):
}
h.peerDownMu.Unlock()
select {
case <-returned:
case <-time.After(stepTimeout):
t.Fatal("HandleMessage did not return after the PeerDown returned")
}
}
// slowWriter takes a millisecond over each write and drops what it is given.
// It closes firstWrite at the first write.
type slowWriter struct {
firstWrite chan struct{}
once sync.Once
}
func (w *slowWriter) Write(p []byte) (int, error) {
w.once.Do(func() { close(w.firstWrite) })
time.Sleep(time.Millisecond)
return len(p), nil
}
// TestPrefixHandlerAppliesAPathWithoutAnOriginToTheRoutingTable hands the
// prefix handler announcements whose AS path gives no origin. One whose path
// is empty or holds AS 0 removes the peer's route, as a withdrawal does; one
@@ -198,8 +423,9 @@ func TestPrefixHandlerKeepsThePathOfAPathEndingInASet(t *testing.T) {
}
}
// checkTableCases hands the messages of each case to a new prefix handler and
// checks its routing table right after, with no flush.
// checkTableCases hands the messages of each case to a new prefix handler. It
// checks the handler's routing table once the handler's PeerDowns have
// returned, with no flush.
func checkTableCases(t *testing.T, tests []tableCase) {
t.Helper()
@@ -212,17 +438,28 @@ func checkTableCases(t *testing.T, tests []tableCase) {
h.HandleMessage(msg)
}
for prefix, want := range tt.want {
got := tableOrigins(h.table, prefix)
if !reflect.DeepEqual(got, want) {
t.Errorf("peers of %s by origin AS = %v, want %v",
prefix, got, want)
}
}
h.peerDowns.Wait()
checkTableOrigins(t, h.table, tt.want)
})
}
}
// checkTableOrigins checks, for each prefix in want, the peers that carry it
// in table with each origin AS.
func checkTableOrigins(
t *testing.T, table *routetable.Table, want map[string]map[int][]string,
) {
t.Helper()
for prefix, wantOrigins := range want {
got := tableOrigins(table, prefix)
if !reflect.DeepEqual(got, wantOrigins) {
t.Errorf("peers of %s by origin AS = %v, want %v",
prefix, got, wantOrigins)
}
}
}
// tableWritesStore is a real database that hands each write of the routing
// table's changes to written once it is made.
type tableWritesStore struct {
@@ -276,9 +513,9 @@ func TestPrefixHandlerWritesTheTableChangesEachSecond(t *testing.T) {
t.Fatalf("wrote %+v, want the pair of %s and AS%d with "+
"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",
tableWriteTimeout)
stepTimeout)
}
}
}
@@ -375,25 +612,32 @@ func (s *tableWriteSizesStore) WriteRouteChanges(
// 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
// 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) {
t.Parallel()
const (
pairs = maxTableChangesPerWrite * 3 / 2
pairs = testMaxTableChangesPerWrite * 3 / 2
flushes = 3
)
db := newTestDatabase(t)
store := &tableWriteSizesStore{Database: db}
h := prefixHandlerWithoutTimer(store)
announceManyPrefixes(h.table, pairs)
table := routetable.New(routetable.Params{})
announceManyPrefixes(table, pairs)
h := newPrefixHandler(store, logger.New(), table, longerThanAnyTest,
testMaxTableChangesPerWrite)
defer stopInTime(t, h)
for range flushes {
h.flushBatch()
}
wantChanges := []int{maxTableChangesPerWrite, pairs - maxTableChangesPerWrite}
wantChanges := []int{
testMaxTableChangesPerWrite, pairs - testMaxTableChangesPerWrite,
}
wantNewPeers := []int{1, 0}
if !slices.Equal(store.changes, wantChanges) ||
@@ -415,11 +659,11 @@ func TestPrefixHandlerWritesManyTableChangesOverSeveralFlushes(t *testing.T) {
// 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
// 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) {
t.Parallel()
const pairs = maxTableChangesPerWrite + 1
const pairs = testMaxTableChangesPerWrite + 1
var logs bytes.Buffer
@@ -427,7 +671,8 @@ func TestPrefixHandlerStopLogsTheTableChangesLeftUnwritten(t *testing.T) {
table := routetable.New(routetable.Params{})
announceManyPrefixes(table, pairs)
h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest)
h := newPrefixHandler(store, newErrorLogger(&logs), table, longerThanAnyTest,
testMaxTableChangesPerWrite)
h.Stop(t.Context())
written := 0
@@ -489,6 +734,7 @@ func TestARestartReadsTheRoutingTableBack(t *testing.T) {
h.HandleMessage(msg)
}
h.peerDowns.Wait()
h.flushBatch()
var logs bytes.Buffer
@@ -527,6 +773,7 @@ func TestARestartReadsTheRoutingTableBack(t *testing.T) {
restarted.HandleMessage(msg)
}
h.peerDowns.Wait()
restarted.Stop(t.Context())
read, err := db.ReadRouteTable(t.Context())
internal/server/home_internal_test.go
+61
View File
@@ -6,6 +6,7 @@ import (
"strconv"
"strings"
"testing"
"testing/synctest"
"time"
"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.
func waitUntil(t *testing.T, maxWait time.Duration, done func() bool) {
t.Helper()
internal/server/server.go
+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
// until the server first works them out
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
@@ -111,7 +117,8 @@ func New(
// 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
// 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 {
const (
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() {
defer close(s.cardsStopped)
s.keepHomeCardsFresh(ctx, homeCardsRefreshInterval)
}()
return nil
}
// Stop stops the HTTP server. It waits, until ctx ends, for the requests in
// progress to finish, and then closes the connections of those still running
// and returns ctx's error.
// Stop stops the goroutine that works out the home page's cards, and waits
// until it has returned or ctx ends. It then stops the HTTP server. It waits,
// 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 {
if s.srv == nil {
return nil
@@ -165,6 +178,16 @@ func (s *Server) Stop(ctx context.Context) error {
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)
if err != nil {
return errors.Join(err, s.srv.Close())
internal/whois/whois.go
+82 -36
View File
@@ -12,6 +12,7 @@ import (
"regexp"
"strings"
"sync"
"syscall"
"time"
"unicode/utf8"
@@ -70,12 +71,12 @@ const nicbrHeader = "% Copyright (c) Nic.br"
// queries to stop the data being taken in bulk.
const nicbrLimitReached = "Query rate limit exceeded"
// NICbrSkipTime is how long the client stops asking NIC.br after NIC.br
// answers that the client is over its query limit. NIC.br publishes
// neither the limit nor how long it lasts. An hour outlasts a limit
// counted per minute or per hour. The ASes looked up meanwhile get
// NIC.br's fields when they are next refreshed.
const NICbrSkipTime = time.Hour
// PauseTime is how long the client sends an RIR's or NIC.br's server no
// query after that server refuses a query. queryRegistry says what a
// refusal is. RIPE refuses an address for good when the address keeps
// passing RIPE's limit (see noPersonalFlag). An hour outlasts a limit
// counted per minute or per hour.
const PauseTime = time.Hour
// RIR identifiers.
const (
@@ -125,6 +126,10 @@ var (
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.
type ASNInfo struct {
ASN int
@@ -156,12 +161,12 @@ type Client struct {
logger *slog.Logger
// mu guards nicbrSkipUntil.
// mu guards pausedUntil.
mu sync.Mutex
// nicbrSkipUntil is when the client next asks NIC.br, after NIC.br
// answered that the client is over its query limit.
nicbrSkipUntil time.Time
// pausedUntil maps a server, given as host:port, to the time the client
// may ask it again.
pausedUntil map[string]time.Time
}
// NewClient creates a new WHOIS client that logs to logger.
@@ -172,23 +177,29 @@ func NewClient(logger *slog.Logger) *Client {
DialContext: dialer.DialContext,
Now: time.Now,
logger: logger,
pausedUntil: make(map[string]time.Time),
}
}
// 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 NIC.br's own server says instead, unless that query fails or
// is skipped, as askNICbr says.
// NIC.br is paused, as askNICbr says.
//
// It returns ErrNoReferral when IANA refers the AS to no RIR, and another
// error when the query to IANA or to the RIR fails, when the RIR refuses
// 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
// answer carries none of ASNInfo's fields.
// It returns ErrNoReferral when IANA refers the AS to no RIR. Its error
// wraps ErrPaused when the RIR is paused. It returns another error in each
// of these cases:
//
// - the query to IANA or to the RIR fails
// - 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) {
// Query IANA first to find the authoritative RIR
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)
if err != nil {
return nil, fmt.Errorf("IANA query failed: %w", err)
@@ -211,7 +222,7 @@ func (c *Client) LookupASN(ctx context.Context, asn int) (*ASNInfo, error) {
}
// Query the authoritative RIR
rirResp, err := c.queryRIR(ctx, whoisServer, rirQuery(rir, query))
rirResp, err := c.queryRegistry(ctx, whoisServer, rirQuery(rir, query))
if err != nil {
return nil, fmt.Errorf("RIR query failed: %w", err)
}
@@ -241,22 +252,19 @@ func (c *Client) LookupASN(ctx context.Context, asn int) (*ASNInfo, error) {
// 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
// client over its limit, or when NIC.br's answer carries none of ASNInfo's
// fields. After the reduced answer, it returns lacnicInfo without asking
// NIC.br until NICbrSkipTime has passed.
// fields. The reduced answer pauses NIC.br, as a refusal does. While NIC.br
// is paused, askNICbr returns lacnicInfo without a warning. The ASes looked
// up meanwhile get NIC.br's fields when they are next refreshed.
func (c *Client) askNICbr(
ctx context.Context, asQuery string, lacnicInfo *ASNInfo,
) *ASNInfo {
asn := lacnicInfo.ASN
c.mu.Lock()
skipping := c.Now().Before(c.nicbrSkipUntil)
c.mu.Unlock()
if skipping {
answer, err := c.queryRegistry(ctx, whoisServerNICbr, asQuery)
if errors.Is(err, ErrPaused) {
return lacnicInfo
}
answer, err := c.query(ctx, whoisServerNICbr, asQuery)
if err != nil {
c.logger.Warn("NIC.br query failed, keeping LACNIC's fields",
"asn", asn, "error", err)
@@ -265,13 +273,11 @@ func (c *Client) askNICbr(
}
if strings.Contains(answer, nicbrLimitReached) {
c.mu.Lock()
c.nicbrSkipUntil = c.Now().Add(NICbrSkipTime)
c.mu.Unlock()
c.pause(whoisServerNICbr)
c.logger.Warn("NIC.br answered that the client is over its query "+
"limit, keeping LACNIC's fields and skipping NIC.br",
"asn", asn, "skip_for", NICbrSkipTime)
"limit, keeping LACNIC's fields and pausing NIC.br",
"asn", asn, "pause_for", PauseTime)
return lacnicInfo
}
@@ -309,25 +315,65 @@ func rirQuery(rir, asQuery string) string {
}
}
// queryRIR sends query to an RIR's server and returns the answer. An answer
// holding an error line with which the server refuses the query is an
// error, and the error holds that line.
func (c *Client) queryRIR(
// queryRegistry sends query to the server of an RIR or of NIC.br, and
// returns the answer. The client pauses the server for PauseTime when the
// server refuses the query. Each of these is a refusal:
//
// - 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,
) (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)
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 {
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") {
if strings.HasPrefix(line, accessDeniedError) ||
strings.HasPrefix(line, accessLimitError) {
return "", fmt.Errorf("%w: %s", errRefused, line)
return fmt.Errorf("%w: %s", errRefused, line)
}
}
return answer, nil
return 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
internal/whois/whois_test.go
+142 -5
View File
@@ -718,9 +718,10 @@ func TestLookupASNKeepsLACNICsAnswerWhenNICbrFails(t *testing.T) {
// TestLookupASNSkipsNICbrAfterItsLimit looks up an AS that NIC.br
// registers three times with one client, while NIC.br answers that the
// client is over its query limit. The first lookup asks NIC.br. The
// second, a second before NICbrSkipTime has passed, does not. The third,
// once NICbrSkipTime has passed, asks again. Each keeps LACNIC's fields.
// client is over its query limit. The test sets the client's clock. The
// first lookup asks NIC.br. The next comes a second before PauseTime has
// passed, and does not ask NIC.br. The last comes at PauseTime and asks
// NIC.br again. Each keeps LACNIC's fields.
func TestLookupASNSkipsNICbrAfterItsLimit(t *testing.T) {
t.Parallel()
@@ -740,8 +741,8 @@ func TestLookupASNSkipsNICbrAfterItsLimit(t *testing.T) {
wantAsks int
}{
{after: 0, wantAsks: 1},
{after: whois.NICbrSkipTime - time.Second, wantAsks: 1},
{after: whois.NICbrSkipTime, wantAsks: 2},
{after: whois.PauseTime - time.Second, wantAsks: 1},
{after: whois.PauseTime, wantAsks: 2},
} {
now = start.Add(step.after)
@@ -785,6 +786,142 @@ 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
// RIPE when RIPE's server refuses the connection, and when it answers only
// with an error. Neither is a lookup that found the AS.