1 Commits
Author SHA1 Message Date
clawbot 070f2025cc Apply each peer's last update for a prefix in a batch (closes #54)
check / check (push) Canceled after 0s
The prefix handler's batch kept one update per prefix across all peers,
chosen by timestamp. One routing change reaches many peers within a second,
so most peers' announcements and withdrawals in a batch were dropped, and
for updates in the same second the first one won.

The batch now keeps the last update per prefix and peer, in the order the
feed delivered them, so updates from every peer are applied. Tests cover two
peers announcing, two peers withdrawing, and one peer announcing then
withdrawing or withdrawing then re-announcing in the same second, and check
the in-memory route count and prefix distribution against the route table.

Model: opus-5-5
2026-10-14 12:39:37 +00:00
2 changed files with 17 additions and 33 deletions
+17 -18
View File
@@ -105,24 +105,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{
@@ -140,6 +123,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()
-15
View File
@@ -100,12 +100,6 @@ func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) {
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
@@ -139,15 +133,6 @@ func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) {
},
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 {