Re-vendor the shared files from sneak/prompts at dd4027b (closes #52)

Takes .golangci.yml, .dockerignore, .gitignore, .editorconfig, the
workflow and REPO_POLICIES.md from sneak/prompts dd4027b, keeping this
repository's own entries after the shared content. Lint (golangci-lint
v2.14.0) and test become Dockerfile phases that script/lint and
script/test build alone and uncached; the image build depends on both,
and script/cibuild bootstraps, runs script/check, then builds the image.
The agent guidance file moves to AGENTS.md. The Go code is fixed for
what the new configuration finds; reads that iterate rows now also
report an iteration error.

Model: opus-5-5

internal/metrics/metrics.go

@@ -60,23 +60,17 @@ func New() *Tracker {
 	}
 }
 
-// SetConnected updates the connection status. Every connection after the
+// SetConnected updates the connection status
-// first counts one reconnect.
 func (t *Tracker) SetConnected(connected bool) {
-	t.isConnected.Store(connected)
+	wasConnected := t.isConnected.Swap(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)
 	}
 }
 

internal/metrics/metrics_test.go

@@ -1,27 +0,0 @@
-package metrics_test
-
-import (
-	"testing"
-
-	"git.eeqj.de/sneak/routewatch/internal/metrics"
-)
-
-// 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) {
-	t.Parallel()
-
-	tracker := metrics.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)
-	}
-}

internal/routewatch/prefixhandler.go

@@ -106,24 +106,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 	h.mu.Lock()
 	defer h.mu.Unlock()
 
-	// Withdrawals are queued before announcements, because the batch keeps
+	// Process announcements
-	// 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).
-	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()
-		}
-	}
-
 	for _, announcement := range msg.Announcements {
 		for _, prefix := range announcement.Prefixes {
 			h.batch = append(h.batch, prefixUpdate{
@@ -141,6 +124,22 @@ 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
 	if len(h.batch) >= prefixBatchSize {
 		h.flushBatchLocked()
@@ -189,16 +188,17 @@ func (h *PrefixHandler) flushBatchLocked() {
 	startTime := time.Now()
 	batchSize := len(h.batch)
 
-	// For each prefix and peer, keep only the last update the feed delivered.
+	// Group updates by prefix to deduplicate
-	// Updates from different peers for the same prefix are all kept. Arrival
+	// For each prefix, keep the latest update
-	// order decides, not the timestamp: timestamps are whole seconds, so
+	prefixMap := make(map[string]prefixUpdate)
-	// updates within one second carry the same one.
-	type prefixAndPeer struct{ prefix, peer string }
-
-	lastUpdates := make(map[prefixAndPeer]prefixUpdate)
 
 	for _, update := range h.batch {
-		lastUpdates[prefixAndPeer{update.prefix, update.peer}] = update
+		key := update.prefix
+
+		existing, ok := prefixMap[key]
+		if !ok || update.timestamp.After(existing.timestamp) {
+			prefixMap[key] = update
+		}
 	}
 
 	// Collect routes to upsert and delete
@@ -210,7 +210,7 @@ func (h *PrefixHandler) flushBatchLocked() {
 	// Collect unique prefixes to update
 	prefixesToUpdate := make(map[string]time.Time)
 
-	for _, update := range lastUpdates {
+	for _, update := range prefixMap {
 		// Track prefix for both announcements and withdrawals
 		seen, exists := prefixesToUpdate[update.prefix]
 		if !exists || update.timestamp.After(seen) {
@@ -248,7 +248,7 @@ func (h *PrefixHandler) flushBatchLocked() {
 	elapsed := time.Since(startTime)
 	h.logger.Debug("Flushed prefix batch",
 		"batch_size", batchSize,
-		"unique_prefixes", len(prefixesToUpdate),
+		"unique_prefixes", len(prefixMap),
 		"success", successCount,
 		"duration_ms", elapsed.Milliseconds(),
 	)

internal/routewatch/prefixhandler_internal_test.go

@@ -1,9 +1,7 @@
 package routewatch
 
 import (
-	"context"
 	"errors"
-	"slices"
 	"testing"
 	"time"
 
@@ -84,185 +82,3 @@ 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) {
-	t.Parallel()
-
-	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) {
-			t.Parallel()
-
-			db := handleBatches(t, tt.batches)
-			checkPrefixRoutes(t, db, prefix, prefixMaskLength, tt.wantPeers)
-		})
-	}
-}
-
-// handleBatches runs each batch of messages through a prefix handler on a new
-// database, flushing after each batch, and returns the database. The database
-// is closed when the test ends.
-func handleBatches(
-	t *testing.T, batches [][]*ristypes.RISMessage,
-) *database.Database {
-	t.Helper()
-
-	db, err := database.New(&config.Config{StateDir: t.TempDir()}, logger.New())
-	if err != nil {
-		t.Fatalf("failed to create database: %v", err)
-	}
-
-	t.Cleanup(func() { _ = db.Close() })
-
-	// Built without NewPrefixHandler's flush timer, so the test flushes itself.
-	h := &PrefixHandler{db: db, logger: logger.New()}
-
-	for _, batch := range batches {
-		for _, msg := range batch {
-			h.HandleMessage(msg)
-		}
-
-		h.mu.Lock()
-		h.flushBatchLocked()
-		h.mu.Unlock()
-	}
-
-	return db
-}
-
-// checkPrefixRoutes checks that exactly wantPeers have a live route for the
-// IPv4 prefix, and that the in-memory route count and prefix distribution
-// match the route table.
-func checkPrefixRoutes(
-	t *testing.T, db *database.Database, prefix string, maskLength int,
-	wantPeers []string,
-) {
-	t.Helper()
-
-	ctx := context.Background()
-
-	routes, err := db.GetPrefixDetailsContext(ctx, prefix)
-	if err != nil && !errors.Is(err, database.ErrNoRoute) {
-		t.Fatalf("GetPrefixDetailsContext: %v", err)
-	}
-
-	gotPeers := make([]string, 0, len(routes))
-	for _, route := range routes {
-		gotPeers = append(gotPeers, route.PeerIP)
-	}
-
-	if !slices.Equal(gotPeers, wantPeers) {
-		t.Errorf("peers with a live route for %s = %v, want %v",
-			prefix, gotPeers, 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(wantPeers) || tableRoutesV4 != len(wantPeers) {
-		t.Errorf("IPv4 live routes: %d in memory, %d in the table, want %d",
-			stats.IPv4Routes, tableRoutesV4, len(wantPeers))
-	}
-
-	var wantDistribution []database.PrefixDistribution
-	if len(wantPeers) > 0 {
-		wantDistribution = []database.PrefixDistribution{
-			{MaskLength: maskLength, Count: 1},
-		}
-	}
-
-	queryDistributionV4, _, err := db.GetPrefixDistributionContext(ctx)
-	if err != nil {
-		t.Fatalf("GetPrefixDistributionContext: %v", err)
-	}
-
-	if !slices.Equal(stats.IPv4PrefixDistribution, wantDistribution) ||
-		!slices.Equal(queryDistributionV4, wantDistribution) {
-		t.Errorf("IPv4 prefix distribution: %v in memory, %v from the table, want %v",
-			stats.IPv4PrefixDistribution, queryDistributionV4, wantDistribution)
-	}
-}