Compare commits
6
Commits
main
..
431d1d0481
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
431d1d0481 | ||
|
|
b4dee48401 | ||
|
|
616f13071f | ||
|
|
521c632087 | ||
|
|
77a708d819 | ||
|
|
a43935308e |
@@ -144,8 +144,9 @@ routewatch/
|
|||||||
- **Reconnection**: Exponential backoff (5s-320s) with reset after 30s of stable connection
|
- **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
|
- **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
|
address's prefix at each mask length, longest first, in the prefix index
|
||||||
(at most 33 lookups for IPv4, 129 for IPv6); prefixes are stored in the
|
(at most 25 lookups for IPv4, 113 for IPv6); prefixes are stored in the
|
||||||
text form Go's `net/netip` prints
|
text form Go's `net/netip` prints. Routes shorter than /8 for IPv4 or /16
|
||||||
|
for IPv6, such as a default route, are not used.
|
||||||
|
|
||||||
### Database Schema
|
### Database Schema
|
||||||
|
|
||||||
|
|||||||
@@ -128,7 +128,7 @@ func TestLiveCountsTrackWritesInRealtime(t *testing.T) {
|
|||||||
|
|
||||||
// A withdrawal removes one route.
|
// A withdrawal removes one route.
|
||||||
if err := db.DeleteLiveRouteBatch([]LiveRouteDeletion{
|
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 {
|
}); err != nil {
|
||||||
t.Fatalf("DeleteLiveRouteBatch: %v", err)
|
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
|
// withdrawRoutes removes routes in one DeleteLiveRouteBatch, or with one
|
||||||
// DeleteLiveRoute each. It names each route by prefix and peer only, with no
|
// DeleteLiveRoute each.
|
||||||
// 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 {
|
func withdrawRoutes(db *Database, batch bool, routes []*LiveRoute) error {
|
||||||
if !batch {
|
if !batch {
|
||||||
for _, route := range routes {
|
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
|
return err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -37,6 +37,14 @@ const (
|
|||||||
ipv6Bits = 128
|
ipv6Bits = 128
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// The IP lookup does not use routes shorter than these mask lengths. A default
|
||||||
|
// route or a route this short only says where a peer sends traffic it has no
|
||||||
|
// better route for; it does not say who holds the address.
|
||||||
|
const (
|
||||||
|
minLookupMaskLengthV4 = 8
|
||||||
|
minLookupMaskLengthV6 = 16
|
||||||
|
)
|
||||||
|
|
||||||
// SQLite memory tuning. cache_size and busy_timeout go in the DSN so every
|
// SQLite memory tuning. cache_size and busy_timeout go in the DSN so every
|
||||||
// pooled connection gets them; the heap limits are process-wide and set once.
|
// pooled connection gets them; the heap limits are process-wide and set once.
|
||||||
const (
|
const (
|
||||||
@@ -220,21 +228,22 @@ func (d *Database) beginTx() (*loggingTx, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// A live-route upsert is an UPDATE followed, only when no row matched, by an
|
// 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
|
// INSERT. A live route is keyed by prefix and peer, so an announcement replaces
|
||||||
// one) is what lets the in-memory route counters stay exact without a COUNT(*).
|
// the peer's route for the prefix whatever its origin AS. The UPDATE's
|
||||||
// Callers hold the database write lock, so no other writer can insert the same
|
// rows-affected count (1 for an existing key, 0 for a new one) is what lets the
|
||||||
// key between the two statements. The id column is set only on INSERT, so an
|
// in-memory route counters stay exact without a COUNT(*). Callers hold the
|
||||||
// updated route keeps its original id, exactly as the previous ON CONFLICT
|
// database write lock, so no other writer can insert the same key between the
|
||||||
// upsert did.
|
// 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 (
|
const (
|
||||||
updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, as_path = ?, next_hop = ?,
|
updateLiveRouteV4SQL = `UPDATE live_routes_v4 SET mask_length = ?, origin_asn = ?, as_path = ?,
|
||||||
last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
|
next_hop = ?, last_updated = ? WHERE prefix = ? AND peer_ip = ?`
|
||||||
|
|
||||||
insertLiveRouteV4SQL = `INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, peer_ip,
|
insertLiveRouteV4SQL = `INSERT INTO live_routes_v4 (id, prefix, mask_length, origin_asn, peer_ip,
|
||||||
as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
|
as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
|
||||||
|
|
||||||
updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, as_path = ?, next_hop = ?,
|
updateLiveRouteV6SQL = `UPDATE live_routes_v6 SET mask_length = ?, origin_asn = ?, as_path = ?,
|
||||||
last_updated = ? WHERE prefix = ? AND origin_asn = ? AND peer_ip = ?`
|
next_hop = ?, last_updated = ? WHERE prefix = ? AND peer_ip = ?`
|
||||||
|
|
||||||
insertLiveRouteV6SQL = `INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip,
|
insertLiveRouteV6SQL = `INSERT INTO live_routes_v6 (id, prefix, mask_length, origin_asn, peer_ip,
|
||||||
as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
|
as_path, next_hop, last_updated) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
|
||||||
@@ -253,8 +262,8 @@ const (
|
|||||||
// live route for its prefix.
|
// live route for its prefix.
|
||||||
func upsertRouteRowV4(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) (
|
func upsertRouteRowV4(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) (
|
||||||
inserted, newPrefix bool, err error) {
|
inserted, newPrefix bool, err error) {
|
||||||
res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
|
res, err := upd.Exec(route.MaskLength, route.OriginASN, pathJSON, route.NextHop, route.LastUpdated,
|
||||||
route.Prefix, route.OriginASN, route.PeerIP)
|
route.Prefix, route.PeerIP)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, false, err
|
return false, false, err
|
||||||
}
|
}
|
||||||
@@ -285,8 +294,8 @@ func upsertRouteRowV4(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string
|
|||||||
// live route for its prefix.
|
// live route for its prefix.
|
||||||
func upsertRouteRowV6(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) (
|
func upsertRouteRowV6(upd, ins, has *sql.Stmt, route *LiveRoute, pathJSON string) (
|
||||||
inserted, newPrefix bool, err error) {
|
inserted, newPrefix bool, err error) {
|
||||||
res, err := upd.Exec(route.MaskLength, pathJSON, route.NextHop, route.LastUpdated,
|
res, err := upd.Exec(route.MaskLength, route.OriginASN, pathJSON, route.NextHop, route.LastUpdated,
|
||||||
route.Prefix, route.OriginASN, route.PeerIP)
|
route.Prefix, route.PeerIP)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, false, err
|
return false, false, err
|
||||||
}
|
}
|
||||||
@@ -433,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
|
// 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 {
|
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 {
|
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() }()
|
defer func() { _ = deleteV6.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)
|
hasV4, err := tx.Prepare(prefixHasLiveRouteV4SQL)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -477,38 +472,18 @@ func (d *Database) DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error {
|
|||||||
// Mask lengths of the prefixes this batch leaves with no live route.
|
// Mask lengths of the prefixes this batch leaves with no live route.
|
||||||
var gonePrefixMaskLengthsV4, gonePrefixMaskLengthsV6 []int
|
var gonePrefixMaskLengthsV4, gonePrefixMaskLengthsV6 []int
|
||||||
for _, del := range deletions {
|
for _, del := range deletions {
|
||||||
var stmt *sql.Stmt
|
stmt := deleteV4
|
||||||
|
if del.IPVersion != ipVersionV4 {
|
||||||
// Select appropriate statement based on IP version and whether we have origin ASN
|
stmt = deleteV6
|
||||||
//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 {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to delete route %s: %w", del.Prefix, err)
|
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
|
// The feed often withdraws a route that is not live, which deletes
|
||||||
// exact rows-affected count to keep the in-memory route counters right.
|
// nothing, so the in-memory route counters use the rows-affected count.
|
||||||
affected, err := res.RowsAffected()
|
affected, err := res.RowsAffected()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to count deleted route %s: %w", del.Prefix, err)
|
return fmt.Errorf("failed to count deleted route %s: %w", del.Prefix, err)
|
||||||
@@ -1274,9 +1249,8 @@ func (d *Database) UpsertLiveRoute(route *LiveRoute) error {
|
|||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// DeleteLiveRoute deletes a live route
|
// DeleteLiveRoute deletes the live route the peer has for the prefix
|
||||||
// If originASN is 0, deletes all routes for the prefix/peer combination
|
func (d *Database) DeleteLiveRoute(prefix, peerIP string) error {
|
||||||
func (d *Database) DeleteLiveRoute(prefix string, originASN int, peerIP string) error {
|
|
||||||
d.lock("DeleteLiveRoute")
|
d.lock("DeleteLiveRoute")
|
||||||
defer d.unlock()
|
defer d.unlock()
|
||||||
|
|
||||||
@@ -1301,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)
|
// 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
|
// so the delete carries no dynamically built SQL.
|
||||||
// ASN can remove several rows.
|
|
||||||
var res sql.Result
|
var res sql.Result
|
||||||
switch {
|
if isV4 {
|
||||||
case isV4 && originASN == 0:
|
|
||||||
res, err = tx.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
|
res, err = tx.Exec(`DELETE FROM live_routes_v4 WHERE prefix = ? AND peer_ip = ?`, prefix, peerIP)
|
||||||
case isV4:
|
} else {
|
||||||
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)
|
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 {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -1876,11 +1840,12 @@ func (d *Database) GetIPInfo(ip string) (*IPInfo, error) {
|
|||||||
// GetIPInfoContext returns comprehensive IP information with context support.
|
// GetIPInfoContext returns comprehensive IP information with context support.
|
||||||
//
|
//
|
||||||
// It finds the most specific live route for the address with point lookups on
|
// It finds the most specific live route for the address with point lookups on
|
||||||
// the prefix index: for each mask length from the longest (32 or 128) down to 0
|
// the prefix index: for each mask length from the longest (32 or 128) down to
|
||||||
// it looks up the address's network prefix at that length, and the first length
|
// minLookupMaskLengthV4 or minLookupMaskLengthV6 it looks up the address's
|
||||||
// with a live route wins. That is at most 33 or 129 indexed lookups, however
|
// network prefix at that length, and the first length with a live route wins.
|
||||||
// many routes are live. The prefix is looked up in the text form net/netip
|
// That is at most 25 or 113 indexed lookups, however many routes are live. The
|
||||||
// prints, which is the form the prefix handler stores.
|
// prefix is looked up in the text form net/netip prints, which is the form the
|
||||||
|
// prefix handler stores.
|
||||||
func (d *Database) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, error) {
|
func (d *Database) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, error) {
|
||||||
addr, err := netip.ParseAddr(ip)
|
addr, err := netip.ParseAddr(ip)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -1891,9 +1856,11 @@ func (d *Database) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, er
|
|||||||
|
|
||||||
info := &IPInfo{IP: ip, IPVersion: ipVersionV6}
|
info := &IPInfo{IP: ip, IPVersion: ipVersionV6}
|
||||||
routesTable, prefixesTable := "live_routes_v6", "prefixes_v6"
|
routesTable, prefixesTable := "live_routes_v6", "prefixes_v6"
|
||||||
|
minMaskLength := minLookupMaskLengthV6
|
||||||
if addr.Is4() {
|
if addr.Is4() {
|
||||||
info.IPVersion = ipVersionV4
|
info.IPVersion = ipVersionV4
|
||||||
routesTable, prefixesTable = "live_routes_v4", "prefixes_v4"
|
routesTable, prefixesTable = "live_routes_v4", "prefixes_v4"
|
||||||
|
minMaskLength = minLookupMaskLengthV4
|
||||||
}
|
}
|
||||||
|
|
||||||
// Route info with peer count and prefix first_seen
|
// Route info with peer count and prefix first_seen
|
||||||
@@ -1934,7 +1901,7 @@ func (d *Database) GetIPInfoContext(ctx context.Context, ip string) (*IPInfo, er
|
|||||||
var whoisUpdatedAt sql.NullTime
|
var whoisUpdatedAt sql.NullTime
|
||||||
|
|
||||||
found := false
|
found := false
|
||||||
for bits := addr.BitLen(); bits >= 0; bits-- {
|
for bits := addr.BitLen(); bits >= minMaskLength; bits-- {
|
||||||
prefix := netip.PrefixFrom(addr, bits).Masked().String()
|
prefix := netip.PrefixFrom(addr, bits).Masked().String()
|
||||||
err := stmt.QueryRowContext(ctx, prefix).Scan(
|
err := stmt.QueryRowContext(ctx, prefix).Scan(
|
||||||
&info.Netblock,
|
&info.Netblock,
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ import (
|
|||||||
"git.eeqj.de/sneak/routewatch/internal/config"
|
"git.eeqj.de/sneak/routewatch/internal/config"
|
||||||
"git.eeqj.de/sneak/routewatch/internal/logger"
|
"git.eeqj.de/sneak/routewatch/internal/logger"
|
||||||
"github.com/google/uuid"
|
"github.com/google/uuid"
|
||||||
|
"github.com/mattn/go-sqlite3"
|
||||||
)
|
)
|
||||||
|
|
||||||
// tempStoreMemory is the PRAGMA temp_store value meaning "hold temp B-trees in
|
// tempStoreMemory is the PRAGMA temp_store value meaning "hold temp B-trees in
|
||||||
@@ -122,8 +123,8 @@ func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) {
|
|||||||
lookup("2001:db9::1", "")
|
lookup("2001:db9::1", "")
|
||||||
|
|
||||||
err = db.DeleteLiveRouteBatch([]LiveRouteDeletion{
|
err = db.DeleteLiveRouteBatch([]LiveRouteDeletion{
|
||||||
{Prefix: "10.1.2.0/24", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV4},
|
{Prefix: "10.1.2.0/24", PeerIP: testPeerIP, IPVersion: ipVersionV4},
|
||||||
{Prefix: "2001:db8:1:2::/64", OriginASN: 64502, PeerIP: testPeerIP, IPVersion: ipVersionV6},
|
{Prefix: "2001:db8:1:2::/64", PeerIP: testPeerIP, IPVersion: ipVersionV6},
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("DeleteLiveRouteBatch: %v", err)
|
t.Fatalf("DeleteLiveRouteBatch: %v", err)
|
||||||
@@ -133,6 +134,112 @@ func TestGetIPInfoFindsMostSpecificLiveRoute(t *testing.T) {
|
|||||||
lookup("2001:db8:1:2::3", "2001:db8:1::/48")
|
lookup("2001:db8:1:2::3", "2001:db8:1::/48")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestGetIPInfoIgnoresShortRoutes checks that the IP lookup does not report a
|
||||||
|
// default route, or a route shorter than /8 for IPv4 or /16 for IPv6, as the
|
||||||
|
// route for an address, and that a route of exactly /8 or /16, or longer, still
|
||||||
|
// is. The prefixes are written out so that a wrong limit fails the test.
|
||||||
|
func TestGetIPInfoIgnoresShortRoutes(t *testing.T) {
|
||||||
|
cfg := &config.Config{StateDir: t.TempDir()}
|
||||||
|
|
||||||
|
db, err := New(cfg, logger.New())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("failed to create database: %v", err)
|
||||||
|
}
|
||||||
|
defer func() { _ = db.Close() }()
|
||||||
|
|
||||||
|
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
announce := func(prefix string, asn int) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
err := db.UpsertLiveRouteBatch([]*LiveRoute{{
|
||||||
|
ID: uuid.New(),
|
||||||
|
Prefix: prefix,
|
||||||
|
MaskLength: netip.MustParsePrefix(prefix).Bits(),
|
||||||
|
IPVersion: detectIPVersion(prefix),
|
||||||
|
OriginASN: asn,
|
||||||
|
PeerIP: testPeerIP,
|
||||||
|
ASPath: []int{asn},
|
||||||
|
NextHop: testPeerIP,
|
||||||
|
LastUpdated: ts,
|
||||||
|
}})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch(%s): %v", prefix, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// lookup checks that ip resolves to the live prefix netblock from asn.
|
||||||
|
lookup := func(ip, netblock string, asn int) {
|
||||||
|
t.Helper()
|
||||||
|
|
||||||
|
info, err := db.GetIPInfo(ip)
|
||||||
|
if err != nil {
|
||||||
|
t.Errorf("GetIPInfo(%s): %v", ip, err)
|
||||||
|
|
||||||
|
return
|
||||||
|
}
|
||||||
|
if info.Netblock != netblock || info.ASN != asn {
|
||||||
|
t.Errorf("GetIPInfo(%s) = %s AS%d, want %s AS%d", ip, info.Netblock, info.ASN, netblock, asn)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Default and very short routes, and routes one bit shorter than each limit.
|
||||||
|
announce("2000::/3", 64500)
|
||||||
|
announce("0.0.0.0/0", 64500)
|
||||||
|
announce("2000::/15", 64500)
|
||||||
|
announce("192.0.0.0/7", 64500)
|
||||||
|
|
||||||
|
for _, ip := range []string{"2001:db8::1", "192.0.2.1"} {
|
||||||
|
if info, err := db.GetIPInfo(ip); !errors.Is(err, ErrNoRoute) {
|
||||||
|
t.Errorf("GetIPInfo(%s) with only short routes = %+v, %v; want ErrNoRoute", ip, info, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Routes of exactly the minimum length are used.
|
||||||
|
announce("2001::/16", 64501)
|
||||||
|
announce("192.0.0.0/8", 64501)
|
||||||
|
|
||||||
|
lookup("2001:db8::1", "2001::/16", 64501)
|
||||||
|
lookup("192.0.2.1", "192.0.0.0/8", 64501)
|
||||||
|
|
||||||
|
announce("2001:db8::/32", 64502)
|
||||||
|
|
||||||
|
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
|
// TestConnectionPoolPragmas holds several pooled connections open at once and
|
||||||
// checks each one carries the per-connection settings from the DSN, plus the
|
// 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
|
// process-wide hard heap limit, and sees the new file with auto_vacuum
|
||||||
@@ -234,7 +341,6 @@ func TestVacuumReturnsFreePages(t *testing.T) {
|
|||||||
routes = append(routes, route)
|
routes = append(routes, route)
|
||||||
deletions = append(deletions, LiveRouteDeletion{
|
deletions = append(deletions, LiveRouteDeletion{
|
||||||
Prefix: route.Prefix,
|
Prefix: route.Prefix,
|
||||||
OriginASN: asn,
|
|
||||||
PeerIP: route.PeerIP,
|
PeerIP: route.PeerIP,
|
||||||
IPVersion: ipVersionV4,
|
IPVersion: ipVersionV4,
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -55,7 +55,7 @@ type Store interface {
|
|||||||
// Live route operations
|
// Live route operations
|
||||||
UpsertLiveRoute(route *LiveRoute) error
|
UpsertLiveRoute(route *LiveRoute) error
|
||||||
UpsertLiveRouteBatch(routes []*LiveRoute) error
|
UpsertLiveRouteBatch(routes []*LiveRoute) error
|
||||||
DeleteLiveRoute(prefix string, originASN int, peerIP string) error
|
DeleteLiveRoute(prefix, peerIP string) error
|
||||||
DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error
|
DeleteLiveRouteBatch(deletions []LiveRouteDeletion) error
|
||||||
GetPrefixDistribution() (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)
|
GetPrefixDistribution() (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)
|
||||||
GetPrefixDistributionContext(ctx context.Context) (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)
|
GetPrefixDistributionContext(ctx context.Context) (ipv4 []PrefixDistribution, ipv6 []PrefixDistribution, err error)
|
||||||
|
|||||||
@@ -114,7 +114,6 @@ type IPInfo struct {
|
|||||||
// LiveRouteDeletion represents parameters for deleting a live route
|
// LiveRouteDeletion represents parameters for deleting a live route
|
||||||
type LiveRouteDeletion struct {
|
type LiveRouteDeletion struct {
|
||||||
Prefix string
|
Prefix string
|
||||||
OriginASN int
|
|
||||||
PeerIP string
|
PeerIP string
|
||||||
IPVersion int
|
IPVersion int
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -107,7 +107,7 @@ CREATE TABLE IF NOT EXISTS live_routes_v4 (
|
|||||||
as_path TEXT NOT NULL, -- JSON array
|
as_path TEXT NOT NULL, -- JSON array
|
||||||
next_hop TEXT NOT NULL,
|
next_hop TEXT NOT NULL,
|
||||||
last_updated DATETIME 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
|
-- 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
|
as_path TEXT NOT NULL, -- JSON array
|
||||||
next_hop TEXT NOT NULL,
|
next_hop TEXT NOT NULL,
|
||||||
last_updated DATETIME 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
|
-- Indexes for live_routes_v4 table
|
||||||
|
|||||||
+15
-10
@@ -60,17 +60,22 @@ func New() *Tracker {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// SetConnected updates the connection status
|
// SetConnected updates the connection status. Every connection after the
|
||||||
|
// first counts one reconnect.
|
||||||
func (t *Tracker) SetConnected(connected bool) {
|
func (t *Tracker) SetConnected(connected bool) {
|
||||||
wasConnected := t.isConnected.Swap(connected)
|
t.isConnected.Store(connected)
|
||||||
if connected {
|
if !connected {
|
||||||
t.mu.Lock()
|
return
|
||||||
t.connectedSince = time.Now()
|
}
|
||||||
t.mu.Unlock()
|
|
||||||
// Increment reconnect count (but not for the initial connection)
|
t.mu.Lock()
|
||||||
if wasConnected || t.reconnectCount.Load() > 0 {
|
// connectedSince is zero until the first connection.
|
||||||
t.reconnectCount.Add(1)
|
connectedBefore := !t.connectedSince.IsZero()
|
||||||
}
|
t.connectedSince = time.Now()
|
||||||
|
t.mu.Unlock()
|
||||||
|
|
||||||
|
if connectedBefore {
|
||||||
|
t.reconnectCount.Add(1)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,21 @@
|
|||||||
|
package metrics
|
||||||
|
|
||||||
|
import "testing"
|
||||||
|
|
||||||
|
// TestReconnectCount connects three times with a disconnect between each, the
|
||||||
|
// way the streamer does, and expects every connection after the first to count
|
||||||
|
// one reconnect.
|
||||||
|
func TestReconnectCount(t *testing.T) {
|
||||||
|
tracker := New()
|
||||||
|
|
||||||
|
tracker.SetConnected(true)
|
||||||
|
tracker.SetConnected(false)
|
||||||
|
tracker.SetConnected(true)
|
||||||
|
tracker.SetConnected(false)
|
||||||
|
tracker.SetConnected(true)
|
||||||
|
|
||||||
|
const wantReconnects = 2
|
||||||
|
if got := tracker.GetReconnectCount(); got != wantReconnects {
|
||||||
|
t.Fatalf("reconnect count = %d, want %d", got, wantReconnects)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -205,7 +205,7 @@ func (m *mockStore) UpsertLiveRoute(route *database.LiveRoute) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// DeleteLiveRoute mock implementation
|
// 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
|
// Simple mock - just return nil
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -20,6 +20,8 @@ const (
|
|||||||
debugInterval = 60 * time.Second
|
debugInterval = 60 * time.Second
|
||||||
// bytesPerMB is bytes per megabyte
|
// bytesPerMB is bytes per megabyte
|
||||||
bytesPerMB = 1024 * 1024
|
bytesPerMB = 1024 * 1024
|
||||||
|
// failureExitCode is the exit status when RouteWatch stops on an error
|
||||||
|
failureExitCode = 1
|
||||||
)
|
)
|
||||||
|
|
||||||
// logDebugStats logs goroutine count and memory usage
|
// logDebugStats logs goroutine count and memory usage
|
||||||
@@ -83,6 +85,9 @@ func CLIEntry() {
|
|||||||
go func() {
|
go func() {
|
||||||
if err := rw.Run(ctx); err != nil {
|
if err := rw.Run(ctx); err != nil {
|
||||||
logger.Error("RouteWatch error", "error", err)
|
logger.Error("RouteWatch error", "error", err)
|
||||||
|
if err := shutdowner.Shutdown(fx.ExitCode(failureExitCode)); err != nil {
|
||||||
|
logger.Error("Failed to shutdown gracefully", "error", err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
|
|||||||
@@ -105,7 +105,24 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
|
|||||||
h.mu.Lock()
|
h.mu.Lock()
|
||||||
defer h.mu.Unlock()
|
defer h.mu.Unlock()
|
||||||
|
|
||||||
// Process announcements
|
// 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),
|
||||||
|
peer: msg.Peer,
|
||||||
|
messageType: "withdrawal",
|
||||||
|
timestamp: timestamp,
|
||||||
|
})
|
||||||
|
// Record withdrawal in metrics
|
||||||
|
if h.metrics != nil {
|
||||||
|
h.metrics.RecordWithdrawal()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
for _, announcement := range msg.Announcements {
|
for _, announcement := range msg.Announcements {
|
||||||
for _, prefix := range announcement.Prefixes {
|
for _, prefix := range announcement.Prefixes {
|
||||||
h.batch = append(h.batch, prefixUpdate{
|
h.batch = append(h.batch, prefixUpdate{
|
||||||
@@ -123,22 +140,6 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Process withdrawals
|
|
||||||
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 {
|
|
||||||
h.metrics.RecordWithdrawal()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// Check if we need to flush
|
// Check if we need to flush
|
||||||
if len(h.batch) >= prefixBatchSize {
|
if len(h.batch) >= prefixBatchSize {
|
||||||
h.flushBatchLocked()
|
h.flushBatchLocked()
|
||||||
@@ -179,14 +180,14 @@ func (h *PrefixHandler) flushBatchLocked() {
|
|||||||
startTime := time.Now()
|
startTime := time.Now()
|
||||||
batchSize := len(h.batch)
|
batchSize := len(h.batch)
|
||||||
|
|
||||||
// Group updates by prefix to deduplicate
|
// For each prefix and peer, keep only the last update the feed delivered.
|
||||||
// For each prefix, keep the latest update
|
// Updates from different peers for the same prefix are all kept. Arrival
|
||||||
prefixMap := make(map[string]prefixUpdate)
|
// order decides, not the timestamp: timestamps are whole seconds, so
|
||||||
|
// updates within one second carry the same one.
|
||||||
|
type prefixAndPeer struct{ prefix, peer string }
|
||||||
|
lastUpdates := make(map[prefixAndPeer]prefixUpdate)
|
||||||
for _, update := range h.batch {
|
for _, update := range h.batch {
|
||||||
key := update.prefix
|
lastUpdates[prefixAndPeer{update.prefix, update.peer}] = update
|
||||||
if existing, ok := prefixMap[key]; !ok || update.timestamp.After(existing.timestamp) {
|
|
||||||
prefixMap[key] = update
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Collect routes to upsert and delete
|
// Collect routes to upsert and delete
|
||||||
@@ -196,7 +197,7 @@ func (h *PrefixHandler) flushBatchLocked() {
|
|||||||
// Collect unique prefixes to update
|
// Collect unique prefixes to update
|
||||||
prefixesToUpdate := make(map[string]time.Time)
|
prefixesToUpdate := make(map[string]time.Time)
|
||||||
|
|
||||||
for _, update := range prefixMap {
|
for _, update := range lastUpdates {
|
||||||
// Track prefix for both announcements and withdrawals
|
// Track prefix for both announcements and withdrawals
|
||||||
if _, exists := prefixesToUpdate[update.prefix]; !exists || update.timestamp.After(prefixesToUpdate[update.prefix]) {
|
if _, exists := prefixesToUpdate[update.prefix]; !exists || update.timestamp.After(prefixesToUpdate[update.prefix]) {
|
||||||
prefixesToUpdate[update.prefix] = update.timestamp
|
prefixesToUpdate[update.prefix] = update.timestamp
|
||||||
@@ -220,7 +221,6 @@ func (h *PrefixHandler) flushBatchLocked() {
|
|||||||
// Create deletion record for batch delete
|
// Create deletion record for batch delete
|
||||||
routesToDelete = append(routesToDelete, database.LiveRouteDeletion{
|
routesToDelete = append(routesToDelete, database.LiveRouteDeletion{
|
||||||
Prefix: update.prefix,
|
Prefix: update.prefix,
|
||||||
OriginASN: update.originASN,
|
|
||||||
PeerIP: update.peer,
|
PeerIP: update.peer,
|
||||||
IPVersion: ipVersion,
|
IPVersion: ipVersion,
|
||||||
})
|
})
|
||||||
@@ -255,7 +255,7 @@ func (h *PrefixHandler) flushBatchLocked() {
|
|||||||
elapsed := time.Since(startTime)
|
elapsed := time.Since(startTime)
|
||||||
h.logger.Debug("Flushed prefix batch",
|
h.logger.Debug("Flushed prefix batch",
|
||||||
"batch_size", batchSize,
|
"batch_size", batchSize,
|
||||||
"unique_prefixes", len(prefixMap),
|
"unique_prefixes", len(prefixesToUpdate),
|
||||||
"success", successCount,
|
"success", successCount,
|
||||||
"duration_ms", elapsed.Milliseconds(),
|
"duration_ms", elapsed.Milliseconds(),
|
||||||
)
|
)
|
||||||
@@ -424,51 +424,24 @@ func (h *PrefixHandler) processAnnouncementDirect(update prefixUpdate) {
|
|||||||
// processWithdrawalDirect handles removing a route directly without prefix table
|
// processWithdrawalDirect handles removing a route directly without prefix table
|
||||||
// nolint:unused // kept for potential future use
|
// nolint:unused // kept for potential future use
|
||||||
func (h *PrefixHandler) processWithdrawalDirect(update prefixUpdate) {
|
func (h *PrefixHandler) processWithdrawalDirect(update prefixUpdate) {
|
||||||
// For withdrawals, we need to delete the route from live_routes
|
if err := h.db.DeleteLiveRoute(update.prefix, update.peer); err != nil {
|
||||||
if update.originASN > 0 {
|
h.logger.Error("Failed to delete live route",
|
||||||
if err := h.db.DeleteLiveRoute(update.prefix, update.originASN, update.peer); err != nil {
|
"prefix", update.prefix,
|
||||||
h.logger.Error("Failed to delete live route",
|
"peer", update.peer,
|
||||||
"prefix", update.prefix,
|
"error", err,
|
||||||
"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
|
// processWithdrawal handles removing a route from the live routing table
|
||||||
// nolint:unused // kept for potential future use
|
// nolint:unused // kept for potential future use
|
||||||
func (h *PrefixHandler) processWithdrawal(_ *database.Prefix, update prefixUpdate) {
|
func (h *PrefixHandler) processWithdrawal(_ *database.Prefix, update prefixUpdate) {
|
||||||
// For withdrawals, we need to delete the route from live_routes
|
if err := h.db.DeleteLiveRoute(update.prefix, update.peer); err != nil {
|
||||||
// Since we have the origin ASN from the update, we can delete the specific route
|
h.logger.Error("Failed to delete live route",
|
||||||
if update.originASN > 0 {
|
"prefix", update.prefix,
|
||||||
if err := h.db.DeleteLiveRoute(update.prefix, update.originASN, update.peer); err != nil {
|
"peer", update.peer,
|
||||||
h.logger.Error("Failed to delete live route",
|
"error", err,
|
||||||
"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,
|
|
||||||
)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -1,7 +1,10 @@
|
|||||||
package routewatch
|
package routewatch
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"slices"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -72,3 +75,263 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestPrefixHandlerAppliesEachPeersLastUpdateInABatch runs updates for one
|
||||||
|
// prefix through the prefix handler into a real database. All updates carry the
|
||||||
|
// same timestamp, as updates within one second of the feed do. Within a batch,
|
||||||
|
// the last update each peer sent for the prefix decides whether that peer has a
|
||||||
|
// live route for it, and updates from different peers are all applied. The
|
||||||
|
// in-memory route count and prefix distribution must match the route table.
|
||||||
|
func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) {
|
||||||
|
const (
|
||||||
|
prefix = "198.51.100.0/24"
|
||||||
|
prefixMaskLength = 24
|
||||||
|
otherPeerIP = "2001:db8:ffff::2"
|
||||||
|
)
|
||||||
|
|
||||||
|
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
announce := func(peer string) *ristypes.RISMessage {
|
||||||
|
return &ristypes.RISMessage{
|
||||||
|
Peer: peer,
|
||||||
|
Path: ristypes.ASPath{testASNA, testASNB},
|
||||||
|
ParsedTimestamp: ts,
|
||||||
|
Announcements: []ristypes.RISAnnouncement{{NextHop: peer, Prefixes: []string{prefix}}},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
withdraw := func(peer string) *ristypes.RISMessage {
|
||||||
|
return &ristypes.RISMessage{Peer: peer, ParsedTimestamp: ts, Withdrawals: []string{prefix}}
|
||||||
|
}
|
||||||
|
announceAndWithdraw := func(peer string) *ristypes.RISMessage {
|
||||||
|
msg := announce(peer)
|
||||||
|
msg.Withdrawals = []string{prefix}
|
||||||
|
|
||||||
|
return msg
|
||||||
|
}
|
||||||
|
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
// Each inner list is handled and then flushed as one batch.
|
||||||
|
batches [][]*ristypes.RISMessage
|
||||||
|
// The peers left with a live route for the prefix, by peer IP.
|
||||||
|
wantPeers []string
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
name: "two peers announce the prefix",
|
||||||
|
batches: [][]*ristypes.RISMessage{{announce(testPeerIP), announce(otherPeerIP)}},
|
||||||
|
wantPeers: []string{testPeerIP, otherPeerIP},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "two peers withdraw the prefix",
|
||||||
|
batches: [][]*ristypes.RISMessage{
|
||||||
|
{announce(testPeerIP)},
|
||||||
|
{announce(otherPeerIP)},
|
||||||
|
{withdraw(testPeerIP), withdraw(otherPeerIP)},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "a peer announces and then withdraws the prefix",
|
||||||
|
batches: [][]*ristypes.RISMessage{{announce(testPeerIP), withdraw(testPeerIP)}},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "a peer withdraws and then re-announces the prefix",
|
||||||
|
batches: [][]*ristypes.RISMessage{
|
||||||
|
{announce(testPeerIP)},
|
||||||
|
{withdraw(testPeerIP), announce(testPeerIP)},
|
||||||
|
},
|
||||||
|
wantPeers: []string{testPeerIP},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
// RFC 4271 section 4.3: the announcement counts, not the withdrawal.
|
||||||
|
name: "one message announces and withdraws the prefix",
|
||||||
|
batches: [][]*ristypes.RISMessage{
|
||||||
|
{announce(testPeerIP)},
|
||||||
|
{announceAndWithdraw(testPeerIP)},
|
||||||
|
},
|
||||||
|
wantPeers: []string{testPeerIP},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
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 _, batch := range tt.batches {
|
||||||
|
for _, msg := range batch {
|
||||||
|
h.HandleMessage(msg)
|
||||||
|
}
|
||||||
|
h.mu.Lock()
|
||||||
|
h.flushBatchLocked()
|
||||||
|
h.mu.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
ctx := context.Background()
|
||||||
|
routes, err := db.GetPrefixDetailsContext(ctx, prefix)
|
||||||
|
if err != nil && !errors.Is(err, database.ErrNoRoute) {
|
||||||
|
t.Fatalf("GetPrefixDetailsContext: %v", err)
|
||||||
|
}
|
||||||
|
var gotPeers []string
|
||||||
|
for _, route := range routes {
|
||||||
|
gotPeers = append(gotPeers, route.PeerIP)
|
||||||
|
}
|
||||||
|
if !slices.Equal(gotPeers, tt.wantPeers) {
|
||||||
|
t.Errorf("peers with a live route for %s = %v, want %v", prefix, gotPeers, 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)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 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)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"net"
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
"net/netip"
|
||||||
"runtime"
|
"runtime"
|
||||||
"sort"
|
"sort"
|
||||||
"strconv"
|
"strconv"
|
||||||
@@ -748,8 +749,15 @@ func (s *Server) handlePrefixDetailJSON() http.HandlerFunc {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Combine prefix and length into CIDR notation
|
// Look the prefix up in the text form net/netip prints, the form the
|
||||||
prefix := prefixParam + "/" + lenParam
|
// tables store, so that any valid way of writing it finds the route.
|
||||||
|
parsed, err := netip.ParsePrefix(prefixParam + "/" + lenParam)
|
||||||
|
if err != nil {
|
||||||
|
writeJSONError(w, http.StatusBadRequest, err.Error())
|
||||||
|
|
||||||
|
return
|
||||||
|
}
|
||||||
|
prefix := parsed.Masked().String()
|
||||||
|
|
||||||
routes, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
|
routes, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -912,8 +920,15 @@ func (s *Server) handlePrefixDetail() http.HandlerFunc {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Combine prefix and length into CIDR notation
|
// Look the prefix up in the text form net/netip prints, the form the
|
||||||
prefix := prefixParam + "/" + lenParam
|
// tables store, so that any valid way of writing it finds the route.
|
||||||
|
parsed, err := netip.ParsePrefix(prefixParam + "/" + lenParam)
|
||||||
|
if err != nil {
|
||||||
|
http.Error(w, "Invalid prefix", http.StatusBadRequest)
|
||||||
|
|
||||||
|
return
|
||||||
|
}
|
||||||
|
prefix := parsed.Masked().String()
|
||||||
|
|
||||||
routes, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
|
routes, err := s.db.GetPrefixDetailsContext(r.Context(), prefix)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -148,6 +148,58 @@ func TestStatsHandlersAnswerFromMemory(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestPrefixDetailHandlersParsePrefix checks that both prefix pages find a
|
||||||
|
// stored route whatever valid text form the URL writes its prefix in, and
|
||||||
|
// answer 400 for a prefix that does not parse. The tables store the form
|
||||||
|
// net/netip prints. Before the fix the URL text was looked up as typed, so
|
||||||
|
// "2001:DB8:0:0::/32" was not found although "2001:db8::/32" was stored, and
|
||||||
|
// "foo/24" answered 500 (https://git.eeqj.de/sneak/routewatch/issues/71).
|
||||||
|
func TestPrefixDetailHandlersParsePrefix(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() })
|
||||||
|
|
||||||
|
ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)
|
||||||
|
if err := db.UpsertLiveRouteBatch([]*database.LiveRoute{
|
||||||
|
{
|
||||||
|
ID: uuid.New(), Prefix: "192.0.2.0/24", MaskLength: 24, IPVersion: 4,
|
||||||
|
OriginASN: 64500, PeerIP: "198.51.100.1", ASPath: []int{64500}, NextHop: "198.51.100.1",
|
||||||
|
LastUpdated: ts,
|
||||||
|
},
|
||||||
|
{
|
||||||
|
ID: uuid.New(), Prefix: "2001:db8::/32", MaskLength: 32, IPVersion: 6,
|
||||||
|
OriginASN: 64501, PeerIP: "2001:db8::1", ASPath: []int{64501}, NextHop: "2001:db8::1",
|
||||||
|
LastUpdated: ts,
|
||||||
|
},
|
||||||
|
}); err != nil {
|
||||||
|
t.Fatalf("UpsertLiveRouteBatch: %v", err)
|
||||||
|
}
|
||||||
|
|
||||||
|
s := New(db, streamer.New(logger.New(), metrics.New()), logger.New(), &config.Config{})
|
||||||
|
|
||||||
|
tests := []struct {
|
||||||
|
path string
|
||||||
|
want int
|
||||||
|
}{
|
||||||
|
{"/api/v1/prefix/foo/24", http.StatusBadRequest},
|
||||||
|
{"/prefix/foo/24", http.StatusBadRequest},
|
||||||
|
{"/api/v1/prefix/2001:DB8:0:0::/32", http.StatusOK},
|
||||||
|
{"/prefix/2001:DB8:0:0::/32", http.StatusOK},
|
||||||
|
{"/api/v1/prefix/192.0.2.1/24", http.StatusOK},
|
||||||
|
{"/prefix/192.0.2.1/24", http.StatusOK},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
s.router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, tt.path, nil))
|
||||||
|
if rec.Code != tt.want {
|
||||||
|
t.Errorf("%s: status %d, want %d; body %s", tt.path, rec.Code, tt.want, rec.Body)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// settledGoroutineCount lets transient goroutines finish, then reports the
|
// settledGoroutineCount lets transient goroutines finish, then reports the
|
||||||
// current count.
|
// current count.
|
||||||
func settledGoroutineCount() int {
|
func settledGoroutineCount() int {
|
||||||
|
|||||||
@@ -3,6 +3,8 @@ package server
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
|
"fmt"
|
||||||
|
"net"
|
||||||
"net/http"
|
"net/http"
|
||||||
"strconv"
|
"strconv"
|
||||||
"time"
|
"time"
|
||||||
@@ -53,9 +55,12 @@ func New(db database.Store, streamer *streamer.Streamer, logger *logger.Logger,
|
|||||||
return s
|
return s
|
||||||
}
|
}
|
||||||
|
|
||||||
// Start starts the HTTP server
|
// 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.
|
||||||
func (s *Server) Start() error {
|
func (s *Server) Start() error {
|
||||||
const (
|
const (
|
||||||
|
network = "tcp"
|
||||||
readHeaderTimeout = 40 * time.Second
|
readHeaderTimeout = 40 * time.Second
|
||||||
readTimeout = 60 * time.Second
|
readTimeout = 60 * time.Second
|
||||||
writeTimeout = 60 * time.Second
|
writeTimeout = 60 * time.Second
|
||||||
@@ -73,10 +78,15 @@ func (s *Server) Start() error {
|
|||||||
|
|
||||||
s.logger.Info("Starting HTTP server", "port", s.port, "addr", s.srv.Addr)
|
s.logger.Info("Starting HTTP server", "port", s.port, "addr", s.srv.Addr)
|
||||||
|
|
||||||
// Start in goroutine but log when actually listening
|
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())
|
||||||
|
|
||||||
go func() {
|
go func() {
|
||||||
s.logger.Info("HTTP server listening", "addr", s.srv.Addr)
|
if err := s.srv.Serve(listener); err != nil && err != http.ErrServerClosed {
|
||||||
if err := s.srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
|
|
||||||
s.logger.Error("HTTP server error", "error", err)
|
s.logger.Error("HTTP server error", "error", err)
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|||||||
@@ -0,0 +1,33 @@
|
|||||||
|
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)
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user