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

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

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

Model: opus-5-5

internal/database/database_test.go

@@ -13,6 +13,7 @@ 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
@@ -205,6 +206,40 @@ 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

internal/routewatch/cli.go

@@ -20,6 +20,8 @@ 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
@@ -83,6 +85,9 @@ 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)
+							}
 						}
 					}()
 

internal/routewatch/prefixhandler_test.go

@@ -3,6 +3,7 @@ package routewatch
 import (
 	"context"
 	"errors"
+	"fmt"
 	"slices"
 	"testing"
 	"time"
@@ -186,24 +187,28 @@ func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) {
 			if len(tt.wantPeers) > 0 {
 				wantDistribution = []database.PrefixDistribution{{MaskLength: prefixMaskLength, Count: 1}}
 			}
-			assertIPv4Counts(t, db, len(tt.wantPeers), wantDistribution)
+			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. 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.
+// 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) {
-	const (
-		prefix      = "198.51.100.0/24"
-		otherPrefix = "203.0.113.0/24"
+	families := []struct {
+		ipVersion           int
+		prefix, otherPrefix string
 		// The mask length of both prefixes.
-		prefixMaskLength = 24
-	)
+		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 {
@@ -214,83 +219,88 @@ func TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer(t *testing.T) {
 			Announcements:   []ristypes.RISAnnouncement{{NextHop: testPeerIP, Prefixes: []string{announced}}},
 		}
 	}
-	// One message that announces otherPrefix with origin AS testASNB and
-	// withdraws prefix.
-	announceOtherAndWithdraw := announce(ristypes.ASPath{testASNA, testASNB}, otherPrefix)
-	announceOtherAndWithdraw.Withdrawals = []string{prefix}
 
-	tests := []struct {
-		name     string
-		messages []*ristypes.RISMessage
-		// The origin AS of the peer's live route for each prefix. A prefix
-		// left out must have no live route.
-		wantOrigins map[string]int
-	}{
-		{
-			name: "the peer announces the prefix with a new origin AS",
-			messages: []*ristypes.RISMessage{
-				announce(ristypes.ASPath{testASNA}, prefix),
-				announce(ristypes.ASPath{testASNA, testASNB}, prefix),
+	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},
 			},
-			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,
+			{
+				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},
 			},
-			wantOrigins: map[string]int{otherPrefix: testASNB},
-		},
-	}
+		}
 
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
-			if err != nil {
-				t.Fatalf("failed to create database: %v", err)
-			}
-			defer func() { _ = db.Close() }()
+		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()
-			}
+				// 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)
+				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)
+					}
 				}
-				var gotOrigins []int
-				for _, route := range routes {
-					gotOrigins = append(gotOrigins, route.OriginASN)
-				}
-				var wantOrigins []int
-				if origin, ok := tt.wantOrigins[p]; ok {
-					wantOrigins = []int{origin}
-				}
-				if !slices.Equal(gotOrigins, wantOrigins) {
-					t.Errorf("origin ASes of the live routes for %s = %v, want %v", p, gotOrigins, wantOrigins)
-				}
-			}
 
-			wantDistribution := []database.PrefixDistribution{
-				{MaskLength: prefixMaskLength, Count: len(tt.wantOrigins)},
-			}
-			assertIPv4Counts(t, db, len(tt.wantOrigins), wantDistribution)
-		})
+				wantDistribution := []database.PrefixDistribution{
+					{MaskLength: family.maskLength, Count: len(tt.wantOrigins)},
+				}
+				assertCounts(t, db, family.ipVersion, len(tt.wantOrigins), wantDistribution)
+			})
+		}
 	}
 }
 
-// assertIPv4Counts checks that the IPv4 live route count and prefix
-// distribution are wantRoutes and wantDistribution, both as held in memory and
-// as read from the route table.
-func assertIPv4Counts(t *testing.T, db *database.Database, wantRoutes int,
+// 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()
 
@@ -299,22 +309,29 @@ func assertIPv4Counts(t *testing.T, db *database.Database, wantRoutes int,
 	if err != nil {
 		t.Fatalf("GetStatsContext: %v", err)
 	}
-	tableRoutesV4, _, err := db.GetLiveRouteCountsContext(ctx)
+	tableRoutesV4, tableRoutesV6, err := db.GetLiveRouteCountsContext(ctx)
 	if err != nil {
 		t.Fatalf("GetLiveRouteCountsContext: %v", err)
 	}
-	if stats.IPv4Routes != wantRoutes || tableRoutesV4 != wantRoutes {
-		t.Errorf("IPv4 live routes: %d in memory, %d in the table, want %d",
-			stats.IPv4Routes, tableRoutesV4, wantRoutes)
-	}
-
-	queryDistributionV4, _, err := db.GetPrefixDistributionContext(ctx)
+	queryDistributionV4, queryDistributionV6, 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)
+
+	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)
 	}
 }

internal/server/server.go

@@ -3,6 +3,8 @@ package server
 
 import (
 	"context"
+	"fmt"
+	"net"
 	"net/http"
 	"strconv"
 	"time"
@@ -53,9 +55,12 @@ func New(db database.Store, streamer *streamer.Streamer, logger *logger.Logger,
 	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 {
 	const (
+		network           = "tcp"
 		readHeaderTimeout = 40 * time.Second
 		readTimeout       = 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)
 
-	// 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() {
-		s.logger.Info("HTTP server listening", "addr", s.srv.Addr)
-		if err := s.srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+		if err := s.srv.Serve(listener); err != nil && err != http.ErrServerClosed {
 			s.logger.Error("HTTP server error", "error", err)
 		}
 	}()

internal/server/server_test.go

@@ -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)
+	}
+}