diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 26d6b69..8e6486c 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|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: diff --git a/config.yaml.example b/config.yaml.example index 80fab13..59a3a90 100644 --- a/config.yaml.example +++ b/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). diff --git a/db/fill_estimate.go b/db/fill_estimate.go new file mode 100644 index 0000000..f5bb269 --- /dev/null +++ b/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) +} diff --git a/db/fill_estimate_test.go b/db/fill_estimate_test.go new file mode 100644 index 0000000..dcd1ce7 --- /dev/null +++ b/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) + } +} diff --git a/db/fill_pacing.go b/db/fill_pacing.go index 7b147d6..8cbfebc 100644 --- a/db/fill_pacing.go +++ b/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()) } diff --git a/db/packet_sites.go b/db/packet_sites.go index 5e095b1..da41803 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 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) diff --git a/db/packet_sites_reconcile_integration_test.go b/db/packet_sites_reconcile_integration_test.go index 352ab27..cdaf5d3 100644 --- a/db/packet_sites_reconcile_integration_test.go +++ b/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) } diff --git a/db/packet_sites_sparse_integration_test.go b/db/packet_sites_sparse_integration_test.go new file mode 100644 index 0000000..b494956 --- /dev/null +++ b/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) + } + +} diff --git a/db/queries/packet_sites.sql b/db/queries/packet_sites.sql index 1f81d50..fe94045 100644 --- a/db/queries/packet_sites.sql +++ b/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. diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index ab93752..9674974 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) } +// 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() diff --git a/db/sqlc/packet_sites.sql.go b/db/sqlc/packet_sites.sql.go index 351ef0a..3dfcdb5 100644 --- a/db/sqlc/packet_sites.sql.go +++ b/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 ` diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 993fdd9..b2af146 100644 --- a/db/sqlc/querier.go +++ b/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) diff --git a/db/store.go b/db/store.go index 4364816..12442b5 100644 --- a/db/store.go +++ b/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 } diff --git a/internal/config/config.go b/internal/config/config.go index 7abab6f..002b592 100644 --- a/internal/config/config.go +++ b/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 }