From 69e7e196af372cc9399f3d54e65f95d022d7f892 Mon Sep 17 00:00:00 2001 From: n30nex Date: Sun, 6 Sep 2026 10:21:18 -0400 Subject: [PATCH] fix(db): tolerate packets before their first observation (#115) --- db/packets_integration_test.go | 79 ++++++++++++++++++++++++++++++++++ db/queries/queries.sql | 18 ++++---- db/sqlc/queries.sql.go | 18 ++++---- 3 files changed, 99 insertions(+), 16 deletions(-) create mode 100644 db/packets_integration_test.go diff --git a/db/packets_integration_test.go b/db/packets_integration_test.go new file mode 100644 index 0000000..10521e3 --- /dev/null +++ b/db/packets_integration_test.go @@ -0,0 +1,79 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "context" + "os" + "testing" + "time" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/jackc/pgx/v5" +) + +// The packet and its observation are separate ingest writes. A list request +// between them must return the packet without inventing an observer or failing. +func TestListPacketsBeforeObservationPostgres(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) + defer cancel() + conn, err := pgx.Connect(ctx, dsn) + if err != nil { + t.Fatal(err) + } + defer conn.Close(context.Background()) + tx, err := conn.Begin(ctx) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(context.Background()) + for _, table := range []string{"packets", "packet_observations", "observers", "transport_scopes"} { + if _, err := tx.Exec(ctx, "CREATE TEMP TABLE "+table+" (LIKE public."+table+" INCLUDING ALL) ON COMMIT DROP"); err != nil { + t.Fatal(err) + } + } + _, err = tx.Exec(ctx, `INSERT INTO packets (packet_hash, payload_type, payload_version, route_type, raw_payload, raw_header, first_heard_at, last_heard_at) +VALUES ('\x01', 4, 0, 1, '\x00', '\x00', NOW(), NOW())`) + if err != nil { + t.Fatal(err) + } + store := &Store{q: sqlc.New(tx)} + page, err := store.ListPackets(ctx, nil, nil, nil, nil, time.Time{}, time.Time{}, 0, 50) + if err != nil { + t.Fatalf("packet without observation must remain readable: %v", err) + } + if len(page.Items) != 1 || page.Items[0].PacketHash != "01" { + t.Fatalf("packet missing: %+v", page) + } + if page.Items[0].LatestObserver != nil || page.Items[0].ObservationCount != 0 { + t.Fatalf("invented observation: %+v", page.Items[0]) + } + _, err = tx.Exec(ctx, ` +INSERT INTO observers (id, public_key) VALUES ('00000000-0000-0000-0000-000000000001', '\x02'); +INSERT INTO packet_observations (id, packet_hash, observer_id, iata, heard_at, path_length_byte, hash_size, hop_count, path_bytes) +VALUES (1, '\x01', '00000000-0000-0000-0000-000000000001', 'YVR', NOW(), 2, 1, 2, '\x1122');`) + if err != nil { + t.Fatal(err) + } + for _, iatas := range [][]string{nil, {"YVR"}} { + page, err := store.ListPackets(ctx, nil, nil, iatas, nil, time.Time{}, time.Time{}, 0, 50) + if err != nil { + t.Fatal(err) + } + if len(page.Items) != 1 || page.Items[0].ObservationCount != 1 { + t.Fatalf("observed packet missing for %v: %+v", iatas, page) + } + observer := page.Items[0].LatestObserver + if observer == nil || observer.IATA != "YVR" || observer.PathLength == nil || observer.PathBytes == nil { + t.Fatalf("observation metadata missing: %+v", observer) + } + if observer.PathLength.Raw != "02" || observer.PathLength.HashSize != 1 || observer.PathLength.HopCount != 2 || *observer.PathBytes != "1122" { + t.Fatalf("observation path changed: %+v", observer) + } + } +} diff --git a/db/queries/queries.sql b/db/queries/queries.sql index b6213db..f637b29 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -391,12 +391,13 @@ SELECT p.scope_id, ts.name AS scope_name, (SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = p.packet_hash) AS observation_count, + -- sqlc loses LATERAL nullability; these scalar defaults are ignored when observer_id is NULL. po.observer_id AS latest_observer_id, o.display_name AS latest_observer_name, - po.iata AS latest_observer_iata, - po.path_length_byte AS latest_observer_path_length_byte, - po.hash_size AS latest_observer_hash_size, - po.hop_count AS latest_observer_hop_count, + COALESCE(po.iata, ''::bpchar) AS latest_observer_iata, + COALESCE(po.path_length_byte, 0::smallint) AS latest_observer_path_length_byte, + 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 FROM packets p LEFT JOIN LATERAL ( @@ -487,12 +488,13 @@ SELECT sat.scan_saturated, sat.scan_floor, (SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = p.packet_hash) AS observation_count, + -- sqlc loses LATERAL nullability; these scalar defaults are ignored when observer_id is NULL. po.observer_id AS latest_observer_id, o.display_name AS latest_observer_name, - po.iata AS latest_observer_iata, - po.path_length_byte AS latest_observer_path_length_byte, - po.hash_size AS latest_observer_hash_size, - po.hop_count AS latest_observer_hop_count, + COALESCE(po.iata, ''::bpchar) AS latest_observer_iata, + COALESCE(po.path_length_byte, 0::smallint) AS latest_observer_path_length_byte, + 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 FROM page sh CROSS JOIN saturation sat diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 670f7ad..7fe72e4 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -2694,12 +2694,13 @@ SELECT p.scope_id, ts.name AS scope_name, (SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = p.packet_hash) AS observation_count, + -- sqlc loses LATERAL nullability; these scalar defaults are ignored when observer_id is NULL. po.observer_id AS latest_observer_id, o.display_name AS latest_observer_name, - po.iata AS latest_observer_iata, - po.path_length_byte AS latest_observer_path_length_byte, - po.hash_size AS latest_observer_hash_size, - po.hop_count AS latest_observer_hop_count, + COALESCE(po.iata, ''::bpchar) AS latest_observer_iata, + COALESCE(po.path_length_byte, 0::smallint) AS latest_observer_path_length_byte, + 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 FROM packets p LEFT JOIN LATERAL ( @@ -2952,12 +2953,13 @@ SELECT sat.scan_saturated, sat.scan_floor, (SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = p.packet_hash) AS observation_count, + -- sqlc loses LATERAL nullability; these scalar defaults are ignored when observer_id is NULL. po.observer_id AS latest_observer_id, o.display_name AS latest_observer_name, - po.iata AS latest_observer_iata, - po.path_length_byte AS latest_observer_path_length_byte, - po.hash_size AS latest_observer_hash_size, - po.hop_count AS latest_observer_hop_count, + COALESCE(po.iata, ''::bpchar) AS latest_observer_iata, + COALESCE(po.path_length_byte, 0::smallint) AS latest_observer_path_length_byte, + 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 FROM page sh CROSS JOIN saturation sat