1 Commits
Author SHA1 Message Date
sneak c152ddd5f4 Re-vendor the shared files from sneak/prompts at dd4027b (closes #52)
check / check (push) Canceled after 0s
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
2026-10-14 13:57:50 +00:00
4 changed files with 38 additions and 255 deletions

No files matched your search

+10 -16
View File
@@ -60,23 +60,17 @@ func New() *Tracker {
}
}
// SetConnected updates the connection status. Every connection after the
// first counts one reconnect.
// SetConnected updates the connection status
func (t *Tracker) SetConnected(connected bool) {
t.isConnected.Store(connected)
if !connected {
return
}
t.mu.Lock()
// connectedSince is zero until the first connection.
connectedBefore := !t.connectedSince.IsZero()
t.connectedSince = time.Now()
t.mu.Unlock()
if connectedBefore {
t.reconnectCount.Add(1)
wasConnected := t.isConnected.Swap(connected)
if connected {
t.mu.Lock()
t.connectedSince = time.Now()
t.mu.Unlock()
// Increment reconnect count (but not for the initial connection)
if wasConnected || t.reconnectCount.Load() > 0 {
t.reconnectCount.Add(1)
}
}
}
-27
View File
@@ -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)
}
}
+28 -28
View File
@@ -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
// 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()
}
}
// Process announcements
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.
// Updates from different peers for the same prefix are all kept. Arrival
// 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)
// Group updates by prefix to deduplicate
// For each prefix, keep the latest update
prefixMap := make(map[string]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(),
)
@@ -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)
}
}