diff --git a/internal/ristypes/ris.go b/internal/ristypes/ris.go index 4b67ee5..f43ca20 100644 --- a/internal/ristypes/ris.go +++ b/internal/ristypes/ris.go @@ -23,26 +23,43 @@ const ( StateDown = "down" ) -// ASPath represents a BGP AS path as a slice of AS numbers. -// It handles JSON unmarshaling of both simple arrays and nested AS sets, -// flattening any nested structures into a single sequence of AS numbers. -type ASPath []int +// ASPath is a BGP AS path as the feed writes it: an array of AS numbers in +// which an AS set is an array of its members, such as [1, 2, [3, 4]]. +type ASPath struct { + // Sequence is the path's AS numbers in order. A set of one member that + // ends the path counts as that AS and is its last element. The members of + // a set anywhere else are flattened into it. + Sequence []int + + // Set holds the members of a set of several members that ends the path. + // Any of them may be the origin, and they are not neighbours of each + // other or of the AS before them. + Set []int +} + +// Origin returns the AS that originated the route: the last AS of the +// sequence, or 0 (none) when the path ends in a set of several members or is +// empty. +func (p *ASPath) Origin() int { + if len(p.Set) > 0 || len(p.Sequence) == 0 { + return 0 + } + + return p.Sequence[len(p.Sequence)-1] +} // UnmarshalJSON implements the json.Unmarshaler interface for ASPath. -// It handles both simple integer arrays [1, 2, 3] and nested AS sets -// like [1, [2, 3], 4], flattening them into a single slice of integers. func (p *ASPath) UnmarshalJSON(data []byte) error { - // First try to unmarshal as a simple array of integers - var simple []int + // Most paths have no AS set and decode as a plain array of integers. + var sequence []int - err := json.Unmarshal(data, &simple) + err := json.Unmarshal(data, &sequence) if err == nil { - *p = ASPath(simple) + *p = ASPath{Sequence: sequence} return nil } - // If that fails, unmarshal as array of interfaces and flatten var raw []any err = json.Unmarshal(data, &raw) @@ -50,24 +67,30 @@ func (p *ASPath) UnmarshalJSON(data []byte) error { return err } - // Flatten the array - result := make([]int, 0) + path := ASPath{Sequence: make([]int, 0, len(raw))} - for _, item := range raw { + for i, item := range raw { switch v := item.(type) { case float64: - result = append(result, int(v)) + path.Sequence = append(path.Sequence, int(v)) case []any: - // Nested array - flatten it - for _, nested := range v { - if num, ok := nested.(float64); ok { - result = append(result, int(num)) + set := make([]int, 0, len(v)) + + for _, member := range v { + if num, ok := member.(float64); ok { + set = append(set, int(num)) } } + + if i == len(raw)-1 && len(set) > 1 { + path.Set = set + } else { + path.Sequence = append(path.Sequence, set...) + } } } - *p = ASPath(result) + *p = path return nil } @@ -97,7 +120,7 @@ type RISMessage struct { RRC string `json:"rrc,omitempty"` MrtTime float64 `json:"mrt_time,omitempty"` SocketTime float64 `json:"socket_time,omitempty"` - Path ASPath `json:"path,omitempty"` + Path ASPath `json:"path,omitzero"` // Community and Raw are present in the feed but read by no handler. // They are the largest fields on a message that lives in up to four // handler queues, so json:"-" keeps them out of the decoded message diff --git a/internal/ristypes/ris_test.go b/internal/ristypes/ris_test.go index 2b3f7d2..9751099 100644 --- a/internal/ristypes/ris_test.go +++ b/internal/ristypes/ris_test.go @@ -4,6 +4,7 @@ import ( "bufio" "encoding/json" "os" + "slices" "testing" "git.eeqj.de/sneak/routewatch/internal/ristypes" @@ -58,7 +59,7 @@ func TestDecodeDropsCommunityAndRaw(t *testing.T) { t.Errorf("message %d: Raw decoded, want empty", messages) } - if len(msg.Path) > 0 { + if len(msg.Path.Sequence) > 0 { withPath++ } @@ -82,3 +83,65 @@ func TestDecodeDropsCommunityAndRaw(t *testing.T) { withPath, withAnnouncements) } } + +// TestASPathKeepsATrailingSetApart decodes AS paths as the feed writes them, an +// AS set as an array inside the path. A set of several members that ends the +// path is kept apart from the sequence, and the path has no origin, since any +// member may be it. A set of one member counts as that AS, and a set anywhere +// else is flattened into the sequence. +func TestASPathKeepsATrailingSetApart(t *testing.T) { + t.Parallel() + + tests := []struct { + path string + wantSequence []int + wantSet []int + wantOrigin int + }{ + { + path: "[64496, 64497]", + wantSequence: []int{64496, 64497}, + wantOrigin: 64497, + }, + { + path: "[64496, 64497, [64498, 64499]]", + wantSequence: []int{64496, 64497}, + wantSet: []int{64498, 64499}, + }, + { + path: "[64496, 64497, [64498]]", + wantSequence: []int{64496, 64497, 64498}, + wantOrigin: 64498, + }, + { + path: "[64496, [64497, 64498], 64499]", + wantSequence: []int{64496, 64497, 64498, 64499}, + wantOrigin: 64499, + }, + {path: "[]"}, + } + + for _, tt := range tests { + t.Run(tt.path, func(t *testing.T) { + t.Parallel() + + var got ristypes.ASPath + + err := json.Unmarshal([]byte(tt.path), &got) + if err != nil { + t.Fatalf("unmarshal %s: %v", tt.path, err) + } + + if !slices.Equal(got.Sequence, tt.wantSequence) || + !slices.Equal(got.Set, tt.wantSet) { + t.Errorf("%s decoded to sequence %v and set %v, want %v and %v", + tt.path, got.Sequence, got.Set, tt.wantSequence, tt.wantSet) + } + + origin := got.Origin() + if origin != tt.wantOrigin { + t.Errorf("origin of %s = %d, want %d", tt.path, origin, tt.wantOrigin) + } + }) + } +} diff --git a/internal/routewatch/ashandler.go b/internal/routewatch/ashandler.go index 27a003f..1484165 100644 --- a/internal/routewatch/ashandler.go +++ b/internal/routewatch/ashandler.go @@ -1,6 +1,7 @@ package routewatch import ( + "slices" "sync" "time" @@ -98,18 +99,14 @@ func (h *ASHandler) QueueCapacity() int { // HandleMessage processes a RIS Live BGP message by extracting all ASNs from // the AS path and queuing them for batch insertion into the database. The -// origin ASN (last element in the path) and all transit ASNs are recorded -// with their associated timestamps. When the batch reaches the configured size -// threshold, the flush goroutine is woken to write it. +// origin ASN and all transit ASNs are recorded with their associated +// timestamps. When the batch reaches the configured size threshold, the flush +// goroutine is woken to write it. func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) { // Use the pre-parsed timestamp timestamp := msg.ParsedTimestamp - // Get origin ASN from path (last element) - var originASN int - if len(msg.Path) > 0 { - originASN = msg.Path[len(msg.Path)-1] - } + originASN := msg.Path.Origin() h.mu.Lock() defer h.mu.Unlock() @@ -129,8 +126,8 @@ func (h *ASHandler) HandleMessage(msg *ristypes.RISMessage) { }) } - // Also track all ASNs in the path - for _, asn := range msg.Path { + // Also track all ASNs in the path, the members of an AS set included + for _, asn := range slices.Concat(msg.Path.Sequence, msg.Path.Set) { if asn > 0 { h.batch = append(h.batch, asnOp{ number: asn, diff --git a/internal/routewatch/batchwrite_internal_test.go b/internal/routewatch/batchwrite_internal_test.go index d6e6d1d..82ebeb0 100644 --- a/internal/routewatch/batchwrite_internal_test.go +++ b/internal/routewatch/batchwrite_internal_test.go @@ -222,7 +222,7 @@ func handleAtMaxBatchSize( func prefixAnnouncement(prefixes []string) *ristypes.RISMessage { return &ristypes.RISMessage{ Peer: testPeerIP, - Path: ristypes.ASPath{testASNA}, + Path: ristypes.ASPath{Sequence: []int{testASNA}}, Announcements: []ristypes.RISAnnouncement{ {NextHop: testPeerIP, Prefixes: prefixes}, }, @@ -255,10 +255,10 @@ func TestASHandlerHandlesMessagesWhileAWriteIsBlocked(t *testing.T) { store := newBlockingStore() h := NewASHandler(store, logger.New()) - fill := slices.Repeat(ristypes.ASPath{testASNA}, asnBatchSize) + fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)} handleWhileAWriteIsBlocked(t, store, h, []*ristypes.RISMessage{{Path: fill}}, - &ristypes.RISMessage{Path: ristypes.ASPath{testASNB}}) + &ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}}) store.mu.Lock() defer store.mu.Unlock() @@ -317,12 +317,12 @@ func TestASHandlerWaitsAtTheMaximumBatchSize(t *testing.T) { store := newBlockingStore() h := NewASHandler(store, logger.New()) - fill := slices.Repeat(ristypes.ASPath{testASNA}, asnBatchSize) - toMax := slices.Repeat(ristypes.ASPath{testASNA}, asnBatchMaxSize) + fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)} + toMax := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchMaxSize)} handleAtMaxBatchSize(t, store, h, []*ristypes.RISMessage{{Path: fill}}, []*ristypes.RISMessage{{Path: toMax}}, - &ristypes.RISMessage{Path: ristypes.ASPath{testASNB}}) + &ristypes.RISMessage{Path: ristypes.ASPath{Sequence: []int{testASNB}}}) store.mu.Lock() defer store.mu.Unlock() @@ -386,7 +386,7 @@ func TestHandlersWakeTheFlushGoroutineWhenTheBatchIsFull(t *testing.T) { h := &ASHandler{batchFullCh: make(chan struct{}, 1)} // The origin AS and each AS in the path are one update each. h.HandleMessage(&ristypes.RISMessage{ - Path: slices.Repeat(ristypes.ASPath{testASNA}, n-1), + Path: ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, n-1)}, }) return h.batchFullCh diff --git a/internal/routewatch/peeringhandler.go b/internal/routewatch/peeringhandler.go index 25df1d7..65ce19a 100644 --- a/internal/routewatch/peeringhandler.go +++ b/internal/routewatch/peeringhandler.go @@ -87,15 +87,20 @@ func (h *PeeringHandler) QueueCapacity() int { // than minPathLengthForPeering are ignored as they cannot contain valid // peering information. func (h *PeeringHandler) HandleMessage(msg *ristypes.RISMessage) { + // Peerings are between neighbours in the path's sequence only: the + // members of an AS set that ends the path are not neighbours of each + // other or of the AS before them. + sequence := msg.Path.Sequence + // Skip if no AS path or only one AS - if len(msg.Path) < minPathLengthForPeering { + if len(sequence) < minPathLengthForPeering { return } timestamp := msg.ParsedTimestamp // Encode AS path as JSON for use as map key - pathJSON, err := json.Marshal(msg.Path) + pathJSON, err := json.Marshal(sequence) if err != nil { h.logger.Error("Failed to encode AS path", "error", err) diff --git a/internal/routewatch/peeringhandler_internal_test.go b/internal/routewatch/peeringhandler_internal_test.go index de4f5dd..32da7e0 100644 --- a/internal/routewatch/peeringhandler_internal_test.go +++ b/internal/routewatch/peeringhandler_internal_test.go @@ -2,6 +2,9 @@ package routewatch import ( "encoding/json" + "fmt" + "maps" + "slices" "strconv" "sync" "testing" @@ -16,6 +19,7 @@ const ( testASNA = 64500 testASNB = 64501 testASNC = 64502 + testASND = 64503 ) // recordingStore wraps mockStore to count every RecordPeering call, so a @@ -64,7 +68,7 @@ func snapshot(h *PeeringHandler) (int, int) { func pathKey(t *testing.T, asns ...int) string { t.Helper() - b, err := json.Marshal(ristypes.ASPath(asns)) + b, err := json.Marshal(asns) if err != nil { t.Fatalf("failed to marshal path: %v", err) } @@ -74,11 +78,80 @@ func pathKey(t *testing.T, asns ...int) string { func handle(h *PeeringHandler, ts time.Time, asns ...int) { h.HandleMessage(&ristypes.RISMessage{ - Path: ristypes.ASPath(asns), + Path: ristypes.ASPath{Sequence: asns}, ParsedTimestamp: ts, }) } +// decodePath decodes an AS path from the JSON the feed writes for it, in which +// an AS set is an array inside the path: decodePath(t, 1, 2, []int{3, 4}) +// decodes [1, 2, [3, 4]]. +func decodePath(t *testing.T, elements ...any) ristypes.ASPath { + t.Helper() + + data, err := json.Marshal(elements) + if err != nil { + t.Fatalf("failed to marshal path: %v", err) + } + + var path ristypes.ASPath + + err = json.Unmarshal(data, &path) + if err != nil { + t.Fatalf("failed to decode path %s: %v", data, err) + } + + return path +} + +// TestPeeringHandlerLeavesATrailingASSetOutOfThePeerings handles AS paths that +// end in an AS set. The members of a set are not neighbours of each other or of +// the AS before the set, so a set of several members is in no peering; a set of +// one member counts as that AS. +func TestPeeringHandlerLeavesATrailingASSetOutOfThePeerings(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + path []any + // The peerings recorded, keyed as the mock store keys them. + wantPeerings []string + }{ + { + name: "a set of several members", + path: []any{testASNA, testASNB, []int{testASNC, testASND}}, + wantPeerings: []string{fmt.Sprintf("%d_%d", testASNA, testASNB)}, + }, + { + name: "a set of one member", + path: []any{testASNA, testASNB, []int{testASNC}}, + wantPeerings: []string{ + fmt.Sprintf("%d_%d", testASNA, testASNB), + fmt.Sprintf("%d_%d", testASNB, testASNC), + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + store := newMockStore() + h := newTestHandler(store) + h.HandleMessage(&ristypes.RISMessage{ + Path: decodePath(t, tt.path...), + ParsedTimestamp: time.Now().UTC(), + }) + h.ProcessPeeringsNow() + + got := slices.Sorted(maps.Keys(store.Peerings)) + if !slices.Equal(got, tt.wantPeerings) { + t.Errorf("peerings recorded = %v, want %v", got, tt.wantPeerings) + } + }) + } +} + // TestPeeringHandlerProcessesEachRunAndEmpties verifies that a processing run // empties the path map (the swap) and that a path seen again after a run is // recorded in the next run too. diff --git a/internal/routewatch/prefixhandler.go b/internal/routewatch/prefixhandler.go index a79238c..48351d1 100644 --- a/internal/routewatch/prefixhandler.go +++ b/internal/routewatch/prefixhandler.go @@ -127,11 +127,11 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { // Use the pre-parsed timestamp timestamp := msg.ParsedTimestamp - // Get origin ASN from path (last element) - var originASN int - if len(msg.Path) > 0 { - originASN = msg.Path[len(msg.Path)-1] - } + originASN := msg.Path.Origin() + + // The stored and shown AS path lists the members of an AS set that ends + // the path after the rest of it. + path := slices.Concat(msg.Path.Sequence, msg.Path.Set) h.mu.Lock() defer h.mu.Unlock() @@ -169,7 +169,7 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) { peer: msg.Peer, messageType: "announcement", timestamp: timestamp, - path: msg.Path, + path: path, }) // Record announcement in metrics if h.metrics != nil { @@ -297,7 +297,9 @@ func (h *PrefixHandler) flushBatch() { prefixesToUpdate[update.prefix] = update.timestamp } - if update.messageType == "announcement" && update.originASN > 0 { + // An announcement whose path gives no origin is stored with origin 0 + // (none). + if update.messageType == "announcement" { // Create live route for batch upsert route := h.createLiveRoute(update) if route != nil { diff --git a/internal/routewatch/prefixhandler_internal_test.go b/internal/routewatch/prefixhandler_internal_test.go index 82c425f..7797aec 100644 --- a/internal/routewatch/prefixhandler_internal_test.go +++ b/internal/routewatch/prefixhandler_internal_test.go @@ -40,7 +40,7 @@ func TestPrefixHandlerStoresPrefixesTheIPLookupFinds(t *testing.T) { ts := time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC) h.HandleMessage(&ristypes.RISMessage{ Peer: testPeerIP, - Path: ristypes.ASPath{testASNA, testASNB}, + Path: ristypes.ASPath{Sequence: []int{testASNA, testASNB}}, ParsedTimestamp: ts, Announcements: []ristypes.RISAnnouncement{{ NextHop: testPeerIP, @@ -99,7 +99,7 @@ func TestPrefixHandlerAppliesEachPeersLastUpdateInABatch(t *testing.T) { announce := func(peer string) *ristypes.RISMessage { return &ristypes.RISMessage{ Peer: peer, - Path: ristypes.ASPath{testASNA, testASNB}, + Path: ristypes.ASPath{Sequence: []int{testASNA, testASNB}}, ParsedTimestamp: ts, Announcements: []ristypes.RISAnnouncement{ {NextHop: peer, Prefixes: []string{prefix}}, @@ -216,7 +216,7 @@ func TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer(t *testing.T) { // One message that announces otherPrefix with origin AS testASNB and // withdraws prefix. announceOtherAndWithdraw := announce( - ristypes.ASPath{testASNA, testASNB}, otherPrefix) + ristypes.ASPath{Sequence: []int{testASNA, testASNB}}, otherPrefix) announceOtherAndWithdraw.Withdrawals = []string{prefix} tests := []struct { @@ -230,8 +230,8 @@ func TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer(t *testing.T) { { name: "the peer announces the prefix with a new origin AS", batches: [][]*ristypes.RISMessage{ - {announce(ristypes.ASPath{testASNA}, prefix)}, - {announce(ristypes.ASPath{testASNA, testASNB}, prefix)}, + {announce(ristypes.ASPath{Sequence: []int{testASNA}}, prefix)}, + {announce(ristypes.ASPath{Sequence: []int{testASNA, testASNB}}, prefix)}, }, wantOrigins: map[string]int{prefix: testASNB}, }, @@ -239,7 +239,7 @@ func TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer(t *testing.T) { name: "one message announces a prefix with another origin AS " + "and withdraws the prefix", batches: [][]*ristypes.RISMessage{ - {announce(ristypes.ASPath{testASNA}, prefix)}, + {announce(ristypes.ASPath{Sequence: []int{testASNA}}, prefix)}, {announceOtherAndWithdraw}, }, wantOrigins: map[string]int{otherPrefix: testASNB}, @@ -264,6 +264,57 @@ func TestPrefixHandlerKeepsOneRoutePerPrefixAndPeer(t *testing.T) { } } +// TestPrefixHandlerStoresNoOriginForAPathEndingInASet announces a prefix with +// AS paths that end in an AS set through the prefix handler into a real +// database. Any member of a set of several may be the origin, so the route is +// stored with origin 0 (none); a set of one member counts as that AS. The +// stored AS path lists the set's members after the rest of the path. +func TestPrefixHandlerStoresNoOriginForAPathEndingInASet(t *testing.T) { + t.Parallel() + + tests := []struct { + name string + path []any + wantOrigin int + wantPath []int + }{ + { + name: "a set of several members", + path: []any{testASNA, testASNB, []int{testASNC, testASND}}, + wantPath: []int{testASNA, testASNB, testASNC, testASND}, + }, + { + name: "a set of one member", + path: []any{testASNA, testASNB, []int{testASNC}}, + wantOrigin: testASNC, + wantPath: []int{testASNA, testASNB, testASNC}, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + + msg := peerAnnouncement(testPeerIP, testV4Prefix) + msg.Path = decodePath(t, tt.path...) + db := handleBatches(t, [][]*ristypes.RISMessage{{msg}}) + + routes, _, err := db.GetPrefixDetailsContext( + context.Background(), testV4Prefix) + if err != nil { + t.Fatalf("GetPrefixDetailsContext(%s): %v", testV4Prefix, err) + } + + if len(routes) != 1 || routes[0].OriginASN != tt.wantOrigin || + !slices.Equal(routes[0].ASPath, tt.wantPath) { + t.Errorf("live routes for %s = %+v, "+ + "want one with origin AS%d and AS path %v", + testV4Prefix, routes, tt.wantOrigin, tt.wantPath) + } + }) + } +} + // TestStateMessagesGoToThePrefixHandlerNotThePeerHandler checks that STATE // messages go to the prefix handler, which acts on a down, and not to the peer // handler, which would record a peer whose session never comes up as active. @@ -540,7 +591,7 @@ func peerAnnouncement(peer string, prefixes ...string) *ristypes.RISMessage { return &ristypes.RISMessage{ Type: ristypes.MessageTypeUpdate, Peer: peer, - Path: ristypes.ASPath{testASNA, testASNB}, + Path: ristypes.ASPath{Sequence: []int{testASNA, testASNB}}, ParsedTimestamp: time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC), Announcements: []ristypes.RISAnnouncement{ {NextHop: peer, Prefixes: prefixes},