diff --git a/README.md b/README.md index 5a5cd36..d25bdfd 100644 --- a/README.md +++ b/README.md @@ -143,13 +143,15 @@ 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. 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: + page also lists the prefix's origin changes, newest first. An origin change is + an AS starting to announce the prefix, when its first peer carries it, or + stopping, when its last peer no longer does. It shows at most the newest + 5,000. Each row shows: - the time - announced or withdrawn - the origin AS - - the peer count + - the peer count: how many peers carry the prefix just after the change, + with any origin AS - `/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 @@ -765,8 +767,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. +- The prefix handler's last write of origin changes takes at most 20,000. The + rest 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 538cf20..e3c6751 100644 --- a/TODO.md +++ b/TODO.md @@ -60,12 +60,15 @@ problem the run finds. 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: + page shows at most the newest 5,000. Each row shows: - the time - announced or withdrawn - the origin AS - the peer count + + A row's peer count is how many peers carry the prefix just after the change, + with any origin AS (closes #161). + - 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/interface.go b/internal/database/interface.go index 717d26a..4186736 100644 --- a/internal/database/interface.go +++ b/internal/database/interface.go @@ -1,6 +1,7 @@ // Package database provides SQLite storage for BGP routing data: ASNs, -// prefixes, announcements, peerings and BGP peers. Its routes table holds a -// copy of the whole routing table that internal/routetable keeps in memory. +// prefixes, announcements, peerings, BGP peers and origin changes. Its routes +// table holds a copy of the whole routing table that internal/routetable +// keeps in memory. package database import ( diff --git a/internal/database/originchanges.go b/internal/database/originchanges.go index 6967d64..62bc957 100644 --- a/internal/database/originchanges.go +++ b/internal/database/originchanges.go @@ -9,13 +9,13 @@ import ( "git.eeqj.de/sneak/routewatch/internal/routetable" ) -// maxOriginChangesPerPrefix is how many origin changes the prune keeps for -// each prefix: the newest ones. A prefix with more has its older rows deleted. -const maxOriginChangesPerPrefix = 500 +// minOriginChangesPerPrefix is how many of a prefix's newest origin changes +// the prune always keeps, whatever their age. +const minOriginChangesPerPrefix = 500 // originChangesRetention is how long the prune keeps an origin change: rows // older than this are deleted, unless they are among the prefix's newest -// maxOriginChangesPerPrefix. +// minOriginChangesPerPrefix. const originChangesRetention = 72 * time.Hour // The statements of the origin_changes table (schema.sql). @@ -122,13 +122,13 @@ func (d *Database) GetOriginChanges( } // PruneOriginChanges deletes, for each prefix with more than -// maxOriginChangesPerPrefix rows, the rows older than originChangesRetention -// that are not among its newest maxOriginChangesPerPrefix. It finds those +// minOriginChangesPerPrefix rows, the rows older than originChangesRetention +// that are not among its newest minOriginChangesPerPrefix. It finds those // prefixes on a reading connection and deletes one prefix per write, so no // write holds the lock for long. func (d *Database) PruneOriginChanges(ctx context.Context) error { rows, err := d.db.QueryContext(ctx, prunePrefixesSQL, - maxOriginChangesPerPrefix) + minOriginChangesPerPrefix) if err != nil { return fmt.Errorf("failed to find prefixes to prune: %w", err) } @@ -158,7 +158,7 @@ func (d *Database) PruneOriginChanges(ctx context.Context) error { d.lock("PruneOriginChanges") _, err := d.writer.ExecContext(ctx, pruneOriginChangesSQL, - prefix, cutoff, prefix, maxOriginChangesPerPrefix-1) + prefix, cutoff, prefix, minOriginChangesPerPrefix-1) d.unlock() diff --git a/internal/database/originchanges_internal_test.go b/internal/database/originchanges_internal_test.go index 9404f16..0dd971c 100644 --- a/internal/database/originchanges_internal_test.go +++ b/internal/database/originchanges_internal_test.go @@ -48,44 +48,30 @@ func TestOriginChangesWriteThenRead(t *testing.T) { } } -// TestOriginChangesPrune checks that the prune keeps every row newer than 72 -// hours, keeps the newest 500 rows of each prefix, deletes the rest, and -// leaves other prefixes alone. +// TestOriginChangesPrune checks that the prune keeps, for a prefix with more +// than 500 rows, its newest 500 rows and every row newer than 72 hours, and +// deletes the rest. It checks which rows survive by their times, not only how +// many. func TestOriginChangesPrune(t *testing.T) { t.Parallel() db := newRoutesTestDatabase(t) oldPrefix := netip.MustParsePrefix(tablePrefixV4) - newPrefix := netip.MustParsePrefix(tablePrefixShortV4) + straddling := netip.MustParsePrefix(tablePrefixShortV4) other := netip.MustParsePrefix("10.20.0.0/16") - now := time.Now() - - changes := make([]routetable.OriginChange, 0, 1201) + // Whole seconds, so the stored times round-trip exactly. + now := time.Now().Truncate(time.Second) // 600 rows for oldPrefix, all older than 72 hours, oldest first. The // prune keeps the newest 500 and deletes the oldest 100. - for i := range 600 { - changes = append(changes, routetable.OriginChange{ - Prefix: oldPrefix, OriginASN: asA, Announced: true, PeerCount: 1, - Time: now.Add(-100 * time.Hour).Add(time.Duration(i) * time.Minute), - }) - } - - // 600 rows for newPrefix, all newer than 72 hours, oldest first. The - // prune keeps all of them. - for i := range 600 { - changes = append(changes, routetable.OriginChange{ - Prefix: newPrefix, OriginASN: asA, Announced: true, PeerCount: 1, - Time: now.Add(-71 * time.Hour).Add(time.Duration(i) * time.Minute), - }) - } - + changes := rows(oldPrefix, now.Add(-100*time.Hour), 600) + // 600 rows for straddling: 400 older than 72 hours, then 200 newer. The + // prune keeps all 200 newer and the newest 300 of the 400 older. + changes = append(changes, rows(straddling, now.Add(-100*time.Hour), 400)...) + changes = append(changes, rows(straddling, now.Add(-71*time.Hour), 200)...) // One row for other, which the prune leaves alone. - changes = append(changes, routetable.OriginChange{ - Prefix: other, OriginASN: asA, Announced: true, PeerCount: 1, - Time: now.Add(-100 * time.Hour), - }) + changes = append(changes, rows(other, now.Add(-100*time.Hour), 1)...) err := db.WriteOriginChanges(changes) if err != nil { @@ -97,22 +83,43 @@ func TestOriginChangesPrune(t *testing.T) { t.Fatalf("PruneOriginChanges: %v", err) } - gotOld, err := db.GetOriginChanges(t.Context(), oldPrefix, 1000) - if err != nil { - t.Fatalf("GetOriginChanges: %v", err) + oldTimes := survivorTimes(t, db, oldPrefix, 1000) + if len(oldTimes) != 500 { + t.Errorf("GetOriginChanges(oldPrefix) = %d rows, want 500", len(oldTimes)) } - if len(gotOld) != 500 { - t.Errorf("GetOriginChanges(oldPrefix) = %d rows, want 500", len(gotOld)) + // The newest row survives, and the oldest survivor is the 100th oldest, + // so a prune that kept the oldest 500 instead would fail. + wantNewest := now.Add(-100 * time.Hour).Add(599 * time.Minute) + if !oldTimes[0].Equal(wantNewest) { + t.Errorf("the newest surviving row's time = %v, want %v", + oldTimes[0], wantNewest) } - gotNew, err := db.GetOriginChanges(t.Context(), newPrefix, 1000) - if err != nil { - t.Fatalf("GetOriginChanges: %v", err) + wantOldest := now.Add(-100 * time.Hour).Add(100 * time.Minute) + if !oldTimes[499].Equal(wantOldest) { + t.Errorf("the oldest surviving row's time = %v, want %v", + oldTimes[499], wantOldest) } - if len(gotNew) != 600 { - t.Errorf("GetOriginChanges(newPrefix) = %d rows, want 600", len(gotNew)) + straddlingTimes := survivorTimes(t, db, straddling, 1000) + if len(straddlingTimes) != 500 { + t.Errorf("GetOriginChanges(straddling) = %d rows, want 500", + len(straddlingTimes)) + } + + // The newest of the newer-than-72-hours rows survives, and the oldest + // survivor is the 100th oldest of the older-than-72-hours rows. + wantNewest = now.Add(-71 * time.Hour).Add(199 * time.Minute) + if !straddlingTimes[0].Equal(wantNewest) { + t.Errorf("the newest surviving row's time = %v, want %v", + straddlingTimes[0], wantNewest) + } + + wantOldest = now.Add(-100 * time.Hour).Add(100 * time.Minute) + if !straddlingTimes[499].Equal(wantOldest) { + t.Errorf("the oldest surviving row's time = %v, want %v", + straddlingTimes[499], wantOldest) } gotOther, err := db.GetOriginChanges(t.Context(), other, 10) @@ -125,6 +132,41 @@ func TestOriginChangesPrune(t *testing.T) { } } +// rows returns n origin changes for prefix, at times from plus one minute per +// step, oldest first. +func rows(prefix netip.Prefix, from time.Time, n int) []routetable.OriginChange { + changes := make([]routetable.OriginChange, 0, n) + + for i := range n { + changes = append(changes, routetable.OriginChange{ + Prefix: prefix, OriginASN: asA, Announced: true, PeerCount: 1, + Time: from.Add(time.Duration(i) * time.Minute), + }) + } + + return changes +} + +// survivorTimes returns the times of prefix's rows after the prune, newest +// first. +func survivorTimes( + t *testing.T, db *Database, prefix netip.Prefix, limit int, +) []time.Time { + t.Helper() + + rows, err := db.GetOriginChanges(t.Context(), prefix, limit) + if err != nil { + t.Fatalf("GetOriginChanges: %v", err) + } + + times := make([]time.Time, len(rows)) + for i, row := range rows { + times[i] = row.Time + } + + return times +} + // TestOriginChangesSurviveRouteDeletion checks that deleting a routes row // leaves the prefix's origin_changes rows. func TestOriginChangesSurviveRouteDeletion(t *testing.T) { diff --git a/internal/routetable/originchanges_test.go b/internal/routetable/originchanges_test.go index 493cd08..cd3fa35 100644 --- a/internal/routetable/originchanges_test.go +++ b/internal/routetable/originchanges_test.go @@ -206,6 +206,33 @@ func TestOriginChangesDropPastTheLimit(t *testing.T) { } } +// TestOriginChangesTakeResetsTheDroppedCount checks that a TakeOriginChanges +// that reports dropped rows leaves the next one reporting none, so each drop +// is logged once, not on every later write. +func TestOriginChangesTakeResetsTheDroppedCount(t *testing.T) { + t.Parallel() + + const limit = 100000 + + table := routetable.New(routetable.Params{}) + addr := netip.MustParseAddr("10.0.0.0") + + for range limit + 1 { + table.Announce(netip.PrefixFrom(addr, 32), asA, routeOf(peerA, asA, 10)) + addr = addr.Next() + } + + _, dropped := table.TakeOriginChanges(limit) + if dropped != 1 { + t.Fatalf("dropped = %d, want 1", dropped) + } + + _, dropped = table.TakeOriginChanges(0) + if dropped != 0 { + t.Errorf("the next take's dropped = %d, want 0", dropped) + } +} + // 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 diff --git a/internal/routetable/routetable.go b/internal/routetable/routetable.go index 983d7a3..3455db9 100644 --- a/internal/routetable/routetable.go +++ b/internal/routetable/routetable.go @@ -287,8 +287,9 @@ func (t *Table) PeersDown(peers []string) { // Load puts one prefix and origin AS pair that the database holds into the // table, as startup does with each row it reads. The change's peers join the // pair's peer set, and its routes are kept as an announcement's would be. -// Unlike an update, Load records no change, so TakeChanges does not hand the -// pair back to be written again. Loading a pair twice changes nothing. +// Unlike an update, Load records no change and no origin change, so +// TakeChanges does not hand the pair back to be written again. Loading a pair +// twice changes nothing. // // The caller makes sure that every peer change names, in its peer set or in // its routes, has a number in Params.PeerNumbers. @@ -479,6 +480,7 @@ func (t *Table) removeOrigin(prefix netip.Prefix, o *origin) { // r in full when it is one of the two newest routes of o. The peer's earlier // route, if o keeps one, is replaced: it is no longer the peer's current route. // The pair is recorded as changed only when its peer set or kept routes change. +// When r is the origin's first peer, an announced origin change is recorded. func (t *Table) join(prefix netip.Prefix, o *origin, r route) { if slices.ContainsFunc(o.routes, r.equal) { // The peer announced its kept route again. @@ -565,7 +567,8 @@ func (t *Table) dropKeptRoute(o *origin, peer uint) bool { // recordChange records that the pair of prefix and the AS of o changed, and // brings the count of origins without a kept route up to date. // wasWithoutKeptRoute says whether o was such an origin before the change. An -// origin that no peer carries any more is removed. +// origin that no peer carries any more is removed, and a withdrawn origin +// change is recorded. func (t *Table) recordChange( prefix netip.Prefix, o *origin, wasWithoutKeptRoute bool, ) { diff --git a/internal/routewatch/batchwrite_internal_test.go b/internal/routewatch/batchwrite_internal_test.go index ecacc8d..d9a2248 100644 --- a/internal/routewatch/batchwrite_internal_test.go +++ b/internal/routewatch/batchwrite_internal_test.go @@ -175,12 +175,13 @@ func (s *blockingStore) releaseWrites() { // //nolint:tagliatelle // field names are the keys the handlers log type stopLog struct { - Msg string `json:"msg"` - Updates int `json:"updates"` - TableChanges int `json:"table_changes"` - HeldMessages int `json:"held_messages"` - Paths int `json:"paths"` - BatchEntries int `json:"batch_entries"` + Msg string `json:"msg"` + Updates int `json:"updates"` + TableChanges int `json:"table_changes"` + HeldMessages int `json:"held_messages"` + OriginChanges int `json:"origin_changes"` + Paths int `json:"paths"` + BatchEntries int `json:"batch_entries"` } // newErrorLogger returns a logger that writes only errors, as JSON lines, to @@ -472,6 +473,42 @@ func (s tableWriteBlockingStore) WriteRouteChanges( return s.mockStore.WriteRouteChanges(changes, newPeers) } +// stopPrefixHandlerWithBlockedWrite stops the prefix handler, whose store +// holds the write that fill starts, after handing it the message extra, and +// returns what Stop logged. blocking is the store the write waits on; store is +// a blockingStore wrapped for the routing table's changes or the origin +// changes. +func stopPrefixHandlerWithBlockedWrite( + t *testing.T, + blocking *blockingStore, + store database.Store, + logs *bytes.Buffer, + fill, extra *ristypes.RISMessage, +) stopLog { + t.Helper() + + h := NewPrefixHandler(store, newErrorLogger(logs), + routetable.New(routetable.Params{})) + stopWhileAWriteIsBlocked(t, blocking, h, h.done, + []*ristypes.RISMessage{fill}, extra) + + return lastLogged(t, logs) +} + +// originChangesBlockingStore is a blockingStore whose writes of the origin +// changes also wait until the test calls releaseWrites. +type originChangesBlockingStore struct { + *blockingStore +} + +func (s originChangesBlockingStore) WriteOriginChanges( + changes []routetable.OriginChange, +) error { + s.waitForRelease() + + return s.mockStore.WriteOriginChanges(changes) +} + // TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress holds the write // of the routing table's changes that a full batch starts, then hands the // prefix handler one more prefix and stops it with a deadline that passes @@ -485,19 +522,67 @@ func TestPrefixHandlerStopCountsTheTableChangesOfAWriteInProgress(t *testing.T) var logs bytes.Buffer blocking := newBlockingStore() - h := NewPrefixHandler(tableWriteBlockingStore{blockingStore: blocking}, - newErrorLogger(&logs), routetable.New(routetable.Params{})) fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize) - stopWhileAWriteIsBlocked(t, blocking, h, h.done, - []*ristypes.RISMessage{prefixAnnouncement(fill)}, - prefixAnnouncement([]string{extraPrefix})) + logged := stopPrefixHandlerWithBlockedWrite(t, blocking, + tableWriteBlockingStore{blockingStore: blocking}, &logs, + prefixAnnouncement(fill), prefixAnnouncement([]string{extraPrefix})) - if logged := lastLogged(t, &logs); logged.TableChanges != 2 { + if logged.TableChanges != 2 { t.Errorf("Stop logged %d table changes not written, want 2", logged.TableChanges) } } +// TestPrefixHandlerStopCountsTheOriginChangesOfAWriteInProgress holds the +// write of the origin changes that a full batch starts, then hands the prefix +// handler one more prefix and stops it with a deadline that passes meanwhile. +// Stop must log as not written both the change the held write took and the one +// since. +func TestPrefixHandlerStopCountsTheOriginChangesOfAWriteInProgress(t *testing.T) { + t.Parallel() + + const extraPrefix = "198.51.100.0/24" + + var logs bytes.Buffer + + blocking := newBlockingStore() + fill := slices.Repeat([]string{testV4Prefix}, prefixBatchSize) + logged := stopPrefixHandlerWithBlockedWrite(t, blocking, + originChangesBlockingStore{blockingStore: blocking}, &logs, + prefixAnnouncement(fill), prefixAnnouncement([]string{extraPrefix})) + + if logged.OriginChanges != 2 { + t.Errorf("Stop logged %d origin changes not written, want 2", + logged.OriginChanges) + } +} + +// TestPrefixHandlerStopCountsTheWaitingOriginChanges stops the prefix handler +// with a deadline that has already passed, so the flush goroutine skips its +// last write. The origin changes still waiting in the routing table are then +// not written, and Stop must log them. +func TestPrefixHandlerStopCountsTheWaitingOriginChanges(t *testing.T) { + t.Parallel() + + var logs bytes.Buffer + + store := newMockStore() + h := newPrefixHandler(store, newErrorLogger(&logs), + routetable.New(routetable.Params{}), longerThanAnyTest, + maxTableChangesPerWrite) + + h.HandleMessage(peerAnnouncement(testPeerIP, testV4Prefix, testV6Prefix)) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + h.Stop(ctx) + + if logged := lastLogged(t, &logs); logged.OriginChanges != 2 { + t.Errorf("Stop logged %d origin changes not written, want 2", + logged.OriginChanges) + } +} + // TestASHandlerSkipsItsLastWriteAfterTheStopDeadline also checks what Stop // logs: the asnBatchSize entries of the held write and the entry for testASNB, // handled since, are not written. diff --git a/internal/routewatch/dbmaintainer.go b/internal/routewatch/dbmaintainer.go index 1384253..379a68c 100644 --- a/internal/routewatch/dbmaintainer.go +++ b/internal/routewatch/dbmaintainer.go @@ -87,6 +87,7 @@ func (m *DBMaintainer) Start(ctx context.Context) { "checkpoint_interval", checkpointInterval, "vacuum_interval", vacuumInterval, "analyze_interval", analyzeInterval, + "origin_changes_prune_interval", originChangesPruneInterval, ) } diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index f701ff3..55f7e81 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -85,8 +85,8 @@ const ( // a time, and takes new messages in between. It holds back at most // maxHeldBackMessages messages. // -// Each flush writes the table's changes to the database. It also stores each -// prefix the updates name with the time it was first seen. +// Each flush writes the table's changes and origin changes to the database. +// It also stores each prefix the updates name with the time it was first seen. type PrefixHandler struct { db database.Store logger *logger.Logger @@ -103,10 +103,10 @@ type PrefixHandler struct { // 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 - // mu, so HandleMessage waits for a database write only while batch is at - // batchMaxSize. + // Batching. mu guards batch, writing, writingTableChanges and + // writingOriginChanges: the flush goroutine takes them under mu and + // writes them after releasing mu, so HandleMessage waits for a database + // write only while batch is at batchMaxSize. mu sync.Mutex batch []seenPrefix // writing is how many updates the flush goroutine took and has not @@ -115,6 +115,9 @@ type PrefixHandler struct { // writingTableChanges is how many changes the flush goroutine took from // the routing table and has not finished writing. writingTableChanges int + // writingOriginChanges is how many origin changes the flush goroutine + // took from the routing table and has not finished writing. + writingOriginChanges int // batchTaken is signalled when the flush goroutine takes batch; its L is // &mu. batchTaken sync.Cond @@ -307,10 +310,9 @@ func (h *PrefixHandler) Stop(ctx context.Context) { unwritten := h.writing + len(h.batch) tableChanges := h.writingTableChanges + h.table.ChangeCount() heldMessages := h.heldCount + originChanges := h.writingOriginChanges + h.table.OriginChangeCount() h.mu.Unlock() - originChanges := h.table.OriginChangeCount() - if unwritten == 0 && tableChanges == 0 && heldMessages == 0 && originChanges == 0 { return @@ -513,9 +515,9 @@ func (h *PrefixHandler) takeBatch() []seenPrefix { return batch } -// flushBatch writes the routing table's changes, then takes the pending batch -// and stores the prefixes it names, without holding the mutex during the -// writes. +// flushBatch writes the routing table's changes and origin changes, then +// takes the pending batch and stores the prefixes it names, without holding +// the mutex during the writes. func (h *PrefixHandler) flushBatch() { h.writeTableChanges() h.writeOriginChanges() @@ -620,12 +622,15 @@ func (h *PrefixHandler) writeTableChanges() { } } -// writeOriginChanges writes at most maxOriginChangesPerWrite of the routing +// writeOriginChanges writes at most originChangesPerWrite of the routing // table's origin changes. When the write fails, they go back to the table, so // that a later write writes them again. A dropped change is logged as a // warning. func (h *PrefixHandler) writeOriginChanges() { + h.mu.Lock() changes, dropped := h.table.TakeOriginChanges(h.originChangesPerWrite) + h.writingOriginChanges = len(changes) + h.mu.Unlock() if dropped > 0 { h.logger.Warn("Origin changes dropped", "dropped", dropped) @@ -636,6 +641,12 @@ func (h *PrefixHandler) writeOriginChanges() { } err := h.db.WriteOriginChanges(changes) + + h.mu.Lock() + defer h.mu.Unlock() + + h.writingOriginChanges = 0 + if err != nil { h.logger.Error("Failed to write the origin changes; the next flush "+ "retries", "changes", len(changes), "error", err) diff --git a/internal/routewatch/prefixhandler_originchanges_internal_test.go b/internal/routewatch/prefixhandler_originchanges_internal_test.go index c2d769d..a40df80 100644 --- a/internal/routewatch/prefixhandler_originchanges_internal_test.go +++ b/internal/routewatch/prefixhandler_originchanges_internal_test.go @@ -1,12 +1,16 @@ package routewatch import ( + "bytes" + "encoding/json" "errors" + "log/slog" "net/netip" "slices" "testing" "git.eeqj.de/sneak/routewatch/internal/database" + "git.eeqj.de/sneak/routewatch/internal/logger" "git.eeqj.de/sneak/routewatch/internal/routetable" ) @@ -125,3 +129,53 @@ func TestPrefixHandlerWritesManyOriginChangesOverSeveralFlushes(t *testing.T) { t.Errorf("WriteOriginChanges held %v changes, want %v", store.counts, want) } } + +// TestPrefixHandlerLogsTheDroppedOriginChanges fills the routing table with +// one origin change more than it keeps, then flushes. The warning the flush +// logs must carry the dropped count. +func TestPrefixHandlerLogsTheDroppedOriginChanges(t *testing.T) { + t.Parallel() + + var logs bytes.Buffer + + db := newTestDatabase(t) + h := prefixHandlerWithoutTimer(db) + h.logger = &logger.Logger{ + Logger: slog.New(slog.NewJSONHandler(&logs, + &slog.HandlerOptions{Level: slog.LevelWarn})), + } + + announceManyPrefixes(h.table, 100001) + + h.flushBatch() + + var warning struct { + Msg string `json:"msg"` + Dropped *int `json:"dropped"` + } + + found := false + + for line := range bytes.SplitSeq(bytes.TrimSpace(logs.Bytes()), []byte("\n")) { + warning.Dropped = nil + + err := json.Unmarshal(line, &warning) + if err != nil { + t.Fatalf("want a JSON log line, got %q: %v", line, err) + } + + if warning.Msg != "Origin changes dropped" { + continue + } + + found = true + + if warning.Dropped == nil || *warning.Dropped != 1 { + t.Errorf("the warning's dropped = %v, want 1", warning.Dropped) + } + } + + if !found { + t.Errorf("no warning logged the dropped origin changes") + } +} diff --git a/internal/server/prefix_originchanges_internal_test.go b/internal/server/prefix_originchanges_internal_test.go index 1d88359..75c7eda 100644 --- a/internal/server/prefix_originchanges_internal_test.go +++ b/internal/server/prefix_originchanges_internal_test.go @@ -145,4 +145,75 @@ func TestPrefixDetailSaysWhenMoreThan5000Changes(t *testing.T) { if !strings.Contains(page, "more than 5000 changes") { t.Errorf("page does not say there are more than 5000 changes") } + + if got := strings.Count(page, "