diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index c8f6974..53eea75 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -179,14 +179,14 @@ func (h *PrefixHandler) flushBatchLocked() { startTime := time.Now() batchSize := len(h.batch) - // Group updates by prefix to deduplicate - // For each prefix, keep the latest update - prefixMap := make(map[string]prefixUpdate) + // 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) for _, update := range h.batch { - key := update.prefix - if existing, ok := prefixMap[key]; !ok || update.timestamp.After(existing.timestamp) { - prefixMap[key] = update - } + lastUpdates[prefixAndPeer{update.prefix, update.peer}] = update } // Collect routes to upsert and delete @@ -196,7 +196,7 @@ func (h *PrefixHandler) flushBatchLocked() { // Collect unique prefixes to update prefixesToUpdate := make(map[string]time.Time) - for _, update := range prefixMap { + for _, update := range lastUpdates { // Track prefix for both announcements and withdrawals if _, exists := prefixesToUpdate[update.prefix]; !exists || update.timestamp.After(prefixesToUpdate[update.prefix]) { prefixesToUpdate[update.prefix] = update.timestamp @@ -255,7 +255,7 @@ func (h *PrefixHandler) flushBatchLocked() { elapsed := time.Since(startTime) h.logger.Debug("Flushed prefix batch", "batch_size", batchSize, - "unique_prefixes", len(prefixMap), + "unique_prefixes", len(prefixesToUpdate), "success", successCount, "duration_ms", elapsed.Milliseconds(), ) diff --git a/internal/routewatch/prefixhandler_test.go b/internal/routewatch/prefixhandler_test.go index 1968f94..af8b76b 100644 --- a/internal/routewatch/prefixhandler_test.go +++ b/internal/routewatch/prefixhandler_test.go @@ -1,7 +1,9 @@ package routewatch import ( + "context" "errors" + "slices" "testing" "time" @@ -72,3 +74,125 @@ 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}} + } + + 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}, + }, + } + + 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) + } + + stats, err := db.GetStatsContext(ctx) + if err != nil { + t.Fatalf("GetStatsContext: %v", err) + } + tableRoutesV4, _, err := db.GetLiveRouteCountsContext(ctx) + if err != nil { + t.Fatalf("GetLiveRouteCountsContext: %v", err) + } + if stats.IPv4Routes != len(tt.wantPeers) || tableRoutesV4 != len(tt.wantPeers) { + t.Errorf("IPv4 live routes: %d in memory, %d in the table, want %d", + stats.IPv4Routes, tableRoutesV4, len(tt.wantPeers)) + } + + var wantDistribution []database.PrefixDistribution + if len(tt.wantPeers) > 0 { + wantDistribution = []database.PrefixDistribution{{MaskLength: prefixMaskLength, Count: 1}} + } + 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) + } + }) + } +}