Fix the second review of the origin changes
check / check (push) Failing after 15m27s

Stop now counts the origin changes inside a running write, in a new
writingOriginChanges field, and adds them to the count it logs. The prune test
checks which rows survive, including a prefix with rows on both sides of 72
hours. The prefix page tests cover 5,000 and 5,001 rows and a row's time and
peer count. New tests cover the dropped-rows warning and the dropped count's
reset. The prune floor is named minOriginChangesPerPrefix. The README, TODO,
package doc and doc comments are corrected to match.

Model: deepseek-v4-pro
This commit is contained in:
2026-11-05 05:08:27 +00:00
parent 56cfc7e554
commit 22783eb425
12 changed files with 382 additions and 82 deletions
+3 -2
View File
@@ -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 (
+8 -8
View File
@@ -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()
@@ -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) {