Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
070f2025cc |
No files matched your search
@@ -105,24 +105,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
|
|||||||
h.mu.Lock()
|
h.mu.Lock()
|
||||||
defer h.mu.Unlock()
|
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 _, announcement := range msg.Announcements {
|
||||||
for _, prefix := range announcement.Prefixes {
|
for _, prefix := range announcement.Prefixes {
|
||||||
h.batch = append(h.batch, prefixUpdate{
|
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
|
// Check if we need to flush
|
||||||
if len(h.batch) >= prefixBatchSize {
|
if len(h.batch) >= prefixBatchSize {
|
||||||
h.flushBatchLocked()
|
h.flushBatchLocked()
|
||||||
|
|||||||
@@ -100,12 +100,6 @@ func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) {
|
|||||||
withdraw := func(peer string) *ristypes.RISMessage {
|
withdraw := func(peer string) *ristypes.RISMessage {
|
||||||
return &ristypes.RISMessage{Peer: peer, ParsedTimestamp: ts, Withdrawals: []string{prefix}}
|
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 {
|
tests := []struct {
|
||||||
name string
|
name string
|
||||||
@@ -139,15 +133,6 @@ func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) {
|
|||||||
},
|
},
|
||||||
wantPeers: []string{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 {
|
for _, tt := range tests {
|
||||||
|
|||||||
Reference in new issue
Block a user