diff --git a/db/packets.go b/db/packets.go index f806a42..2776625 100644 --- a/db/packets.go +++ b/db/packets.go @@ -187,14 +187,6 @@ 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{ @@ -223,14 +215,8 @@ 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 - // 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 + last := rows[len(rows)-1].SiteHeardAt.Time.UnixMilli() + nextCursor = &last } return api.Page[api.PacketSummary]{ Items: items, diff --git a/db/packets_test.go b/db/packets_test.go index f2c8248..9a4a9cb 100644 --- a/db/packets_test.go +++ b/db/packets_test.go @@ -527,107 +527,6 @@ 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 1e54000..22558c4 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -424,14 +424,29 @@ 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 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 +-- 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 FROM unnest(@iatas::bpchar[]) AS req(iata) CROSS JOIN LATERAL ( SELECT po3.packet_hash, po3.heard_at @@ -449,53 +464,15 @@ WITH scanned AS ( ORDER BY po3.heard_at DESC LIMIT @scan_depth ) hits -), --- 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 + GROUP BY hits.packet_hash HAVING (@cursor_ts::timestamptz IS NULL OR NOT EXISTS ( SELECT 1 FROM packet_observations px - WHERE px.packet_hash = scanned.packet_hash + WHERE px.packet_hash = hits.packet_hash AND px.iata = ANY(@iatas::bpchar[]) AND px.heard_at >= @cursor_ts)) ORDER BY site_heard_at DESC LIMIT @page_limit -) -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 +) sh 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 9ee115a..dd96785 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -147,14 +147,10 @@ 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 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. + // 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. ListPacketsByIATAs(ctx context.Context, arg ListPacketsByIATAsParams) ([]ListPacketsByIATAsRow, error) // ============================================================ // REGIONS diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 1fe4a9c..64f3d6b 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -2894,8 +2894,25 @@ func (q *Queries) ListPacketsAfterID(ctx context.Context, arg ListPacketsAfterID } const listPacketsByIATAs = `-- name: ListPacketsByIATAs :many -WITH scanned AS ( - SELECT req.iata AS req_iata, hits.packet_hash, hits.heard_at +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 FROM unnest($1::bpchar[]) AS req(iata) CROSS JOIN LATERAL ( SELECT po3.packet_hash, po3.heard_at @@ -2913,51 +2930,15 @@ WITH scanned AS ( ORDER BY po3.heard_at DESC LIMIT $8 ) hits -), -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 + GROUP BY hits.packet_hash HAVING ($2::timestamptz IS NULL OR NOT EXISTS ( SELECT 1 FROM packet_observations px - WHERE px.packet_hash = scanned.packet_hash + WHERE px.packet_hash = hits.packet_hash AND px.iata = ANY($1::bpchar[]) AND px.heard_at >= $2)) ORDER BY site_heard_at DESC LIMIT $9 -) -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 +) sh 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 @@ -2992,8 +2973,6 @@ 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"` @@ -3009,14 +2988,10 @@ 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 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. +// 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. func (q *Queries) ListPacketsByIATAs(ctx context.Context, arg ListPacketsByIATAsParams) ([]ListPacketsByIATAsRow, error) { rows, err := q.db.Query(ctx, listPacketsByIATAs, arg.Iatas, @@ -3045,8 +3020,6 @@ 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,