diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b309af0..80416a1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -74,7 +74,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)Postgres$' -count=1 -v + run: go test ./internal/background -run '^Test(RouteRetentionTask|PayloadTypeBackfill|PayloadTypeBackfillProbe)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/payload_types.go b/db/payload_types.go index 027cc9f..20e23e4 100644 --- a/db/payload_types.go +++ b/db/payload_types.go @@ -67,6 +67,11 @@ func (s *Store) ObservationIDRange(ctx context.Context) (minID, maxID int64, err return r.MinID, r.MaxID, nil } +// HasUntypedObservations reports whether any observation with id <= maxID lacks a payload type. +func (s *Store) HasUntypedObservations(ctx context.Context, maxID int64) (bool, error) { + return s.q.HasUntypedObservations(ctx, maxID) +} + // BackfillObservationPayloadTypes types, from their own packet, the first maxRows observations in // [fromID, toID) that carry no payload type, in id order. It returns the rows typed and where the // walk continues: after the last row it picked when it picked maxRows, otherwise toID. diff --git a/db/queries/queries.sql b/db/queries/queries.sql index ad45476..0b23aca 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -393,6 +393,14 @@ SELECT (SELECT count(*) FROM typed)::bigint AS typed, COALESCE((SELECT max(id) FROM picked), 0)::bigint AS last_id; +-- name: HasUntypedObservations :one +-- Whether any observation up to @max_id lacks a payload type. A full scan, run at most once per +-- process by the payload-type backfill. +SELECT EXISTS ( + SELECT 1 FROM packet_observations + WHERE payload_type IS NULL AND id <= @max_id::bigint +) AS untyped; + -- name: GetObservationIDRange :one SELECT COALESCE(min(id), 0)::bigint AS min_id, COALESCE(max(id), 0)::bigint AS max_id FROM packet_observations; diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index f294fd2..22607b7 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -1330,6 +1330,21 @@ func (mr *MockQuerierMockRecorder) GetTransportScopes(ctx any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetTransportScopes", reflect.TypeOf((*MockQuerier)(nil).GetTransportScopes), ctx) } +// HasUntypedObservations mocks base method. +func (m *MockQuerier) HasUntypedObservations(ctx context.Context, maxID int64) (bool, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "HasUntypedObservations", ctx, maxID) + ret0, _ := ret[0].(bool) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// HasUntypedObservations indicates an expected call of HasUntypedObservations. +func (mr *MockQuerierMockRecorder) HasUntypedObservations(ctx, maxID any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "HasUntypedObservations", reflect.TypeOf((*MockQuerier)(nil).HasUntypedObservations), ctx, maxID) +} + // InsertChannelMessage mocks base method. func (m *MockQuerier) InsertChannelMessage(ctx context.Context, arg db.InsertChannelMessageParams) (db.InsertChannelMessageRow, error) { m.ctrl.T.Helper() diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index e4a35ca..3644d6b 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -192,6 +192,9 @@ type Querier interface { GetTopNodesProbe(ctx context.Context, arg GetTopNodesProbeParams) ([]GetTopNodesProbeRow, error) GetTransportScopeByName(ctx context.Context, name string) (int32, error) GetTransportScopes(ctx context.Context) ([]GetTransportScopesRow, error) + // Whether any observation up to @max_id lacks a payload type. A full scan, run at most once per + // process by the payload-type backfill. + HasUntypedObservations(ctx context.Context, maxID int64) (bool, error) // ============================================================ // CHANNEL MESSAGES // ============================================================ diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 89b7d1f..ed09c11 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -2694,6 +2694,22 @@ func (q *Queries) GetTransportScopes(ctx context.Context) ([]GetTransportScopesR return items, nil } +const hasUntypedObservations = `-- name: HasUntypedObservations :one +SELECT EXISTS ( + SELECT 1 FROM packet_observations + WHERE payload_type IS NULL AND id <= $1::bigint +) AS untyped +` + +// Whether any observation up to @max_id lacks a payload type. A full scan, run at most once per +// process by the payload-type backfill. +func (q *Queries) HasUntypedObservations(ctx context.Context, maxID int64) (bool, error) { + row := q.db.QueryRow(ctx, hasUntypedObservations, maxID) + var untyped bool + err := row.Scan(&untyped) + return untyped, err +} + const insertChannelMessage = `-- name: InsertChannelMessage :one WITH inserted AS ( diff --git a/internal/background/payload_type_backfill.go b/internal/background/payload_type_backfill.go index 6f93865..0a85daa 100644 --- a/internal/background/payload_type_backfill.go +++ b/internal/background/payload_type_backfill.go @@ -21,6 +21,7 @@ const ( type payloadTypeBackfiller interface { ObservationIDRange(context.Context) (int64, int64, error) + HasUntypedObservations(context.Context, int64) (bool, error) BackfillObservationPayloadTypes(context.Context, int64, int64, int32) (int64, int64, error) LoadState(context.Context, string, any) (bool, error) SaveState(context.Context, string, any) error @@ -36,7 +37,16 @@ func PayloadTypeBackfillTask(store payloadTypeBackfiller, interval time.Duration // maxRows of them; when it hits maxRows it continues after the last row it picked. It saves the // next id after every batch, so a failed or interrupted run resumes where it stopped, and it saves done when a batch // slot finds the walk past the max (at once for an empty table). Once done, a run only reads the state. +// +// Before its first batch, a task value asks once whether any observation up to the max lacks a +// payload type. When none does it saves done at once instead of walking the whole id range. That +// answer stays true: ingest writes payload_type on every observation it stores, so no untyped row +// can appear later, and the walk would have typed nothing. Done only switches the adverts reader to +// the statement that filters on the observation's own type, which returns the same rows once every +// observation carries one; it just switches sooner. When the probe finds an untyped row, or fails +// (logged once), the task walks as before and never probes again. func payloadTypeBackfillTask(store payloadTypeBackfiller, interval time.Duration, span int64, maxRows int32, batches int) Task { + probed := false return Task{ Name: "payload_type_backfill", Interval: interval, @@ -52,6 +62,21 @@ func payloadTypeBackfillTask(store payloadTypeBackfiller, interval time.Duration if err != nil { return fmt.Errorf("payload type backfill: %w", err) } + if maxID > 0 && !probed { + probed = true + untyped, err := store.HasUntypedObservations(ctx, maxID) + switch { + case err != nil: + slog.Warn("observation payload type probe failed; walking the id range", "component", "background", "error", err) + case !untyped: + next := maxID + 1 + if err := store.SaveState(ctx, db.ObservationPayloadTypeKey, db.ObservationPayloadTypeState{NextID: next, Done: true}); err != nil { + return fmt.Errorf("payload type backfill: %w", err) + } + slog.Info("observation payload types complete", "component", "background", "next_id", next, "probe", true) + return nil + } + } next := max(st.NextID, minID) var rows int64 defer func() { diff --git a/internal/background/payload_type_backfill_integration_test.go b/internal/background/payload_type_backfill_integration_test.go index 6106068..8f56bd9 100644 --- a/internal/background/payload_type_backfill_integration_test.go +++ b/internal/background/payload_type_backfill_integration_test.go @@ -6,11 +6,18 @@ package background import ( "context" "maps" + "reflect" "slices" + "strings" "testing" "time" "github.com/MeshCore-Beacon/beacon-server/db" + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/google/uuid" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" ) // Packets 1..2,100 of types 2, 4, 5 and 9 (packet i has type [2,4,5,9][i%4]), one observation each. @@ -260,3 +267,172 @@ WHERE po.id = ANY($1) AND po.payload_type = p.payload_type`, untyped); n != int6 } typedFromPacket("after the capped walk") } + +// advertStatementRecorder records the sqlc name of every adverts statement read through it. +type advertStatementRecorder struct { + sqlc.DBTX + names []string +} + +func (r *advertStatementRecorder) record(sql string) { + if rest, ok := strings.CutPrefix(sql, "-- name: "); ok { + if name, _, _ := strings.Cut(rest, " "); strings.HasPrefix(name, "ListObserverAdverts") { + r.names = append(r.names, name) + } + } +} + +func (r *advertStatementRecorder) Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error) { + r.record(sql) + return r.DBTX.Query(ctx, sql, args...) +} + +func (r *advertStatementRecorder) QueryRow(ctx context.Context, sql string, args ...any) pgx.Row { + r.record(sql) + return r.DBTX.QueryRow(ctx, sql, args...) +} + +// TestPayloadTypeBackfillProbePostgres checks the one-time probe on a real store: an empty and a +// fully typed table are done after one run without a batch, and the adverts gate then switches to +// the observation-typed statement with the same pages; a table with untyped rows still walks. +func TestPayloadTypeBackfillProbePostgres(t *testing.T) { + const ( + first int64 = 2_147_483_641 + span int64 = 500 + maxRows int32 = 50_000 + ) + observer := uuid.MustParse("00000000-0000-0000-0000-0000000000a1") + + setup := func(t *testing.T, fixture string) (context.Context, *pgxpool.Pool, *db.Store, func(string) int64) { + t.Helper() + ctx, pool := migratedPool(t) + if fixture != "" { + if _, err := pool.Exec(ctx, fixture); err != nil { + t.Fatal(err) + } + } + count := func(query string) int64 { + t.Helper() + var n int64 + if err := pool.QueryRow(ctx, query).Scan(&n); err != nil { + t.Fatal(err) + } + return n + } + return ctx, pool, db.New(pool, pool, 0, 0), count + } + state := func(t *testing.T, ctx context.Context, store *db.Store) (db.ObservationPayloadTypeState, bool) { + t.Helper() + var st db.ObservationPayloadTypeState + found, err := store.LoadState(ctx, db.ObservationPayloadTypeKey, &st) + if err != nil { + t.Fatal(err) + } + return st, found + } + + t.Run("empty table", func(t *testing.T) { + ctx, _, store, _ := setup(t, "") + rec := &backfillFromRecorder{Store: store} + if err := payloadTypeBackfillTask(rec, time.Minute, span, maxRows, 10).Run(ctx); err != nil { + t.Fatal(err) + } + if st, found := state(t, ctx, store); !found || st != (db.ObservationPayloadTypeState{Done: true}) || len(rec.from) != 0 { + t.Fatalf("state %+v (found %v), batches from %v", st, found, rec.from) + } + }) + + t.Run("fully typed table", func(t *testing.T) { + ctx, pool, store, count := setup(t, payloadTypeBackfillFixture+` +UPDATE packet_observations po SET payload_type = p.payload_type +FROM packets p WHERE p.packet_hash = po.packet_hash AND po.payload_type IS DISTINCT FROM p.payload_type;`) + if n := count(`SELECT count(*) FROM packet_observations WHERE payload_type IS NULL`); n != 0 { + t.Fatalf("fixture: %d untyped rows", n) + } + maxID := count(`SELECT max(id) FROM packet_observations`) + xmins := count(`SELECT sum(xmin::text::bigint) FROM packet_observations`) + + rec := &advertStatementRecorder{DBTX: pool} + apiStore := db.New(rec, nil, 0, 0, db.WithPayloadTypeGate()) + pages := func(t *testing.T, want string) []api.Page[api.AdvertObservation] { + t.Helper() + var out []api.Page[api.AdvertObservation] + for cursor := int64(0); ; { + p, err := apiStore.ListObserverAdverts(ctx, observer, cursor, 97) + if err != nil { + t.Fatal(err) + } + out = append(out, p) + if !p.HasMore { + break + } + cursor = *p.NextCursor + } + for _, n := range rec.names { + if n != want { + t.Fatalf("statements %v, want only %s", rec.names, want) + } + } + rec.names = nil + return out + } + before := pages(t, "ListObserverAdvertsJoinFilter") + if len(before) < 5 { + t.Fatalf("fixture: %d adverts pages", len(before)) + } + + backfill := &backfillFromRecorder{Store: store} + if err := payloadTypeBackfillTask(backfill, time.Minute, span, maxRows, 10).Run(ctx); err != nil { + t.Fatal(err) + } + if st, found := state(t, ctx, store); !found || st != (db.ObservationPayloadTypeState{NextID: maxID + 1, Done: true}) { + t.Fatalf("state %+v (found %v), want done past %d", st, found, maxID) + } + if len(backfill.from) != 0 { + t.Fatalf("batches from %v, want none", backfill.from) + } + if got := count(`SELECT sum(xmin::text::bigint) FROM packet_observations`); got != xmins { + t.Fatal("the probe rewrote observations") + } + if after := pages(t, "ListObserverAdverts"); !reflect.DeepEqual(after, before) { + t.Fatalf("adverts pages changed after the switch:\n%+v\n%+v", after, before) + } + }) + + t.Run("untyped rows walk", func(t *testing.T) { + ctx, _, store, count := setup(t, payloadTypeBackfillFixture) + if n := count(`SELECT count(*) FROM packet_observations WHERE payload_type IS NULL`); n != 686 { + t.Fatalf("fixture: %d untyped rows", n) + } + rec := &backfillFromRecorder{Store: store} + task := payloadTypeBackfillTask(rec, time.Minute, span, maxRows, 1) + if err := task.Run(ctx); err != nil { + t.Fatal(err) + } + if st, _ := state(t, ctx, store); st != (db.ObservationPayloadTypeState{NextID: first + span}) { + t.Fatalf("after one batch: state %+v", st) + } + for i := 0; i < 10; i++ { + if st, _ := state(t, ctx, store); st.Done { + break + } + if err := task.Run(ctx); err != nil { + t.Fatal(err) + } + } + maxID := count(`SELECT max(id) FROM packet_observations`) + if st, _ := state(t, ctx, store); st != (db.ObservationPayloadTypeState{NextID: maxID + 1, Done: true}) { + t.Fatalf("state %+v", st) + } + if !slices.Equal(rec.from, []int64{first, first + span, first + 2*span, first + 3*span, first + 4*span}) { + t.Fatalf("batches from %v", rec.from) + } + if n := count(`SELECT count(*) FROM packet_observations WHERE payload_type IS NULL`); n != 0 { + t.Fatalf("%d rows still untyped", n) + } + if n := count(`SELECT count(*) FROM packet_observations po JOIN packets p ON p.packet_hash = po.packet_hash +WHERE po.payload_type IS DISTINCT FROM p.payload_type`); n != 1 { + t.Fatalf("%d rows differ from their packet, want only the fixture's mismatch", n) + } + }) +} diff --git a/internal/background/payload_type_backfill_test.go b/internal/background/payload_type_backfill_test.go index 44f2c6c..4f8feea 100644 --- a/internal/background/payload_type_backfill_test.go +++ b/internal/background/payload_type_backfill_test.go @@ -4,9 +4,11 @@ package background import ( + "bytes" "context" "encoding/json" "errors" + "log/slog" "slices" "testing" "time" @@ -21,7 +23,8 @@ type idRange struct{ from, to int64 } // fakeBackfiller records every call and holds the state in memory. A batch types the untyped ids // in its range, in order, up to its row cap, as the SQL does. failCall makes the Nth backfill call -// (1-based) fail with a deadlock; saveErr fails every SaveState. +// (1-based) fail with a deadlock; saveErr fails every SaveState. The probe answers probe, or true +// (some observation is untyped) when probe is nil, so a task walks unless a test says otherwise. type fakeBackfiller struct { state map[string]json.RawMessage minID int64 @@ -32,6 +35,8 @@ type fakeBackfiller struct { caps []int32 failCall int saveErr error + probe func() (bool, error) + probes []int64 // the maxID of every probe } func newFakeBackfiller(minID, maxID int64, untyped ...int64) *fakeBackfiller { @@ -45,6 +50,14 @@ func (f *fakeBackfiller) ObservationIDRange(context.Context) (int64, int64, erro return f.minID, f.maxID, nil } +func (f *fakeBackfiller) HasUntypedObservations(_ context.Context, maxID int64) (bool, error) { + f.probes = append(f.probes, maxID) + if f.probe == nil { + return true, nil + } + return f.probe() +} + func (f *fakeBackfiller) BackfillObservationPayloadTypes(_ context.Context, from, to int64, maxRows int32) (int64, int64, error) { f.batches = append(f.batches, idRange{from, to}) f.caps = append(f.caps, maxRows) @@ -262,3 +275,160 @@ func TestPayloadTypeBackfillTask(t *testing.T) { } }) } + +// captureLogs sends slog to a JSON buffer for the rest of the test and returns the records it holds. +func captureLogs(t *testing.T) func() []map[string]any { + t.Helper() + var out bytes.Buffer + previous := slog.Default() + slog.SetDefault(slog.New(slog.NewJSONHandler(&out, &slog.HandlerOptions{Level: slog.LevelDebug}))) + t.Cleanup(func() { slog.SetDefault(previous) }) + return func() []map[string]any { + var records []map[string]any + for _, line := range bytes.Split(bytes.TrimSpace(out.Bytes()), []byte("\n")) { + if len(line) == 0 { + continue + } + var r map[string]any + if err := json.Unmarshal(line, &r); err != nil { + t.Fatal(err) + } + records = append(records, r) + } + return records + } +} + +func TestPayloadTypeBackfillTaskProbe(t *testing.T) { + ctx := context.Background() + const span = 1_000_000 + run := func(t *testing.T, task Task) { + t.Helper() + if err := task.Run(ctx); err != nil { + t.Fatal(err) + } + } + answer := func(untyped bool, err error) func() (bool, error) { + return func() (bool, error) { return untyped, err } + } + + t.Run("nothing untyped saves done without walking", func(t *testing.T) { + f := newFakeBackfiller(324, 1_000_000) + f.probe = answer(false, nil) + logs := captureLogs(t) + task := payloadTypeBackfillTask(f, time.Minute, span, 1000, 10) + run(t, task) + if len(f.batches) != 0 || !slices.Equal(f.probes, []int64{1_000_000}) { + t.Fatalf("batches %v, probes %v; want no batch and one probe up to the max", f.batches, f.probes) + } + if got := f.saved(t); got != (db.ObservationPayloadTypeState{NextID: 1_000_001, Done: true}) { + t.Fatalf("state %+v, want done past the max", got) + } + complete := 0 + for _, r := range logs() { + if r["msg"] == "observation payload types complete" { + complete++ + if r["probe"] != true || r["next_id"] != float64(1_000_001) { + t.Fatalf("completion record %v", r) + } + } + } + if complete != 1 { + t.Fatalf("%d completion records, want 1", complete) + } + ranges := f.ranges + run(t, task) + if f.ranges != ranges || len(f.probes) != 1 || len(f.batches) != 0 { + t.Fatalf("a done task read the range, probed (%v) or walked (%v)", f.probes, f.batches) + } + }) + + t.Run("nothing untyped after a partial walk saves done", func(t *testing.T) { + f := newFakeBackfiller(324, 3_000_000) + f.probe = answer(false, nil) + if err := f.SaveState(ctx, db.ObservationPayloadTypeKey, db.ObservationPayloadTypeState{NextID: 5000}); err != nil { + t.Fatal(err) + } + run(t, payloadTypeBackfillTask(f, time.Minute, span, 1000, 10)) + if len(f.batches) != 0 || len(f.probes) != 1 { + t.Fatalf("batches %v, probes %v", f.batches, f.probes) + } + if got := f.saved(t); got != (db.ObservationPayloadTypeState{NextID: 3_000_001, Done: true}) { + t.Fatalf("state %+v", got) + } + }) + + t.Run("an untyped row walks and never probes again", func(t *testing.T) { + f := newFakeBackfiller(1, 3_500_000, 2_000_000) + f.probe = answer(true, nil) + task := payloadTypeBackfillTask(f, time.Minute, span, 1000, 2) + run(t, task) + if got := f.takeBatches(); !slices.Equal(got, []idRange{{1, 1_000_001}, {1_000_001, 2_000_001}}) { + t.Fatalf("batches %v", got) + } + f.probe = answer(false, nil) // a second probe would now finish at once + run(t, task) + run(t, task) + if got := f.takeBatches(); !slices.Equal(got, []idRange{{2_000_001, 3_000_001}, {3_000_001, 3_500_001}}) { + t.Fatalf("batches %v", got) + } + if got := f.saved(t); got != (db.ObservationPayloadTypeState{NextID: 3_500_001, Done: true}) { + t.Fatalf("state %+v", got) + } + if len(f.probes) != 1 || len(f.untyped) != 0 { + t.Fatalf("probes %v, still untyped %v", f.probes, f.untyped) + } + }) + + t.Run("a probe error walks and warns once", func(t *testing.T) { + f := newFakeBackfiller(1, 2_500_000) + probeErr := errors.New("probe failed") + f.probe = answer(false, probeErr) + logs := captureLogs(t) + task := payloadTypeBackfillTask(f, time.Minute, span, 1000, 2) + run(t, task) + run(t, task) + if got := f.takeBatches(); !slices.Equal(got, []idRange{{1, 1_000_001}, {1_000_001, 2_000_001}, {2_000_001, 2_500_001}}) { + t.Fatalf("batches %v", got) + } + if got := f.saved(t); got != (db.ObservationPayloadTypeState{NextID: 2_500_001, Done: true}) { + t.Fatalf("state %+v", got) + } + if len(f.probes) != 1 { + t.Fatalf("probes %v, want one", f.probes) + } + var warns []map[string]any + for _, r := range logs() { + if r["level"] == "WARN" { + warns = append(warns, r) + } + } + if len(warns) != 1 || warns[0]["error"] != "probe failed" { + t.Fatalf("warnings %v, want one carrying the probe error", warns) + } + }) + + t.Run("an empty table is done without a probe", func(t *testing.T) { + f := newFakeBackfiller(0, 0) + f.probe = answer(false, nil) + run(t, payloadTypeBackfillTask(f, time.Minute, span, 1000, 2)) + if len(f.probes) != 0 || len(f.batches) != 0 { + t.Fatalf("probes %v, batches %v", f.probes, f.batches) + } + if got := f.saved(t); got != (db.ObservationPayloadTypeState{Done: true}) { + t.Fatalf("state %+v", got) + } + }) + + t.Run("a failed save of done fails the run", func(t *testing.T) { + f := newFakeBackfiller(1, 1_000) + f.probe = answer(false, nil) + f.saveErr = errors.New("save failed") + if err := payloadTypeBackfillTask(f, time.Minute, span, 1000, 2).Run(ctx); !errors.Is(err, f.saveErr) { + t.Fatalf("err %v, want the save error", err) + } + if len(f.batches) != 0 { + t.Fatalf("batches %v", f.batches) + } + }) +}