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_key_bound_109_test.go b/cmd/server/channels_key_bound_109_test.go new file mode 100644 index 000000000..d8fbe0c49 --- /dev/null +++ b/cmd/server/channels_key_bound_109_test.go @@ -0,0 +1,94 @@ +package main + +import ( + "strconv" + "strings" + "sync" + "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) + } + } +} + +// 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 new file mode 100644 index 000000000..6c1f4f880 --- /dev/null +++ b/cmd/server/channels_list_cache.go @@ -0,0 +1,140 @@ +package main + +import ( + "crypto/sha256" + "encoding/hex" + "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 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 +// full slice expression, #98). +type channelListCache struct { + mu sync.Mutex + entries map[string]*channelListEntry + flights singleflight.Group +} + +const ( + channelListTTL = 60 * time.Second + channelListMaxKeys = 64 + // channelListMaxKeyBytes caps a stored key. A longer one becomes + // "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 +) + +// 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. +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) { + key = channelListStoreKey(key) + 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 new file mode 100644 index 000000000..d452d1e15 --- /dev/null +++ b/cmd/server/channels_singleflight_109_test.go @@ -0,0 +1,401 @@ +package main + +import ( + "errors" + "reflect" + "strconv" + "strings" + "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/2)%len(regions)] // every region with both kinds + 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) + } + }) + } +} + +// 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) + } + } +} + +// 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) { + 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 37441f492..d05d7da8c 100644 --- a/cmd/server/db.go +++ b/cmd/server/db.go @@ -95,14 +95,45 @@ 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 + // 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 + // 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. +// Never modified after it is stored (see channelListCache). +type channelListEntry struct { + channels []map[string]interface{} + expires time.Time } // isV3, hasResolvedPath, hasObsRawHex, hasScopeName, hasDefaultScope, @@ -3098,17 +3129,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 + 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 } - db.channelsCacheMu.Unlock() + 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 { @@ -3120,6 +3153,11 @@ func (db *DB) GetChannels(region ...string) ([]map[string]interface{}, error) { regionPlaceholder = strings.Join(placeholders, ",") } + if db.channelsQueryHook != nil { + if err := db.channelsQueryHook("channels", key); 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. @@ -3134,12 +3172,13 @@ func (db *DB) GetChannels(region ...string) ([]map[string]interface{}, 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) @@ -3171,15 +3210,21 @@ func (db *DB) GetChannels(region ...string) ([]map[string]interface{}, 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 + } - // 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). @@ -3189,7 +3234,20 @@ func (db *DB) GetEncryptedChannels(region ...string) ([]map[string]interface{}, if len(region) > 0 { regionParam = region[0] } - regionCodes := normalizeRegionCodes(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 + } + 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)) @@ -3240,17 +3298,23 @@ func (db *DB) GetEncryptedChannels(region ...string) ([]map[string]interface{}, ORDER BY last_activity DESC` } + if db.channelsQueryHook != nil { + if err := db.channelsQueryHook("encrypted", key); err != nil { + return nil, err + } + } rows, err := db.conn.Query(querySQL, args...) if err != nil { 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" @@ -3265,7 +3329,10 @@ func (db *DB) GetEncryptedChannels(region ...string) ([]map[string]interface{}, "encrypted": true, }) } - return channels, nil + if err := it.Err(); err != nil { + return nil, err + } + return &channelListEntry{channels: channels}, nil } // GetChannelMessages returns messages for a specific channel.