Identify a live route by prefix and peer (closes #55)

A peer has at most one route for a prefix, but the live route tables were
keyed by prefix, origin AS and peer. An announcement with a new origin AS
added a second row instead of replacing the first, and a withdrawal sent in
the same message as announcements was given their origin AS and so missed
the withdrawn route.

The tables are now unique on prefix and peer. An announcement updates the
peer's row for the prefix, origin AS included, and a withdrawal deletes it
without looking at an origin AS. The origin-only delete statements, the
deletion's origin field and DeleteLiveRoute's origin parameter are removed.

Model: opus-5-5

internal/database/counts_test.go

@@ -128,7 +128,7 @@ func TestLiveCountsTrackWritesInRealtime(t *testing.T) {
 
 	// A withdrawal removes one route.
 	if err := db.DeleteLiveRouteBatch([]LiveRouteDeletion{
-		{Prefix: "203.0.113.0/24", OriginASN: 64501, PeerIP: "192.0.2.1", IPVersion: ipVersionV4},
+		{Prefix: "203.0.113.0/24", PeerIP: "192.0.2.1", IPVersion: ipVersionV4},
 	}); err != nil {
 		t.Fatalf("DeleteLiveRouteBatch: %v", err)
 	}
@@ -327,13 +327,11 @@ func announceRoutes(db *Database, batch bool, routes []*LiveRoute) error {
 }
 
 // withdrawRoutes removes routes in one DeleteLiveRouteBatch, or with one
-// DeleteLiveRoute each. It names each route by prefix and peer only, with no
-// origin ASN, as a withdrawal from the feed does when its message carries no AS
-// path.
+// DeleteLiveRoute each.
 func withdrawRoutes(db *Database, batch bool, routes []*LiveRoute) error {
 	if !batch {
 		for _, route := range routes {
-			if err := db.DeleteLiveRoute(route.Prefix, 0, route.PeerIP); err != nil {
+			if err := db.DeleteLiveRoute(route.Prefix, route.PeerIP); err != nil {
 				return err
 			}
 		}

internal/database/database.go

@@ -228,21 +228,22 @@ func (d *Database) beginTx() (*loggingTx, error) {
 }
 
 // A live-route upsert is an UPDATE followed, only when no row matched, by an
-// INSERT. The UPDATE's rows-affected count (1 for an existing key, 0 for a new
-// one) is what lets the in-memory route counters stay exact without a COUNT(*).
-// Callers hold the database write lock, so no other writer can insert the same
-// key between the two statements. The id column is set only on INSERT, so an
-// updated route keeps its original id, exactly as the previous ON CONFLICT
-// upsert did.
+// INSERT. A live route is keyed by prefix and peer, so an announcement replaces
+// the peer's route for the prefix whatever its origin AS. The UPDATE's
+// rows-affected count (1 for an existing key, 0 for a new one) is what lets the
+// in-memory route counters stay exact without a COUNT(*). Callers hold the
+// database write lock, so no other writer can insert the same key between the
+// two statements. The id column is set only on INSERT, so an updated route
+// keeps its original id, exactly as the previous ON CONFLICT upsert did.
 const (
-	updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, as_path = ?, next_hop = ?,
-		last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
+	updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, origin_asn = ?, as_path = ?,
+		next_hop = ?, last_updated = ? WHERE prefix = ? AND peer_ip = ?`
 
 	insertLiveRouteV4SQL = `INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, peer_ip,
 		as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
 
-	updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, as_path = ?, next_hop = ?,
-		last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
+	updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, origin_asn = ?, as_path = ?,
+		next_hop = ?, last_updated = ? WHERE prefix = ? AND peer_ip = ?`
 
 	insertLiveRouteV6SQL = `INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip,
 		as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
@@ -261,8 +262,8 @@ const (
 // live route for its prefix.
 func upsertRouteRowV4(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) (
 	inserted, newPrefix bool, err error) {
-	res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
-		route.Prefix, route.OriginASN, route.PeerIP)
+	res, err := upd.Exec(route.MaskLength, route.OriginASN, pathJSON, route.NextHop, route.LastUpdated,
+		route.Prefix, route.PeerIP)
 	if err != nil {
 		return false, false, err
 	}
@@ -293,8 +294,8 @@ func upsertRouteRowV4(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string
 // live route for its prefix.
 func upsertRouteRowV6(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) (
 	inserted, newPrefix bool, err error) {
-	res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
-		route.Prefix, route.OriginASN, route.PeerIP)
+	res, err := upd.Exec(route.MaskLength, route.OriginASN, pathJSON, route.NextHop, route.LastUpdated,
+		route.Prefix, route.PeerIP)
 	if err != nil {
 		return false, false, err
 	}
@@ -441,32 +442,18 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
 		}
 	}()
 
-	// No longer need to separate deletions since we handle them in the loop below
-
 	// Prepare statements for both IPv4 and IPv6 tables
-	stmtV4WithOrigin, err := tx.Prepare(`DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`)
+	deleteV4, err := tx.Prepare(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`)
 	if err != nil {
-		return fmt.Errorf("failed to prepare IPv4 delete with origin statement: %w", err)
+		return fmt.Errorf("failed to prepare IPv4 delete statement: %w", err)
 	}
-	defer func() { _ = stmtV4WithOrigin.Close() }()
+	defer func() { _ = deleteV4.Close() }()
 
-	stmtV4WithoutOrigin, err := tx.Prepare(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`)
+	deleteV6, err := tx.Prepare(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`)
 	if err != nil {
-		return fmt.Errorf("failed to prepare IPv4 delete without origin statement: %w", err)
+		return fmt.Errorf("failed to prepare IPv6 delete statement: %w", err)
 	}
-	defer func() { _ = stmtV4WithoutOrigin.Close() }()
-
-	stmtV6WithOrigin, err := tx.Prepare(`DELETE FROM live_routes_v6 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`)
-	if err != nil {
-		return fmt.Errorf("failed to prepare IPv6 delete with origin statement: %w", err)
-	}
-	defer func() { _ = stmtV6WithOrigin.Close() }()
-
-	stmtV6WithoutOrigin, err := tx.Prepare(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`)
-	if err != nil {
-		return fmt.Errorf("failed to prepare IPv6 delete without origin statement: %w", err)
-	}
-	defer func() { _ = stmtV6WithoutOrigin.Close() }()
+	defer func() { _ = deleteV6.Close() }()
 
 	hasV4, err := tx.Prepare(prefixHasLiveRouteV4SQL)
 	if err != nil {
@@ -485,38 +472,18 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
 	// Mask lengths of the prefixes this batch leaves with no live route.
 	var gonePrefixMaskLengthsV4, gonePrefixMaskLengthsV6 []int
 	for _, del := range deletions {
-		var stmt *sql.Stmt
-
-		// Select appropriate statement based on IP version and whether we have origin ASN
-		//nolint:nestif // Clear logic for selecting the right statement
-		if del.IPVersion == ipVersionV4 {
-			if del.OriginASN == 0 {
-				stmt = stmtV4WithoutOrigin
-			} else {
-				stmt = stmtV4WithOrigin
-			}
-		} else {
-			if del.OriginASN == 0 {
-				stmt = stmtV6WithoutOrigin
-			} else {
-				stmt = stmtV6WithOrigin
-			}
-		}
-
-		// Execute deletion
-		var res sql.Result
-		if del.OriginASN == 0 {
-			res, err = stmt.Exec(del.Prefix, del.PeerIP)
-		} else {
-			res, err = stmt.Exec(del.Prefix, del.OriginASN, del.PeerIP)
+		stmt := deleteV4
+		if del.IPVersion != ipVersionV4 {
+			stmt = deleteV6
 		}
 
+		res, err := stmt.Exec(del.Prefix, del.PeerIP)
 		if err != nil {
 			return fmt.Errorf("failed to delete route %s: %w", del.Prefix, err)
 		}
 
-		// A deletion with no origin ASN can remove several rows, so use the
-		// exact rows-affected count to keep the in-memory route counters right.
+		// The feed often withdraws a route that is not live, which deletes
+		// nothing, so the in-memory route counters use the rows-affected count.
 		affected, err := res.RowsAffected()
 		if err != nil {
 			return fmt.Errorf("failed to count deleted route %s: %w", del.Prefix, err)
@@ -1282,9 +1249,8 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
 	return nil
 }
 
-// DeleteLiveRoute deletes a live route
-// If originASN is 0, deletes all routes for the prefix/peer combination
-func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string) error {
+// DeleteLiveRoute deletes the live route the peer has for the prefix
+func (d *Database) DeleteLiveRoute(prefix, peerIP string) error {
 	d.lock("DeleteLiveRoute")
 	defer d.unlock()
 
@@ -1309,22 +1275,12 @@ func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string)
 	}()
 
 	// Literal per-table queries (rather than one formatted with the table name)
-	// so the delete carries no dynamically built SQL. A delete with no origin
-	// ASN can remove several rows.
+	// so the delete carries no dynamically built SQL.
 	var res sql.Result
-	switch {
-	case isV4 && originASN == 0:
+	if isV4 {
 		res, err = tx.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
-	case isV4:
-		res, err = tx.Exec(
-			`DELETE FROM live_routes_v4 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`,
-			prefix, originASN, peerIP)
-	case originASN == 0:
+	} else {
 		res, err = tx.Exec(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
-	default:
-		res, err = tx.Exec(
-			`DELETE FROM live_routes_v6 WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`,
-			prefix, originASN, peerIP)
 	}
 	if err != nil {
 		return err

internal/database/database_test.go

@@ -122,8 +122,8 @@ func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) {
 	lookup("2001:db9::1", "")
 
 	err = db.DeleteLiveRouteBatch([]LiveRouteDeletion{
-		{Prefix: "10.1.2.0/24", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV4},
-		{Prefix: "2001:db8:1:2::/64", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV6},
+		{Prefix: "10.1.2.0/24", PeerIP: testPeerIP, IPVersion: ipVersionV4},
+		{Prefix: "2001:db8:1:2::/64", PeerIP: testPeerIP, IPVersion: ipVersionV6},
 	})
 	if err != nil {
 		t.Fatalf("DeleteLiveRouteBatch: %v", err)
@@ -306,7 +306,6 @@ func TestVacuumReturnsFreePages(t *testing.T) {
 		routes = append(routes, route)
 		deletions = append(deletions, LiveRouteDeletion{
 			Prefix:    route.Prefix,
-			OriginASN: asn,
 			PeerIP:    route.PeerIP,
 			IPVersion: ipVersionV4,
 		})

internal/database/interface.go

@@ -55,7 +55,7 @@ type Store interface {
 	// Live route operations
 	UpsertLiveRoute(route *LiveRoute) error
 	UpsertLiveRouteBatch(routes []*LiveRoute) error
-	DeleteLiveRoute(prefix string, originASN int, peerIP string) error
+	DeleteLiveRoute(prefix, peerIP string) error
 	DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error
 	GetPrefixDistribution() (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)
 	GetPrefixDistributionContext(ctx context.Context) (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)

internal/database/models.go

@@ -114,7 +114,6 @@ type IPInfo struct {
 // LiveRouteDeletion represents parameters for deleting a live route
 type LiveRouteDeletion struct {
 	Prefix    string
-	OriginASN int
 	PeerIP    string
 	IPVersion int
 }

internal/database/schema.sql

@@ -107,7 +107,7 @@ CREATE TABLE IF NOT EXISTS live_routes_v4 (
 	as_path TEXT NOT NULL, -- JSON array
 	next_hop TEXT NOT NULL,
 	last_updated DATETIME NOT NULL,
-	UNIQUE(prefix, origin_asn, peer_ip)
+	UNIQUE(prefix, peer_ip) -- a peer has at most one route for a prefix
 );
 
 -- IPv6 routing table maintained by PrefixHandler
@@ -120,7 +120,7 @@ CREATE TABLE IF NOT EXISTS live_routes_v6 (
 	as_path TEXT NOT NULL, -- JSON array
 	next_hop TEXT NOT NULL,
 	last_updated DATETIME NOT NULL,
-	UNIQUE(prefix, origin_asn, peer_ip)
+	UNIQUE(prefix, peer_ip) -- a peer has at most one route for a prefix
 );
 
 -- Indexes for live_routes_v4 table

internal/routewatch/app_integration_test.go

@@ -205,7 +205,7 @@ func (m *mockStore) UpsertLiveRoute(route *database.LiveRoute) error {
 }
 
 // DeleteLiveRoute mock implementation
-func (m *mockStore) DeleteLiveRoute(prefix string, originASN int, peerIP string) error {
+func (m *mockStore) DeleteLiveRoute(prefix, peerIP string) error {
 	// Simple mock - just return nil
 	return nil
 }

internal/routewatch/prefixhandler.go

@@ -108,14 +108,14 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	// Withdrawals are queued before announcements, because the batch keeps
 	// each peer's last update for a prefix: a message that lists a prefix as
 	// both withdrawn and announced keeps the route (RFC 4271 section 4.3).
+	// A withdrawal names only the prefix and peer: the message's AS path, and
+	// so its origin AS, belongs to the announced prefixes.
 	for _, prefix := range msg.Withdrawals {
 		h.batch = append(h.batch, prefixUpdate{
 			prefix:      canonicalPrefix(prefix),
-			originASN:   originASN, // Use the originASN from path if available
 			peer:        msg.Peer,
 			messageType: "withdrawal",
 			timestamp:   timestamp,
-			path:        msg.Path,
 		})
 		// Record withdrawal in metrics
 		if h.metrics != nil {
@@ -221,7 +221,6 @@ func (h *PrefixHandler) flushBatchLocked() {
 			// Create deletion record for batch delete
 			routesToDelete = append(routesToDelete, database.LiveRouteDeletion{
 				Prefix:    update.prefix,
-				OriginASN: update.originASN,
 				PeerIP:    update.peer,
 				IPVersion: ipVersion,
 			})
@@ -425,51 +424,24 @@ func (h *PrefixHandler) processAnnouncementDirect(update prefixUpdate) {
 // processWithdrawalDirect handles removing a route directly without prefix table
 // nolint:unused // kept for potential future use
 func (h *PrefixHandler) processWithdrawalDirect(update prefixUpdate) {
-	// For withdrawals, we need to delete the route from live_routes
-	if update.originASN > 0 {
-		if err := h.db.DeleteLiveRoute(update.prefix, update.originASN, update.peer); err != nil {
-			h.logger.Error("Failed to delete live route",
-				"prefix", update.prefix,
-				"origin_asn", update.originASN,
-				"peer", update.peer,
-				"error", err,
-			)
-		}
-	} else {
-		// If no origin ASN, just delete all routes for this prefix from this peer
-		if err := h.db.DeleteLiveRoute(update.prefix, 0, update.peer); err != nil {
-			h.logger.Error("Failed to delete live route (no origin ASN)",
-				"prefix", update.prefix,
-				"peer", update.peer,
-				"error", err,
-			)
-		}
+	if err := h.db.DeleteLiveRoute(update.prefix, update.peer); err != nil {
+		h.logger.Error("Failed to delete live route",
+			"prefix", update.prefix,
+			"peer", update.peer,
+			"error", err,
+		)
 	}
 }
 
 // processWithdrawal handles removing a route from the live routing table
 // nolint:unused // kept for potential future use
 func (h *PrefixHandler) processWithdrawal(_ *database.Prefix, update prefixUpdate) {
-	// For withdrawals, we need to delete the route from live_routes
-	// Since we have the origin ASN from the update, we can delete the specific route
-	if update.originASN > 0 {
-		if err := h.db.DeleteLiveRoute(update.prefix, update.originASN, update.peer); err != nil {
-			h.logger.Error("Failed to delete live route",
-				"prefix", update.prefix,
-				"origin_asn", update.originASN,
-				"peer", update.peer,
-				"error", err,
-			)
-		}
-	} else {
-		// If no origin ASN, just delete all routes for this prefix from this peer
-		if err := h.db.DeleteLiveRoute(update.prefix, 0, update.peer); err != nil {
-			h.logger.Error("Failed to delete live route (no origin ASN)",
-				"prefix", update.prefix,
-				"peer", update.peer,
-				"error", err,
-			)
-		}
+	if err := h.db.DeleteLiveRoute(update.prefix, update.peer); err != nil {
+		h.logger.Error("Failed to delete live route",
+			"prefix", update.prefix,
+			"peer", update.peer,
+			"error", err,
+		)
 	}
 }
 

internal/routewatch/prefixhandler_test.go

@@ -182,32 +182,139 @@ func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) {
 				t.Errorf("peers with a live route for %s = %v, want %v", prefix, gotPeers, tt.wantPeers)
 			}
 
-			stats, err := db.GetStatsContext(ctx)
-			if err != nil {
-				t.Fatalf("GetStatsContext: %v", err)
-			}
-			tableRoutesV4, _, err := db.GetLiveRouteCountsContext(ctx)
-			if err != nil {
-				t.Fatalf("GetLiveRouteCountsContext: %v", err)
-			}
-			if stats.IPv4Routes != len(tt.wantPeers) || tableRoutesV4 != len(tt.wantPeers) {
-				t.Errorf("IPv4 live routes: %d in memory, %d in the table, want %d",
-					stats.IPv4Routes, tableRoutesV4, len(tt.wantPeers))
-			}
-
 			var wantDistribution []database.PrefixDistribution
 			if len(tt.wantPeers) > 0 {
 				wantDistribution = []database.PrefixDistribution{{MaskLength: prefixMaskLength, Count: 1}}
 			}
-			queryDistributionV4, _, err := db.GetPrefixDistributionContext(ctx)
-			if err != nil {
-				t.Fatalf("GetPrefixDistributionContext: %v", err)
-			}
-			if !slices.Equal(stats.IPv4PrefixDistribution, wantDistribution) ||
-				!slices.Equal(queryDistributionV4, wantDistribution) {
-				t.Errorf("IPv4 prefix distribution: %v in memory, %v from the table, want %v",
-					stats.IPv4PrefixDistribution, queryDistributionV4, wantDistribution)
-			}
+			assertIPv4Counts(t, db, len(tt.wantPeers), wantDistribution)
 		})
 	}
 }
+
+// TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer runs updates from one peer
+// through the prefix handler into a real database, flushing each message as its
+// own batch. A peer has at most one route for a prefix: a later announcement
+// replaces it whatever its origin AS, and a withdrawal removes it whatever AS
+// the message's AS path ends in. The in-memory route count and prefix
+// distribution must match the route table.
+func TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer(t *testing.T) {
+	const (
+		prefix      = "198.51.100.0/24"
+		otherPrefix = "203.0.113.0/24"
+		// The mask length of both prefixes.
+		prefixMaskLength = 24
+	)
+
+	ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
+	announce := func(path ristypes.ASPath, announced string) *ristypes.RISMessage {
+		return &ristypes.RISMessage{
+			Peer:            testPeerIP,
+			Path:            path,
+			ParsedTimestamp: ts,
+			Announcements:   []ristypes.RISAnnouncement{{NextHop: testPeerIP, Prefixes: []string{announced}}},
+		}
+	}
+	// One message that announces otherPrefix with origin AS testASNB and
+	// withdraws prefix.
+	announceOtherAndWithdraw := announce(ristypes.ASPath{testASNA, testASNB}, otherPrefix)
+	announceOtherAndWithdraw.Withdrawals = []string{prefix}
+
+	tests := []struct {
+		name     string
+		messages []*ristypes.RISMessage
+		// The origin AS of the peer's live route for each prefix. A prefix
+		// left out must have no live route.
+		wantOrigins map[string]int
+	}{
+		{
+			name: "the peer announces the prefix with a new origin AS",
+			messages: []*ristypes.RISMessage{
+				announce(ristypes.ASPath{testASNA}, prefix),
+				announce(ristypes.ASPath{testASNA, testASNB}, prefix),
+			},
+			wantOrigins: map[string]int{prefix: testASNB},
+		},
+		{
+			name: "one message announces a prefix with another origin AS and withdraws the prefix",
+			messages: []*ristypes.RISMessage{
+				announce(ristypes.ASPath{testASNA}, prefix),
+				announceOtherAndWithdraw,
+			},
+			wantOrigins: map[string]int{otherPrefix: testASNB},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
+			if err != nil {
+				t.Fatalf("failed to create database: %v", err)
+			}
+			defer func() { _ = db.Close() }()
+
+			// Built without NewPrefixHandler's flush timer, so the test flushes itself.
+			h := &PrefixHandler{db: db, logger: logger.New()}
+			for _, msg := range tt.messages {
+				h.HandleMessage(msg)
+				h.mu.Lock()
+				h.flushBatchLocked()
+				h.mu.Unlock()
+			}
+
+			for _, p := range []string{prefix, otherPrefix} {
+				routes, err := db.GetPrefixDetailsContext(context.Background(), p)
+				if err != nil && !errors.Is(err, database.ErrNoRoute) {
+					t.Fatalf("GetPrefixDetailsContext(%s): %v", p, err)
+				}
+				var gotOrigins []int
+				for _, route := range routes {
+					gotOrigins = append(gotOrigins, route.OriginASN)
+				}
+				var wantOrigins []int
+				if origin, ok := tt.wantOrigins[p]; ok {
+					wantOrigins = []int{origin}
+				}
+				if !slices.Equal(gotOrigins, wantOrigins) {
+					t.Errorf("origin ASes of the live routes for %s = %v, want %v", p, gotOrigins, wantOrigins)
+				}
+			}
+
+			wantDistribution := []database.PrefixDistribution{
+				{MaskLength: prefixMaskLength, Count: len(tt.wantOrigins)},
+			}
+			assertIPv4Counts(t, db, len(tt.wantOrigins), wantDistribution)
+		})
+	}
+}
+
+// assertIPv4Counts checks that the IPv4 live route count and prefix
+// distribution are wantRoutes and wantDistribution, both as held in memory and
+// as read from the route table.
+func assertIPv4Counts(t *testing.T, db *database.Database, wantRoutes int,
+	wantDistribution []database.PrefixDistribution) {
+	t.Helper()
+
+	ctx := context.Background()
+	stats, err := db.GetStatsContext(ctx)
+	if err != nil {
+		t.Fatalf("GetStatsContext: %v", err)
+	}
+	tableRoutesV4, _, err := db.GetLiveRouteCountsContext(ctx)
+	if err != nil {
+		t.Fatalf("GetLiveRouteCountsContext: %v", err)
+	}
+	if stats.IPv4Routes != wantRoutes || tableRoutesV4 != wantRoutes {
+		t.Errorf("IPv4 live routes: %d in memory, %d in the table, want %d",
+			stats.IPv4Routes, tableRoutesV4, wantRoutes)
+	}
+
+	queryDistributionV4, _, err := db.GetPrefixDistributionContext(ctx)
+	if err != nil {
+		t.Fatalf("GetPrefixDistributionContext: %v", err)
+	}
+	if !slices.Equal(stats.IPv4PrefixDistribution, wantDistribution) ||
+		!slices.Equal(queryDistributionV4, wantDistribution) {
+		t.Errorf("IPv4 prefix distribution: %v in memory, %v from the table, want %v",
+			stats.IPv4PrefixDistribution, queryDistributionV4, wantDistribution)
+	}
+}

internal/server/handlers.go

@@ -201,16 +201,12 @@ func (s *Server) handleStatusJSON() http.HandlerFunc {
 		select {
 		case <-ctx.Done():
 			s.logger.Error("Database stats timeout in status.json")
-			writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
+			writeJSONError(w, http.StatusRequestTimeout, "Database timeout")
 
 			return
 		case err := <-errChan:
 			s.logger.Error("Failed to get database stats", "error", err)
-			if deadlinePassed(ctx) {
-				writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
-			} else {
-				writeJSONError(w, http.StatusInternalServerError, err.Error())
-			}
+			writeJSONError(w, http.StatusInternalServerError, err.Error())
 
 			return
 		case dbStats = <-statsChan:
@@ -390,7 +386,7 @@ func (s *Server) handleStats() http.HandlerFunc {
 		// Check if context is already cancelled
 		select {
 		case <-ctx.Done():
-			writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
+			http.Error(w, "Request timeout", http.StatusRequestTimeout)
 
 			return
 		default:
@@ -419,16 +415,11 @@ func (s *Server) handleStats() http.HandlerFunc {
 		select {
 		case <-ctx.Done():
 			s.logger.Error("Database stats timeout")
-			writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
-
+			// Don't write response here - timeout middleware already handles it
 			return
 		case err := <-errChan:
 			s.logger.Error("Failed to get database stats", "error", err)
-			if deadlinePassed(ctx) {
-				writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
-			} else {
-				writeJSONError(w, http.StatusInternalServerError, err.Error())
-			}
+			writeJSONError(w, http.StatusInternalServerError, err.Error())
 
 			return
 		case dbStats = <-statsChan:
@@ -631,13 +622,6 @@ func (s *Server) handleIPInfo() http.HandlerFunc {
 			}
 		}
 
-		// The time limit passed while resolving host names.
-		if deadlinePassed(ctx) {
-			writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
-
-			return
-		}
-
 		if len(ipsToLookup) == 0 {
 			writeJSONError(w, http.StatusBadRequest, "No valid IPs or hostnames provided")
 
@@ -680,13 +664,6 @@ func (s *Server) handleIPInfo() http.HandlerFunc {
 			}
 		}
 
-		// The time limit passed while looking up the IPs.
-		if deadlinePassed(ctx) {
-			writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
-
-			return
-		}
-
 		// Return response (even if no results, include errors)
 		if len(response.Results) == 0 && len(response.Errors) > 0 {
 			writeJSONError(w, http.StatusNotFound, "No routes found: "+response.Errors[0])
@@ -726,12 +703,9 @@ func (s *Server) handleASDetailJSON() http.HandlerFunc {
 
 		asInfo, prefixes, err := s.db.GetASDetailsContext(r.Context(), asn)
 		if err != nil {
-			switch {
-			case errors.Is(err, database.ErrNoRoute):
+			if errors.Is(err, database.ErrNoRoute) {
 				writeJSONError(w, http.StatusNotFound, err.Error())
-			case deadlinePassed(r.Context()):
-				writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
-			default:
+			} else {
 				writeJSONError(w, http.StatusInternalServerError, err.Error())
 			}
 
@@ -787,12 +761,9 @@ func (s *Server) handlePrefixDetailJSON() http.HandlerFunc {
 
 		routes, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
 		if err != nil {
-			switch {
-			case errors.Is(err, database.ErrNoRoute):
+			if errors.Is(err, database.ErrNoRoute) {
 				writeJSONError(w, http.StatusNotFound, err.Error())
-			case deadlinePassed(r.Context()):
-				writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
-			default:
+			} else {
 				writeJSONError(w, http.StatusInternalServerError, err.Error())
 			}
 
@@ -832,12 +803,9 @@ func (s *Server) handleASDetail() http.HandlerFunc {
 
 		asInfo, prefixes, err := s.db.GetASDetailsContext(r.Context(), asn)
 		if err != nil {
-			switch {
-			case errors.Is(err, database.ErrNoRoute):
+			if errors.Is(err, database.ErrNoRoute) {
 				http.Error(w, "AS not found", http.StatusNotFound)
-			case deadlinePassed(r.Context()):
-				http.Error(w, requestTimeoutMessage, http.StatusGatewayTimeout)
-			default:
+			} else {
 				s.logger.Error("Failed to get AS details", "error", err)
 				http.Error(w, "Internal server error", http.StatusInternalServerError)
 			}
@@ -964,12 +932,9 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
 
 		routes, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
 		if err != nil {
-			switch {
-			case errors.Is(err, database.ErrNoRoute):
+			if errors.Is(err, database.ErrNoRoute) {
 				http.Error(w, "Prefix not found", http.StatusNotFound)
-			case deadlinePassed(r.Context()):
-				http.Error(w, requestTimeoutMessage, http.StatusGatewayTimeout)
-			default:
+			} else {
 				s.logger.Error("Failed to get prefix details", "error", err)
 				http.Error(w, "Internal server error", http.StatusInternalServerError)
 			}
@@ -1123,12 +1088,8 @@ func (s *Server) handlePrefixLength() http.HandlerFunc {
 		const maxPrefixes = 500
 		prefixes, err := s.db.GetRandomPrefixesByLengthContext(r.Context(), maskLength, ipVersion, maxPrefixes)
 		if err != nil {
-			if deadlinePassed(r.Context()) {
-				http.Error(w, requestTimeoutMessage, http.StatusGatewayTimeout)
-			} else {
-				s.logger.Error("Failed to get prefixes by length", "error", err)
-				http.Error(w, "Internal server error", http.StatusInternalServerError)
-			}
+			s.logger.Error("Failed to get prefixes by length", "error", err)
+			http.Error(w, "Internal server error", http.StatusInternalServerError)
 
 			return
 		}
@@ -1218,12 +1179,8 @@ func (s *Server) handlePrefixLength6() http.HandlerFunc {
 		const maxPrefixes = 500
 		prefixes, err := s.db.GetRandomPrefixesByLengthContext(r.Context(), maskLength, ipVersion, maxPrefixes)
 		if err != nil {
-			if deadlinePassed(r.Context()) {
-				http.Error(w, requestTimeoutMessage, http.StatusGatewayTimeout)
-			} else {
-				s.logger.Error("Failed to get prefixes by length", "error", err)
-				http.Error(w, "Internal server error", http.StatusInternalServerError)
-			}
+			s.logger.Error("Failed to get prefixes by length", "error", err)
+			http.Error(w, "Internal server error", http.StatusInternalServerError)
 
 			return
 		}

internal/server/handlers_test.go

@@ -200,179 +200,6 @@ func TestPrefixDetailHandlersParsePrefix(t *testing.T) {
 	}
 }
 
-// shortRequestDeadline is the deadline the tests below give a request. The
-// server's 30-second limit and the stats handlers' 4-second limit inherit it,
-// so the tests do not wait for either.
-const shortRequestDeadline = 50 * time.Millisecond
-
-// TestRequestPastTimeLimitAnswers504 sends a request through the server's
-// router to a page and to a JSON route whose handlers wait for the time limit
-// to pass and return without answering. The page answers 504; the JSON route
-// answers 504 with the JSON error body. Before the fix both answered 408,
-// written by the time limit middleware from its own goroutine while the
-// handler could still be writing.
-func TestRequestPastTimeLimitAnswers504(t *testing.T) {
-	s := New(nil, streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{})
-
-	waitForTimeLimit := func(_ http.ResponseWriter, r *http.Request) {
-		<-r.Context().Done()
-	}
-	// Registered the way setupRoutes registers pages and JSON routes.
-	s.router.Get("/slow-page", waitForTimeLimit)
-	s.router.Get("/slow.json", JSONValidationMiddleware(http.HandlerFunc(waitForTimeLimit)).ServeHTTP)
-
-	tests := []struct {
-		path         string
-		wantJSONBody bool
-	}{
-		{path: "/slow-page"},
-		{path: "/slow.json", wantJSONBody: true},
-	}
-
-	for _, tt := range tests {
-		ctx, cancel := context.WithTimeout(context.Background(), shortRequestDeadline)
-		rec := httptest.NewRecorder()
-		s.router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, tt.path, nil).WithContext(ctx))
-		cancel()
-
-		if rec.Code != http.StatusGatewayTimeout {
-			t.Errorf("%s: status %d, want %d; body %s", tt.path, rec.Code, http.StatusGatewayTimeout, rec.Body)
-		}
-		if tt.wantJSONBody {
-			checkJSONErrorCode(t, tt.path, rec, http.StatusGatewayTimeout)
-		}
-	}
-}
-
-// TestStatsHandlersAnswer504WhenDatabaseReadRunsOut calls each stats handler
-// with a store that does not answer before the handler's 4-second limit runs
-// out, and checks that the handler itself answers 504 with the JSON error
-// body. Before the fix status.json answered 408, and /api/v1/stats wrote
-// nothing, which the JSON check on its route turned into 500.
-func TestStatsHandlersAnswer504WhenDatabaseReadRunsOut(t *testing.T) {
-	release := make(chan struct{})
-	defer close(release)
-	db := blockingStatsDB{release: release}
-	s := New(db, streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{})
-
-	handlers := map[string]http.HandlerFunc{
-		"status.json": s.handleStatusJSON(),
-		"stats":       s.handleStats(),
-	}
-
-	for name, handler := range handlers {
-		ctx, cancel := context.WithTimeout(context.Background(), shortRequestDeadline)
-		rec := httptest.NewRecorder()
-		handler(rec, httptest.NewRequest(http.MethodGet, "/", nil).WithContext(ctx))
-		cancel()
-
-		if rec.Code != http.StatusGatewayTimeout {
-			t.Errorf("%s: status %d, want %d; body %s", name, rec.Code, http.StatusGatewayTimeout, rec.Body)
-		}
-		checkJSONErrorCode(t, name, rec, http.StatusGatewayTimeout)
-	}
-}
-
-// slowDB embeds database.Store (left nil) and overrides the reads the routes
-// in TestRoutesAnswer504WhenDatabaseReadIsCutOff make. Each read waits until
-// its context ends and returns the context's error, as a slow database read
-// cut off by a time limit does.
-type slowDB struct {
-	database.Store
-}
-
-func (slowDB) GetStatsContext(ctx context.Context) (database.Stats, error) {
-	<-ctx.Done()
-
-	return database.Stats{}, ctx.Err()
-}
-
-func (slowDB) GetIPInfoContext(ctx context.Context, _ string) (*database.IPInfo, error) {
-	<-ctx.Done()
-
-	return nil, ctx.Err()
-}
-
-func (slowDB) GetASDetailsContext(ctx context.Context, _ int) (*database.ASN, []database.LiveRoute, error) {
-	<-ctx.Done()
-
-	return nil, nil, ctx.Err()
-}
-
-func (slowDB) GetPrefixDetailsContext(ctx context.Context, _ string) ([]database.LiveRoute, error) {
-	<-ctx.Done()
-
-	return nil, ctx.Err()
-}
-
-func (slowDB) GetRandomPrefixesByLengthContext(ctx context.Context, _, _, _ int) ([]database.LiveRoute, error) {
-	<-ctx.Done()
-
-	return nil, ctx.Err()
-}
-
-// TestRoutesAnswer504WhenDatabaseReadIsCutOff sends a request through the
-// server's router for each route that reads the database, with a store whose
-// reads fail when the request's time limit passes. Each route answers 504, the
-// JSON routes with the JSON error body. Before the fix they answered 408 or,
-// when the handler wrote first, its own error: 500, or 404 on the IP lookups.
-// The health check is left out: it answers 503 when its own 2-second ping runs
-// out.
-func TestRoutesAnswer504WhenDatabaseReadIsCutOff(t *testing.T) {
-	s := New(slowDB{}, streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{})
-
-	tests := []struct {
-		path         string
-		wantJSONBody bool
-	}{
-		{path: "/as/64500"},
-		{path: "/prefix/192.0.2.0/24"},
-		{path: "/prefixlength/24"},
-		{path: "/prefixlength6/48"},
-		{path: "/status.json", wantJSONBody: true},
-		{path: "/ip/192.0.2.1", wantJSONBody: true},
-		{path: "/api/v1/stats", wantJSONBody: true},
-		{path: "/api/v1/ip/192.0.2.1", wantJSONBody: true},
-		{path: "/api/v1/as/64500", wantJSONBody: true},
-		{path: "/api/v1/prefix/192.0.2.0/24", wantJSONBody: true},
-	}
-
-	for _, tt := range tests {
-		ctx, cancel := context.WithTimeout(context.Background(), shortRequestDeadline)
-		rec := httptest.NewRecorder()
-		s.router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, tt.path, nil).WithContext(ctx))
-		cancel()
-
-		if rec.Code != http.StatusGatewayTimeout {
-			t.Errorf("%s: status %d, want %d; body %s", tt.path, rec.Code, http.StatusGatewayTimeout, rec.Body)
-		}
-		if tt.wantJSONBody {
-			checkJSONErrorCode(t, tt.path, rec, http.StatusGatewayTimeout)
-		}
-	}
-}
-
-// checkJSONErrorCode checks that rec holds the JSON error body with the given
-// code.
-func checkJSONErrorCode(t *testing.T, name string, rec *httptest.ResponseRecorder, want int) {
-	t.Helper()
-
-	var body struct {
-		Status string `json:"status"`
-		Error  struct {
-			Code int `json:"code"`
-		} `json:"error"`
-	}
-	if err := json.Unmarshal(rec.Body.Bytes(), &body); err != nil {
-		t.Errorf("%s: decoding the answer %q: %v", name, rec.Body, err)
-
-		return
-	}
-	if body.Status != "error" || body.Error.Code != want {
-		t.Errorf("%s: answer %s, want status \"error\" and error code %d", name, rec.Body, want)
-	}
-}
-
 // settledGoroutineCount lets transient goroutines finish, then reports the
 // current count.
 func settledGoroutineCount() int {

internal/server/middleware.go

@@ -4,7 +4,6 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"log/slog"
 	"net/http"
@@ -110,24 +109,142 @@ func JSONResponseMiddleware(next http.Handler) http.Handler {
 	})
 }
 
-// requestTimeoutMessage is the error message sent with 504 Gateway Timeout.
-const requestTimeoutMessage = "Request timeout"
+// timeoutWriter wraps ResponseWriter to prevent concurrent writes after timeout
+type timeoutWriter struct {
+	http.ResponseWriter
+	mu      sync.Mutex
+	written bool
+	header  http.Header // cached header to prevent concurrent access
+}
 
-// deadlinePassed reports whether ctx ended because its deadline passed. When
-// a handler's database read or other lookup fails, it uses this to tell a
-// request cut off by its time limit, which answers 504, from other failures.
-func deadlinePassed(ctx context.Context) bool {
-	return errors.Is(ctx.Err(), context.DeadlineExceeded)
+func (tw *timeoutWriter) Write(b []byte) (int, error) {
+	tw.mu.Lock()
+	defer tw.mu.Unlock()
+
+	if tw.written {
+		return 0, nil // Discard writes after timeout
+	}
+
+	return tw.ResponseWriter.Write(b)
+}
+
+func (tw *timeoutWriter) WriteHeader(statusCode int) {
+	tw.mu.Lock()
+	defer tw.mu.Unlock()
+
+	if tw.written {
+		return // Discard writes after timeout
+	}
+
+	tw.ResponseWriter.WriteHeader(statusCode)
+}
+
+func (tw *timeoutWriter) Header() http.Header {
+	tw.mu.Lock()
+	defer tw.mu.Unlock()
+
+	if tw.written {
+		// Return a copy to prevent modifications after timeout
+		if tw.header == nil {
+			tw.header = make(http.Header)
+		}
+
+		return tw.header
+	}
+
+	return tw.ResponseWriter.Header()
+}
+
+func (tw *timeoutWriter) markWritten() {
+	tw.mu.Lock()
+	defer tw.mu.Unlock()
+	tw.written = true
+}
+
+// TimeoutMiddleware creates an HTTP middleware that enforces a request timeout.
+// If the handler does not complete within the specified duration, the middleware
+// returns a JSON error response with HTTP status 408 (Request Timeout).
+//
+// The timeout parameter specifies the maximum duration allowed for request processing.
+// The returned middleware handles panics from the wrapped handler by re-panicking
+// after cleanup, and prevents concurrent writes to the response after timeout occurs.
+func TimeoutMiddleware(timeout time.Duration, logger *slog.Logger) func(http.Handler) http.Handler {
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			startTime := time.Now()
+
+			ctx, cancel := context.WithTimeout(r.Context(), timeout)
+			defer cancel()
+
+			tw := &timeoutWriter{
+				ResponseWriter: w,
+				header:         make(http.Header),
+			}
+
+			done := make(chan struct{})
+			panicChan := make(chan interface{}, 1)
+
+			go func() {
+				defer func() {
+					if p := recover(); p != nil {
+						panicChan <- p
+					}
+				}()
+
+				next.ServeHTTP(tw, r.WithContext(ctx))
+				close(done)
+			}()
+
+			select {
+			case p := <-panicChan:
+				panic(p)
+			case <-done:
+				return
+			case <-ctx.Done():
+				tw.markWritten() // Prevent the handler from writing after timeout
+				execTime := time.Since(startTime)
+
+				// Log the timeout as a warning
+				logger.Warn("Request timeout",
+					"method", r.Method,
+					"path", r.URL.Path,
+					"duration_ms", execTime.Milliseconds(),
+					"remote_addr", r.RemoteAddr,
+				)
+
+				// Write directly to the underlying writer since we've marked tw as written
+				// This is safe because markWritten() prevents the handler from writing
+				tw.mu.Lock()
+				w.Header().Set("Content-Type", "application/json")
+				w.WriteHeader(http.StatusRequestTimeout)
+				tw.mu.Unlock()
+
+				response := map[string]interface{}{
+					"status": "error",
+					"error": map[string]interface{}{
+						"msg":  "Request timeout",
+						"code": http.StatusRequestTimeout,
+					},
+					"@meta": map[string]interface{}{
+						"time_zone":      "UTC",
+						"api_version":    1,
+						"execution_time": fmt.Sprintf("%d ms", execTime.Milliseconds()),
+					},
+				}
+
+				_ = json.NewEncoder(w).Encode(response)
+			}
+		})
+	}
 }
 
 // JSONValidationMiddleware is an HTTP middleware that validates JSON API responses.
 // It ensures that responses with Content-Type "application/json" contain valid JSON.
 //
 // If a response is not valid JSON or is empty when JSON is expected, the middleware
-// returns a properly formatted JSON error response. If the request's time limit
-// has passed, it answers 504 with the error message "Request timeout". For other
-// errors, it returns "Internal server error" with status 500 if the original
-// status was 200.
+// returns a properly formatted JSON error response. For timeout errors (status 408),
+// the error message will be "Request timeout". For other errors, it returns
+// "Internal server error" with status 500 if the original status was 200.
 //
 // Non-JSON responses are passed through unchanged.
 func JSONValidationMiddleware(next http.Handler) http.Handler {
@@ -176,14 +293,11 @@ func JSONValidationMiddleware(next http.Handler) http.Handler {
 		if statusCode == http.StatusOK {
 			statusCode = http.StatusInternalServerError
 		}
-		if deadlinePassed(r.Context()) {
-			statusCode = http.StatusGatewayTimeout
-		}
 		w.WriteHeader(statusCode)
 
 		errorMsg := "Internal server error"
-		if statusCode == http.StatusGatewayTimeout {
-			errorMsg = requestTimeoutMessage
+		if statusCode == http.StatusRequestTimeout {
+			errorMsg = "Request timeout"
 		}
 
 		response := map[string]interface{}{

internal/server/routes.go

@@ -16,12 +16,9 @@ func (s *Server) setupRoutes() {
 	r.Use(middleware.RealIP)
 	r.Use(RequestLoggerMiddleware(s.logger.Logger)) // Structured request logging
 	r.Use(middleware.Recoverer)
-	r.Use(JSONResponseMiddleware)
-	// Timeout must come after JSONResponseMiddleware, which sends its answer
-	// when the handler returns and would send a 200 ahead of Timeout's 504 for
-	// a handler that wrote nothing.
 	const requestTimeout = 30 * time.Second // Increased from 8s for slow queries
-	r.Use(middleware.Timeout(requestTimeout))
+	r.Use(TimeoutMiddleware(requestTimeout, s.logger.Logger))
+	r.Use(JSONResponseMiddleware)
 
 	// Routes
 	r.Get("/", s.handleIndex())