diff --git a/cmd/server/db.go b/cmd/server/db.go index 4a915820..eafd3955 100644 --- a/cmd/server/db.go +++ b/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 } diff --git a/cmd/server/db_channel_messages_perf_test.go b/cmd/server/db_channel_messages_perf_test.go index 76a778ef..b96fc8d4 100644 --- a/cmd/server/db_channel_messages_perf_test.go +++ b/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) diff --git a/internal/dbschema/dbschema.go b/internal/dbschema/dbschema.go index cf28557f..58e37874 100644 --- a/internal/dbschema/dbschema.go +++ b/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,