Slice the packet_sites fill by observations and count progress in rows (#274)

* Slice the packet_sites fill by observations and count progress in rows

* Keep one set of id statistics per fill; count new ids one row each

.github/workflows/ci.yml

@@ -69,7 +69,7 @@ jobs:
       - name: Verify stats and endpoint queries against PostgreSQL 16
         env:
           BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable
-        run: go test ./db -run '^Test(Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|NodeLocationReset|RunMigrations[A-Za-z]*|KnownRoutesIndexSet|PacketsDomainChecks|Reconfirm[A-Za-z]*|BigintCursors|RouteSearchHopOrder|StatsNodeTypesSplit|ListKnownRoutes|ListKnownRoutesMatchesLegacy|KnownRoutesCursorTies|KnownRoutesUnanalyzedRegion|KnownRoutesHopStatementsStayCustom|KnownRoutePaginationTies|RouteRetentionCutoffsExact|DeleteExpiredRoutes|DeleteGraceRoutesSlice|GraceWatermarkHeldByLockedRow|OpenPoolsSettings|PlanRouterClassCUnnamed|PlanRouterParallelWorkers|APIStoreHopStatementsUnnamed|PgStatStatementsMigrationGuarded|DBStatements|DBStatementsNoExtension|PgStatStatementsMigrationNoCreatableSchema|IndexHealthBaselineOnReindex|UnusedIndexes|PgActivity|RollHourSetsWorkMem|RollupWritesKeyOrder|SmallTableIndexSet|NodeReadersAfterIndexDrops|ObserverChannelReadersAfterIndexDrops|UpsertChannelWriteThrough|UpsertChannelHashOnlyWriteThrough|ReadOnlyPoolsRejectWrites|ListPacketsGlobalMatchesLegacy|PacketsGlobalCursorTies|ListPacketsWindowInvertedRows|PacketsLatestObserverTie|NodesLegacyCursorMatchesToday|NodesPageCursorTies|ObserversLegacyCursorMatchesToday|ObserversNextCursorFromLastSeen|ObserverAdvertsPartialIndex|ObserverComparisonMatchesToday|AdvertsWaitForPayloadTypeBackfill|StatsProbeMatchesToday|TraceTagsKeysetMatchesToday|TelemetrySinceMatchesCatchAll|TraceTagPagination|TraceTagPaginationTies|TraceSummarySurvivesMissedHearing|CmiMigrationsIndexSet|InsertChannelMessageSeed[A-Za-z]*|AddChannelMessageIATA|CmiSeedAndAddRace|CmiInitAndBackfill|CmiMessageRangeReconcile|CmiObservationRangeReconcile|CmiBackfillRetries23503|WaitForPreBootWriters|ChannelMessagesByIATAsMatchesLegacy|ChannelMessagesZeroHitRegion|CmiTrustLifecycle|CmiFirstVerification|PacketSitesMigrationsIndexSet|InsertObservationMaintainsPacketSites|InsertObservationResendWritesNothing|PacketSitesRaceMaxMerge|InsertObservationSiteFallback|PacketSitesInitAndBackfill|FillPacketSitesSliceIdempotent|FillPacketSitesSliceRetries23503|PacketSitesReconcile|PacketSitesAudit|PacketSitesAuditTakesNewestHearing|PacketsBySitesMatchesTruth|PacketsBySitesCursorTies|PacketsBySitesManyBogusSites|PacketsBySitesSaturatedPagesFull|PacketsBySitesMatchesLegacy|PacketSitesTrustLifecycle|PacketsBySitesBounded|PacketsBySitesFallbackKeepsPageCursor|SweepPacketSitesFrontsPlan|IndexHealthPacketSitesFreshBaseline|PacketSitesFirstVerification|PacedBackfillSlices|FillSessionsPlanCustom|FillAnalyzeFailureIsIgnored|FillAnalyzeOnlyNeverAnalyzed|HygieneIndexSet|HygieneReadersAfterIndexDrops|KnownRoutesColumns|SequenceUse|SlowPoolIsolation|SlowPoolSize|SequenceUseMixedCaseSchema|ObservationsIndexSet|ReadersAfterIndexDrops)Postgres$' -count=1 -v
+        run: go test ./db -run '^Test(Signal|Paths|PacketSummaries|PacketEndpointsResolveLive|ObserverMetrics|RouteEvidence|RouteEvidenceIndex|AnalyticsRetention|AnalyticsRetentionConcurrent|DeleteOldPacketsBatches|MeshMapperCatalogue|ChannelMessageScopes|NodeLocationReset|RunMigrations[A-Za-z]*|KnownRoutesIndexSet|PacketsDomainChecks|Reconfirm[A-Za-z]*|BigintCursors|RouteSearchHopOrder|StatsNodeTypesSplit|ListKnownRoutes|ListKnownRoutesMatchesLegacy|KnownRoutesCursorTies|KnownRoutesUnanalyzedRegion|KnownRoutesHopStatementsStayCustom|KnownRoutePaginationTies|RouteRetentionCutoffsExact|DeleteExpiredRoutes|DeleteGraceRoutesSlice|GraceWatermarkHeldByLockedRow|OpenPoolsSettings|PlanRouterClassCUnnamed|PlanRouterParallelWorkers|APIStoreHopStatementsUnnamed|PgStatStatementsMigrationGuarded|DBStatements|DBStatementsNoExtension|PgStatStatementsMigrationNoCreatableSchema|IndexHealthBaselineOnReindex|UnusedIndexes|PgActivity|RollHourSetsWorkMem|RollupWritesKeyOrder|SmallTableIndexSet|NodeReadersAfterIndexDrops|ObserverChannelReadersAfterIndexDrops|UpsertChannelWriteThrough|UpsertChannelHashOnlyWriteThrough|ReadOnlyPoolsRejectWrites|ListPacketsGlobalMatchesLegacy|PacketsGlobalCursorTies|ListPacketsWindowInvertedRows|PacketsLatestObserverTie|NodesLegacyCursorMatchesToday|NodesPageCursorTies|ObserversLegacyCursorMatchesToday|ObserversNextCursorFromLastSeen|ObserverAdvertsPartialIndex|ObserverComparisonMatchesToday|AdvertsWaitForPayloadTypeBackfill|StatsProbeMatchesToday|TraceTagsKeysetMatchesToday|TelemetrySinceMatchesCatchAll|TraceTagPagination|TraceTagPaginationTies|TraceSummarySurvivesMissedHearing|CmiMigrationsIndexSet|InsertChannelMessageSeed[A-Za-z]*|AddChannelMessageIATA|CmiSeedAndAddRace|CmiInitAndBackfill|CmiMessageRangeReconcile|CmiObservationRangeReconcile|CmiBackfillRetries23503|WaitForPreBootWriters|ChannelMessagesByIATAsMatchesLegacy|ChannelMessagesZeroHitRegion|CmiTrustLifecycle|CmiFirstVerification|PacketSitesMigrationsIndexSet|InsertObservationMaintainsPacketSites|InsertObservationResendWritesNothing|PacketSitesRaceMaxMerge|InsertObservationSiteFallback|PacketSitesInitAndBackfill|FillPacketSitesSliceIdempotent|FillPacketSitesSliceRetries23503|PacketSitesReconcile|PacketSitesAudit|PacketSitesAuditTakesNewestHearing|PacketsBySitesMatchesTruth|PacketsBySitesCursorTies|PacketsBySitesManyBogusSites|PacketsBySitesSaturatedPagesFull|PacketsBySitesMatchesLegacy|PacketSitesTrustLifecycle|PacketsBySitesBounded|PacketsBySitesFallbackKeepsPageCursor|SweepPacketSitesFrontsPlan|IndexHealthPacketSitesFreshBaseline|PacketSitesFirstVerification|PacketSitesBackfillSlicesSparseIDs|PacketSitesBackfillResumesAcrossGaps|PacketSitesProgressCountsRows|PacedBackfillSlices|FillSessionsPlanCustom|FillAnalyzeFailureIsIgnored|FillAnalyzeOnlyNeverAnalyzed|HygieneIndexSet|HygieneReadersAfterIndexDrops|KnownRoutesColumns|SequenceUse|SlowPoolIsolation|SlowPoolSize|SequenceUseMixedCaseSchema|ObservationsIndexSet|ReadersAfterIndexDrops)Postgres$' -count=1 -v
 
       - name: Verify the channel message site rule against PostgreSQL 16
         env:

config.yaml.example

@@ -100,7 +100,7 @@ ratelimit:
 #  fill:
 #    duty: 0.5                  # share of time a fill may be busy, 0.01 to 1; 0.25 on a slow (rotational) disk
 #    min_pause: 100ms           # shortest pause after a slice or batch, 0s to 1m
-#    packet_sites_slice: 500000 # observation ids per packet_sites backfill slice, 1 to 10000000
+#    packet_sites_slice: 500000 # observations per packet_sites backfill slice, 1 to 10000000
 #    cmi_batch: 10000           # message ids per cmi backfill batch, 1 to 1000000
 
 # MeshMapper integrations (optional, require a regional or grouped-region API key).

db/fill_estimate.go

@@ -0,0 +1,75 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package db
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v5"
+)
+
+// idHistogram is the planner's view of a table's ids: the equal-frequency histogram of a unique id column and the
+// table's estimated row count. A fill uses it to report progress in rows rather than ids, so an id range that
+// retention emptied long ago counts for as little as it costs.
+type idHistogram struct {
+	bounds []int64 // ascending; each adjacent pair holds the same share of rows
+	rows   float64
+}
+
+// loadObservationIDHistogram reads packet_observations' id statistics. Before the first ANALYZE there are none:
+// the histogram is empty and rows reports !ok.
+func loadObservationIDHistogram(ctx context.Context, conn interface {
+	QueryRow(context.Context, string, ...any) pgx.Row
+}) (idHistogram, error) {
+	var h idHistogram
+	err := conn.QueryRow(ctx, `SELECT
+  COALESCE((SELECT s.histogram_bounds::text::bigint[] FROM pg_stats s
+            WHERE s.schemaname = current_schema() AND s.tablename = 'packet_observations' AND s.attname = 'id'),
+           '{}'::bigint[]),
+  COALESCE((SELECT c.reltuples FROM pg_class c WHERE c.oid = 'packet_observations'::regclass), 0)::float8`).Scan(&h.bounds, &h.rows)
+	return h, err
+}
+
+// below is the share of rows with an id at or below x, interpolating linearly inside a bucket.
+func (h idHistogram) below(x int64) float64 {
+	b := h.bounds
+	n := len(b) - 1
+	switch {
+	case x < b[0]:
+		return 0
+	case x >= b[n]:
+		return 1
+	}
+	i := 0
+	for i < n-1 && x >= b[i+1] {
+		i++
+	}
+	within := 1.0
+	if b[i+1] > b[i] {
+		within = float64(x-b[i]) / float64(b[i+1]-b[i])
+	}
+	return (float64(i) + within) / float64(n)
+}
+
+// estimate is the estimated number of rows with ids in (after, through]; ok is false without statistics. Ids above
+// the histogram's last bound arrived since the last ANALYZE and are new, so dense: each counts as one row.
+func (h idHistogram) estimate(after, through int64) (int64, bool) {
+	if len(h.bounds) < 2 || h.rows <= 0 {
+		return 0, false
+	}
+	if through <= after {
+		return 0, true
+	}
+	last := h.bounds[len(h.bounds)-1]
+	in := h.rows * (h.below(min(through, last)) - h.below(min(after, last)))
+	return int64(in) + max(through-max(after, last), 0), true
+}
+
+// rowsOrIDs is estimate, or the id count when there are no statistics.
+func (h idHistogram) rowsOrIDs(after, through int64) int64 {
+	if n, ok := h.estimate(after, through); ok {
+		return n
+	}
+	return max(through-after, 0)
+}

db/fill_estimate_test.go

@@ -0,0 +1,66 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package db
+
+import (
+	"testing"
+	"time"
+)
+
+func TestIDHistogramEstimate(t *testing.T) {
+	// 1,000 rows in 4 equal buckets: 250 in (0, 100], 250 in (100, 1,000,000] (a sparse old range), 500 in the dense
+	// top ids.
+	h := idHistogram{bounds: []int64{1, 100, 1_000_000, 1_000_100, 1_000_200}, rows: 1000}
+	for _, c := range []struct {
+		after, through, want int64
+	}{
+		{0, 1_000_200, 1000},
+		{1_000_000, 1_000_200, 500},
+		{100, 1_000_000, 250},
+		{1_000_100, 1_000_150, 125},
+		{500_000, 1_000_000, 125},
+		{1_000_200, 1_000_300, 100}, // above the histogram: new since ANALYZE, one row an id
+		{1_000_150, 1_000_250, 175}, // 125 inside it and 50 above
+		{-5, 0, 0},                  // below it
+		{50, 50, 0},
+	} {
+		if got, ok := h.estimate(c.after, c.through); !ok || got != c.want {
+			t.Errorf("estimate(%d, %d) = %d, %v; want %d", c.after, c.through, got, ok, c.want)
+		}
+	}
+	if _, ok := (idHistogram{}).estimate(0, 10); ok {
+		t.Error("an empty histogram estimated rows")
+	}
+	if got := (idHistogram{}).rowsOrIDs(10, 25); got != 15 {
+		t.Errorf("rowsOrIDs without statistics = %d, want the id count 15", got)
+	}
+}
+
+// On ids like production's, a few old rows far below a dense live range, an ETA in rows follows the work left
+// where one in ids counts the empty ids as work.
+func TestIDHistogramETAFollowsRows(t *testing.T) {
+	// 100 rows spread over ids (0, 770M], then 10M dense rows in (770M, 780M].
+	bounds := []int64{1, 770_000_000}
+	for id := int64(770_100_000); id <= 780_000_000; id += 100_000 {
+		bounds = append(bounds, id)
+	}
+	h := idHistogram{bounds: bounds, rows: float64(len(bounds)-1) * 100_000}
+	writer, pos := int64(780_000_000), int64(775_000_000) // half the dense rows filled in an hour
+	var p FillProgress
+	start := time.Unix(0, 0)
+	p.now = func() time.Time { return start }
+	p.Start(h.rowsOrIDs(writer, writer))
+	p.now = func() time.Time { return start.Add(time.Hour) }
+	eta, ok := p.ETA(h.rowsOrIDs(pos, writer), h.rowsOrIDs(0, pos))
+	if !ok || eta < 55*time.Minute || eta > 65*time.Minute {
+		t.Fatalf("ETA in rows %v, %v; want about an hour", eta, ok)
+	}
+	var ids FillProgress
+	ids.now = func() time.Time { return start }
+	ids.Start(0)
+	ids.now = func() time.Time { return start.Add(time.Hour) }
+	if eta, _ := ids.ETA(writer-pos, pos); eta < 100*time.Hour {
+		t.Fatalf("ETA in ids %v; the test's ids should make it about 155 h", eta)
+	}
+}

db/fill_pacing.go

@@ -20,7 +20,7 @@ import (
 type FillPacing struct {
 	Duty             float64       // the share of time a fill may be busy, in (0, 1]
 	MinPause         time.Duration // the shortest pause after a slice or batch
-	PacketSitesSlice int64         // observation ids one packet_sites backfill slice covers
+	PacketSitesSlice int64         // observations one packet_sites backfill slice covers
 	CmiBatch         int64         // message ids one cmi backfill batch covers
 }
 
@@ -103,6 +103,13 @@ func (p *FillProgress) Start(pos int64) {
 	}
 }
 
+// Started reports whether the clock is running.
+func (p *FillProgress) Started() bool {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+	return !p.started.IsZero()
+}
+
 // Reset forgets the clock, for a fill that starts over (a rebuild) or has completed.
 func (p *FillProgress) Reset() {
 	p.mu.Lock()
@@ -135,12 +142,17 @@ func (p *FillProgress) eta(now time.Time, pos, remaining int64) (time.Duration,
 // Log logs msg with the ids done and remaining and, once the position moved, the ETA. It returns the ETA and
 // whether there was one.
 func (p *FillProgress) Log(msg string, pos, done, remaining int64, attrs ...any) (time.Duration, bool) {
+	return p.LogLeft(msg, pos, remaining, append(attrs, "ids_done", done, "ids_remaining", remaining)...)
+}
+
+// LogLeft logs msg with attrs and, once the position moved, the ETA to cover left more units of the position.
+func (p *FillProgress) LogLeft(msg string, pos, left int64, attrs ...any) (time.Duration, bool) {
 	p.mu.Lock()
 	now := p.clock()
 	p.lastLogged = now
-	eta, ok := p.eta(now, pos, remaining)
+	eta, ok := p.eta(now, pos, left)
 	p.mu.Unlock()
-	attrs = append(attrs, "component", "readmodel", "ids_done", done, "ids_remaining", remaining)
+	attrs = append([]any{"component", "readmodel"}, attrs...)
 	if ok {
 		attrs = append(attrs, "eta", eta.Round(time.Second).String())
 	}

db/packet_sites.go

@@ -22,8 +22,8 @@ import (
 
 const (
 	packetSitesStateKey = "packet_sites"
-	// packetSitesSlice is the observation ids one fill covers in the reconcile and the audit, and the backfill's
-	// default slice (readmodels.fill.packet_sites_slice).
+	// packetSitesSlice is the observation ids one fill covers in the reconcile and the audit, and the observations
+	// in the backfill's default slice (readmodels.fill.packet_sites_slice).
 	packetSitesSlice int64 = 500_000
 	// packetSitesFillAttempts bounds the retries of a fill that lost a packet to a concurrent cascade (23503).
 	packetSitesFillAttempts = 3
@@ -56,7 +56,8 @@ type PacketSitesSession struct {
 	duty     float64
 	slice    int64
 	progress *FillProgress
-	sliced   bool // this session's Backfill has filled a slice
+	sliced   bool         // this session's Backfill has filled a slice
+	ids      *idHistogram // the statistics progress counts in, kept for the whole fill so its units never change
 
 	// fillStmt runs one fill; tests replace it. fills lists the (after, through] ranges filled, and retries counts
 	// 23503 retries, for tests.
@@ -91,7 +92,8 @@ func (s *Store) BeginPacketSites(ctx context.Context) (*PacketSitesSession, bool
 	}
 	pace := s.FillPacing()
 	p := &PacketSitesSession{store: s, conn: conn, q: sqlc.New(conn),
-		pause: pace.MinPause, duty: pace.Duty, slice: pace.PacketSitesSlice, progress: &s.packetSitesProgress}
+		pause: pace.MinPause, duty: pace.Duty, slice: pace.PacketSitesSlice, progress: &s.packetSitesProgress,
+		ids: &s.packetSitesIDs}
 	p.fillStmt = func(ctx context.Context, after, through int64) (int64, error) {
 		return p.q.FillPacketSitesSlice(ctx, sqlc.FillPacketSitesSliceParams{AfterID: after, ThroughID: through})
 	}
@@ -168,10 +170,12 @@ func (p *PacketSitesSession) fillWith(ctx context.Context, after, through int64,
 	return 0, err
 }
 
-// Backfill fills packet_sites from WriterFromID down, one slice of observation ids at a time (the session's slice,
+// Backfill fills packet_sites from WriterFromID down, one slice of observations at a time (the session's slice,
 // packetSitesSlice by default), newest first, until it passes the oldest observation (then Complete) or budget is
-// spent. Progress is saved after every slice, and a failed slice is retried from the same place on the next call.
-// After each slice it waits as FillPacing.Wait does, so it is busy at most duty of the time.
+// spent. A slice is the next slice observations by id, so a long, sparse id range costs what its rows cost. Progress
+// is saved after every slice, and a failed slice is retried from the same place on the next call. After each slice
+// it waits as FillPacing.Wait does, so it is busy at most duty of the time. Its progress log and ETA count
+// estimated observations, not ids.
 func (p *PacketSitesSession) Backfill(ctx context.Context, budget time.Duration) (PacketSitesState, error) {
 	deadline := time.Now().Add(budget)
 	slice := p.slice
@@ -198,10 +202,17 @@ func (p *PacketSitesSession) Backfill(ctx context.Context, budget time.Duration)
 			}
 			break
 		}
-		if p.progress != nil {
-			p.progress.Start(p.state.WriterFromID - p.state.BackfilledDownToID)
+		if p.progress != nil && !p.progress.Started() {
+			p.loadIDs(ctx)
+			p.progress.Start(p.ids.rowsOrIDs(p.state.BackfilledDownToID, p.state.WriterFromID))
 		}
-		from := max(p.state.BackfilledDownToID-slice, lo-1)
+		// A slice holds slice observations, however far apart their ids are: retention leaves a few old observations
+		// spread over a long id range, which would otherwise take a paced slice per slice ids.
+		from, err := p.q.ObservationIDBelow(ctx, sqlc.ObservationIDBelowParams{ThroughID: p.state.BackfilledDownToID, N: slice})
+		if err != nil {
+			return p.state, err
+		}
+		from = max(from, lo-1)
 		if _, err := p.fill(ctx, from, p.state.BackfilledDownToID); err != nil {
 			return p.state, fmt.Errorf("packet_sites backfill (%d, %d]: %w", from, p.state.BackfilledDownToID, err)
 		}
@@ -215,7 +226,10 @@ func (p *PacketSitesSession) Backfill(ctx context.Context, budget time.Duration)
 			p.sliced = true
 		}
 		if p.progress != nil && p.progress.Due() {
-			p.progress.Log("packet_sites backfill progress", p.state.WriterFromID-from, p.state.WriterFromID-from, from-(lo-1),
+			done, left := p.ids.rowsOrIDs(from, p.state.WriterFromID), p.ids.rowsOrIDs(lo-1, from)
+			p.progress.LogLeft("packet_sites backfill progress", done, left,
+				"ids_done", p.state.WriterFromID-from, "ids_remaining", from-(lo-1),
+				"observations_done", done, "observations_remaining", left,
 				"backfilled_down_to_id", from, "writer_from_id", p.state.WriterFromID)
 		}
 		pace := FillPacing{Duty: p.duty, MinPause: p.pause}
@@ -226,6 +240,17 @@ func (p *PacketSitesSession) Backfill(ctx context.Context, budget time.Duration)
 	return p.state, nil
 }
 
+// loadIDs reads the observation id statistics the progress log counts in, when its clock starts; without them
+// (a failed read, or no ANALYZE yet) progress counts ids.
+func (p *PacketSitesSession) loadIDs(ctx context.Context) {
+	h, err := loadObservationIDHistogram(ctx, p.conn)
+	if err != nil {
+		slog.Warn("packet_sites backfill: cannot read observation statistics; progress counts ids", "component", "readmodel", "error", err)
+		h = idHistogram{}
+	}
+	*p.ids = h
+}
+
 // sleepCtx waits d or until ctx ends; it reports whether ctx is still live.
 func sleepCtx(ctx context.Context, d time.Duration) bool {
 	t := time.NewTimer(d)

db/packet_sites_reconcile_integration_test.go

@@ -120,15 +120,15 @@ func TestPacketSitesInitAndBackfillPostgres(t *testing.T) {
 	if err != nil || !st.Complete {
 		t.Fatalf("Backfill = %+v, %v", st, err)
 	}
-	// 1.2M ids from the oldest: three slices, none below the oldest observation.
-	if fmt.Sprint(p.fills) != "[[3000700000 3001200000] [3000200000 3000700000] [3000000000 3000200000]]" {
+	// 1.2M ids from the oldest but 600 observations: one slice, not below the oldest observation.
+	if fmt.Sprint(p.fills) != "[[3000000000 3001200000]]" {
 		t.Fatalf("fills %v", p.fills)
 	}
 	f.wantClean("after backfill")
 	if n, err := sqlc.New(f.pool).FillPacketSitesSlice(f.ctx, sqlc.FillPacketSitesSliceParams{AfterID: 0, ThroughID: f.maxID()}); err != nil || n != 0 {
 		t.Fatalf("a second fill wrote %d, %v", n, err)
 	}
-	if st, err := p.Backfill(f.ctx, time.Minute); err != nil || len(p.fills) != 3 || !st.Complete {
+	if st, err := p.Backfill(f.ctx, time.Minute); err != nil || len(p.fills) != 1 || !st.Complete {
 		t.Fatalf("a complete backfill ran again: %+v, %v, %v", st, err, p.fills)
 	}
 

db/packet_sites_sparse_integration_test.go

@@ -0,0 +1,164 @@
+// Copyright 2026 Beacon Contributors
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package db
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc"
+)
+
+// sparseObservations stores three clusters of observations with 100,000 unused ids between them, the way a long
+// retention run leaves a few old observations far below the live ids.
+func (f *psFixture) sparseObservations() {
+	f.t.Helper()
+	stream := f.packets(60)
+	third := len(stream) / 3
+	f.unsited(stream[:third]...)
+	f.exec("SELECT setval('packet_observations_id_seq', (SELECT max(id) FROM packet_observations) + 100000)")
+	f.unsited(stream[third : 2*third]...)
+	f.exec("SELECT setval('packet_observations_id_seq', (SELECT max(id) FROM packet_observations) + 100000)")
+	f.unsited(stream[2*third:]...)
+}
+
+// observationIDs lists every observation id, ascending.
+func (f *psFixture) observationIDs() []int64 {
+	f.t.Helper()
+	rows, err := f.pool.Query(f.ctx, "SELECT id FROM packet_observations ORDER BY id")
+	if err != nil {
+		f.t.Fatal(err)
+	}
+	defer rows.Close()
+	var ids []int64
+	for rows.Next() {
+		var id int64
+		if err := rows.Scan(&id); err != nil {
+			f.t.Fatal(err)
+		}
+		ids = append(ids, id)
+	}
+	if err := rows.Err(); err != nil {
+		f.t.Fatal(err)
+	}
+	return ids
+}
+
+// wantSlicesCover checks that fills run newest first without overlap, that each starts at an observation, and
+// that together they cover every observation id.
+func wantSlicesCover(t *testing.T, fills [][2]int64, ids []int64) {
+	t.Helper()
+	exists := map[int64]bool{}
+	for _, id := range ids {
+		exists[id] = true
+	}
+	covered := 0
+	for i, r := range fills {
+		if !exists[r[1]] || (i > 0 && r[1] > fills[i-1][0]) {
+			t.Fatalf("fill %d is %v: %v", i, r, fills)
+		}
+		for _, id := range ids {
+			if id > r[0] && id <= r[1] {
+				covered++
+			}
+		}
+	}
+	if covered != len(ids) {
+		t.Fatalf("fills %v cover %d of %d observations", fills, covered, len(ids))
+	}
+}
+
+// A backfill slice holds a slice of observations however sparse their ids are, so long id ranges retention left
+// nearly empty cost what their rows cost, and the slices fill exactly what a dense walk does.
+func TestPacketSitesBackfillSlicesSparseIDsPostgres(t *testing.T) {
+	f := newPsFixture(t)
+	f.sparseObservations()
+	ids := f.observationIDs()
+	p := f.begin()
+	p.slice = 50
+	if _, err := p.Init(f.ctx); err != nil {
+		t.Fatal(err)
+	}
+	st, err := p.Backfill(f.ctx, time.Minute)
+	if err != nil || !st.Complete {
+		t.Fatalf("Backfill = %+v, %v", st, err)
+	}
+	// 50 observations a slice, where walking 50 ids a slice would take more than 4,000.
+	if want := (len(ids) + 49) / 50; len(p.fills) != want {
+		t.Fatalf("%d fills for %d observations in three clusters, want %d: %v", len(p.fills), len(ids), want, p.fills)
+	}
+	wantSlicesCover(t, p.fills, ids)
+	f.wantClean("after the sparse backfill")
+}
+
+// A backfill stopped by a failed slice resumes on the next call from its saved position and still covers every
+// observation once.
+func TestPacketSitesBackfillResumesAcrossGapsPostgres(t *testing.T) {
+	f := newPsFixture(t)
+	f.sparseObservations()
+	ids := f.observationIDs()
+	p := f.begin()
+	p.slice = 50
+	if _, err := p.Init(f.ctx); err != nil {
+		t.Fatal(err)
+	}
+	fill, calls := p.fillStmt, 0
+	p.fillStmt = func(ctx context.Context, after, through int64) (int64, error) {
+		if calls++; calls == 3 {
+			return 0, errors.New("disk on fire")
+		}
+		return fill(ctx, after, through)
+	}
+	if _, err := p.Backfill(f.ctx, time.Minute); err == nil {
+		t.Fatal("Backfill ignored a failed slice")
+	}
+	done := p.fills[:2]
+	var saved PacketSitesState
+	if found, err := f.store.LoadState(f.ctx, "packet_sites", &saved); err != nil || !found || saved.Complete || saved.BackfilledDownToID != done[1][0] {
+		t.Fatalf("saved %+v, %v, %v; want the position after two slices, %d", saved, found, err, done[1][0])
+	}
+	p.Close(f.ctx)
+
+	q := f.begin()
+	q.slice = 50
+	st, err := q.Backfill(f.ctx, time.Minute)
+	if err != nil || !st.Complete {
+		t.Fatalf("resumed Backfill = %+v, %v", st, err)
+	}
+	if q.fills[0][1] != saved.BackfilledDownToID {
+		t.Fatalf("resumed at %v, want the saved position %d", q.fills[0], saved.BackfilledDownToID)
+	}
+	wantSlicesCover(t, append(append([][2]int64{}, done...), q.fills...), ids)
+	f.wantClean("after the resumed backfill")
+}
+
+// Progress counts estimated observations: after ANALYZE the estimate of the whole range is close to the row count,
+// and the empty ranges between clusters count for nearly nothing.
+func TestPacketSitesProgressCountsRowsPostgres(t *testing.T) {
+	f := newPsFixture(t)
+	f.sparseObservations()
+	f.exec("ANALYZE packet_observations")
+	ids := f.observationIDs()
+	p := f.begin()
+	p.loadIDs(f.ctx)
+	lo, hi := ids[0], ids[len(ids)-1]
+	all, ok := p.ids.estimate(lo-1, hi)
+	if !ok || all < int64(len(ids))*9/10 || all > int64(len(ids))*11/10 {
+		t.Fatalf("estimate of all %d observations = %d, %v", len(ids), all, ok)
+	}
+	third := len(ids) / 3
+	if gap, _ := p.ids.estimate(ids[third-1], ids[third]-1); gap > int64(len(ids))/20 {
+		t.Fatalf("the empty 100,000 ids between clusters count %d of %d observations", gap, len(ids))
+	}
+	// The 10 observations at or below the first of the second cluster reach back across the gap.
+	if n, err := sqlc.New(f.pool).ObservationIDBelow(f.ctx, sqlc.ObservationIDBelowParams{ThroughID: ids[third], N: 10}); err != nil || n != ids[third-10] {
+		t.Fatalf("ObservationIDBelow across the gap = %d, %v; want %d", n, err, ids[third-10])
+	}
+	if n, err := sqlc.New(f.pool).ObservationIDBelow(f.ctx, sqlc.ObservationIDBelowParams{ThroughID: ids[5], N: 6}); err != nil || n != 0 {
+		t.Fatalf("ObservationIDBelow past the oldest = %d, %v; want 0", n, err)
+	}
+
+}

db/queries/packet_sites.sql

@@ -39,6 +39,12 @@ ON CONFLICT (packet_hash, iata) DO UPDATE SET site_heard_at = EXCLUDED.site_hear
 -- name: MinObservationID :one
 SELECT COALESCE(min(id), 0)::bigint AS id FROM packet_observations;
 
+-- name: ObservationIDBelow :one
+-- The id that bounds the newest n observations at or below through_id from below: (id, through_id] holds exactly n
+-- of them. 0 when fewer than n + 1 remain.
+SELECT COALESCE((SELECT po.id FROM packet_observations po WHERE po.id <= @through_id::bigint
+                 ORDER BY po.id DESC OFFSET @n::bigint LIMIT 1), 0)::bigint AS id;
+
 -- name: CheckPacketSitesSlice :one
 -- Exact check of every packet with an observation in (after_id, through_id]: site rows the observations imply
 -- and the table lacks, rows whose hearing time or packet columns differ, and rows no observation backs.

db/sqlc/mock/querier.go

@@ -2588,6 +2588,21 @@ func (mr *MockQuerierMockRecorder) NextChannelMessageID(ctx, afterID any) *gomoc
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "NextChannelMessageID", reflect.TypeOf((*MockQuerier)(nil).NextChannelMessageID), ctx, afterID)
 }
 
+// ObservationIDBelow mocks base method.
+func (m *MockQuerier) ObservationIDBelow(ctx context.Context, arg db.ObservationIDBelowParams) (int64, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ObservationIDBelow", ctx, arg)
+	ret0, _ := ret[0].(int64)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ObservationIDBelow indicates an expected call of ObservationIDBelow.
+func (mr *MockQuerierMockRecorder) ObservationIDBelow(ctx, arg any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ObservationIDBelow", reflect.TypeOf((*MockQuerier)(nil).ObservationIDBelow), ctx, arg)
+}
+
 // OldestMissingRollupHour mocks base method.
 func (m *MockQuerier) OldestMissingRollupHour(ctx context.Context) (pgtype.Timestamptz, error) {
 	m.ctrl.T.Helper()

db/sqlc/packet_sites.sql.go

@@ -1142,6 +1142,25 @@ func (q *Queries) MinObservationID(ctx context.Context) (int64, error) {
 	return id, err
 }
 
+const observationIDBelow = `-- name: ObservationIDBelow :one
+SELECT COALESCE((SELECT po.id FROM packet_observations po WHERE po.id <= $1::bigint
+                 ORDER BY po.id DESC OFFSET $2::bigint LIMIT 1), 0)::bigint AS id
+`
+
+type ObservationIDBelowParams struct {
+	ThroughID int64 `json:"through_id"`
+	N         int64 `json:"n"`
+}
+
+// The id that bounds the newest n observations at or below through_id from below: (id, through_id] holds exactly n
+// of them. 0 when fewer than n + 1 remain.
+func (q *Queries) ObservationIDBelow(ctx context.Context, arg ObservationIDBelowParams) (int64, error) {
+	row := q.db.QueryRow(ctx, observationIDBelow, arg.ThroughID, arg.N)
+	var id int64
+	err := row.Scan(&id)
+	return id, err
+}
+
 const packetSitesTotalBytes = `-- name: PacketSitesTotalBytes :one
 SELECT pg_total_relation_size('packet_sites'::regclass)::bigint AS bytes
 `

db/sqlc/querier.go

@@ -433,6 +433,9 @@ type Querier interface {
 	MinObservationID(ctx context.Context) (int64, error)
 	// The first message id above after_id, or 0 when there is none.
 	NextChannelMessageID(ctx context.Context, afterID int64) (int64, error)
+	// The id that bounds the newest n observations at or below through_id from below: (id, through_id] holds exactly n
+	// of them. 0 when fewer than n + 1 remain.
+	ObservationIDBelow(ctx context.Context, arg ObservationIDBelowParams) (int64, error)
 	OldestMissingRollupHour(ctx context.Context) (pgtype.Timestamptz, error)
 	PacketSitesTotalBytes(ctx context.Context) (int64, error)
 	PruneImportedRegions(ctx context.Context, keep []string) ([]string, error)

db/store.go

@@ -35,6 +35,7 @@ type Store struct {
 	packetSitesVacuums  atomic.Int32    // VACUUM (ANALYZE) runs of packet_sites by the first-time verification
 	fill                *FillPacing     // see WithFillPacing; nil: DefaultFillPacing
 	packetSitesProgress FillProgress    // the packet_sites backfill's progress log, across sessions
+	packetSitesIDs      idHistogram     // the id statistics that progress counts in, read when it started
 	cmiProgress         FillProgress    // the cmi backfill's progress log, across sessions
 }
 

internal/config/config.go

@@ -69,7 +69,7 @@ type ReadModelsConfig struct {
 type FillConfig struct {
 	Duty             *float64  `yaml:"duty"`               // 0.01 to 1; default 0.5
 	MinPause         *duration `yaml:"min_pause"`          // 0 to 1m; default 100ms
-	PacketSitesSlice *int64    `yaml:"packet_sites_slice"` // observation ids per packet_sites slice; default 500000
+	PacketSitesSlice *int64    `yaml:"packet_sites_slice"` // observations per packet_sites slice; default 500000
 	CmiBatch         *int64    `yaml:"cmi_batch"`          // message ids per cmi batch; default 10000
 }