Read every route from the routing table in memory (part of #122)

The IP lookups, the AS and prefix pages and their API answers, the prefix-length pages and the stats read routes only from the routing table the prefix handler keeps. fx reads the table at startup and gives it to both. AS names, WHOIS details and first-seen times still come from the database.

The IP lookup names the origin AS most peers carry, with the time of the newest kept route. The prefix answers list peers whose routes are not kept without a path or time. The stats gain origins_without_kept_route. They no longer read route times from the per-peer tables, so the two helpers that did are gone.

Model: opus-5-5

internal/routewatch/app.go

@@ -31,11 +31,13 @@ import (
 // are disabled, since only the batched handlers exist.
 var errNonBatchedNotImplemented = errors.New("non-batched handlers not implemented")
 
-// Dependencies contains all dependencies for RouteWatch
+// Dependencies contains all dependencies for RouteWatch. Table is the routing
+// table read from the database at startup, which the server reads too.
 type Dependencies struct {
 	fx.In
 
 	DB       database.Store
+	Table    *routetable.Table
 	Streamer *streamer.Streamer
 	Server   *server.Server
 	Logger   *logger.Logger
@@ -45,6 +47,7 @@ type Dependencies struct {
 // RouteWatch represents the main application instance
 type RouteWatch struct {
 	db             database.Store
+	table          *routetable.Table
 	streamer       *streamer.Streamer
 	server         *server.Server
 	logger         *logger.Logger
@@ -69,6 +72,7 @@ const blocklistsDirName = "blocklists"
 func New(deps Dependencies) *RouteWatch {
 	rw := &RouteWatch{
 		db:         deps.DB,
+		table:      deps.Table,
 		streamer:   deps.Streamer,
 		server:     deps.Server,
 		logger:     deps.Logger,
@@ -97,13 +101,6 @@ func (rw *RouteWatch) Run(ctx context.Context) error {
 	if rw.config.EnableBatchedDatabaseWrites {
 		rw.logger.Info("Using batched database handlers for improved performance")
 
-		// The routing table is read before the streamer connects, so that
-		// the prefix handler applies the feed's updates to all of it.
-		table, err := rw.readRouteTable(ctx)
-		if err != nil {
-			return err
-		}
-
 		// ASHandler maintains the asns table
 		rw.dbHandler = NewASHandler(rw.db, rw.logger)
 		rw.streamer.RegisterHandler(rw.dbHandler)
@@ -114,7 +111,7 @@ func (rw *RouteWatch) Run(ctx context.Context) error {
 
 		// PrefixHandler maintains the prefixes and live_routes tables, and
 		// the routing table in memory and its copy in the routes table
-		rw.prefixHandler = NewPrefixHandler(rw.db, rw.logger, table)
+		rw.prefixHandler = NewPrefixHandler(rw.db, rw.logger, rw.table)
 		rw.prefixHandler.SetMetricsTracker(rw.streamer.GetMetricsTracker())
 		rw.streamer.RegisterHandler(rw.prefixHandler)
 
@@ -269,20 +266,21 @@ func (rw *RouteWatch) closeDatabase(ctx context.Context) {
 
 // readRouteTable reads the routing table from the database and logs how many
 // prefix and origin AS pairs and peers it holds and how long the read took.
-// ctx is fx's start context, which ends 15 seconds after the start. Reading a
-// full table can take longer, so the read does not stop with ctx.
-func (rw *RouteWatch) readRouteTable(
-	ctx context.Context,
+// fx calls it once, to build the server and RouteWatch, before the streamer
+// connects, so the prefix handler applies the feed's updates to all of it. No
+// start timeout limits the read: fx builds them before it runs a start hook.
+func readRouteTable(
+	db database.Store, logger *logger.Logger,
 ) (*routetable.Table, error) {
 	start := time.Now()
 
-	table, err := rw.db.ReadRouteTable(context.WithoutCancel(ctx))
+	table, err := db.ReadRouteTable(context.Background())
 	if err != nil {
 		return nil, fmt.Errorf("failed to read the routing table: %w", err)
 	}
 
 	counts := table.Counts()
-	rw.logger.Info("Read the routing table", "pairs", counts.Pairs,
+	logger.Info("Read the routing table", "pairs", counts.Pairs,
 		"peers", counts.Peers, "duration_ms", time.Since(start).Milliseconds())
 
 	return table, nil
@@ -299,6 +297,7 @@ func getModule() []fx.Option {
 				database.New,
 				fx.As(new(database.Store)),
 			),
+			readRouteTable,
 			streamer.New,
 			server.New,
 			New,

internal/routewatch/app_integration_internal_test.go

@@ -7,6 +7,7 @@ import (
 	"net"
 	"net/http"
 	"net/http/httptest"
+	"net/netip"
 	"slices"
 	"strconv"
 	"strings"
@@ -265,6 +266,32 @@ func (m *mockStore) GetASDetailsContext(
 	return m.GetASDetails(asn)
 }
 
+// GetAS returns the stored AS, or only its number when none is stored.
+func (m *mockStore) GetAS(_ context.Context, asn int) (*database.ASN, bool, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if asnInfo, exists := m.ASNs[asn]; exists {
+		return asnInfo, true, nil
+	}
+
+	return &database.ASN{ASN: asn}, false, nil
+}
+
+// GetASWHOIS returns the stored AS, or only its number when none is stored.
+func (m *mockStore) GetASWHOIS(ctx context.Context, asn int) (*database.ASN, error) {
+	asnInfo, _, err := m.GetAS(ctx, asn)
+
+	return asnInfo, err
+}
+
+// GetPrefixFirstSeen returns the zero time: the mock keeps no prefix times.
+func (m *mockStore) GetPrefixFirstSeen(
+	_ context.Context, _ netip.Prefix,
+) (time.Time, error) {
+	return time.Time{}, nil
+}
+
 // GetPrefixDetails mock implementation
 func (m *mockStore) GetPrefixDetails(
 	_ string,
@@ -502,12 +529,15 @@ func TestRouteWatchLiveFeed(t *testing.T) {
 		EnableBatchedDatabaseWrites: true,
 	}
 
+	table := routetable.New(routetable.Params{})
+
 	// Create server
-	srv := server.New(mockDB, s, logger, cfg)
+	srv := server.New(mockDB, table, s, logger, cfg)
 
 	// Create RouteWatch with 5 second limit
 	deps := Dependencies{
 		DB:       mockDB,
+		Table:    table,
 		Streamer: s,
 		Server:   srv,
 		Logger:   logger,
@@ -626,7 +656,7 @@ func TestShutdownWritesTheUpdatesQueuedInTheStreamer(t *testing.T) {
 	rw := New(Dependencies{
 		DB:       store,
 		Streamer: s,
-		Server:   server.New(store, s, log, cfg),
+		Server:   server.New(store, nil, s, log, cfg),
 		Logger:   log,
 		Config:   cfg,
 	})
@@ -728,7 +758,7 @@ func TestShutdownReturnsByItsDeadlineWhileAWriteIsBlocked(t *testing.T) {
 	rw := New(Dependencies{
 		DB:       store,
 		Streamer: s,
-		Server:   server.New(store, s, log, cfg),
+		Server:   server.New(store, nil, s, log, cfg),
 		Logger:   log,
 		Config:   cfg,
 	})
@@ -842,7 +872,7 @@ func TestShutdownCheckpointsAndClosesTheDatabaseLast(t *testing.T) {
 	rw := New(Dependencies{
 		DB:       store,
 		Streamer: s,
-		Server:   server.New(store, s, log, cfg),
+		Server:   server.New(store, nil, s, log, cfg),
 		Logger:   log,
 		Config:   cfg,
 	})

internal/routewatch/prefixhandler_table_internal_test.go

@@ -416,12 +416,8 @@ func TestARestartReadsTheRoutingTableBack(t *testing.T) {
 
 	var logs bytes.Buffer
 
-	rw := &RouteWatch{
-		db:     db,
-		logger: &logger.Logger{Logger: slog.New(slog.NewJSONHandler(&logs, nil))},
-	}
-
-	table, err := rw.readRouteTable(t.Context())
+	table, err := readRouteTable(db,
+		&logger.Logger{Logger: slog.New(slog.NewJSONHandler(&logs, nil))})
 	if err != nil {
 		t.Fatalf("readRouteTable: %v", err)
 	}