Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
bc7c5e18a5 |
@@ -141,10 +141,7 @@ routewatch/
|
||||
- **Batched Writes**: All database operations are batched for performance
|
||||
- **Backpressure**: Probabilistic message dropping when queues exceed 50% capacity
|
||||
- **Graceful Shutdown**: 60-second timeout, flushes all pending batches
|
||||
- **Reconnection**: Any end of the stream other than shutdown, including RIS Live
|
||||
ending the response normally, marks the feed disconnected and reconnects with
|
||||
exponential backoff (5s-320s). The backoff resets after a stream that lasted
|
||||
more than 30s. A stream with no line for 60s is dropped and reconnected.
|
||||
- **Reconnection**: Exponential backoff (5s-320s) with reset after 30s of stable connection
|
||||
- **IP Lookup**: the most specific live route is found by looking up the
|
||||
address's prefix at each mask length, longest first, in the prefix index
|
||||
(at most 25 lookups for IPv4, 113 for IPv6); prefixes are stored in the
|
||||
|
||||
@@ -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", PeerIP: "192.0.2.1", IPVersion: ipVersionV4},
|
||||
{Prefix: "203.0.113.0/24", OriginASN: 64501, PeerIP: "192.0.2.1", IPVersion: ipVersionV4},
|
||||
}); err != nil {
|
||||
t.Fatalf("DeleteLiveRouteBatch: %v", err)
|
||||
}
|
||||
@@ -327,11 +327,13 @@ func announceRoutes(db *Database, batch bool, routes []*LiveRoute) error {
|
||||
}
|
||||
|
||||
// withdrawRoutes removes routes in one DeleteLiveRouteBatch, or with one
|
||||
// DeleteLiveRoute each.
|
||||
// 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.
|
||||
func withdrawRoutes(db *Database, batch bool, routes []*LiveRoute) error {
|
||||
if !batch {
|
||||
for _, route := range routes {
|
||||
if err := db.DeleteLiveRoute(route.Prefix, route.PeerIP); err != nil {
|
||||
if err := db.DeleteLiveRoute(route.Prefix, 0, route.PeerIP); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
@@ -228,22 +228,21 @@ func (d *Database) beginTx() (*loggingTx, error) {
|
||||
}
|
||||
|
||||
// A live-route upsert is an UPDATE followed, only when no row matched, by an
|
||||
// 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.
|
||||
// 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.
|
||||
const (
|
||||
updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, origin_asn = ?, as_path = ?,
|
||||
next_hop = ?, last_updated = ? WHERE prefix = ? AND peer_ip = ?`
|
||||
updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, as_path = ?, next_hop = ?,
|
||||
last_updated = ? WHERE prefix = ? AND origin_asn = ? 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 = ?, origin_asn = ?, as_path = ?,
|
||||
next_hop = ?, last_updated = ? WHERE prefix = ? AND peer_ip = ?`
|
||||
updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, as_path = ?, next_hop = ?,
|
||||
last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
|
||||
|
||||
insertLiveRouteV6SQL = `INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip,
|
||||
as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
|
||||
@@ -262,8 +261,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, route.OriginASN, pathJSON, route.NextHop, route.LastUpdated,
|
||||
route.Prefix, route.PeerIP)
|
||||
res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
|
||||
route.Prefix, route.OriginASN, route.PeerIP)
|
||||
if err != nil {
|
||||
return false, false, err
|
||||
}
|
||||
@@ -294,8 +293,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, route.OriginASN, pathJSON, route.NextHop, route.LastUpdated,
|
||||
route.Prefix, route.PeerIP)
|
||||
res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
|
||||
route.Prefix, route.OriginASN, route.PeerIP)
|
||||
if err != nil {
|
||||
return false, false, err
|
||||
}
|
||||
@@ -442,18 +441,32 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
|
||||
}
|
||||
}()
|
||||
|
||||
// Prepare statements for both IPv4 and IPv6 tables
|
||||
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 statement: %w", err)
|
||||
}
|
||||
defer func() { _ = deleteV4.Close() }()
|
||||
// No longer need to separate deletions since we handle them in the loop below
|
||||
|
||||
deleteV6, err := tx.Prepare(`DELETE FROM live_routes_v6 WHERE prefix = ? AND peer_ip = ?`)
|
||||
// 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 = ?`)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to prepare IPv6 delete statement: %w", err)
|
||||
return fmt.Errorf("failed to prepare IPv4 delete with origin statement: %w", err)
|
||||
}
|
||||
defer func() { _ = deleteV6.Close() }()
|
||||
defer func() { _ = stmtV4WithOrigin.Close() }()
|
||||
|
||||
stmtV4WithoutOrigin, err := tx.Prepare(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to prepare IPv4 delete without origin 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() }()
|
||||
|
||||
hasV4, err := tx.Prepare(prefixHasLiveRouteV4SQL)
|
||||
if err != nil {
|
||||
@@ -472,18 +485,38 @@ 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 {
|
||||
stmt := deleteV4
|
||||
if del.IPVersion != ipVersionV4 {
|
||||
stmt = deleteV6
|
||||
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)
|
||||
}
|
||||
|
||||
res, err := stmt.Exec(del.Prefix, del.PeerIP)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to delete route %s: %w", del.Prefix, err)
|
||||
}
|
||||
|
||||
// The feed often withdraws a route that is not live, which deletes
|
||||
// nothing, so the in-memory route counters use the rows-affected count.
|
||||
// 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.
|
||||
affected, err := res.RowsAffected()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to count deleted route %s: %w", del.Prefix, err)
|
||||
@@ -1249,8 +1282,9 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// DeleteLiveRoute deletes the live route the peer has for the prefix
|
||||
func (d *Database) DeleteLiveRoute(prefix, peerIP string) error {
|
||||
// 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 {
|
||||
d.lock("DeleteLiveRoute")
|
||||
defer d.unlock()
|
||||
|
||||
@@ -1275,12 +1309,22 @@ func (d *Database) DeleteLiveRoute(prefix, peerIP string) error {
|
||||
}()
|
||||
|
||||
// Literal per-table queries (rather than one formatted with the table name)
|
||||
// so the delete carries no dynamically built SQL.
|
||||
// so the delete carries no dynamically built SQL. A delete with no origin
|
||||
// ASN can remove several rows.
|
||||
var res sql.Result
|
||||
if isV4 {
|
||||
switch {
|
||||
case isV4 && originASN == 0:
|
||||
res, err = tx.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
|
||||
} else {
|
||||
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:
|
||||
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
|
||||
@@ -1588,17 +1632,16 @@ func (d *Database) GetASPeersContext(ctx context.Context, asn int) ([]ASPeer, er
|
||||
}
|
||||
|
||||
// GetPrefixDetails returns detailed information about a prefix
|
||||
func (d *Database) GetPrefixDetails(prefix string) ([]LiveRoute, map[int]ASN, error) {
|
||||
func (d *Database) GetPrefixDetails(prefix string) ([]LiveRoute, error) {
|
||||
return d.GetPrefixDetailsContext(context.Background(), prefix)
|
||||
}
|
||||
|
||||
// GetPrefixDetailsContext returns the live routes for a prefix and, keyed by
|
||||
// AS number, the handle and description of each AS that originates it.
|
||||
func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, map[int]ASN, error) {
|
||||
// GetPrefixDetailsContext returns detailed information about a prefix with context support
|
||||
func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, error) {
|
||||
// Determine if it's IPv4 or IPv6 by parsing the prefix
|
||||
_, ipnet, err := net.ParseCIDR(prefix)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("invalid prefix format: %w", err)
|
||||
return nil, fmt.Errorf("invalid prefix format: %w", err)
|
||||
}
|
||||
|
||||
tableName := "live_routes_v4"
|
||||
@@ -1620,12 +1663,11 @@ func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) (
|
||||
|
||||
rows, err := d.db.QueryContext(ctx, query, prefix)
|
||||
if err != nil {
|
||||
return nil, nil, fmt.Errorf("failed to query prefix details: %w", err)
|
||||
return nil, fmt.Errorf("failed to query prefix details: %w", err)
|
||||
}
|
||||
defer func() { _ = rows.Close() }()
|
||||
|
||||
var routes []LiveRoute
|
||||
origins := make(map[int]ASN)
|
||||
for rows.Next() {
|
||||
var route LiveRoute
|
||||
var pathJSON string
|
||||
@@ -1648,18 +1690,13 @@ func (d *Database) GetPrefixDetailsContext(ctx context.Context, prefix string) (
|
||||
route.IPVersion = ipVersion
|
||||
route.MaskLength, _ = ipnet.Mask.Size()
|
||||
routes = append(routes, route)
|
||||
origins[route.OriginASN] = ASN{
|
||||
ASN: route.OriginASN,
|
||||
Handle: handle.String,
|
||||
Description: description.String,
|
||||
}
|
||||
}
|
||||
|
||||
if len(routes) == 0 {
|
||||
return nil, nil, fmt.Errorf("%w: %s", ErrNoRoute, prefix)
|
||||
return nil, fmt.Errorf("%w: %s", ErrNoRoute, prefix)
|
||||
}
|
||||
|
||||
return routes, origins, nil
|
||||
return routes, nil
|
||||
}
|
||||
|
||||
// GetRandomPrefixesByLength returns a random sample of prefixes with the specified mask length
|
||||
|
||||
@@ -13,7 +13,6 @@ import (
|
||||
"git.eeqj.de/sneak/routewatch/internal/config"
|
||||
"git.eeqj.de/sneak/routewatch/internal/logger"
|
||||
"github.com/google/uuid"
|
||||
"github.com/mattn/go-sqlite3"
|
||||
)
|
||||
|
||||
// tempStoreMemory is the PRAGMA temp_store value meaning "hold temp B-trees in
|
||||
@@ -123,8 +122,8 @@ func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) {
|
||||
lookup("2001:db9::1", "")
|
||||
|
||||
err = db.DeleteLiveRouteBatch([]LiveRouteDeletion{
|
||||
{Prefix: "10.1.2.0/24", PeerIP: testPeerIP, IPVersion: ipVersionV4},
|
||||
{Prefix: "2001:db8:1:2::/64", PeerIP: testPeerIP, IPVersion: ipVersionV6},
|
||||
{Prefix: "10.1.2.0/24", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV4},
|
||||
{Prefix: "2001:db8:1:2::/64", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV6},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("DeleteLiveRouteBatch: %v", err)
|
||||
@@ -206,40 +205,6 @@ func TestGetIPInfoIgnoresShortRoutes(t *testing.T) {
|
||||
lookup("2001:db8::1", "2001:db8::/32", 64502)
|
||||
}
|
||||
|
||||
// TestLiveRouteTablesHoldOneRoutePerPrefixAndPeer checks that each live route
|
||||
// table refuses a second route from a peer for a prefix, even with another
|
||||
// origin AS. An upsert updates the peer's route before it would insert one, so
|
||||
// only an insert straight into the table shows what the table allows.
|
||||
func TestLiveRouteTablesHoldOneRoutePerPrefixAndPeer(t *testing.T) {
|
||||
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() }()
|
||||
|
||||
const otherOriginASN = 64501
|
||||
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||
for _, tt := range []struct {
|
||||
route *LiveRoute
|
||||
insertSQL string
|
||||
}{
|
||||
{route: mkV4Route(t, "198.51.100.0/24", 64500, ts), insertSQL: insertLiveRouteV4SQL},
|
||||
{route: mkV6Route("2001:db8::/32", 64500, ts), insertSQL: insertLiveRouteV6SQL},
|
||||
} {
|
||||
if err := db.UpsertLiveRoute(tt.route); err != nil {
|
||||
t.Fatalf("UpsertLiveRoute(%s): %v", tt.route.Prefix, err)
|
||||
}
|
||||
|
||||
_, err := db.db.Exec(tt.insertSQL, uuid.NewString(), tt.route.Prefix, tt.route.MaskLength,
|
||||
otherOriginASN, tt.route.PeerIP, "[]", tt.route.NextHop, ts)
|
||||
var sqliteErr sqlite3.Error
|
||||
if !errors.As(err, &sqliteErr) || sqliteErr.ExtendedCode != sqlite3.ErrConstraintUnique {
|
||||
t.Errorf("second route from peer %s for %s: error %v, want a UNIQUE constraint error",
|
||||
tt.route.PeerIP, tt.route.Prefix, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestConnectionPoolPragmas holds several pooled connections open at once and
|
||||
// checks each one carries the per-connection settings from the DSN, plus the
|
||||
// process-wide hard heap limit, and sees the new file with auto_vacuum
|
||||
@@ -341,6 +306,7 @@ func TestVacuumReturnsFreePages(t *testing.T) {
|
||||
routes = append(routes, route)
|
||||
deletions = append(deletions, LiveRouteDeletion{
|
||||
Prefix: route.Prefix,
|
||||
OriginASN: asn,
|
||||
PeerIP: route.PeerIP,
|
||||
IPVersion: ipVersionV4,
|
||||
})
|
||||
|
||||
@@ -55,7 +55,7 @@ type Store interface {
|
||||
// Live route operations
|
||||
UpsertLiveRoute(route *LiveRoute) error
|
||||
UpsertLiveRouteBatch(routes []*LiveRoute) error
|
||||
DeleteLiveRoute(prefix, peerIP string) error
|
||||
DeleteLiveRoute(prefix string, originASN int, peerIP string) error
|
||||
DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error
|
||||
GetPrefixDistribution() (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)
|
||||
GetPrefixDistributionContext(ctx context.Context) (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)
|
||||
@@ -76,8 +76,8 @@ type Store interface {
|
||||
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)
|
||||
GetPrefixDetails(prefix string) ([]LiveRoute, error)
|
||||
GetPrefixDetailsContext(ctx context.Context, prefix string) ([]LiveRoute, error)
|
||||
GetRandomPrefixesByLength(maskLength, ipVersion, limit int) ([]LiveRoute, error)
|
||||
GetRandomPrefixesByLengthContext(ctx context.Context, maskLength, ipVersion, limit int) ([]LiveRoute, error)
|
||||
|
||||
|
||||
@@ -114,6 +114,7 @@ type IPInfo struct {
|
||||
// LiveRouteDeletion represents parameters for deleting a live route
|
||||
type LiveRouteDeletion struct {
|
||||
Prefix string
|
||||
OriginASN int
|
||||
PeerIP string
|
||||
IPVersion int
|
||||
}
|
||||
|
||||
@@ -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, peer_ip) -- a peer has at most one route for a prefix
|
||||
UNIQUE(prefix, origin_asn, peer_ip)
|
||||
);
|
||||
|
||||
-- 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, peer_ip) -- a peer has at most one route for a prefix
|
||||
UNIQUE(prefix, origin_asn, peer_ip)
|
||||
);
|
||||
|
||||
-- Indexes for live_routes_v4 table
|
||||
|
||||
@@ -205,7 +205,7 @@ func (m *mockStore) UpsertLiveRoute(route *database.LiveRoute) error {
|
||||
}
|
||||
|
||||
// DeleteLiveRoute mock implementation
|
||||
func (m *mockStore) DeleteLiveRoute(prefix, peerIP string) error {
|
||||
func (m *mockStore) DeleteLiveRoute(prefix string, originASN int, peerIP string) error {
|
||||
// Simple mock - just return nil
|
||||
return nil
|
||||
}
|
||||
@@ -252,15 +252,13 @@ func (m *mockStore) GetASDetailsContext(ctx context.Context, asn int) (*database
|
||||
}
|
||||
|
||||
// GetPrefixDetails mock implementation
|
||||
func (m *mockStore) GetPrefixDetails(prefix string) ([]database.LiveRoute, map[int]database.ASN, error) {
|
||||
func (m *mockStore) GetPrefixDetails(prefix string) ([]database.LiveRoute, error) {
|
||||
// Return empty routes for now
|
||||
return []database.LiveRoute{}, map[int]database.ASN{}, nil
|
||||
return []database.LiveRoute{}, nil
|
||||
}
|
||||
|
||||
// GetPrefixDetailsContext mock implementation with context support
|
||||
func (m *mockStore) GetPrefixDetailsContext(
|
||||
ctx context.Context, prefix string,
|
||||
) ([]database.LiveRoute, map[int]database.ASN, error) {
|
||||
func (m *mockStore) GetPrefixDetailsContext(ctx context.Context, prefix string) ([]database.LiveRoute, error) {
|
||||
return m.GetPrefixDetails(prefix)
|
||||
}
|
||||
|
||||
|
||||
@@ -20,8 +20,6 @@ const (
|
||||
debugInterval = 60 * time.Second
|
||||
// bytesPerMB is bytes per megabyte
|
||||
bytesPerMB = 1024 * 1024
|
||||
// failureExitCode is the exit status when RouteWatch stops on an error
|
||||
failureExitCode = 1
|
||||
)
|
||||
|
||||
// logDebugStats logs goroutine count and memory usage
|
||||
@@ -85,9 +83,6 @@ func CLIEntry() {
|
||||
go func() {
|
||||
if err := rw.Run(ctx); err != nil {
|
||||
logger.Error("RouteWatch error", "error", err)
|
||||
if err := shutdowner.Shutdown(fx.ExitCode(failureExitCode)); err != nil {
|
||||
logger.Error("Failed to shutdown gracefully", "error", err)
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
|
||||
@@ -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,6 +221,7 @@ 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,
|
||||
})
|
||||
@@ -424,24 +425,51 @@ 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) {
|
||||
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,
|
||||
)
|
||||
// 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,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 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) {
|
||||
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,
|
||||
)
|
||||
// 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,
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -3,7 +3,6 @@ package routewatch
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -171,7 +170,7 @@ func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) {
|
||||
}
|
||||
|
||||
ctx := context.Background()
|
||||
routes, _, err := db.GetPrefixDetailsContext(ctx, prefix)
|
||||
routes, err := db.GetPrefixDetailsContext(ctx, prefix)
|
||||
if err != nil && !errors.Is(err, database.ErrNoRoute) {
|
||||
t.Fatalf("GetPrefixDetailsContext: %v", err)
|
||||
}
|
||||
@@ -183,155 +182,32 @@ 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}}
|
||||
}
|
||||
assertCounts(t, db, ipv4Version, len(tt.wantPeers), wantDistribution)
|
||||
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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer runs updates from one peer
|
||||
// through the prefix handler into a real database, flushing each message as its
|
||||
// own batch, once with IPv4 prefixes and once with IPv6 prefixes. 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) {
|
||||
families := []struct {
|
||||
ipVersion int
|
||||
prefix, otherPrefix string
|
||||
// The mask length of both prefixes.
|
||||
maskLength int
|
||||
}{
|
||||
{ipVersion: ipv4Version, prefix: "198.51.100.0/24", otherPrefix: "203.0.113.0/24", maskLength: 24},
|
||||
{ipVersion: ipv6Version, prefix: "2001:db8:1::/48", otherPrefix: "2001:db8:2::/48", maskLength: 48},
|
||||
}
|
||||
|
||||
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}}},
|
||||
}
|
||||
}
|
||||
|
||||
for _, family := range families {
|
||||
prefix, otherPrefix := family.prefix, family.otherPrefix
|
||||
|
||||
// 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(fmt.Sprintf("IPv%d %s", family.ipVersion, 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: family.maskLength, Count: len(tt.wantOrigins)},
|
||||
}
|
||||
assertCounts(t, db, family.ipVersion, len(tt.wantOrigins), wantDistribution)
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// assertCounts checks that the live route count and prefix distribution of one
|
||||
// IP version are wantRoutes and wantDistribution, both as held in memory and as
|
||||
// read from the route table.
|
||||
func assertCounts(t *testing.T, db *database.Database, ipVersion, wantRoutes int,
|
||||
wantDistribution []database.PrefixDistribution) {
|
||||
t.Helper()
|
||||
|
||||
ctx := context.Background()
|
||||
stats, err := db.GetStatsContext(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("GetStatsContext: %v", err)
|
||||
}
|
||||
tableRoutesV4, tableRoutesV6, err := db.GetLiveRouteCountsContext(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("GetLiveRouteCountsContext: %v", err)
|
||||
}
|
||||
queryDistributionV4, queryDistributionV6, err := db.GetPrefixDistributionContext(ctx)
|
||||
if err != nil {
|
||||
t.Fatalf("GetPrefixDistributionContext: %v", err)
|
||||
}
|
||||
|
||||
memoryRoutes, tableRoutes := stats.IPv4Routes, tableRoutesV4
|
||||
memoryDistribution, queryDistribution := stats.IPv4PrefixDistribution, queryDistributionV4
|
||||
if ipVersion == ipv6Version {
|
||||
memoryRoutes, tableRoutes = stats.IPv6Routes, tableRoutesV6
|
||||
memoryDistribution, queryDistribution = stats.IPv6PrefixDistribution, queryDistributionV6
|
||||
}
|
||||
|
||||
if memoryRoutes != wantRoutes || tableRoutes != wantRoutes {
|
||||
t.Errorf("IPv%d live routes: %d in memory, %d in the table, want %d",
|
||||
ipVersion, memoryRoutes, tableRoutes, wantRoutes)
|
||||
}
|
||||
if !slices.Equal(memoryDistribution, wantDistribution) ||
|
||||
!slices.Equal(queryDistribution, wantDistribution) {
|
||||
t.Errorf("IPv%d prefix distribution: %v in memory, %v from the table, want %v",
|
||||
ipVersion, memoryDistribution, queryDistribution, wantDistribution)
|
||||
}
|
||||
}
|
||||
|
||||
+72
-21
@@ -201,12 +201,16 @@ func (s *Server) handleStatusJSON() http.HandlerFunc {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
s.logger.Error("Database stats timeout in status.json")
|
||||
writeJSONError(w, http.StatusRequestTimeout, "Database timeout")
|
||||
writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
|
||||
|
||||
return
|
||||
case err := <-errChan:
|
||||
s.logger.Error("Failed to get database stats", "error", err)
|
||||
writeJSONError(w, http.StatusInternalServerError, err.Error())
|
||||
if deadlinePassed(ctx) {
|
||||
writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
|
||||
} else {
|
||||
writeJSONError(w, http.StatusInternalServerError, err.Error())
|
||||
}
|
||||
|
||||
return
|
||||
case dbStats = <-statsChan:
|
||||
@@ -386,7 +390,7 @@ func (s *Server) handleStats() http.HandlerFunc {
|
||||
// Check if context is already cancelled
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
http.Error(w, "Request timeout", http.StatusRequestTimeout)
|
||||
writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
|
||||
|
||||
return
|
||||
default:
|
||||
@@ -415,11 +419,16 @@ func (s *Server) handleStats() http.HandlerFunc {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
s.logger.Error("Database stats timeout")
|
||||
// Don't write response here - timeout middleware already handles it
|
||||
writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
|
||||
|
||||
return
|
||||
case err := <-errChan:
|
||||
s.logger.Error("Failed to get database stats", "error", err)
|
||||
writeJSONError(w, http.StatusInternalServerError, err.Error())
|
||||
if deadlinePassed(ctx) {
|
||||
writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
|
||||
} else {
|
||||
writeJSONError(w, http.StatusInternalServerError, err.Error())
|
||||
}
|
||||
|
||||
return
|
||||
case dbStats = <-statsChan:
|
||||
@@ -622,6 +631,13 @@ 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")
|
||||
|
||||
@@ -664,6 +680,13 @@ 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])
|
||||
@@ -703,9 +726,12 @@ func (s *Server) handleASDetailJSON() http.HandlerFunc {
|
||||
|
||||
asInfo, prefixes, err := s.db.GetASDetailsContext(r.Context(), asn)
|
||||
if err != nil {
|
||||
if errors.Is(err, database.ErrNoRoute) {
|
||||
switch {
|
||||
case errors.Is(err, database.ErrNoRoute):
|
||||
writeJSONError(w, http.StatusNotFound, err.Error())
|
||||
} else {
|
||||
case deadlinePassed(r.Context()):
|
||||
writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
|
||||
default:
|
||||
writeJSONError(w, http.StatusInternalServerError, err.Error())
|
||||
}
|
||||
|
||||
@@ -759,11 +785,14 @@ func (s *Server) handlePrefixDetailJSON() http.HandlerFunc {
|
||||
}
|
||||
prefix := parsed.Masked().String()
|
||||
|
||||
routes, _, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
|
||||
routes, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
|
||||
if err != nil {
|
||||
if errors.Is(err, database.ErrNoRoute) {
|
||||
switch {
|
||||
case errors.Is(err, database.ErrNoRoute):
|
||||
writeJSONError(w, http.StatusNotFound, err.Error())
|
||||
} else {
|
||||
case deadlinePassed(r.Context()):
|
||||
writeJSONError(w, http.StatusGatewayTimeout, requestTimeoutMessage)
|
||||
default:
|
||||
writeJSONError(w, http.StatusInternalServerError, err.Error())
|
||||
}
|
||||
|
||||
@@ -803,9 +832,12 @@ func (s *Server) handleASDetail() http.HandlerFunc {
|
||||
|
||||
asInfo, prefixes, err := s.db.GetASDetailsContext(r.Context(), asn)
|
||||
if err != nil {
|
||||
if errors.Is(err, database.ErrNoRoute) {
|
||||
switch {
|
||||
case errors.Is(err, database.ErrNoRoute):
|
||||
http.Error(w, "AS not found", http.StatusNotFound)
|
||||
} else {
|
||||
case deadlinePassed(r.Context()):
|
||||
http.Error(w, requestTimeoutMessage, http.StatusGatewayTimeout)
|
||||
default:
|
||||
s.logger.Error("Failed to get AS details", "error", err)
|
||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
||||
}
|
||||
@@ -930,11 +962,14 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
|
||||
}
|
||||
prefix := parsed.Masked().String()
|
||||
|
||||
routes, originASNs, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
|
||||
routes, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
|
||||
if err != nil {
|
||||
if errors.Is(err, database.ErrNoRoute) {
|
||||
switch {
|
||||
case errors.Is(err, database.ErrNoRoute):
|
||||
http.Error(w, "Prefix not found", http.StatusNotFound)
|
||||
} else {
|
||||
case deadlinePassed(r.Context()):
|
||||
http.Error(w, requestTimeoutMessage, http.StatusGatewayTimeout)
|
||||
default:
|
||||
s.logger.Error("Failed to get prefix details", "error", err)
|
||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
||||
}
|
||||
@@ -952,10 +987,18 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
|
||||
originMap := make(map[int]*ASNInfo)
|
||||
for _, route := range routes {
|
||||
if _, exists := originMap[route.OriginASN]; !exists {
|
||||
// Get AS info from database
|
||||
asInfo, _, _ := s.db.GetASDetailsContext(r.Context(), route.OriginASN)
|
||||
handle := ""
|
||||
description := ""
|
||||
if asInfo != nil {
|
||||
handle = asInfo.Handle
|
||||
description = asInfo.Description
|
||||
}
|
||||
originMap[route.OriginASN] = &ASNInfo{
|
||||
ASN: route.OriginASN,
|
||||
Handle: originASNs[route.OriginASN].Handle,
|
||||
Description: originASNs[route.OriginASN].Description,
|
||||
Handle: handle,
|
||||
Description: description,
|
||||
PeerCount: 0,
|
||||
}
|
||||
}
|
||||
@@ -1080,8 +1123,12 @@ func (s *Server) handlePrefixLength() http.HandlerFunc {
|
||||
const maxPrefixes = 500
|
||||
prefixes, err := s.db.GetRandomPrefixesByLengthContext(r.Context(), maskLength, ipVersion, maxPrefixes)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to get prefixes by length", "error", err)
|
||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
||||
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)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
@@ -1171,8 +1218,12 @@ func (s *Server) handlePrefixLength6() http.HandlerFunc {
|
||||
const maxPrefixes = 500
|
||||
prefixes, err := s.db.GetRandomPrefixesByLengthContext(r.Context(), maskLength, ipVersion, maxPrefixes)
|
||||
if err != nil {
|
||||
s.logger.Error("Failed to get prefixes by length", "error", err)
|
||||
http.Error(w, "Internal server error", http.StatusInternalServerError)
|
||||
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)
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
@@ -3,13 +3,10 @@ package server
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"html"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"runtime"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -18,7 +15,6 @@ import (
|
||||
"git.eeqj.de/sneak/routewatch/internal/logger"
|
||||
"git.eeqj.de/sneak/routewatch/internal/metrics"
|
||||
"git.eeqj.de/sneak/routewatch/internal/streamer"
|
||||
"git.eeqj.de/sneak/routewatch/pkg/asinfo"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
@@ -204,72 +200,179 @@ func TestPrefixDetailHandlersParsePrefix(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// failingASDetailsDB wraps a real store and fails every AS details lookup.
|
||||
type failingASDetailsDB struct {
|
||||
// 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 (failingASDetailsDB) GetASDetailsContext(_ context.Context, _ int) (*database.ASN, []database.LiveRoute, error) {
|
||||
return nil, nil, errors.New("AS details lookup failed")
|
||||
func (slowDB) GetStatsContext(ctx context.Context) (database.Stats, error) {
|
||||
<-ctx.Done()
|
||||
|
||||
return database.Stats{}, ctx.Err()
|
||||
}
|
||||
|
||||
// TestPrefixDetailShowsOriginNames checks that the prefix page shows each
|
||||
// origin AS's handle and description without looking up the AS's details,
|
||||
// which also read every route the AS originates. Before the fix the page did
|
||||
// that lookup once per origin, so with it failing the names were missing
|
||||
// (https://git.eeqj.de/sneak/routewatch/issues/67). The AS path column shows
|
||||
// handles too, so the test looks for the origin list's "(handle)" form.
|
||||
func TestPrefixDetailShowsOriginNames(t *testing.T) {
|
||||
db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
|
||||
if err != nil {
|
||||
t.Fatalf("database.New: %v", err)
|
||||
}
|
||||
t.Cleanup(func() { _ = db.Close() })
|
||||
func (slowDB) GetIPInfoContext(ctx context.Context, _ string) (*database.IPInfo, error) {
|
||||
<-ctx.Done()
|
||||
|
||||
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||
origins := []int{15169, 13335}
|
||||
for _, asn := range origins {
|
||||
if _, err := db.GetOrCreateASN(asn, ts); err != nil {
|
||||
t.Fatalf("GetOrCreateASN(%d): %v", asn, err)
|
||||
}
|
||||
}
|
||||
if err := db.UpsertLiveRouteBatch([]*database.LiveRoute{
|
||||
{
|
||||
ID: uuid.New(), Prefix: "192.0.2.0/24", MaskLength: 24, IPVersion: 4,
|
||||
OriginASN: origins[0], PeerIP: "198.51.100.1", ASPath: []int{origins[0]}, NextHop: "198.51.100.1",
|
||||
LastUpdated: ts,
|
||||
},
|
||||
{
|
||||
ID: uuid.New(), Prefix: "192.0.2.0/24", MaskLength: 24, IPVersion: 4,
|
||||
OriginASN: origins[1], PeerIP: "198.51.100.2", ASPath: []int{origins[1]}, NextHop: "198.51.100.2",
|
||||
LastUpdated: ts,
|
||||
},
|
||||
}); err != nil {
|
||||
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||
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},
|
||||
}
|
||||
|
||||
s := New(failingASDetailsDB{db}, streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{})
|
||||
rec := httptest.NewRecorder()
|
||||
s.router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, "/prefix/192.0.2.0/24", nil))
|
||||
if rec.Code != http.StatusOK {
|
||||
t.Fatalf("status %d, want %d; body %s", rec.Code, http.StatusOK, rec.Body)
|
||||
}
|
||||
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()
|
||||
|
||||
page := rec.Body.String()
|
||||
for _, asn := range origins {
|
||||
info, ok := asinfo.Get(asn)
|
||||
if !ok {
|
||||
t.Fatalf("AS%d is not in the AS info data", asn)
|
||||
if rec.Code != http.StatusGatewayTimeout {
|
||||
t.Errorf("%s: status %d, want %d; body %s", tt.path, rec.Code, http.StatusGatewayTimeout, rec.Body)
|
||||
}
|
||||
if want := "(" + html.EscapeString(info.Handle) + ")"; !strings.Contains(page, want) {
|
||||
t.Errorf("page does not show AS%d's handle %q", asn, want)
|
||||
}
|
||||
if want := html.EscapeString(info.Description); !strings.Contains(page, want) {
|
||||
t.Errorf("page does not show AS%d's description %q", asn, want)
|
||||
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 {
|
||||
|
||||
+17
-131
@@ -4,6 +4,7 @@ import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
@@ -109,142 +110,24 @@ func JSONResponseMiddleware(next http.Handler) http.Handler {
|
||||
})
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
// requestTimeoutMessage is the error message sent with 504 Gateway Timeout.
|
||||
const requestTimeoutMessage = "Request timeout"
|
||||
|
||||
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)
|
||||
}
|
||||
})
|
||||
}
|
||||
// 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)
|
||||
}
|
||||
|
||||
// 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. 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.
|
||||
// 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.
|
||||
//
|
||||
// Non-JSON responses are passed through unchanged.
|
||||
func JSONValidationMiddleware(next http.Handler) http.Handler {
|
||||
@@ -293,11 +176,14 @@ 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.StatusRequestTimeout {
|
||||
errorMsg = "Request timeout"
|
||||
if statusCode == http.StatusGatewayTimeout {
|
||||
errorMsg = requestTimeoutMessage
|
||||
}
|
||||
|
||||
response := map[string]interface{}{
|
||||
|
||||
@@ -16,9 +16,12 @@ func (s *Server) setupRoutes() {
|
||||
r.Use(middleware.RealIP)
|
||||
r.Use(RequestLoggerMiddleware(s.logger.Logger)) // Structured request logging
|
||||
r.Use(middleware.Recoverer)
|
||||
const requestTimeout = 30 * time.Second // Increased from 8s for slow queries
|
||||
r.Use(TimeoutMiddleware(requestTimeout, s.logger.Logger))
|
||||
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))
|
||||
|
||||
// Routes
|
||||
r.Get("/", s.handleIndex())
|
||||
|
||||
@@ -3,8 +3,6 @@ package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"time"
|
||||
@@ -55,12 +53,9 @@ func New(db database.Store, streamer *streamer.Streamer, logger *logger.Logger,
|
||||
return s
|
||||
}
|
||||
|
||||
// Start listens on the configured port and serves HTTP in the background.
|
||||
// It returns an error if it cannot listen, for example because the port is
|
||||
// already in use.
|
||||
// Start starts the HTTP server
|
||||
func (s *Server) Start() error {
|
||||
const (
|
||||
network = "tcp"
|
||||
readHeaderTimeout = 40 * time.Second
|
||||
readTimeout = 60 * time.Second
|
||||
writeTimeout = 60 * time.Second
|
||||
@@ -78,15 +73,10 @@ func (s *Server) Start() error {
|
||||
|
||||
s.logger.Info("Starting HTTP server", "port", s.port, "addr", s.srv.Addr)
|
||||
|
||||
listener, err := net.Listen(network, s.srv.Addr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("starting HTTP server: %w", err)
|
||||
}
|
||||
|
||||
s.logger.Info("HTTP server listening", "addr", listener.Addr().String())
|
||||
|
||||
// Start in goroutine but log when actually listening
|
||||
go func() {
|
||||
if err := s.srv.Serve(listener); err != nil && err != http.ErrServerClosed {
|
||||
s.logger.Info("HTTP server listening", "addr", s.srv.Addr)
|
||||
if err := s.srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
||||
s.logger.Error("HTTP server error", "error", err)
|
||||
}
|
||||
}()
|
||||
|
||||
@@ -1,33 +0,0 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"testing"
|
||||
|
||||
"git.eeqj.de/sneak/routewatch/internal/config"
|
||||
"git.eeqj.de/sneak/routewatch/internal/logger"
|
||||
"git.eeqj.de/sneak/routewatch/internal/metrics"
|
||||
"git.eeqj.de/sneak/routewatch/internal/streamer"
|
||||
)
|
||||
|
||||
// TestStartFailsWhenPortIsHeld holds a port with a listener of its own and
|
||||
// checks that Start on that port returns an error. Before the fix Start
|
||||
// listened in a goroutine, only logged the failure and returned nil, so the
|
||||
// daemon kept running without its web server.
|
||||
func TestStartFailsWhenPortIsHeld(t *testing.T) {
|
||||
held, err := net.Listen("tcp", ":0")
|
||||
if err != nil {
|
||||
t.Fatalf("failed to open the test listener: %v", err)
|
||||
}
|
||||
defer func() { _ = held.Close() }()
|
||||
|
||||
port := held.Addr().(*net.TCPAddr).Port
|
||||
log := logger.New()
|
||||
s := New(nil, streamer.New(log, metrics.New()), log, &config.Config{Port: port})
|
||||
defer func() { _ = s.Stop(context.Background()) }()
|
||||
|
||||
if err := s.Start(); err == nil {
|
||||
t.Fatalf("Start returned nil while port %d was held by another listener", port)
|
||||
}
|
||||
}
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
"compress/gzip"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"math"
|
||||
@@ -49,7 +48,6 @@ const (
|
||||
metricsUpdateRate = time.Second
|
||||
minBackoffDelay = 5 * time.Second
|
||||
maxBackoffDelay = 320 * time.Second
|
||||
streamIdleTimeout = 60 * time.Second // Reconnect when no line arrives for this long
|
||||
metricsLogInterval = 10 * time.Second
|
||||
bytesPerKB = 1024
|
||||
bytesPerMB = 1024 * 1024
|
||||
@@ -60,9 +58,6 @@ const (
|
||||
backpressureSlope = 2.0 // Slope for linear drop probability increase
|
||||
)
|
||||
|
||||
// errStreamEnded is returned by stream when RIS Live ends the response.
|
||||
var errStreamEnded = errors.New("RIS Live ended the stream")
|
||||
|
||||
// MessageHandler defines the interface for processing RIS messages.
|
||||
// Implementations must specify which message types they want to receive,
|
||||
// how to process messages, and their desired queue capacity.
|
||||
@@ -112,8 +107,6 @@ type Streamer struct {
|
||||
logger *logger.Logger
|
||||
client *http.Client
|
||||
url string
|
||||
idleTimeout time.Duration // Drop the connection when no line arrives for this long
|
||||
minBackoff time.Duration // First wait before reconnecting; doubles up to maxBackoffDelay
|
||||
handlers []*handlerInfo
|
||||
rawHandler RawMessageHandler
|
||||
mu sync.RWMutex
|
||||
@@ -131,10 +124,8 @@ type Streamer struct {
|
||||
// The metrics tracker is used to record message counts, bytes received, and connection status.
|
||||
func New(logger *logger.Logger, metrics *metrics.Tracker) *Streamer {
|
||||
return &Streamer{
|
||||
logger: logger,
|
||||
url: risLiveURL,
|
||||
idleTimeout: streamIdleTimeout,
|
||||
minBackoff: minBackoffDelay,
|
||||
logger: logger,
|
||||
url: risLiveURL,
|
||||
client: &http.Client{
|
||||
Timeout: 0, // No timeout for streaming
|
||||
Transport: &http.Transport{
|
||||
@@ -414,7 +405,7 @@ func (s *Streamer) updateMetrics(messageBytes int) {
|
||||
|
||||
// streamWithReconnect handles streaming with automatic reconnection and exponential backoff
|
||||
func (s *Streamer) streamWithReconnect(ctx context.Context) {
|
||||
backoffDelay := s.minBackoff
|
||||
backoffDelay := minBackoffDelay
|
||||
consecutiveFailures := 0
|
||||
|
||||
for {
|
||||
@@ -431,9 +422,8 @@ func (s *Streamer) streamWithReconnect(ctx context.Context) {
|
||||
err := s.stream(ctx)
|
||||
streamDuration := time.Since(startTime)
|
||||
|
||||
// Only Stop ends the streamer. A stream that ends any other way,
|
||||
// with or without an error, is a disconnect and is reconnected.
|
||||
if ctx.Err() != nil {
|
||||
if err == nil {
|
||||
// Clean exit (context cancelled)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -444,13 +434,17 @@ func (s *Streamer) streamWithReconnect(ctx context.Context) {
|
||||
"stream_duration", streamDuration)
|
||||
s.metrics.SetConnected(false)
|
||||
|
||||
// A stream that lasted more than 30 seconds resets the backoff. That
|
||||
// includes a stream dropped after streamIdleTimeout with no line, so
|
||||
// a silent feed is retried after the first backoff delay every time.
|
||||
// Check if context is cancelled
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
|
||||
// If we streamed for more than 30 seconds, reset the backoff
|
||||
// This indicates we had a successful connection that received data
|
||||
if streamDuration > 30*time.Second {
|
||||
s.logger.Info("Resetting backoff delay due to successful connection",
|
||||
"stream_duration", streamDuration)
|
||||
backoffDelay = s.minBackoff
|
||||
backoffDelay = minBackoffDelay
|
||||
consecutiveFailures = 0
|
||||
} else {
|
||||
// Increment consecutive failures
|
||||
@@ -483,13 +477,7 @@ func (s *Streamer) stream(ctx context.Context) error {
|
||||
connCtx, connCancel := context.WithCancel(ctx)
|
||||
defer connCancel()
|
||||
|
||||
// The idle timer drops the connection, by cancelling connCtx, when no
|
||||
// line arrives for s.idleTimeout. Each line read starts it over. It
|
||||
// also covers the wait for the server to answer.
|
||||
idleTimer := time.AfterFunc(s.idleTimeout, connCancel)
|
||||
defer idleTimer.Stop()
|
||||
|
||||
req, err := http.NewRequestWithContext(connCtx, "GET", s.url, nil)
|
||||
req, err := http.NewRequestWithContext(ctx, "GET", s.url, nil)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to create request: %w", err)
|
||||
}
|
||||
@@ -571,8 +559,6 @@ func (s *Streamer) stream(ctx context.Context) error {
|
||||
scanner := bufio.NewScanner(reader)
|
||||
|
||||
for scanner.Scan() {
|
||||
idleTimer.Reset(s.idleTimeout)
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
s.logger.Info("Stream stopped by context")
|
||||
@@ -721,19 +707,11 @@ func (s *Streamer) stream(ctx context.Context) error {
|
||||
s.mu.RUnlock()
|
||||
}
|
||||
|
||||
if ctx.Err() != nil {
|
||||
return ctx.Err()
|
||||
}
|
||||
|
||||
if connCtx.Err() != nil {
|
||||
return fmt.Errorf("no line received for %s", s.idleTimeout)
|
||||
}
|
||||
|
||||
if err := scanner.Err(); err != nil {
|
||||
return fmt.Errorf("scanner error: %w", err)
|
||||
}
|
||||
|
||||
return errStreamEnded
|
||||
return nil
|
||||
}
|
||||
|
||||
// shouldDropForBackpressure determines if a message should be dropped based on queue utilization
|
||||
|
||||
@@ -7,7 +7,6 @@ import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"runtime"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -40,14 +39,6 @@ func TestNewStreamer(t *testing.T) {
|
||||
if s.metrics != metricsTracker {
|
||||
t.Error("metrics tracker not set correctly")
|
||||
}
|
||||
|
||||
if s.idleTimeout != streamIdleTimeout {
|
||||
t.Errorf("idleTimeout = %s, want %s", s.idleTimeout, streamIdleTimeout)
|
||||
}
|
||||
|
||||
if s.minBackoff != minBackoffDelay {
|
||||
t.Errorf("minBackoff = %s, want %s", s.minBackoff, minBackoffDelay)
|
||||
}
|
||||
}
|
||||
|
||||
// TestStreamDoesNotLeakTickersAcrossReconnects drives many short-lived
|
||||
@@ -66,8 +57,8 @@ func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) {
|
||||
|
||||
// One warm-up connection so any persistent HTTP transport goroutine exists
|
||||
// before we take the baseline.
|
||||
if err := s.stream(context.Background()); !errors.Is(err, errStreamEnded) {
|
||||
t.Fatalf("warm-up stream returned %v, want %v", err, errStreamEnded)
|
||||
if err := s.stream(context.Background()); err != nil {
|
||||
t.Fatalf("warm-up stream returned error: %v", err)
|
||||
}
|
||||
s.client.CloseIdleConnections()
|
||||
|
||||
@@ -75,8 +66,8 @@ func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) {
|
||||
|
||||
const reconnects = 20
|
||||
for range reconnects {
|
||||
if err := s.stream(context.Background()); !errors.Is(err, errStreamEnded) {
|
||||
t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
|
||||
if err := s.stream(context.Background()); err != nil {
|
||||
t.Fatalf("stream returned error: %v", err)
|
||||
}
|
||||
}
|
||||
s.client.CloseIdleConnections()
|
||||
@@ -87,35 +78,23 @@ func TestStreamDoesNotLeakTickersAcrossReconnects(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// updateLine is one UPDATE message as RIS Live sends it.
|
||||
const updateLine = `{"type":"ris_message","data":{"type":"UPDATE","peer":"192.0.2.1",` +
|
||||
`"peer_asn":"64496","timestamp":1700000000}}` + "\n"
|
||||
|
||||
// updateHandler wants UPDATE messages. For each one it handles it sends on
|
||||
// handled if there is room, and otherwise does nothing; a nil handled is
|
||||
// never sent on.
|
||||
type updateHandler struct {
|
||||
handled chan struct{}
|
||||
}
|
||||
// updateHandler wants UPDATE messages and does nothing with them.
|
||||
type updateHandler struct{}
|
||||
|
||||
func (updateHandler) WantsMessage(messageType string) bool { return messageType == "UPDATE" }
|
||||
func (updateHandler) HandleMessage(*ristypes.RISMessage) {}
|
||||
func (updateHandler) QueueCapacity() int { return 10 }
|
||||
|
||||
func (h updateHandler) HandleMessage(*ristypes.RISMessage) {
|
||||
select {
|
||||
case h.handled <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
// TestStopBeforeMessageReachesQueues stops the streamer after the read loop
|
||||
// has checked for cancellation but before it hands the message to the handler
|
||||
// queues. That is the gap a stop from another goroutine can land in, and it
|
||||
// used to end in "send on closed channel". The raw handler runs in that gap on
|
||||
// the read loop itself, so calling Stop from it hits the gap every time.
|
||||
func TestStopBeforeMessageReachesQueues(t *testing.T) {
|
||||
const line = `{"type":"ris_message","data":{"type":"UPDATE","peer":"192.0.2.1",` +
|
||||
`"peer_asn":"64496","timestamp":1700000000}}` + "\n"
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
_, _ = io.WriteString(w, updateLine)
|
||||
_, _ = io.WriteString(w, line)
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
@@ -138,117 +117,6 @@ func TestStopBeforeMessageReachesQueues(t *testing.T) {
|
||||
s.Stop()
|
||||
}
|
||||
|
||||
// TestReconnectWhenStreamEnds has the server end the first response
|
||||
// normally, as when RIS Live restarts. The streamer must treat that as a
|
||||
// disconnect and connect again.
|
||||
func TestReconnectWhenStreamEnds(t *testing.T) {
|
||||
endResponse := func(http.ResponseWriter, *http.Request) {}
|
||||
checkReconnect(t, endResponse, streamIdleTimeout)
|
||||
}
|
||||
|
||||
// TestReconnectWhenStreamGoesIdle has the server hold the first response
|
||||
// open without sending anything. The streamer must drop it once the idle
|
||||
// timeout passes and connect again.
|
||||
func TestReconnectWhenStreamGoesIdle(t *testing.T) {
|
||||
const idleTimeout = 50 * time.Millisecond
|
||||
holdResponse := func(w http.ResponseWriter, r *http.Request) {
|
||||
w.(http.Flusher).Flush()
|
||||
<-r.Context().Done()
|
||||
}
|
||||
checkReconnect(t, holdResponse, idleTimeout)
|
||||
}
|
||||
|
||||
// TestReconnectWhenServerNeverAnswers has the server accept the first
|
||||
// request and never answer it, not even with headers. The streamer must give
|
||||
// up on it once the idle timeout passes and connect again.
|
||||
func TestReconnectWhenServerNeverAnswers(t *testing.T) {
|
||||
const idleTimeout = 50 * time.Millisecond
|
||||
neverAnswer := func(_ http.ResponseWriter, r *http.Request) {
|
||||
<-r.Context().Done()
|
||||
}
|
||||
checkReconnect(t, neverAnswer, idleTimeout)
|
||||
}
|
||||
|
||||
// TestStreamNotDroppedWhileLinesArrive has the server send lines well within
|
||||
// the idle timeout for several times the timeout, then end the response. Each
|
||||
// line must start the idle timeout over, so the stream lasts until the server
|
||||
// ends it instead of being dropped as idle.
|
||||
func TestStreamNotDroppedWhileLinesArrive(t *testing.T) {
|
||||
const (
|
||||
idleTimeout = 100 * time.Millisecond
|
||||
lineInterval = idleTimeout / 10
|
||||
sendFor = 5 * idleTimeout
|
||||
)
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
|
||||
deadline := time.Now().Add(sendFor)
|
||||
for time.Now().Before(deadline) {
|
||||
_, _ = io.WriteString(w, updateLine)
|
||||
w.(http.Flusher).Flush()
|
||||
time.Sleep(lineInterval)
|
||||
}
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
s := New(logger.New(), metrics.New())
|
||||
s.url = srv.URL
|
||||
s.idleTimeout = idleTimeout
|
||||
|
||||
if err := s.stream(context.Background()); !errors.Is(err, errStreamEnded) {
|
||||
t.Fatalf("stream returned %v, want %v", err, errStreamEnded)
|
||||
}
|
||||
}
|
||||
|
||||
// checkReconnect runs a streamer against a fake RIS Live server. The first
|
||||
// response is served by firstResponse; every later one carries one UPDATE
|
||||
// line and then stays open. The streamer must come back for a second
|
||||
// response, must not be marked connected when it does, and must handle the
|
||||
// UPDATE in it.
|
||||
func checkReconnect(t *testing.T, firstResponse http.HandlerFunc, idleTimeout time.Duration) {
|
||||
t.Helper()
|
||||
|
||||
const maxWait = 5 * time.Second
|
||||
|
||||
tracker := metrics.New()
|
||||
connectedAtReconnect := make(chan bool, 1)
|
||||
var requests atomic.Int32
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if requests.Add(1) == 1 {
|
||||
firstResponse(w, r)
|
||||
|
||||
return
|
||||
}
|
||||
select {
|
||||
case connectedAtReconnect <- tracker.IsConnected():
|
||||
default:
|
||||
}
|
||||
_, _ = io.WriteString(w, updateLine)
|
||||
w.(http.Flusher).Flush()
|
||||
<-r.Context().Done()
|
||||
}))
|
||||
defer srv.Close()
|
||||
|
||||
s := New(logger.New(), tracker)
|
||||
s.url = srv.URL
|
||||
s.idleTimeout = idleTimeout
|
||||
s.minBackoff = time.Millisecond
|
||||
handler := updateHandler{handled: make(chan struct{}, 1)}
|
||||
s.RegisterHandler(handler)
|
||||
if err := s.Start(); err != nil {
|
||||
t.Fatalf("Start returned error: %v", err)
|
||||
}
|
||||
defer s.Stop()
|
||||
|
||||
select {
|
||||
case <-handler.handled:
|
||||
case <-time.After(maxWait):
|
||||
t.Fatalf("no message from a second response handled within %s", maxWait)
|
||||
}
|
||||
|
||||
if <-connectedAtReconnect {
|
||||
t.Error("streamer was still marked connected when it came back for a second response")
|
||||
}
|
||||
}
|
||||
|
||||
// settledGoroutineCount lets transient goroutines finish, then reports the
|
||||
// current count.
|
||||
func settledGoroutineCount() int {
|
||||
|
||||
@@ -283,7 +283,6 @@
|
||||
<div class="origin-item">
|
||||
<a href="/as/{{.ASN}}" class="as-link">AS{{.ASN}}</a>
|
||||
{{if .Handle}} ({{.Handle}}){{end}}
|
||||
{{if .Description}} {{.Description}}{{end}}
|
||||
<span style="color: #7f8c8d; margin-left: 10px;">{{.PeerCount}} peer{{if ne .PeerCount 1}}s{{end}}</span>
|
||||
</div>
|
||||
{{end}}
|
||||
|
||||
Reference in New Issue
Block a user