Compare commits

...
3 Commits
Author SHA1 Message Date
clawbot 1dea2210ef Rewrite TODO.md sentence by sentence (part of #157)
check / check (push) Canceled after 0s
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 22:12:27 +00:00
clawbot e5afd94a39 Take every down peer out of the table in one walk (closes #225)
check / check (push) Canceled after 0s
The prefix handler ran one PeerDown per peer, one after another. When
the sessions of several peers went down together, their walks queued up.
A returning full-table peer's messages stayed held back while the walks
ahead of its own ran, and could reach the limit of 20,000.

The table's PeersDown now takes several peers out of every prefix in one
walk. One walk runs at a time. The next one takes every peer whose
session went down since the previous one began. The doc of
maxHeldBackMessages says how long a peer's messages are held back. The
limit is now 150,000 messages, about 150 MiB. The status page shows how
many messages each handler holds back.

Model: opus-5-5
2026-11-02 23:10:09 +01:00
clawbot 9bd0e91a3b Let the lock go for as long as it was held during Overview's walk (closes #226)
check / check (push) Canceled after 0s
The home page's cards are worked out once a minute by Overview, which
walks every prefix of the routing table under the table's read lock.
After each batch of 10,000 prefixes, the walk let the lock go and took
it back at once. The prefix handler, waiting for the write lock, then got
in only once per batch. During a walk, its queue could fill, and the
streamer then dropped updates for it.

The walk now lets the lock go for as long as it held the lock for the
batch, as PeerDown does. A test applies another peer's 1,000
announcements while Overview walks a table with more prefixes than five
of the walk's batches.

Model: opus-5-5
2026-11-02 08:00:00 +01:00
17 changed files with 842 additions and 404 deletions

No files matched your search

+8 -6
View File
@@ -445,8 +445,9 @@ the routing table, and everything else from the database.
- A withdrawal removes the peer's route. So does an announcement whose - A withdrawal removes the peer's route. So does an announcement whose
AS path is empty or holds AS 0 anywhere. AS path is empty or holds AS 0 anywhere.
- A peer whose BGP session goes down leaves every prefix. The handler - 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 holds back the peer's later updates. It applies them in order after
prefix, then applies them in order. the peer has left every prefix and after the held-back updates of
every peer whose session went down before it.
Every second it writes the changed prefix and origin AS pairs to the 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 `routes` table. A write takes at most 20,000 pairs, and the rest follow
@@ -651,10 +652,11 @@ What happens at each limit:
at most about 0.1 GiB of the Go heap. at most about 0.1 GiB of the Go heap.
- Held-back messages: the prefix handler holds back a peer's messages after the - 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. peer's session goes down. It applies them after the peer has left every prefix
It holds at most 20,000 such messages (`maxHeldBackMessages`), about 20 MiB. and after the held-back messages of every peer whose session went down before
At that number it takes nothing more off its queue until it has applied one of it. It holds at most 150,000 such messages (`maxHeldBackMessages`), about 150
them. 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 `DEBUG=routewatch` the daemon logs a `System stats` line every 60 seconds
with the goroutine count and Go memory figures. with the goroutine count and Go memory figures.
+345 -279
View File
@@ -10,171 +10,223 @@
# Status # Status
pre-1.0. No git tags. `main` last took `next` on 2026-10-06, through RouteWatch is pre-1.0, with no git tags. `main` last took `next` on 2026-10-27,
https://git.eeqj.de/sneak/routewatch/pulls/6. The linter and the tests are through https://git.eeqj.de/sneak/routewatch/pulls/166.
phases of the Docker build, and the Gitea workflow runs `script/cibuild` (the https://git.eeqj.de/sneak/routewatch/pulls/202 carries the commits since then
format check of the Go code and the markdown, both phases, then the image) on into `main`. It is labelled `merge-ready` and waits for sneak.
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 The linter and the tests are phases of the Docker build. The Gitea workflow runs
upaas"). A 25-hour run of `a9b6860` on the live feed peaked at about 0.6 GiB `script/cibuild` on every push. That script runs the two phases and checks the
(https://git.eeqj.de/sneak/routewatch/issues/3). A run of `825afa2` on the live formatting of the Go code and the markdown. Then it builds the image.
feed found the prefix handler dropping about a quarter of its updates because
its route writes were slower than the feed The image sets memory ceilings for a 5 GiB container (README "Memory"). README
(https://git.eeqj.de/sneak/routewatch/issues/96); the cheaper writes since then "Running under upaas" says how to run it.
are measured by a benchmark, not yet on the live feed.
A five-day run of `1c65656` on the live feed, from 2026-10-27 to 2026-11-01,
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 run of `26d939a` on the live feed showed the prefix handler still dropping
updates while the home page's cards were worked out.
https://git.eeqj.de/sneak/routewatch/issues/226 fixed that on `next`.
https://git.eeqj.de/sneak/routewatch/issues/225, also on `next`, takes the peers
whose sessions go down together out of the table in one walk. A 24-hour run of
`9bd0e91` on the live feed is checking whether any handler still drops updates.
The README rewrite, https://git.eeqj.de/sneak/routewatch/pulls/210, is being
reworked. 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 # Next Step
Run `next` on the live feed, as README "Running under upaas" describes, until Post the outcome of the 24-hour run of `9bd0e91` on
the routing table stops growing. Post its size and memory figures on https://git.eeqj.de/sneak/routewatch/issues/226: whether any handler still drops
https://git.eeqj.de/sneak/routewatch/issues/122. The same run makes the final updates, and the process's peak memory against 5 GiB. File an issue for each
check https://git.eeqj.de/sneak/routewatch/issues/96 names: whether the prefix problem the run finds.
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.
# Completed Steps # Completed Steps
- 2026-11-02: like a failed WHOIS lookup, a lookup skipped because its RIR is - 2026-11-02: `TODO.md` is rewritten from scratch, sentence by sentence, against
paused makes the fetcher wait longer before the next lookup. The longest wait the rules of https://git.eeqj.de/sneak/routewatch/issues/157. Every Completed
is one minute. The status page does not count a skip as a failure. A skip used Steps entry keeps its date, its facts and its issue. Status and Next Step
to leave the wait as it was. Each lookup asks IANA before it finds its RIR describe the repo as the tracker shows it (part of #157)
paused. So while most stale ASes belonged to a paused RIR, the fetcher could - 2026-11-02: when the sessions of several peers go down, the prefix handler
ask IANA every second (closes #222) takes the peers out of the routing table together. The table's `PeersDown`
- 2026-11-02: a peer's session going down no longer stops the prefix handler. takes several peers out of every prefix in one walk. One walk runs at a time.
The handler runs the table's `PeerDown` on a separate goroutine, one peer at a The next one takes every peer whose session went down since the previous one
time, and goes on taking messages. `PeerDown` holds the table's lock for began. Each peer used to have a walk of its own. A peer's messages stayed held
10,000 prefixes at a time, then lets the lock go for as long again. The back while the walks queued ahead of its own ran. The handler holds back at
handler holds back the down peer's later messages until that peer's `PeerDown` most 150,000 messages, about 150 MiB. It used to hold back at most 20,000. The
returns. Then the handler applies them in order, one at a time, and takes new status page shows how many messages the handler holds back (closes #225)
messages in between. It holds back at most 20,000 messages. At that number, - 2026-11-02: working out the home page's cards no longer stops the prefix
`HandleMessage` waits until one of them is applied. When `Stop` runs past its handler. The walk over the routing table holds the table's lock for 10,000
deadline, it logs how many held-back messages were not applied (closes #216) prefixes at a time, then lets the lock go for as long again, as `PeerDown`
does. It used to take the lock back at once. The prefix handler then applied
about one update per 10,000 prefixes walked. During a walk, its queue could
fill, and the streamer then dropped updates for it (closes #226)
- 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 - 2026-11-02: RouteWatch pauses an RIR's or NIC.br's WHOIS server for an hour
after the server refuses a query. A refused or reset connection is a refusal. after the server refuses a query. NIC.br is Brazil's registry. A refused or
So is an answer with an `%ERROR:201:` or `%ERROR:202:` line. Lookups at the reset connection is a refusal. So is an answer with an `%ERROR:201:` or
other servers go on. An AS whose RIR is paused is looked up after the pause. `%ERROR:202:` line. Lookups at the other servers go on. An AS whose RIR is
IANA is never paused, because a pause of IANA would stop every lookup. A paused is looked up after the pause. While NIC.br is paused, a Brazilian AS
refusal by IANA is an ordinary failure. NIC.br is also paused for an hour gets the details from LACNIC's answer. IANA is never paused, because a pause
after it says RouteWatch has passed NIC.br's query limit. Before, the fetcher of IANA would stop every lookup. A refusal by IANA is an ordinary failure.
asked a server again within seconds of a refusal (closes #218) NIC.br is also paused for an hour after it says RouteWatch has passed its
- 2026-11-02: the AS handler never waits for a database write. Its batch holds 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 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 batch no longer grows with the feed's rate. It used to hold one entry for each
AS of each path and to wait once it held 300,000. In a burst of the feed its AS of each path. At 300,000 entries it waited for the write in progress. In a
queue then filled, and the streamer dropped messages for it (closes #217) burst of the feed its queue then filled, and the streamer dropped messages for
- 2026-10-27: the tests of `internal/routewatch` no longer fail on a loaded 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 host. Five of them gave `HandleMessage` two seconds to apply a full prefix
batch, which under the race detector can take longer. A test now waits up to batch. Under the race detector that can take longer. A test now waits up to 15
15 seconds for each step it expects. The tests of the handlers' limits give seconds for each step it expects. The tests of the handlers' limits now give
the handlers small ones, so the package no longer runs past its 90-second the handlers small limits. So the package no longer runs past its 90-second
limit (closes #213, #215) limit (closes #213, #215)
- 2026-10-27: the server works out the home page's cards every minute until it - 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 stops. It used to work them out only once, at startup. The end of the context
that `Start` is given stopped the refreshes (closes #211) given to `Start` stopped the refreshes (closes #211)
- 2026-10-27: README "Rationale" says why RouteWatch exists: to answer our other - 2026-10-27: README "Rationale" says why RouteWatch exists. RouteWatch answers
services' questions about IP addresses, from its own routing table and the our other services' questions about IP addresses, from its own routing table
WHOIS details it keeps. It still says what RIS Live does not give (closes and the WHOIS details it keeps. The section still says what RIS Live does not
#204) give (closes #204)
- 2026-10-27: every JSON answer carries `@meta`. A panic on a JSON route answers - 2026-10-27: Every JSON answer carries `@meta`. A panic on a JSON route answers
500 with the JSON error body. `/status.json` and the health check answer 405 500 with the JSON error body. `/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 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 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 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 the message of an error answer. A test checks `@meta` on every JSON route, in
its successes and its errors (closes #201) its successes and its errors (closes #201)
- 2026-10-27: every value in a JSON answer that has a unit is a number, with the - 2026-10-27: Every value with a unit in a JSON answer is a number. The unit is
unit at the end of its field name. `@meta.execution_time` is now at the end of the field name. `@meta.execution_time` is now
`execution_time_ms`. The stats' `uptime`, `connection_duration` and `execution_time_ms`. In the stats, `uptime` is now `uptime_seconds`,
`memory_usage` are now `uptime_seconds`, `connection_duration_seconds` and `connection_duration` is now `connection_duration_seconds`, and `memory_usage`
`memory_usage_bytes`. `whois_stats.current_interval` is now is now `memory_usage_bytes`. `whois_stats.current_interval` is now
`current_interval_seconds`. They used to be strings such as "12 ms", `whois_stats.current_interval_seconds`. They used to be strings such as "12
"1h18m34s" or "1.1 GB". `connection_duration_seconds` is null while the feed ms", "1h18m34s" or "1.1 GB". `connection_duration_seconds` is null while the
is not connected. `API_POLICIES.md` is vendored from `sneak/prompts`, and feed is not connected. `API_POLICIES.md` is vendored from `sneak/prompts`, and
`REPO_POLICIES.md` re-vendored (closes #199) `REPO_POLICIES.md` is re-vendored (closes #199)
- 2026-10-27: every result of the IP lookup has the same fields. A field without - 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 a value is null. It used to be left out. An address with no route in the
every route and AS field set to null, and `ptr` is an empty list when there routing table has every route and AS field set to null. `ptr` is an empty list
are no PTR names. New fields give the number of origin ASes of the netblock, when there are no PTR names. The new field `num_origins` gives the number of
in `num_origins`, and which they are, in `origin_asns` (closes #200) origin ASes of the netblock, and `origin_asns` gives which they are (closes
- 2026-10-27: with `TRUSTED_PROXIES` unset, the server trusts reverse proxies at #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 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 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, 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) `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 - 2026-10-27: The home page has cards with figures of the routing table. One
newest and oldest kept route, the newest and oldest updated AS, the AS with card shows the newest and the oldest kept route. One shows the newest and the
the most prefixes, the ASes with a single prefix, and fifteen well-known oldest updated AS. One shows the AS with the most prefixes. One shows the ASes
networks. Two cards count the bogon announcements and the prefixes with more with a single prefix. One shows fifteen well-known networks. Two cards count
than one origin AS, and link to the new pages `/bogons` and the bogon announcements and the prefixes with more than one origin AS. They
`/multiple-origins` that list them. A new package, `internal/bogon`, tells link to the new pages `/bogons` and `/multiple-origins`, which list them. A
which prefixes and AS numbers are reserved. The server works the figures out new package, `internal/bogon`, tells which prefixes and AS numbers are
once a minute (closes #154) 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, - 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 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 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 are set each new prefix with its first-seen time. The image's memory limits fit a full
for a full routing table of about 1 GB: `GOMEMLIMIT` is 2.5 GiB, the writing routing table of about 1 GB. `GOMEMLIMIT` is 2.5 GiB. The writing connection's
connection's page cache 512 MiB, and SQLite's heap limits 768 MiB and 1 GiB page cache is 512 MiB. SQLite's soft and hard heap limits are 768 MiB and 1
(closes #122) GiB (closes #122)
- 2026-10-27: the pages, the API and the stats read every route from the routing - 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 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 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, prefix page and API list every peer that carries the prefix. Only the kept
next hop and time only for the kept routes. The stats count the origins whose routes show an AS path, a next hop and a time. The stats count the origins
kept routes were all withdrawn, in `origins_without_kept_route` (part of #122) whose kept routes were all withdrawn, in `origins_without_kept_route` (part of
- 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
#122) #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 `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, 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 Another reads the whole table back into memory at startup. Nothing uses them
them yet (part of #122) yet (part of #122)
- 2026-10-27: a new package, `internal/routetable`, holds the whole routing - 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 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 it, one bit per peer. It also keeps the two newest of their routes. Nothing
(part of #122) uses it yet (part of #122)
- 2026-10-27: each result of the IP lookup lists the public blocklists the - 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 address is on, in `blocklists`. The lists are DShield, Feodo Tracker, two
FireHOL lists, Spamhaus DROP and the Tor exit list. RouteWatch downloads each FireHOL lists, two Spamhaus DROP lists and the Tor exit list. RouteWatch
list on its own schedule and answers from copies in memory. The copies are downloads each list on its own schedule and answers from copies in memory. The
also kept in the state directory's `blocklists` directory. IP lookup answers copies are also kept in the state directory's `blocklists` directory. IP
carry `Cache-Control`. An address with no live route now has a result too, lookup answers carry `Cache-Control`. An address with no route now has a
without route fields, so the IP lookup no longer answers 404 (closes #158) result too, without route fields. So the IP lookup no longer answers 404
- 2026-10-27: the home page has an IP information API card. It shows the `curl` (closes #158)
command that looks up the visitor's own address at `/ip/{addr}`, and that - 2026-10-27: The home page has an IP information API card. It shows the `curl`
command's live answer (closes #156) command that looks up the visitor's own address at `/ip/{addr}`. It also shows
- 2026-10-27: on the AS page, a click on the heading of the IPv4 or the IPv6 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) 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, - 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 shortest first. Prefixes of one mask length are listed by address. They used
listed by address (closes #153) to be sorted by address first, then by mask length (closes #153)
- 2026-10-27: on `/` and `/status`, a spinner beside the status turns one frame - 2026-10-27: The connection status on `/` and `/status` gets a spinner. It
each time the stats show more messages than the time before. A page that has turns one frame each time the stats show more messages than the time before.
not seen more messages for three refreshes shows "stale" and the time since After three refreshes with no new messages, the page shows "stale" in red
then instead, in red (closes #152) 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. - 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 LACNIC's answer for such an AS is NIC.br's answer with each accented letter
accented letter replaced. The details now come from NIC.br's own server, replaced. NIC.br is Brazil's registry. The details now come from NIC.br's own
`whois.registro.br`. When that query fails, the details from LACNIC's answer server, `whois.registro.br`. When that query fails, the details from LACNIC's
are kept and a warning is logged. After NIC.br answers that RouteWatch is over answer are kept and a warning is logged. RouteWatch stops asking NIC.br for an
its query limit, RouteWatch stops asking NIC.br for an hour (closes #174) 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` - 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 breaks off. The generator used to retry the failed read forever. A record that
does not parse is still logged and skipped (closes #177) does not parse is still logged and skipped (closes #177)
@@ -183,182 +235,196 @@ tested.
one AS. A failed download used to replace the list with an empty file. The 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 generator reads `as.csv` from `ipverse/as-metadata`, the source's new name
(closes #173) (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 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 start their answer with the `as-block` object of the AS's block. Its dates
were stored as the AS's (closes #168) were stored as the AS's (closes #168)
- 2026-10-27: AS names keep their accented letters. The AS list built into - 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 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 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 WHOIS answer that is not valid UTF-8 is decoded as Latin-1 before it is parsed
APNIC send, before it is parsed and stored (closes #165) 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 - 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 "5 minutes ago" or "2 weeks ago". It used to print Go durations such as
ago". An age under a minute reads in seconds, and one over a month in months "360h0m0s ago". An age under a minute reads "now" or in seconds. An age over a
or years instead of a date (closes #163) 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 - 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 two first-seen times. It keeps the later of the two last-seen times. An old
from a collector catching up on a backlog no longer moves last seen back update from a collector catching up on a backlog no longer moves last seen
(closes #162) back (closes #162)
- 2026-10-26: the status page shows "Route Churn/min", the number of prefix - 2026-10-26: The status page shows "Route Churn/min". That is the number of
announcements and withdrawals, from all peers, in the last full clock minute. prefix announcements and withdrawals from all peers in the last full clock
`/api/v1/stats` reports it as `route_churn_per_min` in `stream`, which 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 replaces the moving average `route_churn_per_sec`. The messages and updates
per second on `/` and `/status` are whole numbers (closes #160) 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, 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 APNIC and AFRINIC leave out person and role objects. RIPE counts those objects
a daily limit per address. A refused query is logged with the RIR's error line against a daily limit per address. A refused query is logged with the RIR's
(closes #159) error line (closes #159)
- 2026-10-26: the home page's AS name search lists the ASes whose handle or - 2026-10-26: The home page's AS name search works. It lists the ASes whose
description holds the text, as links to their pages, from the new handle or description holds the text, as links to their pages. The new
`/api/v1/search/as`; the search ignores case. When the database has no record `/api/v1/search/as` answers it. The search ignores case. When the database has
of an AS, its page takes the handle and description from the AS list built no record of an AS, its page takes the handle and description from the AS list
into RouteWatch and still lists the AS's prefixes (closes #151) 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, - 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 README "Database Schema" matches `schema.sql`. The rest of the README matches
code; this file brought up to date (closes #72) 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 - 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 their random UUID keys. The prefix tables lose their duplicate index. A route
update sets only the columns that change, and every write goes through one 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 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) 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 - 2026-10-19: The checkpoint of the WAL (SQLite's write-ahead log), the
database lock every batch write holds, so a vacuum no longer fails with incremental vacuum and `ANALYZE` take the database lock that every batch write
"database is locked" and an ANALYZE no longer makes a batch write lose its holds. So a vacuum no longer fails with "database is locked". An `ANALYZE` no
updates; ANALYZE runs as `PRAGMA optimize` with an analysis limit (closes #94) longer makes a batch write lose its updates. `ANALYZE` runs as
- 2026-10-19: shutdown stops the streamer first, and each handler writes `PRAGMA optimize` with an analysis limit (closes #94)
everything left on its queue before it stops, so the updates waiting there are - 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) no longer lost (closes #69)
- 2026-10-19: a path that ends in an AS set of several members has origin AS 0, - 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 which means no known origin. It used to have the set's last member as its
peerings; the pages show such a route as having no known origin (closes #75) 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 - 2026-10-19: `bgp_peer_count` drops to 0 when the last peer goes down (closes
#92) #92)
- 2026-10-19: `bgp_peer_count` counts every peer the feed has heard from, not - 2026-10-19: `bgp_peer_count` counts every peer the feed has heard from. It
only those whose session opened since startup, and drops a peer whose session used to count only the peers whose session opened since startup. It drops a
goes down (closes #74) peer whose session goes down (closes #74)
- 2026-10-19: the shared files are re-vendored from `sneak/prompts` at - 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 `0653579`. The linter is `golangci-lint` v2.14.0 with its configuration. Lint
`Dockerfile` phases that keep Go's and golangci-lint's caches, markdown and test are `Dockerfile` phases that keep Go's and `golangci-lint`'s caches.
formatted with prettier, agent guidance in `AGENTS.md`; the code is fixed for `make fmt` also formats the markdown with `prettier`. The agent guidance moves
what the new linter configuration finds (closes #52) to `AGENTS.md`. The code is fixed for what the new linter configuration finds
- 2026-10-19: the streamer reads feed lines up to 1 MiB, so a line over 64 KiB (closes #52)
no longer ends the connection (closes #76) - 2026-10-19: The streamer reads feed lines up to 1 MiB. A line over 64 KiB used
- 2026-10-19: a peer's live routes are removed when its BGP session goes down; to end the connection (closes #76)
`STATE` messages are never dropped for backpressure, and one that finds its - 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) 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 - 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 index lookups. They used to read every route of that length. They list each
prefix once (closes #66) 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 `/api/v1/stats` answers 504 instead of 500 when its read runs out of time
(closes #70) (closes #70)
- 2026-10-19: the AS, peer and prefix handlers write each batch outside the lock - 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 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 a write. Meanwhile the next batch grows to at most ten batch sizes (closes
#65) #65)
- 2026-10-19: a live route is identified by prefix and peer, so an announcement - 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 with a new origin AS replaces the peer's route. A withdrawal removes it
whatever its origin AS (closes #55) whatever its origin AS (closes #55)
- 2026-10-14: the streamer reconnects when RIS Live ends the stream or sends - 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 nothing for 60 seconds. The health check reports the feed disconnected
meanwhile (closes #53) meanwhile (closes #53)
- 2026-10-14: the prefix page takes the names of the origin ASes from the prefix - 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) 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 - 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) 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 - 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 does not parse answers 400. Any valid way of writing a prefix finds its routes
routes (closes #71) (closes #71)
- 2026-10-14: the IP lookup ignores routes shorter than /8 for IPv4 and /16 for - 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) 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 - 2026-10-14: A prefix batch keeps each peer's last update for a prefix. It used
of one update per prefix across all peers (closes #54) to keep one update per prefix across all peers (closes #54)
- 2026-10-14: `reconnect_count` in `/api/v1/stats` counts every reconnect - 2026-10-14: `reconnect_count` in `/api/v1/stats` counts every reconnect
(closes #68) (closes #68)
- 2026-10-03: `/api/v1/stats` serves the prefix distribution from memory, seeded - 2026-10-03: `/api/v1/stats` serves the prefix distribution from memory. The
at startup and adjusted on every live-route write, so a request no longer distribution is seeded at startup and adjusted on every live-route write. So a
reads every live route (closes #30) request no longer reads every live route (closes #30)
- 2026-10-03: a new database is created with `auto_vacuum` set to incremental, - 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, The setting goes in the connection string, so it is set before the file is
and each periodic incremental vacuum now returns up to 1000 free pages; the first written. Each periodic incremental vacuum now returns up to 1000 free
late `PRAGMA auto_vacuum` in `Initialize`, which SQLite ignored, is gone pages. The late `PRAGMA auto_vacuum` in `Initialize`, which SQLite ignored, is
(closes #43) gone (closes #43)
- 2026-10-03: looking up an IP address no longer reads every IPv6 route: both - 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 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 the prefix index. The IPv4 range columns are gone. Prefixes from the feed are
are stored in one text form, so an IPv6 withdrawal, which the feed sends stored in one text form. So an IPv6 withdrawal, which the feed sends
uncompressed, now removes its route (closes #48) uncompressed, now removes its route (closes #48)
- 2026-10-02: a plain `docker build .` stamps the commit's tag or short commit - 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`: into the page footer. It used to stamp `unknown`. The value comes from
`.dockerignore` sends `.git` without `.git/config`, a `VERSION` build argument `git describe --tags --always`. `.dockerignore` sends `.git` without
takes precedence, and `make build` stamps the same value (closes #46) `.git/config`. A `VERSION` build argument takes precedence. `make build`
- 2026-09-29: the entrypoint creates the data directory if it is missing and stamps the same value (closes #46)
stops the start if a step fails; README "Running under upaas" no longer asks - 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) for the host directory to be created first (closes #42)
- 2026-09-29: `.dockerignore` keeps `.git`, local build output, local databases - 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 and `.env` out of the Docker build context. So they stay out of the source
in the image (closes #39) archive in the image (closes #39)
- 2026-09-29: README first line names the MIT license and the author; the - 2026-09-29: The README's first line names the MIT licence and the author. The
License section now says MIT and links `LICENSE`, and an Author section was License section says MIT and links `LICENSE`. An Author section is added. This
added; this file brought up to date (closes #38) file is brought up to date (closes #38)
- 2026-09-29: MIT `LICENSE` (closes #1) - 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 - 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 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 handing a message to the handler queues. A second `Stop` no longer closes the
the queues again (closes #34) queues again (closes #34)
- 2026-09-28: `docker stop` no longer kills the daemon 2 seconds after the stop - 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` signal. The entrypoint switches to the `routewatch` user with `setpriv`
instead of `runuser`, so the daemon receives the signal itself and gets the 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) 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`, - 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 `XDG_DATA_HOME` or `MALLOC_ARENA_MAX` stops the start. The health check
follows `PORT`, README "Running under upaas" section (closes #31) follows `PORT`. README gains a "Running under upaas" section (closes #31)
- 2026-09-22: realtime in-memory database statistics: counts seeded at startup - 2026-09-22: The database statistics are kept in memory. The counts are seeded
and adjusted on every write, oldest/newest route timestamps via index-end at startup and adjusted on every write. The oldest and newest route times come
lookups; `/api/v1/stats` no longer scans the tables (closes #27) from a lookup at each end of an index. `/api/v1/stats` no longer scans the
- 2026-09-21: batch writes take the write lock when their transaction begins tables (closes #27)
(`_txlock=immediate`), so they wait out a WAL checkpoint instead of failing - 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) with "database is locked" (closes #25)
- 2026-09-21: `MALLOC_ARENA_MAX=2` in the image caps glibc malloc arenas, so - 2026-09-21: `MALLOC_ARENA_MAX=2` in the image caps the number of glibc malloc
memory outside the Go runtime no longer grows with the core count (closes #23) arenas. So memory outside the Go runtime no longer grows with the core count
- 2026-09-21: `GOMEMLIMIT=1536MiB` in the image; README Memory section with the (closes #23)
- 2026-09-21: The image sets `GOMEMLIMIT=1536MiB`. README "Memory" gives the
memory budget and the 5 GiB container limit (closes #13) 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) from 100,000 (closes #11)
- 2026-09-21: two goroutine leaks fixed: the stats handlers after a timeout and - 2026-09-21: Two goroutine leaks are fixed. The stats handlers leaked after a
the streamer's tickers on every reconnect (closes #12) timeout. The streamer's tickers leaked on every reconnect (closes #12)
- 2026-09-21: parsed RIS messages no longer keep the unused `Community` and - 2026-09-21: Parsed RIS messages no longer keep the unused `Community` and
`Raw` fields (closes #9) `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) on every push (closes #14)
- 2026-09-21: the peering handler's AS-path map holds at most 500,000 paths and - 2026-09-21: The peering handler's AS-path map holds at most 500,000 paths.
is swapped for an empty one every 30 seconds instead of copied (closes #10) Every 30 seconds it is swapped for an empty one instead of copied (closes #10)
- 2026-09-21: SQLite memory bounded across the whole connection pool: a 64 MiB - 2026-09-21: SQLite's memory is bounded across the whole connection pool. Each
page cache on each connection, 1 GiB soft and 1.5 GiB hard heap limits (closes connection has a 64 MiB page cache. The heap limits are 1 GiB soft and 1.5 GiB
#8) hard (closes #8)
- 2026-09-21: the Docker build runs the format check, the linter and the tests, - 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 Linting runs in a separate stage on a `golangci-lint` image pinned by digest
#5) (closes #5)
- 2026-09-21: `make test` skips the live-network feed test (`-short`), so - 2026-09-21: `make test` skips the live feed test, with `-short`. So
`make check` no longer depends on the network (closes #2) `make check` no longer depends on the network (closes #2)
- 2026-07-07 Adopted scripts-to-rule-them-all: `script/` entrypoints, Makefile - 2026-07-07: The scripts-to-rule-them-all layout is adopted. `script/` holds
shims, README Entrypoints section the entrypoints. The `Makefile` targets `check`, `test`, `lint`, `fmt` and
- 2026-02-22: repo policy compliance: required policy files, .gitignore update, `fmt-check` call them. So do `bootstrap`, `setup`, `docker` and `hooks`.
Makefile fmt-check/check/docker/hooks targets, gofmt pass README gains an "Entrypoints" section
(repo-policies-compliance, unmerged) - 2026-02-22: The repo is brought in line with the repo policies. The required
- 2026-01-01: WAL checkpointing: periodic (5s), on startup with logging, policy files are added. `.gitignore` is updated. The `Makefile` gains
TRUNCATE mode; fixed slow queries `fmt-check`, `check`, `docker` and `hooks` targets. The Go code is formatted
- 2025-12-31: status page: navbar, home page with search, oldest/newest route with `gofmt`. The work was on the branch `repo-policies-compliance`, unmerged
timestamps, NULL handling in WHOIS stats; container runs as routewatch user when this entry was written
with proper state dir - 2026-01-01: The WAL is checkpointed every 5 seconds, and at startup with a log
- 2025-12-30: structured HTTP request logging, increased timeouts, CIDR prefix line. Checkpoints run in `TRUNCATE` mode. This fixed slow queries
URL routing fixes, status page metrics and footer - 2025-12-31: The pages get a navigation bar and a home page with a search. The
- 2025-12-29: Dockerfile multi-stage build with source archive; SQLite status page shows the oldest and newest route times. The WHOIS stats query
incremental vacuum for non-blocking space reclamation handles `NULL`. The container runs as the `routewatch` user with the right
- 2025-07-27: initial RouteWatch BGP stream monitor 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 # Future Steps
None. - Merge https://git.eeqj.de/sneak/routewatch/pulls/210 once a review passes. As
the last of the two PRs for https://git.eeqj.de/sneak/routewatch/issues/157 to
merge, it closes the issue.
+1 -1
View File
@@ -273,7 +273,7 @@ func TestVacuumReturnsFreePages(t *testing.T) {
writeTableChanges(t, db, table) writeTableChanges(t, db, table)
// The peer going down removes every prefix, and so every row. // The peer going down removes every prefix, and so every row.
table.PeerDown(peerA) table.PeersDown([]string{peerA})
writeTableChanges(t, db, table) writeTableChanges(t, db, table)
var before int var before int
+2 -2
View File
@@ -107,7 +107,7 @@ func TestReadRouteTableGivesBackWhatWasWritten(t *testing.T) {
// A peer moves to another origin, one goes down and a new one comes. // A peer moves to another origin, one goes down and a new one comes.
announceTo(source, tablePrefixV4, asB, peerA, 60) announceTo(source, tablePrefixV4, asB, peerA, 60)
source.PeerDown(peerC) source.PeersDown([]string{peerC})
announceTo(source, tablePrefixV6, asB, peerE, 70) announceTo(source, tablePrefixV6, asB, peerE, 70)
writeTableChanges(t, db, source) writeTableChanges(t, db, source)
@@ -133,7 +133,7 @@ func TestReadRouteTableGivesATableThatTakesLaterUpdates(t *testing.T) {
// another one goes down, and a new peer comes. // another one goes down, and a new peer comes.
for _, table := range []*routetable.Table{source, read} { for _, table := range []*routetable.Table{source, read} {
table.Withdraw(netip.MustParsePrefix(tablePrefixV4), peerA) table.Withdraw(netip.MustParsePrefix(tablePrefixV4), peerA)
table.PeerDown(peerB) table.PeersDown([]string{peerB})
announceTo(table, tablePrefixShortV4, asB, peerD, 50) announceTo(table, tablePrefixShortV4, asB, peerD, 50)
} }
+15 -6
View File
@@ -79,8 +79,9 @@ type OriginPeers struct {
// Overview works out the table's overview. It lists at most maxListed bogon // Overview works out the table's overview. It lists at most maxListed bogon
// announcements and at most maxListed prefixes with more than one origin AS: // announcements and at most maxListed prefixes with more than one origin AS:
// the first in prefix order. It walks over every prefix, but holds the read // the first in prefix order. It visits every prefix in the table, in batches
// lock for walkBatch prefixes at a time. // of walkBatch. After each batch, it lets the table's lock go for as long as
// it held the lock for the batch.
func (t *Table) Overview(maxListed int) Overview { func (t *Table) Overview(maxListed int) Overview {
var overview Overview var overview Overview
@@ -110,15 +111,18 @@ func (t *Table) Overview(maxListed int) Overview {
} }
// walkPrefixes calls visit with every prefix of the table and its origins. It // walkPrefixes calls visit with every prefix of the table and its origins. It
// holds the read lock, but lets it go after every walkBatch prefixes, so the // holds the read lock for walkBatch prefixes at a time. After each batch, it
// updates waiting for it get in. A prefix that is added or removed in such a // lets the table's lock go for as long as it held the lock for the batch.
// gap may or may not be visited, as Go allows for a map that changes during a // During a walk over many batches, updates can then take the lock about half
// range over it. visit must not keep origins once it returns. // of the time. The walk may or may not visit a prefix that is added or removed
// while the lock is let go. Go allows this in a range over a map that changes.
// visit must not keep origins once it returns.
func (t *Table) walkPrefixes(visit func(prefix netip.Prefix, origins []*origin)) { func (t *Table) walkPrefixes(visit func(prefix netip.Prefix, origins []*origin)) {
t.mu.RLock() t.mu.RLock()
defer t.mu.RUnlock() defer t.mu.RUnlock()
visited := 0 visited := 0
batchStart := time.Now()
for _, f := range []*family{&t.ipv4, &t.ipv6} { for _, f := range []*family{&t.ipv4, &t.ipv6} {
for _, prefixes := range f.byLength { for _, prefixes := range f.byLength {
@@ -127,8 +131,13 @@ func (t *Table) walkPrefixes(visit func(prefix netip.Prefix, origins []*origin))
visited++ visited++
if visited%walkBatch == 0 { if visited%walkBatch == 0 {
held := time.Since(batchStart)
t.mu.RUnlock() t.mu.RUnlock()
time.Sleep(held)
t.mu.RLock() t.mu.RLock()
batchStart = time.Now()
} }
} }
} }
@@ -131,3 +131,61 @@ func TestWalkPrefixesLetsUpdatesIn(t *testing.T) {
"before its second batch") "before its second batch")
} }
} }
// TestOverviewLetsUpdatesIn applies another peer's 1,000 announcements while
// Overview walks a table with more prefixes than five of the walk's batches.
// After each batch, the walk lets the lock go for as long as it held the lock.
// So all 1,000 announcements must be applied before Overview returns. A walk
// that takes the lock back at once lets in about one update per batch.
func TestOverviewLetsUpdatesIn(t *testing.T) {
t.Parallel()
const (
asn = 1000
peer = "192.0.2.1"
otherPeer = "192.0.2.2"
// prefixCount is more prefixes than five of the walk's batches.
prefixCount = 5*walkBatch + 1
announcements = 1000
)
routeTime := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
table := New(Params{})
prefixes := make([]netip.Prefix, 0, prefixCount)
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})
prefixes = append(prefixes, prefix)
addr = addr.Next()
}
overviewDone := make(chan struct{})
go func() {
defer close(overviewDone)
table.Overview(0)
}()
// Wait until Overview holds the read lock. TryLock fails while a reader
// holds it.
for table.mu.TryLock() {
table.mu.Unlock()
runtime.Gosched()
}
for _, prefix := range prefixes[:announcements] {
table.Announce(prefix, asn, Route{Peer: otherPeer, Time: routeTime})
}
select {
case <-overviewDone:
t.Fatal("Overview returned before another peer's 1,000 announcements " +
"were applied")
default:
}
<-overviewDone
}
+1 -1
View File
@@ -227,7 +227,7 @@ func TestCounts(t *testing.T) {
}) })
// peerC was the last peer of the IPv6 prefix. It keeps its number. // peerC was the last peer of the IPv6 prefix. It keeps its number.
table.PeerDown(peerC) table.PeersDown([]string{peerC})
checkCounts(t, table, routetable.Counts{ checkCounts(t, table, routetable.Counts{
Pairs: 2, Pairs: 2,
Peers: 3, Peers: 3,
+26 -11
View File
@@ -28,7 +28,7 @@ const (
keptTimesDegree = 32 keptTimesDegree = 32
// walkBatch is how many prefixes a walk over the whole table visits in // 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. // one hold of the lock. Overview and PeersDown each walk the whole table.
// Callers waiting for the lock get it between batches. // Callers waiting for the lock get it between batches.
walkBatch = 10000 walkBatch = 10000
@@ -205,20 +205,29 @@ func (t *Table) Withdraw(prefix netip.Prefix, peer string) {
} }
} }
// PeerDown applies a peer's session going down: the peer leaves every peer // PeersDown applies the sessions of peers going down: each of the peers leaves
// set, and every route of the peer is dropped. When the peer carries any // every peer set, and every route of each is dropped. When any of them carries
// prefix, PeerDown visits every prefix in the table, in batches of walkBatch. // a prefix, PeersDown visits every prefix in the table once, in batches of
// After each batch, it lets the table's lock go for as long as it held the // walkBatch. After each batch, it lets the table's lock go for as long as it
// lock for the batch. During a walk over many batches, reads and updates can // held the lock for the batch. During a walk over many batches, reads and
// then take the lock about half of the time. An announcement by the peer // updates can then take the lock about half of the time. An announcement by
// applied during the walk may be dropped too. So the caller applies the peer's // one of the peers applied during the walk may be dropped too. So the caller
// later updates only once PeerDown has returned. // applies those peers' later updates only once PeersDown has returned.
func (t *Table) PeerDown(peer string) { func (t *Table) PeersDown(peers []string) {
t.mu.Lock() t.mu.Lock()
defer t.mu.Unlock() defer t.mu.Unlock()
// numbers holds the number of each of the peers that carries a prefix.
numbers := make([]uint, 0, len(peers))
for _, peer := range peers {
number, known := t.peerNumbers[peer] number, known := t.peerNumbers[peer]
if !known || t.peerPrefixes[number] == 0 { if known && t.peerPrefixes[number] > 0 {
numbers = append(numbers, number)
}
}
if len(numbers) == 0 {
return return
} }
@@ -228,9 +237,15 @@ func (t *Table) PeerDown(peer string) {
for _, f := range []*family{&t.ipv4, &t.ipv6} { for _, f := range []*family{&t.ipv4, &t.ipv6} {
for _, prefixes := range f.byLength { for _, prefixes := range f.byLength {
for prefix, origins := range prefixes { for prefix, origins := range prefixes {
for _, number := range numbers {
o := originWithPeer(origins, number) o := originWithPeer(origins, number)
if o != nil { if o != nil {
t.leave(prefix, o, number) t.leave(prefix, o, number)
// leave may have removed an origin of the
// prefix, or the prefix itself.
origins = prefixes[prefix]
}
} }
// Go allows the map to change before the next step of // Go allows the map to change before the next step of
@@ -7,11 +7,11 @@ import (
"time" "time"
) )
// TestPeerDownLetsReadsIn reads the table's counts again and again while a peer // TestPeersDownLetsReadsIn reads the table's counts again and again while a
// that carries more than two batches of prefixes goes down. PeerDown lets the // peer that carries more than two batches of prefixes goes down. PeersDown lets
// lock go after every walkBatch prefixes, so a read must get in while the peer // the lock go after every walkBatch prefixes, so a read must get in while the
// still carries some of its prefixes but no longer all. // peer still carries some of its prefixes but no longer all.
func TestPeerDownLetsReadsIn(t *testing.T) { func TestPeersDownLetsReadsIn(t *testing.T) {
t.Parallel() t.Parallel()
const ( const (
@@ -32,8 +32,8 @@ func TestPeerDownLetsReadsIn(t *testing.T) {
addr = addr.Next() addr = addr.Next()
} }
// Hold a read lock until PeerDown waits for the lock, so that the first // Hold a read lock until PeersDown waits for the lock, so that the first
// read below starts once PeerDown holds it. TryRLock fails once a writer // read below starts once PeersDown holds it. TryRLock fails once a writer
// waits. // waits.
table.mu.RLock() table.mu.RLock()
@@ -42,7 +42,7 @@ func TestPeerDownLetsReadsIn(t *testing.T) {
go func() { go func() {
defer close(peerDownDone) defer close(peerDownDone)
table.PeerDown(peer) table.PeersDown([]string{peer})
}() }()
for table.mu.TryRLock() { for table.mu.TryRLock() {
@@ -60,7 +60,7 @@ func TestPeerDownLetsReadsIn(t *testing.T) {
select { select {
case <-peerDownDone: case <-peerDownDone:
t.Fatal("no read got in while PeerDown ran") t.Fatal("no read got in while PeersDown ran")
default: default:
} }
} }
+44 -11
View File
@@ -69,10 +69,10 @@ func withdraw(peer string) step {
} }
} }
// peerDown is the step of peer's session going down. // peersDown is the step of the sessions of peers going down.
func peerDown(peer string) step { func peersDown(peers ...string) step {
return func(table *routetable.Table, _ netip.Prefix) { return func(table *routetable.Table, _ netip.Prefix) {
table.PeerDown(peer) table.PeersDown(peers)
} }
} }
@@ -272,21 +272,28 @@ func TestWithdrawalsAndPeerDowns(t *testing.T) {
{ {
name: "a peer going down leaves the prefix", name: "a peer going down leaves the prefix",
steps: []step{announce(peerA, asA, 10), announce(peerB, asA, 20), steps: []step{announce(peerA, asA, 10), announce(peerB, asA, 20),
peerDown(peerB)}, peersDown(peerB)},
want: []routetable.Origin{{ASN: asA, Peers: []string{peerA}, want: []routetable.Origin{{ASN: asA, Peers: []string{peerA},
Routes: []routetable.Route{routeOf(peerA, asA, 10)}}}, Routes: []routetable.Route{routeOf(peerA, asA, 10)}}},
}, },
{ {
name: "a peer going down after it withdrew changes nothing", name: "a peer going down after it withdrew changes nothing",
steps: []step{announce(peerA, asA, 10), announce(peerB, asA, 20), steps: []step{announce(peerA, asA, 10), announce(peerB, asA, 20),
withdraw(peerB), peerDown(peerB)}, withdraw(peerB), peersDown(peerB)},
want: []routetable.Origin{{ASN: asA, Peers: []string{peerA}, want: []routetable.Origin{{ASN: asA, Peers: []string{peerA},
Routes: []routetable.Route{routeOf(peerA, asA, 10)}}}, Routes: []routetable.Route{routeOf(peerA, asA, 10)}}},
}, },
{
name: "peers going down together each leave the prefix",
steps: []step{announce(peerA, asA, 10), announce(peerB, asB, 20),
announce(peerC, asB, 30), peersDown(peerA, peerB)},
want: []routetable.Origin{{ASN: asB, Peers: []string{peerC},
Routes: []routetable.Route{routeOf(peerC, asB, 30)}}},
},
}) })
} }
func TestPeerDownLeavesEveryPrefix(t *testing.T) { func TestPeersDownLeavesEveryPrefix(t *testing.T) {
t.Parallel() t.Parallel()
table := routetable.New(routetable.Params{}) table := routetable.New(routetable.Params{})
@@ -303,7 +310,7 @@ func TestPeerDownLeavesEveryPrefix(t *testing.T) {
} }
table.Announce(prefixes[0], asB, routeOf(peerB, asB, 20)) table.Announce(prefixes[0], asB, routeOf(peerB, asB, 20))
table.PeerDown(peerA) table.PeersDown([]string{peerA})
want := []routetable.Origin{{ASN: asB, Peers: []string{peerB}, want := []routetable.Origin{{ASN: asB, Peers: []string{peerB},
Routes: []routetable.Route{routeOf(peerB, asB, 20)}}} Routes: []routetable.Route{routeOf(peerB, asB, 20)}}}
@@ -333,6 +340,32 @@ func TestPeerDownLeavesEveryPrefix(t *testing.T) {
} }
} }
// TestPeersDownTakesOutPeersOfDifferentPrefixes takes peerA and peerB out in
// one PeersDown. peerA carries only prefixV4, and peerB only prefixV6. So
// peerA's leaving removes prefixV4 before PeersDown looks for peerB there. Both
// prefixes must then be gone.
func TestPeersDownTakesOutPeersOfDifferentPrefixes(t *testing.T) {
t.Parallel()
table := routetable.New(routetable.Params{})
prefixes := []netip.Prefix{
netip.MustParsePrefix(prefixV4),
netip.MustParsePrefix(prefixV6),
}
table.Announce(prefixes[0], asA, routeOf(peerA, asA, 10))
table.Announce(prefixes[1], asB, routeOf(peerB, asB, 20))
table.PeersDown([]string{peerA, peerB})
for _, prefix := range prefixes {
got := table.Origins(prefix)
if got != nil {
t.Errorf("Origins(%s) = %+v after its only peer went down, want nil",
prefix, got)
}
}
}
func TestCountsOriginsWithoutKeptRoute(t *testing.T) { func TestCountsOriginsWithoutKeptRoute(t *testing.T) {
t.Parallel() t.Parallel()
@@ -358,7 +391,7 @@ func TestCountsOriginsWithoutKeptRoute(t *testing.T) {
{announce(peerC, asA, 30), 0}, {announce(peerC, asA, 30), 0},
{withdraw(peerB), 0}, {withdraw(peerB), 0},
{withdraw(peerC), 1}, {withdraw(peerC), 1},
{peerDown(peerA), 0}, {peersDown(peerA), 0},
} }
for i, s := range steps { for i, s := range steps {
@@ -464,7 +497,7 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) {
checkChanges(t, table, []change{}, map[string]uint{}) checkChanges(t, table, []change{}, map[string]uint{})
// peerA going down changes v4, the one prefix it carries. // peerA going down changes v4, the one prefix it carries.
table.PeerDown(peerA) table.PeersDown([]string{peerA})
checkChanges(t, table, []change{ checkChanges(t, table, []change{
{Prefix: prefixV4, OriginASN: asA, Peers: []uint{1, 2}, {Prefix: prefixV4, OriginASN: asA, Peers: []uint{1, 2},
Routes: []routetable.Route{ Routes: []routetable.Route{
@@ -473,7 +506,7 @@ func TestTakeChangesHoldsOnlyPairsThatChanged(t *testing.T) {
}, map[string]uint{}) }, map[string]uint{})
// peerA, which carries nothing now, goes down again. // peerA, which carries nothing now, goes down again.
table.PeerDown(peerA) table.PeersDown([]string{peerA})
checkChanges(t, table, []change{}, map[string]uint{}) checkChanges(t, table, []change{}, map[string]uint{})
} }
@@ -617,7 +650,7 @@ func TestConcurrentUse(t *testing.T) {
func(round int) { table.Announce(prefix, asA, routeOf(peerA, asA, round)) }, func(round int) { table.Announce(prefix, asA, routeOf(peerA, asA, round)) },
func(round int) { table.Announce(prefix, asB, routeOf(peerB, asB, round)) }, func(round int) { table.Announce(prefix, asB, routeOf(peerB, asB, round)) },
func(int) { table.Withdraw(prefix, peerA) }, func(int) { table.Withdraw(prefix, peerA) },
func(int) { table.PeerDown(peerB) }, func(int) { table.PeersDown([]string{peerB}) },
func(int) { table.PutBackChanges(table.TakeChanges(math.MaxInt)) }, func(int) { table.PutBackChanges(table.TakeChanges(math.MaxInt)) },
func(int) { table.ChangeCount() }, func(int) { table.ChangeCount() },
func(int) { table.Lookup(prefix.Addr()) }, func(int) { table.Lookup(prefix.Addr()) },
+97 -41
View File
@@ -46,10 +46,19 @@ const (
maxTableChangesPerWrite = 20000 maxTableChangesPerWrite = 20000
// maxHeldBackMessages is the most messages the handler holds back from // maxHeldBackMessages is the most messages the handler holds back from
// peers whose PeerDown has not returned. A message takes about 1 KiB, so // peers whose session went down. A peer's messages are held back from
// that many take about 20 MiB (README "Memory"). At this number // the moment its session goes down until peerDown has applied them. If
// HandleMessage waits until one of them is applied. // peerDown is running at that moment, the peer waits for the rest of
maxHeldBackMessages = 20000 // peerDown's current PeersDown call. The peer also waits while peerDown
// applies the messages held back from that call's peers, those that
// arrive meanwhile included. The next PeersDown call takes the peer out.
// Once that call returns, peerDown applies the messages held back from
// that call's peers, one peer after another, in the order the call lists
// them. A held-back message that says the peer's session went down again
// makes the peer's later messages wait for the next call. A message takes
// about 1 KiB, so 150,000 take about 150 MiB (README "Memory"). At this
// number HandleMessage waits until one of them is applied.
maxHeldBackMessages = 150000
) )
// PrefixHandler keeps the routing table in memory current from the feed. It // PrefixHandler keeps the routing table in memory current from the feed. It
@@ -58,14 +67,16 @@ const (
// does an announcement whose AS path is empty or holds AS 0 anywhere. A peer // does an announcement whose AS path is empty or holds AS 0 anywhere. A peer
// whose BGP session goes down leaves every prefix. // whose BGP session goes down leaves every prefix.
// //
// The table's PeerDown applies a session going down. It visits every prefix, // The table's PeersDown applies sessions going down. It visits every prefix,
// which takes seconds on a full table. So the handler runs PeerDown on // which takes seconds on a full table. So the handler runs PeersDown on a
// a separate goroutine, one peer at a time, and goes on taking messages // separate goroutine and goes on taking messages meanwhile. One PeersDown runs
// meanwhile. PeerDown lets the table's lock go about half of the time, so the // at a time. The next one takes every peer whose session went down since the
// handler applies other peers' messages while PeerDown runs. The handler holds // previous one began. PeersDown lets the table's lock go about half of the
// back the down peer's later messages until PeerDown returns. Then the handler // time, so the handler applies other peers' messages while PeersDown runs. The
// applies them in order, one at a time, and takes new messages in between. It // handler holds back a down peer's later messages until peerDown applies them,
// holds back at most maxHeldBackMessages messages. // as the doc of maxHeldBackMessages describes. It 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 // 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. // prefix the updates name with the time it was first seen.
@@ -103,20 +114,25 @@ type PrefixHandler struct {
stopCh chan context.Context stopCh chan context.Context
done chan struct{} // closed when the flush goroutine returns done chan struct{} // closed when the flush goroutine returns
// Peers whose session went down. peerDownMu lets one PeerDown run at a // Peers whose session went down. heldBack, under mu, holds each such
// time, so the handler's own updates wait for the table's lock behind // peer's later messages, in order, until the peer has left every prefix
// at most one. heldBack, under mu, holds each such peer's later // and they are applied. A peer with no such messages yet has an empty
// messages, in order, until its PeerDown has returned and they are // entry. heldCount, under mu, is how many messages heldBack holds.
// 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 // heldTaken is signalled each time a held-back message is taken out to
// be applied; its L is &mu. peerDowns counts the peerDown goroutines // be applied; its L is &mu.
// that have not returned.
peerDownMu sync.Mutex
heldBack map[string][]*ristypes.RISMessage heldBack map[string][]*ristypes.RISMessage
heldCount int heldCount int
heldTaken sync.Cond heldTaken sync.Cond
// downPeers, under mu, lists the peers whose session went down and that
// the next PeersDown takes. peerDownRunning, under mu, says whether
// peerDown is running. peerDowns counts the peerDown goroutines that have
// not returned.
downPeers []string
peerDownRunning bool
peerDowns sync.WaitGroup peerDowns sync.WaitGroup
// tablePeersDown is the table's PeersDown. Only peerDown calls it, so a
// test may replace it before it hands the handler a message.
tablePeersDown func(peers []string)
} }
// seenPrefix is what the batch keeps of an update: the prefix it names and its // seenPrefix is what the batch keeps of an update: the prefix it names and its
@@ -181,6 +197,7 @@ func newPrefixHandler(
stopCh: make(chan context.Context, 1), stopCh: make(chan context.Context, 1),
done: make(chan struct{}), done: make(chan struct{}),
heldBack: make(map[string][]*ristypes.RISMessage), heldBack: make(map[string][]*ristypes.RISMessage),
tablePeersDown: table.PeersDown,
} }
h.batchTaken.L = &h.mu h.batchTaken.L = &h.mu
h.heldTaken.L = &h.mu h.heldTaken.L = &h.mu
@@ -212,7 +229,7 @@ func (h *PrefixHandler) QueueCapacity() int {
} }
// HandleMessage processes a message to track prefix information. A message // HandleMessage processes a message to track prefix information. A message
// from a peer whose PeerDown has not returned is held back until it has. // from a peer whose session went down is held back until peerDown applies it.
func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
h.mu.Lock() h.mu.Lock()
defer h.mu.Unlock() defer h.mu.Unlock()
@@ -235,7 +252,16 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
h.applyMessage(msg) h.applyMessage(msg)
} }
// Stop waits for every PeerDown the handler started and for the messages held // HeldBackCount returns how many messages the handler holds back from peers
// whose session went down. The status page shows it.
func (h *PrefixHandler) HeldBackCount() int {
h.mu.Lock()
defer h.mu.Unlock()
return h.heldCount
}
// Stop waits for every PeersDown the handler started and for the messages held
// back meanwhile to be applied. It gives up this wait once ctx ends. Then Stop // 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 // 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 // goroutine makes one last flush before it returns. That flush writes at most
@@ -290,8 +316,8 @@ func (h *PrefixHandler) Stop(ctx context.Context) {
} }
// applyMessage applies msg. The caller holds mu. A peer's session going down // 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 // has the peer leave every prefix. An update is applied to the routing table,
// its prefixes join the batch. // and its prefixes join the batch.
func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) { func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) {
if msg.Type == ristypes.MessageTypeState { if msg.Type == ristypes.MessageTypeState {
if msg.State == ristypes.StateDown { if msg.State == ristypes.StateDown {
@@ -370,34 +396,64 @@ func (h *PrefixHandler) applyMessage(msg *ristypes.RISMessage) {
} }
} }
// startPeerDown starts peerDown for peer on a new goroutine. The caller holds // startPeerDown adds peer to downPeers. It starts peerDown on a new goroutine
// mu. The peer's messages are held back until peerDown has applied them. // unless peerDown is running. The caller holds mu. The peer's messages are
// Messages held back for an earlier PeerDown of the peer, and not yet applied, // held back until peerDown has applied them. Messages held back for an earlier
// stay held back. // session of the peer going down, and not yet applied, stay held back.
func (h *PrefixHandler) startPeerDown(peer string) { func (h *PrefixHandler) startPeerDown(peer string) {
_, goingDown := h.heldBack[peer] _, goingDown := h.heldBack[peer]
if !goingDown { if !goingDown {
h.heldBack[peer] = []*ristypes.RISMessage{} h.heldBack[peer] = []*ristypes.RISMessage{}
} }
h.peerDowns.Add(1) h.downPeers = append(h.downPeers, peer)
go h.peerDown(peer) if h.peerDownRunning {
return
} }
// peerDown runs the table's PeerDown for peer, once no other PeerDown is h.peerDownRunning = true
// running. Then peerDown applies the messages held back from the peer, in h.peerDowns.Add(1)
// 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 go h.peerDown()
// 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) { // peerDown takes every peer in downPeers and runs the table's PeersDown for
// them. Then it applies the messages held back from each of those peers. It
// repeats this until downPeers is empty.
func (h *PrefixHandler) peerDown() {
defer h.peerDowns.Done() defer h.peerDowns.Done()
h.peerDownMu.Lock() for {
h.table.PeerDown(peer) h.mu.Lock()
h.peerDownMu.Unlock()
peers := h.downPeers
h.downPeers = nil
if len(peers) == 0 {
h.peerDownRunning = false
h.mu.Unlock()
return
}
h.mu.Unlock()
h.tablePeersDown(peers)
for _, peer := range peers {
h.applyHeldBack(peer)
}
}
}
// applyHeldBack applies the messages held back from 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 adds the
// peer to downPeers. The rest stay held back until the next PeersDown has
// returned.
func (h *PrefixHandler) applyHeldBack(peer string) {
for { for {
h.mu.Lock() h.mu.Lock()
@@ -238,7 +238,7 @@ func TestPrefixHandlerStopLogsNothingWhenItWroteEverything(t *testing.T) {
} }
// TestPrefixHandlerStopLogsTheMessagesHeldBack stops a prefix handler with a // TestPrefixHandlerStopLogsTheMessagesHeldBack stops a prefix handler with a
// deadline that has passed, while a peer's PeerDown is held up and one message // deadline that has passed, while a peer's PeersDown is held up and one message
// of the peer is held back. Stop must log that message as not applied. // of the peer is held back. Stop must log that message as not applied.
func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) { func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
t.Parallel() t.Parallel()
@@ -249,8 +249,8 @@ func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
routetable.New(routetable.Params{}), longerThanAnyTest, routetable.New(routetable.Params{}), longerThanAnyTest,
maxTableChangesPerWrite) maxTableChangesPerWrite)
h.peerDownMu.Lock() release := holdUpPeersDown(h)
defer h.peerDownMu.Unlock() defer close(release)
h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix)) h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
h.HandleMessage(sessionDown()) h.HandleMessage(sessionDown())
@@ -268,8 +268,8 @@ func TestPrefixHandlerStopLogsTheMessagesHeldBack(t *testing.T) {
} }
// TestPrefixHandlerStopWaitsForTheMessagesHeldBack stops a prefix handler // TestPrefixHandlerStopWaitsForTheMessagesHeldBack stops a prefix handler
// while a peer's PeerDown is held up and an announcement of the peer is held // while a peer's PeersDown 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 // back. Stop must wait. Once the PeersDown is let go and Stop returns, the
// announcement must be in the database. // announcement must be in the database.
func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) { func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) {
t.Parallel() t.Parallel()
@@ -278,7 +278,7 @@ func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) {
h := newPrefixHandler(db, logger.New(), routetable.New(routetable.Params{}), h := newPrefixHandler(db, logger.New(), routetable.New(routetable.Params{}),
longerThanAnyTest, maxTableChangesPerWrite) longerThanAnyTest, maxTableChangesPerWrite)
h.peerDownMu.Lock() release := holdUpPeersDown(h)
h.HandleMessage(sessionDown()) h.HandleMessage(sessionDown())
h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix)) h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
@@ -291,16 +291,16 @@ func TestPrefixHandlerStopWaitsForTheMessagesHeldBack(t *testing.T) {
select { select {
case <-stopped: case <-stopped:
t.Error("Stop returned while a PeerDown was held up") t.Error("Stop returned while a PeersDown was held up")
case <-time.After(stillWaitingAfter): case <-time.After(stillWaitingAfter):
} }
h.peerDownMu.Unlock() close(release)
select { select {
case <-stopped: case <-stopped:
case <-time.After(stepTimeout): case <-time.After(stepTimeout):
t.Fatalf("Stop did not return within %s of the PeerDown's release", t.Fatalf("Stop did not return within %s of the PeersDown's release",
stepTimeout) stepTimeout)
} }
@@ -324,16 +324,32 @@ const longerThanAnyTest = time.Hour
// applies the updates to an empty routing table. It is built without // applies the updates to an empty routing table. It is built without
// NewPrefixHandler's flush goroutine, so the test flushes itself. // NewPrefixHandler's flush goroutine, so the test flushes itself.
func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler { func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler {
table := routetable.New(routetable.Params{})
return &PrefixHandler{ return &PrefixHandler{
db: store, db: store,
logger: logger.New(), logger: logger.New(),
table: routetable.New(routetable.Params{}), table: table,
batchMaxSize: prefixBatchMaxSize, batchMaxSize: prefixBatchMaxSize,
tableChangesPerWrite: maxTableChangesPerWrite, tableChangesPerWrite: maxTableChangesPerWrite,
heldBack: make(map[string][]*ristypes.RISMessage), heldBack: make(map[string][]*ristypes.RISMessage),
tablePeersDown: table.PeersDown,
} }
} }
// holdUpPeersDown makes each PeersDown of h wait until the returned channel is
// closed, as another peer's long PeersDown would hold it up. The test calls it
// before it hands h a message.
func holdUpPeersDown(h *PrefixHandler) chan struct{} {
release := make(chan struct{})
h.tablePeersDown = func(peers []string) {
<-release
h.table.PeersDown(peers)
}
return release
}
// newTestDatabase opens a database in a new temporary directory. It is closed // newTestDatabase opens a database in a new temporary directory. It is closed
// when the test ends. // when the test ends.
func newTestDatabase(t *testing.T) *database.Database { func newTestDatabase(t *testing.T) *database.Database {
@@ -103,23 +103,24 @@ func TestPrefixHandlerAppliesUpdatesToTheRoutingTable(t *testing.T) {
}) })
} }
// TestPrefixHandlerTakesMessagesWhileAPeerGoesDown holds up the PeerDown of a // TestPrefixHandlerTakesMessagesWhileAPeerGoesDown holds up the PeersDown of a
// peer whose session went down, as another peer's long PeerDown would. The // peer whose session went down, as another peer's long PeersDown would. The
// handler must go on taking messages. It applies another peer's announcement // 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: // at once. It holds back these messages of the down peer, in this order:
// - an announcement; // - an announcement;
// - its session going down again; // - its session going down again;
// - a second announcement. // - a second announcement.
// //
// Once the PeerDowns have returned, the down peer must carry only the prefix // HeldBackCount must count them. Once the PeersDown calls have returned, the
// of the second announcement. // down peer must carry only the prefix of the second announcement.
func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) { func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
t.Parallel() t.Parallel()
h := prefixHandlerWithoutTimer(newMockStore()) const heldMessages = 3
h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
h.peerDownMu.Lock() h := prefixHandlerWithoutTimer(newMockStore())
release := holdUpPeersDown(h)
h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix))
handled := returnsInTime(func() { handled := returnsInTime(func() {
for _, msg := range []*ristypes.RISMessage{ for _, msg := range []*ristypes.RISMessage{
@@ -144,10 +145,15 @@ func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
setV4Prefix: {}, setV4Prefix: {},
}) })
h.peerDownMu.Unlock() if h.HeldBackCount() != heldMessages {
t.Errorf("HeldBackCount() = %d, want %d", h.HeldBackCount(),
heldMessages)
}
close(release)
if !returnsInTime(h.peerDowns.Wait) { if !returnsInTime(h.peerDowns.Wait) {
t.Fatalf("the PeerDowns did not return within %s", stepTimeout) t.Fatalf("the PeersDown calls did not return within %s", stepTimeout)
} }
checkTableOrigins(t, h.table, map[string]map[int][]string{ checkTableOrigins(t, h.table, map[string]map[int][]string{
@@ -158,21 +164,85 @@ func TestPrefixHandlerTakesMessagesWhileAPeerGoesDown(t *testing.T) {
}) })
} }
// TestPrefixHandlerAppliesMessagesWhilePeerDownRuns gives a peer 50,000 // TestPrefixHandlerTakesPeersDownTogether has four peers announce testV4Prefix.
// It holds up the PeersDown of the first peer, whose session goes down.
// Meanwhile the sessions of the three others go down, and each of them
// announces otherV4Prefix. Once the first PeersDown is let go, the next one
// must take the three others together. No peer may then carry testV4Prefix,
// and the three others must carry otherV4Prefix.
func TestPrefixHandlerTakesPeersDownTogether(t *testing.T) {
t.Parallel()
others := []string{secondPeerIP, thirdPeerIP, fourthPeerIP}
h := prefixHandlerWithoutTimer(newMockStore())
// calls receives the peers of each PeersDown, with room for one call
// for each peer.
calls := make(chan []string, len(others)+1)
release := make(chan struct{})
h.tablePeersDown = func(peers []string) {
calls <- slices.Clone(peers)
<-release
h.table.PeersDown(peers)
}
for _, peer := range append([]string{testPeerIP}, others...) {
h.HandleMessage(peerAnnouncement(peer, testV4Prefix))
}
h.HandleMessage(sessionDown())
select {
case <-calls:
case <-time.After(stepTimeout):
t.Fatalf("PeersDown did not begin within %s", stepTimeout)
}
for _, peer := range others {
h.HandleMessage(sessionDownOf(peer))
h.HandleMessage(peerAnnouncement(peer, otherV4Prefix))
}
close(release)
if !returnsInTime(h.peerDowns.Wait) {
t.Fatalf("the PeersDown calls did not return within %s", stepTimeout)
}
close(calls)
later := make([][]string, 0, len(calls))
for peers := range calls {
later = append(later, peers)
}
if !reflect.DeepEqual(later, [][]string{others}) {
t.Errorf("the PeersDown calls after the first took %v, want %v "+
"together", later, others)
}
checkTableOrigins(t, h.table, map[string]map[int][]string{
testV4Prefix: {},
otherV4Prefix: {testASNB: others},
})
}
// TestPrefixHandlerAppliesMessagesWhilePeersDownRuns gives a peer 50,000
// prefixes and hands the prefix handler that peer's session going down. The // 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 // table's PeersDown then visits the prefixes in five batches. Once PeersDown
// begun, the handler is handed 1,000 announcements of another peer. The // 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 // handler must apply them all before PeersDown ends, while the down peer still
// carries some of its prefixes. // carries some of its prefixes.
func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) { func TestPrefixHandlerAppliesMessagesWhilePeersDownRuns(t *testing.T) {
t.Parallel() t.Parallel()
const ( const (
// downPeerPrefixes is five times the 10,000 prefixes PeerDown visits // downPeerPrefixes is five times the 10,000 prefixes PeersDown visits
// in one hold of the table's lock. // in one hold of the table's lock.
downPeerPrefixes = 50000 downPeerPrefixes = 50000
// otherPeerMessages is far more messages than the handler applies // otherPeerMessages is far more messages than the handler applies
// while PeerDown runs, if PeerDown takes the lock back at once after // while PeersDown runs, if PeersDown takes the lock back at once after
// each batch. The handler then applies only a few for each batch. // each batch. The handler then applies only a few for each batch.
otherPeerMessages = 1000 otherPeerMessages = 1000
// firstOtherAddr is the address of the other peer's first prefix. // firstOtherAddr is the address of the other peer's first prefix.
@@ -184,7 +254,7 @@ func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) {
h.HandleMessage(sessionDown()) h.HandleMessage(sessionDown())
// Wait until PeerDown has begun: the down peer has left some of its // Wait until PeersDown has begun: the down peer has left some of its
// prefixes. The down peer's prefixes are IPv4 and the other peer's are // prefixes. The down peer's prefixes are IPv4 and the other peer's are
// IPv6, so the counts tell them apart. // IPv6, so the counts tell them apart.
for h.table.Counts().IPv4Routes == downPeerPrefixes { for h.table.Counts().IPv4Routes == downPeerPrefixes {
@@ -207,15 +277,76 @@ func TestPrefixHandlerAppliesMessagesWhilePeerDownRuns(t *testing.T) {
} }
if counts.IPv4Routes == 0 { if counts.IPv4Routes == 0 {
t.Errorf("PeerDown ended before the handler applied the other peer's "+ t.Errorf("PeersDown ended before the handler applied the other peer's "+
"%d announcements", otherPeerMessages) "%d announcements", otherPeerMessages)
} }
if !returnsInTime(h.peerDowns.Wait) { if !returnsInTime(h.peerDowns.Wait) {
t.Fatalf("the PeerDown did not return within %s", stepTimeout) t.Fatalf("the PeersDown did not return within %s", stepTimeout)
} }
} }
// TestPrefixHandlerTakesMessagesWhileAReturningPeerSendsItsTable gives a peer
// 50,000 prefixes and hands the prefix handler that peer's session going down.
// Once PeersDown has begun, the peer comes back and sends more announcements
// than the handler's queue holds, as a peer that sends its whole table again
// does. The handler must hold them back and go on taking messages. Another
// peer's announcement, handed next, must be applied before PeersDown ends.
func TestPrefixHandlerTakesMessagesWhileAReturningPeerSendsItsTable(t *testing.T) {
t.Parallel()
const (
// downPeerPrefixes is five times the 10,000 prefixes PeersDown visits
// in one hold of the table's lock.
downPeerPrefixes = 50000
returningMessages = prefixHandlerQueueSize + 1
)
table := routetable.New(routetable.Params{})
announceManyPrefixes(table, downPeerPrefixes)
h := newPrefixHandler(newMockStore(), logger.New(), table,
longerThanAnyTest, maxTableChangesPerWrite)
defer stopInTime(t, h)
h.HandleMessage(sessionDown())
// Wait until PeersDown has begun: the down peer has left some of its
// prefixes, which are all IPv4.
for h.table.Counts().IPv4Routes == downPeerPrefixes {
runtime.Gosched()
}
returning := peerAnnouncement(testPeerIP, testV6Prefix)
if !returnsInTime(func() {
for range returningMessages {
h.HandleMessage(returning)
}
h.HandleMessage(peerAnnouncement(secondPeerIP, testV6Prefix))
}) {
t.Fatalf("the handler did not take the messages within %s", stepTimeout)
}
if h.table.Counts().IPv4Routes == 0 {
t.Error("PeersDown ended before the handler took the other peer's " +
"announcement")
}
checkTableOrigins(t, h.table, map[string]map[int][]string{
testV6Prefix: {testASNB: {secondPeerIP}},
})
if !returnsInTime(h.peerDowns.Wait) {
t.Fatalf("the PeersDown did not return within %s", stepTimeout)
}
checkTableOrigins(t, h.table, map[string]map[int][]string{
testV6Prefix: {testASNB: {testPeerIP, secondPeerIP}},
})
}
// TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes holds back many // TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes holds back many
// messages of a peer whose session went down. Each names a prefix that does // 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 // not parse, so applying it logs an error. Each log write takes a
@@ -237,14 +368,14 @@ func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) {
defer stopInTime(t, h) defer stopInTime(t, h)
h.peerDownMu.Lock() release := holdUpPeersDown(h)
h.HandleMessage(sessionDown()) h.HandleMessage(sessionDown())
for range heldMessages { for range heldMessages {
h.HandleMessage(peerAnnouncement(testPeerIP, badPrefix)) h.HandleMessage(peerAnnouncement(testPeerIP, badPrefix))
} }
h.peerDownMu.Unlock() close(release)
select { select {
case <-logs.firstWrite: case <-logs.firstWrite:
@@ -268,10 +399,10 @@ func TestPrefixHandlerTakesMessagesWhileItAppliesHeldBackOnes(t *testing.T) {
} }
} }
// TestPrefixHandlerWaitsAtTheMaximumHeldBack holds up the PeerDown of a peer // TestPrefixHandlerWaitsAtTheMaximumHeldBack holds up the PeersDown of a peer
// whose session went down. It hands the handler maxHeldBackMessages messages // whose session went down. It hands the handler maxHeldBackMessages messages
// of that peer. HandleMessage must then wait, even for another peer's // of that peer. HandleMessage must then wait, even for another peer's
// message, until the PeerDown returns and a held-back message is applied. // message, until the PeersDown returns and a held-back message is applied.
func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) { func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
t.Parallel() t.Parallel()
@@ -280,7 +411,7 @@ func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
maxTableChangesPerWrite) maxTableChangesPerWrite)
defer stopInTime(t, h) defer stopInTime(t, h)
h.peerDownMu.Lock() release := holdUpPeersDown(h)
h.HandleMessage(sessionDown()) h.HandleMessage(sessionDown())
held := peerAnnouncement(testPeerIP, testV6Prefix) held := peerAnnouncement(testPeerIP, testV6Prefix)
@@ -302,12 +433,12 @@ func TestPrefixHandlerWaitsAtTheMaximumHeldBack(t *testing.T) {
case <-time.After(stillWaitingAfter): case <-time.After(stillWaitingAfter):
} }
h.peerDownMu.Unlock() close(release)
select { select {
case <-returned: case <-returned:
case <-time.After(stepTimeout): case <-time.After(stepTimeout):
t.Fatal("HandleMessage did not return after the PeerDown returned") t.Fatal("HandleMessage did not return after the PeersDown returned")
} }
} }
@@ -424,7 +555,7 @@ func TestPrefixHandlerKeepsThePathOfAPathEndingInASet(t *testing.T) {
} }
// checkTableCases hands the messages of each case to a new prefix handler. It // 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 // checks the handler's routing table once the handler's PeersDown calls have
// returned, with no flush. // returned, with no flush.
func checkTableCases(t *testing.T, tests []tableCase) { func checkTableCases(t *testing.T, tests []tableCase) {
t.Helper() t.Helper()
+2
View File
@@ -486,6 +486,7 @@ type HandlerStatsInfo struct {
QueueHighWaterMark int `json:"queue_high_water_mark"` QueueHighWaterMark int `json:"queue_high_water_mark"`
ProcessedCount uint64 `json:"processed_count"` ProcessedCount uint64 `json:"processed_count"`
DroppedCount uint64 `json:"dropped_count"` DroppedCount uint64 `json:"dropped_count"`
HeldBackCount int `json:"held_back_count"`
AvgProcessTimeMs float64 `json:"avg_process_time_ms"` AvgProcessTimeMs float64 `json:"avg_process_time_ms"`
MinProcessTimeMs float64 `json:"min_process_time_ms"` MinProcessTimeMs float64 `json:"min_process_time_ms"`
MaxProcessTimeMs float64 `json:"max_process_time_ms"` MaxProcessTimeMs float64 `json:"max_process_time_ms"`
@@ -570,6 +571,7 @@ func handlerStatsInfo(handlerStats []streamer.HandlerStats) []HandlerStatsInfo {
QueueHighWaterMark: hs.QueueHighWaterMark, QueueHighWaterMark: hs.QueueHighWaterMark,
ProcessedCount: hs.ProcessedCount, ProcessedCount: hs.ProcessedCount,
DroppedCount: hs.DroppedCount, DroppedCount: hs.DroppedCount,
HeldBackCount: hs.HeldBackCount,
AvgProcessTimeMs: float64(hs.AvgProcessTime.Microseconds()) / AvgProcessTimeMs: float64(hs.AvgProcessTime.Microseconds()) /
microsecondsPerMillisecond, microsecondsPerMillisecond,
MinProcessTimeMs: float64(hs.MinProcessTime.Microseconds()) / MinProcessTimeMs: float64(hs.MinProcessTime.Microseconds()) /
+37
View File
@@ -295,6 +295,43 @@ func TestStatsAnswerCarriesChurnPerMinute(t *testing.T) {
} }
} }
// heldBackMessages is how many messages heldBackHandler holds back.
const heldBackMessages = 7
// heldBackHandler is a stream handler that wants no message and holds back
// heldBackMessages messages.
type heldBackHandler struct{}
func (heldBackHandler) WantsMessage(string) bool { return false }
func (heldBackHandler) HandleMessage(*ristypes.RISMessage) {}
func (heldBackHandler) QueueCapacity() int { return 1 }
func (heldBackHandler) HeldBackCount() int { return heldBackMessages }
// TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack registers a handler
// that holds back messages. Its figures in /api/v1/stats, which the status
// page shows, must give how many.
func TestStatsAnswerGivesHowManyMessagesAHandlerHoldsBack(t *testing.T) {
t.Parallel()
s := newServerWithTestRoutes(t)
s.streamer.RegisterHandler(heldBackHandler{})
var body struct {
Data StatsResponse `json:"data"`
}
err := json.Unmarshal(get(t, s, statsPath).Body.Bytes(), &body)
if err != nil {
t.Fatalf("decoding the answer: %v", err)
}
stats := body.Data.HandlerStats
if len(stats) != 1 || stats[0].HeldBackCount != heldBackMessages {
t.Errorf("handler figures %+v, want one handler holding back %d "+
"messages", stats, heldBackMessages)
}
}
// fixedIntervalASNFetcher is a WHOIS fetcher that does nothing and reports // fixedIntervalASNFetcher is a WHOIS fetcher that does nothing and reports
// interval as its wait between fetches. // interval as its wait between fetches.
type fixedIntervalASNFetcher struct { type fixedIntervalASNFetcher struct {
+9
View File
@@ -351,6 +351,10 @@ type HandlerStats struct {
AvgProcessTime time.Duration AvgProcessTime time.Duration
MinProcessTime time.Duration MinProcessTime time.Duration
MaxProcessTime time.Duration MaxProcessTime time.Duration
// HeldBackCount is how many messages the handler holds back to apply
// later: what its HeldBackCount method returns, or 0 for a handler
// without one.
HeldBackCount int
} }
// GetHandlerStats returns a snapshot of performance statistics for all registered // GetHandlerStats returns a snapshot of performance statistics for all registered
@@ -392,6 +396,11 @@ func (s *Streamer) GetHandlerStats() []HandlerStats {
info.metrics.mu.Unlock() info.metrics.mu.Unlock()
holder, holdsBack := info.handler.(interface{ HeldBackCount() int })
if holdsBack {
hs.HeldBackCount = holder.HeldBackCount()
}
stats = append(stats, hs) stats = append(stats, hs)
} }
+4
View File
@@ -541,6 +541,10 @@
<span class="metric-label">Dropped</span> <span class="metric-label">Dropped</span>
<span class="metric-value ${handler.dropped_count > 0 ? 'disconnected' : ''}">${formatNumber(handler.dropped_count)}</span> <span class="metric-value ${handler.dropped_count > 0 ? 'disconnected' : ''}">${formatNumber(handler.dropped_count)}</span>
</div> </div>
<div class="metric">
<span class="metric-label">Held Back</span>
<span class="metric-value">${formatNumber(handler.held_back_count)}</span>
</div>
<div class="metric"> <div class="metric">
<span class="metric-label">Avg Time</span> <span class="metric-label">Avg Time</span>
<span class="metric-value">${formatProcessingTime(handler.avg_process_time_ms)}</span> <span class="metric-value">${formatProcessingTime(handler.avg_process_time_ms)}</span>