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}}
There are more than 5,000 changes. Showing the newest 5,000.
+There are more than {{.MaxOriginChanges}} changes. Showing the newest {{.MaxOriginChanges}}.
{{end}}