Record origin changes and show them on the prefix page (closes #161)

The routing table records a row each time a prefix and origin AS gets its first peer (announced) or loses its last (withdrawn), with the time and how many peers carry the prefix just after. The prefix handler writes the rows to a new origin_changes table, apart from routes. Every hour the database maintainer keeps each prefix's rows from the last 72 hours or its newest 500, whichever keeps more. The prefix page lists them newest first, at most 5,000.

At most 100,000 rows wait in memory to be written; past that the oldest are dropped and logged. A stop loses the rows past its last write.

Model: deepseek-v4-pro (implementation); opus-5-5 (commit message)

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 (
@@ -55,6 +56,14 @@ type Store interface {
 	WriteRouteChanges(changes []routetable.Change, newPeers map[string]uint) error
 	ReadRouteTable(ctx context.Context) (*routetable.Table, error)
 
+	// Origin change operations: the prefix and origin AS pairs' first and
+	// last peers, as the routing table records them.
+	WriteOriginChanges(changes []routetable.OriginChange) error
+	GetOriginChanges(
+		ctx context.Context, prefix netip.Prefix, limit int,
+	) ([]routetable.OriginChange, error)
+	PruneOriginChanges(ctx context.Context) error
+
 	// ASN WHOIS operations
 	GetNextStaleASN(ctx context.Context, staleThreshold time.Duration) (int, error)
 	UpdateASNWHOIS(ctx context.Context, update *ASNWHOISUpdate) error

internal/database/originchanges.go

@@ -0,0 +1,172 @@
+package database
+
+import (
+	"context"
+	"fmt"
+	"net/netip"
+	"time"
+
+	"git.eeqj.de/sneak/routewatch/internal/routetable"
+)
+
+// 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
+// minOriginChangesPerPrefix.
+const originChangesRetention = 72 * time.Hour
+
+// The statements of the origin_changes table (schema.sql).
+const (
+	insertOriginChangeSQL = `INSERT INTO origin_changes
+		(prefix, origin_asn, announced, peer_count, time)
+		VALUES (?, ?, ?, ?, ?)`
+	originChangesSQL = `SELECT prefix, origin_asn, announced, peer_count, time
+		FROM origin_changes WHERE prefix = ? ORDER BY id DESC LIMIT ?`
+	prunePrefixesSQL = `SELECT prefix FROM origin_changes
+		GROUP BY prefix HAVING COUNT(*) > ?`
+	pruneOriginChangesSQL = `DELETE FROM origin_changes
+		WHERE prefix = ? AND time < ? AND id < (
+			SELECT id FROM origin_changes WHERE prefix = ?
+			ORDER BY id DESC LIMIT 1 OFFSET ?)`
+)
+
+// WriteOriginChanges writes, in one transaction, the origin changes the
+// routing table handed over. When it fails, it writes nothing; the caller then
+// writes them again.
+func (d *Database) WriteOriginChanges(changes []routetable.OriginChange) error {
+	if len(changes) == 0 {
+		return nil
+	}
+
+	d.lock("WriteOriginChanges")
+	defer d.unlock()
+
+	tx, err := d.beginTx()
+	if err != nil {
+		return fmt.Errorf("failed to begin transaction: %w", err)
+	}
+	defer d.rollback(tx)
+
+	ctx := context.Background()
+
+	stmt, err := tx.PrepareContext(ctx, insertOriginChangeSQL)
+	if err != nil {
+		return fmt.Errorf("failed to prepare origin change insert: %w", err)
+	}
+	defer func() { _ = stmt.Close() }()
+
+	for _, change := range changes {
+		_, err := stmt.ExecContext(ctx, change.Prefix.String(),
+			change.OriginASN, change.Announced, change.PeerCount,
+			change.Time.Unix())
+		if err != nil {
+			return fmt.Errorf("failed to write origin change %s: %w",
+				change.Prefix, err)
+		}
+	}
+
+	err = tx.Commit()
+	if err != nil {
+		return fmt.Errorf("failed to commit transaction: %w", err)
+	}
+
+	return nil
+}
+
+// GetOriginChanges returns at most limit of prefix's origin changes, newest
+// first.
+func (d *Database) GetOriginChanges(
+	ctx context.Context, prefix netip.Prefix, limit int,
+) ([]routetable.OriginChange, error) {
+	rows, err := d.db.QueryContext(ctx, originChangesSQL,
+		prefix.Masked().String(), limit)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read origin changes: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	var changes []routetable.OriginChange
+
+	for rows.Next() {
+		var (
+			change     routetable.OriginChange
+			prefixText string
+			timeSec    int64
+		)
+
+		err := rows.Scan(&prefixText, &change.OriginASN, &change.Announced,
+			&change.PeerCount, &timeSec)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read an origin change: %w", err)
+		}
+
+		change.Prefix, err = netip.ParsePrefix(prefixText)
+		if err != nil {
+			return nil, fmt.Errorf("failed to read origin change %s: %w",
+				prefixText, err)
+		}
+
+		change.Time = time.Unix(timeSec, 0).UTC()
+		changes = append(changes, change)
+	}
+
+	err = rows.Err()
+	if err != nil {
+		return nil, fmt.Errorf("failed to read origin changes: %w", err)
+	}
+
+	return changes, nil
+}
+
+// PruneOriginChanges deletes, for each prefix with more than
+// 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,
+		minOriginChangesPerPrefix)
+	if err != nil {
+		return fmt.Errorf("failed to find prefixes to prune: %w", err)
+	}
+	defer func() { _ = rows.Close() }()
+
+	var prefixes []string
+
+	for rows.Next() {
+		var prefix string
+
+		err := rows.Scan(&prefix)
+		if err != nil {
+			return fmt.Errorf("failed to read a prefix to prune: %w", err)
+		}
+
+		prefixes = append(prefixes, prefix)
+	}
+
+	err = rows.Err()
+	if err != nil {
+		return fmt.Errorf("failed to find prefixes to prune: %w", err)
+	}
+
+	cutoff := time.Now().Add(-originChangesRetention).Unix()
+
+	for _, prefix := range prefixes {
+		d.lock("PruneOriginChanges")
+
+		_, err := d.writer.ExecContext(ctx, pruneOriginChangesSQL,
+			prefix, cutoff, prefix, minOriginChangesPerPrefix-1)
+
+		d.unlock()
+
+		if err != nil {
+			return fmt.Errorf("failed to prune origin changes of %s: %w",
+				prefix, err)
+		}
+	}
+
+	return nil
+}

internal/database/originchanges_internal_test.go

@@ -0,0 +1,217 @@
+package database
+
+import (
+	"net/netip"
+	"reflect"
+	"testing"
+	"time"
+
+	"git.eeqj.de/sneak/routewatch/internal/routetable"
+)
+
+// TestOriginChangesWriteThenRead writes three origin changes and reads them
+// back newest first, honouring the limit.
+func TestOriginChangesWriteThenRead(t *testing.T) {
+	t.Parallel()
+
+	db := newRoutesTestDatabase(t)
+	prefix := netip.MustParsePrefix(tablePrefixV4)
+
+	changes := []routetable.OriginChange{
+		{Prefix: prefix, OriginASN: asA, Announced: true, PeerCount: 1,
+			Time: tableTime(10)},
+		{Prefix: prefix, OriginASN: asA, Announced: false, PeerCount: 0,
+			Time: tableTime(20)},
+		{Prefix: prefix, OriginASN: asB, Announced: true, PeerCount: 1,
+			Time: tableTime(30)},
+	}
+
+	err := db.WriteOriginChanges(changes)
+	if err != nil {
+		t.Fatalf("WriteOriginChanges: %v", err)
+	}
+
+	got, err := db.GetOriginChanges(t.Context(), prefix, 2)
+	if err != nil {
+		t.Fatalf("GetOriginChanges: %v", err)
+	}
+
+	want := []routetable.OriginChange{
+		{Prefix: prefix, OriginASN: asB, Announced: true, PeerCount: 1,
+			Time: tableTime(30)},
+		{Prefix: prefix, OriginASN: asA, Announced: false, PeerCount: 0,
+			Time: tableTime(20)},
+	}
+
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("GetOriginChanges = %+v, want %+v", got, want)
+	}
+}
+
+// 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)
+	straddling := netip.MustParsePrefix(tablePrefixShortV4)
+	busy := netip.MustParsePrefix(tablePrefixBusyV4)
+	other := netip.MustParsePrefix("10.20.0.0/16")
+
+	// 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.
+	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)...)
+	// 100 rows older than 72 hours and minOriginChangesPerPrefix+100 newer
+	// rows for busy. The prune keeps all the newer rows and none of the older,
+	// because the newer rows alone are more than minOriginChangesPerPrefix.
+	changes = append(changes, rows(busy, now.Add(-100*time.Hour), 100)...)
+	changes = append(changes, rows(busy, now.Add(-71*time.Hour),
+		minOriginChangesPerPrefix+100)...)
+	// One row for other, which the prune leaves alone.
+	changes = append(changes, rows(other, now.Add(-100*time.Hour), 1)...)
+
+	err := db.WriteOriginChanges(changes)
+	if err != nil {
+		t.Fatalf("WriteOriginChanges: %v", err)
+	}
+
+	err = db.PruneOriginChanges(t.Context())
+	if err != nil {
+		t.Fatalf("PruneOriginChanges: %v", err)
+	}
+
+	// The newest row survives, and the oldest survivor is the 100th oldest,
+	// so a prune that kept the oldest 500 instead would fail.
+	checkSurvivors(t, db, oldPrefix, 500,
+		now.Add(-100*time.Hour).Add(599*time.Minute),
+		now.Add(-100*time.Hour).Add(100*time.Minute))
+
+	// 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.
+	checkSurvivors(t, db, straddling, 500,
+		now.Add(-71*time.Hour).Add(199*time.Minute),
+		now.Add(-100*time.Hour).Add(100*time.Minute))
+
+	// Every newer-than-72-hours row survives, so the oldest survivor is the
+	// first of them and the newest is minOriginChangesPerPrefix+99 minutes
+	// after it.
+	checkSurvivors(t, db, busy, minOriginChangesPerPrefix+100,
+		now.Add(-71*time.Hour).
+			Add(time.Duration(minOriginChangesPerPrefix+99)*time.Minute),
+		now.Add(-71*time.Hour))
+
+	gotOther, err := db.GetOriginChanges(t.Context(), other, 10)
+	if err != nil {
+		t.Fatalf("GetOriginChanges: %v", err)
+	}
+
+	if len(gotOther) != 1 {
+		t.Errorf("GetOriginChanges(other) = %d rows, want 1", len(gotOther))
+	}
+}
+
+// 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
+}
+
+// checkSurvivors checks that prefix kept exactly wantCount rows, with the
+// newest survivor at wantNewest and the oldest at wantOldest.
+func checkSurvivors(
+	t *testing.T, db *Database, prefix netip.Prefix,
+	wantCount int, wantNewest, wantOldest time.Time,
+) {
+	t.Helper()
+
+	times := survivorTimes(t, db, prefix, wantCount+1)
+	if len(times) != wantCount {
+		t.Errorf("GetOriginChanges(%s) = %d rows, want %d",
+			prefix, len(times), wantCount)
+
+		return
+	}
+
+	if !times[0].Equal(wantNewest) {
+		t.Errorf("the newest surviving row's time = %v, want %v",
+			times[0], wantNewest)
+	}
+
+	if !times[wantCount-1].Equal(wantOldest) {
+		t.Errorf("the oldest surviving row's time = %v, want %v",
+			times[wantCount-1], wantOldest)
+	}
+}
+
+// TestOriginChangesSurviveRouteDeletion checks that deleting a routes row
+// leaves the prefix's origin_changes rows.
+func TestOriginChangesSurviveRouteDeletion(t *testing.T) {
+	t.Parallel()
+
+	db := newRoutesTestDatabase(t)
+	prefix := netip.MustParsePrefix(tablePrefixV4)
+
+	err := db.WriteOriginChanges([]routetable.OriginChange{{
+		Prefix: prefix, OriginASN: asA, Announced: true, PeerCount: 1,
+		Time: tableTime(10),
+	}})
+	if err != nil {
+		t.Fatalf("WriteOriginChanges: %v", err)
+	}
+
+	// Write a route, then delete it.
+	table := routetable.New(routetable.Params{})
+	announceTo(table, tablePrefixV4, asA, peerA, 10)
+	writeTableChanges(t, db, table)
+
+	table.Withdraw(prefix, peerA)
+	writeTableChanges(t, db, table)
+
+	got, err := db.GetOriginChanges(t.Context(), prefix, 10)
+	if err != nil {
+		t.Fatalf("GetOriginChanges: %v", err)
+	}
+
+	if len(got) != 1 {
+		t.Errorf("GetOriginChanges = %d rows, want 1", len(got))
+	}
+}

internal/database/routes_internal_test.go

@@ -34,6 +34,7 @@ const (
 
 	tablePrefixV4      = "10.1.2.0/24"
 	tablePrefixShortV4 = "10.9.0.0/16"
+	tablePrefixBusyV4  = "10.10.0.0/16"
 	tablePrefixV6      = "2001:db8:1::/48"
 )
 

internal/database/schema.sql

@@ -119,4 +119,20 @@ CREATE TABLE IF NOT EXISTS routes (
 CREATE TABLE IF NOT EXISTS route_peers (
 	number INTEGER PRIMARY KEY,
 	address TEXT NOT NULL UNIQUE
-);
+);
+
+-- 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 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,
+	origin_asn INTEGER NOT NULL,
+	announced BOOLEAN NOT NULL,
+	peer_count INTEGER NOT NULL,
+	time INTEGER NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_origin_changes_prefix ON origin_changes(prefix);