Small tables: drop HOT-blocking indexes, channel write-through (#239)

* perf(db): drop small-table indexes that block HOT updates

* perf(db): touch known channels by id instead of upserting

* feat: preload channel ids at boot for ingest and maintenance

.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:

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.

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
+}

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
 }
 

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)
+				}
+			}
+		}
+	})
+}

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")
+	}
+}

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);

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;

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;

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;

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;

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;

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;

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;

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;

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;

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;

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;

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;

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);

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)

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)
+}

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()

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).

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),

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) {