Sample the audit and verification slices by row; count payload-type progress in rows (#280)

* Sample the audit and verification slices by row; count payload-type progress in rows

* Read sample statistics outside the verify transaction; anchor payload-type progress at id 0

.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|PacketSitesBackfillSlicesSparseIDs|PacketSitesBackfillResumesAcrossGaps|PacketSitesProgressCountsRows|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|PacketSitesAuditSamplesRows|PacketSitesVerifySamplesRows|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:
@@ -79,7 +79,7 @@ jobs:
       - name: Verify route retention task against PostgreSQL 16
         env:
           BEACON_TEST_POSTGRES_DSN: postgres://postgres:backup-ci-only@127.0.0.1:5432/postgres?sslmode=disable
-        run: go test ./internal/background -run '^Test(RouteRetentionTask|PayloadTypeBackfill|PayloadTypeBackfillProbe|CmiBootRepair|CmiReadOnlyTrust|PacketSitesRollbackRollForward|PacketSitesTrustDropsOnBypass|PacketSitesReadOnlyTrust|PacedFillsEarnTrust)Postgres$' -count=1 -v
+        run: go test ./internal/background -run '^Test(RouteRetentionTask|PayloadTypeBackfill|PayloadTypeBackfillProbe|PayloadTypeBackfillProgressCountsRows|CmiBootRepair|CmiReadOnlyTrust|PacketSitesRollbackRollForward|PacketSitesTrustDropsOnBypass|PacketSitesReadOnlyTrust|PacedFillsEarnTrust)Postgres$' -count=1 -v
 
       # Builds the fixture once; the planguard tests below reuse it (PLANGUARD_FIXTURE_DSN) after checking that its
       # marker records a build of the same files.

db/fill_estimate.go

@@ -5,6 +5,7 @@ package db
 
 import (
 	"context"
+	"log/slog"
 
 	"github.com/jackc/pgx/v5"
 )
@@ -73,3 +74,49 @@ func (h idHistogram) rowsOrIDs(after, through int64) int64 {
 	}
 	return max(through-after, 0)
 }
+
+// idAtRows is the lowest id x in [after, through] with rowsOrIDs(after, x) >= k: the id about k rows above after.
+func (h idHistogram) idAtRows(after, through, k int64) int64 {
+	lo, hi := after, through
+	for lo < hi {
+		mid := lo + (hi-lo)/2
+		if h.rowsOrIDs(after, mid) >= k {
+			hi = mid
+		} else {
+			lo = mid + 1
+		}
+	}
+	return lo
+}
+
+// sliceStart picks the lower bound of a sampled slice in (after, through]: the id u of the way (u in [0, 1))
+// through the range's estimated rows, leaving room for slice rows above it, so each row is as likely to be sampled
+// as any other however sparse its ids are (except the range's first slice rows, which only a start near after
+// covers). Without statistics it picks by ids, as rows and ids then coincide.
+func (h idHistogram) sliceStart(after, through, slice int64, u float64) int64 {
+	room := h.rowsOrIDs(after, through) - slice
+	if room <= 0 {
+		return after
+	}
+	return h.idAtRows(after, through, int64(u*float64(room)))
+}
+
+// ObservationRows estimates the observations in an id range from the planner's statistics on packet_observations,
+// counting ids where there are none. Progress logs use it, so long empty id ranges count for what they cost.
+type ObservationRows struct{ h idHistogram }
+
+// In is the estimated number of observations with ids in (after, through].
+func (r ObservationRows) In(after, through int64) int64 { return r.h.rowsOrIDs(after, through) }
+
+// ObservationRows reads the statistics for an estimate; a failed read is logged and estimates by ids.
+func (s *Store) ObservationRows(ctx context.Context) ObservationRows {
+	if s.pool == nil {
+		return ObservationRows{}
+	}
+	h, err := loadObservationIDHistogram(ctx, s.pool)
+	if err != nil {
+		slog.Warn("cannot read observation statistics; progress counts ids", "component", "db", "error", err)
+		return ObservationRows{}
+	}
+	return ObservationRows{h}
+}

db/fill_estimate_test.go

@@ -64,3 +64,31 @@ func TestIDHistogramETAFollowsRows(t *testing.T) {
 		t.Fatalf("ETA in ids %v; the test's ids should make it about 155 h", eta)
 	}
 }
+
+// A sampled slice starts at a uniformly random row: a sparse old id range holding 10 % of the rows over 90 % of
+// the ids gets about 10 % of the samples, where picking by id would give it about 90 %.
+func TestSliceStartSamplesRowsProportionally(t *testing.T) {
+	// 100 rows a bucket: 1 bucket over ids (0, 9,000,000], 9 buckets over (9,000,000, 10,000,000].
+	bounds := []int64{1, 9_000_000}
+	for id := int64(9_111_111); id < 10_000_000; id += 111_111 {
+		bounds = append(bounds, id)
+	}
+	bounds = append(bounds, 10_000_000)
+	h := idHistogram{bounds: bounds, rows: float64(len(bounds)-1) * 100}
+	const draws = 10_000
+	sparse := 0
+	for i := range draws {
+		if h.sliceStart(0, 10_000_000, 10, float64(i)/draws) < 9_000_000 {
+			sparse++
+		}
+	}
+	if share := float64(sparse) / draws; share < 0.08 || share > 0.12 {
+		t.Fatalf("sparse range got %.1f %% of the samples, want about 10 %%", 100*share)
+	}
+	if got := h.sliceStart(0, 10_000_000, 2000, 0.5); got != 0 {
+		t.Fatalf("a range holding fewer rows than the slice starts at %d, want its start", got)
+	}
+	if got := (idHistogram{}).sliceStart(100, 200, 10, 0.5); got != 145 {
+		t.Fatalf("without statistics the start is %d, want 145 (by ids)", got)
+	}
+}

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 observations
-	// in the backfill's default slice (readmodels.fill.packet_sites_slice).
+	// packetSitesSlice is the observation ids one fill covers in the reconcile, and the observations in the backfill's
+	// default slice (readmodels.fill.packet_sites_slice) and in the audit's and verification's sampled slice.
 	packetSitesSlice int64 = 500_000
 	// packetSitesFillAttempts bounds the retries of a fill that lost a packet to a concurrent cascade (23503).
 	packetSitesFillAttempts = 3
@@ -58,11 +58,13 @@ type PacketSitesSession struct {
 	progress *FillProgress
 	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
+	sample   int64        // observations in the audit's and verification's sampled slice; tests shrink it
 
 	// fillStmt runs one fill; tests replace it. fills lists the (after, through] ranges filled, and retries counts
 	// 23503 retries, for tests.
 	fillStmt func(ctx context.Context, after, through int64) (int64, error)
 	fills    [][2]int64
+	checks   [][2]int64 // the (after, through] ranges Verify checked, for tests
 	retries  int
 }
 
@@ -93,7 +95,7 @@ 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,
-		ids: &s.packetSitesIDs}
+		ids: &s.packetSitesIDs, sample: packetSitesSlice}
 	p.fillStmt = func(ctx context.Context, after, through int64) (int64, error) {
 		return p.q.FillPacketSitesSlice(ctx, sqlc.FillPacketSitesSliceParams{AfterID: after, ThroughID: through})
 	}
@@ -240,6 +242,31 @@ func (p *PacketSitesSession) Backfill(ctx context.Context, budget time.Duration)
 	return p.state, nil
 }
 
+// loadSampleIDs reads the statistics a sampled slice is picked with, outside any transaction, so a failed read
+// only falls back to picking by ids.
+func (p *PacketSitesSession) loadSampleIDs(ctx context.Context) idHistogram {
+	ids, err := loadObservationIDHistogram(ctx, p.conn)
+	if err != nil {
+		slog.Warn("packet_sites: cannot read observation statistics; the sampled slice is picked by ids", "component", "readmodel", "error", err)
+		return idHistogram{}
+	}
+	return ids
+}
+
+// sampleSlice picks a slice of p.sample observations in (after, through], starting u of the way through the
+// range by row (see idHistogram.sliceStart); a range holding fewer observations is sampled whole.
+func (p *PacketSitesSession) sampleSlice(ctx context.Context, q sqlc.Querier, ids idHistogram, after, through int64, u float64) (int64, int64, error) {
+	from := ids.sliceStart(after, through, p.sample, u)
+	to, err := q.ObservationIDAbove(ctx, sqlc.ObservationIDAboveParams{AfterID: from, Skip: p.sample - 1})
+	if err != nil {
+		return 0, 0, err
+	}
+	if to == 0 || to > through {
+		to = through
+	}
+	return from, to, 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) {
@@ -289,10 +316,11 @@ func (p *PacketSitesSession) Reconcile(ctx context.Context) (int64, error) {
 	return rows, p.save(ctx)
 }
 
-// Audit refills one slice of packetSitesSlice ids below ReconPrevHi, the ids no reconcile will see again, each
-// pair at its latest hearing among all the packet's observations; start picks its lower bound in [lo, hi]. It returns the rows written: anything above zero was missed by every writer
-// and reconcile.
-func (p *PacketSitesSession) Audit(ctx context.Context, start func(lo, hi int64) int64) (int64, error) {
+// Audit refills one sampled slice (packetSitesSlice observations) at or below ReconPrevHi, the ids no reconcile will see
+// again, each pair at its latest hearing among all the packet's observations. The slice starts u (in [0, 1)) of the
+// way through those observations by row, so every observation is as likely to be audited however sparse its ids
+// are. It returns the rows written: anything above zero was missed by every writer and reconcile.
+func (p *PacketSitesSession) Audit(ctx context.Context, u float64) (int64, error) {
 	lo, err := p.q.MinObservationID(ctx)
 	if err != nil {
 		return 0, err
@@ -301,10 +329,9 @@ func (p *PacketSitesSession) Audit(ctx context.Context, start func(lo, hi int64)
 	if lo == 0 || top < lo {
 		return 0, nil
 	}
-	after, through := lo-1, top
-	if top-(lo-1) > packetSitesSlice {
-		after = min(max(start(lo-1, top-packetSitesSlice), lo-1), top-packetSitesSlice)
-		through = after + packetSitesSlice
+	after, through, err := p.sampleSlice(ctx, p.q, p.loadSampleIDs(ctx), lo-1, top, u)
+	if err != nil {
+		return 0, err
 	}
 	return p.fillWith(ctx, after, through, func(ctx context.Context, after, through int64) (int64, error) {
 		return p.q.FillPacketSitesAudit(ctx, sqlc.FillPacketSitesAuditParams{AfterID: after, ThroughID: through})
@@ -341,7 +368,7 @@ func (v *PacketSitesVerification) add(c sqlc.CheckPacketSitesSliceRow) {
 // Verify is the first-time verification. The caller has just reconciled (Verify does not reconcile again, which
 // would shrink the next run's overlap). It VACUUMs and ANALYZEs the table once per process (index-only scans need
 // its visibility map; a retry every interval does not repeat it), then in one read-only snapshot runs the exact check
-// over the newest slice of observation ids and one random slice below it, and compares region pages of the busiest
+// over the newest slice of observation ids and a random slice of the observations below it, picked by row, and compares region pages of the busiest
 // recent sites with pages read straight from the observations. The first pass is persisted as FirstVerifiedAt. The
 // caller retries a failed pass every interval.
 func (p *PacketSitesSession) Verify(ctx context.Context) (PacketSitesVerification, error) {
@@ -353,6 +380,7 @@ func (p *PacketSitesSession) Verify(ctx context.Context) (PacketSitesVerificatio
 			p.store.packetSitesVacuums.Add(1)
 		}
 	}
+	ids := p.loadSampleIDs(ctx)
 	tx, err := p.conn.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.RepeatableRead, AccessMode: pgx.ReadOnly})
 	if err != nil {
 		return PacketSitesVerification{}, err
@@ -365,11 +393,17 @@ func (p *PacketSitesSession) Verify(ctx context.Context) (PacketSitesVerificatio
 		return PacketSitesVerification{}, err
 	}
 	ranges := [][2]int64{{max(hi-packetSitesSlice, 0), hi}}
-	if below := hi - packetSitesSlice - (lo - 1); lo > 0 && below > packetSitesSlice {
-		start := lo - 1 + rand.Int64N(below-packetSitesSlice)
-		ranges = append(ranges, [2]int64{start, start + packetSitesSlice})
+	if lo > 0 && hi-packetSitesSlice >= lo {
+		// A random slice of the observations below, picked by row: in a long, sparse old id range a slice of ids
+		// would hold almost nothing.
+		after, through, err := p.sampleSlice(ctx, q, ids, lo-1, hi-packetSitesSlice, rand.Float64())
+		if err != nil {
+			return PacketSitesVerification{}, err
+		}
+		ranges = append(ranges, [2]int64{after, through})
 	}
 	var v PacketSitesVerification
+	p.checks = append(p.checks, ranges...)
 	for _, r := range ranges {
 		c, err := q.CheckPacketSitesSlice(ctx, sqlc.CheckPacketSitesSliceParams{AfterID: r[0], ThroughID: r[1]})
 		if err != nil {

db/packet_sites_reconcile_integration_test.go

@@ -374,19 +374,19 @@ func TestPacketSitesAuditPostgres(t *testing.T) {
 	if _, err := p.Backfill(f.ctx, time.Minute); err != nil {
 		t.Fatal(err)
 	}
-	lowest := func(lo, _ int64) int64 { return lo }
-	if n, err := p.Audit(f.ctx, lowest); err != nil || n != 0 || fmt.Sprint(p.fills[len(p.fills)-1]) != "[3000000000 3000500000]" {
+	// 600 observations, fewer than a sampled slice: the audit covers all of them.
+	if n, err := p.Audit(f.ctx, 0); err != nil || n != 0 || fmt.Sprint(p.fills[len(p.fills)-1]) != "[3000000000 3001200000]" {
 		t.Fatalf("clean audit %d, %v, %v", n, err, p.fills)
 	}
 	// A row lost below the watermark (manual SQL) comes back from the audit of its slice.
 	f.exec(`DELETE FROM packet_sites WHERE (packet_hash, iata) = (
   SELECT packet_hash, iata FROM packet_observations WHERE id = 3000000001)`)
-	if n, err := p.Audit(f.ctx, lowest); err != nil || n != 1 {
+	if n, err := p.Audit(f.ctx, 0); err != nil || n != 1 {
 		t.Fatalf("audit repaired %d, %v; want 1", n, err)
 	}
 	f.wantClean("audited")
 
-	// Less than one slice below the watermark: the audit covers (min - 1, ReconPrevHi].
+	// Fewer observations than a slice below the watermark: the audit covers (min - 1, ReconPrevHi] wherever u points.
 	g := newPsFixture(t)
 	g.sited(g.packets(20)...)
 	s := g.begin()
@@ -394,7 +394,7 @@ func TestPacketSitesAuditPostgres(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if _, err := s.Audit(g.ctx, func(int64, int64) int64 { t.Fatal("start called for a short range"); return 0 }); err != nil {
+	if _, err := s.Audit(g.ctx, 0.99); err != nil {
 		t.Fatal(err)
 	}
 	if fmt.Sprint(s.fills) != fmt.Sprint([][2]int64{{3_000_000_000, st.ReconPrevHi}}) {
@@ -416,7 +416,7 @@ func TestPacketSitesAuditTakesNewestHearingPostgres(t *testing.T) {
 		t.Fatal(err)
 	}
 	f.exec("DELETE FROM packet_sites WHERE packet_hash = $1", p)
-	n, err := s.Audit(f.ctx, func(lo, _ int64) int64 { return lo })
+	n, err := s.Audit(f.ctx, 0)
 	if err != nil || n != 1 {
 		t.Fatalf("audit repaired %d, %v; want 1", n, err)
 	}

db/packet_sites_sparse_integration_test.go

@@ -162,3 +162,82 @@ func TestPacketSitesProgressCountsRowsPostgres(t *testing.T) {
 	}
 
 }
+
+// The audit samples observations, not ids: over three equal clusters 100,000 ids apart each cluster gets about a
+// third of the audited slices, and every slice holds a full sample, where slices of ids would mostly land in the
+// empty ranges between them.
+func TestPacketSitesAuditSamplesRowsPostgres(t *testing.T) {
+	f := newPsFixture(t)
+	f.sparseObservations()
+	f.exec("ANALYZE packet_observations")
+	ids := f.observationIDs()
+	p := f.begin()
+	p.slice = 50
+	if _, err := p.Init(f.ctx); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := p.Backfill(f.ctx, time.Minute); err != nil {
+		t.Fatal(err)
+	}
+	p.sample = 20
+	third := len(ids) / 3
+	clusters := [3]int{}
+	const draws = 60
+	for i := range draws {
+		p.fills = nil
+		if n, err := p.Audit(f.ctx, float64(i)/draws); err != nil || n != 0 {
+			t.Fatalf("audit %d: %d, %v", i, n, err)
+		}
+		r := p.fills[0]
+		held := 0
+		for k, id := range ids {
+			if id > r[0] && id <= r[1] {
+				held++
+				clusters[min(k/third, 2)]++
+			}
+		}
+		if held != 20 {
+			t.Fatalf("audit %d slice %v holds %d observations, want 20", i, r, held)
+		}
+	}
+	for c, n := range clusters {
+		if share := float64(n) / (draws * 20); share < 0.2 || share > 0.47 {
+			t.Fatalf("cluster %d got %.0f %% of the audited observations %v, want about a third", c, 100*share, clusters)
+		}
+	}
+}
+
+// The verification's random slice below the newest is a sample of observations too: it holds a full sample even
+// when the ids below are mostly empty.
+func TestPacketSitesVerifySamplesRowsPostgres(t *testing.T) {
+	f := newPsFixture(t)
+	f.sparseObservations()
+	f.exec("ANALYZE packet_observations")
+	ids := f.observationIDs()
+	p := f.begin()
+	p.slice, p.sample = 50, 20
+	if _, err := p.Init(f.ctx); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := p.Backfill(f.ctx, time.Minute); err != nil {
+		t.Fatal(err)
+	}
+	// Raise the watermark far above the observations, so the newest 500,000-id slice is empty and everything is
+	// below it.
+	p.state.ReconHi = ids[len(ids)-1] + packetSitesSlice + 1
+	for range 10 {
+		if v, err := p.Verify(f.ctx); err != nil || !v.OK() {
+			t.Fatalf("Verify = %+v, %v", v, err)
+		}
+		r := p.checks[len(p.checks)-1]
+		held := 0
+		for _, id := range ids {
+			if id > r[0] && id <= r[1] {
+				held++
+			}
+		}
+		if held != 20 {
+			t.Fatalf("sampled range %v holds %d observations, want 20", r, held)
+		}
+	}
+}

db/queries/packet_sites.sql

@@ -45,6 +45,11 @@ SELECT COALESCE(min(id), 0)::bigint AS id FROM packet_observations;
 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: ObservationIDAbove :one
+-- The id skip + 1 observations above after_id: (after_id, id] holds exactly skip + 1 of them. 0 when fewer remain.
+SELECT COALESCE((SELECT po.id FROM packet_observations po WHERE po.id > @after_id::bigint
+                 ORDER BY po.id OFFSET @skip::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)
 }
 
+// ObservationIDAbove mocks base method.
+func (m *MockQuerier) ObservationIDAbove(ctx context.Context, arg db.ObservationIDAboveParams) (int64, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "ObservationIDAbove", ctx, arg)
+	ret0, _ := ret[0].(int64)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// ObservationIDAbove indicates an expected call of ObservationIDAbove.
+func (mr *MockQuerierMockRecorder) ObservationIDAbove(ctx, arg any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ObservationIDAbove", reflect.TypeOf((*MockQuerier)(nil).ObservationIDAbove), ctx, arg)
+}
+
 // ObservationIDBelow mocks base method.
 func (m *MockQuerier) ObservationIDBelow(ctx context.Context, arg db.ObservationIDBelowParams) (int64, error) {
 	m.ctrl.T.Helper()

db/sqlc/packet_sites.sql.go

@@ -1142,6 +1142,24 @@ func (q *Queries) MinObservationID(ctx context.Context) (int64, error) {
 	return id, err
 }
 
+const observationIDAbove = `-- name: ObservationIDAbove :one
+SELECT COALESCE((SELECT po.id FROM packet_observations po WHERE po.id > $1::bigint
+                 ORDER BY po.id OFFSET $2::bigint LIMIT 1), 0)::bigint AS id
+`
+
+type ObservationIDAboveParams struct {
+	AfterID int64 `json:"after_id"`
+	Skip    int64 `json:"skip"`
+}
+
+// The id skip + 1 observations above after_id: (after_id, id] holds exactly skip + 1 of them. 0 when fewer remain.
+func (q *Queries) ObservationIDAbove(ctx context.Context, arg ObservationIDAboveParams) (int64, error) {
+	row := q.db.QueryRow(ctx, observationIDAbove, arg.AfterID, arg.Skip)
+	var id int64
+	err := row.Scan(&id)
+	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

db/sqlc/querier.go

@@ -433,6 +433,8 @@ 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 skip + 1 observations above after_id: (after_id, id] holds exactly skip + 1 of them. 0 when fewer remain.
+	ObservationIDAbove(ctx context.Context, arg ObservationIDAboveParams) (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)

internal/background/payload_type_backfill.go

@@ -27,6 +27,12 @@ type payloadTypeBackfiller interface {
 	SaveState(context.Context, string, any) error
 }
 
+// observationRows is a store that estimates observations by id range from the planner's statistics (db.Store
+// does); without it progress counts ids.
+type observationRows interface {
+	ObservationRows(context.Context) db.ObservationRows
+}
+
 // fillPacer is a store that carries the backfills' pacing (db.Store does); a store without it is paced with
 // db.DefaultFillPacing.
 type fillPacer interface {
@@ -59,6 +65,7 @@ func payloadTypeBackfillTask(store payloadTypeBackfiller, interval time.Duration
 		pace = p.FillPacing()
 	}
 	var progress db.FillProgress
+	var rows db.ObservationRows // read when progress starts, so its units stay the same for the whole walk
 	return Task{
 		Name:     "payload_type_backfill",
 		Interval: interval,
@@ -90,14 +97,23 @@ func payloadTypeBackfillTask(store payloadTypeBackfiller, interval time.Duration
 				}
 			}
 			next := max(st.NextID, minID)
-			progress.Start(next) // the ETA uses the rate next moves at and each run's own remaining count
-			var rows int64
+			if !progress.Started() {
+				if r, ok := store.(observationRows); ok {
+					rows = r.ObservationRows(ctx)
+				}
+			}
+			// The ETA uses the rate the observations below next grow at and each run's own remaining count, both
+			// estimated from the statistics, so a long, sparse old id range counts for what it holds.
+			progress.Start(rows.In(0, next-1)) // from id 0: retention raising minID must not move the position back
+			var typed int64
 			defer func() {
-				metrics.JobRows.WithLabelValues("payload_type_backfill").Add(float64(rows))
+				metrics.JobRows.WithLabelValues("payload_type_backfill").Add(float64(typed))
 				remaining := max(maxID+1-next, 0)
-				attrs := []any{"component", "background", "next_id", next, "max_id", maxID, "rows", rows,
-					"ids_done", max(next-minID, 0), "ids_remaining", remaining}
-				if eta, ok := progress.ETA(next, remaining); ok {
+				done, left := rows.In(minID-1, next-1), rows.In(next-1, maxID)
+				attrs := []any{"component", "background", "next_id", next, "max_id", maxID, "rows", typed,
+					"ids_done", max(next-minID, 0), "ids_remaining", remaining,
+					"observations_done", done, "observations_remaining", left}
+				if eta, ok := progress.ETA(rows.In(0, next-1), left); ok {
 					attrs = append(attrs, "eta", eta.Round(time.Second).String())
 				}
 				slog.Info("observation payload type backfill progress: observer advert pages answer 503 until it is done", attrs...)
@@ -115,7 +131,7 @@ func payloadTypeBackfillTask(store payloadTypeBackfiller, interval time.Duration
 				if err != nil {
 					return fmt.Errorf("payload type backfill: %w", err)
 				}
-				rows += n
+				typed += n
 				next = to
 				if err := store.SaveState(ctx, db.ObservationPayloadTypeKey, db.ObservationPayloadTypeState{NextID: next}); err != nil {
 					return fmt.Errorf("payload type backfill: %w", err)

internal/background/payload_type_backfill_integration_test.go

@@ -5,7 +5,9 @@ package background
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
+	"log/slog"
 	"maps"
 	"os"
 	"reflect"
@@ -468,3 +470,49 @@ WHERE po.payload_type IS DISTINCT FROM p.payload_type`); n != 1 {
 		}
 	})
 }
+
+// The progress line counts observations from the planner's statistics: 200 untyped observations split by 10
+// million unused ids report about 150 left after the first 50, not 10 million ids.
+func TestPayloadTypeBackfillProgressCountsRowsPostgres(t *testing.T) {
+	ctx, pool := migratedPool(t)
+	if _, err := pool.Exec(ctx, `
+INSERT INTO iata_codes(iata) VALUES ('YOW');
+INSERT INTO observers(id,public_key) VALUES ('00000000-0000-0000-0000-0000000000a1','\xa1');
+INSERT INTO packets(packet_hash,payload_type,payload_version,route_type,raw_payload,raw_header,first_heard_at,last_heard_at)
+SELECT int4send(i),4,0,1,'\x00','\x00',now(),now() FROM generate_series(1,200) i;
+INSERT INTO packet_observations(packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count)
+SELECT int4send(i),'00000000-0000-0000-0000-0000000000a1','YOW',now(),0,1,0 FROM generate_series(1,100) i ORDER BY i;
+SELECT setval('packet_observations_id_seq', (SELECT max(id) FROM packet_observations) + 10000000);
+INSERT INTO packet_observations(packet_hash,observer_id,iata,heard_at,path_length_byte,hash_size,hop_count)
+SELECT int4send(i),'00000000-0000-0000-0000-0000000000a1','YOW',now(),0,1,0 FROM generate_series(101,200) i ORDER BY i;
+ANALYZE packet_observations;`); err != nil {
+		t.Fatal(err)
+	}
+	var logs strings.Builder
+	prev := slog.Default()
+	slog.SetDefault(slog.New(slog.NewJSONHandler(&logs, nil)))
+	t.Cleanup(func() { slog.SetDefault(prev) })
+	task := payloadTypeBackfillTask(db.New(pool, pool, 0, 0, db.WithFillPacing(db.FillPacing{Duty: 1})), time.Minute, 50, 1000, 1)
+	if err := task.Run(ctx); err != nil {
+		t.Fatal(err)
+	}
+	var line struct {
+		Msg                   string `json:"msg"`
+		IDsRemaining          int64  `json:"ids_remaining"`
+		ObservationsDone      int64  `json:"observations_done"`
+		ObservationsRemaining int64  `json:"observations_remaining"`
+	}
+	for _, l := range strings.Split(strings.TrimSpace(logs.String()), "\n") {
+		if strings.Contains(l, "payload type backfill progress") {
+			if err := json.Unmarshal([]byte(l), &line); err != nil {
+				t.Fatal(err)
+			}
+		}
+	}
+	if line.Msg == "" || line.IDsRemaining < 10_000_000 {
+		t.Fatalf("progress line %+v in %s", line, logs.String())
+	}
+	if line.ObservationsDone < 40 || line.ObservationsDone > 60 || line.ObservationsRemaining < 135 || line.ObservationsRemaining > 165 {
+		t.Fatalf("observations done %d, remaining %d; want about 50 and 150", line.ObservationsDone, line.ObservationsRemaining)
+	}
+}

internal/background/readmodel_packet_sites.go

@@ -47,7 +47,7 @@ type PacketSitesSession interface {
 	Init(ctx context.Context) (db.PacketSitesState, error)
 	Reconcile(ctx context.Context) (int64, error)
 	Backfill(ctx context.Context, budget time.Duration) (db.PacketSitesState, error)
-	Audit(ctx context.Context, start func(lo, hi int64) int64) (int64, error)
+	Audit(ctx context.Context, u float64) (int64, error)
 	Verify(ctx context.Context) (db.PacketSitesVerification, error)
 	Close(ctx context.Context)
 }
@@ -286,7 +286,7 @@ func (j *PacketSitesJobs) AuditTask() Task {
 			if !st.Complete {
 				return nil // a rebuild is backfilling: its missing slices are not misses
 			}
-			n, err := s.Audit(ctx, func(lo, hi int64) int64 { return lo + rand.Int64N(hi-lo+1) })
+			n, err := s.Audit(ctx, rand.Float64())
 			if err != nil {
 				return err
 			}

internal/background/readmodel_packet_sites_test.go

@@ -76,9 +76,9 @@ func (f *fakeSites) Backfill(_ context.Context, budget time.Duration) (db.Packet
 	}
 	return f.state, nil
 }
-func (f *fakeSites) Audit(_ context.Context, start func(lo, hi int64) int64) (int64, error) {
+func (f *fakeSites) Audit(_ context.Context, u float64) (int64, error) {
 	f.log("audit")
-	if a := start(10, 20); a < 10 || a > 20 {
+	if u < 0 || u >= 1 {
 		f.log("bad-start")
 	}
 	if len(f.audits) == 0 {