2 Commits
Author SHA1 Message Date
sneak ac338d8f10 Keep a trailing AS set out of the origin and the peerings (closes #75)
check / check (push) Canceled after 0s
A path that ends in an AS set of several members, such as
[64496, 64497, [64498, 64499]], now has no origin: its route is stored
with origin 0 (none), and no member of the set is in a peering. A set of
one member counts as that AS. An announcement with an empty path is
still not stored.

ASPath keeps such a set apart from the sequence and gives the origin.
The prefix and AS handlers take the origin from it; the peering handler
uses the sequence only. The stored AS path still lists the set's members
at the end.

The prefix and prefix length pages show origin 0 as no known origin, the
IP lookup asks no WHOIS lookup for it, and the README says what it means.

Model: opus-5-5
2026-10-19 10:35:31 +00:00
clawbot 3667d09800 Count every peer the feed has heard from in bgp_peer_count (closes #74)
check / check (push) Canceled after 0s
bgp_peer_count grew only on OPEN messages, so it started at 0 while
updates arrived from hundreds of peers whose sessions were already up, and
it kept peers whose sessions had gone down. Now any message from a peer
adds it to the streamer's peer set, and a NOTIFICATION or a STATE message
whose state is down removes it without adding it first. The rule is in
updatePeers, called for every message.

Model: opus-5-5
2026-10-19 12:24:49 +02:00
10 changed files with 231 additions and 23 deletions
+5
View File
@@ -54,6 +54,11 @@ DEBUG=routewatch ./routewatch
- `GET /api/v1/as/{asn}` - Get prefixes announced by an AS
- `GET /api/v1/prefix/{prefix}` - Get routes for a specific prefix
An origin AS of 0 (`origin_asn` in a route, `asn` in an IP lookup, the key `0`
in a prefix's `origins`) means the route has no known origin: its AS path ends
in an AS set of several members, any of which may have originated it. The pages
show such a route as having no known origin.
## Code Structure
```
+4 -3
View File
@@ -2141,11 +2141,12 @@ func (d *Database) GetIPInfoContext(
info.FirstSeen = prefixFirstSeen.Time
}
// Check if WHOIS data needs refresh (never fetched or older than 30 days)
// Check if WHOIS data needs refresh (never fetched or older than 30 days).
// A route with origin 0 has no known origin, so no AS to look up.
const staleThreshold = 30 * 24 * time.Hour
info.NeedsWHOISRefresh = !whoisUpdatedAt.Valid ||
time.Since(whoisUpdatedAt.Time) > staleThreshold
info.NeedsWHOISRefresh = info.ASN != 0 && (!whoisUpdatedAt.Valid ||
time.Since(whoisUpdatedAt.Time) > staleThreshold)
return info, nil
}
@@ -213,6 +213,33 @@ func TestGetIPInfoIgnoresShortRoutes(t *testing.T) {
lookup("2001:db8::1", "2001:db8::/32", 64502)
}
// TestGetIPInfoNeedsNoWHOISRefreshForOriginZero checks that the IP lookup asks
// for no WHOIS lookup when the address's route has origin 0, no known origin,
// and still asks for one when its origin AS has no WHOIS data.
func TestGetIPInfoNeedsNoWHOISRefreshForOriginZero(t *testing.T) {
t.Parallel()
db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
writeOriginRoutes(t, db, map[string]int{"10.0.0.0/8": 0, "11.0.0.0/8": 64500})
for ip, want := range map[string]bool{"10.0.0.1": false, "11.0.0.1": true} {
info, err := db.GetIPInfo(ip)
if err != nil {
t.Fatalf("GetIPInfo(%s): %v", ip, err)
}
if info.NeedsWHOISRefresh != want {
t.Errorf("GetIPInfo(%s) asks for a WHOIS lookup of AS%d: %t, want %t",
ip, info.ASN, info.NeedsWHOISRefresh, want)
}
}
}
// TestLiveRouteTablesHoldOneRoutePerPrefixAndPeer checks that each live route
// table refuses a second route from a peer for a prefix, even with another
// origin AS. An upsert updates the peer's route before it would insert one, so
+9 -3
View File
@@ -73,6 +73,9 @@ type prefixUpdate struct {
messageType string
timestamp time.Time
path []int
// endsInSet says the path ends in an AS set of several members, so it
// gives no origin.
endsInSet bool
}
// NewPrefixHandler creates a new batched prefix tracking handler
@@ -170,6 +173,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
messageType: "announcement",
timestamp: timestamp,
path: path,
endsInSet: len(msg.Path.Set) > 0,
})
// Record announcement in metrics
if h.metrics != nil {
@@ -297,9 +301,11 @@ func (h *PrefixHandler) flushBatch() {
prefixesToUpdate[update.prefix] = update.timestamp
}
// An announcement whose path gives no origin is stored with origin 0
// (none).
if update.messageType == "announcement" {
// An announcement is stored when its path gives an origin, or when it
// ends in an AS set of several members: that route is stored with
// origin 0 (none). An announcement with an empty path is not stored.
if update.messageType == "announcement" &&
(update.originASN > 0 || update.endsInSet) {
// Create live route for batch upsert
route := h.createLiveRoute(update)
if route != nil {
@@ -315,6 +315,19 @@ func TestPrefixHandlerStoresNoOriginForAPathEndingInASet(t *testing.T) {
}
}
// TestPrefixHandlerDoesNotStoreAnAnnouncementWithAnEmptyPath announces a
// prefix with an empty AS path, which gives no origin and does not end in an
// AS set, so no live route is stored.
func TestPrefixHandlerDoesNotStoreAnAnnouncementWithAnEmptyPath(t *testing.T) {
t.Parallel()
msg := peerAnnouncement(testPeerIP, testV4Prefix)
msg.Path = ristypes.ASPath{}
db := handleBatches(t, [][]*ristypes.RISMessage{{msg}})
assertPeers(t, db, map[string][]string{testV4Prefix: nil})
}
// TestStateMessagesGoToThePrefixHandlerNotThePeerHandler checks that STATE
// messages go to the prefix handler, which acts on a down, and not to the peer
// handler, which would record a peer whose session never comes up as active.
+57
View File
@@ -331,6 +331,63 @@ func TestPrefixDetailShowsOriginNames(t *testing.T) {
}
}
// TestPagesSayOriginZeroIsNoKnownOrigin stores a route with origin 0, as the
// prefix handler stores a route whose AS path ends in an AS set of several
// members, and checks that the prefix page and the prefix length page say it
// has no known origin, without a link to AS0 or the description of the
// reserved AS 0 (https://git.eeqj.de/sneak/routewatch/issues/75).
func TestPagesSayOriginZeroIsNoKnownOrigin(t *testing.T) {
t.Parallel()
db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
if err != nil {
t.Fatalf("database.New: %v", err)
}
t.Cleanup(func() { _ = db.Close() })
err = db.UpsertLiveRouteBatch([]*database.LiveRoute{{
ID: uuid.New(), Prefix: testPrefixV4, MaskLength: 24, IPVersion: 4,
OriginASN: 0, PeerIP: testPeerV4, ASPath: []int{64500, 64501, 64502},
NextHop: testPeerV4, LastUpdated: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC),
}})
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
reserved, ok := asinfo.Get(0)
if !ok {
t.Fatal("AS0 is not in the AS info data")
}
s := New(db, streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{})
for _, path := range []string{"/prefix/" + testPrefixV4, "/prefixlength/24"} {
rec := httptest.NewRecorder()
s.router.ServeHTTP(rec, httptest.NewRequestWithContext(
t.Context(), http.MethodGet, path, nil))
if rec.Code != http.StatusOK {
t.Fatalf("%s: status %d, want %d; body %s",
path, rec.Code, http.StatusOK, rec.Body)
}
page := rec.Body.String()
if !strings.Contains(page, "No known origin") {
t.Errorf("%s does not say the route has no known origin", path)
}
if strings.Contains(page, `href="/as/0"`) {
t.Errorf("%s links to AS0", path)
}
if strings.Contains(page, html.EscapeString(reserved.Description)) {
t.Errorf("%s shows AS0's description %q", path, reserved.Description)
}
}
}
// shortRequestDeadline is the deadline the tests below give a request. The
// server's 30-second limit and the stats handlers' 4-second limit inherit it,
// so the tests do not wait for either.
+32 -17
View File
@@ -64,6 +64,10 @@ const (
backpressureSlope = 2.0 // Slope for linear drop probability increase
)
// messageTypeNotification is the type of a BGP NOTIFICATION message, which ends
// the session with the peer.
const messageTypeNotification = "NOTIFICATION"
var (
errAlreadyRunning = errors.New("streamer already running")
errUnexpectedStatus = errors.New("unexpected status code")
@@ -665,8 +669,8 @@ func (s *Streamer) readStream(
}
// processLine passes one line of the stream to the raw handler, parses it,
// and dispatches an UPDATE or STATE message to the handlers. An error ends the
// connection.
// updates the peer set (updatePeers), and dispatches an UPDATE or STATE
// message to the handlers. An error ends the connection.
func (s *Streamer) processLine(ctx context.Context, line []byte) error {
// Update metrics with decompressed message size
s.updateMetrics(len(line))
@@ -711,6 +715,8 @@ func (s *Streamer) processLine(ctx context.Context, line []byte) error {
// Parse the timestamp
msg.ParsedTimestamp = time.Unix(int64(msg.Timestamp), 0).UTC()
peerCount := s.updatePeers(&msg)
// Process based on message type
switch msg.Type {
case ristypes.MessageTypeUpdate:
@@ -723,13 +729,6 @@ func (s *Streamer) processLine(ctx context.Context, line []byte) error {
// BGP keepalive messages - silently process
return nil
case "OPEN":
// BGP open messages - track peer as active
s.bgpPeersMu.Lock()
s.bgpPeers[msg.Peer] = true
peerCount := len(s.bgpPeers)
s.bgpPeersMu.Unlock()
s.metrics.SetBGPPeerCount(peerCount)
s.logger.Info("BGP session opened",
"peer", msg.Peer,
"peer_asn", msg.PeerASN,
@@ -737,14 +736,7 @@ func (s *Streamer) processLine(ctx context.Context, line []byte) error {
)
return nil
case "NOTIFICATION":
// BGP notification messages (session closed)
s.bgpPeersMu.Lock()
delete(s.bgpPeers, msg.Peer)
peerCount := len(s.bgpPeers)
s.bgpPeersMu.Unlock()
s.metrics.SetBGPPeerCount(peerCount)
case messageTypeNotification:
s.logger.Warn("BGP notification",
"peer", msg.Peer,
"peer_asn", msg.PeerASN,
@@ -766,6 +758,29 @@ func (s *Streamer) processLine(ctx context.Context, line []byte) error {
return s.dispatchMessage(ctx, &msg)
}
// updatePeers adds the peer msg came from to the peer set, or removes it if
// msg ends the peer's session: a NOTIFICATION, or a STATE message whose state
// is down. It records the size of the set as the BGP peer count and returns
// it.
func (s *Streamer) updatePeers(msg *ristypes.RISMessage) int {
sessionEnded := msg.Type == messageTypeNotification ||
(msg.Type == ristypes.MessageTypeState && msg.State == ristypes.StateDown)
s.bgpPeersMu.Lock()
if sessionEnded {
delete(s.bgpPeers, msg.Peer)
} else {
s.bgpPeers[msg.Peer] = true
}
peerCount := len(s.bgpPeers)
s.bgpPeersMu.Unlock()
s.metrics.SetBGPPeerCount(peerCount)
return peerCount
}
// dispatchMessage queues msg for each handler that wants it, unless
// queueMessage drops it for that handler.
func (s *Streamer) dispatchMessage(
@@ -518,6 +518,78 @@ func TestStreamReadsLongLine(t *testing.T) {
}
}
// TestPeerCountCountsEveryPeerHeardFrom has the server send an UPDATE from
// stateDownPeer and a KEEPALIVE from another peer, with no OPEN from either,
// and then a message that ends stateDownPeer's session: a NOTIFICATION, or a
// STATE message whose state is down. The peer count must be 2 after the first
// two messages and 1 after the third.
func TestPeerCountCountsEveryPeerHeardFrom(t *testing.T) {
t.Parallel()
const (
otherPeer = "192.0.2.2"
peerUpdateLine = `{"type":"ris_message","data":{"type":"UPDATE",` +
`"peer":"` + stateDownPeer + `",` +
`"peer_asn":"64496","timestamp":1700000000}}` + "\n"
otherPeerKeepaliveLine = `{"type":"ris_message","data":{"type":"KEEPALIVE",` +
`"peer":"` + otherPeer + `",` +
`"peer_asn":"64497","timestamp":1700000000}}` + "\n"
peerNotificationLine = `{"type":"ris_message","data":{"type":"NOTIFICATION",` +
`"peer":"` + stateDownPeer + `",` +
`"peer_asn":"64496","timestamp":1700000000}}` + "\n"
bothPeers = 2
)
sessionEndLines := map[string]string{
"NOTIFICATION": peerNotificationLine,
"STATE down": stateDownLine,
}
for name, sessionEndLine := range sessionEndLines {
t.Run(name, func(t *testing.T) {
t.Parallel()
tracker := metrics.New()
s := New(logger.New(), tracker)
streamLines(t, s, peerUpdateLine+otherPeerKeepaliveLine)
got := tracker.GetBGPPeerCount()
if got != bothPeers {
t.Fatalf("peer count is %d after messages from two peers, want %d",
got, bothPeers)
}
streamLines(t, s, sessionEndLine)
got = tracker.GetBGPPeerCount()
if got != 1 {
t.Errorf("peer count is %d after one of two peers' sessions "+
"ended, want 1", got)
}
})
}
}
// streamLines has a server send lines as its whole response and runs one
// stream of s against it.
func streamLines(t *testing.T, s *Streamer, lines string) {
t.Helper()
srv := httptest.NewServer(http.HandlerFunc(
func(w http.ResponseWriter, _ *http.Request) {
_, _ = io.WriteString(w, lines)
}))
defer srv.Close()
s.url = srv.URL
err := s.stream(context.Background())
if !errors.Is(err, errStreamEnded) {
t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
}
}
// checkReconnect runs a streamer against a fake RIS Live server. The first
// response is served by firstResponse; every later one carries one UPDATE
// line and then stays open. The streamer must come back for a second
+8
View File
@@ -281,9 +281,13 @@
<div class="origin-list">
{{range .Origins}}
<div class="origin-item">
{{if .ASN}}
<a href="/as/{{.ASN}}" class="as-link">AS{{.ASN}}</a>
{{if .Handle}} ({{.Handle}}){{end}}
{{if .Description}} {{.Description}}{{end}}
{{else}}
No known origin
{{end}}
<span style="color: #7f8c8d; margin-left: 10px;">{{.PeerCount}} peer{{if ne .PeerCount 1}}s{{end}}</span>
</div>
{{end}}
@@ -312,7 +316,11 @@
{{range .Routes}}
<tr>
<td>
{{if .OriginASN}}
<a href="/as/{{.OriginASN}}" class="as-link">AS{{.OriginASN}}</a>
{{else}}
No known origin
{{end}}
</td>
<td class="peer-ip">{{.PeerIP}}</td>
<td class="as-path">{{range $i, $as := .ASPathWithHandle}}{{if $i}} → {{end}}<a href="/as/{{$as.ASN}}" class="as-link">{{if $as.Handle}}{{$as.Handle}}{{else}}AS{{$as.ASN}}{{end}}</a>{{end}}</td>
+4
View File
@@ -172,9 +172,13 @@
<td><a href="{{ .Prefix | prefixURL }}" class="prefix-link">{{ .Prefix }}</a></td>
<td class="age">{{ .Age }}</td>
<td>
{{ if .OriginASN }}
<a href="/as/{{ .OriginASN }}" class="as-link">
AS{{ .OriginASN }}{{ if .OriginASDescription }} ({{ .OriginASDescription }}){{ end }}
</a>
{{ else }}
No known origin
{{ end }}
</td>
</tr>
{{ end }}