Skip to content

Revert "fix: IATA-filtered packet lists stop paginating early" #101

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
446564 merged 1 commit into from
Aug 20, 2026
Merged

Revert "fix: IATA-filtered packet lists stop paginating early" #101

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 2 additions & 16 deletions db/packets.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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{
Expand Down Expand Up @@ -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,
Expand Down
101 changes: 0 additions & 101 deletions db/packets_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
75 changes: 26 additions & 49 deletions db/queries/queries.sql
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down
12 changes: 4 additions & 8 deletions db/sqlc/querier.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

79 changes: 26 additions & 53 deletions db/sqlc/queries.sql.go
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading