From 4abd9caf596925b19dad3bfd11511224de360e9e Mon Sep 17 00:00:00 2001 From: dborup Date: Tue, 29 Sep 2026 08:23:06 +0000 Subject: [PATCH 1/5] test(channels): reproduce uncoalesced GetChannels/GetEncryptedChannels misses (#109) Adds two test seams to DB, nil in production: channelsMissHook (after a cache miss) and channelsQueryHook (right before each real query), and channels_singleflight_109_test.go, which counts real queries on a seeded fixture (pinTestDB/pinTestSeed, the #107 fixture). On master all 6 fail: 16 concurrent cold GetChannels("AAR") run 16 queries, normalized region forms are not shared, GetEncryptedChannels has no coalescing and no cache, a failing flight runs 8 queries for 8 callers, and a caller held past its miss queries again after another caller has filled the cache. Results per region are already correct; the tests pin that too. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019TcZHooUiiknVWbECVWzk8 --- cmd/server/channels_singleflight_109_test.go | 268 +++++++++++++++++++ cmd/server/db.go | 22 ++ 2 files changed, 290 insertions(+) create mode 100644 cmd/server/channels_singleflight_109_test.go diff --git a/cmd/server/channels_singleflight_109_test.go b/cmd/server/channels_singleflight_109_test.go new file mode 100644 index 000000000..7754dcb2b --- /dev/null +++ b/cmd/server/channels_singleflight_109_test.go @@ -0,0 +1,268 @@ +package main + +import ( + "errors" + "reflect" + "sync" + "sync/atomic" + "testing" + "time" +) + +// Issue #109: concurrent cache misses for one region must run one SQLite +// query, for GetChannels and GetEncryptedChannels, without mixing regions, +// caching errors or losing #107's index pinning and #98's slice protection. +// The counts come from channelsQueryHook, which runs once per real query. + +func singleflightDB(t *testing.T) *DB { + t.Helper() + db := pinTestDB(t, true, false) + pinTestSeed(t, db, true, 2000, 109) + return db +} + +// queryCounter counts real queries per kind. The first query of a flight +// waits until either `want` queries are running at once (the no-coalescing +// failure mode: every caller queries) or a short grace period passes, so +// concurrent callers really overlap it. +type queryCounter struct { + mu sync.Mutex + n map[string]int + running atomic.Int32 + want int32 + failAll error +} + +func (q *queryCounter) hook(kind, _ string) error { + q.mu.Lock() + if q.n == nil { + q.n = map[string]int{} + } + q.n[kind]++ + q.mu.Unlock() + q.running.Add(1) + deadline := time.Now().Add(300 * time.Millisecond) + for q.running.Load() < q.want && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + return q.failAll +} + +func (q *queryCounter) count(kind string) int { + q.mu.Lock() + defer q.mu.Unlock() + return q.n[kind] +} + +// callConcurrently starts n callers at once and returns their results. +func callConcurrently(n int, fn func(i int) (interface{}, error)) ([]interface{}, []error) { + res := make([]interface{}, n) + errs := make([]error, n) + start := make(chan struct{}) + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + <-start + res[i], errs[i] = fn(i) + }(i) + } + close(start) + wg.Wait() + return res, errs +} + +func TestGetChannelsCoalescesConcurrentMisses_109(t *testing.T) { + db := singleflightDB(t) + q := &queryCounter{want: 16} + db.channelsQueryHook = q.hook + res, errs := callConcurrently(16, func(int) (interface{}, error) { return db.GetChannels("AAR") }) + for i, err := range errs { + if err != nil { + t.Fatalf("caller %d: %v", i, err) + } + } + if got := q.count("channels"); got != 1 { + t.Fatalf("16 concurrent cold GetChannels(AAR) ran %d queries, want 1", got) + } + for i := 1; i < len(res); i++ { + if !reflect.DeepEqual(res[i], res[0]) { + t.Fatalf("caller %d got a different result", i) + } + } +} + +// " aar ", "AAR" and "aar,AAR" are one normalized region. +func TestGetChannelsCoalescesOnTheNormalizedRegion_109(t *testing.T) { + db := singleflightDB(t) + q := &queryCounter{want: 12} + db.channelsQueryHook = q.hook + forms := []string{"AAR", " aar ", "aar,AAR", "Aar"} + _, errs := callConcurrently(12, func(i int) (interface{}, error) { return db.GetChannels(forms[i%len(forms)]) }) + for _, err := range errs { + if err != nil { + t.Fatal(err) + } + } + if got := q.count("channels"); got != 1 { + t.Fatalf("12 concurrent cold calls for one normalized region ran %d queries, want 1", got) + } +} + +func TestGetEncryptedChannelsCoalescesConcurrentMisses_109(t *testing.T) { + db := singleflightDB(t) + q := &queryCounter{want: 16} + db.channelsQueryHook = q.hook + res, errs := callConcurrently(16, func(int) (interface{}, error) { return db.GetEncryptedChannels("CPH") }) + for i, err := range errs { + if err != nil { + t.Fatalf("caller %d: %v", i, err) + } + } + if got := q.count("encrypted"); got != 1 { + t.Fatalf("16 concurrent cold GetEncryptedChannels(CPH) ran %d queries, want 1", got) + } + if len(res[0].([]map[string]interface{})) == 0 { + t.Fatal("fixture: no encrypted channels for CPH") + } + // and within the TTL a later call is served from the cache + if _, err := db.GetEncryptedChannels("CPH"); err != nil { + t.Fatal(err) + } + if got := q.count("encrypted"); got != 1 { + t.Fatalf("a call within the TTL ran a query again (%d queries)", got) + } +} + +// Different regions are separate flights and each gets its own result, +// identical to what an uncoalesced call returns. +func TestDifferentRegionsAreNeverMixed_109(t *testing.T) { + regions := []string{"AAR", "CPH", "", "OSL"} + want := map[string][2]interface{}{} + for _, r := range regions { + fresh := singleflightDB(t) + ch, err := fresh.GetChannels(r) + if err != nil { + t.Fatal(err) + } + enc, err := fresh.GetEncryptedChannels(r) + if err != nil { + t.Fatal(err) + } + want[r] = [2]interface{}{ch, enc} + } + if reflect.DeepEqual(want["AAR"][0], want["CPH"][0]) || reflect.DeepEqual(want["AAR"][1], want["CPH"][1]) { + t.Fatal("fixture: AAR and CPH give the same channels; the test could not tell them apart") + } + + db := singleflightDB(t) + q := &queryCounter{want: 32} + db.channelsQueryHook = q.hook + res, errs := callConcurrently(32, func(i int) (interface{}, error) { + r := regions[i%len(regions)] + if i%2 == 0 { + ch, err := db.GetChannels(r) + return [2]interface{}{r, ch}, err + } + enc, err := db.GetEncryptedChannels(r) + return [2]interface{}{r, enc}, err + }) + for i, err := range errs { + if err != nil { + t.Fatal(err) + } + got := res[i].([2]interface{}) + r := got[0].(string) + if !reflect.DeepEqual(got[1], want[r][i%2]) { + t.Fatalf("caller %d (region %q, kind %d) got another region's result", i, r, i%2) + } + } + if c, e := q.count("channels"), q.count("encrypted"); c != len(regions) || e != len(regions) { + t.Fatalf("queries: channels=%d encrypted=%d, want %d each (one per region)", c, e, len(regions)) + } +} + +// An error is shared only by the callers of that flight and is not cached: +// the next call queries again and succeeds. +func TestChannelErrorsAreSharedButNotCached_109(t *testing.T) { + for _, kind := range []string{"channels", "encrypted"} { + t.Run(kind, func(t *testing.T) { + db := singleflightDB(t) + boom := errors.New("injected query failure") + q := &queryCounter{want: 8, failAll: boom} + db.channelsQueryHook = q.hook + call := func() (interface{}, error) { + if kind == "channels" { + return db.GetChannels("AAR") + } + return db.GetEncryptedChannels("AAR") + } + _, errs := callConcurrently(8, func(int) (interface{}, error) { return call() }) + for i, err := range errs { + if !errors.Is(err, boom) { + t.Fatalf("caller %d: err %v, want the flight's error", i, err) + } + } + if got := q.count(kind); got != 1 { + t.Fatalf("the failing flight ran %d queries, want 1", got) + } + q.failAll = nil + res, err := call() + if err != nil { + t.Fatalf("after the error: %v", err) + } + if res == nil { + t.Fatal("after the error: nil result") + } + if got := q.count(kind); got != 2 { + t.Fatalf("the error was cached: %d queries after a failed and a fresh call, want 2", got) + } + }) + } +} + +// A caller that missed the cache before a flight finished, and reaches +// singleflight only after it has, must find the fresh result on the second +// cache check inside its own flight instead of querying again. +func TestSecondCacheCheckInsideTheFlight_109(t *testing.T) { + for _, kind := range []string{"channels", "encrypted"} { + t.Run(kind, func(t *testing.T) { + db := singleflightDB(t) + q := &queryCounter{want: 1} + db.channelsQueryHook = q.hook + var misses atomic.Int32 + release := make(chan struct{}) + db.channelsMissHook = func(k, _ string) { + if k == kind && misses.Add(1) == 1 { + <-release // the first caller is held right after its miss + } + } + call := func() error { + var err error + if kind == "channels" { + _, err = db.GetChannels("OSL") + } else { + _, err = db.GetEncryptedChannels("OSL") + } + return err + } + held := make(chan error, 1) + go func() { held <- call() }() + for misses.Load() < 1 { + time.Sleep(time.Millisecond) + } + // a second caller misses too, runs the query and fills the cache + if err := call(); err != nil { + t.Fatal(err) + } + close(release) + if err := <-held; err != nil { + t.Fatal(err) + } + if got := q.count(kind); got != 1 { + t.Fatalf("%d queries: the held caller queried again instead of re-checking the cache inside its flight", got) + } + }) + } +} diff --git a/cmd/server/db.go b/cmd/server/db.go index 37441f492..b68cb88e7 100644 --- a/cmd/server/db.go +++ b/cmd/server/db.go @@ -103,6 +103,12 @@ type DB struct { // channelsPinFallbacks counts GetChannels queries that fell back to the // unpinned SQL because channelHashIndex was not usable; the first is logged. channelsPinFallbacks atomic.Int64 + // Test seams (#109), nil in production. channelsMissHook runs after a + // cache miss, before the query is built; channelsQueryHook runs right + // before the query executes, once per real query. kind is "channels" + // (GetChannels) or "encrypted" (GetEncryptedChannels). + channelsMissHook func(kind, region string) + channelsQueryHook func(kind, region string) error } // isV3, hasResolvedPath, hasObsRawHex, hasScopeName, hasDefaultScope, @@ -3107,6 +3113,9 @@ func (db *DB) GetChannels(region ...string) ([]map[string]interface{}, error) { return res, nil } db.channelsCacheMu.Unlock() + if db.channelsMissHook != nil { + db.channelsMissHook("channels", regionParam) + } regionCodes := normalizeRegionCodes(regionParam) regionPlaceholder := "" @@ -3120,6 +3129,11 @@ func (db *DB) GetChannels(region ...string) ([]map[string]interface{}, error) { regionPlaceholder = strings.Join(placeholders, ",") } + if db.channelsQueryHook != nil { + if err := db.channelsQueryHook("channels", regionParam); err != nil { + return nil, err + } + } // Pinned to channelHashIndex first; INDEXED BY fails at prepare time // (before any row is read) when that index is missing or unusable, and // then the unpinned query - the pre-pinning SQL - runs instead. @@ -3189,6 +3203,9 @@ func (db *DB) GetEncryptedChannels(region ...string) ([]map[string]interface{}, if len(region) > 0 { regionParam = region[0] } + if db.channelsMissHook != nil { + db.channelsMissHook("encrypted", regionParam) + } regionCodes := normalizeRegionCodes(regionParam) var querySQL string @@ -3240,6 +3257,11 @@ func (db *DB) GetEncryptedChannels(region ...string) ([]map[string]interface{}, ORDER BY last_activity DESC` } + if db.channelsQueryHook != nil { + if err := db.channelsQueryHook("encrypted", regionParam); err != nil { + return nil, err + } + } rows, err := db.conn.Query(querySQL, args...) if err != nil { return nil, err From 048db3c9dcf08ac639cfe83013ab4d4b6229245d Mon Sep 17 00:00:00 2001 From: dborup Date: Tue, 29 Sep 2026 08:29:43 +0000 Subject: [PATCH 2/5] perf(channels): coalesce GetChannels/GetEncryptedChannels misses per normalized region (#109) GetChannels had a single-slot 60s cache and no request coalescing; GetEncryptedChannels had neither. Every request that missed ran the GROUP BY scan itself. Both now go through channelListCache (channels_list_cache.go): - keyed by the normalized region (sorted, de-duplicated codes), so " aar " and "AAR,aar" share one entry and one flight, and different regions never share a result; - per-key singleflight with a second cache check inside the flight; - only successful results are stored: an error goes to that flight's callers and the next call queries again; - bounded to 64 keys (the region is a query parameter): expired entries go first, then the one closest to expiry. The query bodies are unchanged: #107's pinned-then-unpinned fallback is the same code, now in queryChannels(key); #98's full-slice append in handleChannels is untouched and the cached slices are never modified. No new map[string]interface{} (the cached slice moved from a DB field to channelListEntry in db.go; count unchanged at 66). Benchmark, 16 concurrent cold callers (8 GetChannels, 8 GetEncryptedChannels, region AAR, 20k-row fixture): master 16 queries and about 440 ms per round, this branch 2 queries and about 54 ms. TestDifferentRegionsAreNeverMixed_109 now spreads both kinds over every region (the first version paired each region with one kind only). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_019TcZHooUiiknVWbECVWzk8 --- cmd/server/channel_proposals_test.go | 4 +- cmd/server/channels_cache_append_test.go | 14 +-- cmd/server/channels_list_cache.go | 121 +++++++++++++++++++ cmd/server/channels_singleflight_109_test.go | 65 +++++++++- cmd/server/db.go | 77 +++++++----- 5 files changed, 239 insertions(+), 42 deletions(-) create mode 100644 cmd/server/channels_list_cache.go diff --git a/cmd/server/channel_proposals_test.go b/cmd/server/channel_proposals_test.go index d6f6a9161..23e029bf1 100644 --- a/cmd/server/channel_proposals_test.go +++ b/cmd/server/channel_proposals_test.go @@ -384,9 +384,7 @@ func TestChannelsListDoesNotMutateCaches(t *testing.T) { // capacity, so pad the decrypted channels until it does. var cached []map[string]interface{} for i := 0; ; i++ { - db.channelsCacheMu.Lock() - db.channelsCacheRes = nil - db.channelsCacheMu.Unlock() + db.channelsCache.reset() var err error if cached, err = db.GetChannels(""); err != nil { t.Fatal(err) diff --git a/cmd/server/channels_cache_append_test.go b/cmd/server/channels_cache_append_test.go index 728fd6617..170f53200 100644 --- a/cmd/server/channels_cache_append_test.go +++ b/cmd/server/channels_cache_append_test.go @@ -91,11 +91,8 @@ func insertChannelTx(t *testing.T, db *DB, hash, channelHash, decodedJSON string } func resetDBChannelsCache(db *DB) { - db.channelsCacheMu.Lock() - db.channelsCacheRes = nil - db.channelsCacheKey = "" - db.channelsCacheExp = time.Time{} - db.channelsCacheMu.Unlock() + db.channelsCache.reset() + db.encChannelsCache.reset() } // TestChannelsCacheAppend_DBPath drives GET /api/channels through the real @@ -153,9 +150,10 @@ func TestChannelsCacheAppend_DBPath(t *testing.T) { } } - db.channelsCacheMu.Lock() - final := db.channelsCacheRes - db.channelsCacheMu.Unlock() + var final []map[string]interface{} + if e, ok := db.channelsCache.get("", time.Now()); ok { + final = e.channels + } if cap(final) != cap(cached) || len(final) != len(cached) { t.Fatalf("cache slice header changed unexpectedly: before cap=%d len=%d, after cap=%d len=%d (cache may have expired/refreshed — this test assumes it didn't)", cap(cached), len(cached), cap(final), len(final)) } diff --git a/cmd/server/channels_list_cache.go b/cmd/server/channels_list_cache.go new file mode 100644 index 000000000..64c7d2f3c --- /dev/null +++ b/cmd/server/channels_list_cache.go @@ -0,0 +1,121 @@ +package main + +import ( + "slices" + "sort" + "strings" + "sync" + "time" + + "golang.org/x/sync/singleflight" +) + +// channelListCache backs DB.GetChannels and DB.GetEncryptedChannels (#109). +// +// - Keyed by the normalized region (channelsRegionKey), so " aar " and +// "AAR,aar" share an entry and a flight; different regions never do. +// - Concurrent misses for one key run the query once (singleflight). The +// flight checks the cache again first: a caller that missed before +// another flight stored a result must not query again. +// - Only successful results are stored; an error is returned to the +// callers of that flight and the next call queries again. +// - Bounded: at most channelListMaxKeys entries. The region comes from a +// query parameter, so the key space is caller-controlled; when full, +// expired entries are dropped first, then the one closest to expiry. +// +// A stored entry is never modified. Its channels slice is shared by every +// caller, so callers must not write into it (handleChannels appends with a +// full slice expression, #98). +type channelListCache struct { + mu sync.Mutex + entries map[string]*channelListEntry + flights singleflight.Group +} + +const ( + channelListTTL = 60 * time.Second + channelListMaxKeys = 64 +) + +// channelsRegionKey is the cache and flight key for a region parameter: the +// normalized codes (normalizeRegionCodes), sorted and de-duplicated. The +// query runs with normalizeRegionCodes(key), the same set of codes. +func channelsRegionKey(region string) string { + codes := normalizeRegionCodes(region) + if len(codes) == 0 { + return "" + } + sort.Strings(codes) + return strings.Join(slices.Compact(codes), ",") +} + +func (c *channelListCache) get(key string, now time.Time) (*channelListEntry, bool) { + c.mu.Lock() + defer c.mu.Unlock() + e := c.entries[key] + if e == nil || !now.Before(e.expires) { + return nil, false + } + return e, true +} + +func (c *channelListCache) put(key string, e *channelListEntry, now time.Time) { + c.mu.Lock() + defer c.mu.Unlock() + if c.entries == nil { + c.entries = make(map[string]*channelListEntry) + } + if _, ok := c.entries[key]; !ok && len(c.entries) >= channelListMaxKeys { + for k, old := range c.entries { + if !now.Before(old.expires) { + delete(c.entries, k) + } + } + if len(c.entries) >= channelListMaxKeys { + victim, first := "", true + var soonest time.Time + for k, old := range c.entries { + if first || old.expires.Before(soonest) { + victim, soonest, first = k, old.expires, false + } + } + delete(c.entries, victim) + } + } + c.entries[key] = e +} + +// reset drops every entry (tests). +func (c *channelListCache) reset() { + c.mu.Lock() + c.entries = nil + c.mu.Unlock() +} + +// load returns the cached entry for key or, on a miss, the result of one +// query shared by every concurrent caller of key. missed runs after the +// first cache check fails (a test seam; nil in production). +func (c *channelListCache) load(key string, missed func(), query func() (*channelListEntry, error)) (*channelListEntry, error) { + if e, ok := c.get(key, time.Now()); ok { + return e, nil + } + if missed != nil { + missed() + } + v, err, _ := c.flights.Do(key, func() (interface{}, error) { + if e, ok := c.get(key, time.Now()); ok { + return e, nil + } + e, err := query() + if err != nil { + return nil, err + } + e.expires = time.Now().Add(channelListTTL) + c.put(key, e, time.Now()) + return e, nil + }) + if err != nil { + return nil, err + } + return v.(*channelListEntry), nil +} diff --git a/cmd/server/channels_singleflight_109_test.go b/cmd/server/channels_singleflight_109_test.go index 7754dcb2b..1c3a7169b 100644 --- a/cmd/server/channels_singleflight_109_test.go +++ b/cmd/server/channels_singleflight_109_test.go @@ -3,6 +3,8 @@ package main import ( "errors" "reflect" + "strconv" + "strings" "sync" "sync/atomic" "testing" @@ -160,7 +162,7 @@ func TestDifferentRegionsAreNeverMixed_109(t *testing.T) { q := &queryCounter{want: 32} db.channelsQueryHook = q.hook res, errs := callConcurrently(32, func(i int) (interface{}, error) { - r := regions[i%len(regions)] + r := regions[(i/2)%len(regions)] // every region with both kinds if i%2 == 0 { ch, err := db.GetChannels(r) return [2]interface{}{r, ch}, err @@ -266,3 +268,64 @@ func TestSecondCacheCheckInsideTheFlight_109(t *testing.T) { }) } } + +// The region comes from a query parameter, so the cache must stay bounded +// however many distinct regions callers send. +func TestChannelListCacheIsBounded_109(t *testing.T) { + var c channelListCache + now := time.Now() + for i := 0; i < 1000; i++ { + key := channelsRegionKey(strings.Repeat("X", 1+i%5) + strconv.Itoa(i)) + c.put(key, &channelListEntry{expires: now.Add(channelListTTL)}, now) + if len(c.entries) > channelListMaxKeys { + t.Fatalf("cache grew to %d entries", len(c.entries)) + } + } + // expired entries go first: a fresh key replaces an expired one, not a live one + c.reset() + for i := 0; i < channelListMaxKeys; i++ { + exp := now.Add(channelListTTL) + if i == 7 { + exp = now.Add(-time.Second) + } + c.put(strconv.Itoa(i), &channelListEntry{expires: exp}, now) + } + c.put("new", &channelListEntry{expires: now.Add(channelListTTL)}, now) + if _, ok := c.entries["7"]; ok { + t.Error("the expired entry survived") + } + if len(c.entries) != channelListMaxKeys { + t.Errorf("%d entries, want %d", len(c.entries), channelListMaxKeys) + } +} + +func TestChannelsRegionKey_109(t *testing.T) { + for in, want := range map[string]string{ + "": "", "all": "", " aar ": "AAR", "cph,AAR,aar": "AAR,CPH", "AAR, ,CPH": "AAR,CPH", + } { + if got := channelsRegionKey(in); got != want { + t.Errorf("channelsRegionKey(%q) = %q, want %q", in, got, want) + } + } +} + +// BenchmarkColdConcurrentGetChannels_109: 16 callers hit a cold (reset) +// cache for one region at once; reports the real queries per round. +func BenchmarkColdConcurrentGetChannels_109(b *testing.B) { + db := pinTestDB(b, true, false) + pinTestSeed(b, db, true, 20000, 109) + var queries atomic.Int64 + db.channelsQueryHook = func(string, string) error { queries.Add(1); return nil } + b.ResetTimer() + for i := 0; i < b.N; i++ { + db.channelsCache.reset() + db.encChannelsCache.reset() + callConcurrently(16, func(j int) (interface{}, error) { + if j%2 == 0 { + return db.GetChannels("AAR") + } + return db.GetEncryptedChannels("AAR") + }) + } + b.ReportMetric(float64(queries.Load())/float64(b.N), "queries/round") +} diff --git a/cmd/server/db.go b/cmd/server/db.go index b68cb88e7..311e0b684 100644 --- a/cmd/server/db.go +++ b/cmd/server/db.go @@ -95,11 +95,11 @@ type DB struct { routeMaskStatus RouteMaskBackfillStatus routeMaskStatusExp time.Time - // 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 (60s TTL, #762): keyed by normalized region, with + // concurrent misses coalesced into one query per key (#109). See + // channels_list_cache.go. + channelsCache channelListCache + encChannelsCache channelListCache // channelsPinFallbacks counts GetChannels queries that fell back to the // unpinned SQL because channelHashIndex was not usable; the first is logged. channelsPinFallbacks atomic.Int64 @@ -111,6 +111,13 @@ type DB struct { channelsQueryHook func(kind, region string) error } +// channelListEntry is one cached GetChannels / GetEncryptedChannels result. +// Never modified after it is stored (see channelListCache). +type channelListEntry struct { + channels []map[string]interface{} + expires time.Time +} + // isV3, hasResolvedPath, hasObsRawHex, hasScopeName, hasDefaultScope, // hasConfiguredScope, hasDefaultScopeConfirmedAt, hasMultibyteSupCols, and // hasLastSeen are pure atomic-load accessors over the corresponding @@ -3104,20 +3111,19 @@ func (db *DB) GetChannels(region ...string) ([]map[string]interface{}, error) { if len(region) > 0 { 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 - } - db.channelsCacheMu.Unlock() - if db.channelsMissHook != nil { - db.channelsMissHook("channels", regionParam) + key := channelsRegionKey(regionParam) + e, err := db.channelsCache.load(key, db.channelsMissFunc("channels", key), func() (*channelListEntry, error) { + return db.queryChannels(key) + }) + if err != nil { + return nil, err } + return e.channels, nil +} - regionCodes := normalizeRegionCodes(regionParam) +// queryChannels runs the GetChannels query for a normalized region key. +func (db *DB) queryChannels(key string) (*channelListEntry, error) { + regionCodes := normalizeRegionCodes(key) regionPlaceholder := "" args := make([]interface{}, 0, len(regionCodes)) if len(regionCodes) > 0 { @@ -3130,7 +3136,7 @@ func (db *DB) GetChannels(region ...string) ([]map[string]interface{}, error) { } if db.channelsQueryHook != nil { - if err := db.channelsQueryHook("channels", regionParam); err != nil { + if err := db.channelsQueryHook("channels", key); err != nil { return nil, err } } @@ -3186,14 +3192,15 @@ 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() + return &channelListEntry{channels: channels}, nil +} - return channels, nil +// channelsMissFunc returns the channelsMissHook call for one lookup, or nil. +func (db *DB) channelsMissFunc(kind, key string) func() { + if db.channelsMissHook == nil { + return nil + } + return func() { db.channelsMissHook(kind, key) } } // GetEncryptedChannels returns channels where all messages are undecryptable (no key). @@ -3203,10 +3210,20 @@ func (db *DB) GetEncryptedChannels(region ...string) ([]map[string]interface{}, if len(region) > 0 { regionParam = region[0] } - if db.channelsMissHook != nil { - db.channelsMissHook("encrypted", regionParam) + key := channelsRegionKey(regionParam) + e, err := db.encChannelsCache.load(key, db.channelsMissFunc("encrypted", key), func() (*channelListEntry, error) { + return db.queryEncryptedChannels(key) + }) + if err != nil { + return nil, err } - regionCodes := normalizeRegionCodes(regionParam) + return e.channels, nil +} + +// queryEncryptedChannels runs the GetEncryptedChannels query for a +// normalized region key. +func (db *DB) queryEncryptedChannels(key string) (*channelListEntry, error) { + regionCodes := normalizeRegionCodes(key) var querySQL string args := make([]interface{}, 0, len(regionCodes)) @@ -3258,7 +3275,7 @@ func (db *DB) GetEncryptedChannels(region ...string) ([]map[string]interface{}, } if db.channelsQueryHook != nil { - if err := db.channelsQueryHook("encrypted", regionParam); err != nil { + if err := db.channelsQueryHook("encrypted", key); err != nil { return nil, err } } @@ -3287,7 +3304,7 @@ func (db *DB) GetEncryptedChannels(region ...string) ([]map[string]interface{}, "encrypted": true, }) } - return channels, nil + return &channelListEntry{channels: channels}, nil } // GetChannelMessages returns messages for a specific channel. From a39b2fd50584000d37348599e63b345a4c0e87a0 Mon Sep 17 00:00:00 2001 From: dborup Date: Wed, 30 Sep 2026 08:21:40 +0000 Subject: [PATCH 3/5] fix(channels): fail a channel list query whose rows stop on an error (#109) queryChannels and queryEncryptedChannels did not check rows.Err() after the row loop. A SQLite step that fails part-way ends rows.Next() like the last row does, so the truncated list was returned as success and, since #109, cached for the TTL and shared with every waiter of the flight. Both now return the error, which the cache already never stores. Test: TestChannelRowsErrorIsNotCachedAsSuccess_109 fails the iteration after one row (both kinds) through a new test seam, channelsRowsHook, and asserts the error, a re-query on the next call and the full list. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011FcyXW5RdFzLZhuL1ntAsY --- cmd/server/channels_singleflight_109_test.go | 70 ++++++++++++++++++++ cmd/server/db.go | 36 ++++++++-- 2 files changed, 102 insertions(+), 4 deletions(-) diff --git a/cmd/server/channels_singleflight_109_test.go b/cmd/server/channels_singleflight_109_test.go index 1c3a7169b..d452d1e15 100644 --- a/cmd/server/channels_singleflight_109_test.go +++ b/cmd/server/channels_singleflight_109_test.go @@ -309,6 +309,76 @@ func TestChannelsRegionKey_109(t *testing.T) { } } +// failingRows passes `ok` rows through, then stops as a failed SQLite step +// does: Next returns false and Err reports the failure. +type failingRows struct { + channelRows + ok int + err error +} + +func (r *failingRows) Next() bool { + if r.ok == 0 { + return false + } + r.ok-- + return r.channelRows.Next() +} + +func (r *failingRows) Err() error { + if r.ok == 0 { + return r.err + } + return r.channelRows.Err() +} + +// A failure part-way through the rows must fail the call, not return (and +// cache for the TTL) the truncated list as a success. +func TestChannelRowsErrorIsNotCachedAsSuccess_109(t *testing.T) { + for _, kind := range []string{"channels", "encrypted"} { + t.Run(kind, func(t *testing.T) { + db := singleflightDB(t) + call := func() ([]map[string]interface{}, error) { + if kind == "channels" { + return db.GetChannels("") + } + return db.GetEncryptedChannels("") + } + full, err := call() + if err != nil { + t.Fatal(err) + } + if len(full) < 2 { + t.Fatalf("fixture gives %d rows, need at least 2 to truncate", len(full)) + } + db.channelsCache.reset() + db.encChannelsCache.reset() + + q := &queryCounter{want: 1} + db.channelsQueryHook = q.hook + boom := errors.New("injected step failure") + db.channelsRowsHook = func(_ string, rows channelRows) channelRows { + return &failingRows{channelRows: rows, ok: 1, err: boom} + } + res, err := call() + if !errors.Is(err, boom) { + t.Fatalf("iteration failed after 1 of %d rows: got %d rows, err %v; want the step error", len(full), len(res), err) + } + db.channelsRowsHook = nil + res, err = call() + if err != nil { + t.Fatalf("after the failed iteration: %v", err) + } + if got := q.count(kind); got != 2 { + t.Fatalf("%d queries after a failed and a fresh call, want 2: the truncated list was cached", got) + } + if !reflect.DeepEqual(res, full) { + t.Fatalf("after the failed iteration: %d rows, want the full %d", len(res), len(full)) + } + }) + } +} + // BenchmarkColdConcurrentGetChannels_109: 16 callers hit a cold (reset) // cache for one region at once; reports the real queries per round. func BenchmarkColdConcurrentGetChannels_109(b *testing.B) { diff --git a/cmd/server/db.go b/cmd/server/db.go index 311e0b684..d05d7da8c 100644 --- a/cmd/server/db.go +++ b/cmd/server/db.go @@ -109,6 +109,24 @@ type DB struct { // (GetChannels) or "encrypted" (GetEncryptedChannels). channelsMissHook func(kind, region string) channelsQueryHook func(kind, region string) error + // channelsRowsHook wraps the result rows of each real query, so a test + // can fail the iteration part-way. + channelsRowsHook func(kind string, rows channelRows) channelRows +} + +// channelRows is the part of *sql.Rows the channel list scans use. +type channelRows interface { + Next() bool + Scan(dest ...interface{}) error + Err() error +} + +// channelsRows returns rows, wrapped by channelsRowsHook when a test set it. +func (db *DB) channelsRows(kind string, rows *sql.Rows) channelRows { + if db.channelsRowsHook == nil { + return rows + } + return db.channelsRowsHook(kind, rows) } // channelListEntry is one cached GetChannels / GetEncryptedChannels result. @@ -3154,12 +3172,13 @@ func (db *DB) queryChannels(key string) (*channelListEntry, error) { return nil, err } defer rows.Close() + it := db.channelsRows("channels", rows) channels := make([]map[string]interface{}, 0) - for rows.Next() { + for it.Next() { var chHash, lastActivity, sampleJSON sql.NullString var msgCount int - if err := rows.Scan(&chHash, &msgCount, &lastActivity, &sampleJSON); err != nil { + if err := it.Scan(&chHash, &msgCount, &lastActivity, &sampleJSON); err != nil { continue } channelName := nullStr(chHash) @@ -3191,6 +3210,11 @@ func (db *DB) queryChannels(key string) (*channelListEntry, error) { "messageCount": msgCount, "lastActivity": nullStr(lastActivity), }) } + // A step that fails part-way ends the loop like the last row does; the + // truncated list must not be returned (and cached) as a success. + if err := it.Err(); err != nil { + return nil, err + } return &channelListEntry{channels: channels}, nil } @@ -3284,12 +3308,13 @@ func (db *DB) queryEncryptedChannels(key string) (*channelListEntry, error) { return nil, err } defer rows.Close() + it := db.channelsRows("encrypted", rows) channels := make([]map[string]interface{}, 0) - for rows.Next() { + for it.Next() { var chHash, lastActivity sql.NullString var msgCount int - if err := rows.Scan(&chHash, &msgCount, &lastActivity); err != nil { + if err := it.Scan(&chHash, &msgCount, &lastActivity); err != nil { continue } fullHash := nullStrVal(chHash) // e.g. "enc_3A" @@ -3304,6 +3329,9 @@ func (db *DB) queryEncryptedChannels(key string) (*channelListEntry, error) { "encrypted": true, }) } + if err := it.Err(); err != nil { + return nil, err + } return &channelListEntry{channels: channels}, nil } From 7f32845bb7577b94c671b9de37b6a48a16dd12ac Mon Sep 17 00:00:00 2001 From: dborup Date: Wed, 30 Sep 2026 08:22:56 +0000 Subject: [PATCH 4/5] fix(channels): bound the byte size of channel list cache keys (#109) The cache key is the normalized region query parameter, whose length is caller-controlled (up to the 1 MiB header limit). The 64-entry bound alone let 64 such keys per list stay in memory for the TTL. A key over 256 bytes is now stored, and used as the flight key, as "sha256:" (71 bytes). Long regions are still cached, coalesced and kept apart; the 64-entry bound is unchanged. Test: TestChannelListCacheKeyLengthIsBounded_109 (300 codes, two 1 MiB codes differing in the last byte, and a short code; both lists). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011FcyXW5RdFzLZhuL1ntAsY --- cmd/server/channels_key_bound_109_test.go | 55 +++++++++++++++++++++++ cmd/server/channels_list_cache.go | 18 ++++++++ 2 files changed, 73 insertions(+) create mode 100644 cmd/server/channels_key_bound_109_test.go diff --git a/cmd/server/channels_key_bound_109_test.go b/cmd/server/channels_key_bound_109_test.go new file mode 100644 index 000000000..1797e1ba4 --- /dev/null +++ b/cmd/server/channels_key_bound_109_test.go @@ -0,0 +1,55 @@ +package main + +import ( + "strconv" + "strings" + "testing" +) + +// Every code in the region parameter becomes part of the key, so a long +// parameter must not be kept as a map key (and flight key) byte for byte. +// Long keys are replaced by a fixed-size digest; they are still cached, +// coalesced and kept apart per region, and the entry bound still holds. +func TestChannelListCacheKeyLengthIsBounded_109(t *testing.T) { + db := singleflightDB(t) + q := &queryCounter{want: 1} + db.channelsQueryHook = q.hook + many := make([]string, 300) + for i := range many { + many[i] = "R" + strconv.Itoa(i) + } + regions := []string{ + strings.Join(many, ","), // many short codes + strings.Repeat("Y", 1<<20), // one 1 MiB code + strings.Repeat("Y", 1<<20) + "Z", // differs only in the last byte + "AAR", + } + for _, r := range regions { + for i := 0; i < 2; i++ { // the second call is served from the cache + if _, err := db.GetChannels(r); err != nil { + t.Fatal(err) + } + if _, err := db.GetEncryptedChannels(r); err != nil { + t.Fatal(err) + } + } + } + for _, kind := range []string{"channels", "encrypted"} { + if got := q.count(kind); got != len(regions) { + t.Errorf("%s: %d queries for %d regions called twice each, want one per region", kind, got, len(regions)) + } + } + for name, c := range map[string]*channelListCache{"channels": &db.channelsCache, "encrypted": &db.encChannelsCache} { + if len(c.entries) != len(regions) { + t.Errorf("%s: %d entries for %d regions", name, len(c.entries), len(regions)) + } + for k := range c.entries { + if len(k) > channelListMaxKeyBytes { + t.Errorf("%s: cache key of %d bytes kept, limit %d", name, len(k), channelListMaxKeyBytes) + } + } + if _, ok := c.entries["AAR"]; !ok { + t.Errorf("%s: a short key is not stored as itself", name) + } + } +} diff --git a/cmd/server/channels_list_cache.go b/cmd/server/channels_list_cache.go index 64c7d2f3c..fd734b9f3 100644 --- a/cmd/server/channels_list_cache.go +++ b/cmd/server/channels_list_cache.go @@ -1,6 +1,8 @@ package main import ( + "crypto/sha256" + "encoding/hex" "slices" "sort" "strings" @@ -22,6 +24,8 @@ import ( // - Bounded: at most channelListMaxKeys entries. The region comes from a // query parameter, so the key space is caller-controlled; when full, // expired entries are dropped first, then the one closest to expiry. +// A key longer than channelListMaxKeyBytes is stored (and used as the +// flight key) as its SHA-256 digest, so each stored key is small too. // // A stored entry is never modified. Its channels slice is shared by every // caller, so callers must not write into it (handleChannels appends with a @@ -35,8 +39,21 @@ type channelListCache struct { const ( channelListTTL = 60 * time.Second channelListMaxKeys = 64 + // channelListMaxKeyBytes caps a stored key. A longer one becomes + // "sha256:" + 64 hex digits (71 bytes); no normalized region key + // contains ':', so a digest never equals a short key. + channelListMaxKeyBytes = 256 ) +// channelListStoreKey is the key the cache and the flight use for key. +func channelListStoreKey(key string) string { + if len(key) <= channelListMaxKeyBytes { + return key + } + sum := sha256.Sum256([]byte(key)) + return "sha256:" + hex.EncodeToString(sum[:]) +} + // channelsRegionKey is the cache and flight key for a region parameter: the // normalized codes (normalizeRegionCodes), sorted and de-duplicated. The // query runs with normalizeRegionCodes(key), the same set of codes. @@ -96,6 +113,7 @@ func (c *channelListCache) reset() { // query shared by every concurrent caller of key. missed runs after the // first cache check fails (a test seam; nil in production). func (c *channelListCache) load(key string, missed func(), query func() (*channelListEntry, error)) (*channelListEntry, error) { + key = channelListStoreKey(key) if e, ok := c.get(key, time.Now()); ok { return e, nil } From 390af21f4a9108f9eb130ffd65bbd9396ebe9cd8 Mon Sep 17 00:00:00 2001 From: dborup Date: Wed, 30 Sep 2026 15:16:19 +0200 Subject: [PATCH 5/5] test(channels): pin that a long region key is queried unhashed (#109) Only the cache and the flight use the bounded (sha256) key; the query must still get the full normalized region. Without this, passing the digest to the query would look up the region "sha256:..." and cache an empty list for 60 s, and no test caught it (review mutant G3). Also correct the comment on why a digest never equals a short key: region codes are upper-case, so none starts with the lower-case "sha256:". Co-Authored-By: Claude Opus 5.5 --- cmd/server/channels_key_bound_109_test.go | 39 +++++++++++++++++++++++ cmd/server/channels_list_cache.go | 5 +-- 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/cmd/server/channels_key_bound_109_test.go b/cmd/server/channels_key_bound_109_test.go index 1797e1ba4..d8fbe0c49 100644 --- a/cmd/server/channels_key_bound_109_test.go +++ b/cmd/server/channels_key_bound_109_test.go @@ -3,6 +3,7 @@ package main import ( "strconv" "strings" + "sync" "testing" ) @@ -53,3 +54,41 @@ func TestChannelListCacheKeyLengthIsBounded_109(t *testing.T) { } } } + +// Only the cache and the flight use the bounded key: the query itself must +// still get the full normalized region, or a long region parameter would be +// looked up as the region "sha256:…" and cache an empty list. +func TestChannelListLongKeyQueriesFullRegion_109(t *testing.T) { + db := singleflightDB(t) + var mu sync.Mutex + got := map[string][]string{} + db.channelsQueryHook = func(kind, region string) error { + mu.Lock() + got[kind] = append(got[kind], region) + mu.Unlock() + return nil + } + many := make([]string, 300) + for i := range many { + many[i] = "r" + strconv.Itoa(i) + } + param := strings.Join(many, ",") + want := channelsRegionKey(param) + if len(want) <= channelListMaxKeyBytes { + t.Fatalf("test region key is %d bytes, want more than %d", len(want), channelListMaxKeyBytes) + } + if _, err := db.GetChannels(param); err != nil { + t.Fatal(err) + } + if _, err := db.GetEncryptedChannels(param); err != nil { + t.Fatal(err) + } + for _, kind := range []string{"channels", "encrypted"} { + if len(got[kind]) != 1 { + t.Fatalf("%s: %d queries, want 1", kind, len(got[kind])) + } + if got[kind][0] != want { + t.Errorf("%s: query got region %.40q…, want the full normalized key %.40q…", kind, got[kind][0], want) + } + } +} diff --git a/cmd/server/channels_list_cache.go b/cmd/server/channels_list_cache.go index fd734b9f3..6c1f4f880 100644 --- a/cmd/server/channels_list_cache.go +++ b/cmd/server/channels_list_cache.go @@ -40,8 +40,9 @@ const ( channelListTTL = 60 * time.Second channelListMaxKeys = 64 // channelListMaxKeyBytes caps a stored key. A longer one becomes - // "sha256:" + 64 hex digits (71 bytes); no normalized region key - // contains ':', so a digest never equals a short key. + // "sha256:" + 64 hex digits (71 bytes). Normalized region codes are + // upper-case (normalizeRegionCodes), so no short key starts with the + // lower-case "sha256:" and a digest never equals a short key. channelListMaxKeyBytes = 256 )