Keep a trailing AS set out of the origin and the peerings (closes #75)

A path ending in an AS set, such as [64496, 64497, [64498, 64499]], was
flattened into the sequence, so the route was stored as originated by the
last set member and the peering handler recorded peerings the path does not
show. ASPath now keeps the sequence and a trailing set of several members
apart; Origin() gives the last AS of the sequence, or 0 (no known origin)
for such a set. The handlers use it; peerings come from the sequence only.
The pages show origin 0 as no known origin, and the IP lookup asks no WHOIS
lookup for it. Empty paths are unchanged.

Model: opus-5-5

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,

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

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)
 

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.

internal/routewatch/prefixhandler.go

@@ -73,6 +73,9 @@ type prefixUpdate struct {
 	messageType string
 	timestamp   time.Time
 	path        []int
+	// endsInSet says the path ends in an AS set of several members, so it
+	// gives no origin.
+	endsInSet bool
 }
 
 // NewPrefixHandler creates a new batched prefix tracking handler
@@ -127,11 +130,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 +172,8 @@ func (h *PrefixHandler) HandleMessage(msg *ristypes.RISMessage) {
 				peer:        msg.Peer,
 				messageType: "announcement",
 				timestamp:   timestamp,
-				path:        msg.Path,
+				path:        path,
+				endsInSet:   len(msg.Path.Set) > 0,
 			})
 			// Record announcement in metrics
 			if h.metrics != nil {
@@ -297,7 +301,11 @@ func (h *PrefixHandler) flushBatch() {
 			prefixesToUpdate[update.prefix] = update.timestamp
 		}
 
-		if update.messageType == "announcement" && update.originASN > 0 {
+		// An announcement is stored when its path gives an origin, or when it
+		// ends in an AS set of several members: that route is stored with
+		// origin 0 (none). An announcement with an empty path is not stored.
+		if update.messageType == "announcement" &&
+			(update.originASN > 0 || update.endsInSet) {
 			// Create live route for batch upsert
 			route := h.createLiveRoute(update)
 			if route != nil {

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,70 @@ 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)
+			}
+		})
+	}
+}
+
+// TestPrefixHandlerDoesNotStoreAnAnnouncementWithAnEmptyPath announces a
+// prefix with an empty AS path, which gives no origin and does not end in an
+// AS set, so no live route is stored.
+func TestPrefixHandlerDoesNotStoreAnAnnouncementWithAnEmptyPath(t *testing.T) {
+	t.Parallel()
+
+	msg := peerAnnouncement(testPeerIP, testV4Prefix)
+	msg.Path = ristypes.ASPath{}
+	db := handleBatches(t, [][]*ristypes.RISMessage{{msg}})
+
+	assertPeers(t, db, map[string][]string{testV4Prefix: nil})
+}
+
 // 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 +604,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},