From 4483c6de3d2cc0193f43e104ef88f7ca4f213f86 Mon Sep 17 00:00:00 2001 From: MrAlders0n <55921894+MrAlders0n@users.noreply.github.com> Date: Sat, 24 Oct 2026 13:28:15 -0400 Subject: [PATCH] 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 | 4 +- db/fill_estimate.go | 47 +++++++++++ db/fill_estimate_test.go | 28 +++++++ db/packet_sites.go | 64 +++++++++++---- db/packet_sites_reconcile_integration_test.go | 12 +-- db/packet_sites_sparse_integration_test.go | 79 +++++++++++++++++++ db/queries/packet_sites.sql | 5 ++ db/sqlc/mock/querier.go | 15 ++++ db/sqlc/packet_sites.sql.go | 18 +++++ db/sqlc/querier.go | 2 + internal/background/payload_type_backfill.go | 30 +++++-- .../payload_type_backfill_integration_test.go | 48 +++++++++++ internal/background/readmodel_packet_sites.go | 4 +- .../background/readmodel_packet_sites_test.go | 4 +- 14 files changed, 326 insertions(+), 34 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8e6486c..1581cf7 100644 --- a/.github/workflows/ci.yml +++ b/.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. diff --git a/db/fill_estimate.go b/db/fill_estimate.go index f5bb269..199d302 100644 --- a/db/fill_estimate.go +++ b/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} +} diff --git a/db/fill_estimate_test.go b/db/fill_estimate_test.go index dcd1ce7..a4e849f 100644 --- a/db/fill_estimate_test.go +++ b/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) + } +} diff --git a/db/packet_sites.go b/db/packet_sites.go index da41803..38a00c0 100644 --- a/db/packet_sites.go +++ b/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 { diff --git a/db/packet_sites_reconcile_integration_test.go b/db/packet_sites_reconcile_integration_test.go index cdaf5d3..f47f3d2 100644 --- a/db/packet_sites_reconcile_integration_test.go +++ b/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) } diff --git a/db/packet_sites_sparse_integration_test.go b/db/packet_sites_sparse_integration_test.go index b494956..0e3a2a4 100644 --- a/db/packet_sites_sparse_integration_test.go +++ b/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) + } + } +} diff --git a/db/queries/packet_sites.sql b/db/queries/packet_sites.sql index fe94045..409de0c 100644 --- a/db/queries/packet_sites.sql +++ b/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. diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 9674974..4d30ae5 100644 --- a/db/sqlc/mock/querier.go +++ b/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() diff --git a/db/sqlc/packet_sites.sql.go b/db/sqlc/packet_sites.sql.go index 3dfcdb5..d864e37 100644 --- a/db/sqlc/packet_sites.sql.go +++ b/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 diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index b2af146..6d414bb 100644 --- a/db/sqlc/querier.go +++ b/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) diff --git a/internal/background/payload_type_backfill.go b/internal/background/payload_type_backfill.go index cdbd784..b377c5c 100644 --- a/internal/background/payload_type_backfill.go +++ b/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) diff --git a/internal/background/payload_type_backfill_integration_test.go b/internal/background/payload_type_backfill_integration_test.go index ae2a7cf..8f2d646 100644 --- a/internal/background/payload_type_backfill_integration_test.go +++ b/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) + } +} diff --git a/internal/background/readmodel_packet_sites.go b/internal/background/readmodel_packet_sites.go index 07ff7d5..7c9f5ea 100644 --- a/internal/background/readmodel_packet_sites.go +++ b/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 } diff --git a/internal/background/readmodel_packet_sites_test.go b/internal/background/readmodel_packet_sites_test.go index 1518033..80b6c37 100644 --- a/internal/background/readmodel_packet_sites_test.go +++ b/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 {