From 56cfc7e554cff68664da153753f35eb5694e71fa Mon Sep 17 00:00:00 2001 From: sneak Date: Thu, 5 Nov 2026 04:33:51 +0000 Subject: [PATCH] Fix the origin-changes review findings Hold the 100,000-row cap after a failed write's rows are put back, dropping and counting the oldest rows past it. Stop counts the unwritten origin changes. The prefix handler's per-write origin change limit is a field, so a test checks that one write leaves the rest. The server and README now say where origin changes come from. README, TODO and schema say 72 hours or 500 rows, whichever keeps more. The prefix page and TODO list the origin change columns and say what the peer count counts. The page takes its 5,000-row note from the constant. Model: deepseek-v4-pro --- README.md | 29 ++++++---- TODO.md | 17 +++--- internal/database/schema.sql | 5 +- internal/routetable/originchanges.go | 31 ++++++++--- internal/routetable/originchanges_test.go | 54 +++++++++++++++++++ internal/routewatch/prefixhandler.go | 51 ++++++++++-------- .../routewatch/prefixhandler_internal_test.go | 15 +++--- ...efixhandler_originchanges_internal_test.go | 43 +++++++++++++++ internal/server/handlers.go | 5 +- .../prefix_originchanges_internal_test.go | 4 +- internal/server/server.go | 4 +- internal/templates/prefix_detail.html | 4 +- 12 files changed, 199 insertions(+), 63 deletions(-) diff --git a/README.md b/README.md index bb61bf2..5a5cd36 100644 --- a/README.md +++ b/README.md @@ -143,10 +143,13 @@ and origin AS can have two kept routes, one or none. its origin ASes and every peer that carries it. Only the kept routes show an AS path, a next hop and a time. The next hop is the router the peer sends the prefix's traffic to. RouteWatch records the peer itself as the next hop. The - page also lists the prefix's origin changes, newest first: each time an origin - AS gained its first peer or lost its last, with the time, announced or - withdrawn, the origin AS and the peer count. It shows at most the newest - 5,000. + page also lists the prefix's origin changes, newest first. It shows at most + the newest 5,000. The peer count counts the peers of every origin AS that + carries the prefix. Each row shows: + - the time + - announced or withdrawn + - the origin AS + - the peer count - `/prefixlength/{length}`: up to 500 IPv4 prefixes with that mask length, picked at random. Each shows its origin AS and the age of its newest kept route. When peers give a prefix different origin ASes, the page shows the one @@ -625,8 +628,9 @@ The parts: messages. It gives each message to the handlers that want it, through one queue per handler. - The prefix handler keeps the routing table in memory current. Every second it - writes the table's changes to the `routes` table of the database. It also - stores each new prefix with the time it was first seen. + writes the table's changes to the `routes` table and the table's origin + changes to the `origin_changes` table of the database. It also stores each new + prefix with the time it was first seen. - The AS handler records every AS seen in an AS path in the `asns` table, with the times it was first and last seen. - The peer handler records every peer in the `bgp_peers` table. @@ -643,11 +647,14 @@ The parts: - The database maintainer checkpoints the database every 5 seconds: it folds the write-ahead log, where SQLite first writes changes, into the database file. It reclaims free space every 10 minutes. Every hour it refreshes the statistics - SQLite uses to plan its queries, and prunes the origin changes table. + SQLite uses to plan its queries, and prunes the `origin_changes` table. A + prune keeps each prefix's rows from the last 72 hours, or its newest 500, + whichever keeps more. - The HTTP server serves the pages and the API. Every route it shows comes from - the routing table in memory. WHOIS details and first-seen times come from the - database. AS names come from the database or from the AS list built into - RouteWatch. The AS name search reads the AS list built into RouteWatch. + the routing table in memory. WHOIS details, first-seen times and origin + changes come from the database. AS names come from the database or from the AS + list built into RouteWatch. The AS name search reads the AS list built into + RouteWatch. ### The routing table @@ -758,6 +765,8 @@ What a stop can lose: changed pairs. A full-table peer is a peer that carries the whole routing table. Its session going down changes about a million pairs, which take about a minute to write. A stop in that minute leaves the rest unwritten. +- The same last write takes at most 20,000 of the routing table's origin + changes. Origin changes past them are lost. - A withdrawal, or a session going down, that is lost this way leaves its routes in the `routes` table. The next start reads them back. They stay until the peer sends another update for the prefix, or its session goes down again. diff --git a/TODO.md b/TODO.md index e3353ad..538cf20 100644 --- a/TODO.md +++ b/TODO.md @@ -56,13 +56,16 @@ problem the run finds. # Completed Steps -- 2026-11-05: the prefix page lists the prefix's origin changes, newest first: - each time an origin AS gained its first peer or lost its last, with the time, - announced or withdrawn, the origin AS and the peer count. The routing table - records each change, and the prefix handler writes them to a new - `origin_changes` table. The database maintainer prunes each prefix to its - newest 500 rows and 72 hours. The page shows at most the newest 5,000 (closes - #161) +- 2026-11-05: the prefix page lists the prefix's origin changes, newest first. + The routing table records each change, and the prefix handler writes them to a + new `origin_changes` table. The database maintainer prunes each prefix to its + newest 500 rows or its rows from the last 72 hours, whichever keeps more. The + page shows at most the newest 5,000 (closes #161). The peer count counts the + peers of every origin AS. Each row shows: + - the time + - announced or withdrawn + - the origin AS + - the peer count - 2026-11-03: the streamer drops an update for a handler only when that handler's queue is full. It used to drop a random share of the updates once the queue was half full. The prefix handler's queue now holds 100,000 diff --git a/internal/database/schema.sql b/internal/database/schema.sql index edb556a..4730298 100644 --- a/internal/database/schema.sql +++ b/internal/database/schema.sql @@ -123,8 +123,9 @@ CREATE TABLE IF NOT EXISTS route_peers ( -- The origin changes of the routing table: each time a prefix and origin AS -- gained its first peer or lost its last. The prefix handler writes them, and --- the database maintainer prunes each prefix to its newest 500 rows and 72 --- hours. time is whole seconds since 1970, UTC. +-- the database maintainer prunes each prefix to its newest 500 rows or its +-- rows from the last 72 hours, whichever keeps more. time is whole seconds +-- since 1970, UTC. CREATE TABLE IF NOT EXISTS origin_changes ( id INTEGER PRIMARY KEY, prefix TEXT NOT NULL, diff --git a/internal/routetable/originchanges.go b/internal/routetable/originchanges.go index c1361cc..6cd3b34 100644 --- a/internal/routetable/originchanges.go +++ b/internal/routetable/originchanges.go @@ -7,7 +7,8 @@ import ( ) // maxWaitingOriginChanges is the most origin changes the table keeps waiting -// for the prefix handler to write. A change past it is dropped and counted. +// for the prefix handler to write. The oldest change past it is dropped and +// counted. const maxWaitingOriginChanges = 100000 // OriginChange is one change of a prefix and origin AS pair: the origin AS @@ -39,13 +40,8 @@ func (t *Table) recordOriginChange( Time: t.now(), } - if len(t.originChanges) == maxWaitingOriginChanges { - // The oldest change is dropped to make room. - t.originChanges = t.originChanges[1:] - t.originChangesDropped++ - } - t.originChanges = append(t.originChanges, change) + t.trimOriginChanges() } // TakeOriginChanges returns at most limit of the origin changes that wait to @@ -67,12 +63,31 @@ func (t *Table) TakeOriginChanges(limit int) ([]OriginChange, int) { // PutBackOriginChanges records again what one TakeOriginChanges call returned, // for a caller that could not write it. The next TakeOriginChanges returns -// those changes first, before the ones that arrived since. +// those changes first, before the ones that arrived since. The oldest changes +// past the cap are dropped and counted. func (t *Table) PutBackOriginChanges(changes []OriginChange) { t.mu.Lock() defer t.mu.Unlock() t.originChanges = append(changes, t.originChanges...) + t.trimOriginChanges() +} + +// trimOriginChanges drops the oldest origin changes past the cap and counts +// them. The caller holds the lock. +func (t *Table) trimOriginChanges() { + if extra := len(t.originChanges) - maxWaitingOriginChanges; extra > 0 { + t.originChangesDropped += extra + t.originChanges = t.originChanges[extra:] + } +} + +// OriginChangeCount returns how many origin changes wait for TakeOriginChanges. +func (t *Table) OriginChangeCount() int { + t.mu.RLock() + defer t.mu.RUnlock() + + return len(t.originChanges) } // prefixPeerCount returns how many peers carry prefix, counting every origin diff --git a/internal/routetable/originchanges_test.go b/internal/routetable/originchanges_test.go index b6b97ed..493cd08 100644 --- a/internal/routetable/originchanges_test.go +++ b/internal/routetable/originchanges_test.go @@ -206,6 +206,60 @@ func TestOriginChangesDropPastTheLimit(t *testing.T) { } } +// TestOriginChangesPutBackTrimsToTheLimit checks that putting back what a +// failed write took, after new changes filled the table again, drops the +// oldest changes past the cap and counts them, so the cap still holds. The +// clock advances one second per change, so a change's time says its age. +func TestOriginChangesPutBackTrimsToTheLimit(t *testing.T) { + t.Parallel() + + const ( + limit = 100000 + taken = 20000 + added = 20000 + ) + + now := time.Unix(0, 0) + table := routetable.New(routetable.Params{ + Now: func() time.Time { + now = now.Add(time.Second) + + return now + }, + }) + addr := netip.MustParseAddr("10.0.0.0") + + announce := func(n int) { + for range n { + table.Announce(netip.PrefixFrom(addr, 32), asA, routeOf(peerA, asA, 10)) + addr = addr.Next() + } + } + + announce(limit) + + takenChanges, _ := table.TakeOriginChanges(taken) + + announce(added) + + table.PutBackOriginChanges(takenChanges) + + changes, dropped := table.TakeOriginChanges(math.MaxInt) + if dropped != added { + t.Errorf("dropped = %d, want %d", dropped, added) + } + + if len(changes) != limit { + t.Errorf("len(changes) = %d, want %d", len(changes), limit) + } + + // The put-back rows are the oldest, so they are the ones dropped. The + // first remaining change is the one that followed them. + if want := time.Unix(int64(added+1), 0); !changes[0].Time.Equal(want) { + t.Errorf("first change's time = %v, want %v", changes[0].Time, want) + } +} + // checkOriginChanges checks that TakeOriginChanges, taking every change, // returns want. func checkOriginChanges( diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index bb36a32..f701ff3 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -99,6 +99,9 @@ type PrefixHandler struct { // tableChangesPerWrite is maxTableChangesPerWrite, except in tests that // give a smaller one. tableChangesPerWrite int + // originChangesPerWrite is maxOriginChangesPerWrite, except in tests that + // give a smaller one. + originChangesPerWrite int // Batching. mu guards batch, writing and writingTableChanges: the // flush goroutine takes them under mu and writes them after releasing @@ -194,17 +197,18 @@ func newPrefixHandler( batchTimeout time.Duration, tableChangesPerWrite int, ) *PrefixHandler { h := &PrefixHandler{ - db: db, - logger: logger, - table: table, - batch: make([]seenPrefix, 0, prefixBatchSize), - batchMaxSize: prefixBatchMaxSize, - tableChangesPerWrite: tableChangesPerWrite, - batchFullCh: make(chan struct{}, 1), - stopCh: make(chan context.Context, 1), - done: make(chan struct{}), - heldBack: make(map[string][]*ristypes.RISMessage), - tablePeersDown: table.PeersDown, + db: db, + logger: logger, + table: table, + batch: make([]seenPrefix, 0, prefixBatchSize), + batchMaxSize: prefixBatchMaxSize, + tableChangesPerWrite: tableChangesPerWrite, + originChangesPerWrite: maxOriginChangesPerWrite, + batchFullCh: make(chan struct{}, 1), + stopCh: make(chan context.Context, 1), + done: make(chan struct{}), + heldBack: make(map[string][]*ristypes.RISMessage), + tablePeersDown: table.PeersDown, } h.batchTaken.L = &h.mu h.heldTaken.L = &h.mu @@ -272,13 +276,13 @@ func (h *PrefixHandler) HeldBackCount() int { // back meanwhile to be applied. It gives up this wait once ctx ends. Then Stop // stops the flush goroutine and waits for it until ctx ends. The flush // goroutine makes one last flush before it returns. That flush writes at most -// tableChangesPerWrite of the routing table's changes, and the prefixes of the -// updates the handler holds. Once ctx has ended the flush goroutine skips that -// last flush, and a write still running is lost when the process exits. A -// failed write of the table's changes is not retried either. Stop logs how -// many updates and table changes are not written. It also logs how many -// held-back messages are not applied. The caller calls Stop once its last -// HandleMessage has returned. +// tableChangesPerWrite of the routing table's changes, originChangesPerWrite of +// its origin changes, and the prefixes of the updates the handler holds. Once +// ctx has ended the flush goroutine skips that last flush, and a write still +// running is lost when the process exits. A failed write of the table's +// changes is not retried either. Stop logs how many updates, table changes and +// origin changes are not written. It also logs how many held-back messages are +// not applied. The caller calls Stop once its last HandleMessage has returned. func (h *PrefixHandler) Stop(ctx context.Context) { peerDownsDone := make(chan struct{}) @@ -305,21 +309,24 @@ func (h *PrefixHandler) Stop(ctx context.Context) { heldMessages := h.heldCount h.mu.Unlock() - if unwritten == 0 && tableChanges == 0 && heldMessages == 0 { + originChanges := h.table.OriginChangeCount() + + if unwritten == 0 && tableChanges == 0 && heldMessages == 0 && + originChanges == 0 { return } if ctx.Err() != nil { h.logger.Error("Stop deadline passed before the prefix handler wrote "+ "its updates", "updates", unwritten, "table_changes", tableChanges, - "held_messages", heldMessages) + "held_messages", heldMessages, "origin_changes", originChanges) return } h.logger.Error("The prefix handler stopped before it wrote everything", "updates", unwritten, "table_changes", tableChanges, - "held_messages", heldMessages) + "held_messages", heldMessages, "origin_changes", originChanges) } // applyMessage applies msg. The caller holds mu. A peer's session going down @@ -618,7 +625,7 @@ func (h *PrefixHandler) writeTableChanges() { // that a later write writes them again. A dropped change is logged as a // warning. func (h *PrefixHandler) writeOriginChanges() { - changes, dropped := h.table.TakeOriginChanges(maxOriginChangesPerWrite) + changes, dropped := h.table.TakeOriginChanges(h.originChangesPerWrite) if dropped > 0 { h.logger.Warn("Origin changes dropped", "dropped", dropped) diff --git a/internal/routewatch/prefixhandler_internal_test.go b/internal/routewatch/prefixhandler_internal_test.go index d7ce0a7..2b3d245 100644 --- a/internal/routewatch/prefixhandler_internal_test.go +++ b/internal/routewatch/prefixhandler_internal_test.go @@ -327,13 +327,14 @@ func prefixHandlerWithoutTimer(store database.Store) *PrefixHandler { table := routetable.New(routetable.Params{}) return &PrefixHandler{ - db: store, - logger: logger.New(), - table: table, - batchMaxSize: prefixBatchMaxSize, - tableChangesPerWrite: maxTableChangesPerWrite, - heldBack: make(map[string][]*ristypes.RISMessage), - tablePeersDown: table.PeersDown, + db: store, + logger: logger.New(), + table: table, + batchMaxSize: prefixBatchMaxSize, + tableChangesPerWrite: maxTableChangesPerWrite, + originChangesPerWrite: maxOriginChangesPerWrite, + heldBack: make(map[string][]*ristypes.RISMessage), + tablePeersDown: table.PeersDown, } } diff --git a/internal/routewatch/prefixhandler_originchanges_internal_test.go b/internal/routewatch/prefixhandler_originchanges_internal_test.go index 7bc88ab..c2d769d 100644 --- a/internal/routewatch/prefixhandler_originchanges_internal_test.go +++ b/internal/routewatch/prefixhandler_originchanges_internal_test.go @@ -3,6 +3,7 @@ package routewatch import ( "errors" "net/netip" + "slices" "testing" "git.eeqj.de/sneak/routewatch/internal/database" @@ -82,3 +83,45 @@ func TestPrefixHandlerWritesAFailedOriginChangeWriteAgain(t *testing.T) { got, testASNB) } } + +// originChangesWriteSizesStore is a real database that records how many +// changes each write of the origin changes holds. +type originChangesWriteSizesStore struct { + *database.Database + + counts []int +} + +func (s *originChangesWriteSizesStore) WriteOriginChanges( + changes []routetable.OriginChange, +) error { + s.counts = append(s.counts, len(changes)) + + return s.Database.WriteOriginChanges(changes) +} + +// TestPrefixHandlerWritesManyOriginChangesOverSeveralFlushes has more origin +// changes than one write takes, then flushes three times. The first write must +// hold one write's worth, the second the rest, and the third nothing. The +// handler's limit is small. +func TestPrefixHandlerWritesManyOriginChangesOverSeveralFlushes(t *testing.T) { + t.Parallel() + + const perWrite = 10 + + db := newTestDatabase(t) + store := &originChangesWriteSizesStore{Database: db} + h := prefixHandlerWithoutTimer(store) + h.originChangesPerWrite = perWrite + + announceManyPrefixes(h.table, perWrite*3/2) + + h.flushBatch() + h.flushBatch() + h.flushBatch() + + want := []int{perWrite, perWrite*3/2 - perWrite} + if !slices.Equal(store.counts, want) { + t.Errorf("WriteOriginChanges held %v changes, want %v", store.counts, want) + } +} diff --git a/internal/server/handlers.go b/internal/server/handlers.go index be9d897..815ad1f 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -1637,7 +1637,9 @@ type prefixDetailData struct { OriginCount int OriginChanges []routetable.OriginChange MoreOriginChanges bool - CSPNonce string + // MaxOriginChanges is the most origin changes the page shows. + MaxOriginChanges int + CSPNonce string } // newPrefixDetailData builds the prefix detail page's data. @@ -1655,6 +1657,7 @@ func newPrefixDetailData( OriginCount: len(originInfos), OriginChanges: changes, MoreOriginChanges: moreChanges, + MaxOriginChanges: maxOriginChangesOnPage, CSPNonce: nonce, } } diff --git a/internal/server/prefix_originchanges_internal_test.go b/internal/server/prefix_originchanges_internal_test.go index 58dbbb7..1d88359 100644 --- a/internal/server/prefix_originchanges_internal_test.go +++ b/internal/server/prefix_originchanges_internal_test.go @@ -142,7 +142,7 @@ func TestPrefixDetailSaysWhenMoreThan5000Changes(t *testing.T) { page := rec.Body.String() - if !strings.Contains(page, "more than 5,000 changes") { - t.Errorf("page does not say there are more than 5,000 changes") + if !strings.Contains(page, "more than 5000 changes") { + t.Errorf("page does not say there are more than 5000 changes") } } diff --git a/internal/server/server.go b/internal/server/server.go index 7bf4758..13a8a67 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -7,8 +7,8 @@ // - the AS name search // // Every route it shows comes from the routing table in memory. AS names, -// WHOIS details and the times a prefix and an AS were first seen come from -// the database. +// WHOIS details, the times a prefix and an AS were first seen, and a prefix's +// origin changes come from the database. package server import ( diff --git a/internal/templates/prefix_detail.html b/internal/templates/prefix_detail.html index a1f8824..7afb629 100644 --- a/internal/templates/prefix_detail.html +++ b/internal/templates/prefix_detail.html @@ -297,13 +297,13 @@ {{end}} -{{end}} + {{end}} {{if .OriginChanges}}

Origin Changes

{{if .MoreOriginChanges}} -

There are more than 5,000 changes. Showing the newest 5,000.

+

There are more than {{.MaxOriginChanges}} changes. Showing the newest {{.MaxOriginChanges}}.

{{end}}