diff --git a/db/packets.go b/db/packets.go index 2776625..f806a42 100644 --- a/db/packets.go +++ b/db/packets.go @@ -187,6 +187,14 @@ func (s *Store) listPacketsByIATAs(ctx context.Context, payloadTypes, routeTypes if hasMore { rows = rows[:limit] } + // Observers duplicate a packet across the scan, so the page can collapse + // short while the site still has history below scan_floor. Stopping here + // would strand every older packet behind a hasMore=false. + var scanFloor pgtype.Timestamptz + if len(rows) > 0 && rows[0].ScanSaturated { + hasMore = true + scanFloor = rows[0].ScanFloor + } items := make([]api.PacketSummary, 0, len(rows)) for _, v := range rows { item := api.PacketSummary{ @@ -215,8 +223,14 @@ func (s *Store) listPacketsByIATAs(ctx context.Context, payloadTypes, routeTypes // Cursor follows site-local recency, not the packet's global last_heard_at. var nextCursor *int64 if hasMore && len(rows) > 0 { - last := rows[len(rows)-1].SiteHeardAt.Time.UnixMilli() - nextCursor = &last + last := rows[len(rows)-1].SiteHeardAt.Time + // A saturated scan never read below its floor. Paging past it would + // skip that band; resuming at it only repeats packets already shown. + if scanFloor.Valid && scanFloor.Time.After(last) { + last = scanFloor.Time + } + ms := last.UnixMilli() + nextCursor = &ms } return api.Page[api.PacketSummary]{ Items: items, diff --git a/db/packets_test.go b/db/packets_test.go index 9a4a9cb..f2c8248 100644 --- a/db/packets_test.go +++ b/db/packets_test.go @@ -527,6 +527,107 @@ func TestListPackets_IATAFilterRoutesToObservationIndex(t *testing.T) { } } +// A site whose packets are heard by more observers than scan_depth allows +// for collapses to a short page while history remains below the floor. +// The page must still report more, or the client stops paging for good. +func TestListPackets_SaturatedShortPageKeepsPaging(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + + oldest := time.Date(2026, 8, 7, 12, 57, 25, 0, time.UTC) + floor := time.Date(2026, 8, 7, 12, 50, 0, 0, time.UTC) + + // Two rows for a limit of 5: the scan filled up but collapsed to a short page. + mock.EXPECT(). + ListPacketsByIATAs(gomock.Any(), gomock.Any()). + Return([]sqlc.ListPacketsByIATAsRow{ + { + PacketHash: []byte{0x01}, + SiteHeardAt: pgtype.Timestamptz{Time: oldest.Add(time.Minute), Valid: true}, + ScanSaturated: true, + ScanFloor: pgtype.Timestamptz{Time: floor, Valid: true}, + }, + { + PacketHash: []byte{0x02}, + SiteHeardAt: pgtype.Timestamptz{Time: oldest, Valid: true}, + ScanSaturated: true, + ScanFloor: pgtype.Timestamptz{Time: floor, Valid: true}, + }, + }, nil) + + store := &Store{q: mock} + page, err := store.ListPackets(context.Background(), nil, nil, []string{"YOW"}, nil, time.Time{}, time.Time{}, 0, 5) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if len(page.Items) != 2 { + t.Fatalf("got %d items, want 2", len(page.Items)) + } + if !page.HasMore { + t.Error("hasMore = false on a saturated short page, want true") + } + // Oldest returned item sits above the floor, so it is the safe cursor. + if page.NextCursor == nil || *page.NextCursor != oldest.UnixMilli() { + t.Errorf("next cursor = %v, want %d (oldest item)", page.NextCursor, oldest.UnixMilli()) + } +} + +// The floor is newer than the oldest item, so paging past it would skip the +// band the saturated site never read. +func TestListPackets_CursorClampsToScanFloor(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + + floor := time.Date(2026, 8, 7, 12, 50, 0, 0, time.UTC) + oldest := floor.Add(-30 * time.Minute) + + mock.EXPECT(). + ListPacketsByIATAs(gomock.Any(), gomock.Any()). + Return([]sqlc.ListPacketsByIATAsRow{{ + PacketHash: []byte{0x01}, + SiteHeardAt: pgtype.Timestamptz{Time: oldest, Valid: true}, + ScanSaturated: true, + ScanFloor: pgtype.Timestamptz{Time: floor, Valid: true}, + }}, nil) + + store := &Store{q: mock} + page, err := store.ListPackets(context.Background(), nil, nil, []string{"YOW", "YYZ"}, nil, time.Time{}, time.Time{}, 0, 5) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if page.NextCursor == nil || *page.NextCursor != floor.UnixMilli() { + t.Errorf("next cursor = %v, want %d (clamped to floor)", page.NextCursor, floor.UnixMilli()) + } +} + +// An unsaturated scan read the site dry, so paging has to stop. +func TestListPackets_UnsaturatedShortPageEndsPaging(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + + oldest := time.Date(2026, 8, 7, 12, 57, 25, 0, time.UTC) + + mock.EXPECT(). + ListPacketsByIATAs(gomock.Any(), gomock.Any()). + Return([]sqlc.ListPacketsByIATAsRow{{ + PacketHash: []byte{0x01}, + SiteHeardAt: pgtype.Timestamptz{Time: oldest, Valid: true}, + ScanSaturated: false, + }}, nil) + + store := &Store{q: mock} + page, err := store.ListPackets(context.Background(), nil, nil, []string{"YOW"}, nil, time.Time{}, time.Time{}, 0, 5) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if page.HasMore { + t.Error("hasMore = true on an exhausted site, want false") + } + if page.NextCursor != nil { + t.Errorf("next cursor = %v, want nil", page.NextCursor) + } +} + func TestListPackets_UnfilteredKeepsGlobalQuery(t *testing.T) { ctrl := gomock.NewController(t) mock := mockdb.NewMockQuerier(ctrl) diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 22558c4..1e54000 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -424,29 +424,14 @@ LIMIT $6; -- to fill a page for a quiet site; walking the site's own observation log is -- proportional to the page size instead. Results are ordered by when the -- requested sites heard the packet (site-local recency) and the cursor --- follows that ordering. scan_depth is a multiple of the page size to --- absorb per-observer duplicates; if duplication exceeds it across a --- page, pagination ends early (hasMore=false) rather than returning a --- short page, even though deeper matches exist. -SELECT - p.packet_hash, - p.payload_type, - p.route_type, - p.first_heard_at, - p.last_heard_at, - p.scope_id, - ts.name AS scope_name, - sh.site_heard_at, - (SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = p.packet_hash) AS observation_count, - 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, - po.path_bytes AS latest_observer_path_bytes -FROM ( - SELECT hits.packet_hash, MAX(hits.heard_at)::timestamptz AS site_heard_at +-- follows that ordering. scan_depth caps how deep each site's observation +-- log is walked. A packet repeats once per observer that heard it, so a +-- page can collapse to fewer distinct packets than were asked for without +-- the site being exhausted. scan_saturated reports whether any site hit +-- that cap and scan_floor the oldest heard_at they all cover, so a short +-- page can keep paging instead of reading as the end of the data. +WITH scanned AS ( + SELECT req.iata AS req_iata, hits.packet_hash, hits.heard_at FROM unnest(@iatas::bpchar[]) AS req(iata) CROSS JOIN LATERAL ( SELECT po3.packet_hash, po3.heard_at @@ -464,15 +449,53 @@ FROM ( ORDER BY po3.heard_at DESC LIMIT @scan_depth ) hits - GROUP BY hits.packet_hash +), +-- A site that filled scan_depth still has unread history below its floor. +-- The newest such floor is the point above which every site is covered. +saturation AS ( + SELECT + COUNT(*) > 0 AS scan_saturated, + MAX(floor_ts)::timestamptz AS scan_floor + FROM ( + SELECT MIN(heard_at) AS floor_ts + FROM scanned + GROUP BY req_iata + HAVING COUNT(*) >= @scan_depth + ) filled +), +page AS ( + SELECT scanned.packet_hash, MAX(scanned.heard_at)::timestamptz AS site_heard_at + FROM scanned + GROUP BY scanned.packet_hash HAVING (@cursor_ts::timestamptz IS NULL OR NOT EXISTS ( SELECT 1 FROM packet_observations px - WHERE px.packet_hash = hits.packet_hash + WHERE px.packet_hash = scanned.packet_hash AND px.iata = ANY(@iatas::bpchar[]) AND px.heard_at >= @cursor_ts)) ORDER BY site_heard_at DESC LIMIT @page_limit -) sh +) +SELECT + p.packet_hash, + p.payload_type, + p.route_type, + p.first_heard_at, + p.last_heard_at, + p.scope_id, + ts.name AS scope_name, + sh.site_heard_at, + sat.scan_saturated, + sat.scan_floor, + (SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = p.packet_hash) AS observation_count, + 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, + po.path_bytes AS latest_observer_path_bytes +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 diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index dd96785..9ee115a 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -147,10 +147,14 @@ type Querier interface { // to fill a page for a quiet site; walking the site's own observation log is // proportional to the page size instead. Results are ordered by when the // requested sites heard the packet (site-local recency) and the cursor - // follows that ordering. scan_depth is a multiple of the page size to - // absorb per-observer duplicates; if duplication exceeds it across a - // page, pagination ends early (hasMore=false) rather than returning a - // short page, even though deeper matches exist. + // follows that ordering. scan_depth caps how deep each site's observation + // log is walked. A packet repeats once per observer that heard it, so a + // page can collapse to fewer distinct packets than were asked for without + // the site being exhausted. scan_saturated reports whether any site hit + // that cap and scan_floor the oldest heard_at they all cover, so a short + // page can keep paging instead of reading as the end of the data. + // A site that filled scan_depth still has unread history below its floor. + // The newest such floor is the point above which every site is covered. ListPacketsByIATAs(ctx context.Context, arg ListPacketsByIATAsParams) ([]ListPacketsByIATAsRow, error) // ============================================================ // REGIONS diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 64f3d6b..1fe4a9c 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -2894,25 +2894,8 @@ func (q *Queries) ListPacketsAfterID(ctx context.Context, arg ListPacketsAfterID } const listPacketsByIATAs = `-- name: ListPacketsByIATAs :many -SELECT - p.packet_hash, - p.payload_type, - p.route_type, - p.first_heard_at, - p.last_heard_at, - p.scope_id, - ts.name AS scope_name, - sh.site_heard_at, - (SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = p.packet_hash) AS observation_count, - 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, - po.path_bytes AS latest_observer_path_bytes -FROM ( - SELECT hits.packet_hash, MAX(hits.heard_at)::timestamptz AS site_heard_at +WITH scanned AS ( + SELECT req.iata AS req_iata, hits.packet_hash, hits.heard_at FROM unnest($1::bpchar[]) AS req(iata) CROSS JOIN LATERAL ( SELECT po3.packet_hash, po3.heard_at @@ -2930,15 +2913,51 @@ FROM ( ORDER BY po3.heard_at DESC LIMIT $8 ) hits - GROUP BY hits.packet_hash +), +saturation AS ( + SELECT + COUNT(*) > 0 AS scan_saturated, + MAX(floor_ts)::timestamptz AS scan_floor + FROM ( + SELECT MIN(heard_at) AS floor_ts + FROM scanned + GROUP BY req_iata + HAVING COUNT(*) >= $8 + ) filled +), +page AS ( + SELECT scanned.packet_hash, MAX(scanned.heard_at)::timestamptz AS site_heard_at + FROM scanned + GROUP BY scanned.packet_hash HAVING ($2::timestamptz IS NULL OR NOT EXISTS ( SELECT 1 FROM packet_observations px - WHERE px.packet_hash = hits.packet_hash + WHERE px.packet_hash = scanned.packet_hash AND px.iata = ANY($1::bpchar[]) AND px.heard_at >= $2)) ORDER BY site_heard_at DESC LIMIT $9 -) sh +) +SELECT + p.packet_hash, + p.payload_type, + p.route_type, + p.first_heard_at, + p.last_heard_at, + p.scope_id, + ts.name AS scope_name, + sh.site_heard_at, + sat.scan_saturated, + sat.scan_floor, + (SELECT COUNT(*) FROM packet_observations po2 WHERE po2.packet_hash = p.packet_hash) AS observation_count, + 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, + po.path_bytes AS latest_observer_path_bytes +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 @@ -2973,6 +2992,8 @@ type ListPacketsByIATAsRow struct { 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"` @@ -2988,10 +3009,14 @@ type ListPacketsByIATAsRow struct { // to fill a page for a quiet site; walking the site's own observation log is // proportional to the page size instead. Results are ordered by when the // requested sites heard the packet (site-local recency) and the cursor -// follows that ordering. scan_depth is a multiple of the page size to -// absorb per-observer duplicates; if duplication exceeds it across a -// page, pagination ends early (hasMore=false) rather than returning a -// short page, even though deeper matches exist. +// follows that ordering. scan_depth caps how deep each site's observation +// log is walked. A packet repeats once per observer that heard it, so a +// page can collapse to fewer distinct packets than were asked for without +// the site being exhausted. scan_saturated reports whether any site hit +// that cap and scan_floor the oldest heard_at they all cover, so a short +// page can keep paging instead of reading as the end of the data. +// A site that filled scan_depth still has unread history below its floor. +// The newest such floor is the point above which every site is covered. func (q *Queries) ListPacketsByIATAs(ctx context.Context, arg ListPacketsByIATAsParams) ([]ListPacketsByIATAsRow, error) { rows, err := q.db.Query(ctx, listPacketsByIATAs, arg.Iatas, @@ -3020,6 +3045,8 @@ func (q *Queries) ListPacketsByIATAs(ctx context.Context, arg ListPacketsByIATAs &i.ScopeID, &i.ScopeName, &i.SiteHeardAt, + &i.ScanSaturated, + &i.ScanFloor, &i.ObservationCount, &i.LatestObserverID, &i.LatestObserverName, diff --git a/go.mod b/go.mod index 6f981fa..ac5e38c 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/MeshCore-Beacon/beacon-server -go 1.26.5 +go 1.26.6 require ( github.com/alicebob/miniredis/v2 v2.38.0