perf: index, cache, and deflake /api/channels queries (rebase of #1887) (#1936)

Continues #1887 by @Jonher937. The commit is theirs, authorship
unchanged; I only rebased it onto master.

It went CONFLICTING because #1934 (prepared statements, originally
@Joel-Claw's #1878) landed in the same `DB` struct. Both PRs add fields
there and this one also replaces the single-slot channels cache.

Resolution: kept this PR's keyed caches (`channelsCache`,
`encChannelsCache`, `msgCache` plus their entry types and TTL constants)
and kept master's thirteen prepared-statement fields alongside them. The
old single-slot `channelsCacheKey`/`channelsCacheRes`/`channelsCacheExp`
trio is gone, which is the point of this PR. Nothing else touched.

Verified: `cmd/server` builds and the **full suite passes**, not just
the channel tests.

My review stands: approve, with two questions that do not block and are
worth a look at some point.

1. `msgCache` is keyed by `hash|limit|offset|region`, and `offset` grows
without bound as someone pages through a channel. Each entry also holds
a full page of message maps, so a full 256-entry cache at `limit=50`
holds around 12,800 maps. The other two caches are keyed by region only
and genuinely low-cardinality as your comment says; this one is the odd
one out.
2. `getMsgCache` returns the cached slice directly, so every hit hands
the caller the same message maps. If any handler mutates one before
serialising, it corrupts the cache for the next ten seconds. Same class
as the finding on #1871, which was fixed there by copying at the two
broadcast sites.

Co-authored-by: Jonathan Herlin <jonte@jherlin.se>

cmd/server/db.go

@@ -45,11 +45,20 @@ type DB struct {
 	hasMultibyteSupCols bool   // nodes/inactive_nodes have multibyte_sup/multibyte_evidence (#903)
 	hasLastSeen         bool   // transmissions.last_seen column exists (#1690)
 
-	// Channel list cache (60s TTL) — avoids repeated GROUP BY scans (#762)
-	channelsCacheMu  sync.Mutex
-	channelsCacheKey string
-	channelsCacheRes []map[string]interface{}
-	channelsCacheExp time.Time
+	// Channel list caches, keyed by region param — avoids repeated GROUP BY
+	// scans (#762). Keyed per-region (not a single slot) so mixed-region
+	// traffic doesn't evict and re-run the query on every request.
+	channelsCacheMu sync.Mutex
+	channelsCache   map[string]channelsCacheEntry
+
+	encChannelsCacheMu sync.Mutex
+	encChannelsCache   map[string]channelsCacheEntry
+
+	// Channel messages cache, keyed by hash+limit+offset+region. Unlike
+	// GetChannels, this previously had no cache at all — every page
+	// view/poll re-ran the full paginated query.
+	msgCacheMu sync.Mutex
+	msgCache   map[string]channelMessagesCacheEntry
 
 	// Prepared statements for frequently-called queries.
 	// Prepared once at initDB time, reused across all requests.
@@ -68,6 +77,30 @@ type DB struct {
 	stmtMaxObsID            *sql.Stmt // COALESCE(MAX(id), 0) FROM observations
 }
 
+// channelsCacheTTL is shared by the GetChannels and GetEncryptedChannels
+// caches — both are cheap to keep fresh at the same cadence as before.
+const channelsCacheTTL = 60 * time.Second
+
+// msgCacheTTL is shorter than channelsCacheTTL: channel message pages are
+// polled more aggressively (e.g. an open channel view) and staleness is
+// more visible to users than in the channel list.
+const msgCacheTTL = 10 * time.Second
+
+// maxCacheEntries bounds a keyed cache's size. Region/pagination keys are
+// low-cardinality in practice; this is a defensive reset, not a real LRU.
+const maxCacheEntries = 256
+
+type channelsCacheEntry struct {
+	res []map[string]interface{}
+	exp time.Time
+}
+
+type channelMessagesCacheEntry struct {
+	msgs  []map[string]interface{}
+	total int
+	exp   time.Time
+}
+
 // OpenDB opens a read-only SQLite connection with WAL mode.
 func OpenDB(path string) (*DB, error) {
 	dsn := fmt.Sprintf("file:%s?mode=ro&_journal_mode=WAL&_busy_timeout=5000", path)
@@ -1601,6 +1634,48 @@ func (db *DB) GetTraces(hash string) ([]map[string]interface{}, error) {
 	return traces, nil
 }
 
+// getChannelsCache returns the cached GetChannels result for key, if present
+// and unexpired.
+func (db *DB) getChannelsCache(key string) ([]map[string]interface{}, bool) {
+	db.channelsCacheMu.Lock()
+	defer db.channelsCacheMu.Unlock()
+	e, ok := db.channelsCache[key]
+	if !ok || time.Now().After(e.exp) {
+		return nil, false
+	}
+	return e.res, true
+}
+
+func (db *DB) setChannelsCache(key string, res []map[string]interface{}) {
+	db.channelsCacheMu.Lock()
+	defer db.channelsCacheMu.Unlock()
+	if db.channelsCache == nil || len(db.channelsCache) > maxCacheEntries {
+		db.channelsCache = make(map[string]channelsCacheEntry)
+	}
+	db.channelsCache[key] = channelsCacheEntry{res: res, exp: time.Now().Add(channelsCacheTTL)}
+}
+
+// getEncChannelsCache/setEncChannelsCache mirror getChannelsCache/
+// setChannelsCache for GetEncryptedChannels, which previously had no cache.
+func (db *DB) getEncChannelsCache(key string) ([]map[string]interface{}, bool) {
+	db.encChannelsCacheMu.Lock()
+	defer db.encChannelsCacheMu.Unlock()
+	e, ok := db.encChannelsCache[key]
+	if !ok || time.Now().After(e.exp) {
+		return nil, false
+	}
+	return e.res, true
+}
+
+func (db *DB) setEncChannelsCache(key string, res []map[string]interface{}) {
+	db.encChannelsCacheMu.Lock()
+	defer db.encChannelsCacheMu.Unlock()
+	if db.encChannelsCache == nil || len(db.encChannelsCache) > maxCacheEntries {
+		db.encChannelsCache = make(map[string]channelsCacheEntry)
+	}
+	db.encChannelsCache[key] = channelsCacheEntry{res: res, exp: time.Now().Add(channelsCacheTTL)}
+}
+
 // GetChannels returns channel list from GRP_TXT packets.
 // Queries transmissions directly (not a VIEW) to avoid observation-level
 // duplicates that could cause stale lastMessage when an older message has
@@ -1611,14 +1686,9 @@ func (db *DB) GetChannels(region ...string) ([]map[string]interface{}, error) {
 		regionParam = region[0]
 	}
 
-	// Check cache (60s TTL)
-	db.channelsCacheMu.Lock()
-	if db.channelsCacheRes != nil && db.channelsCacheKey == regionParam && time.Now().Before(db.channelsCacheExp) {
-		res := db.channelsCacheRes
-		db.channelsCacheMu.Unlock()
-		return res, nil
+	if cached, ok := db.getChannelsCache(regionParam); ok {
+		return cached, nil
 	}
-	db.channelsCacheMu.Unlock()
 
 	regionCodes := normalizeRegionCodes(regionParam)
 
@@ -1726,12 +1796,7 @@ func (db *DB) GetChannels(region ...string) ([]map[string]interface{}, error) {
 		})
 	}
 
-	// Store in cache (60s TTL)
-	db.channelsCacheMu.Lock()
-	db.channelsCacheRes = channels
-	db.channelsCacheKey = regionParam
-	db.channelsCacheExp = time.Now().Add(60 * time.Second)
-	db.channelsCacheMu.Unlock()
+	db.setChannelsCache(regionParam, channels)
 
 	return channels, nil
 }
@@ -1743,6 +1808,11 @@ func (db *DB) GetEncryptedChannels(region ...string) ([]map[string]interface{},
 	if len(region) > 0 {
 		regionParam = region[0]
 	}
+
+	if cached, ok := db.getEncChannelsCache(regionParam); ok {
+		return cached, nil
+	}
+
 	regionCodes := normalizeRegionCodes(regionParam)
 
 	var querySQL string
@@ -1819,9 +1889,35 @@ func (db *DB) GetEncryptedChannels(region ...string) ([]map[string]interface{},
 			"encrypted":    true,
 		})
 	}
+
+	db.setEncChannelsCache(regionParam, channels)
+
 	return channels, nil
 }
 
+// getMsgCache/setMsgCache cache GetChannelMessages results, keyed by
+// hash+limit+offset+region. GetChannelMessages previously had no cache at
+// all, so every page view/poll re-ran the full paginated query even though
+// polling the same page repeatedly is the common case.
+func (db *DB) getMsgCache(key string) ([]map[string]interface{}, int, bool) {
+	db.msgCacheMu.Lock()
+	defer db.msgCacheMu.Unlock()
+	e, ok := db.msgCache[key]
+	if !ok || time.Now().After(e.exp) {
+		return nil, 0, false
+	}
+	return e.msgs, e.total, true
+}
+
+func (db *DB) setMsgCache(key string, msgs []map[string]interface{}, total int) {
+	db.msgCacheMu.Lock()
+	defer db.msgCacheMu.Unlock()
+	if db.msgCache == nil || len(db.msgCache) > maxCacheEntries {
+		db.msgCache = make(map[string]channelMessagesCacheEntry)
+	}
+	db.msgCache[key] = channelMessagesCacheEntry{msgs: msgs, total: total, exp: time.Now().Add(msgCacheTTL)}
+}
+
 // GetChannelMessages returns messages for a specific channel.
 // Uses transmission-level ordering (first_seen) to ensure correct message
 // sequence even when observations arrive out of order.
@@ -1845,6 +1941,12 @@ func (db *DB) GetChannelMessages(channelHash string, limit, offset int, region .
 	if len(region) > 0 {
 		regionParam = region[0]
 	}
+
+	cacheKey := fmt.Sprintf("%s|%d|%d|%s", channelHash, limit, offset, regionParam)
+	if msgs, total, ok := db.getMsgCache(cacheKey); ok {
+		return msgs, total, nil
+	}
+
 	regionCodes := normalizeRegionCodes(regionParam)
 	regionArgs := make([]interface{}, 0, len(regionCodes))
 	regionPlaceholders := ""
@@ -1938,7 +2040,9 @@ func (db *DB) GetChannelMessages(channelHash string, limit, offset int, region .
 	idRows.Close()
 
 	if len(pageIDs) == 0 {
-		return []map[string]interface{}{}, total, nil
+		empty := []map[string]interface{}{}
+		db.setMsgCache(cacheKey, empty, total)
+		return empty, total, nil
 	}
 
 	// 3) Fetch observations for just this page of transmissions. We keep
@@ -2088,6 +2192,8 @@ func (db *DB) GetChannelMessages(channelHash string, limit, offset int, region .
 		messages = append(messages, e.data)
 	}
 
+	db.setMsgCache(cacheKey, messages, total)
+
 	return messages, total, nil
 }
 

cmd/server/db_channel_messages_perf_test.go

@@ -76,6 +76,13 @@ func TestGetChannelMessagesPerfLargeChannel(t *testing.T) {
 		t.Fatal(err)
 	}
 
+	// Clear the results cache so the timed call below still measures the
+	// real SQL path rather than a cache hit — this test's whole point is
+	// catching a regression in that query.
+	db.msgCacheMu.Lock()
+	db.msgCache = nil
+	db.msgCacheMu.Unlock()
+
 	start := time.Now()
 	msgs, total, err := db.GetChannelMessages("#perf", 50, 0)
 	elapsed := time.Since(start)

internal/dbschema/dbschema.go

@@ -61,6 +61,11 @@ func Apply(rw *sql.DB, logf Logger) error {
 	if err := ensureObserverIATAColumn(rw, logf); err != nil {
 		return fmt.Errorf("ensure observers.iata: %w", err)
 	}
+	// Must run after ensureObserverIATAColumn: the expression index below
+	// is on observers.iata, which that step guarantees exists.
+	if err := ensureChannelIndexes(rw); err != nil {
+		return fmt.Errorf("ensure channel indexes: %w", err)
+	}
 	if err := ensureForeignAdvertColumn(rw, logf); err != nil {
 		return fmt.Errorf("ensure foreign_advert: %w", err)
 	}
@@ -243,6 +248,41 @@ func ensureServerIndexes(rw *sql.DB) error {
 	return nil
 }
 
+// ensureChannelIndexes speeds up /api/channels and
+// /api/channels/{hash}/messages.
+//
+//   - idx_transmissions_channel_hash_payload adds first_seen to the
+//     existing channel_hash index (cmd/ingestor/db.go migration
+//     'channel_hash_v1', idx_tx_channel_hash) so GetChannels' "latest
+//     message" correlated subquery (channel_hash + payload_type, ORDER BY
+//     first_seen DESC LIMIT 1) and GetChannelMessages' count/page queries
+//     resolve as index-order scans instead of sorting per channel.
+//   - idx_observers_iata_norm indexes the UPPER(TRIM(iata)) expression
+//     that GetChannels/GetEncryptedChannels/GetChannelMessages use for
+//     region filtering; a plain index on iata can't be used under that
+//     expression, so the expression itself is indexed.
+//
+// transmissions.channel_hash itself is added by the ingestor's legacy
+// 'channel_hash_v1' migration (cmd/ingestor/db.go), not by this package,
+// so it isn't guaranteed to exist yet on every DB Apply runs against —
+// probe before indexing, same pattern as the observer_idx/observer_id
+// probe in ensureServerIndexes above.
+func ensureChannelIndexes(rw *sql.DB) error {
+	hasChannelHash, err := TableHasColumn(rw, "transmissions", "channel_hash")
+	if err != nil {
+		return err
+	}
+	if hasChannelHash {
+		if _, err := rw.Exec(`CREATE INDEX IF NOT EXISTS idx_transmissions_channel_hash_payload ON transmissions(channel_hash, payload_type, first_seen)`); err != nil {
+			return fmt.Errorf("ensure idx_transmissions_channel_hash_payload: %w", err)
+		}
+	}
+	if _, err := rw.Exec(`CREATE INDEX IF NOT EXISTS idx_observers_iata_norm ON observers(UPPER(TRIM(iata)))`); err != nil {
+		return fmt.Errorf("ensure idx_observers_iata_norm: %w", err)
+	}
+	return nil
+}
+
 func ensureNeighborEdgesTable(rw *sql.DB) error {
 	_, err := rw.Exec(`CREATE TABLE IF NOT EXISTS neighbor_edges (
 		node_a TEXT NOT NULL,