diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 43b9d5c5..beb7284f 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -52,10 +52,10 @@ jobs: BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable run: go test ./cmd/beacon -run '^TestBackupConfigurationPostgres$' -count=1 -v - - name: Verify signal and path aggregates against PostgreSQL 16 + - name: Verify stats and endpoint queries against PostgreSQL 16 env: BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable - run: go test ./db -run '^Test(Signal|Paths)Postgres$' -count=1 -v + run: go test ./db -run '^Test(Signal|Paths|PacketEndpointsResolveLive)Postgres$' -count=1 -v - name: Verify backup command against PostgreSQL 16 run: | diff --git a/db/endpoints.go b/db/endpoints.go new file mode 100644 index 00000000..457ff421 --- /dev/null +++ b/db/endpoints.go @@ -0,0 +1,151 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "context" + "encoding/hex" + "log/slog" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/meshcore-go/meshcore-go" +) + +// packetEndpoints is what a packet's payload says about its sender and recipient. +type packetEndpoints struct { + advert bool // source is the advert's own key, not a hash + advertKey []byte // nil for legacy adverts stored without one + source, destination []byte // one-byte hashes +} + +func parsePacketEndpoints(payloadType int16, raw, originPubkey []byte) packetEndpoints { + var ep packetEndpoints + switch payloadType { + case int16(meshcore.PayloadTypeAdvert): + ep.advert, ep.advertKey = true, originPubkey + case int16(meshcore.PayloadTypeAnonReq): + if p, err := meshcore.AnonReqFromBytes(raw); err == nil { + ep.destination = []byte{p.Destination} + } + case int16(meshcore.PayloadTypeReq): + if p, err := meshcore.RequestFromBytes(raw); err == nil { + ep.source, ep.destination = []byte{p.Source}, []byte{p.Destination} + } + case int16(meshcore.PayloadTypeResponse): + if p, err := meshcore.ResponseFromBytes(raw); err == nil { + ep.source, ep.destination = []byte{p.Source}, []byte{p.Destination} + } + case int16(meshcore.PayloadTypeTxtMsg): + if p, err := meshcore.TextMessageFromBytes(raw); err == nil { + ep.source, ep.destination = []byte{p.Source}, []byte{p.Destination} + } + case int16(meshcore.PayloadTypePath): + if p, err := meshcore.PathFromBytes(raw); err == nil { + ep.source, ep.destination = []byte{p.Source}, []byte{p.Destination} + } + } + return ep +} + +type endpointLookup struct { + iata string + ep packetEndpoints +} + +// resolveEndpoints resolves a whole page of endpoints in at most two queries. +// Results line up with lookups; nil means no such endpoint or a failed lookup. +func (s *Store) resolveEndpoints(ctx context.Context, lookups []endpointLookup) (sources, destinations []*api.ResolvedHop) { + iatas, hashes, keys := map[string]bool{}, map[byte]bool{}, map[string][]byte{} + for _, l := range lookups { + for _, h := range [][]byte{l.ep.source, l.ep.destination} { + if len(h) == 1 { + iatas[l.iata], hashes[h[0]] = true, true + } + } + if len(l.ep.advertKey) > 0 { + keys[string(l.ep.advertKey)] = l.ep.advertKey + } + } + + candidates := map[string]map[string][]api.ResolvedPathEntry{} // iata -> hash hex + hashesOK := len(hashes) > 0 + if hashesOK { + params := sqlc.ResolveEndpointHashPairsParams{} + for iata := range iatas { + params.Iatas = append(params.Iatas, iata) + } + for h := range hashes { + params.Hashes = append(params.Hashes, []byte{h}) + } + rows, err := s.q.ResolveEndpointHashPairs(ctx, params) + if err != nil { + slog.Error("store: endpoint resolution failed", "component", "db", "error", err) + hashesOK = false + } + for _, r := range rows { + if candidates[r.Iata] == nil { + candidates[r.Iata] = map[string][]api.ResolvedPathEntry{} + } + key := hex.EncodeToString(r.Hash) + candidates[r.Iata][key] = append(candidates[r.Iata][key], api.ResolvedPathEntry{ + NodeID: r.NodeID, Name: r.Name, Latitude: r.Latitude, + Longitude: r.Longitude, PublicKey: r.PublicKey, + }) + } + } + + adverts := map[string]*api.ResolvedNode{} + advertsOK := true + if len(keys) > 0 { + pubkeys := make([][]byte, 0, len(keys)) + for _, k := range keys { + pubkeys = append(pubkeys, k) + } + rows, err := s.q.GetNodesByPubkeys(ctx, pubkeys) + if err != nil { + slog.Error("store: advert source lookup failed", "component", "db", "error", err) + advertsOK = false + } + for _, r := range rows { + adverts[string(r.PublicKey)] = &api.ResolvedNode{ + ID: r.ID, Name: r.Name, PublicKey: hex.EncodeToString(r.PublicKey), + Latitude: r.Latitude, Longitude: r.Longitude, + } + } + } + + hop := func(iata string, h []byte) *api.ResolvedHop { + if len(h) != 1 || !hashesOK { + return nil + } + r := api.BuildResolvedPath([][]byte{h}, candidates[iata])[0] + return &r + } + sources = make([]*api.ResolvedHop, len(lookups)) + destinations = make([]*api.ResolvedHop, len(lookups)) + for i, l := range lookups { + if l.ep.advert { + if advertsOK { + r := api.ResolveExactNode(adverts[string(l.ep.advertKey)]) + sources[i] = &r + } + } else { + sources[i] = hop(l.iata, l.ep.source) + } + destinations[i] = hop(l.iata, l.ep.destination) + } + return sources, destinations +} + +// fillLatestEndpoints resolves the latest observer's endpoints for a page of packets. +// lookups line up with items; rows without a latest observer are skipped. +func (s *Store) fillLatestEndpoints(ctx context.Context, items []api.PacketSummary, lookups []endpointLookup) { + sources, destinations := s.resolveEndpoints(ctx, lookups) + for i := range items { + if lo := items[i].LatestObserver; lo != nil { + lo.ResolvedSource, lo.ResolvedDestination = sources[i], destinations[i] + } + } +} diff --git a/db/endpoints_test.go b/db/endpoints_test.go new file mode 100644 index 00000000..a896d2e4 --- /dev/null +++ b/db/endpoints_test.go @@ -0,0 +1,140 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "context" + "errors" + "sort" + "testing" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + mockdb "github.com/MeshCore-Beacon/beacon-server/db/sqlc/mock" + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/google/uuid" + "github.com/meshcore-go/meshcore-go" + "go.uber.org/mock/gomock" +) + +func TestParsePacketEndpoints(t *testing.T) { + msg := append([]byte{0xbb, 0xaa, 0, 0}, make([]byte, 16)...) + key := []byte{1, 2, 3} + for _, tc := range []struct { + name string + kind uint8 + raw, key []byte + want packetEndpoints + }{ + {"direct message", meshcore.PayloadTypeTxtMsg, msg, nil, packetEndpoints{source: []byte{0xaa}, destination: []byte{0xbb}}}, + {"request", meshcore.PayloadTypeReq, msg, nil, packetEndpoints{source: []byte{0xaa}, destination: []byte{0xbb}}}, + {"advert", meshcore.PayloadTypeAdvert, nil, key, packetEndpoints{advert: true, advertKey: key}}, + {"channel message", meshcore.PayloadTypeGrpTxt, msg, nil, packetEndpoints{}}, + {"truncated", meshcore.PayloadTypeTxtMsg, []byte{0xbb}, nil, packetEndpoints{}}, + } { + t.Run(tc.name, func(t *testing.T) { + got := parsePacketEndpoints(int16(tc.kind), tc.raw, tc.key) + if got.advert != tc.want.advert || string(got.advertKey) != string(tc.want.advertKey) || + string(got.source) != string(tc.want.source) || string(got.destination) != string(tc.want.destination) { + t.Fatalf("got %+v, want %+v", got, tc.want) + } + }) + } +} + +func hopNames(h *api.ResolvedHop) []string { + if h == nil { + return nil + } + names := []string{h.Confidence} + for _, n := range h.Nodes { + names = append(names, *n.Name) + } + sort.Strings(names[1:]) + return names +} + +func TestResolveEndpointsBatchesAPage(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + store := &Store{q: mock} + name := func(s string) *string { return &s } + alice, bob, carol := uuid.New(), uuid.New(), uuid.New() + advertKey := []byte{0xaa, 0x11} + + mock.EXPECT().ResolveEndpointHashPairs(gomock.Any(), gomock.Any()).DoAndReturn( + func(_ context.Context, p sqlc.ResolveEndpointHashPairsParams) ([]sqlc.ResolveEndpointHashPairsRow, error) { + if len(p.Iatas) != 2 || len(p.Hashes) != 2 { + t.Errorf("want distinct IATAs and hashes, got %v %v", p.Iatas, p.Hashes) + } + return []sqlc.ResolveEndpointHashPairsRow{ + {Iata: "YYZ", Hash: []byte{0xaa}, NodeID: alice, Name: name("Alice")}, + {Iata: "YYZ", Hash: []byte{0xaa}, NodeID: bob, Name: name("Bob")}, + {Iata: "YYZ", Hash: []byte{0xbb}, NodeID: carol, Name: name("Carol")}, + {Iata: "YVR", Hash: []byte{0xaa}, NodeID: alice, Name: name("Alice")}, + {Iata: "YVR", Hash: []byte{0xbb}, NodeID: carol, Name: name("Carol")}, // cross-product extra + }, nil + }) + mock.EXPECT().GetNodesByPubkeys(gomock.Any(), [][]byte{advertKey}).Return( + []sqlc.GetNodesByPubkeysRow{{ID: alice, PublicKey: advertKey, Name: name("Alice")}}, nil) + + sources, destinations := store.resolveEndpoints(context.Background(), []endpointLookup{ + {iata: "YYZ", ep: packetEndpoints{source: []byte{0xaa}, destination: []byte{0xbb}}}, + {iata: "YVR", ep: packetEndpoints{source: []byte{0xaa}}}, + {iata: "YYZ", ep: packetEndpoints{advert: true, advertKey: advertKey}}, + {iata: "YYZ", ep: packetEndpoints{advert: true, advertKey: advertKey}}, + {iata: "YYZ", ep: packetEndpoints{advert: true}}, + {}, + }) + + for i, tc := range []struct{ source, destination []string }{ + {[]string{"ambiguous", "Alice", "Bob"}, []string{"high", "Carol"}}, + {[]string{"high", "Alice"}, nil}, + {[]string{"high", "Alice"}, nil}, + {[]string{"high", "Alice"}, nil}, + {[]string{"none"}, nil}, + {nil, nil}, + } { + if got := hopNames(sources[i]); !equalStrings(got, tc.source) { + t.Errorf("row %d source = %v, want %v", i, got, tc.source) + } + if got := hopNames(destinations[i]); !equalStrings(got, tc.destination) { + t.Errorf("row %d destination = %v, want %v", i, got, tc.destination) + } + } +} + +func TestResolveEndpointsFailedLookupLeavesHopsEmpty(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + store := &Store{q: mock} + mock.EXPECT().ResolveEndpointHashPairs(gomock.Any(), gomock.Any()).Return(nil, errors.New("boom")) + + sources, destinations := store.resolveEndpoints(context.Background(), []endpointLookup{ + {iata: "YYZ", ep: packetEndpoints{source: []byte{0xaa}, destination: []byte{0xbb}}}, + }) + if sources[0] != nil || destinations[0] != nil { + t.Fatal("a failed lookup must not report endpoints as unresolved") + } +} + +func TestResolveEndpointsSkipsQueriesWhenNothingToResolve(t *testing.T) { + ctrl := gomock.NewController(t) + store := &Store{q: mockdb.NewMockQuerier(ctrl)} // no EXPECT: any query fails the test + sources, _ := store.resolveEndpoints(context.Background(), []endpointLookup{{}, {iata: "YYZ"}}) + if len(sources) != 2 || sources[0] != nil || sources[1] != nil { + t.Fatal("rows without endpoints must stay empty") + } +} + +func equalStrings(a, b []string) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} diff --git a/db/migrations/038_drop_resolved_endpoints.sql b/db/migrations/038_drop_resolved_endpoints.sql new file mode 100644 index 00000000..e69190ec --- /dev/null +++ b/db/migrations/038_drop_resolved_endpoints.sql @@ -0,0 +1,6 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- Endpoints now resolve at read time like path hops. The per-observation +-- snapshots were ~60% of packet_observations on prod. +ALTER TABLE packet_observations DROP COLUMN IF EXISTS resolved_endpoints; diff --git a/db/packet_endpoints_integration_test.go b/db/packet_endpoints_integration_test.go index 756d9cdf..50d2c0bb 100644 --- a/db/packet_endpoints_integration_test.go +++ b/db/packet_endpoints_integration_test.go @@ -5,18 +5,18 @@ package db import ( "context" - "encoding/json" + "crypto/rand" "fmt" "os" - "reflect" + "strings" "testing" "time" 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/google/uuid" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" ) type endpointQueryCounter struct { @@ -33,225 +33,134 @@ func (q *endpointQueryCounter) QueryRow(ctx context.Context, sql string, args .. return q.DBTX.QueryRow(ctx, sql, args...) } -func TestPacketEndpointSnapshotsPostgres(t *testing.T) { +// Endpoints resolve live against today's nodes, in a fixed number of queries per page. +func TestPacketEndpointsResolveLivePostgres(t *testing.T) { dsn := os.Getenv("BEACON_TEST_POSTGRES_DSN") if dsn == "" { t.Skip("set BEACON_TEST_POSTGRES_DSN for the PostgreSQL regression test") } - ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) defer cancel() - conn, err := pgx.Connect(ctx, dsn) + // Own throwaway database so CI can run it against an empty server. + admin, err := pgx.Connect(ctx, dsn) if err != nil { t.Fatal(err) } - defer conn.Close(context.Background()) - tx, err := conn.Begin(ctx) + defer admin.Close(context.Background()) + name := "beacon_endpoints_" + strings.ToLower(rand.Text()) + ident := pgx.Identifier{name}.Sanitize() + if _, err := admin.Exec(ctx, "CREATE DATABASE "+ident); err != nil { + t.Fatal(err) + } + defer admin.Exec(context.Background(), "DROP DATABASE "+ident+" WITH (FORCE)") + cfg, err := pgxpool.ParseConfig(dsn) if err != nil { t.Fatal(err) } - defer tx.Rollback(context.Background()) + cfg.ConnConfig.Database = name + pool, err := pgxpool.NewWithConfig(ctx, cfg) + if err != nil { + t.Fatal(err) + } + defer pool.Close() + if err := RunMigrations(ctx, pool); err != nil { + t.Fatal(err) + } exec := func(query string, args ...any) { t.Helper() - if _, err := tx.Exec(ctx, query, args...); err != nil { + if _, err := pool.Exec(ctx, query, args...); err != nil { t.Fatal(err) } } - for _, table := range []string{"packets", "packet_observations", "observers", "transport_scopes", "nodes", "channel_messages"} { - exec("CREATE TEMP TABLE " + table + " (LIKE public." + table + " INCLUDING ALL) ON COMMIT DROP") - } - // Also lets the unchanged Store run against the regression fixture before migration. - exec("ALTER TABLE pg_temp.packet_observations ADD COLUMN IF NOT EXISTS resolved_endpoints JSONB") - exec("CREATE TEMP SEQUENCE endpoint_observation_id START 1000") - exec("ALTER TABLE pg_temp.packet_observations ALTER COLUMN id SET DEFAULT nextval('pg_temp.endpoint_observation_id')") - observer1 := uuid.MustParse("00000000-0000-0000-0000-000000000001") + exec("INSERT INTO iata_codes(iata) VALUES ('YYZ'),('YVR')") + + observer := uuid.MustParse("00000000-0000-0000-0000-000000000001") + alice, bob, carol := uuid.New(), uuid.New(), uuid.New() + aliceKey := "aa11" + fmt.Sprintf("%060d", 0) observer2 := uuid.MustParse("00000000-0000-0000-0000-000000000002") - nodeID := uuid.MustParse("00000000-0000-0000-0000-000000000003") - key := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" - name, earlier := "Captured 👋 companion", "Earlier companion" - high := &api.ResolvedHop{Confidence: "high", Nodes: []api.ResolvedNode{{ID: nodeID, Name: &name, PublicKey: key}}} - old := &api.ResolvedHop{Confidence: "high", Nodes: []api.ResolvedNode{{ID: nodeID, Name: &earlier, PublicKey: key}}} - ambiguous := &api.ResolvedHop{Confidence: "ambiguous", Nodes: []api.ResolvedNode{high.Nodes[0], {ID: observer2, PublicKey: "bb"}}} - none := &api.ResolvedHop{Confidence: "none", Nodes: []api.ResolvedNode{}} - snapshot := func(source, destination *api.ResolvedHop) []byte { - value, err := json.Marshal(struct { - Source *api.ResolvedHop `json:"source,omitempty"` - Destination *api.ResolvedHop `json:"destination,omitempty"` - }{source, destination}) - if err != nil { - t.Fatal(err) - } - return value - } - check := func(source, destination, wantSource, wantDestination *api.ResolvedHop) { - t.Helper() - if !reflect.DeepEqual(source, wantSource) || !reflect.DeepEqual(destination, wantDestination) { - t.Error("historical packet lost endpoint snapshot") - } - } - exec("INSERT INTO observers(id,public_key) VALUES ($1,'\\x01'),($2,'\\x02')", observer1, observer2) - exec("INSERT INTO nodes(id,public_key,name,node_type) VALUES ($1,decode($2,'hex'),$3,1)", nodeID, key, name) + exec("INSERT INTO observers(id,public_key) VALUES ($1,'\\x01'),($2,'\\x02')", observer, observer2) + exec(`INSERT INTO nodes(id,public_key,name,node_type) VALUES + ($1,decode($4,'hex'),'Alice',1), ($2,'\xaa22','Bob',1), ($3,'\xbb33','Carol',2)`, alice, bob, carol, aliceKey) + // Alice and Bob share prefix aa in YYZ; YVR only knows Alice. + exec(`INSERT INTO node_short_ids(node_id,iata,prefix_4) VALUES + ($1,'YYZ','\xaa110000'), ($2,'YYZ','\xaa220000'), ($3,'YYZ','\xbb330000'), ($1,'YVR','\xaa110000')`, alice, bob, carol) + anchor := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) - exec(`INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,origin_pubkey,first_heard_at,last_heard_at) -SELECT decode(lpad(to_hex(i),64,'0'),'hex'),kind,0,1,'\x00','\x00',decode($2,'hex'), - $1::timestamptz+i*interval '1 second',$1::timestamptz+i*interval '1 second' -FROM (VALUES (1,4),(2,2),(3,3),(4,4),(5,3)) v(i,kind)`, anchor, key) - for _, obs := range []struct { - id, packet int - observer uuid.UUID - iata string - data []byte + msg := append([]byte{0xbb, 0xaa, 0, 0}, make([]byte, 16)...) + packets := []struct { + kind int + raw []byte + origin any + iata string }{ - {1, 1, observer1, "YYZ", snapshot(old, nil)}, - {2, 1, observer2, "YVR", snapshot(high, nil)}, - {3, 2, observer1, "YYZ", snapshot(ambiguous, none)}, - {4, 3, observer2, "YVR", snapshot(nil, nil)}, - {5, 4, observer1, "YYZ", nil}, - } { - exec(`INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count,source_broker,resolved_endpoints) -VALUES ($1,decode($2,'hex'),$3,$4,$5,0,1,0,'fixture',$6)`, obs.id, fmt.Sprintf("%064x", obs.packet), obs.observer, obs.iata, anchor.Add(time.Duration(obs.id)*time.Second), obs.data) - } - exec("UPDATE nodes SET name='Current name'") - q := &endpointQueryCounter{DBTX: tx} + {2, msg, nil, "YYZ"}, // 1: direct message, ambiguous source + {4, []byte{0}, aliceKey, "YYZ"}, // 2: advert from Alice + {4, []byte{0}, "cc" + aliceKey[2:], "YYZ"}, // 3: advert from an unknown node + {2, msg, nil, "YVR"}, // 4: same message bytes, other IATA + } + for i, p := range packets { + n := i + 1 + exec(`INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,origin_pubkey,first_heard_at,last_heard_at) +VALUES (decode($1,'hex'),$2,0,1,$3,'\x00',decode($4,'hex'),$5,$5)`, fmt.Sprintf("%064x", n), p.kind, p.raw, p.origin, anchor.Add(time.Duration(n)*time.Second)) + exec(`INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count) +VALUES ($1,decode($2,'hex'),$3,$4,$5,0,1,0)`, n, fmt.Sprintf("%064x", n), observer, p.iata, anchor.Add(time.Duration(n)*time.Second)) + } + exec("UPDATE nodes SET name='Alice renamed' WHERE id=$1", alice) + + q := &endpointQueryCounter{DBTX: pool} store := &Store{q: sqlc.New(q)} - for _, iatas := range [][]string{nil, {"YYZ"}, {"YVR"}, {"YYZ", "YVR", "YYZ"}} { - q.calls = 0 - page, err := store.ListPackets(ctx, nil, nil, iatas, nil, time.Time{}, time.Time{}, 0, 50) - if err != nil { - t.Fatal(err) - } - if q.calls != 1 || len(page.Items) == 0 { - t.Fatalf("list used %d queries or lost its page", q.calls) + check := func(label string, got *api.ResolvedHop, want ...string) { + t.Helper() + if names := hopNames(got); !equalStrings(names, want) { + t.Errorf("%s = %v, want %v", label, names, want) } - for _, item := range page.Items { + } + checkPage := func(items []api.PacketSummary) { + t.Helper() + for _, item := range items { + lo := item.LatestObserver switch item.PacketHash { case fmt.Sprintf("%064x", 1): - // The existing regional roll-up also selects the global latest observer. - if item.LatestObserver == nil || item.LatestObserver.ID != observer2 || item.LatestObserver.IATA != "YVR" { - t.Fatal("latest observer selection changed") - } - check(item.LatestObserver.ResolvedSource, item.LatestObserver.ResolvedDestination, high, nil) + check("dm source", lo.ResolvedSource, "ambiguous", "Alice renamed", "Bob") + check("dm destination", lo.ResolvedDestination, "high", "Carol") case fmt.Sprintf("%064x", 2): - check(item.LatestObserver.ResolvedSource, item.LatestObserver.ResolvedDestination, ambiguous, none) - case fmt.Sprintf("%064x", 3), fmt.Sprintf("%064x", 4): - check(item.LatestObserver.ResolvedSource, item.LatestObserver.ResolvedDestination, nil, nil) - case fmt.Sprintf("%064x", 5): - if item.LatestObserver != nil { - t.Fatal("invented an observation for a packet without one") - } + check("advert source", lo.ResolvedSource, "high", "Alice renamed") + case fmt.Sprintf("%064x", 3): + check("unknown advert", lo.ResolvedSource, "none") + case fmt.Sprintf("%064x", 4): + check("YVR source", lo.ResolvedSource, "high", "Alice renamed") + check("YVR destination", lo.ResolvedDestination, "none") } } } - q.calls = 0 - backfill, err := store.ListPacketsAfterID(ctx, 0, -1, -1, nil, "", 50) - if err != nil || len(backfill) != 5 || q.calls != 1 { - t.Fatalf("backfill: rows=%d queries=%d err=%v", len(backfill), q.calls, err) - } - check(backfill[0].LatestObserver.ResolvedSource, backfill[0].LatestObserver.ResolvedDestination, old, nil) - check(backfill[1].LatestObserver.ResolvedSource, backfill[1].LatestObserver.ResolvedDestination, high, nil) - check(backfill[2].LatestObserver.ResolvedSource, backfill[2].LatestObserver.ResolvedDestination, ambiguous, none) - packet1 := append(make([]byte, 31), 1) - for pass := 0; pass < 2; pass++ { + + // One page query plus one hash batch and one advert batch. + for _, iatas := range [][]string{nil, {"YYZ", "YVR"}} { q.calls = 0 - packet, err := store.GetPacket(ctx, packet1) - if err != nil || len(packet.Observations) != 2 { - t.Fatalf("packet detail: %v", err) - } - check(packet.Observations[0].ResolvedSource, packet.Observations[0].ResolvedDestination, old, nil) - check(packet.Observations[1].ResolvedSource, packet.Observations[1].ResolvedDestination, high, nil) - if q.calls != 2 { - t.Errorf("snapshot detail used %d queries, want packet plus observations only", q.calls) - } - if pass == 0 { - legacy, err := store.GetPacket(ctx, append(make([]byte, 31), 4)) - if err != nil || legacy.Observations[0].ResolvedSource == nil || *legacy.Observations[0].ResolvedSource.Nodes[0].Name != "Current name" { - t.Fatalf("legacy detail lookup changed: %v", err) - } - exec("DELETE FROM nodes") + page, err := store.ListPackets(ctx, nil, nil, iatas, nil, time.Time{}, time.Time{}, 0, 50) + if err != nil || len(page.Items) != 4 || q.calls != 3 { + t.Fatalf("list %v: rows=%d queries=%d err=%v", iatas, len(page.Items), q.calls, err) } + checkPage(page.Items) } - exec(`INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at) -SELECT decode(lpad(to_hex(i),64,'0'),'hex'),4,0,1,'\x00','\x00',$1::timestamptz+i*interval '1 second',$1::timestamptz+i*interval '1 second' -FROM generate_series(6,55) i`, anchor) - exec(`INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count,source_broker,resolved_endpoints) -SELECT i,decode(lpad(to_hex(i),64,'0'),'hex'),$1,'YYZ',$2::timestamptz+i*interval '1 second',0,1,0,'fixture',$3 -FROM generate_series(6,55) i`, observer1, anchor, snapshot(high, nil)) q.calls = 0 - page, err := store.ListPackets(ctx, nil, nil, nil, nil, time.Time{}, time.Time{}, 0, 50) - if err != nil || len(page.Items) != 50 || q.calls != 1 { - t.Fatalf("50-row page: queries=%d err=%v", q.calls, err) - } - for _, item := range page.Items { - check(item.LatestObserver.ResolvedSource, item.LatestObserver.ResolvedDestination, high, nil) + backfill, err := store.ListPacketsAfterID(ctx, 0, -1, -1, nil, "", 50) + if err != nil || len(backfill) != 4 || q.calls != 3 { + t.Fatalf("backfill: rows=%d queries=%d err=%v", len(backfill), q.calls, err) } - t.Logf("50 historical rows use %d database query", q.calls) + checkPage(backfill) - // Capture uses the ordinary insert; duplicate delivery cannot replace its snapshot. - exec(`INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at) -VALUES (decode($1,'hex'),4,0,1,'\x00','\x00',NOW(),NOW())`, fmt.Sprintf("%064x", 56)) - observation := ingest.InsertObservationParams{PacketHash: append(make([]byte, 31), 56), - ObserverID: observer1, IATA: "YYZ", HeardAt: anchor, HashSize: 1, SourceBroker: "fixture", - PayloadType: 4, ResolvedEndpoints: snapshot(high, none)} - for _, wantInserted := range []bool{true, false} { - q.calls = 0 - inserted, err := store.InsertObservation(ctx, observation) - if err != nil || inserted != wantInserted || q.calls != 1 { - t.Fatalf("snapshot insert: inserted=%v queries=%d err=%v", inserted, q.calls, err) - } - observation.ResolvedEndpoints = snapshot(old, nil) - observation.IATA = "YVR" - } - stored, err := store.GetPacket(ctx, observation.PacketHash) - if err != nil || len(stored.Observations) != 1 || stored.Observations[0].IATA != "YYZ" { - t.Fatalf("duplicate observation changed: %v", err) - } - check(stored.Observations[0].ResolvedSource, stored.Observations[0].ResolvedDestination, high, none) - // First adverts can be captured before their node exists. Empty old captures - // must allow detail-time lookup once that node has been advertised. - exec("INSERT INTO nodes(id,public_key,name,node_type) VALUES ($1,decode($2,'hex'),'Later discovered companion',1)", nodeID, key) - exec("UPDATE packets SET origin_pubkey=decode($2,'hex') WHERE packet_hash=$1", observation.PacketHash, key) - for _, raw := range []string{`{}`, `{"source":{"confidence":"none","nodes":[]}}`} { - exec("UPDATE packet_observations SET resolved_endpoints=$2::jsonb,source_broker=NULL WHERE packet_hash=$1", observation.PacketHash, raw) - q.calls = 0 - stored, err = store.GetPacket(ctx, observation.PacketHash) - if err != nil || q.calls != 4 || stored.Observations[0].SourceBroker != "" { - t.Fatalf("empty snapshot fallback or nullable legacy broker: queries=%d err=%v", q.calls, err) - } - if source := stored.Observations[0].ResolvedSource; source == nil || len(source.Nodes) != 1 || source.Nodes[0].Name == nil || *source.Nodes[0].Name != "Later discovered companion" { - t.Fatal("empty capture suppressed the later advertised node") - } - } - exec(`INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,origin_pubkey,first_heard_at,last_heard_at) -VALUES (decode($1,'hex'),4,0,1,'\x00','\x00',decode($2,'hex'),$3,$3)`, fmt.Sprintf("%064x", 57), key, anchor) - observation.PacketHash, observation.ResolvedEndpoints = append(make([]byte, 31), 57), nil - inserted, err := store.InsertObservation(ctx, observation) - if err != nil || !inserted { - t.Fatalf("uncaptured observation insert: %v", err) - } - var isNull bool - if err := tx.QueryRow(ctx, "SELECT resolved_endpoints IS NULL FROM packet_observations WHERE packet_hash=$1", observation.PacketHash).Scan(&isNull); err != nil || !isNull { - t.Fatalf("uncaptured resolution was not SQL NULL: %v", err) - } - empty, err := store.GetPacket(ctx, append(make([]byte, 31), 5)) - if err != nil || len(empty.Observations) != 0 { - t.Fatalf("packet awaiting its first observation: %v", err) + // Detail resolves every observation in one batch, per its own IATA. + exec(`INSERT INTO packet_observations(id,packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count) +VALUES (5,decode($1,'hex'),$2,'YVR',$3,0,1,0)`, fmt.Sprintf("%064x", 1), observer2, anchor.Add(10*time.Second)) + q.calls = 0 + packet, err := store.GetPacket(ctx, append(make([]byte, 31), 1)) + if err != nil || len(packet.Observations) != 2 { + t.Fatalf("detail: %v", err) } -} - -func TestDecodePacketEndpointSnapshot(t *testing.T) { - for _, tc := range []struct { - raw string - captured bool - }{ - {"", false}, {"null", false}, {"[", false}, {"[]", false}, {"{}", false}, - {`{"source":{"confidence":"none","nodes":[]}}`, false}, - {`{"destination":{"confidence":"none","nodes":[]}}`, false}, - {`{"source":{},"destination":null}`, false}, - {`{"source":{"confidence":"high","nodes":[{"publicKey":"aa"}]}}`, true}, - {`{"destination":{"confidence":"ambiguous","nodes":[{"publicKey":"aa"},{"publicKey":"ab"}]}}`, true}, - } { - _, captured := decodePacketEndpointSnapshot(json.RawMessage(tc.raw)) - if captured != tc.captured { - t.Errorf("snapshot %q: captured=%v, want %v", tc.raw, captured, tc.captured) - } + if q.calls != 3 { + t.Errorf("detail used %d queries, want packet, observations and one endpoint batch", q.calls) } + check("YYZ observation", packet.Observations[0].ResolvedSource, "ambiguous", "Alice renamed", "Bob") + check("YVR observation", packet.Observations[1].ResolvedSource, "high", "Alice renamed") } diff --git a/db/packets.go b/db/packets.go index 8172c66f..2acc5d42 100644 --- a/db/packets.go +++ b/db/packets.go @@ -84,14 +84,6 @@ func buildLatestObserverPath(pathLengthByte, hashSize, hopCount *int16, pathByte return pathLength, pathBytesHex } -func decodePacketEndpointSnapshot(raw json.RawMessage) (api.PacketEndpointSnapshot, bool) { - var snapshot *api.PacketEndpointSnapshot - if len(raw) == 0 || json.Unmarshal(raw, &snapshot) != nil || snapshot == nil || !snapshot.HasResolvedNodes() { - return api.PacketEndpointSnapshot{}, false - } - return *snapshot, true -} - func (s *Store) ListPackets(ctx context.Context, payloadTypes, routeTypes []int16, iatas []string, scopes []string, since, until time.Time, cursor int64, limit int32) (api.Page[api.PacketSummary], error) { if len(iatas) > 0 { return s.listPacketsByIATAs(ctx, payloadTypes, routeTypes, iatas, scopes, since, until, cursor, limit) @@ -125,7 +117,9 @@ func (s *Store) ListPackets(ctx context.Context, payloadTypes, routeTypes []int1 rows = rows[:limit] } items := make([]api.PacketSummary, 0, len(rows)) + lookups := make([]endpointLookup, 0, len(rows)) for _, v := range rows { + var lookup endpointLookup item := api.PacketSummary{ PacketHash: hex.EncodeToString(v.PacketHash), PayloadType: v.PayloadType, @@ -141,20 +135,20 @@ func (s *Store) ListPackets(ctx context.Context, payloadTypes, routeTypes []int1 item.Summary = &v.Summary } if v.LatestObserverID != (uuid.UUID{}) { - endpoints, _ := decodePacketEndpointSnapshot(v.LatestObserverResolvedEndpoints) item.LatestObserver = &api.PacketLatestObserver{ - ID: v.LatestObserverID, - DisplayName: v.LatestObserverName, - IATA: v.LatestObserverIata, - ResolvedSource: endpoints.Source, - ResolvedDestination: endpoints.Destination, + ID: v.LatestObserverID, + DisplayName: v.LatestObserverName, + IATA: v.LatestObserverIata, } + lookup = endpointLookup{iata: v.LatestObserverIata, ep: parsePacketEndpoints(v.PayloadType, v.RawPayload, v.OriginPubkey)} item.LatestObserver.PathLength, item.LatestObserver.PathBytes = buildLatestObserverPath( &v.LatestObserverPathLengthByte, &v.LatestObserverHashSize, &v.LatestObserverHopCount, v.LatestObserverPathBytes, ) } items = append(items, item) + lookups = append(lookups, lookup) } + s.fillLatestEndpoints(ctx, items, lookups) var nextCursor *int64 if hasMore && len(items) > 0 { last := items[len(items)-1].LastHeardAt @@ -210,7 +204,9 @@ func (s *Store) listPacketsByIATAs(ctx context.Context, payloadTypes, routeTypes scanFloor = rows[0].ScanFloor } items := make([]api.PacketSummary, 0, len(rows)) + lookups := make([]endpointLookup, 0, len(rows)) for _, v := range rows { + var lookup endpointLookup item := api.PacketSummary{ PacketHash: hex.EncodeToString(v.PacketHash), PayloadType: v.PayloadType, @@ -226,20 +222,20 @@ func (s *Store) listPacketsByIATAs(ctx context.Context, payloadTypes, routeTypes item.Summary = &v.Summary } if v.LatestObserverID != (uuid.UUID{}) { - endpoints, _ := decodePacketEndpointSnapshot(v.LatestObserverResolvedEndpoints) item.LatestObserver = &api.PacketLatestObserver{ - ID: v.LatestObserverID, - DisplayName: v.LatestObserverName, - IATA: v.LatestObserverIata, - ResolvedSource: endpoints.Source, - ResolvedDestination: endpoints.Destination, + ID: v.LatestObserverID, + DisplayName: v.LatestObserverName, + IATA: v.LatestObserverIata, } + lookup = endpointLookup{iata: v.LatestObserverIata, ep: parsePacketEndpoints(v.PayloadType, v.RawPayload, v.OriginPubkey)} item.LatestObserver.PathLength, item.LatestObserver.PathBytes = buildLatestObserverPath( &v.LatestObserverPathLengthByte, &v.LatestObserverHashSize, &v.LatestObserverHopCount, v.LatestObserverPathBytes, ) } items = append(items, item) + lookups = append(lookups, lookup) } + s.fillLatestEndpoints(ctx, items, lookups) // Cursor follows site-local recency, not the packet's global last_heard_at. var nextCursor *int64 if hasMore && len(rows) > 0 { @@ -272,7 +268,9 @@ func (s *Store) ListPacketsAfterID(ctx context.Context, afterObservationID int64 return nil, err } items := make([]api.PacketSummary, 0, len(rows)) + lookups := make([]endpointLookup, 0, len(rows)) for _, v := range rows { + var lookup endpointLookup item := api.PacketSummary{ PacketHash: hex.EncodeToString(v.PacketHash), PayloadType: v.PayloadType, @@ -288,14 +286,12 @@ func (s *Store) ListPacketsAfterID(ctx context.Context, afterObservationID int64 item.Summary = &v.Summary } if v.LatestObserverID != (uuid.UUID{}) { - endpoints, _ := decodePacketEndpointSnapshot(v.LatestObserverResolvedEndpoints) item.LatestObserver = &api.PacketLatestObserver{ - ID: v.LatestObserverID, - DisplayName: v.LatestObserverName, - IATA: v.LatestObserverIata, - ResolvedSource: endpoints.Source, - ResolvedDestination: endpoints.Destination, + ID: v.LatestObserverID, + DisplayName: v.LatestObserverName, + IATA: v.LatestObserverIata, } + lookup = endpointLookup{iata: v.LatestObserverIata, ep: parsePacketEndpoints(v.PayloadType, v.RawPayload, v.OriginPubkey)} // Inner join here (unlike ListPackets/listPacketsByIATAs' LEFT JOIN LATERAL), so // these are never nil when an observer was joined at all. item.LatestObserver.PathLength, item.LatestObserver.PathBytes = buildLatestObserverPath( @@ -303,7 +299,9 @@ func (s *Store) ListPacketsAfterID(ctx context.Context, afterObservationID int64 ) } items = append(items, item) + lookups = append(lookups, lookup) } + s.fillLatestEndpoints(ctx, items, lookups) return items, nil } @@ -417,42 +415,14 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, } } } - // 1-byte source/destination hashes for the payload types that carry a resolvable - // endpoint (REQUEST, RESPONSE, TEXT_MESSAGE, PATH, ANON_REQ's destination). Constant - // for this packet hash, so parsed once; resolution itself still happens per-observation - // below since candidates depend on the observation's IATA, same as the intermediate - // hop resolution already does. - var sourceHashByte, destHashByte []byte - switch row.PayloadType { - case int16(meshcore.PayloadTypeAnonReq): - if anonReq, err := meshcore.AnonReqFromBytes(row.RawPayload); err == nil { - destHashByte = []byte{anonReq.Destination} - } - case int16(meshcore.PayloadTypeReq): - if req, err := meshcore.RequestFromBytes(row.RawPayload); err == nil { - sourceHashByte = []byte{req.Source} - destHashByte = []byte{req.Destination} - } - case int16(meshcore.PayloadTypeResponse): - if resp, err := meshcore.ResponseFromBytes(row.RawPayload); err == nil { - sourceHashByte = []byte{resp.Source} - destHashByte = []byte{resp.Destination} - } - case int16(meshcore.PayloadTypeTxtMsg): - if txt, err := meshcore.TextMessageFromBytes(row.RawPayload); err == nil { - sourceHashByte = []byte{txt.Source} - destHashByte = []byte{txt.Destination} - } - case int16(meshcore.PayloadTypePath): - if path, err := meshcore.PathFromBytes(row.RawPayload); err == nil { - sourceHashByte = []byte{path.Source} - destHashByte = []byte{path.Destination} - } + // Endpoints resolve against each observation's IATA, all in one batch. + ep := parsePacketEndpoints(row.PayloadType, row.RawPayload, row.OriginPubkey) + lookups := make([]endpointLookup, len(obsRows)) + for i, v := range obsRows { + lookups[i] = endpointLookup{iata: v.Iata, ep: ep} } - // Legacy ADVERT sources need an exact lookup, at most once for this packet. - var resolvedAdvertSource *api.ResolvedNode - advertSourceLookedUp := false - for _, v := range obsRows { + sources, destinations := s.resolveEndpoints(ctx, lookups) + for i, v := range obsRows { obs := api.PacketObservationDetail{ ID: v.ID, ObserverID: v.ObserverID, @@ -496,38 +466,7 @@ func (s *Store) GetPacket(ctx context.Context, packetHash []byte) (*api.Packet, } } obs.ResolvedPath = resolvedPath - if endpoints, captured := decodePacketEndpointSnapshot(v.ResolvedEndpoints); captured { - obs.ResolvedSource = endpoints.Source - obs.ResolvedDestination = endpoints.Destination - } else { - if row.PayloadType == int16(meshcore.PayloadTypeAdvert) { - if !advertSourceLookedUp && row.OriginPubkey != nil { - if nodeID, err := s.GetNodeByPubkey(ctx, row.OriginPubkey); err == nil { - if nodes, err := s.GetNodesByIDs(ctx, []uuid.UUID{nodeID}); err == nil { - resolvedAdvertSource = nodes[nodeID] - } - } - advertSourceLookedUp = true - } - hop := api.ResolveExactNode(resolvedAdvertSource) - obs.ResolvedSource = &hop - } else if len(sourceHashByte) == 1 { - if r, err := s.ResolveEndpointHashes(ctx, v.Iata, [][]byte{sourceHashByte}); err != nil { - slog.Error(fmt.Sprintf("store: source resolution failed for observation %d", v.ID), "component", "db", "error", err) - } else { - hop := api.BuildResolvedPath([][]byte{sourceHashByte}, r)[0] - obs.ResolvedSource = &hop - } - } - if len(destHashByte) == 1 { - if r, err := s.ResolveEndpointHashes(ctx, v.Iata, [][]byte{destHashByte}); err != nil { - slog.Error(fmt.Sprintf("store: destination resolution failed for observation %d", v.ID), "component", "db", "error", err) - } else { - hop := api.BuildResolvedPath([][]byte{destHashByte}, r)[0] - obs.ResolvedDestination = &hop - } - } - } + obs.ResolvedSource, obs.ResolvedDestination = sources[i], destinations[i] if row.PayloadType == int16(meshcore.PayloadTypeTrace) && len(traceRawHashes) > 0 { // Swap in the trace's own path hashes so PathData's hop-block split (driven by // pathBytes + hashSize) lines up 1:1 with resolvedPath -- the raw SNR bytes @@ -594,7 +533,6 @@ func (s *Store) InsertObservation(ctx context.Context, o ingest.InsertObservatio CodingRate: &o.CodingRate, SourceBroker: &o.SourceBroker, PayloadType: &o.PayloadType, - ResolvedEndpoints: o.ResolvedEndpoints, AirtimeMs: o.AirtimeMs, } row, err := s.q.InsertObservation(ctx, params) diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 1f84bdf3..9c467944 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -489,10 +489,11 @@ SELECT COALESCE(po.hash_size, 0::smallint) AS latest_observer_hash_size, COALESCE(po.hop_count, 0::smallint) AS latest_observer_hop_count, po.path_bytes AS latest_observer_path_bytes, - po.resolved_endpoints AS latest_observer_resolved_endpoints + p.raw_payload, + p.origin_pubkey FROM packets p LEFT JOIN LATERAL ( - SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes, resolved_endpoints + SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes FROM packet_observations WHERE packet_hash = p.packet_hash ORDER BY heard_at DESC @@ -599,12 +600,13 @@ SELECT COALESCE(po.hash_size, 0::smallint) AS latest_observer_hash_size, COALESCE(po.hop_count, 0::smallint) AS latest_observer_hop_count, po.path_bytes AS latest_observer_path_bytes, - po.resolved_endpoints AS latest_observer_resolved_endpoints + p.raw_payload, + p.origin_pubkey FROM page sh CROSS JOIN saturation sat JOIN packets p ON p.packet_hash = sh.packet_hash LEFT JOIN LATERAL ( - SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes, resolved_endpoints + SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes FROM packet_observations WHERE packet_hash = p.packet_hash ORDER BY heard_at DESC @@ -643,7 +645,8 @@ SELECT po.hash_size AS latest_observer_hash_size, po.hop_count AS latest_observer_hop_count, po.path_bytes AS latest_observer_path_bytes, - po.resolved_endpoints AS latest_observer_resolved_endpoints, + p.raw_payload, + p.origin_pubkey, ts.name AS scope_name FROM packets p JOIN packet_observations po ON po.packet_hash = p.packet_hash @@ -729,10 +732,9 @@ INSERT INTO packet_observations ( coding_rate, source_broker, payload_type, - resolved_endpoints, airtime_ms ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19 + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18 ) ON CONFLICT (packet_hash, observer_id) DO NOTHING RETURNING *; @@ -795,6 +797,11 @@ WHERE id = ANY($1::uuid[]); -- name: GetNodeByPubkey :one SELECT id FROM nodes WHERE public_key = $1; +-- name: GetNodesByPubkeys :many +SELECT id, public_key, name, latitude, longitude +FROM nodes +WHERE public_key = ANY(@pubkeys::bytea[]); + -- name: ListNodes :many -- Limit the filtered node page before enriching IATA membership and neighbours. WITH page AS ( @@ -1413,6 +1420,19 @@ CROSS JOIN LATERAL ( WHERE ns.iata = $1 AND ns.prefix_1 = ANY($2::bytea[]); +-- name: ResolveEndpointHashPairs :many +-- Batch form of ResolveEndpointHashes for a page of packets. Matches the cross +-- product of IATAs and hashes; callers pick out the pairs they asked for. +SELECT ns.iata, ns.prefix_1 AS hash, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key +FROM node_short_ids ns +CROSS JOIN LATERAL ( + SELECT id, name, latitude, longitude, public_key + FROM nodes WHERE id = ns.node_id + LIMIT 1 +) n +WHERE ns.iata = ANY(@iatas::bpchar[]) + AND ns.prefix_1 = ANY(@hashes::bytea[]); + -- name: ResolvePathHashesP2 :many SELECT ns.prefix_4 AS hash, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key FROM node_short_ids ns diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 4efbfaa5..6a665e77 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -339,6 +339,21 @@ func (mr *MockQuerierMockRecorder) GetNodesByIDs(ctx, dollar_1 any) *gomock.Call return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetNodesByIDs", reflect.TypeOf((*MockQuerier)(nil).GetNodesByIDs), ctx, dollar_1) } +// GetNodesByPubkeys mocks base method. +func (m *MockQuerier) GetNodesByPubkeys(ctx context.Context, pubkeys [][]byte) ([]db.GetNodesByPubkeysRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "GetNodesByPubkeys", ctx, pubkeys) + ret0, _ := ret[0].([]db.GetNodesByPubkeysRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// GetNodesByPubkeys indicates an expected call of GetNodesByPubkeys. +func (mr *MockQuerierMockRecorder) GetNodesByPubkeys(ctx, pubkeys any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetNodesByPubkeys", reflect.TypeOf((*MockQuerier)(nil).GetNodesByPubkeys), ctx, pubkeys) +} + // GetObserverActivityHourly mocks base method. func (m *MockQuerier) GetObserverActivityHourly(ctx context.Context, arg db.GetObserverActivityHourlyParams) ([]db.GetObserverActivityHourlyRow, error) { m.ctrl.T.Helper() @@ -1391,6 +1406,21 @@ func (mr *MockQuerierMockRecorder) RefreshTopTalkers(ctx any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RefreshTopTalkers", reflect.TypeOf((*MockQuerier)(nil).RefreshTopTalkers), ctx) } +// ResolveEndpointHashPairs mocks base method. +func (m *MockQuerier) ResolveEndpointHashPairs(ctx context.Context, arg db.ResolveEndpointHashPairsParams) ([]db.ResolveEndpointHashPairsRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ResolveEndpointHashPairs", ctx, arg) + ret0, _ := ret[0].([]db.ResolveEndpointHashPairsRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ResolveEndpointHashPairs indicates an expected call of ResolveEndpointHashPairs. +func (mr *MockQuerierMockRecorder) ResolveEndpointHashPairs(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ResolveEndpointHashPairs", reflect.TypeOf((*MockQuerier)(nil).ResolveEndpointHashPairs), ctx, arg) +} + // ResolveEndpointHashes mocks base method. func (m *MockQuerier) ResolveEndpointHashes(ctx context.Context, arg db.ResolveEndpointHashesParams) ([]db.ResolveEndpointHashesRow, error) { m.ctrl.T.Helper() diff --git a/db/sqlc/models.go b/db/sqlc/models.go index 126b37a4..b5ffb054 100644 --- a/db/sqlc/models.go +++ b/db/sqlc/models.go @@ -335,7 +335,6 @@ type PacketObservation struct { CodingRate *int16 `json:"coding_rate"` SourceBroker *string `json:"source_broker"` PayloadType *int16 `json:"payload_type"` - ResolvedEndpoints []byte `json:"resolved_endpoints"` AirtimeMs *float32 `json:"airtime_ms"` } diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index d3913c42..232eae3b 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -54,6 +54,7 @@ type Querier interface { // Returns the neighbors of a node with details, ordered by most recently seen. GetNodeNeighbors(ctx context.Context, nodeID uuid.UUID) ([]GetNodeNeighborsRow, error) GetNodesByIDs(ctx context.Context, dollar_1 []uuid.UUID) ([]GetNodesByIDsRow, error) + GetNodesByPubkeys(ctx context.Context, pubkeys [][]byte) ([]GetNodesByPubkeysRow, error) // Hour-or-coarser buckets summed from the hourly rollup; same COALESCE-plus-count shape as the raw query. GetObserverActivityHourly(ctx context.Context, arg GetObserverActivityHourlyParams) ([]GetObserverActivityHourlyRow, error) GetObserverActivityHourlyPayloadTypes(ctx context.Context, arg GetObserverActivityHourlyPayloadTypesParams) ([]GetObserverActivityHourlyPayloadTypesRow, error) @@ -224,6 +225,9 @@ type Querier interface { RefreshTopNodes(ctx context.Context) error RefreshTopObservers(ctx context.Context) error RefreshTopTalkers(ctx context.Context) error + // Batch form of ResolveEndpointHashes for a page of packets. Matches the cross + // product of IATAs and hashes; callers pick out the pairs they asked for. + ResolveEndpointHashPairs(ctx context.Context, arg ResolveEndpointHashPairsParams) ([]ResolveEndpointHashPairsRow, error) // Logical endpoints can be any advertised role, unlike intermediate relay hops. // Endpoint hashes are always one byte; use the existing (iata, prefix_1) index. // LIMIT 1 keeps generic plans on a node PK lookup per candidate instead of diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index d454af2d..38d299c6 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -608,6 +608,46 @@ func (q *Queries) GetNodesByIDs(ctx context.Context, dollar_1 []uuid.UUID) ([]Ge return items, nil } +const getNodesByPubkeys = `-- name: GetNodesByPubkeys :many +SELECT id, public_key, name, latitude, longitude +FROM nodes +WHERE public_key = ANY($1::bytea[]) +` + +type GetNodesByPubkeysRow struct { + ID uuid.UUID `json:"id"` + PublicKey []byte `json:"public_key"` + Name *string `json:"name"` + Latitude *float64 `json:"latitude"` + Longitude *float64 `json:"longitude"` +} + +func (q *Queries) GetNodesByPubkeys(ctx context.Context, pubkeys [][]byte) ([]GetNodesByPubkeysRow, error) { + rows, err := q.db.Query(ctx, getNodesByPubkeys, pubkeys) + if err != nil { + return nil, err + } + defer rows.Close() + items := []GetNodesByPubkeysRow{} + for rows.Next() { + var i GetNodesByPubkeysRow + if err := rows.Scan( + &i.ID, + &i.PublicKey, + &i.Name, + &i.Latitude, + &i.Longitude, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const getObserverActivityHourly = `-- name: GetObserverActivityHourly :many SELECT date_bin($3::interval, bucket, TIMESTAMPTZ 'epoch')::timestamptz AS bucket, @@ -2049,13 +2089,12 @@ INSERT INTO packet_observations ( coding_rate, source_broker, payload_type, - resolved_endpoints, airtime_ms ) VALUES ( - $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19 + $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18 ) ON CONFLICT (packet_hash, observer_id) DO NOTHING -RETURNING id, packet_hash, observer_id, iata, heard_at, path_length_byte, hash_size, hop_count, path_bytes, rssi, snr, propagation_time_ms, radio_freq_mhz, spread_factor, bandwidth_khz, coding_rate, source_broker, payload_type, resolved_endpoints, airtime_ms +RETURNING id, packet_hash, observer_id, iata, heard_at, path_length_byte, hash_size, hop_count, path_bytes, rssi, snr, propagation_time_ms, radio_freq_mhz, spread_factor, bandwidth_khz, coding_rate, source_broker, payload_type, airtime_ms ` type InsertObservationParams struct { @@ -2076,7 +2115,6 @@ type InsertObservationParams struct { CodingRate *int16 `json:"coding_rate"` SourceBroker *string `json:"source_broker"` PayloadType *int16 `json:"payload_type"` - ResolvedEndpoints []byte `json:"resolved_endpoints"` AirtimeMs *float32 `json:"airtime_ms"` } @@ -2102,7 +2140,6 @@ func (q *Queries) InsertObservation(ctx context.Context, arg InsertObservationPa arg.CodingRate, arg.SourceBroker, arg.PayloadType, - arg.ResolvedEndpoints, arg.AirtimeMs, ) var i PacketObservation @@ -2125,7 +2162,6 @@ func (q *Queries) InsertObservation(ctx context.Context, arg InsertObservationPa &i.CodingRate, &i.SourceBroker, &i.PayloadType, - &i.ResolvedEndpoints, &i.AirtimeMs, ) return i, err @@ -2923,7 +2959,7 @@ func (q *Queries) ListNodes(ctx context.Context, arg ListNodesParams) ([]ListNod } const listObservationsForPacket = `-- name: ListObservationsForPacket :many -SELECT po.id, po.packet_hash, po.observer_id, po.iata, po.heard_at, po.path_length_byte, po.hash_size, po.hop_count, po.path_bytes, po.rssi, po.snr, po.propagation_time_ms, po.radio_freq_mhz, po.spread_factor, po.bandwidth_khz, po.coding_rate, po.source_broker, po.payload_type, po.resolved_endpoints, po.airtime_ms, o.display_name AS observer_name +SELECT po.id, po.packet_hash, po.observer_id, po.iata, po.heard_at, po.path_length_byte, po.hash_size, po.hop_count, po.path_bytes, po.rssi, po.snr, po.propagation_time_ms, po.radio_freq_mhz, po.spread_factor, po.bandwidth_khz, po.coding_rate, po.source_broker, po.payload_type, po.airtime_ms, o.display_name AS observer_name FROM packet_observations po LEFT JOIN observers o ON o.id = po.observer_id WHERE po.packet_hash = $1 @@ -2949,7 +2985,6 @@ type ListObservationsForPacketRow struct { CodingRate *int16 `json:"coding_rate"` SourceBroker *string `json:"source_broker"` PayloadType *int16 `json:"payload_type"` - ResolvedEndpoints []byte `json:"resolved_endpoints"` AirtimeMs *float32 `json:"airtime_ms"` ObserverName *string `json:"observer_name"` } @@ -2982,7 +3017,6 @@ func (q *Queries) ListObservationsForPacket(ctx context.Context, packetHash []by &i.CodingRate, &i.SourceBroker, &i.PayloadType, - &i.ResolvedEndpoints, &i.AirtimeMs, &i.ObserverName, ); err != nil { @@ -3216,10 +3250,11 @@ SELECT COALESCE(po.hash_size, 0::smallint) AS latest_observer_hash_size, COALESCE(po.hop_count, 0::smallint) AS latest_observer_hop_count, po.path_bytes AS latest_observer_path_bytes, - po.resolved_endpoints AS latest_observer_resolved_endpoints + p.raw_payload, + p.origin_pubkey FROM packets p LEFT JOIN LATERAL ( - SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes, resolved_endpoints + SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes FROM packet_observations WHERE packet_hash = p.packet_hash ORDER BY heard_at DESC @@ -3249,23 +3284,24 @@ type ListPacketsParams struct { } type ListPacketsRow struct { - PacketHash []byte `json:"packet_hash"` - PayloadType int16 `json:"payload_type"` - Summary string `json:"summary"` - RouteType int16 `json:"route_type"` - FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` - LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` - ScopeID *int32 `json:"scope_id"` - ScopeName *string `json:"scope_name"` - ObservationCount int64 `json:"observation_count"` - LatestObserverID uuid.UUID `json:"latest_observer_id"` - LatestObserverName *string `json:"latest_observer_name"` - LatestObserverIata string `json:"latest_observer_iata"` - LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"` - LatestObserverHashSize int16 `json:"latest_observer_hash_size"` - LatestObserverHopCount int16 `json:"latest_observer_hop_count"` - LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"` - LatestObserverResolvedEndpoints []byte `json:"latest_observer_resolved_endpoints"` + PacketHash []byte `json:"packet_hash"` + PayloadType int16 `json:"payload_type"` + Summary string `json:"summary"` + RouteType int16 `json:"route_type"` + FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` + LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` + ScopeID *int32 `json:"scope_id"` + ScopeName *string `json:"scope_name"` + ObservationCount int64 `json:"observation_count"` + LatestObserverID uuid.UUID `json:"latest_observer_id"` + LatestObserverName *string `json:"latest_observer_name"` + LatestObserverIata string `json:"latest_observer_iata"` + LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"` + LatestObserverHashSize int16 `json:"latest_observer_hash_size"` + LatestObserverHopCount int16 `json:"latest_observer_hop_count"` + LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"` + RawPayload []byte `json:"raw_payload"` + OriginPubkey []byte `json:"origin_pubkey"` } // Returns packets with the latest observation rolled in for display. @@ -3305,7 +3341,8 @@ func (q *Queries) ListPackets(ctx context.Context, arg ListPacketsParams) ([]Lis &i.LatestObserverHashSize, &i.LatestObserverHopCount, &i.LatestObserverPathBytes, - &i.LatestObserverResolvedEndpoints, + &i.RawPayload, + &i.OriginPubkey, ); err != nil { return nil, err } @@ -3344,7 +3381,8 @@ SELECT po.hash_size AS latest_observer_hash_size, po.hop_count AS latest_observer_hop_count, po.path_bytes AS latest_observer_path_bytes, - po.resolved_endpoints AS latest_observer_resolved_endpoints, + p.raw_payload, + p.origin_pubkey, ts.name AS scope_name FROM packets p JOIN packet_observations po ON po.packet_hash = p.packet_hash @@ -3369,22 +3407,23 @@ type ListPacketsAfterIDParams struct { } type ListPacketsAfterIDRow struct { - PacketHash []byte `json:"packet_hash"` - PayloadType int16 `json:"payload_type"` - Summary string `json:"summary"` - RouteType int16 `json:"route_type"` - FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` - LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` - ObservationCount int64 `json:"observation_count"` - LatestObserverID uuid.UUID `json:"latest_observer_id"` - LatestObserverName *string `json:"latest_observer_name"` - LatestObserverIata string `json:"latest_observer_iata"` - LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"` - LatestObserverHashSize int16 `json:"latest_observer_hash_size"` - LatestObserverHopCount int16 `json:"latest_observer_hop_count"` - LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"` - LatestObserverResolvedEndpoints []byte `json:"latest_observer_resolved_endpoints"` - ScopeName *string `json:"scope_name"` + PacketHash []byte `json:"packet_hash"` + PayloadType int16 `json:"payload_type"` + Summary string `json:"summary"` + RouteType int16 `json:"route_type"` + FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` + LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` + ObservationCount int64 `json:"observation_count"` + LatestObserverID uuid.UUID `json:"latest_observer_id"` + LatestObserverName *string `json:"latest_observer_name"` + LatestObserverIata string `json:"latest_observer_iata"` + LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"` + LatestObserverHashSize int16 `json:"latest_observer_hash_size"` + LatestObserverHopCount int16 `json:"latest_observer_hop_count"` + LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"` + RawPayload []byte `json:"raw_payload"` + OriginPubkey []byte `json:"origin_pubkey"` + ScopeName *string `json:"scope_name"` } // Returns packets with observations after the given observation ID, ordered oldest first. @@ -3420,7 +3459,8 @@ func (q *Queries) ListPacketsAfterID(ctx context.Context, arg ListPacketsAfterID &i.LatestObserverHashSize, &i.LatestObserverHopCount, &i.LatestObserverPathBytes, - &i.LatestObserverResolvedEndpoints, + &i.RawPayload, + &i.OriginPubkey, &i.ScopeName, ); err != nil { return nil, err @@ -3509,12 +3549,13 @@ SELECT COALESCE(po.hash_size, 0::smallint) AS latest_observer_hash_size, COALESCE(po.hop_count, 0::smallint) AS latest_observer_hop_count, po.path_bytes AS latest_observer_path_bytes, - po.resolved_endpoints AS latest_observer_resolved_endpoints + p.raw_payload, + p.origin_pubkey FROM page sh CROSS JOIN saturation sat JOIN packets p ON p.packet_hash = sh.packet_hash LEFT JOIN LATERAL ( - SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes, resolved_endpoints + SELECT observer_id, iata, path_length_byte, hash_size, hop_count, path_bytes FROM packet_observations WHERE packet_hash = p.packet_hash ORDER BY heard_at DESC @@ -3538,26 +3579,27 @@ type ListPacketsByIATAsParams struct { } type ListPacketsByIATAsRow struct { - PacketHash []byte `json:"packet_hash"` - PayloadType int16 `json:"payload_type"` - Summary string `json:"summary"` - RouteType int16 `json:"route_type"` - FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` - LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` - ScopeID *int32 `json:"scope_id"` - ScopeName *string `json:"scope_name"` - SiteHeardAt pgtype.Timestamptz `json:"site_heard_at"` - ScanSaturated bool `json:"scan_saturated"` - ScanFloor pgtype.Timestamptz `json:"scan_floor"` - ObservationCount int64 `json:"observation_count"` - LatestObserverID uuid.UUID `json:"latest_observer_id"` - LatestObserverName *string `json:"latest_observer_name"` - LatestObserverIata string `json:"latest_observer_iata"` - LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"` - LatestObserverHashSize int16 `json:"latest_observer_hash_size"` - LatestObserverHopCount int16 `json:"latest_observer_hop_count"` - LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"` - LatestObserverResolvedEndpoints []byte `json:"latest_observer_resolved_endpoints"` + PacketHash []byte `json:"packet_hash"` + PayloadType int16 `json:"payload_type"` + Summary string `json:"summary"` + RouteType int16 `json:"route_type"` + FirstHeardAt pgtype.Timestamptz `json:"first_heard_at"` + LastHeardAt pgtype.Timestamptz `json:"last_heard_at"` + ScopeID *int32 `json:"scope_id"` + ScopeName *string `json:"scope_name"` + SiteHeardAt pgtype.Timestamptz `json:"site_heard_at"` + ScanSaturated bool `json:"scan_saturated"` + ScanFloor pgtype.Timestamptz `json:"scan_floor"` + ObservationCount int64 `json:"observation_count"` + LatestObserverID uuid.UUID `json:"latest_observer_id"` + LatestObserverName *string `json:"latest_observer_name"` + LatestObserverIata string `json:"latest_observer_iata"` + LatestObserverPathLengthByte int16 `json:"latest_observer_path_length_byte"` + LatestObserverHashSize int16 `json:"latest_observer_hash_size"` + LatestObserverHopCount int16 `json:"latest_observer_hop_count"` + LatestObserverPathBytes []byte `json:"latest_observer_path_bytes"` + RawPayload []byte `json:"raw_payload"` + OriginPubkey []byte `json:"origin_pubkey"` } // IATA-filtered packet list, driven from idx_observations_iata_heard. @@ -3612,7 +3654,8 @@ func (q *Queries) ListPacketsByIATAs(ctx context.Context, arg ListPacketsByIATAs &i.LatestObserverHashSize, &i.LatestObserverHopCount, &i.LatestObserverPathBytes, - &i.LatestObserverResolvedEndpoints, + &i.RawPayload, + &i.OriginPubkey, ); err != nil { return nil, err } @@ -3951,6 +3994,63 @@ func (q *Queries) RefreshTopTalkers(ctx context.Context) error { return err } +const resolveEndpointHashPairs = `-- name: ResolveEndpointHashPairs :many +SELECT ns.iata, ns.prefix_1 AS hash, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key +FROM node_short_ids ns +CROSS JOIN LATERAL ( + SELECT id, name, latitude, longitude, public_key + FROM nodes WHERE id = ns.node_id + LIMIT 1 +) n +WHERE ns.iata = ANY($1::bpchar[]) + AND ns.prefix_1 = ANY($2::bytea[]) +` + +type ResolveEndpointHashPairsParams struct { + Iatas []string `json:"iatas"` + Hashes [][]byte `json:"hashes"` +} + +type ResolveEndpointHashPairsRow struct { + Iata string `json:"iata"` + Hash []byte `json:"hash"` + NodeID uuid.UUID `json:"node_id"` + Name *string `json:"name"` + Latitude *float64 `json:"latitude"` + Longitude *float64 `json:"longitude"` + PublicKey []byte `json:"public_key"` +} + +// Batch form of ResolveEndpointHashes for a page of packets. Matches the cross +// product of IATAs and hashes; callers pick out the pairs they asked for. +func (q *Queries) ResolveEndpointHashPairs(ctx context.Context, arg ResolveEndpointHashPairsParams) ([]ResolveEndpointHashPairsRow, error) { + rows, err := q.db.Query(ctx, resolveEndpointHashPairs, arg.Iatas, arg.Hashes) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ResolveEndpointHashPairsRow{} + for rows.Next() { + var i ResolveEndpointHashPairsRow + if err := rows.Scan( + &i.Iata, + &i.Hash, + &i.NodeID, + &i.Name, + &i.Latitude, + &i.Longitude, + &i.PublicKey, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const resolveEndpointHashes = `-- name: ResolveEndpointHashes :many SELECT ns.prefix_1 AS hash, n.id AS node_id, n.name, n.latitude, n.longitude, n.public_key FROM node_short_ids ns diff --git a/internal/api/packets.go b/internal/api/packets.go index bb4ac37e..2b2e4ebf 100644 --- a/internal/api/packets.go +++ b/internal/api/packets.go @@ -94,21 +94,6 @@ type ResolvedHop struct { Nodes []ResolvedNode `json:"nodes"` // empty for "none", one for "high", multiple for "ambiguous" } -// PacketEndpointSnapshot is the internal storage shape for an observation's -// endpoint resolution at ingest. Names and confidence are preserved even when -// the current node registry changes. REST/WS expose the existing per-endpoint fields. -type PacketEndpointSnapshot struct { - Source *ResolvedHop `json:"source,omitempty"` - Destination *ResolvedHop `json:"destination,omitempty"` -} - -// HasResolvedNodes distinguishes a historical capture from an unresolved lookup -// that should be retried when a node becomes known (including a first advert). -func (s PacketEndpointSnapshot) HasResolvedNodes() bool { - return (s.Source != nil && len(s.Source.Nodes) > 0) || - (s.Destination != nil && len(s.Destination.Nodes) > 0) -} - // ResolvedNode is a node reference within a resolved path hop. type ResolvedNode struct { ID uuid.UUID `json:"id"` diff --git a/internal/api/packets_test.go b/internal/api/packets_test.go index 2befa568..f430f60a 100644 --- a/internal/api/packets_test.go +++ b/internal/api/packets_test.go @@ -10,28 +10,6 @@ import ( "github.com/meshcore-go/meshcore-go" ) -func TestPacketEndpointSnapshotHasResolvedNodes(t *testing.T) { - none := &api.ResolvedHop{Confidence: "none", Nodes: []api.ResolvedNode{}} - known := &api.ResolvedHop{Confidence: "high", Nodes: []api.ResolvedNode{{PublicKey: "aa"}}} - for _, tc := range []struct { - name string - snapshot api.PacketEndpointSnapshot - want bool - }{ - {"absent", api.PacketEndpointSnapshot{}, false}, - {"unresolved source", api.PacketEndpointSnapshot{Source: none}, false}, - {"both unresolved", api.PacketEndpointSnapshot{Source: none, Destination: none}, false}, - {"source only", api.PacketEndpointSnapshot{Source: known, Destination: none}, true}, - {"destination only", api.PacketEndpointSnapshot{Destination: known}, true}, - } { - t.Run(tc.name, func(t *testing.T) { - if got := tc.snapshot.HasResolvedNodes(); got != tc.want { - t.Fatalf("resolved nodes = %v, want %v", got, tc.want) - } - }) - } -} - func TestPayloadTypeName(t *testing.T) { tests := []struct { input int16 diff --git a/internal/ingest/packet.go b/internal/ingest/packet.go index d5176c1d..657da999 100644 --- a/internal/ingest/packet.go +++ b/internal/ingest/packet.go @@ -55,7 +55,6 @@ type InsertObservationParams struct { CodingRate int16 SourceBroker string PayloadType int16 - ResolvedEndpoints json.RawMessage AirtimeMs *float32 // nil when the observer never reported costable radio settings } @@ -763,7 +762,7 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ if err != nil { w.log.Error(fmt.Sprintf("db: get observer radio failed for %s", pubkeyHex), "error", err) } - // Save the same endpoint resolution used by the live event in the observation INSERT. + // Endpoints for the live event only; stored rows resolve them at read time. var resolvedSource, resolvedDestination *api.ResolvedHop if packet.PayloadType() == meshcore.PayloadTypeAdvert && originPubkey != nil { // Exact match: ADVERT carries the sender's real identity pubkey, not a @@ -786,15 +785,6 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ resolvedDestination = &hop } } - var resolvedEndpoints json.RawMessage - snapshot := api.PacketEndpointSnapshot{Source: resolvedSource, Destination: resolvedDestination} - if snapshot.HasResolvedNodes() { - resolvedEndpoints, err = json.Marshal(snapshot) - if err != nil { - w.log.Error("endpoint snapshot encoding failed", "error", err) - resolvedEndpoints = nil // optional enrichment must not discard the observation - } - } // Airtime is costed from the frame as received; zero radio columns mean the // observer never reported its settings, so there is nothing to cost. var airtimeMs *float32 @@ -820,7 +810,6 @@ func (w *Worker) handlePacket(ctx context.Context, iata, pubkeyHex string, raw [ CodingRate: radio.CR, SourceBroker: w.cfg.BrokerName, PayloadType: int16(packet.PayloadType()), - ResolvedEndpoints: resolvedEndpoints, AirtimeMs: airtimeMs, } inserted, err := w.db.InsertObservation(ctx, oParams) diff --git a/internal/ingest/packet_endpoints_test.go b/internal/ingest/packet_endpoints_test.go index 61a47063..3611f6f2 100644 --- a/internal/ingest/packet_endpoints_test.go +++ b/internal/ingest/packet_endpoints_test.go @@ -5,118 +5,94 @@ package ingest import ( "context" - "encoding/hex" "encoding/json" - "errors" - "math" "reflect" "testing" + "time" "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/MeshCore-Beacon/beacon-server/internal/hub" "github.com/google/uuid" "github.com/meshcore-go/meshcore-go" ) type endpointCaptureDB struct { *stubDB - node api.ResolvedNode - observed []InsertObservationParams - lookups int - pathLookups int - missingNode bool + node api.ResolvedNode + observed int } func (s *endpointCaptureDB) GetNodeByPubkey(context.Context, []byte) (uuid.UUID, error) { - if s.missingNode { - return uuid.Nil, errors.New("node not yet advertised") - } return s.node.ID, nil } - func (s *endpointCaptureDB) UpsertNode(context.Context, UpsertNodeParams, RadioSettings) (uuid.UUID, error) { - s.missingNode = false return s.node.ID, nil } func (s *endpointCaptureDB) GetNodesByIDs(context.Context, []uuid.UUID) (map[uuid.UUID]*api.ResolvedNode, error) { return map[uuid.UUID]*api.ResolvedNode{s.node.ID: &s.node}, nil } -func (s *endpointCaptureDB) ResolvePathHashes(_ context.Context, _ string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) { - if len(hashes) > 0 { - s.pathLookups++ - } - return nil, nil // a companion must not be found by the relay-only resolver -} - -func (s *endpointCaptureDB) ResolveEndpointHashes(_ context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) { - if len(hashes) == 0 { - return nil, nil - } - s.lookups++ - if iata != "YYZ" || s.missingNode { - return nil, nil +func (s *endpointCaptureDB) ResolveEndpointHashes(_ context.Context, _ string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) { + if len(hashes) == 1 && hashes[0][0] == 0xaa { + return map[string][]api.ResolvedPathEntry{"aa": {{NodeID: s.node.ID, Name: s.node.Name, PublicKey: []byte{0xaa}}}}, nil } - return map[string][]api.ResolvedPathEntry{ - "aa": {{NodeID: s.node.ID, Name: s.node.Name, PublicKey: []byte{0xaa}}, {NodeID: uuid.Nil, PublicKey: []byte{0xab}}}, - }, nil + return nil, nil } - -func (s *endpointCaptureDB) InsertObservation(_ context.Context, observation InsertObservationParams) (bool, error) { - s.observed = append(s.observed, observation) +func (s *endpointCaptureDB) InsertObservation(context.Context, InsertObservationParams) (bool, error) { + s.observed++ return true, nil } -func TestHandlePacketCapturesEndpoints(t *testing.T) { - name := "Companion 👋" - for _, kind := range []string{"advert", "first advert", "direct message", "unresolved direct", "unaddressed", "encoding failure"} { +// Stored rows resolve endpoints at read time, but live events still carry them. +func TestHandlePacketBroadcastsEndpoints(t *testing.T) { + name := "Companion" + for _, kind := range []string{"advert", "direct message"} { t.Run(kind, func(t *testing.T) { w, base := newTestWorker() db := &endpointCaptureDB{stubDB: base, node: api.ResolvedNode{ID: uuid.New(), Name: &name, PublicKey: "aa"}} w.db = db + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + go w.hub.Run() + client := w.hub.NewClient() + w.hub.AddScope(client, "endpoints", hub.Scope{Events: []hub.EventType{hub.EventPacketObservation, hub.EventObserverStatus}}) + defer w.hub.Remove(client) + waitForSummarySubscriber(t, ctx, w.hub, client) + packet := buildAdvertPacket(t, false) - want := api.PacketEndpointSnapshot{} - switch kind { - case "advert": - hop := api.ResolveExactNode(&db.node) - want.Source = &hop - case "first advert": - db.missingNode = true - case "direct message", "unresolved direct": - // Destination, source, MAC and a minimal ciphertext envelope. + wantSource := api.ResolveExactNode(&db.node) + var wantDestination *api.ResolvedHop + if kind == "direct message" { packet = &meshcore.Packet{Header: meshcore.MakeHeader(meshcore.RouteTypeFlood, meshcore.PayloadTypeTxtMsg, 0), Payload: append([]byte{0xbb, 0xaa, 0, 0}, make([]byte, 16)...)} - db.missingNode = kind == "unresolved direct" - resolved, _ := db.ResolveEndpointHashes(context.Background(), "YYZ", [][]byte{{0xaa}}) - source := api.BuildResolvedPath([][]byte{{0xaa}}, resolved)[0] - destination := api.BuildResolvedPath([][]byte{{0xbb}}, nil)[0] - want.Source, want.Destination = &source, &destination - db.lookups = 0 - case "unaddressed": - packet = buildTracePacket(t) - case "encoding failure": - nan := math.NaN() - db.node.Latitude = &nan + wantSource = api.BuildResolvedPath([][]byte{{0xaa}}, map[string][]api.ResolvedPathEntry{"aa": {{NodeID: db.node.ID, Name: &name, PublicKey: []byte{0xaa}}}})[0] + d := api.BuildResolvedPath([][]byte{{0xbb}}, nil)[0] + wantDestination = &d } - w.handlePacket(context.Background(), "YYZ", hex.EncodeToString([]byte{1, 2}), packetEnvelope(t, packet)) - if len(db.observed) != 1 { - t.Fatalf("observation was lost or written twice: %d", len(db.observed)) + w.handlePacket(ctx, "YYZ", "0102", packetEnvelope(t, packet)) + if db.observed != 1 { + t.Fatalf("observation written %d times", db.observed) } - if kind == "first advert" || kind == "unresolved direct" || kind == "unaddressed" || kind == "encoding failure" { - if db.observed[0].ResolvedEndpoints != nil { - t.Fatal("empty or failed endpoint resolution must remain SQL NULL") - } - if kind == "first advert" && db.missingNode { - t.Fatal("advert side effects did not make the node available for a later lookup") + for { + select { + case event := <-client.Send: + if event.Type != hub.EventPacketObservation { + continue + } + var got struct { + Observation struct { + ResolvedSource *api.ResolvedHop `json:"resolvedSource"` + ResolvedDestination *api.ResolvedHop `json:"resolvedDestination"` + } `json:"observation"` + } + if err := json.Unmarshal(event.Payload, &got); err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(got.Observation.ResolvedSource, &wantSource) || !reflect.DeepEqual(got.Observation.ResolvedDestination, wantDestination) { + t.Fatalf("live endpoints: got %+v / %+v", got.Observation.ResolvedSource, got.Observation.ResolvedDestination) + } + return + case <-ctx.Done(): + t.Fatal("packet observation event missing") } - return - } - var got api.PacketEndpointSnapshot - if err := json.Unmarshal(db.observed[0].ResolvedEndpoints, &got); err != nil { - t.Fatal(err) - } - if !reflect.DeepEqual(got, want) { - t.Fatalf("captured endpoints differ: got %+v, want %+v", got, want) - } - if kind == "direct message" && (db.lookups != 2 || db.pathLookups != 0) { - t.Fatalf("wrong resolver or repeated work: endpoint=%d relay=%d", db.lookups, db.pathLookups) } }) }