sneak/routewatch
1
0
Fork 0
Code Issues 4 Pull Requests 2 Actions 19 Packages Projects Releases Wiki Activity

Remove the per-peer route tables (closes #122)
check / check (push) Canceled after 0s
Details

Browse Source 
The routing table in memory now holds every route, so the two per-peer route
tables go, with every write, delete, count and read that served them. The
prefix handler applies each update to the table, writes the table's changes
every second, and stores each new prefix with its first-seen time. The peer
delete, its retry on the next flush and the updates held for it go too.

The image's memory settings fit a full table of about 1 GB: GOMEMLIMIT of
2.5 GiB, a 512 MiB page cache for the writing connection, and SQLite heap
limits of 768 MiB and 1 GiB. The README describes the new table, its
database copy and the memory budget.

Model: opus-5-5
This commit is contained in:
sneak
2026-10-27 15:48:18 +00:00
parent 506e289a63
commit 555f01920d
20 changed files with 512 additions and 5105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/database/counts.go
+9 -100
View File
@@ -6,11 +6,10 @@ import (
"sync"
)
// liveCounts holds the running row counts and the prefix distribution that the
// stats endpoints report. They are seeded once at startup from the tables and
// then adjusted on every write, so a stats read serves them from memory instead
// of running a query over the tables. The COUNT(*) scans (issue 27) and then
// the prefix distribution query (issue 30) each grew with the database until
// liveCounts holds the running row counts that the stats endpoints report.
// They are seeded once at startup from the tables and then adjusted on every
// write, so a stats read serves them from memory instead of running a query
// over the tables. The COUNT(*) scans (issue 27) grew with the database until
// they took the whole request timeout and made /api/v1/stats return 500.
//
// A single mutex guards all fields so the stats reader takes a consistent
@@ -23,20 +22,11 @@ type liveCounts struct {
prefixesV6 int
peerings int
peers int
routesV4 int
routesV6 int
// The prefix distribution: for each mask length, the number of distinct
// prefixes that have at least one live route.
distributionV4 [ipv4Bits + 1]int
distributionV6 [ipv6Bits + 1]int
}
// seed sets every count to the value read from the tables at startup. It runs
// before any writer, so it needs no coordination with the adjust methods.
func (c *liveCounts) seed(
asns, prefixesV4, prefixesV6, peerings, peers, routesV4, routesV6 int,
distributionV4, distributionV6 []PrefixDistribution,
) {
func (c *liveCounts) seed(asns, prefixesV4, prefixesV6, peerings, peers int) {
c.mu.Lock()
defer c.mu.Unlock()
@@ -45,16 +35,6 @@ func (c *liveCounts) seed(
c.prefixesV6 = prefixesV6
c.peerings = peerings
c.peers = peers
c.routesV4 = routesV4
c.routesV6 = routesV6
for _, entry := range distributionV4 {
addAtMaskLength(c.distributionV4[:], entry.MaskLength, entry.Count)
}
for _, entry := range distributionV6 {
addAtMaskLength(c.distributionV6[:], entry.MaskLength, entry.Count)
}
}
// addASNs adds n to the ASN count.
@@ -86,41 +66,7 @@ func (c *liveCounts) addPeers(n int) {
c.mu.Unlock()
}
// addRoutes adds to the IPv4 and IPv6 live-route counts. Deletions pass
// negative values.
func (c *liveCounts) addRoutes(v4, v6 int) {
c.mu.Lock()
c.routesV4 += v4
c.routesV6 += v6
c.mu.Unlock()
}
// addToDistribution adds n to the IPv4 and IPv6 prefix distributions once for
// each listed mask length. A write lists the mask lengths of the prefixes it
// gave their first live route with n = 1, and of the prefixes it left with no
// live route with n = -1.
func (c *liveCounts) addToDistribution(maskLengthsV4, maskLengthsV6 []int, n int) {
c.mu.Lock()
defer c.mu.Unlock()
for _, maskLength := range maskLengthsV4 {
addAtMaskLength(c.distributionV4[:], maskLength, n)
}
for _, maskLength := range maskLengthsV6 {
addAtMaskLength(c.distributionV6[:], maskLength, n)
}
}
// addAtMaskLength adds n to counts[maskLength]. A mask length the array has no
// entry for is ignored, so a malformed route cannot crash the daemon.
func addAtMaskLength(counts []int, maskLength, n int) {
if maskLength >= 0 && maskLength < len(counts) {
counts[maskLength] += n
}
}
// fill copies the counts into a Stats, including the derived totals, under a
// fill copies the counts into a Stats, including the derived total, under a
// single read lock so the reader sees one consistent snapshot.
func (c *liveCounts) fill(s *Stats) {
c.mu.RLock()
@@ -132,26 +78,6 @@ func (c *liveCounts) fill(s *Stats) {
s.Prefixes = c.prefixesV4 + c.prefixesV6
s.Peerings = c.peerings
s.Peers = c.peers
s.IPv4Routes = c.routesV4
s.IPv6Routes = c.routesV6
s.LiveRoutes = c.routesV4 + c.routesV6
s.IPv4PrefixDistribution = distributionList(c.distributionV4[:])
s.IPv6PrefixDistribution = distributionList(c.distributionV6[:])
}
// distributionList lists the mask lengths that have at least one prefix, in
// ascending order, the way the distribution query returns them.
func distributionList(counts []int) []PrefixDistribution {
var list []PrefixDistribution
for maskLength, count := range counts {
if count > 0 {
list = append(list,
PrefixDistribution{MaskLength: maskLength, Count: count})
}
}
return list
}
// countRows returns the number of rows in the named table. It is used only at
@@ -167,9 +93,8 @@ func (d *Database) countRows(ctx context.Context, table string) (int, error) {
return n, nil
}
// seedCounts reads the current row counts and prefix distribution from the
// tables into the in-memory counters. It runs once at startup, before the
// streamer begins writing.
// seedCounts reads the current row counts from the tables into the in-memory
// counters. It runs once at startup, before the streamer begins writing.
func (d *Database) seedCounts(ctx context.Context) error {
asns, err := d.countRows(ctx, "asns")
if err != nil {
@@ -196,23 +121,7 @@ func (d *Database) seedCounts(ctx context.Context) error {
return err
}
routesV4, err := d.countRows(ctx, tableLiveRoutesV4)
if err != nil {
return err
}
routesV6, err := d.countRows(ctx, tableLiveRoutesV6)
if err != nil {
return err
}
distributionV4, distributionV6, err := d.GetPrefixDistributionContext(ctx)
if err != nil {
return err
}
d.counts.seed(asns, prefixesV4, prefixesV6, peerings, peers, routesV4, routesV6,
distributionV4, distributionV6)
d.counts.seed(asns, prefixesV4, prefixesV6, peerings, peers)
return nil
}
internal/database/counts_internal_test.go
+25 -577
View File
@@ -3,62 +3,23 @@ package database
import (
"context"
"fmt"
"slices"
"sync"
"testing"
"time"
"git.eeqj.de/sneak/routewatch/internal/config"
"git.eeqj.de/sneak/routewatch/internal/logger"
"github.com/google/uuid"
)
// testV4PeerIP is the peer the IPv4 routes in these tests are learned from.
// testV4PeerIP is the peer the IPv4 rows in these tests are learned from.
const testV4PeerIP = "192.0.2.1"
// testV6Prefix is the prefix of the IPv6 routes in these tests.
// testV6Prefix is the IPv6 prefix of these tests.
const testV6Prefix = "2001:db8::/32"
// mkV4Route builds an IPv4 live route with its mask length taken from the
// prefix.
func mkV4Route(t *testing.T, prefix string, asn int, ts time.Time) *LiveRoute {
t.Helper()
maskLength, err := prefixMaskLength(prefix)
if err != nil {
t.Fatalf("prefixMaskLength(%s): %v", prefix, err)
}
return &LiveRoute{
Prefix: prefix,
MaskLength: maskLength,
IPVersion: ipVersionV4,
OriginASN: asn,
PeerIP: testV4PeerIP,
ASPath: []int{asn},
NextHop: "192.0.2.254",
LastUpdated: ts,
}
}
// mkV6Route builds an IPv6 live route.
func mkV6Route(prefix string, asn int, ts time.Time) *LiveRoute {
return &LiveRoute{
Prefix: prefix,
MaskLength: 32,
IPVersion: ipVersionV6,
OriginASN: asn,
PeerIP: "2001:db8::1",
ASPath: []int{asn},
NextHop: "2001:db8::ffff",
LastUpdated: ts,
}
}
// TestLiveCountsTrackWritesInRealtime checks that the stats counts start at
// zero, reflect each write the moment it commits (no recompute, no timer), do
// not move when a route is merely re-announced, and drop when a route is
// deleted. These counts are what /api/v1/stats reports; before this change the
// zero and reflect each write the moment it commits (no recompute, no timer).
// These counts are what /api/v1/stats reports; before this change the
// endpoint recomputed them with a COUNT(*) over each table on every request.
func TestLiveCountsTrackWritesInRealtime(t *testing.T) {
t.Parallel()
@@ -79,7 +40,7 @@ func TestLiveCountsTrackWritesInRealtime(t *testing.T) {
}
if empty.ASNs != 0 || empty.Prefixes != 0 || empty.Peerings != 0 ||
empty.Peers != 0 || empty.LiveRoutes != 0 {
empty.Peers != 0 {
t.Fatalf("empty database counts nonzero: %+v", empty)
}
@@ -87,17 +48,6 @@ func TestLiveCountsTrackWritesInRealtime(t *testing.T) {
writeASNsPrefixesAndPeers(t, db, ts)
routes := []*LiveRoute{
mkV4Route(t, "198.51.100.0/24", 64500, ts),
mkV4Route(t, "203.0.113.0/24", 64501, ts.Add(time.Minute)),
mkV6Route(testV6Prefix, 64502, ts.Add(2*time.Minute)),
}
err = db.UpsertLiveRouteBatch(routes)
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
stats, err := db.GetStatsContext(ctx)
if err != nil {
t.Fatalf("GetStatsContext: %v", err)
@@ -105,46 +55,11 @@ func TestLiveCountsTrackWritesInRealtime(t *testing.T) {
assertCounts(t, "after inserts", stats, wantCounts{
asns: 2, prefixes: 2, peerings: 1, peers: 1,
ipv4Routes: 2, ipv6Routes: 1, liveRoutes: 3,
})
// Re-announcing the same routes is an update, not an insert: counts hold.
err = db.UpsertLiveRouteBatch(routes)
if err != nil {
t.Fatalf("UpsertLiveRouteBatch (re-announce): %v", err)
}
stats, err = db.GetStatsContext(ctx)
if err != nil {
t.Fatalf("GetStatsContext: %v", err)
}
assertCounts(t, "after re-announce", stats, wantCounts{
asns: 2, prefixes: 2, peerings: 1, peers: 1,
ipv4Routes: 2, ipv6Routes: 1, liveRoutes: 3,
})
// A withdrawal removes one route.
err = db.DeleteLiveRouteBatch([]LiveRouteDeletion{
{Prefix: "203.0.113.0/24", PeerIP: testV4PeerIP, IPVersion: ipVersionV4},
})
if err != nil {
t.Fatalf("DeleteLiveRouteBatch: %v", err)
}
stats, err = db.GetStatsContext(ctx)
if err != nil {
t.Fatalf("GetStatsContext: %v", err)
}
assertCounts(t, "after delete", stats, wantCounts{
asns: 2, prefixes: 2, peerings: 1, peers: 1,
ipv4Routes: 1, ipv6Routes: 1, liveRoutes: 2,
})
}
// writeASNsPrefixesAndPeers writes two ASNs, two prefixes, one peer and one
// peering: the rows besides live routes that the stats count.
// peering: the rows the stats count.
func writeASNsPrefixesAndPeers(t *testing.T, db *Database, ts time.Time) {
t.Helper()
@@ -180,8 +95,8 @@ func writeASNsPrefixesAndPeers(t *testing.T, db *Database, ts time.Time) {
}
// TestLiveCountsSeededFromDatabaseAtStartup writes rows, reopens the same
// database file, and checks the counts and the prefix distribution come back
// from the seed scan rather than starting at zero.
// database file, and checks the counts come back from the seed scan rather
// than starting at zero.
func TestLiveCountsSeededFromDatabaseAtStartup(t *testing.T) {
t.Parallel()
@@ -192,23 +107,7 @@ func TestLiveCountsSeededFromDatabaseAtStartup(t *testing.T) {
t.Fatalf("failed to create database: %v", err)
}
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
seen := SeenTimes{First: ts, Last: ts}
err = db.GetOrCreateASNBatch(
map[int]SeenTimes{64500: seen, 64501: seen, 64502: seen})
if err != nil {
t.Fatalf("GetOrCreateASNBatch: %v", err)
}
err = db.UpsertLiveRouteBatch([]*LiveRoute{
mkV4Route(t, "198.51.100.0/24", 64500, ts),
mkV6Route(testV6Prefix, 64502, ts),
})
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
writeASNsPrefixesAndPeers(t, db, time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC))
err = db.Close()
if err != nil {
@@ -226,435 +125,9 @@ func TestLiveCountsSeededFromDatabaseAtStartup(t *testing.T) {
t.Fatalf("GetStatsContext after reopen: %v", err)
}
if stats.ASNs != 3 {
t.Errorf("seeded ASNs = %d, want 3", stats.ASNs)
}
if stats.IPv4Routes != 1 || stats.IPv6Routes != 1 || stats.LiveRoutes != 2 {
t.Errorf("seeded routes = (v4 %d, v6 %d, total %d), want (1, 1, 2)",
stats.IPv4Routes, stats.IPv6Routes, stats.LiveRoutes)
}
assertDistribution(t, "seeded IPv4 distribution", stats.IPv4PrefixDistribution,
[]PrefixDistribution{{MaskLength: 24, Count: 1}})
assertDistribution(t, "seeded IPv6 distribution", stats.IPv6PrefixDistribution,
[]PrefixDistribution{{MaskLength: 32, Count: 1}})
}
// TestPrefixDistributionTracksWrites checks that the prefix distribution the
// stats read reports stays exact across each kind of live-route write, and that
// after every step it equals what the distribution query reads from the route
// tables. The steps run once through the batch methods the prefix handler uses
// and once through the single-route methods.
func TestPrefixDistributionTracksWrites(t *testing.T) {
t.Parallel()
steps := prefixDistributionSteps(t)
for _, batch := range []bool{true, false} {
name := "single-route writes"
if batch {
name = "batch writes"
}
t.Run(name, func(t *testing.T) {
t.Parallel()
db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
ctx := context.Background()
for _, step := range steps {
err := announceRoutes(db, batch, step.announce)
if err != nil {
t.Fatalf("%s: announce: %v", step.name, err)
}
err = withdrawRoutes(db, batch, step.withdraw)
if err != nil {
t.Fatalf("%s: withdraw: %v", step.name, err)
}
stats, err := db.GetStatsContext(ctx)
if err != nil {
t.Fatalf("%s: GetStatsContext: %v", step.name, err)
}
queryV4, queryV6, err := db.GetPrefixDistributionContext(ctx)
if err != nil {
t.Fatalf("%s: GetPrefixDistributionContext: %v", step.name, err)
}
assertDistribution(t, step.name+": IPv4 distribution",
stats.IPv4PrefixDistribution, step.wantV4)
assertDistribution(t, step.name+": IPv6 distribution",
stats.IPv6PrefixDistribution, step.wantV6)
assertDistribution(t, step.name+": IPv4 distribution query",
queryV4, step.wantV4)
assertDistribution(t, step.name+": IPv6 distribution query",
queryV6, step.wantV6)
}
})
}
}
// prefixDistributionStep is one step of TestPrefixDistributionTracksWrites:
// the routes it announces, then the routes it withdraws, then the
// distributions both the stats and the distribution query must report.
type prefixDistributionStep struct {
name string
announce []*LiveRoute
withdraw []*LiveRoute
wantV4 []PrefixDistribution
wantV6 []PrefixDistribution
}
// prefixDistributionSteps returns the steps of
// TestPrefixDistributionTracksWrites, in the order they run.
func prefixDistributionSteps(t *testing.T) []prefixDistributionStep {
t.Helper()
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
shared := mkV4Route(t, "198.51.100.0/24", 64500, ts)
sharedSecondPeer := mkV4Route(t, "198.51.100.0/24", 64500, ts)
sharedSecondPeer.PeerIP = "192.0.2.2"
other := mkV4Route(t, "203.0.113.0/24", 64501, ts)
wide := mkV4Route(t, "172.16.0.0/16", 64502, ts)
v6 := mkV6Route(testV6Prefix, 64503, ts)
v6SecondPeer := mkV6Route(testV6Prefix, 64503, ts)
v6SecondPeer.PeerIP = "2001:db8::2"
all := []PrefixDistribution{{MaskLength: 16, Count: 1}, {MaskLength: 24, Count: 2}}
v6Only := []PrefixDistribution{{MaskLength: 32, Count: 1}}
return []prefixDistributionStep{
{
name: "new routes",
announce: []*LiveRoute{shared, other, wide, v6},
wantV4: all, wantV6: v6Only,
},
{
name: "re-announcement",
announce: []*LiveRoute{shared, other, wide, v6},
wantV4: all, wantV6: v6Only,
},
{
name: "second peer announces a prefix that has a live route",
announce: []*LiveRoute{sharedSecondPeer, v6SecondPeer},
wantV4: all, wantV6: v6Only,
},
{
name: "withdrawal of a route that is not the last for its prefix",
withdraw: []*LiveRoute{shared, v6},
wantV4: all, wantV6: v6Only,
},
{
name: "withdrawal of the last route for a prefix",
withdraw: []*LiveRoute{sharedSecondPeer, v6SecondPeer},
wantV4: []PrefixDistribution{
{MaskLength: 16, Count: 1}, {MaskLength: 24, Count: 1},
},
},
{
name: "withdrawal of every remaining route",
withdraw: []*LiveRoute{other, wide},
},
{
name: "two peers announce a new prefix together",
announce: []*LiveRoute{shared, sharedSecondPeer, v6, v6SecondPeer},
wantV4: []PrefixDistribution{{MaskLength: 24, Count: 1}},
wantV6: v6Only,
},
{
name: "both routes for a prefix withdrawn together",
withdraw: []*LiveRoute{shared, sharedSecondPeer, v6, v6SecondPeer},
},
// The feed often withdraws a route that is not live. That must not take
// the prefix out of the distribution. A count wrongly taken below zero is
// left out of the answer, so the next step shows it: its announcement
// at the same mask length would then not be counted.
{
name: "withdrawal of routes that are not live",
withdraw: []*LiveRoute{other, v6},
},
{
name: "announcement after a withdrawal of routes that are not live",
announce: []*LiveRoute{other, v6},
wantV4: []PrefixDistribution{{MaskLength: 24, Count: 1}},
wantV6: v6Only,
},
}
}
// announceRoutes writes routes in one UpsertLiveRouteBatch, or with one
// UpsertLiveRoute each.
func announceRoutes(db *Database, batch bool, routes []*LiveRoute) error {
if batch {
return db.UpsertLiveRouteBatch(routes)
}
for _, route := range routes {
err := db.UpsertLiveRoute(route)
if err != nil {
return err
}
}
return nil
}
// withdrawRoutes removes routes in one DeleteLiveRouteBatch, or with one
// DeleteLiveRoute each.
func withdrawRoutes(db *Database, batch bool, routes []*LiveRoute) error {
if !batch {
for _, route := range routes {
err := db.DeleteLiveRoute(route.Prefix, route.PeerIP)
if err != nil {
return err
}
}
return nil
}
deletions := make([]LiveRouteDeletion, 0, len(routes))
for _, route := range routes {
deletions = append(deletions, LiveRouteDeletion{
Prefix: route.Prefix, PeerIP: route.PeerIP, IPVersion: route.IPVersion,
})
}
return db.DeleteLiveRouteBatch(deletions)
}
// TestDeleteLiveRoutesOfPeer gives two peers IPv4 and IPv6 live routes, some
// for the same prefixes, and deletes all of one peer's routes. The other peer's
// routes stay, and the route counts and prefix distribution held in memory
// equal what is read from the route tables.
func TestDeleteLiveRoutesOfPeer(t *testing.T) {
t.Parallel()
const (
peerIP = "192.0.2.1"
otherPeerIP = "192.0.2.2"
sharedV4 = "198.51.100.0/24"
sharedV6 = "2001:db8::/32"
)
db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
ofPeer := func(peer string, route *LiveRoute) *LiveRoute {
route.PeerIP = peer
return route
}
onlyV6 := mkV6Route("2001:db8:1::/48", 64503, ts)
onlyV6.MaskLength = 48
err = db.UpsertLiveRouteBatch([]*LiveRoute{
ofPeer(peerIP, mkV4Route(t, sharedV4, 64500, ts)),
ofPeer(otherPeerIP, mkV4Route(t, sharedV4, 64500, ts)),
ofPeer(peerIP, mkV4Route(t, "203.0.113.0/24", 64501, ts)),
ofPeer(peerIP, mkV6Route(sharedV6, 64502, ts)),
ofPeer(otherPeerIP, mkV6Route(sharedV6, 64502, ts)),
ofPeer(peerIP, onlyV6),
assertCounts(t, "after reopen", stats, wantCounts{
asns: 2, prefixes: 2, peerings: 1, peers: 1,
})
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
// peerIP has two IPv4 and two IPv6 routes; a second delete finds none.
for _, want := range []int{4, 0} {
deleted, err := db.DeleteLiveRoutesOfPeer(peerIP)
if err != nil {
t.Fatalf("DeleteLiveRoutesOfPeer: %v", err)
}
if deleted != want {
t.Errorf("DeleteLiveRoutesOfPeer deleted %d routes, want %d",
deleted, want)
}
}
for _, prefix := range []string{sharedV4, sharedV6} {
routes, _, err := db.GetPrefixDetailsContext(context.Background(), prefix)
if err != nil {
t.Fatalf("GetPrefixDetailsContext(%s): %v", prefix, err)
}
if len(routes) != 1 || routes[0].PeerIP != otherPeerIP {
t.Errorf("live routes for %s = %+v, want only the one from %s",
prefix, routes, otherPeerIP)
}
}
checkOneLiveRoutePerIPVersion(t, db)
}
// TestDeleteLiveRoutesOfPeerDeletesInChunks gives a peer one IPv4 route more
// than a chunk holds, two of them for prefixes another peer also has. One chunk
// deletes peerRouteChunkSize of the peer's routes and commits, leaving the last
// one. The routes are written again, and DeleteLiveRoutesOfPeer deletes all of
// them, a chunk and then the last one. After each step the route count and
// prefix distribution held in memory and those read from the route table are
// the ones wanted.
func TestDeleteLiveRoutesOfPeerDeletesInChunks(t *testing.T) {
t.Parallel()
const (
peerIP = "192.0.2.1"
otherPeerIP = "192.0.2.2"
// How many of the peer's prefixes the other peer also has, the first
// ones in prefix order.
sharedPrefixes = 2
)
db, err := New(&config.Config{StateDir: t.TempDir()}, logger.New())
if err != nil {
t.Fatalf("failed to create database: %v", err)
}
defer func() { _ = db.Close() }()
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
peerRoutes := make([]*LiveRoute, 0, peerRouteChunkSize+1)
otherPeerRoutes := make([]*LiveRoute, 0, sharedPrefixes)
for i := range peerRouteChunkSize + 1 {
prefix := fmt.Sprintf("10.%d.%d.0/24", i/256, i%256)
route := mkV4Route(t, prefix, 64500, ts)
route.PeerIP = peerIP
peerRoutes = append(peerRoutes, route)
if i < sharedPrefixes {
route := mkV4Route(t, prefix, 64500, ts)
route.PeerIP = otherPeerIP
otherPeerRoutes = append(otherPeerRoutes, route)
}
}
err = db.UpsertLiveRouteBatch(slices.Concat(peerRoutes, otherPeerRoutes))
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
deletedV4, deletedV6, err := db.deletePeerRouteChunk(peerIP)
if err != nil {
t.Fatalf("deletePeerRouteChunk: %v", err)
}
if deletedV4 != peerRouteChunkSize || deletedV6 != 0 {
t.Errorf("deletePeerRouteChunk deleted (%d, %d) routes, want (%d, 0)",
deletedV4, deletedV6, peerRouteChunkSize)
}
// The shared prefixes went with the chunk, so each route left has a
// prefix of its own.
checkV4RoutesOfOwnPrefixes(t, db, "after one chunk", 1+sharedPrefixes)
err = db.UpsertLiveRouteBatch(peerRoutes)
if err != nil {
t.Fatalf("UpsertLiveRouteBatch: %v", err)
}
deleted, err := db.DeleteLiveRoutesOfPeer(peerIP)
if err != nil {
t.Fatalf("DeleteLiveRoutesOfPeer: %v", err)
}
if deleted != len(peerRoutes) {
t.Errorf("DeleteLiveRoutesOfPeer deleted %d routes, want %d",
deleted, len(peerRoutes))
}
checkV4RoutesOfOwnPrefixes(t, db, "after DeleteLiveRoutesOfPeer",
sharedPrefixes)
}
// checkV4RoutesOfOwnPrefixes checks that the database has wantRoutes IPv4 live
// routes, each for a /24 no other live route has, both as counted in memory and
// as read from the route table. when says which step of the test checks.
func checkV4RoutesOfOwnPrefixes(
t *testing.T, db *Database, when string, wantRoutes int,
) {
t.Helper()
ctx := context.Background()
stats, err := db.GetStatsContext(ctx)
if err != nil {
t.Fatalf("GetStatsContext: %v", err)
}
tableRoutes, _, err := db.GetLiveRouteCountsContext(ctx)
if err != nil {
t.Fatalf("GetLiveRouteCountsContext: %v", err)
}
if stats.IPv4Routes != wantRoutes || tableRoutes != wantRoutes {
t.Errorf("%s: IPv4 live routes: %d in memory, %d in the table, want %d",
when, stats.IPv4Routes, tableRoutes, wantRoutes)
}
queryDistribution, _, err := db.GetPrefixDistributionContext(ctx)
if err != nil {
t.Fatalf("GetPrefixDistributionContext: %v", err)
}
want := []PrefixDistribution{{MaskLength: 24, Count: wantRoutes}}
assertDistribution(t, when+": IPv4 distribution",
stats.IPv4PrefixDistribution, want)
assertDistribution(t, when+": IPv4 distribution query",
queryDistribution, want)
}
// checkOneLiveRoutePerIPVersion checks that the database has one IPv4 live
// route, for a /24, and one IPv6 live route, for a /32, both as counted in
// memory and as read from the route tables.
func checkOneLiveRoutePerIPVersion(t *testing.T, db *Database) {
t.Helper()
ctx := context.Background()
stats, err := db.GetStatsContext(ctx)
if err != nil {
t.Fatalf("GetStatsContext: %v", err)
}
tableV4, tableV6, err := db.GetLiveRouteCountsContext(ctx)
if err != nil {
t.Fatalf("GetLiveRouteCountsContext: %v", err)
}
if stats.IPv4Routes != 1 || tableV4 != 1 || stats.IPv6Routes != 1 ||
tableV6 != 1 {
t.Errorf("live routes (IPv4, IPv6): (%d, %d) in memory, (%d, %d) in "+
"the tables, want (1, 1)",
stats.IPv4Routes, stats.IPv6Routes, tableV4, tableV6)
}
queryV4, queryV6, err := db.GetPrefixDistributionContext(ctx)
if err != nil {
t.Fatalf("GetPrefixDistributionContext: %v", err)
}
wantV4 := []PrefixDistribution{{MaskLength: 24, Count: 1}}
wantV6 := []PrefixDistribution{{MaskLength: 32, Count: 1}}
assertDistribution(t, "IPv4 distribution", stats.IPv4PrefixDistribution, wantV4)
assertDistribution(t, "IPv6 distribution", stats.IPv6PrefixDistribution, wantV6)
assertDistribution(t, "IPv4 distribution query", queryV4, wantV4)
assertDistribution(t, "IPv6 distribution query", queryV6, wantV6)
}
// TestLiveCountsConcurrentReadWrite runs writers and stats readers at once so
@@ -672,7 +145,10 @@ func TestLiveCountsConcurrentReadWrite(t *testing.T) {
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
const writers = 4
const (
writers = 4
prefixesPerWrite = 25
)
var wg sync.WaitGroup
wg.Add(writers)
@@ -681,14 +157,12 @@ func TestLiveCountsConcurrentReadWrite(t *testing.T) {
go func(base int) {
defer wg.Done()
for i := range 25 {
asn := 65000 + base*100 + i
route := mkV6Route(testV6Prefix, asn, ts)
route.PeerIP = "2001:db8::" + uuid.NewString()
for i := range prefixesPerWrite {
prefix := fmt.Sprintf("10.%d.%d.0/24", base, i)
err := db.UpsertLiveRoute(route)
err := db.UpdatePrefixesBatch(map[string]time.Time{prefix: ts})
if err != nil {
t.Errorf("UpsertLiveRoute: %v", err)
t.Errorf("UpdatePrefixesBatch: %v", err)
return
}
@@ -728,22 +202,16 @@ func TestLiveCountsConcurrentReadWrite(t *testing.T) {
t.Fatalf("GetStatsContext: %v", err)
}
if want := writers * 25; stats.IPv6Routes != want {
t.Errorf("IPv6Routes = %d, want %d", stats.IPv6Routes, want)
if want := writers * prefixesPerWrite; stats.IPv4Prefixes != want {
t.Errorf("IPv4Prefixes = %d, want %d", stats.IPv4Prefixes, want)
}
// Every writer announced the same prefix, so it counts once.
assertDistribution(t, "IPv6 distribution", stats.IPv6PrefixDistribution,
[]PrefixDistribution{{MaskLength: 32, Count: 1}})
}
type wantCounts struct {
asns int
prefixes int
peerings int
peers int
ipv4Routes int
ipv6Routes int
liveRoutes int
asns int
prefixes int
peerings int
peers int
}
func assertCounts(t *testing.T, when string, got Stats, want wantCounts) {
@@ -764,24 +232,4 @@ func assertCounts(t *testing.T, when string, got Stats, want wantCounts) {
if got.Peers != want.peers {
t.Errorf("%s: Peers = %d, want %d", when, got.Peers, want.peers)
}
if got.IPv4Routes != want.ipv4Routes {
t.Errorf("%s: IPv4Routes = %d, want %d", when, got.IPv4Routes, want.ipv4Routes)
}
if got.IPv6Routes != want.ipv6Routes {
t.Errorf("%s: IPv6Routes = %d, want %d", when, got.IPv6Routes, want.ipv6Routes)
}
if got.LiveRoutes != want.liveRoutes {
t.Errorf("%s: LiveRoutes = %d, want %d", when, got.LiveRoutes, want.liveRoutes)
}
}
func assertDistribution(t *testing.T, what string, got, want []PrefixDistribution) {
t.Helper()
if !slices.Equal(got, want) {
t.Errorf("%s = %v, want %v", what, got, want)
}
}
internal/database/database.go
+19 -1452
View File
File diff suppressed because it is too large Load Diff
internal/database/database_internal_test.go
+55 -1270
View File
File diff suppressed because it is too large Load Diff
internal/database/interface.go
+14 -47
View File
@@ -1,7 +1,6 @@
// Package database provides SQLite storage for BGP routing data including ASNs,
// prefixes, announcements, peerings, and live route tables. Its routes table
// holds a copy of the whole routing table that internal/routetable keeps in
// memory.
// 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.
package database
import (
@@ -12,26 +11,21 @@ import (
"git.eeqj.de/sneak/routewatch/internal/routetable"
)
// Stats contains database statistics including counts for ASNs, prefixes,
// peerings, peers, and live routes, as well as file size and prefix distribution data.
// Stats contains database statistics: counts of ASNs, prefixes, peerings and
// peers, and the file size.
type Stats struct {
ASNs int
Prefixes int
IPv4Prefixes int
IPv6Prefixes int
Peerings int
Peers int
FileSizeBytes int64
LiveRoutes int
IPv4Routes int
IPv6Routes int
IPv4PrefixDistribution []PrefixDistribution
IPv6PrefixDistribution []PrefixDistribution
ASNs int
Prefixes int
IPv4Prefixes int
IPv6Prefixes int
Peerings int
Peers int
FileSizeBytes int64
}
// Store defines the interface for database operations. It provides methods for
// managing ASNs, prefixes, announcements, peerings, BGP peers, and live routes.
// Implementations must be safe for concurrent use.
// managing ASNs, prefixes, announcements, peerings, BGP peers, and the copy of
// the routing table. Implementations must be safe for concurrent use.
//
//nolint:interfacebloat // the one seam handlers and tests depend on
type Store interface {
@@ -57,27 +51,10 @@ type Store interface {
UpdatePeer(peerIP string, peerASN int, messageType string, timestamp time.Time) error
UpdatePeerBatch(peers map[string]PeerUpdate) error
// Live route operations
UpsertLiveRoute(route *LiveRoute) error
UpsertLiveRouteBatch(routes []*LiveRoute) error
DeleteLiveRoute(prefix, peerIP string) error
DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error
DeleteLiveRoutesOfPeer(peerIP string) (deleted int, err error)
GetPrefixDistribution() (
ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)
GetPrefixDistributionContext(ctx context.Context) (
ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)
GetLiveRouteCounts() (ipv4Count, ipv6Count int, err error)
GetLiveRouteCountsContext(ctx context.Context) (ipv4Count, ipv6Count int, err error)
// Routing table operations: the database's copy of the table in memory
WriteRouteChanges(changes []routetable.Change, newPeers map[string]uint) error
ReadRouteTable(ctx context.Context) (*routetable.Table, error)
// IP lookup operations
GetIPInfo(ip string) (*IPInfo, error)
GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, error)
// ASN WHOIS operations
GetNextStaleASN(ctx context.Context, staleThreshold time.Duration) (int, error)
UpdateASNWHOIS(ctx context.Context, update *ASNWHOISUpdate) error
@@ -87,18 +64,8 @@ type Store interface {
GetAS(ctx context.Context, asn int) (*ASN, bool, error)
GetASWHOIS(ctx context.Context, asn int) (*ASN, error)
GetPrefixFirstSeen(ctx context.Context, prefix netip.Prefix) (time.Time, error)
GetASDetails(asn int) (*ASN, []LiveRoute, error)
GetASDetailsContext(ctx context.Context, asn int) (*ASN, []LiveRoute, error)
GetASPeers(asn int) ([]ASPeer, error)
GetASPeersContext(ctx context.Context, asn int) ([]ASPeer, error)
GetPrefixDetails(prefix string) ([]LiveRoute, map[int]ASN, error)
GetPrefixDetailsContext(
ctx context.Context, prefix string,
) ([]LiveRoute, map[int]ASN, error)
GetRandomPrefixesByLength(maskLength, ipVersion, limit int) ([]LiveRoute, error)
GetRandomPrefixesByLengthContext(
ctx context.Context, maskLength, ipVersion, limit int,
) ([]LiveRoute, error)
// Lifecycle
Close() error
internal/database/models.go
-7
View File
@@ -118,13 +118,6 @@ type IPInfo struct {
NeedsWHOISRefresh bool `json:"-"`
}
// LiveRouteDeletion represents parameters for deleting a live route
type LiveRouteDeletion struct {
Prefix string
PeerIP string
IPVersion int
}
// SeenTimes is the earliest and the latest time one batch of updates saw an AS,
// a peer or a peering. A row the batch adds is stored first seen at First and
// last seen at Last. A row already stored takes First as its first_seen only
internal/database/schema.sql
-59
View File
@@ -89,65 +89,6 @@ CREATE INDEX IF NOT EXISTS idx_bgp_peers_asn ON bgp_peers(peer_asn);
CREATE INDEX IF NOT EXISTS idx_bgp_peers_last_seen ON bgp_peers(last_seen);
CREATE INDEX IF NOT EXISTS idx_bgp_peers_ip ON bgp_peers(peer_ip);
-- IPv4 routing table maintained by PrefixHandler. A route write changes the
-- table and its entry in every index on a column the write sets, so the table
-- has only the indexes that the queries named with them use.
CREATE TABLE IF NOT EXISTS live_routes_v4 (
-- The row id, which the prefix-length pages' sample draws through the
-- mask_length index. Declared as a column, it stays the same for as long
-- as the route is live, also through a VACUUM.
id INTEGER PRIMARY KEY,
prefix TEXT NOT NULL,
mask_length INTEGER NOT NULL, -- CIDR mask length (0-32)
origin_asn INTEGER NOT NULL,
peer_ip TEXT NOT NULL,
as_path TEXT NOT NULL, -- JSON array
next_hop TEXT NOT NULL,
last_updated DATETIME NOT NULL,
-- A peer has at most one route for a prefix. This key finds the route a
-- write updates or deletes and, peer_ip coming first, all of a peer's
-- routes when its BGP session goes down.
UNIQUE(peer_ip, prefix)
);
-- IPv6 routing table maintained by PrefixHandler, as live_routes_v4.
CREATE TABLE IF NOT EXISTS live_routes_v6 (
id INTEGER PRIMARY KEY,
prefix TEXT NOT NULL,
mask_length INTEGER NOT NULL, -- CIDR mask length (0-128)
origin_asn INTEGER NOT NULL,
peer_ip TEXT NOT NULL,
as_path TEXT NOT NULL, -- JSON array
next_hop TEXT NOT NULL,
last_updated DATETIME NOT NULL,
UNIQUE(peer_ip, prefix)
);
-- Indexes for both live route tables, and the queries that use each.
--
-- prefix, last_updated: a prefix's routes, for the prefix page and its API
-- answer, the IP lookup and its peer count, and the check whether a prefix
-- still has a live route; and a prefix's newest route for the prefix-length
-- pages and the AS page, in one search whatever the number of peers.
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_prefix_last_updated ON live_routes_v4(prefix, last_updated);
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_prefix_last_updated ON live_routes_v6(prefix, last_updated);
-- mask_length: the prefix-length pages' random sample, the row id range of a
-- mask length and the first route of it at or after a row id.
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_length ON live_routes_v4(mask_length);
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_length ON live_routes_v6(mask_length);
-- mask_length, prefix: the prefix-length pages' list of a mask length's
-- prefixes in text order, and the prefix distribution read at startup.
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_mask_prefix ON live_routes_v4(mask_length, prefix);
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_mask_prefix ON live_routes_v6(mask_length, prefix);
-- origin_asn, prefix: the prefixes an AS originates, each in one search
-- whatever the number of peers, for the AS page and its API answer.
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_origin_prefix ON live_routes_v4(origin_asn, prefix);
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_origin_prefix ON live_routes_v6(origin_asn, prefix);
-- last_updated: the oldest and newest route times in /api/v1/stats and
-- /status.json.
CREATE INDEX IF NOT EXISTS idx_live_routes_v4_last_updated ON live_routes_v4(last_updated);
CREATE INDEX IF NOT EXISTS idx_live_routes_v6_last_updated ON live_routes_v6(last_updated);
-- The whole routing table, as internal/routetable keeps it in memory: one row
-- for each prefix and origin AS that some peer carries. Startup reads every
-- row, and the rows of the pairs that change are written back.
internal/database/utils.go
-14
View File
@@ -1,8 +1,6 @@
package database
import (
"fmt"
"net"
"strings"
"github.com/google/uuid"
@@ -20,15 +18,3 @@ func detectIPVersion(prefix string) int {
return ipVersionV4
}
// prefixMaskLength returns the mask length of a prefix such as 192.0.2.0/24.
func prefixMaskLength(prefix string) (int, error) {
_, network, err := net.ParseCIDR(prefix)
if err != nil {
return 0, fmt.Errorf("invalid prefix %s: %w", prefix, err)
}
maskLength, _ := network.Mask.Size()
return maskLength, nil
}