From fba9515757271a7f1269a4e48e4bac3aeaf8e01d Mon Sep 17 00:00:00 2001 From: MrAlders0n Date: Fri, 25 Sep 2026 13:44:26 -0400 Subject: [PATCH] perf(db): resolve packet endpoints at read time Storing a full candidate snapshot on every observation was the wrong call. The same node lists got copied thousands of times, about 22 GB on prod. Resolve endpoints live like path hops and drop the column. --- .github/workflows/ci.yml | 4 +- db/endpoints.go | 151 +++++++++ db/endpoints_test.go | 140 +++++++++ db/migrations/038_drop_resolved_endpoints.sql | 6 + db/packet_endpoints_integration_test.go | 287 ++++++------------ db/packets.go | 126 ++------ db/queries/queries.sql | 34 ++- db/sqlc/mock/querier.go | 30 ++ db/sqlc/models.go | 1 - db/sqlc/querier.go | 4 + db/sqlc/queries.sql.go | 240 ++++++++++----- internal/api/packets.go | 15 - internal/api/packets_test.go | 22 -- internal/ingest/packet.go | 13 +- internal/ingest/packet_endpoints_test.go | 128 ++++---- 15 files changed, 713 insertions(+), 488 deletions(-) create mode 100644 db/endpoints.go create mode 100644 db/endpoints_test.go create mode 100644 db/migrations/038_drop_resolved_endpoints.sql 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) } }) }