Experimental route evidence and MeshMapper scope context

cmd/beacon/main.go

@@ -429,6 +429,7 @@ func main() {
 	// CachedReader. Only response projections receive the geographic annotation.
 	reader = api.WithLocalBorders(reader, localBorders)
 	r := router.New(h, reader, []*ingest.Worker{broker1, broker2}, router.Options{
+		ScopeCatalogues:      scopeImporter,
 		MaxConnsPerIP:        resolved.MaxConnsPerIP,
 		MaxConnectsPerMinute: resolved.MaxConnectsPerMinute,
 		WSAllowedOrigins:     cfg.WebSocket.AllowedOrigins,

db/nodes.go

@@ -127,6 +127,7 @@ func (s *Store) ListNodes(ctx context.Context, nodeType int16, iatas []string, s
 			Longitude:          v.Longitude,
 			IsObserver:         v.IsObserver,
 			ObserverID:         nullableUUID(v.ObserverID),
+			DefaultScope:       v.DefaultScopeName,
 			KnownNeighborCount: v.KnownNeighborCount,
 			NeighborIDs:        v.NeighborIds,
 			Stale:              v.LastSeen.Valid && v.LastSeen.Time.Before(time.Now().Add(-s.staleThreshold)),

db/nodes_test.go

@@ -706,3 +706,15 @@ func TestDeleteOldNodes(t *testing.T) {
 		t.Fatalf("unexpected error: %v", err)
 	}
 }
+
+func TestListNodes_DefaultScopePreservesCaseAndMissingValue(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	mock := mockdb.NewMockQuerier(ctrl)
+	scope := "#YOW"
+	mock.EXPECT().ListNodes(gomock.Any(), gomock.Any()).Return([]sqlc.ListNodesRow{{DefaultScopeName: &scope}, {}}, nil)
+	store := &Store{q: mock}
+	page, err := store.ListNodes(context.Background(), 0, nil, nil, nil, nil, "", "", "", 0, 10, false)
+	if err != nil || len(page.Items) != 2 || page.Items[0].DefaultScope == nil || *page.Items[0].DefaultScope != scope || page.Items[1].DefaultScope != nil {
+		t.Fatalf("scope metadata lost or invented: %+v %v", page, err)
+	}
+}

db/packets.go

@@ -15,6 +15,7 @@ import (
 	sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
 	"github.com/MeshCore-Beacon/beacon-server/internal/api"
 	"github.com/MeshCore-Beacon/beacon-server/internal/ingest"
+	"github.com/MeshCore-Beacon/beacon-server/internal/tracequality"
 	"github.com/google/uuid"
 	"github.com/jackc/pgx/v5/pgtype"
 	"github.com/meshcore-go/meshcore-go"
@@ -406,7 +407,15 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet,
 	// compute it once rather than per observation.
 	var traceRawHashes [][]byte
 	if row.PayloadType == int16(meshcore.PayloadTypeTrace) {
-		if trace, err := meshcore.TraceFromBytes(row.RawPayload); err == nil {
+		quality := tracequality.Assess(row.RawPayload, int(row.PayloadVersion), int(row.RouteType), -1)
+		for _, obs := range obsRows {
+			q := tracequality.Assess(row.RawPayload, int(row.PayloadVersion), int(row.RouteType), int(obs.PathLengthByte))
+			for _, reason := range q.Reasons {
+				quality.Add(reason)
+			}
+		}
+		p.TraceQuality = &quality
+		if trace, err := meshcore.TraceFromBytes(row.RawPayload); err == nil && quality.Status == "supported" {
 			hashSize := int(trace.PathHashSize())
 			for i := 0; i+hashSize <= len(trace.PathHashes); i += hashSize {
 				traceRawHashes = append(traceRawHashes, trace.PathHashes[i:i+hashSize])
@@ -442,8 +451,11 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet,
 		obs.PropagationTimeMs = &prop
 		resolvedPath := []api.ResolvedHop{}
 		if row.PayloadType == int16(meshcore.PayloadTypeTrace) {
+			rawPath := hex.EncodeToString(v.PathBytes)
+			obs.WirePathBytes = &rawPath
+			obs.TraceQuality = p.TraceQuality
 			if len(traceRawHashes) > 0 {
-				resolved, err := s.ResolvePathHashes(ctx, v.Iata, traceRawHashes)
+				resolved, err := s.ResolveTracePathHashes(ctx, v.Iata, traceRawHashes)
 				if err != nil {
 					slog.Error(fmt.Sprintf("store: path resolution failed for observation %d", v.ID), "component", "db", "error", err)
 				} else {
@@ -491,7 +503,7 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet,
 		}
 		p.Observations = append(p.Observations, obs)
 	}
-	if row.PayloadType == 9 && len(obsRows) > 0 {
+	if row.PayloadType == 9 && len(obsRows) > 0 && p.TraceQuality.Status == "supported" {
 		iatas := make([]string, 0, len(obsRows))
 		seen := make(map[string]struct{})
 		for _, v := range obsRows {
@@ -503,6 +515,11 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet,
 		var parsed tracePayload
 		if err := json.Unmarshal(row.ParsedPayload, &parsed); err == nil {
 			p.ResolvedRoute = s.resolveTraceRoute(ctx, &parsed, iatas)
+			for _, hop := range p.ResolvedRoute {
+				if hop.Confidence == "ambiguous" {
+					p.TraceQuality.Add("ambiguous_prefix")
+				}
+			}
 		}
 	}
 	return p, nil

db/queries/queries.sql

@@ -455,6 +455,8 @@ WHERE p.packet_hash = $1;
 -- Return distinct observation IATAs in first-heard order for path resolution,
 -- without fetching full observations separately for every trace packet.
 SELECT encode(p.packet_hash, 'hex') AS packet_hash_hex,
+    p.raw_payload, p.payload_version,
+    EXISTS(SELECT 1 FROM packet_observations po WHERE po.packet_hash=p.packet_hash AND po.path_length_byte>=64) AS invalid_snr_path,
     p.route_type,
     p.first_heard_at,
     p.last_heard_at,
@@ -1422,7 +1424,11 @@ SELECT
      WHERE ti.trace_tag = t.trace_tag
        AND (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR ti.iata = ANY($1::bpchar[]))) AS iata_count,
     COALESCE(t.trace_type, '')::text AS trace_type,
-    t.best_payload
+    t.best_payload,
+    ARRAY(SELECT ti.iata FROM trace_iatas ti WHERE ti.trace_tag=t.trace_tag)::bpchar[] AS iatas,
+    EXISTS(SELECT 1 FROM packets p WHERE p.trace_tag=t.trace_tag AND p.payload_version<>0) AS unsupported_version,
+    EXISTS(SELECT 1 FROM packets p WHERE p.trace_tag=t.trace_tag AND p.route_type NOT IN (2,3)) AS non_direct,
+    EXISTS(SELECT 1 FROM packets p JOIN packet_observations po ON po.packet_hash=p.packet_hash WHERE p.trace_tag=t.trace_tag AND po.path_length_byte>=64) AS invalid_snr_path
 FROM trace_tags t
 WHERE (COALESCE(cardinality($1::bpchar[]), 0) = 0 OR EXISTS (
         SELECT 1 FROM trace_iatas ti WHERE ti.trace_tag = t.trace_tag AND ti.iata = ANY($1::bpchar[])))
@@ -1834,3 +1840,31 @@ ON CONFLICT (country) DO UPDATE SET
 -- name: PruneZoneBoundaries :many
 -- Drops imports for IATAs no longer configured, so their manual border returns.
 DELETE FROM meshmapper_zone_boundaries WHERE NOT (iata = ANY(@keep::text[])) RETURNING iata;
+
+-- name: ResolveTraceHashes :many
+-- TRACE can terminate at companions. Preserve the full prefix (including eight
+-- bytes), and return every candidate across heard regions instead of picking one.
+SELECT h.hash::bytea AS hash, si.iata, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key
+FROM unnest(@hashes::bytea[]) AS h(hash)
+JOIN node_short_ids si ON si.iata = ANY(@iatas::bpchar[]) AND
+ ((octet_length(h.hash)=1 AND si.prefix_1=h.hash) OR
+  (octet_length(h.hash)=2 AND si.prefix_2=h.hash) OR
+  (octet_length(h.hash) IN (4,8) AND si.prefix_4=substring(h.hash FROM 1 FOR 4)))
+JOIN nodes n ON n.id=si.node_id
+WHERE substring(n.public_key FROM 1 FOR octet_length(h.hash))=h.hash
+ORDER BY si.iata, h.hash, n.id;
+
+-- name: GetTopologyLinks :many
+-- Reduce route history to unique adjacent pairs before transferring it to a browser.
+WITH pairs AS (
+ SELECT h.node_id AS a, r.node_ids[(h.ord+1)::int] AS b
+ FROM known_routes r CROSS JOIN LATERAL unnest(r.node_ids) WITH ORDINALITY h(node_id,ord)
+ WHERE r.last_seen >= @since::timestamptz AND r.last_seen < @until::timestamptz
+   AND (COALESCE(cardinality(@iatas::bpchar[]),0)=0 OR r.iata=ANY(@iatas::bpchar[]))
+   AND h.ord < cardinality(r.node_ids)
+)
+SELECT LEAST(a,b)::uuid AS from_id, GREATEST(a,b)::uuid AS to_id
+FROM pairs WHERE a IS NOT NULL AND b IS NOT NULL AND a<>b
+GROUP BY LEAST(a,b), GREATEST(a,b)
+ORDER BY LEAST(a,b), GREATEST(a,b)
+LIMIT @link_limit::int;

db/route_evidence.go

@@ -39,6 +39,16 @@ func (s *Store) GetRouteEvidence(ctx context.Context, iata, key string, query ap
 		if c.IATA != iata || c.PathKey != key || !c.Since.Equal(query.Since) || !c.Until.Equal(query.Until) || c.ID <= 0 || c.HeardAt.Before(query.Since) || !c.HeardAt.Before(query.Until) {
 			return nil, api.ErrRouteEvidenceInput
 		}
+		if c.HashSize != 0 || c.PathBytes != "" {
+			if (query.HashSize != 0 || query.PathBytes != "") && (query.HashSize != c.HashSize || query.PathBytes != c.PathBytes) {
+				return nil, api.ErrRouteEvidenceInput
+			}
+			query.HashSize, query.PathBytes = c.HashSize, c.PathBytes
+		}
+	}
+	pinned := query.HashSize != 0 || query.PathBytes != ""
+	if pinned && !api.ValidRouteEvidencePath(query.HashSize, query.PathBytes) {
+		return nil, api.ErrRouteEvidenceInput
 	}
 	keyBytes, _ := hex.DecodeString(key)
 	row, err := s.q.GetRouteEvidenceRoute(ctx, sqlc.GetRouteEvidenceRouteParams{Iata: iata, PathKey: keyBytes})
@@ -49,10 +59,35 @@ func (s *Store) GetRouteEvidence(ctx context.Context, iata, key string, query ap
 	if err != nil {
 		return nil, err
 	}
+	width, path, valid := savedRoutePath(row.HashPrefix, row.HopCount)
+	if pinned {
+		selected, _ := hex.DecodeString(query.PathBytes)
+		hops := len(selected) / int(query.HashSize)
+		if hops != len(row.NodeIds) || hops != int(row.HopCount) {
+			return nil, api.ErrRouteEvidenceInput
+		}
+		prefixes := make([][]byte, hops)
+		for i, id := range row.NodeIds {
+			prefixes[i] = selected[i*int(query.HashSize) : (i+1)*int(query.HashSize)]
+			// A changed representation must still belong to this node chain.
+			// Reuse the already-loaded keys; never widen the evidence query.
+			if !valid || width != query.HashSize || !bytes.Equal(path, selected) {
+				node := nodes[id]
+				if node == nil {
+					return nil, api.ErrRouteEvidenceInput
+				}
+				pubkey, err := hex.DecodeString(node.PublicKey)
+				if err != nil || !bytes.HasPrefix(pubkey, prefixes[i]) {
+					return nil, api.ErrRouteEvidenceInput
+				}
+			}
+		}
+		row.HashPrefix = prefixes
+		width, path, valid = query.HashSize, selected, true
+	}
 	route := toKnownRoutes([]knownRouteRow{{ID: row.ID, NodeIds: row.NodeIds, HashPrefix: row.HashPrefix, Iata: row.Iata, HopCount: row.HopCount, FirstSeen: row.FirstSeen, LastSeen: row.LastSeen, ObservationCount: row.ObservationCount}}, nodes)[0]
 	route.PathKey = hex.EncodeToString(row.PathKey)
 	out := &api.RouteEvidence{Items: []api.RouteObservation{}, Route: route, WindowStart: query.Since.UnixMilli(), WindowEnd: query.Until.UnixMilli(), GeneratedAt: time.Now().UnixMilli(), MatchType: "saved_path_prefixes"}
-	width, path, valid := savedRoutePath(row.HashPrefix, row.HopCount)
 	if !valid {
 		return out, nil
 	}
@@ -86,7 +121,7 @@ func (s *Store) GetRouteEvidence(ctx context.Context, iata, key string, query ap
 	}
 	if out.HasMore {
 		last := rows[len(rows)-1]
-		cursor := (api.RouteEvidenceCursor{IATA: iata, PathKey: key, Since: query.Since, Until: query.Until, HeardAt: last.HeardAt.Time, ID: last.ID}).String()
+		cursor := (api.RouteEvidenceCursor{IATA: iata, PathKey: key, Since: query.Since, Until: query.Until, HeardAt: last.HeardAt.Time, ID: last.ID, HashSize: width, PathBytes: out.PathBytes}).String()
 		out.NextPageCursor = &cursor
 	}
 	return out, nil

db/route_evidence_integration_test.go

@@ -36,12 +36,13 @@ func TestRouteEvidencePostgres(t *testing.T) {
 	applyBaseline(t, ctx, tx)
 	_, err := tx.Exec(ctx, `
  INSERT INTO observers(id,public_key,display_name) VALUES ('00000000-0000-0000-0000-000000000001','\x01','One'),('00000000-0000-0000-0000-000000000002','\x02','Two');
- INSERT INTO nodes(id,public_key,node_type,name) VALUES ('00000000-0000-0000-0000-000000000001','\xaa',2,'A'),('00000000-0000-0000-0000-000000000002','\xbb',2,'B');
+ INSERT INTO nodes(id,public_key,node_type,name) VALUES ('00000000-0000-0000-0000-000000000001','\xaa0102',2,'A'),('00000000-0000-0000-0000-000000000002','\xbb0304',2,'B');
  INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at)
  SELECT int4send(i),4,0,1,'\x00','\x00',NOW(),NOW() FROM generate_series(1,10) i;`)
 	if err != nil {
 		t.Fatal(err)
 	}
+
 	capture := &evidenceCapture{Tx: tx}
 	store := &Store{q: sqlc.New(capture)}
 	ids := []uuid.UUID{uuid.MustParse("00000000-0000-0000-0000-000000000001"), uuid.MustParse("00000000-0000-0000-0000-000000000002")}
@@ -80,6 +81,36 @@ func TestRouteEvidencePostgres(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
+	if query.Cursor.HashSize != 1 || query.Cursor.PathBytes != "aabb" {
+		t.Fatalf("cursor did not pin the original bytes: %+v", query.Cursor)
+	}
+	// A new representation arrives while page one is open. New reads follow it,
+	// but pagination and copied links retain the exact original representation.
+	if _, err = tx.Exec(ctx, `INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at) SELECT int4send(i),4,0,1,'\x00','\x00',NOW(),NOW() FROM generate_series(12,14) i`); err != nil {
+		t.Fatal(err)
+	}
+	_, err = tx.Exec(ctx, `INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,hash_size,hop_count,path_bytes,payload_type,path_length_byte)
+ SELECT i,int4send(i),'00000000-0000-0000-0000-000000000001','YOW',$1,2,2,'\xaa01bb03',4,66 FROM generate_series(12,14) i`, at)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := store.UpsertKnownRoute(ctx, ids, [][]byte{{0xaa, 1}, {0xbb, 3}}, "YOW", 2); err != nil {
+		t.Fatal(err)
+	}
+	freshQuery := api.RouteEvidenceQuery{Since: since, Until: until, Limit: 2}
+	fresh, err := store.GetRouteEvidence(ctx, "YOW", key, freshQuery)
+	if err != nil || fresh.HashSize != 2 || fresh.PathBytes != "aa01bb03" || len(fresh.Items) != 2 || fresh.Items[0].ID != 14 || fresh.Route.Hops[0].HashBytes != "aa01" {
+		t.Fatalf("new representation lost: %+v %v", fresh, err)
+	}
+	if fresh.Route.ID != first.Route.ID || fresh.Route.PathKey != first.Route.PathKey || fresh.Route.FirstSeen != first.Route.FirstSeen || fresh.Route.ObservationCount != first.Route.ObservationCount+1 {
+		t.Fatalf("route identity/counters changed: %+v", fresh.Route)
+	}
+	shared := freshQuery
+	shared.HashSize, shared.PathBytes = first.HashSize, first.PathBytes
+	copyPage, err := store.GetRouteEvidence(ctx, "YOW", key, shared)
+	if err != nil || len(copyPage.Items) != 2 || copyPage.Items[0].ID != 3 || copyPage.Route.Hops[0].HashBytes != "aa" {
+		t.Fatalf("shared representation changed: %+v %v", copyPage, err)
+	}
 	second, err := store.GetRouteEvidence(ctx, "YOW", key, query)
 	if err != nil {
 		t.Fatal(err)
@@ -90,6 +121,44 @@ func TestRouteEvidencePostgres(t *testing.T) {
 	if _, err = json.Marshal(second); err != nil {
 		t.Fatalf("non-finite JSON: %v", err)
 	}
+	for _, prefixes := range [][][]byte{{{0xaa, 1, 2}, {0xbb, 3, 4}}, {{0xaa, 1, 2}, {0xbb, 3, 4}}} {
+		if err := store.UpsertKnownRoute(ctx, ids, prefixes, "YOW", 2); err != nil {
+			t.Fatal(err)
+		}
+	}
+	thirdWidth, err := store.GetRouteEvidence(ctx, "YOW", key, freshQuery)
+	if err != nil || thirdWidth.HashSize != 3 || thirdWidth.PathBytes != "aa0102bb0304" || len(thirdWidth.Items) != 0 || thirdWidth.Route.ID != first.Route.ID || thirdWidth.Route.ObservationCount != first.Route.ObservationCount+3 {
+		t.Fatalf("third width or repeated upsert: %+v %v", thirdWidth, err)
+	}
+	// Missing identities or mismatched bytes cannot make a pinned query search
+	// a different chain. Prefix collisions likewise never widen the SQL match.
+	wrong := shared
+	wrong.PathBytes = "aacc"
+	if _, err := store.GetRouteEvidence(ctx, "YOW", key, wrong); !errors.Is(err, api.ErrRouteEvidenceInput) {
+		t.Fatalf("unrelated path accepted: %v", err)
+	}
+	wrong.PathBytes = "aabbcc"
+	if _, err := store.GetRouteEvidence(ctx, "YOW", key, wrong); !errors.Is(err, api.ErrRouteEvidenceInput) {
+		t.Fatalf("wrong hop count accepted: %v", err)
+	}
+	_, err = tx.Exec(ctx, `INSERT INTO nodes(id,public_key,node_type,name) VALUES ('00000000-0000-0000-0000-000000000003','\xaa9988',2,'Collision');
+ DELETE FROM nodes WHERE id='00000000-0000-0000-0000-000000000001'`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := store.GetRouteEvidence(ctx, "YOW", key, shared); !errors.Is(err, api.ErrRouteEvidenceInput) {
+		t.Fatalf("missing node substituted by colliding node: %v", err)
+	}
+	if _, err = tx.Exec(ctx, `INSERT INTO nodes(id,public_key,node_type,name) VALUES ('00000000-0000-0000-0000-000000000001','\xaa0102',2,'A')`); err != nil {
+		t.Fatal(err)
+	}
+	copyPage, err = store.GetRouteEvidence(ctx, "YOW", key, shared)
+	if err != nil || len(copyPage.Items) != 2 || copyPage.Items[0].ID != 3 {
+		t.Fatalf("collision broadened evidence: %+v %v", copyPage, err)
+	}
+	if err := store.UpsertKnownRoute(ctx, ids, [][]byte{{0xaa}, {0xbb}}, "YOW", 2); err != nil {
+		t.Fatal(err)
+	}
 	query.Cursor.IATA = "YVR"
 	if _, err = store.GetRouteEvidence(ctx, "YOW", key, query); !errors.Is(err, api.ErrRouteEvidenceInput) {
 		t.Fatal("cross-route cursor accepted")