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

A path that ends in an AS set of several members, such as
[64496, 64497, [64498, 64499]], now has no origin: its route is stored
with origin 0 (none), and no member of the set is in a peering. A set of
one member counts as that AS.

ASPath keeps such a set apart from the sequence and gives the origin.
The prefix and AS handlers take the origin from it; the peering handler
uses the sequence only. The stored AS path still lists the set's members
at the end.

Model: opus-5-5

internal/ristypes/ris.go

@@ -23,26 +23,43 @@ const (
 	StateDown = "down"
 )
 
-// ASPath represents a BGP AS path as a slice of AS numbers.
+// ASPath is a BGP AS path as the feed writes it: an array of AS numbers in
-// It handles JSON unmarshaling of both simple arrays and nested AS sets,
+// which an AS set is an array of its members, such as [1, 2, [3, 4]].
-// flattening any nested structures into a single sequence of AS numbers.
+type ASPath struct {
-type ASPath []int
+	// 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
+	// Most paths have no AS set and decode as a plain array of integers.
-	var simple []int
+	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
+	path := ASPath{Sequence: make([]int, 0, len(raw))}
-	result := make([]int, 0)
 
-	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
+			set := make([]int, 0, len(v))
-			for _, nested := range v {
+
-				if num, ok := nested.(float64); ok {
+			for _, member := range v {
-					result = append(result, int(num))
+				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

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)
+			}
+		})
+	}
+}

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
+// origin ASN and all transit ASNs are recorded with their associated
-// with their associated timestamps. When the batch reaches the configured size
+// timestamps. When the batch reaches the configured size threshold, the flush
-// threshold, the flush goroutine is woken to write it.
+// 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)
+	originASN := msg.Path.Origin()
-	var originASN int
-	if len(msg.Path) > 0 {
-		originASN = msg.Path[len(msg.Path)-1]
-	}
 
 	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
+	// Also track all ASNs in the path, the members of an AS set included
-	for _, asn := range msg.Path {
+	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)
+	fill := ristypes.ASPath{Sequence: slices.Repeat([]int{testASNA}, asnBatchSize)}
-	toMax := slices.Repeat(ristypes.ASPath{testASNA}, asnBatchMaxSize)
+	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

@@ -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)
+	originASN := msg.Path.Origin()
-	var originASN int
+
-	if len(msg.Path) > 0 {
+	// The stored and shown AS path lists the members of an AS set that ends
-		originASN = msg.Path[len(msg.Path)-1]
+	// 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 {

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{Sequence: []int{testASNA}}, prefix)},
-					{announce(ristypes.ASPath{testASNA, testASNB}, 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},