fix: keep IATA packet paging alive past short pages

A packet repeats once per observer in the site scan, so grouping can collapse a scan_depth window to fewer packets than the page asked for. That short page set hasMore=false and stranded all older history: YOW dead-ended after 45 packets, ~6 minutes back. Report scan saturation and the floor the scans covered, and treat a saturated short page as more data. Clamp the cursor to the floor so paging on cannot skip the band the scan never read.
2026-09-09 21:45:35 +00:00 · 2026-08-19 16:21:26 -07:00
parent 7d759d0a0f
commit ddc6dc1ca9
5 changed files with 227 additions and 58 deletions
@@ -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,
@@ -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)
@@ -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
@@ -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
@@ -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,