From 77a708d81967b652eb3c7f8f5245e7aa4575218c Mon Sep 17 00:00:00 2001 From: clawbot <35+clawbot@noreply.example.org> Date: Wed, 14 Oct 2026 17:09:13 +0200 Subject: [PATCH] Apply each peer's last update for a prefix in a batch (closes #54) The prefix handler kept one update per prefix in each batch across all peers, chosen by a timestamp cut to whole seconds, so most peers' announcements and withdrawals were lost and the first update in a second won. A batch now keeps the last update per prefix and peer in feed order. A message's withdrawals are queued before its announcements, so a prefix both withdrawn and announced in one message keeps its route (RFC 4271 section 4.3). Tests run each case through the handler into a real database and check the in-memory counts. Model: opus-5-5 --- internal/routewatch/prefixhandler.go | 53 +++++---- internal/routewatch/prefixhandler_test.go | 139 ++++++++++++++++++++++ 2 files changed, 166 insertions(+), 26 deletions(-) diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index c8f6974..59f3991 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -105,7 +105,24 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { h.mu.Lock() 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). + 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{ @@ -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 if len(h.batch) >= prefixBatchSize { h.flushBatchLocked() @@ -179,14 +180,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 +197,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 +256,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..0d30e77 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,140 @@ 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) + } + + 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) + } + }) + } +}