diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b225269..b0c45cd 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -48,7 +48,7 @@ jobs: run: go test ./... - name: Verify scope refresh concurrency - run: go test -race ./internal/meshmapper ./internal/scopestore + run: go test -race ./internal/meshmapper ./internal/scopestore && go test -race ./db -run '^TestChannelIDCache' - name: Verify backup startup against PostgreSQL 16 env: @@ -58,7 +58,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|Reconfirm[A-Za-z]*|BigintCursors|RouteSearchHopOrder|StatsNodeTypesSplit|ListKnownRoutes|ListKnownRoutesMatchesLegacy|KnownRoutesCursorTies|KnownRoutesUnanalyzedRegion|KnownRoutesHopStatementsStayCustom|KnownRoutePaginationTies|RouteRetentionCutoffsExact|DeleteExpiredRoutes|DeleteGraceRoutesSlice|GraceWatermarkHeldByLockedRow|OpenPoolsSettings|PlanRouterClassCUnnamed|APIStoreHopStatementsUnnamed|PgStatStatementsMigrationGuarded|DBStatements|DBStatementsNoExtension|PgStatStatementsMigrationNoCreatableSchema|IndexHealthBaselineOnReindex|UnusedIndexes|PgActivity|RollHourSetsWorkMem)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|Reconfirm[A-Za-z]*|BigintCursors|RouteSearchHopOrder|StatsNodeTypesSplit|ListKnownRoutes|ListKnownRoutesMatchesLegacy|KnownRoutesCursorTies|KnownRoutesUnanalyzedRegion|KnownRoutesHopStatementsStayCustom|KnownRoutePaginationTies|RouteRetentionCutoffsExact|DeleteExpiredRoutes|DeleteGraceRoutesSlice|GraceWatermarkHeldByLockedRow|OpenPoolsSettings|PlanRouterClassCUnnamed|APIStoreHopStatementsUnnamed|PgStatStatementsMigrationGuarded|DBStatements|DBStatementsNoExtension|PgStatStatementsMigrationNoCreatableSchema|IndexHealthBaselineOnReindex|UnusedIndexes|PgActivity|RollHourSetsWorkMem|SmallTableIndexSet|NodeReadersAfterIndexDrops|ObserverChannelReadersAfterIndexDrops|UpsertChannelWriteThrough|UpsertChannelHashOnlyWriteThrough)Postgres$' -count=1 -v - name: Verify route retention task against PostgreSQL 16 env: diff --git a/cmd/beacon/main.go b/cmd/beacon/main.go index daef262..6efebc1 100644 --- a/cmd/beacon/main.go +++ b/cmd/beacon/main.go @@ -158,9 +158,16 @@ func main() { // routes each statement by plan class through the guards; it never runs background jobs. enforce, _ := guards.ParseFamilies(resolved.GuardsEnforce) // validated by config.Load guard := guards.New(guards.Options{Enforce: enforce, ForceObserve: layout == planpolicy.Single, Budget: resolved.GuardsBudget}) - ingestStore := db.New(pools.Ingest, pools.Ingest, resolved.ClockDriftThreshold, resolved.NodeStaleThreshold) - maintStore := db.New(pools.Maint, pools.Maint, resolved.ClockDriftThreshold, resolved.NodeStaleThreshold) + // Ingest and maintenance share one channel id cache; the API never writes channels. + channelIDs := db.NewChannelIDCache() + ingestStore := db.New(pools.Ingest, pools.Ingest, resolved.ClockDriftThreshold, resolved.NodeStaleThreshold, db.WithChannelIDs(channelIDs)) + maintStore := db.New(pools.Maint, pools.Maint, resolved.ClockDriftThreshold, resolved.NodeStaleThreshold, db.WithChannelIDs(channelIDs)) apiStore := db.NewAPIStore(pools, layout, guard, resolved.ClockDriftThreshold, resolved.NodeStaleThreshold) + if err := maintStore.PreloadChannelIDs(ctx); err != nil { + slog.Warn("channel id preload failed; known channels upsert until cached", "component", "startup", "error", err) + } else { + slog.Info(fmt.Sprintf("preloaded %d channel ids", channelIDs.Len()), "component", "startup") + } // ── Presence write coalescing ──────────────────────────────────────────── // Ingest writes go through the coalescer; reads use the API store. diff --git a/db/channel_ids.go b/db/channel_ids.go new file mode 100644 index 0000000..cc4c2c0 --- /dev/null +++ b/db/channel_ids.go @@ -0,0 +1,93 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "context" + "sync" +) + +// channelKey identifies a channels row by its conflict key. Hash-only rows have a +// NULL key_fingerprint, which is distinct from an empty one. +type channelKey struct { + hash, fp string + hashOnly bool +} + +// ChannelIDCache maps channel keys to row ids so a hearing of a known channel can +// update its row by id instead of running an upsert, which burns a sequence value +// every time. It has no eviction: hash-only rows are bounded by the 256 one-byte +// hashes and keyed rows by the configured and imported keys. Safe for concurrent use. +type ChannelIDCache struct { + mu sync.RWMutex + ids map[channelKey]int32 +} + +func NewChannelIDCache() *ChannelIDCache { + return &ChannelIDCache{ids: make(map[channelKey]int32)} +} + +// Len returns the number of cached keys. +func (c *ChannelIDCache) Len() int { + c.mu.RLock() + defer c.mu.RUnlock() + return len(c.ids) +} + +func (c *ChannelIDCache) get(k channelKey) (int32, bool) { + c.mu.RLock() + defer c.mu.RUnlock() + id, ok := c.ids[k] + return id, ok +} + +func (c *ChannelIDCache) put(k channelKey, id int32) { + c.mu.Lock() + defer c.mu.Unlock() + c.ids[k] = id +} + +func (c *ChannelIDCache) forget(k channelKey) { + c.mu.Lock() + defer c.mu.Unlock() + delete(c.ids, k) +} + +// replace swaps in m as the whole cache; the caller must not use m afterwards. +func (c *ChannelIDCache) replace(m map[channelKey]int32) { + c.mu.Lock() + defer c.mu.Unlock() + c.ids = m +} + +// Option configures a Store built by New. +type Option func(*Store) + +// WithChannelIDs makes the store write channel hearings through c. Stores that write +// channels share one cache. +func WithChannelIDs(c *ChannelIDCache) Option { + return func(s *Store) { s.channelIDs = c } +} + +// PreloadChannelIDs loads every channel key and id into the store's cache. A store +// without a cache does nothing. On error the cache is left as it was. +func (s *Store) PreloadChannelIDs(ctx context.Context) error { + if s.channelIDs == nil { + return nil + } + rows, err := s.q.ListChannelIDs(ctx) + if err != nil { + return err + } + m := make(map[channelKey]int32, len(rows)) + for _, r := range rows { + if r.KeyFingerprint == nil { + m[channelKey{hash: string(r.ChannelHash), hashOnly: true}] = r.ID + } else { + m[channelKey{hash: string(r.ChannelHash), fp: string(r.KeyFingerprint)}] = r.ID + } + } + s.channelIDs.replace(m) + return nil +} diff --git a/db/channels.go b/db/channels.go index c292b76..d988112 100644 --- a/db/channels.go +++ b/db/channels.go @@ -16,6 +16,9 @@ import ( "github.com/jackc/pgx/v5/pgtype" ) +// UpsertChannel records a hearing of a keyed (or, with a nil fingerprint, hash-only) +// channel and returns its id. With a cache, a known key is touched by id; a touch that +// finds no row (deleted or re-keyed since cached) falls back to the upsert. func (s *Store) UpsertChannel(ctx context.Context, channelHash []byte, keyFingerprint []byte, name string, hashtag string) (int, error) { var namePtr, hashtagPtr *string if name != "" { @@ -24,6 +27,25 @@ func (s *Store) UpsertChannel(ctx context.Context, channelHash []byte, keyFinger if hashtag != "" { hashtagPtr = &hashtag } + cached := s.channelIDs != nil && keyFingerprint != nil + key := channelKey{hash: string(channelHash), fp: string(keyFingerprint)} + if cached { + if id, ok := s.channelIDs.get(key); ok { + touched, err := s.q.TouchChannel(ctx, sqlc.TouchChannelParams{ + Name: namePtr, + ID: id, + ChannelHash: channelHash, + KeyFingerprint: keyFingerprint, + }) + if err == nil { + return int(touched), nil + } + if !errors.Is(err, pgx.ErrNoRows) { + return 0, err + } + s.channelIDs.forget(key) + } + } isHashtag := hashtag != "" row, err := s.q.UpsertChannel(ctx, sqlc.UpsertChannelParams{ ChannelHash: channelHash, @@ -35,14 +57,34 @@ func (s *Store) UpsertChannel(ctx context.Context, channelHash []byte, keyFinger if err != nil { return 0, err } + if cached { + s.channelIDs.put(key, row.ID) + } return int(row.ID), nil } +// UpsertChannelHashOnly is UpsertChannel for a channel whose key is unknown. func (s *Store) UpsertChannelHashOnly(ctx context.Context, channelHash []byte) (int, error) { + key := channelKey{hash: string(channelHash), hashOnly: true} + if s.channelIDs != nil { + if id, ok := s.channelIDs.get(key); ok { + touched, err := s.q.TouchChannelHashOnly(ctx, sqlc.TouchChannelHashOnlyParams{ID: id, ChannelHash: channelHash}) + if err == nil { + return int(touched), nil + } + if !errors.Is(err, pgx.ErrNoRows) { + return 0, err + } + s.channelIDs.forget(key) + } + } rowID, err := s.q.UpsertChannelHashOnly(ctx, channelHash) if err != nil { return 0, err } + if s.channelIDs != nil { + s.channelIDs.put(key, rowID) + } return int(rowID), nil } diff --git a/db/channels_integration_test.go b/db/channels_integration_test.go new file mode 100644 index 0000000..9e31817 --- /dev/null +++ b/db/channels_integration_test.go @@ -0,0 +1,285 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "bytes" + "context" + "fmt" + "reflect" + "testing" + + sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" + "github.com/jackc/pgx/v5" +) + +// channelFixture inserts 10 keyed channels (0x01..0x0a, fingerprint = the hash byte +// repeated 8 times, name "ch-N") and 5 hash-only channels (0x21..0x25) through the +// sequence default, all last seen long before the test transaction's NOW(). +func channelFixture(t *testing.T, ctx context.Context, tx pgx.Tx) { + t.Helper() + mustExec(t, ctx, tx, ` +INSERT INTO channels (channel_hash, key_fingerprint, name, key_known, first_seen, last_seen) +SELECT set_byte('\x00'::bytea, 0, h), decode(repeat(lpad(to_hex(h), 2, '0'), 8), 'hex'), 'ch-' || h, true, + '2026-01-01'::timestamptz, '2026-01-02'::timestamptz +FROM generate_series(1, 10) h ORDER BY h`) + mustExec(t, ctx, tx, ` +INSERT INTO channels (channel_hash, first_seen, last_seen) +SELECT set_byte('\x00'::bytea, 0, h), '2026-01-01'::timestamptz, '2026-01-02'::timestamptz +FROM generate_series(33, 37) h ORDER BY h`) +} + +func fp8(h byte) []byte { return bytes.Repeat([]byte{h}, 8) } + +func channelSnap(t *testing.T, ctx context.Context, tx pgx.Tx) []sqlc.Channel { + t.Helper() + rows, err := tx.Query(ctx, `SELECT * FROM channels ORDER BY channel_hash, key_fingerprint NULLS FIRST`) + if err != nil { + t.Fatal(err) + } + out, err := pgx.CollectRows(rows, pgx.RowToStructByPos[sqlc.Channel]) + if err != nil { + t.Fatal(err) + } + return out +} + +func channelSeq(t *testing.T, ctx context.Context, tx pgx.Tx) int64 { + t.Helper() + var v int64 + if err := tx.QueryRow(ctx, `SELECT last_value FROM channels_id_seq`).Scan(&v); err != nil { + t.Fatal(err) + } + return v +} + +func maxChannelID(t *testing.T, ctx context.Context, tx pgx.Tx) int32 { + t.Helper() + var v int32 + if err := tx.QueryRow(ctx, `SELECT max(id) FROM channels`).Scan(&v); err != nil { + t.Fatal(err) + } + return v +} + +// legacyWant runs hear through the legacy store inside a savepoint and returns the +// table it leaves, then rolls the rows back (the sequence keeps what it burned). +func legacyWant(t *testing.T, ctx context.Context, tx pgx.Tx, hear func(*Store)) []sqlc.Channel { + t.Helper() + mustExec(t, ctx, tx, `SAVEPOINT ref`) + hear(&Store{q: sqlc.New(tx)}) + want := channelSnap(t, ctx, tx) + mustExec(t, ctx, tx, `ROLLBACK TO SAVEPOINT ref`) + return want +} + +// sameChannels compares every column of every row; newRow (if set) matches its row on +// every column but id, and that id must be above floor. +func sameChannels(t *testing.T, got, want []sqlc.Channel, newRow func(sqlc.Channel) bool, floor int32) { + t.Helper() + if len(got) != len(want) { + t.Fatalf("got %d rows, want %d", len(got), len(want)) + } + for i := range got { + g, w := got[i], want[i] + if newRow != nil && newRow(g) { + if g.ID <= floor { + t.Errorf("new row id %d not above %d", g.ID, floor) + } + g.ID, w.ID = 0, 0 + } + if !reflect.DeepEqual(g, w) { + t.Errorf("row %d:\n got %+v\nwant %+v", i, g, w) + } + } +} + +func isChannel(hash byte, fp []byte) func(sqlc.Channel) bool { + return func(c sqlc.Channel) bool { + return bytes.Equal(c.ChannelHash, []byte{hash}) && bytes.Equal(c.KeyFingerprint, fp) && (fp == nil) == (c.KeyFingerprint == nil) + } +} + +// After the boot preload, known-key hearings burn no sequence value and leave today's rows. +func TestUpsertChannelWriteThroughPostgres(t *testing.T) { + ctx, tx := retentionTx(t) + applyBaseline(t, ctx, tx) + channelFixture(t, ctx, tx) + fixtureMax := maxChannelID(t, ctx, tx) + cache := NewChannelIDCache() + wt := &Store{q: sqlc.New(tx), channelIDs: cache} + if err := wt.PreloadChannelIDs(ctx); err != nil { + t.Fatal(err) + } + if cache.Len() != 15 { + t.Fatalf("preloaded %d keys, want 15", cache.Len()) + } + + hear := func(s *Store, h byte, name string, n int) { + for i := 0; i < n; i++ { + if _, err := s.UpsertChannel(ctx, []byte{h}, fp8(h), name, ""); err != nil { + t.Fatal(err) + } + } + } + stream := func(s *Store) { + for i := 0; i < 1000; i++ { + name := "" + if i%3 == 0 { + name = fmt.Sprintf("name-%d", i%7) + } + hear(s, byte(1+i%10), name, 1) + } + } + + want := legacyWant(t, ctx, tx, stream) + seq0 := channelSeq(t, ctx, tx) + stream(wt) + if seq1 := channelSeq(t, ctx, tx); seq1 != seq0 { + t.Fatalf("known keys burned %d sequence values", seq1-seq0) + } + sameChannels(t, channelSnap(t, ctx, tx), want, nil, 0) + + t.Run("unknown key", func(t *testing.T) { + want := legacyWant(t, ctx, tx, func(s *Store) { hear(s, 0x0b, "new", 5) }) + seq0 := channelSeq(t, ctx, tx) + hear(wt, 0x0b, "new", 5) + if d := channelSeq(t, ctx, tx) - seq0; d != 1 { + t.Fatalf("sequence moved %d, want 1", d) + } + if cache.Len() != 16 { + t.Fatalf("cache.Len() = %d, want 16", cache.Len()) + } + sameChannels(t, channelSnap(t, ctx, tx), want, isChannel(0x0b, fp8(0x0b)), fixtureMax) + }) + + t.Run("deleted", func(t *testing.T) { + mustExec(t, ctx, tx, `DELETE FROM channels WHERE channel_hash = '\x03'`) + want := legacyWant(t, ctx, tx, func(s *Store) { hear(s, 0x03, "back", 5) }) + seq0 := channelSeq(t, ctx, tx) + hear(wt, 0x03, "back", 5) + if d := channelSeq(t, ctx, tx) - seq0; d != 1 { + t.Fatalf("sequence moved %d, want 1", d) + } + got := channelSnap(t, ctx, tx) + sameChannels(t, got, want, isChannel(0x03, fp8(0x03)), fixtureMax) + for _, c := range got { + if isChannel(0x03, fp8(0x03))(c) { + if id, ok := cache.get(channelKey{hash: "\x03", fp: string(fp8(0x03))}); !ok || id != c.ID { + t.Fatalf("cache holds %d, %v; want new id %d", id, ok, c.ID) + } + } + } + }) + + t.Run("re-keyed", func(t *testing.T) { + mustExec(t, ctx, tx, `UPDATE channels SET key_fingerprint = '\xffffffffffffffff', name = 'rekeyed', + last_seen = '2000-01-01' WHERE channel_hash = '\x04'`) + before := channelSnap(t, ctx, tx) + want := legacyWant(t, ctx, tx, func(s *Store) { hear(s, 0x04, "fresh", 5) }) + seq0 := channelSeq(t, ctx, tx) + hear(wt, 0x04, "fresh", 5) + if d := channelSeq(t, ctx, tx) - seq0; d != 1 { + t.Fatalf("sequence moved %d, want 1", d) + } + got := channelSnap(t, ctx, tx) + if len(got) != len(before)+1 { + t.Fatalf("got %d rows, want %d (a new (0x04, fp4) row)", len(got), len(before)+1) + } + sameChannels(t, got, want, isChannel(0x04, fp8(0x04)), fixtureMax) + rekeyed := isChannel(0x04, fp8(0xff)) + for i := range before { + if rekeyed(before[i]) { + for _, c := range got { + if rekeyed(c) && !reflect.DeepEqual(c, before[i]) { + t.Fatalf("re-keyed row changed:\n got %+v\nwant %+v", c, before[i]) + } + } + } + } + }) +} + +func TestUpsertChannelHashOnlyWriteThroughPostgres(t *testing.T) { + ctx, tx := retentionTx(t) + applyBaseline(t, ctx, tx) + channelFixture(t, ctx, tx) + // A keyed row beside the hash-only 0x21 that the hash-only path must never touch. + mustExec(t, ctx, tx, `INSERT INTO channels (channel_hash, key_fingerprint, name, key_known, first_seen, last_seen) + VALUES ('\x21', '\x2121212121212121', 'keyed-21', true, '2026-01-01', '2026-01-02')`) + fixtureMax := maxChannelID(t, ctx, tx) + keyedBefore := channelSnap(t, ctx, tx) + cache := NewChannelIDCache() + wt := &Store{q: sqlc.New(tx), channelIDs: cache} + if err := wt.PreloadChannelIDs(ctx); err != nil { + t.Fatal(err) + } + if cache.Len() != 16 { + t.Fatalf("preloaded %d keys, want 16", cache.Len()) + } + + hear := func(s *Store, h byte, n int) { + for i := 0; i < n; i++ { + if _, err := s.UpsertChannelHashOnly(ctx, []byte{h}); err != nil { + t.Fatal(err) + } + } + } + stream := func(s *Store) { + for i := 0; i < 1000; i++ { + hear(s, byte(0x21+i%5), 1) + } + } + + want := legacyWant(t, ctx, tx, stream) + seq0 := channelSeq(t, ctx, tx) + stream(wt) + if seq1 := channelSeq(t, ctx, tx); seq1 != seq0 { + t.Fatalf("known keys burned %d sequence values", seq1-seq0) + } + got := channelSnap(t, ctx, tx) + sameChannels(t, got, want, nil, 0) + keyed21 := isChannel(0x21, fp8(0x21)) + for _, b := range keyedBefore { + if keyed21(b) { + for _, c := range got { + if keyed21(c) && !reflect.DeepEqual(c, b) { + t.Fatalf("keyed 0x21 row touched:\n got %+v\nwant %+v", c, b) + } + } + } + } + + t.Run("unknown key", func(t *testing.T) { + want := legacyWant(t, ctx, tx, func(s *Store) { hear(s, 0x26, 5) }) + seq0 := channelSeq(t, ctx, tx) + hear(wt, 0x26, 5) + if d := channelSeq(t, ctx, tx) - seq0; d != 1 { + t.Fatalf("sequence moved %d, want 1", d) + } + if cache.Len() != 17 { + t.Fatalf("cache.Len() = %d, want 17", cache.Len()) + } + sameChannels(t, channelSnap(t, ctx, tx), want, isChannel(0x26, nil), fixtureMax) + }) + + t.Run("deleted", func(t *testing.T) { + mustExec(t, ctx, tx, `DELETE FROM channels WHERE channel_hash = '\x23' AND key_fingerprint IS NULL`) + want := legacyWant(t, ctx, tx, func(s *Store) { hear(s, 0x23, 5) }) + seq0 := channelSeq(t, ctx, tx) + hear(wt, 0x23, 5) + if d := channelSeq(t, ctx, tx) - seq0; d != 1 { + t.Fatalf("sequence moved %d, want 1", d) + } + got := channelSnap(t, ctx, tx) + sameChannels(t, got, want, isChannel(0x23, nil), fixtureMax) + for _, c := range got { + if isChannel(0x23, nil)(c) { + if id, ok := cache.get(channelKey{hash: "\x23", hashOnly: true}); !ok || id != c.ID { + t.Fatalf("cache holds %d, %v; want new id %d", id, ok, c.ID) + } + } + } + }) +} diff --git a/db/channels_test.go b/db/channels_test.go index be0fd46..4807f06 100644 --- a/db/channels_test.go +++ b/db/channels_test.go @@ -6,12 +6,14 @@ package db import ( "context" "errors" + "sync" "testing" "time" sqlc "github.com/MeshCore-Beacon/beacon-server/db/sqlc" mockdb "github.com/MeshCore-Beacon/beacon-server/db/sqlc/mock" "github.com/MeshCore-Beacon/beacon-server/internal/api" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgtype" "go.uber.org/mock/gomock" ) @@ -353,3 +355,338 @@ func TestListChannelMessages_DBError(t *testing.T) { t.Fatal("expected error, got nil") } } + +var testFP = []byte{1, 2, 3, 4, 5, 6, 7, 8} + +func strPtr(s string) *string { return &s } + +func TestUpsertChannelWriteThroughCached(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + cache := NewChannelIDCache() + cache.put(channelKey{hash: "\x11", fp: string(testFP)}, 7) + store := &Store{q: mock, channelIDs: cache} + + mock.EXPECT().UpsertChannel(gomock.Any(), gomock.Any()).Times(0) + gomock.InOrder( + mock.EXPECT().TouchChannel(gomock.Any(), sqlc.TouchChannelParams{ + Name: strPtr("pub"), ID: 7, ChannelHash: []byte{0x11}, KeyFingerprint: testFP, + }).Return(int32(7), nil), + mock.EXPECT().TouchChannel(gomock.Any(), sqlc.TouchChannelParams{ + Name: nil, ID: 7, ChannelHash: []byte{0x11}, KeyFingerprint: testFP, + }).Return(int32(7), nil), + ) + + for _, name := range []string{"pub", ""} { + id, err := store.UpsertChannel(context.Background(), []byte{0x11}, testFP, name, "") + if err != nil { + t.Fatal(err) + } + if id != 7 { + t.Fatalf("name %q: id = %d, want 7", name, id) + } + } +} + +func upsertParams(hash, fp []byte, name, hashtag string) sqlc.UpsertChannelParams { + p := sqlc.UpsertChannelParams{ChannelHash: hash, Column2: fp} + if name != "" { + p.Name = &name + } + if hashtag != "" { + p.Hashtag = &hashtag + } + isHashtag := hashtag != "" + p.IsHashtag = &isHashtag + return p +} + +func TestUpsertChannelWriteThroughFallback(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + cache := NewChannelIDCache() + key := channelKey{hash: "\x11", fp: string(testFP)} + cache.put(key, 7) + store := &Store{q: mock, channelIDs: cache} + + gomock.InOrder( + mock.EXPECT().TouchChannel(gomock.Any(), sqlc.TouchChannelParams{ + ID: 7, ChannelHash: []byte{0x11}, KeyFingerprint: testFP, + }).Return(int32(0), pgx.ErrNoRows), + mock.EXPECT().UpsertChannel(gomock.Any(), upsertParams([]byte{0x11}, testFP, "", "#tag")). + Return(sqlc.Channel{ID: 9}, nil), + mock.EXPECT().TouchChannel(gomock.Any(), sqlc.TouchChannelParams{ + ID: 9, ChannelHash: []byte{0x11}, KeyFingerprint: testFP, + }).Return(int32(9), nil), + ) + + for i := 0; i < 2; i++ { + id, err := store.UpsertChannel(context.Background(), []byte{0x11}, testFP, "", "#tag") + if err != nil { + t.Fatal(err) + } + if id != 9 { + t.Fatalf("call %d: id = %d, want 9", i, id) + } + } + if got, ok := cache.get(key); !ok || got != 9 { + t.Fatalf("cache = %d, %v; want 9", got, ok) + } +} + +func TestUpsertChannelWriteThroughUnknownKeyCached(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + cache := NewChannelIDCache() + store := &Store{q: mock, channelIDs: cache} + + gomock.InOrder( + mock.EXPECT().UpsertChannel(gomock.Any(), upsertParams([]byte{0x11}, testFP, "pub", "")). + Return(sqlc.Channel{ID: 4}, nil), + mock.EXPECT().TouchChannel(gomock.Any(), sqlc.TouchChannelParams{ + Name: strPtr("pub"), ID: 4, ChannelHash: []byte{0x11}, KeyFingerprint: testFP, + }).Return(int32(4), nil), + ) + for i := 0; i < 2; i++ { + id, err := store.UpsertChannel(context.Background(), []byte{0x11}, testFP, "pub", "") + if err != nil || id != 4 { + t.Fatalf("call %d: %d, %v; want 4", i, id, err) + } + } + if cache.Len() != 1 { + t.Fatalf("cache.Len() = %d, want 1", cache.Len()) + } +} + +func TestUpsertChannelTouchErrorReturned(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + cache := NewChannelIDCache() + key := channelKey{hash: "\x11", fp: string(testFP)} + cache.put(key, 7) + store := &Store{q: mock, channelIDs: cache} + boom := errors.New("boom") + + mock.EXPECT().TouchChannel(gomock.Any(), gomock.Any()).Return(int32(0), boom) + mock.EXPECT().UpsertChannel(gomock.Any(), gomock.Any()).Times(0) + + if _, err := store.UpsertChannel(context.Background(), []byte{0x11}, testFP, "", ""); !errors.Is(err, boom) { + t.Fatalf("err = %v, want boom", err) + } + if got, ok := cache.get(key); !ok || got != 7 { + t.Fatalf("cache = %d, %v; want 7 kept", got, ok) + } +} + +func TestUpsertChannelNilFingerprintBypassesCache(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + cache := NewChannelIDCache() + store := &Store{q: mock, channelIDs: cache} + + mock.EXPECT().UpsertChannel(gomock.Any(), upsertParams([]byte{0x11}, nil, "pub", "")). + Return(sqlc.Channel{ID: 5}, nil).Times(2) + mock.EXPECT().TouchChannel(gomock.Any(), gomock.Any()).Times(0) + mock.EXPECT().TouchChannelHashOnly(gomock.Any(), gomock.Any()).Times(0) + + for i := 0; i < 2; i++ { + id, err := store.UpsertChannel(context.Background(), []byte{0x11}, nil, "pub", "") + if err != nil || id != 5 { + t.Fatalf("call %d: %d, %v; want 5", i, id, err) + } + } + if cache.Len() != 0 { + t.Fatalf("cache.Len() = %d, want 0", cache.Len()) + } +} + +func TestUpsertChannelWithoutCacheUnchanged(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + store := &Store{q: mock} + + mock.EXPECT().UpsertChannel(gomock.Any(), upsertParams([]byte{0x11}, testFP, "pub", "")). + Return(sqlc.Channel{ID: 3}, nil).Times(2) + mock.EXPECT().UpsertChannelHashOnly(gomock.Any(), []byte{0x22}).Return(int32(6), nil).Times(2) + mock.EXPECT().TouchChannel(gomock.Any(), gomock.Any()).Times(0) + mock.EXPECT().TouchChannelHashOnly(gomock.Any(), gomock.Any()).Times(0) + + for i := 0; i < 2; i++ { + if id, err := store.UpsertChannel(context.Background(), []byte{0x11}, testFP, "pub", ""); err != nil || id != 3 { + t.Fatalf("UpsertChannel: %d, %v", id, err) + } + if id, err := store.UpsertChannelHashOnly(context.Background(), []byte{0x22}); err != nil || id != 6 { + t.Fatalf("UpsertChannelHashOnly: %d, %v", id, err) + } + } + if err := store.PreloadChannelIDs(context.Background()); err != nil { + t.Fatal(err) + } +} + +func TestUpsertChannelHashOnlyWriteThrough(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + cache := NewChannelIDCache() + key := channelKey{hash: "\x22", hashOnly: true} + cache.put(key, 5) + store := &Store{q: mock, channelIDs: cache} + + mock.EXPECT().UpsertChannel(gomock.Any(), gomock.Any()).Times(0) + gomock.InOrder( + mock.EXPECT().TouchChannelHashOnly(gomock.Any(), sqlc.TouchChannelHashOnlyParams{ID: 5, ChannelHash: []byte{0x22}}). + Return(int32(5), nil), + mock.EXPECT().TouchChannelHashOnly(gomock.Any(), sqlc.TouchChannelHashOnlyParams{ID: 5, ChannelHash: []byte{0x22}}). + Return(int32(0), pgx.ErrNoRows), + mock.EXPECT().UpsertChannelHashOnly(gomock.Any(), []byte{0x22}).Return(int32(8), nil), + mock.EXPECT().TouchChannelHashOnly(gomock.Any(), sqlc.TouchChannelHashOnlyParams{ID: 8, ChannelHash: []byte{0x22}}). + Return(int32(8), nil), + ) + for i, want := range []int{5, 8, 8} { + id, err := store.UpsertChannelHashOnly(context.Background(), []byte{0x22}) + if err != nil || id != want { + t.Fatalf("call %d: %d, %v; want %d", i, id, err, want) + } + } + if got, ok := cache.get(key); !ok || got != 8 { + t.Fatalf("cache = %d, %v; want 8", got, ok) + } +} + +func TestChannelIDCacheKeys(t *testing.T) { + c := NewChannelIDCache() + hashOnly := channelKey{hash: "\x11", hashOnly: true} + empty := channelKey{hash: "\x11", fp: ""} + keyed := channelKey{hash: "\x11", fp: string(testFP)} + c.put(hashOnly, 1) + c.put(empty, 2) + c.put(keyed, 3) + if c.Len() != 3 { + t.Fatalf("Len = %d, want 3 distinct keys", c.Len()) + } + for k, want := range map[channelKey]int32{hashOnly: 1, empty: 2, keyed: 3} { + if got, ok := c.get(k); !ok || got != want { + t.Fatalf("get(%+v) = %d, %v; want %d", k, got, ok, want) + } + } + c.forget(empty) + if _, ok := c.get(empty); ok { + t.Fatal("forgotten key still cached") + } + if _, ok := c.get(hashOnly); !ok { + t.Fatal("forget removed a different key") + } + c.put(keyed, 4) + if got, _ := c.get(keyed); got != 4 { + t.Fatalf("put did not overwrite: %d", got) + } + c.replace(map[channelKey]int32{{hash: "\x99", hashOnly: true}: 9}) + if c.Len() != 1 { + t.Fatalf("Len after replace = %d, want 1", c.Len()) + } + if _, ok := c.get(keyed); ok { + t.Fatal("replace kept an old entry") + } + if got, ok := c.get(channelKey{hash: "\x99", hashOnly: true}); !ok || got != 9 { + t.Fatalf("replaced entry = %d, %v", got, ok) + } +} + +func TestChannelIDCacheConcurrent(t *testing.T) { + c := NewChannelIDCache() + keys := make([]channelKey, 64) + for i := range keys { + keys[i] = channelKey{hash: string([]byte{byte(i)}), fp: string([]byte{byte(i), 1}), hashOnly: i%2 == 0} + } + var wg sync.WaitGroup + for g := 0; g < 16; g++ { + wg.Add(1) + go func(g int) { + defer wg.Done() + for i := 0; i < 1000; i++ { + k := keys[(g*31+i)%len(keys)] + switch i % 4 { + case 0: + c.get(k) + case 1: + c.put(k, int32(i)) + case 2: + c.forget(k) + case 3: + if i%100 == 3 { + c.replace(map[channelKey]int32{k: int32(g)}) + } else { + c.Len() + } + } + } + }(g) + } + wg.Wait() + if c.Len() > len(keys) { + t.Fatalf("Len = %d > %d keys", c.Len(), len(keys)) + } +} + +func TestPreloadChannelIDs(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + cache := NewChannelIDCache() + store := &Store{q: mock, channelIDs: cache} + + mock.EXPECT().ListChannelIDs(gomock.Any()).Return([]sqlc.ListChannelIDsRow{ + {ID: 1, ChannelHash: []byte{0x11}, KeyFingerprint: testFP}, + {ID: 2, ChannelHash: []byte{0x12}, KeyFingerprint: []byte{}}, + {ID: 3, ChannelHash: []byte{0x11}, KeyFingerprint: nil}, + }, nil) + if err := store.PreloadChannelIDs(context.Background()); err != nil { + t.Fatal(err) + } + if cache.Len() != 3 { + t.Fatalf("Len = %d, want 3", cache.Len()) + } + for k, want := range map[channelKey]int32{ + {hash: "\x11", fp: string(testFP)}: 1, + {hash: "\x12", fp: ""}: 2, + {hash: "\x11", hashOnly: true}: 3, + } { + if got, ok := cache.get(k); !ok || got != want { + t.Fatalf("get(%+v) = %d, %v; want %d", k, got, ok, want) + } + } + + boom := errors.New("boom") + mock.EXPECT().ListChannelIDs(gomock.Any()).Return(nil, boom) + if err := store.PreloadChannelIDs(context.Background()); !errors.Is(err, boom) { + t.Fatalf("err = %v, want boom", err) + } + if cache.Len() != 3 { + t.Fatalf("cache changed on error: Len = %d", cache.Len()) + } + + nilStore := &Store{q: mock} + mock.EXPECT().ListChannelIDs(gomock.Any()).Times(0) + if err := nilStore.PreloadChannelIDs(context.Background()); err != nil { + t.Fatal(err) + } +} + +func TestUpsertChannelFallbackErrorForgetsKey(t *testing.T) { + ctrl := gomock.NewController(t) + mock := mockdb.NewMockQuerier(ctrl) + cache := NewChannelIDCache() + key := channelKey{hash: "\x11", fp: string(testFP)} + cache.put(key, 7) + store := &Store{q: mock, channelIDs: cache} + boom := errors.New("boom") + + mock.EXPECT().TouchChannel(gomock.Any(), gomock.Any()).Return(int32(0), pgx.ErrNoRows) + mock.EXPECT().UpsertChannel(gomock.Any(), gomock.Any()).Return(sqlc.Channel{}, boom) + + if _, err := store.UpsertChannel(context.Background(), []byte{0x11}, testFP, "", ""); !errors.Is(err, boom) { + t.Fatalf("err = %v, want boom", err) + } + if _, ok := cache.get(key); ok { + t.Fatal("a key whose row is gone stayed cached") + } +} diff --git a/db/migrations/011_node_iatas_iata_v2.sql b/db/migrations/011_node_iatas_iata_v2.sql new file mode 100644 index 0000000..0c474a4 --- /dev/null +++ b/db/migrations/011_node_iatas_iata_v2.sql @@ -0,0 +1,5 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- Region filters probe iata only; last_heard in the key made every node_iatas update non-HOT. +CREATE INDEX CONCURRENTLY idx_node_iatas_iata_v2 ON node_iatas USING btree (iata); diff --git a/db/migrations/012_drop_node_iatas_iata.sql b/db/migrations/012_drop_node_iatas_iata.sql new file mode 100644 index 0000000..5758b2a --- /dev/null +++ b/db/migrations/012_drop_node_iatas_iata.sql @@ -0,0 +1,5 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- Replaced by idx_node_iatas_iata_v2, renamed next. +DROP INDEX CONCURRENTLY IF EXISTS idx_node_iatas_iata; diff --git a/db/migrations/013_rename_node_iatas_iata_v2.sql b/db/migrations/013_rename_node_iatas_iata_v2.sql new file mode 100644 index 0000000..f167cf6 --- /dev/null +++ b/db/migrations/013_rename_node_iatas_iata_v2.sql @@ -0,0 +1,5 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- Keep the index name the (iata) foreign key cascade and the plans have always used. +ALTER INDEX idx_node_iatas_iata_v2 RENAME TO idx_node_iatas_iata; diff --git a/db/migrations/014_drop_nodes_type_last_seen.sql b/db/migrations/014_drop_nodes_type_last_seen.sql new file mode 100644 index 0000000..1122add --- /dev/null +++ b/db/migrations/014_drop_nodes_type_last_seen.sql @@ -0,0 +1,5 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- Custom typeName pages seq-scan instead; every advert update was non-HOT for it. +DROP INDEX CONCURRENTLY IF EXISTS idx_nodes_type_last_seen; diff --git a/db/migrations/015_drop_nodes_pubkey.sql b/db/migrations/015_drop_nodes_pubkey.sql new file mode 100644 index 0000000..52601b6 --- /dev/null +++ b/db/migrations/015_drop_nodes_pubkey.sql @@ -0,0 +1,5 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- Duplicates nodes_public_key_key. +DROP INDEX CONCURRENTLY IF EXISTS idx_nodes_pubkey; diff --git a/db/migrations/016_drop_nodes_location.sql b/db/migrations/016_drop_nodes_location.sql new file mode 100644 index 0000000..7cfdac2 --- /dev/null +++ b/db/migrations/016_drop_nodes_location.sql @@ -0,0 +1,5 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- No statement filters nodes by coordinates. +DROP INDEX CONCURRENTLY IF EXISTS idx_nodes_location; diff --git a/db/migrations/017_drop_nodes_min_firmware.sql b/db/migrations/017_drop_nodes_min_firmware.sql new file mode 100644 index 0000000..28bb0df --- /dev/null +++ b/db/migrations/017_drop_nodes_min_firmware.sql @@ -0,0 +1,5 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- No statement filters on min_firmware_version. +DROP INDEX CONCURRENTLY IF EXISTS idx_nodes_min_firmware; diff --git a/db/migrations/018_drop_nodes_multibyte_paths.sql b/db/migrations/018_drop_nodes_multibyte_paths.sql new file mode 100644 index 0000000..82703cc --- /dev/null +++ b/db/migrations/018_drop_nodes_multibyte_paths.sql @@ -0,0 +1,5 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- The multibyte filters cost a few dozen extra buffers on a seq scan. +DROP INDEX CONCURRENTLY IF EXISTS idx_nodes_multibyte_paths; diff --git a/db/migrations/019_drop_nodes_multibyte_traces.sql b/db/migrations/019_drop_nodes_multibyte_traces.sql new file mode 100644 index 0000000..2cefb2a --- /dev/null +++ b/db/migrations/019_drop_nodes_multibyte_traces.sql @@ -0,0 +1,5 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- The multibyte filters cost a few dozen extra buffers on a seq scan. +DROP INDEX CONCURRENTLY IF EXISTS idx_nodes_multibyte_traces; diff --git a/db/migrations/020_drop_observers_last_seen.sql b/db/migrations/020_drop_observers_last_seen.sql new file mode 100644 index 0000000..b9c1c21 --- /dev/null +++ b/db/migrations/020_drop_observers_last_seen.sql @@ -0,0 +1,5 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- Observer pages sort a small heap; the index made every presence flush non-HOT. +DROP INDEX CONCURRENTLY IF EXISTS idx_observers_last_seen; diff --git a/db/migrations/021_drop_observers_pubkey.sql b/db/migrations/021_drop_observers_pubkey.sql new file mode 100644 index 0000000..1921f77 --- /dev/null +++ b/db/migrations/021_drop_observers_pubkey.sql @@ -0,0 +1,5 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- Duplicates observers_public_key_key. +DROP INDEX CONCURRENTLY IF EXISTS idx_observers_pubkey; diff --git a/db/migrations/022_drop_observers_type.sql b/db/migrations/022_drop_observers_type.sql new file mode 100644 index 0000000..8e5493b --- /dev/null +++ b/db/migrations/022_drop_observers_type.sql @@ -0,0 +1,5 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- Never scanned; the type filter reads a small heap. +DROP INDEX CONCURRENTLY IF EXISTS idx_observers_type; diff --git a/db/migrations/023_drop_channels_last_seen_id.sql b/db/migrations/023_drop_channels_last_seen_id.sql new file mode 100644 index 0000000..11a7099 --- /dev/null +++ b/db/migrations/023_drop_channels_last_seen_id.sql @@ -0,0 +1,5 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- Channel pages sort a small heap; the index made every channel touch non-HOT. +DROP INDEX CONCURRENTLY IF EXISTS idx_channels_last_seen_id; diff --git a/db/migrations/024_observers_channels_fillfactor.sql b/db/migrations/024_observers_channels_fillfactor.sql new file mode 100644 index 0000000..c551a80 --- /dev/null +++ b/db/migrations/024_observers_channels_fillfactor.sql @@ -0,0 +1,6 @@ +-- Copyright 2026 Beacon Contributors +-- SPDX-License-Identifier: AGPL-3.0-or-later + +-- Room on each page for HOT updates of the hottest rows. +ALTER TABLE observers SET (fillfactor = 50); +ALTER TABLE channels SET (fillfactor = 50); diff --git a/db/queries/queries.sql b/db/queries/queries.sql index 59ae9e2..74422dc 100644 --- a/db/queries/queries.sql +++ b/db/queries/queries.sql @@ -1035,6 +1035,25 @@ ON CONFLICT (channel_hash) WHERE key_fingerprint IS NULL DO UPDATE SET last_seen = NOW() RETURNING id; +-- name: ListChannelIDs :many +-- Every channel key and its id, preloaded at boot for the ingest write-through. +SELECT id, channel_hash, key_fingerprint FROM channels; + +-- name: TouchChannel :one +-- Write-through for a cached keyed channel: the UpsertChannel conflict update, by id. +-- No row (deleted or re-keyed since cached) sends the caller back to UpsertChannel. +UPDATE channels SET + last_seen = NOW(), + name = COALESCE(sqlc.narg(name), name) +WHERE id = sqlc.arg(id) AND channel_hash = sqlc.arg(channel_hash) AND key_fingerprint = sqlc.arg(key_fingerprint) +RETURNING id; + +-- name: TouchChannelHashOnly :one +-- Write-through for a cached hash-only channel: the UpsertChannelHashOnly conflict update, by id. +UPDATE channels SET last_seen = NOW() +WHERE id = sqlc.arg(id) AND channel_hash = sqlc.arg(channel_hash) AND key_fingerprint IS NULL +RETURNING id; + -- name: ListUndecryptedGroupTextPackets :many -- Returns GRP_TXT packets (payload_type=5) never successfully decrypted. Used at boot to -- retry decryption against the current keystore for packets whose channel key was only added @@ -1108,7 +1127,7 @@ LIMIT @page_limit; -- name: ListChannelsAfter :many -- Keep the non-null tuple boundary separate from the legacy optional cursor so --- generic prepared plans can seek directly into the composite ordered index. +-- generic prepared plans see a plain row comparison on (last_seen, id). SELECT c.* FROM channels c WHERE (c.last_seen, c.id) < (@cursor_ts::timestamptz, @cursor_id::integer) AND (@channel_hash::bytea IS NULL OR c.channel_hash = @channel_hash) diff --git a/db/small_tables_integration_test.go b/db/small_tables_integration_test.go new file mode 100644 index 0000000..b03fb71 --- /dev/null +++ b/db/small_tables_integration_test.go @@ -0,0 +1,491 @@ +// Copyright 2026 Beacon Contributors +// SPDX-License-Identifier: AGPL-3.0-or-later + +package db + +import ( + "context" + "errors" + "reflect" + "slices" + "strconv" + "strings" + "testing" + "time" + + 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" +) + +// smallTableMigrations rebuild node_iatas' region index and drop the small-table indexes +// that kept ingest updates from being HOT. +var smallTableMigrations = []string{ + "011_node_iatas_iata_v2.sql", + "012_drop_node_iatas_iata.sql", + "013_rename_node_iatas_iata_v2.sql", + "014_drop_nodes_type_last_seen.sql", + "015_drop_nodes_pubkey.sql", + "016_drop_nodes_location.sql", + "017_drop_nodes_min_firmware.sql", + "018_drop_nodes_multibyte_paths.sql", + "019_drop_nodes_multibyte_traces.sql", + "020_drop_observers_last_seen.sql", + "021_drop_observers_pubkey.sql", + "022_drop_observers_type.sql", + "023_drop_channels_last_seen_id.sql", + "024_observers_channels_fillfactor.sql", +} + +// sortTieRuns returns rows with each maximal run of equal key sorted by id; the order +// across runs is left alone. +func sortTieRuns[T any](rows []T, key func(T) string, id func(T) string) []T { + out := slices.Clone(rows) + for start := 0; start < len(out); { + end := start + 1 + for end < len(out) && key(out[end]) == key(out[start]) { + end++ + } + slices.SortStableFunc(out[start:end], func(a, b T) int { return strings.Compare(id(a), id(b)) }) + start = end + } + return out +} + +func TestSortTieRuns(t *testing.T) { + type row struct{ k, id string } + in := []row{{"b", "2"}, {"b", "1"}, {"a", "9"}, {"c", "5"}, {"c", "3"}, {"b", "0"}} + got := sortTieRuns(in, func(r row) string { return r.k }, func(r row) string { return r.id }) + want := []row{{"b", "1"}, {"b", "2"}, {"a", "9"}, {"c", "3"}, {"c", "5"}, {"b", "0"}} + if !reflect.DeepEqual(got, want) || in[0].id != "2" { + t.Fatalf("got %v, want %v; input changed: %v", got, want, in) + } +} + +var smallTableIndexSets = []struct { + table string + indexes []string + reloptions []string +}{ + {"nodes", []string{"nodes_pkey", "nodes_public_key_key"}, []string{"autovacuum_vacuum_scale_factor=0.05"}}, + {"node_iatas", []string{"idx_node_iatas_iata", "node_iatas_pkey"}, []string{"autovacuum_vacuum_scale_factor=0.05"}}, + {"observers", []string{"idx_observers_last_iata", "observers_pkey", "observers_public_key_key"}, []string{"fillfactor=50"}}, + {"channels", []string{"channels_channel_hash_key_fingerprint_key", "channels_hashtag_key", "channels_pkey", + "idx_channels_hash", "idx_channels_hash_no_key", "idx_channels_hashtag"}, []string{"fillfactor=50"}}, +} + +func checkSmallTableIndexSets(t *testing.T, ctx context.Context, pool *pgxpool.Pool) { + t.Helper() + for _, want := range smallTableIndexSets { + var got, bad, rel []string + if err := pool.QueryRow(ctx, `SELECT coalesce(array_agg(indexrelid::regclass::text ORDER BY indexrelid::regclass::text), '{}'), + coalesce(array_agg(indexrelid::regclass::text) FILTER (WHERE NOT (indisvalid AND indisready)), '{}'), + (SELECT coalesce(reloptions, '{}') FROM pg_class WHERE oid = $1::regclass) +FROM pg_index WHERE indrelid = $1::regclass`, want.table).Scan(&got, &bad, &rel); err != nil { + t.Fatal(err) + } + if !slices.Equal(got, want.indexes) || len(bad) != 0 || !slices.Equal(rel, want.reloptions) { + t.Errorf("%s indexes = %v, want %v; not valid and ready: %v; reloptions = %v, want %v", + want.table, got, want.indexes, bad, rel, want.reloptions) + } + } + var def string + var v2Gone bool + if err := pool.QueryRow(ctx, "SELECT pg_get_indexdef('idx_node_iatas_iata'::regclass), to_regclass('idx_node_iatas_iata_v2') IS NULL").Scan(&def, &v2Gone); err != nil { + t.Fatal(err) + } + if !strings.HasSuffix(def, "USING btree (iata)") || !v2Gone { + t.Errorf("idx_node_iatas_iata = %q, idx_node_iatas_iata_v2 gone = %v", def, v2Gone) + } +} + +func checkRecorded(t *testing.T, ctx context.Context, pool *pgxpool.Pool, files ...string) { + t.Helper() + for _, f := range files { + var recorded bool + if err := pool.QueryRow(ctx, "SELECT EXISTS(SELECT 1 FROM schema_migrations WHERE filename = $1)", f).Scan(&recorded); err != nil || !recorded { + t.Fatalf("%s not recorded: %v", f, err) + } + } +} + +// The migrations leave the four ingest-churned tables with exactly the indexes a reader or a constraint needs. +func TestSmallTableIndexSetPostgres(t *testing.T) { + ctx, pool := schemaPool(t) + if err := RunMigrations(ctx, pool); err != nil { + t.Fatal(err) + } + checkSmallTableIndexSets(t, ctx, pool) + + // An operator already dropped these by hand, or a crash lost the ledger rows: the drops re-run as no-ops. + if _, err := pool.Exec(ctx, "DELETE FROM schema_migrations WHERE filename IN ('014_drop_nodes_type_last_seen.sql', '023_drop_channels_last_seen_id.sql')"); err != nil { + t.Fatal(err) + } + if err := RunMigrations(ctx, pool); err != nil { + t.Fatalf("re-running dropped-index migrations: %v", err) + } + checkRecorded(t, ctx, pool, "014_drop_nodes_type_last_seen.sql", "023_drop_channels_last_seen_id.sql") + checkSmallTableIndexSets(t, ctx, pool) + + // A crash after the replacement index was built but before its ledger row: 011 records the valid index. + if _, err := pool.Exec(ctx, "CREATE INDEX CONCURRENTLY idx_node_iatas_iata_v2 ON node_iatas USING btree (iata)"); err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(ctx, "DELETE FROM schema_migrations WHERE filename = '011_node_iatas_iata_v2.sql'"); err != nil { + t.Fatal(err) + } + if err := RunMigrations(ctx, pool); err != nil { + t.Fatalf("011 over a pre-built index: %v", err) + } + checkRecorded(t, ctx, pool, "011_node_iatas_iata_v2.sql") + if _, err := pool.Exec(ctx, "DROP INDEX idx_node_iatas_iata_v2"); err != nil { + t.Fatal(err) + } + checkSmallTableIndexSets(t, ctx, pool) +} + +// applyMigrationsTx applies migration files inside tx, CONCURRENTLY removed so they can run there. +func applyMigrationsTx(t *testing.T, ctx context.Context, tx pgx.Tx, files ...string) { + t.Helper() + for _, f := range files { + body, err := migrationFiles.ReadFile("migrations/" + f) + if err != nil { + t.Fatal(err) + } + if _, err := tx.Exec(ctx, strings.ReplaceAll(string(body), "CONCURRENTLY ", "")); err != nil { + t.Fatalf("%s: %v", f, err) + } + } +} + +// planSettings are the four plan modes each reader runs under, before and after the drops. +// enable_seqscan off steers the old plans onto idx_nodes_type_last_seen and the node_iatas index. +var planSettings = []string{ + "SET LOCAL plan_cache_mode = force_custom_plan; SET LOCAL enable_seqscan = on", + "SET LOCAL plan_cache_mode = force_custom_plan; SET LOCAL enable_seqscan = off", + "SET LOCAL plan_cache_mode = force_generic_plan; SET LOCAL enable_seqscan = on", + "SET LOCAL plan_cache_mode = force_generic_plan; SET LOCAL enable_seqscan = off", +} + +// readerShape is one reader call; norm sorts its tie runs so equal results compare equal. +type readerShape struct { + name string + run func() (any, error) + norm func(any) any +} + +type shapeResult struct { + val any + err error +} + +func runShapes(t *testing.T, ctx context.Context, tx pgx.Tx, shapes []readerShape) map[string][]shapeResult { + t.Helper() + out := make(map[string][]shapeResult) + for _, setting := range planSettings { + mustExec(t, ctx, tx, setting) + for _, s := range shapes { + v, err := s.run() + if err != nil && !errors.Is(err, pgx.ErrNoRows) { + t.Fatalf("%s under %q: %v", s.name, setting, err) + } + out[s.name] = append(out[s.name], shapeResult{v, err}) + } + } + mustExec(t, ctx, tx, "RESET plan_cache_mode; RESET enable_seqscan") + return out +} + +func compareShapes(t *testing.T, shapes []readerShape, before, after map[string][]shapeResult, skip ...string) { + t.Helper() + for _, s := range shapes { + if slices.Contains(skip, s.name) { + continue + } + for i := range planSettings { + b, a := before[s.name][i], after[s.name][i] + if (a.err == nil) != (b.err == nil) || (a.err != nil && a.err.Error() != b.err.Error()) { + t.Errorf("%s under %q: error %v, before %v", s.name, planSettings[i], a.err, b.err) + continue + } + if a.err != nil { + continue + } + if !reflect.DeepEqual(s.norm(a.val), s.norm(b.val)) { + t.Errorf("%s under %q: rows differ after the drops\nafter: %+v\nbefore: %+v", s.name, planSettings[i], a.val, b.val) + } + } + } +} + +func timeKey(ts time.Time) string { return ts.UTC().Format(time.RFC3339Nano) } + +func queryTimes(t *testing.T, ctx context.Context, tx pgx.Tx, sql string) map[uuid.UUID]string { + t.Helper() + rows, err := tx.Query(ctx, sql) + if err != nil { + t.Fatal(err) + } + out := make(map[uuid.UUID]string) + var id uuid.UUID + var ts time.Time + if _, err := pgx.ForEachRow(rows, []any{&id, &ts}, func() error { out[id] = timeKey(ts); return nil }); err != nil { + t.Fatal(err) + } + return out +} + +// Nodes 1..3000: node i has last_seen base - i s, except nodes 140-164, which share base - 140 s +// (global ranks 140-164). +const smallTableReaderFixture = ` +INSERT INTO transport_scopes (id, name, transport_key, key_fingerprint) VALUES (1, '#test', '\x01', '\x02'); +INSERT INTO nodes (id, public_key, node_type, name, latitude, longitude, location_source, last_advert_at, + supports_multibyte_paths, supports_multibyte_traces, default_scope_id, first_seen, last_seen, + radio_freq_mhz, radio_sf, radio_bw_khz, device_clock_drift_seconds) +SELECT md5('node' || i)::uuid, + CASE WHEN i % 100 = 0 THEN '\x0000'::bytea || int4send(i) ELSE sha256(int4send(i)) END, + 1 + i % 4, 'node-' || i, + CASE WHEN i % 3 = 0 THEN 49 + i / 10000.0 END, CASE WHEN i % 3 = 0 THEN -123 - i / 10000.0 END, + CASE WHEN i % 3 = 0 THEN 'advert' END, + '2026-06-01'::timestamptz - (CASE WHEN i BETWEEN 140 AND 164 THEN 140 ELSE i END) * interval '1 second', + i % 7 = 0, i % 5 = 0, CASE WHEN i % 4 = 0 THEN 1 END, + '2026-05-01'::timestamptz + i * interval '1 second', + '2026-06-01'::timestamptz - (CASE WHEN i BETWEEN 140 AND 164 THEN 140 ELSE i END) * interval '1 second', + CASE WHEN i % 2 = 0 THEN 910.525 END, CASE WHEN i % 2 = 0 THEN 7 END, CASE WHEN i % 2 = 0 THEN 62.5 END, + CASE WHEN 1 + i % 4 NOT IN (2, 3) THEN NULL + WHEN i BETWEEN 1000 AND 1019 THEN CASE WHEN i % 2 = 0 THEN 99999 ELSE -99999 END + ELSE CASE WHEN i % 2 = 0 THEN i ELSE -i END END +FROM generate_series(1, 3000) i; +INSERT INTO node_iatas (node_id, iata, first_heard, last_heard, observation_count) +SELECT md5('node' || i)::uuid, 'YVR', '2026-05-01', '2026-06-01'::timestamptz - i * interval '2 seconds', 1 + i % 97 +FROM generate_series(1, 3000) i WHERE i % 3 <> 0; +INSERT INTO node_iatas (node_id, iata, first_heard, last_heard, observation_count) +SELECT md5('node' || i)::uuid, 'YYJ', '2026-05-01', '2026-06-01'::timestamptz - i * interval '2 seconds' - interval '1 second', 1 + i % 13 +FROM generate_series(1, 3000) i WHERE i % 5 = 0; +INSERT INTO node_neighbors (node_id, neighbor_id, iata, first_seen, last_seen, snr) +SELECT md5('node' || i)::uuid, md5('node' || (i + d))::uuid, 'YVR', '2026-05-01', '2026-06-01', d +FROM generate_series(1, 50) i, generate_series(1, 2) d; +INSERT INTO observers (id, public_key, display_name, first_seen, last_seen) +SELECT md5('observer' || k)::uuid, n.public_key, 'obs-' || k, '2026-05-01', '2026-06-01' +FROM generate_series(1, 20) k JOIN nodes n ON n.id = md5('node' || (k * 13))::uuid; +ANALYZE nodes, node_iatas, node_neighbors, observers;` + +// Rows from every reader of the dropped indexes are today's, before and after the drops, in both plan modes. +func TestNodeReadersAfterIndexDropsPostgres(t *testing.T) { + ctx, tx := retentionTx(t) + applyBaseline(t, ctx, tx) + applyMigrationsTx(t, ctx, tx, "003_meshmapper_repeaters.sql") // GetNode reads the MeshMapper tables + mustExec(t, ctx, tx, smallTableReaderFixture) + store := &Store{q: sqlc.New(tx)} + + lastSeen := queryTimes(t, ctx, tx, "SELECT id, last_seen FROM nodes") + tieRun := queryTimes(t, ctx, tx, "SELECT id, last_seen FROM nodes WHERE last_seen = '2026-06-01'::timestamptz - interval '140 seconds'") + if len(tieRun) != 25 { + t.Fatalf("tie run has %d nodes, want 25", len(tieRun)) + } + var type2Cursor, globalCursor time.Time + var busiestYVR uuid.UUID + if err := tx.QueryRow(ctx, `SELECT (SELECT last_seen FROM nodes WHERE node_type = 2 ORDER BY last_seen DESC OFFSET 399 LIMIT 1), + (SELECT last_seen FROM nodes ORDER BY last_seen DESC OFFSET 1999 LIMIT 1), + (SELECT node_id FROM node_iatas WHERE iata = 'YVR' ORDER BY observation_count DESC, node_id LIMIT 1)`).Scan(&type2Cursor, &globalCursor, &busiestYVR); err != nil { + t.Fatal(err) + } + yes := true + nodeKey := func(n api.NodeSummary) string { return lastSeen[n.ID] } + nodeID := func(n api.NodeSummary) string { return n.ID.String() } + normNodes := func(v any) any { + p := v.(api.Page[api.NodeSummary]) + p.Items = sortTieRuns(p.Items, nodeKey, nodeID) + return p + } + listNodes := func(name string, nodeType int16, iatas []string, mbPaths, mbTraces *bool, prefix, nameFilter string, cursor time.Time, limit int32, neighbors bool) readerShape { + var ms int64 + if !cursor.IsZero() { + ms = cursor.UnixMilli() + } + return readerShape{name: name, norm: normNodes, run: func() (any, error) { + return store.ListNodes(ctx, nodeType, iatas, mbPaths, mbTraces, nil, prefix, nameFilter, "", ms, limit, neighbors) + }} + } + var none time.Time + normDrift := func(v any) any { + return sortTieRuns(v.([]api.ClockDriftEntry), + func(e api.ClockDriftEntry) string { + return strconv.Itoa(max(e.ClockDriftSeconds, -e.ClockDriftSeconds)) + }, + func(e api.ClockDriftEntry) string { return e.NodeID.String() }) + } + normTypes := func(v any) any { + return sortTieRuns(v.([]api.NodeTypeCount), + func(c api.NodeTypeCount) string { return strconv.FormatInt(c.Count, 10) }, + func(c api.NodeTypeCount) string { return strconv.Itoa(int(c.NodeType)) }) + } + same := func(v any) any { return v } + shapes := []readerShape{ + listNodes("global l50", 0, nil, nil, nil, "", "", none, 50, true), + listNodes("global l150", 0, nil, nil, nil, "", "", none, 150, true), + listNodes("type1", 1, nil, nil, nil, "", "", none, 50, false), + listNodes("type2", 2, nil, nil, nil, "", "", none, 50, false), + listNodes("type3", 3, nil, nil, nil, "", "", none, 50, false), + listNodes("type2 deep cursor", 2, nil, nil, nil, "", "", type2Cursor, 50, false), + listNodes("global deep cursor l200", 0, nil, nil, nil, "", "", globalCursor, 200, false), + listNodes("YVR", 0, []string{"YVR"}, nil, nil, "", "", none, 50, false), + listNodes("YVR type2", 2, []string{"YVR"}, nil, nil, "", "", none, 50, false), + listNodes("name node-1", 0, nil, nil, nil, "", "node-1", none, 50, false), + listNodes("mbPaths", 0, nil, &yes, nil, "", "", none, 50, false), + listNodes("mbTraces", 0, nil, nil, &yes, "", "", none, 50, false), + listNodes("pubkey prefix 0000", 0, nil, nil, nil, "0000", "", none, 50, false), + listNodes("type9", 9, nil, nil, nil, "", "", none, 50, false), + {name: "clock drift", norm: normDrift, run: func() (any, error) { return store.GetStatsClockDrift(ctx, nil, 500) }}, + {name: "clock drift YVR", norm: normDrift, run: func() (any, error) { return store.GetStatsClockDrift(ctx, []string{"YVR"}, 500) }}, + {name: "node types", norm: normTypes, run: func() (any, error) { return store.GetStatsNodeTypes(ctx, nil) }}, + {name: "node types YVR", norm: normTypes, run: func() (any, error) { return store.GetStatsNodeTypes(ctx, []string{"YVR"}) }}, + {name: "node busiest YVR", norm: same, run: func() (any, error) { return store.GetNode(ctx, busiestYVR) }}, + {name: "node not found", norm: same, run: func() (any, error) { return store.GetNode(ctx, uuid.Nil) }}, + } + + before := runShapes(t, ctx, tx, shapes) + for _, r := range before["type9"] { + if items := r.val.(api.Page[api.NodeSummary]).Items; items == nil || len(items) != 0 { + t.Fatalf("type9 items = %#v, want empty non-nil", items) + } + } + for _, r := range before["node not found"] { + if !errors.Is(r.err, pgx.ErrNoRows) { + t.Fatalf("GetNode(uuid.Nil) error = %v, want no rows", r.err) + } + } + if n := len(before["clock drift"][0].val.([]api.ClockDriftEntry)); n != 500 { + t.Fatalf("clock drift returned %d rows, want 500", n) + } + + applyMigrationsTx(t, ctx, tx, smallTableMigrations...) + mustExec(t, ctx, tx, "ANALYZE nodes, node_iatas, node_neighbors, observers") + after := runShapes(t, ctx, tx, shapes) + + compareShapes(t, shapes, before, after, "global l150") + // LIMIT 150 splits the 25-node tie run at ranks 140-164: rows 1-139 are fixed, the last 11 are any 11 of the run. + for i := range planSettings { + for _, r := range []shapeResult{before["global l150"][i], after["global l150"][i]} { + p := r.val.(api.Page[api.NodeSummary]) + if len(p.Items) != 150 || !p.HasMore || p.NextCursor == nil { + t.Fatalf("global l150 under %q: %d items, hasMore %v", planSettings[i], len(p.Items), p.HasMore) + } + seen := map[uuid.UUID]bool{} + for _, n := range p.Items[139:] { + if _, ok := tieRun[n.ID]; !ok || seen[n.ID] { + t.Errorf("global l150 under %q: row %s is not a distinct member of the tie run", planSettings[i], n.ID) + } + seen[n.ID] = true + } + } + b, a := before["global l150"][i].val.(api.Page[api.NodeSummary]), after["global l150"][i].val.(api.Page[api.NodeSummary]) + if !reflect.DeepEqual(a.Items[:139], b.Items[:139]) || *a.NextCursor != *b.NextCursor { + t.Errorf("global l150 under %q: first 139 rows or cursor differ after the drops", planSettings[i]) + } + } +} + +// Observers 1..400: observer i has last_seen base - i s, except 50-61, which share base - 50 s. +// Channels 1..120: channel i has last_seen base - i s, except 30-35, which share base - 30 s. +const observerChannelReaderFixture = ` +INSERT INTO transport_scopes (id, name, transport_key, key_fingerprint) VALUES (1, '#test', '\x01', '\x02'); +INSERT INTO observers (id, public_key, display_name, observer_type, radio_freq_mhz, radio_sf, radio_bw_khz, + last_status_at, first_seen, last_seen, last_iata, last_iata_at) +SELECT md5('observer' || i)::uuid, sha256(int4send(i)), 'obs-' || i, + CASE i % 3 WHEN 0 THEN 'repeater' WHEN 1 THEN 'companion' END, + CASE WHEN i % 4 = 0 THEN 910.525 END, CASE WHEN i % 4 = 0 THEN 7 END, CASE WHEN i % 4 = 0 THEN 62.5 END, + CASE WHEN i % 2 = 0 THEN now() ELSE '2026-05-01'::timestamptz + i * interval '1 second' END, + '2026-05-01', + '2026-06-01'::timestamptz - (CASE WHEN i BETWEEN 50 AND 61 THEN 50 ELSE i END) * interval '1 second', + CASE WHEN i % 5 < 3 THEN 'YVR' WHEN i % 5 = 3 THEN 'YYJ' END, + CASE WHEN i % 5 < 4 THEN '2026-06-01'::timestamptz END +FROM generate_series(1, 400) i; +INSERT INTO observer_brokers (observer_id, broker_name, first_seen, last_seen) +SELECT md5('observer' || i)::uuid, 'mqtt1', '2026-05-01', '2026-06-01' FROM generate_series(1, 400) i WHERE i % 2 = 1; +INSERT INTO observer_scopes (observer_id, scope_id) VALUES (md5('observer7')::uuid, 1); +INSERT INTO channels (id, channel_hash, key_fingerprint, name, is_hashtag, key_known, first_seen, last_seen, message_count) +SELECT i, set_byte('\x00'::bytea, 0, i % 40), CASE WHEN i % 3 <> 0 THEN sha256(int4send(i)) END, + CASE WHEN i % 3 <> 0 THEN 'chan-' || i END, i % 6 = 1, i % 3 = 1, + '2026-05-01', + '2026-06-01'::timestamptz - (CASE WHEN i BETWEEN 30 AND 35 THEN 30 ELSE i END) * interval '1 second', i +FROM generate_series(1, 120) i; +ANALYZE observers, observer_brokers, observer_scopes, channels;` + +// ListObservers (idx_observers_last_seen, _type, _pubkey) and ListChannels (idx_channels_last_seen_id) keep their rows. +func TestObserverChannelReadersAfterIndexDropsPostgres(t *testing.T) { + ctx, tx := retentionTx(t) + applyBaseline(t, ctx, tx) + mustExec(t, ctx, tx, observerChannelReaderFixture) + store := &Store{q: sqlc.New(tx)} + + lastSeen := queryTimes(t, ctx, tx, "SELECT id, last_seen FROM observers") + var obsCursor, chanCursor time.Time + var page60 api.ChannelCursor + if err := tx.QueryRow(ctx, `SELECT (SELECT last_seen FROM observers ORDER BY last_seen DESC OFFSET 249 LIMIT 1), + (SELECT last_seen FROM channels ORDER BY last_seen DESC, id DESC OFFSET 39 LIMIT 1), + (SELECT last_seen FROM channels ORDER BY last_seen DESC, id DESC OFFSET 59 LIMIT 1), + (SELECT id FROM channels ORDER BY last_seen DESC, id DESC OFFSET 59 LIMIT 1)`).Scan(&obsCursor, &chanCursor, &page60.LastSeen, &page60.ID); err != nil { + t.Fatal(err) + } + normObservers := func(v any) any { + p := v.(api.Page[api.ObserverSummary]) + p.Items = sortTieRuns(p.Items, func(o api.ObserverSummary) string { return lastSeen[o.ID] }, + func(o api.ObserverSummary) string { return o.ID.String() }) + return p + } + listObservers := func(name string, iatas []string, observerType, broker, status, nameFilter, scope string, cursor time.Time, limit int32) readerShape { + var ms int64 + if !cursor.IsZero() { + ms = cursor.UnixMilli() + } + return readerShape{name: name, norm: normObservers, run: func() (any, error) { + return store.ListObservers(ctx, iatas, observerType, broker, status, nameFilter, scope, ms, limit) + }} + } + exact := func(v any) any { return v } + listChannels := func(name string, hash []byte, keyKnown *bool, cursor time.Time, pageCursor *api.ChannelCursor) readerShape { + var ms int64 + if !cursor.IsZero() { + ms = cursor.UnixMilli() + } + return readerShape{name: name, norm: exact, run: func() (any, error) { + return store.ListChannels(ctx, 50, hash, nil, keyKnown, ms, pageCursor) + }} + } + var none time.Time + yes := true + shapes := []readerShape{ + listObservers("observers first l201", nil, "", "", "", "", "", none, 201), + listObservers("observers legacy cursor", nil, "", "", "", "", "", obsCursor, 50), + listObservers("observers repeater", nil, "repeater", "", "", "", "", none, 50), + listObservers("observers YVR", []string{"YVR"}, "", "", "", "", "", none, 50), + listObservers("observers online", nil, "", "", "online", "", "", none, 50), + listObservers("observers mqtt1", nil, "", "mqtt1", "", "", "", none, 50), + listObservers("observers name obs-1", nil, "", "", "", "obs-1", "", none, 50), + listObservers("observers scope #test", nil, "", "", "", "", "#test", none, 50), + listChannels("channels first l50", nil, nil, none, nil), + listChannels("channels legacy cursor", nil, nil, chanCursor, nil), + listChannels("channels page cursor", nil, nil, none, &page60), + listChannels("channels key known", nil, &yes, none, nil), + listChannels("channels hash 05", []byte{0x05}, nil, none, nil), + } + + before := runShapes(t, ctx, tx, shapes) + if n := len(before["observers first l201"][0].val.(api.Page[api.ObserverSummary]).Items); n != 201 { + t.Fatalf("observers first page has %d rows, want 201", n) + } + if n := len(before["observers scope #test"][0].val.(api.Page[api.ObserverSummary]).Items); n != 1 { + t.Fatalf("observers in #test = %d, want 1", n) + } + if n := len(before["channels hash 05"][0].val.(api.ChannelPage).Items); n != 3 { + t.Fatalf("channels with hash 05 = %d, want 3", n) + } + + applyMigrationsTx(t, ctx, tx, smallTableMigrations...) + mustExec(t, ctx, tx, "ANALYZE observers, observer_brokers, observer_scopes, channels") + after := runShapes(t, ctx, tx, shapes) + compareShapes(t, shapes, before, after) +} diff --git a/db/sqlc/mock/querier.go b/db/sqlc/mock/querier.go index 96645b2..143eea6 100644 --- a/db/sqlc/mock/querier.go +++ b/db/sqlc/mock/querier.go @@ -1329,6 +1329,21 @@ func (mr *MockQuerierMockRecorder) ListChannelCatalogues(ctx any) *gomock.Call { return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListChannelCatalogues", reflect.TypeOf((*MockQuerier)(nil).ListChannelCatalogues), ctx) } +// ListChannelIDs mocks base method. +func (m *MockQuerier) ListChannelIDs(ctx context.Context) ([]db.ListChannelIDsRow, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "ListChannelIDs", ctx) + ret0, _ := ret[0].([]db.ListChannelIDsRow) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// ListChannelIDs indicates an expected call of ListChannelIDs. +func (mr *MockQuerierMockRecorder) ListChannelIDs(ctx any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ListChannelIDs", reflect.TypeOf((*MockQuerier)(nil).ListChannelIDs), ctx) +} + // ListChannelMessages mocks base method. func (m *MockQuerier) ListChannelMessages(ctx context.Context, arg db.ListChannelMessagesParams) ([]db.ListChannelMessagesRow, error) { m.ctrl.T.Helper() @@ -2514,6 +2529,36 @@ func (mr *MockQuerierMockRecorder) SetRollupHourPartial(ctx, hour any) *gomock.C return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SetRollupHourPartial", reflect.TypeOf((*MockQuerier)(nil).SetRollupHourPartial), ctx, hour) } +// TouchChannel mocks base method. +func (m *MockQuerier) TouchChannel(ctx context.Context, arg db.TouchChannelParams) (int32, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "TouchChannel", ctx, arg) + ret0, _ := ret[0].(int32) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// TouchChannel indicates an expected call of TouchChannel. +func (mr *MockQuerierMockRecorder) TouchChannel(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TouchChannel", reflect.TypeOf((*MockQuerier)(nil).TouchChannel), ctx, arg) +} + +// TouchChannelHashOnly mocks base method. +func (m *MockQuerier) TouchChannelHashOnly(ctx context.Context, arg db.TouchChannelHashOnlyParams) (int32, error) { + m.ctrl.T.Helper() + ret := m.ctrl.Call(m, "TouchChannelHashOnly", ctx, arg) + ret0, _ := ret[0].(int32) + ret1, _ := ret[1].(error) + return ret0, ret1 +} + +// TouchChannelHashOnly indicates an expected call of TouchChannelHashOnly. +func (mr *MockQuerierMockRecorder) TouchChannelHashOnly(ctx, arg any) *gomock.Call { + mr.mock.ctrl.T.Helper() + return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "TouchChannelHashOnly", reflect.TypeOf((*MockQuerier)(nil).TouchChannelHashOnly), ctx, arg) +} + // TouchObserverBrokers mocks base method. func (m *MockQuerier) TouchObserverBrokers(ctx context.Context, arg db.TouchObserverBrokersParams) error { m.ctrl.T.Helper() diff --git a/db/sqlc/querier.go b/db/sqlc/querier.go index 2448ff2..b01660f 100644 --- a/db/sqlc/querier.go +++ b/db/sqlc/querier.go @@ -196,6 +196,8 @@ type Querier interface { // Pass cursor=0 to start from the beginning. ListAllChannelMessages(ctx context.Context, arg ListAllChannelMessagesParams) ([]ListAllChannelMessagesRow, error) ListChannelCatalogues(ctx context.Context) ([]MeshmapperChannelCatalogue, error) + // Every channel key and its id, preloaded at boot for the ingest write-through. + ListChannelIDs(ctx context.Context) ([]ListChannelIDsRow, error) // Returns messages for a channel identified by integer ID. // Pass a zero/null timestamp for since to return all messages up to limit. // Pass empty string for iata to skip IATA filtering. @@ -212,7 +214,7 @@ type Querier interface { // Pass cursor=0 to start from the beginning (cursor is last_seen epoch ms). ListChannels(ctx context.Context, arg ListChannelsParams) ([]Channel, error) // Keep the non-null tuple boundary separate from the legacy optional cursor so - // generic prepared plans can seek directly into the composite ordered index. + // generic prepared plans see a plain row comparison on (last_seen, id). ListChannelsAfter(ctx context.Context, arg ListChannelsAfterParams) ([]Channel, error) // Only complete hours whose raw rows are still intact can be re-rolled. Queued missing hours // wait: their first roll consumes the entry only if the entry predates its snapshot. @@ -397,6 +399,11 @@ type Querier interface { SetNodeMultibyteTraces(ctx context.Context, id uuid.UUID) error SetPacketDecrypted(ctx context.Context, packetHash []byte) error SetRollupHourPartial(ctx context.Context, hour pgtype.Timestamptz) error + // Write-through for a cached keyed channel: the UpsertChannel conflict update, by id. + // No row (deleted or re-keyed since cached) sends the caller back to UpsertChannel. + TouchChannel(ctx context.Context, arg TouchChannelParams) (int32, error) + // Write-through for a cached hash-only channel: the UpsertChannelHashOnly conflict update, by id. + TouchChannelHashOnly(ctx context.Context, arg TouchChannelHashOnlyParams) (int32, error) TouchObserverBrokers(ctx context.Context, arg TouchObserverBrokersParams) error // Batched flush of coalesced presence bumps. GREATEST keeps a late flush // from regressing a newer write-through (e.g. a status update). diff --git a/db/sqlc/queries.sql.go b/db/sqlc/queries.sql.go index 337f430..d0e8455 100644 --- a/db/sqlc/queries.sql.go +++ b/db/sqlc/queries.sql.go @@ -2736,6 +2736,37 @@ func (q *Queries) ListChannelCatalogues(ctx context.Context) ([]MeshmapperChanne return items, nil } +const listChannelIDs = `-- name: ListChannelIDs :many +SELECT id, channel_hash, key_fingerprint FROM channels +` + +type ListChannelIDsRow struct { + ID int32 `json:"id"` + ChannelHash []byte `json:"channel_hash"` + KeyFingerprint []byte `json:"key_fingerprint"` +} + +// Every channel key and its id, preloaded at boot for the ingest write-through. +func (q *Queries) ListChannelIDs(ctx context.Context) ([]ListChannelIDsRow, error) { + rows, err := q.db.Query(ctx, listChannelIDs) + if err != nil { + return nil, err + } + defer rows.Close() + items := []ListChannelIDsRow{} + for rows.Next() { + var i ListChannelIDsRow + if err := rows.Scan(&i.ID, &i.ChannelHash, &i.KeyFingerprint); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const listChannelMessages = `-- name: ListChannelMessages :many SELECT DISTINCT ON (cm.id) cm.id, cm.channel_id, cm.packet_hash, cm.sender_name, cm.sender_pubkey, cm.content, cm.sent_at, encode(cm.packet_hash, 'hex') as packet_hash_hex, c.channel_hash, ts.name AS scope_name, p.transport_codes_present, p.observation_count AS observation_count @@ -2996,7 +3027,7 @@ type ListChannelsAfterParams struct { } // Keep the non-null tuple boundary separate from the legacy optional cursor so -// generic prepared plans can seek directly into the composite ordered index. +// generic prepared plans see a plain row comparison on (last_seen, id). func (q *Queries) ListChannelsAfter(ctx context.Context, arg ListChannelsAfterParams) ([]Channel, error) { rows, err := q.db.Query(ctx, listChannelsAfter, arg.CursorTs, @@ -5840,6 +5871,54 @@ func (q *Queries) SetPacketDecrypted(ctx context.Context, packetHash []byte) err return err } +const touchChannel = `-- name: TouchChannel :one +UPDATE channels SET + last_seen = NOW(), + name = COALESCE($1, name) +WHERE id = $2 AND channel_hash = $3 AND key_fingerprint = $4 +RETURNING id +` + +type TouchChannelParams struct { + Name *string `json:"name"` + ID int32 `json:"id"` + ChannelHash []byte `json:"channel_hash"` + KeyFingerprint []byte `json:"key_fingerprint"` +} + +// Write-through for a cached keyed channel: the UpsertChannel conflict update, by id. +// No row (deleted or re-keyed since cached) sends the caller back to UpsertChannel. +func (q *Queries) TouchChannel(ctx context.Context, arg TouchChannelParams) (int32, error) { + row := q.db.QueryRow(ctx, touchChannel, + arg.Name, + arg.ID, + arg.ChannelHash, + arg.KeyFingerprint, + ) + var id int32 + err := row.Scan(&id) + return id, err +} + +const touchChannelHashOnly = `-- name: TouchChannelHashOnly :one +UPDATE channels SET last_seen = NOW() +WHERE id = $1 AND channel_hash = $2 AND key_fingerprint IS NULL +RETURNING id +` + +type TouchChannelHashOnlyParams struct { + ID int32 `json:"id"` + ChannelHash []byte `json:"channel_hash"` +} + +// Write-through for a cached hash-only channel: the UpsertChannelHashOnly conflict update, by id. +func (q *Queries) TouchChannelHashOnly(ctx context.Context, arg TouchChannelHashOnlyParams) (int32, error) { + row := q.db.QueryRow(ctx, touchChannelHashOnly, arg.ID, arg.ChannelHash) + var id int32 + err := row.Scan(&id) + return id, err +} + const touchObserverBrokers = `-- name: TouchObserverBrokers :exec UPDATE observer_brokers ob SET last_seen = GREATEST(ob.last_seen, v.seen), diff --git a/db/store.go b/db/store.go index a4b82c9..88b7e36 100644 --- a/db/store.go +++ b/db/store.go @@ -23,10 +23,11 @@ import ( // Store wraps the sqlc-generated Queries and implements both ingest.DB and api.Reader. type Store struct { q sqlc.Querier - pool *pgxpool.Pool // dedicated connections (rollups, reconfirm, catalog reads); nil for the API store and mocks - clockDriftThreshold time.Duration // see api.Node.ClockOutOfSync - staleThreshold time.Duration // see api.NodeSummary.Stale - pgVersion atomic.Int32 // server_version_num, read by the first PgActivity call + pool *pgxpool.Pool // dedicated connections (rollups, reconfirm, catalog reads); nil for the API store and mocks + clockDriftThreshold time.Duration // see api.Node.ClockOutOfSync + staleThreshold time.Duration // see api.NodeSummary.Stale + pgVersion atomic.Int32 // server_version_num, read by the first PgActivity call + channelIDs *ChannelIDCache // shared by the stores that write channels; nil for the API store and mocks } // New creates a Store whose statements run on dbtx (a pool, or a PlanRouter for the API). @@ -35,8 +36,12 @@ type Store struct { // server's clockOutOfSync is reported true; see internal/config.ResolvedConfig.ClockDriftThreshold. // staleThreshold is how long since last_seen before a node's Stale is reported true; see // internal/config.ResolvedConfig.NodeStaleThreshold. -func New(dbtx sqlc.DBTX, pool *pgxpool.Pool, clockDriftThreshold, staleThreshold time.Duration) *Store { - return &Store{q: sqlc.New(dbtx), pool: pool, clockDriftThreshold: clockDriftThreshold, staleThreshold: staleThreshold} +func New(dbtx sqlc.DBTX, pool *pgxpool.Pool, clockDriftThreshold, staleThreshold time.Duration, opts ...Option) *Store { + s := &Store{q: sqlc.New(dbtx), pool: pool, clockDriftThreshold: clockDriftThreshold, staleThreshold: staleThreshold} + for _, o := range opts { + o(s) + } + return s } func (s *Store) ResolvePathHashes(ctx context.Context, iata string, hashes [][]byte) (map[string][]api.ResolvedPathEntry, error) {