From df3c87f30420c1f05b5f4364786ee80cf419d7bf Mon Sep 17 00:00:00 2001 From: R0ck Date: Fri, 2 Oct 2026 00:45:58 +0100 Subject: [PATCH] perf(nodes): cache region membership for /api/nodes?region= (#2101) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `/api/nodes?region=X` filtered with an inline `public_key IN (SELECT DISTINCT from_pubkey FROM transmissions ⋈ observations ⋈ observers ...)` subquery. It was uncached and evaluated twice per request (COUNT(*) and the page), and fetchAllNodes() pages at 500, so each Nodes or Map view ran it several times. Measured on a 1.4 GB database (1.47M observations, 2 vCPU): ~10s per evaluation, 18-21s per region-filtered page on v3.12.0, and one open tab refreshing every minute kept the reader pool and both cores busy. Membership is cached per canonical region set (sorted, de-duplicated codes) with an observations.id watermark: - fresh for 30s; after that the cached set is served while one background refresh (singleflight, #1910 pattern) scans only observations past the watermark, with the join order forced to drive from the rowid range; - every 30 minutes the refresh is a full rebuild instead, so retention pruning, observer IATA changes and late from_pubkey backfills are not held forever. Full rebuilds are serialised across region sets, since each holds a pooled connection for seconds; - only the first request for a region set waits on a full scan, and concurrent first requests share it; - a failed refresh is logged and leaves the previous entry serving; scans run under a context bounded by the rebuild interval; - entries are immutable once stored and the cache holds at most 32. GetNodes binds the set through `public_key IN (SELECT value FROM json_each(?))`, so COUNT and SELECT no longer re-run the join. Results match the previous subquery (the tests use it as the oracle, on v2 and v3 schemas); the only change is that a node newly heard in a region can take up to one refresh interval to appear. Co-Authored-By: Claude Opus 5.5 --- cmd/server/db.go | 50 ++-- cmd/server/nodes_region_cache.go | 316 +++++++++++++++++++++ cmd/server/nodes_region_cache_test.go | 387 ++++++++++++++++++++++++++ 3 files changed, 727 insertions(+), 26 deletions(-) create mode 100644 cmd/server/nodes_region_cache.go create mode 100644 cmd/server/nodes_region_cache_test.go diff --git a/cmd/server/db.go b/cmd/server/db.go index e6f1a3dd3..e51cba30c 100644 --- a/cmd/server/db.go +++ b/cmd/server/db.go @@ -78,6 +78,20 @@ type DB struct { // GetEncryptedChannels. Nil in production. encChannelsQueryHook func() + // GetNodes region membership (public keys heard per region set), with an + // observations.id watermark for incremental refresh. See + // nodes_region_cache.go. + nodeRegionCacheMu sync.Mutex + nodeRegionCache map[string]*nodeRegionEntry + nodeRegionSF singleflight.Group + // Serialises full membership rebuilds across region sets: each holds one + // of the 4 pooled connections for seconds on a large database, and + // entries built together at startup fall due together. + nodeRegionFullMu sync.Mutex + // Test-only hook fired at the start of each real membership scan (full or + // delta), inside nodeRegionSF's flight. Nil in production. + nodeRegionQueryHook func() + // 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. @@ -1130,33 +1144,17 @@ func (db *DB) GetNodes(limit, offset int, role, search, before, lastHeard, sortB } } - if region != "" { - codes := normalizeRegionCodes(region) - if len(codes) > 0 { - placeholders := make([]string, len(codes)) - regionArgs := make([]interface{}, len(codes)) - for i, c := range codes { - placeholders[i] = "?" - regionArgs[i] = c - } - joinCond := "obs.rowid = o.observer_idx" - if !db.isV3 { - joinCond = "obs.id = o.observer_id" - } - // #1143: from_pubkey is a dedicated, indexed column populated at - // ingest (and backfilled) for ADVERT rows specifically so pubkey - // lookups don't need to JSON_EXTRACT + parse decoded_json per row. - subq := fmt.Sprintf(`public_key IN ( - SELECT DISTINCT t.from_pubkey - FROM transmissions t - JOIN observations o ON o.transmission_id = t.id - JOIN observers obs ON %s - WHERE t.payload_type = 4 - AND UPPER(TRIM(obs.iata)) IN (%s) - )`, joinCond, strings.Join(placeholders, ",")) - where = append(where, subq) - args = append(args, regionArgs...) + if codes := normalizeRegionCodes(region); len(codes) > 0 { + // Nodes whose ADVERTs were heard by an observer in the region. The + // membership set is cached and refreshed incrementally rather than + // re-derived by a transmissions ⋈ observations scan on every page + // (twice: COUNT and SELECT). See nodes_region_cache.go. + keysJSON, err := db.nodeRegionKeysJSON(codes) + if err != nil { + return nil, 0, nil, err } + where = append(where, "public_key IN (SELECT value FROM json_each(?))") + args = append(args, keysJSON) } w := "" diff --git a/cmd/server/nodes_region_cache.go b/cmd/server/nodes_region_cache.go new file mode 100644 index 000000000..723e3fab7 --- /dev/null +++ b/cmd/server/nodes_region_cache.go @@ -0,0 +1,316 @@ +package main + +import ( + "context" + "database/sql" + "encoding/json" + "errors" + "fmt" + "log" + "sort" + "strings" + "time" +) + +// Region-scoped node membership cache for GetNodes (#2101). +// +// `/api/nodes?region=X` restricts the node list to nodes whose ADVERTs were +// heard by an observer in region X. That used to be an inline +// `public_key IN (SELECT DISTINCT from_pubkey FROM transmissions ⋈ +// observations ⋈ observers ...)` subquery, uncached, evaluated twice per +// request (COUNT(*) and the page). On a 1.4 GB database (1.47M observations) +// each evaluation took ~10s, and fetchAllNodes() pages at 500, so a single +// open Nodes or Map tab kept a 2-vCPU host busy. +// +// Membership only grows as observations arrive, so it is cached per region +// set with an observations.id watermark: +// - fresh (< nodeRegionFreshTTL): served as-is; +// - stale: served as-is while ONE background refresh scans only +// observations with id > watermark (a rowid range, milliseconds); +// - every nodeRegionRebuildInterval the refresh is a full rebuild instead, +// so retention pruning, observer IATA changes and late from_pubkey +// backfills are picked up. +// +// Only the first request for a region set waits on a full scan, and +// concurrent first requests share it (the #1910 singleflight pattern, as +// channelsSF). + +const ( + nodeRegionFreshTTL = 30 * time.Second + nodeRegionRebuildInterval = 30 * time.Minute + // nodeRegionMaxEntries bounds the cache. Entries hold up to a few thousand + // pubkeys each, so this is far below the shared maxCacheEntries; in + // practice an instance sees a handful of distinct region sets. + nodeRegionMaxEntries = 32 +) + +// errNodeRegionFlightResult means nodeRegionSF returned something other than +// a *nodeRegionEntry, which only a programming error can cause. +var errNodeRegionFlightResult = errors.New("unexpected node region flight result") + +// nodeRegionEntry is immutable once stored: refreshes build a new entry, so +// readers can use one without holding nodeRegionCacheMu. +type nodeRegionEntry struct { + keys []string // sorted from_pubkeys heard in the region set + keysJSON string // keys as a JSON array, bound to json_each(?) + watermark int64 // max observations.id covered by keys + refreshed time.Time + built time.Time // last full rebuild +} + +func (e *nodeRegionEntry) fresh() bool { + return time.Since(e.refreshed) < nodeRegionFreshTTL +} + +func (e *nodeRegionEntry) rebuildDue() bool { + return time.Since(e.built) >= nodeRegionRebuildInterval +} + +// newNodeRegionEntry snapshots set into an entry. +func newNodeRegionEntry(set map[string]struct{}, watermark int64, built time.Time) (*nodeRegionEntry, error) { + keys := make([]string, 0, len(set)) + for k := range set { + keys = append(keys, k) + } + sort.Strings(keys) + buf, err := json.Marshal(keys) + if err != nil { + return nil, fmt.Errorf("encode region keys: %w", err) + } + return &nodeRegionEntry{ + keys: keys, + keysJSON: string(buf), + watermark: watermark, + refreshed: time.Now(), + built: built, + }, nil +} + +// nodeRegionKey canonicalises normalised region codes so "EDI,GLA", +// "gla, edi" and "EDI,EDI,GLA" share one cache entry. +func nodeRegionKey(codes []string) string { + seen := make(map[string]struct{}, len(codes)) + uniq := make([]string, 0, len(codes)) + for _, c := range codes { + if _, ok := seen[c]; !ok { + seen[c] = struct{}{} + uniq = append(uniq, c) + } + } + sort.Strings(uniq) + return strings.Join(uniq, ",") +} + +func (db *DB) getNodeRegionEntry(key string) *nodeRegionEntry { + db.nodeRegionCacheMu.Lock() + defer db.nodeRegionCacheMu.Unlock() + return db.nodeRegionCache[key] +} + +func (db *DB) setNodeRegionEntry(key string, e *nodeRegionEntry) { + db.nodeRegionCacheMu.Lock() + defer db.nodeRegionCacheMu.Unlock() + if db.nodeRegionCache == nil || len(db.nodeRegionCache) >= nodeRegionMaxEntries { + db.nodeRegionCache = make(map[string]*nodeRegionEntry) + } + db.nodeRegionCache[key] = e +} + +// nodeRegionKeysJSON returns the JSON array of node public keys heard in the +// given (normalised, non-empty) region codes, for binding to +// `public_key IN (SELECT value FROM json_each(?))`. +func (db *DB) nodeRegionKeysJSON(codes []string) (string, error) { + key := nodeRegionKey(codes) + + if e := db.getNodeRegionEntry(key); e != nil { + if !e.fresh() { + // Stale-while-revalidate: kick one refresh, don't wait on it. + // DoChan dedups against an in-flight refresh for the same key, + // and its buffered result channel may be left unread. + db.nodeRegionSF.DoChan(key, func() (any, error) { + return db.refreshNodeRegion(key, codes) + }) + } + return e.keysJSON, nil + } + + v, err, _ := db.nodeRegionSF.Do(key, func() (any, error) { + return db.refreshNodeRegion(key, codes) + }) + if err != nil { + return "", err + } + e, ok := v.(*nodeRegionEntry) + if !ok { + return "", fmt.Errorf("nodes-region %s: %w: %T", key, errNodeRegionFlightResult, v) + } + return e.keysJSON, nil +} + +// refreshNodeRegion brings the cache entry for key up to date and stores it: +// a delta scan past the watermark when an entry exists and is not due a +// rebuild, a full scan otherwise. It runs inside nodeRegionSF, so at most one +// refresh per key is in flight. Failures are logged here because a background +// refresh has no caller to report to; the previous entry stays in place. +func (db *DB) refreshNodeRegion(key string, codes []string) (*nodeRegionEntry, error) { + prev := db.getNodeRegionEntry(key) + if prev != nil && prev.fresh() { + return prev, nil // a flight that finished just before this one got here + } + if db.nodeRegionQueryHook != nil { + db.nodeRegionQueryHook() + } + + // A scan still running when the next rebuild would be due is abandoned + // rather than left holding one of the pooled connections. + ctx, cancel := context.WithTimeout(context.Background(), nodeRegionRebuildInterval) + defer cancel() + + var e *nodeRegionEntry + var err error + if prev == nil || prev.rebuildDue() { + e, err = db.buildNodeRegion(ctx, key, codes) + } else { + e, err = db.extendNodeRegion(ctx, key, codes, prev) + } + if err != nil { + err = fmt.Errorf("nodes-region %s: %w", key, err) + log.Printf("[nodes-region] refresh failed: %v", err) + return nil, err + } + db.setNodeRegionEntry(key, e) + return e, nil +} + +// buildNodeRegion scans the full observation history for the region set. +// Full builds are serialised across region sets: each holds a pooled +// connection for seconds on a large database, and entries built together at +// startup fall due together. +func (db *DB) buildNodeRegion(ctx context.Context, key string, codes []string) (*nodeRegionEntry, error) { + db.nodeRegionFullMu.Lock() + defer db.nodeRegionFullMu.Unlock() + + wm, err := db.maxObservationID(ctx) + if err != nil { + return nil, err + } + start := time.Now() + set := make(map[string]struct{}) + if _, err := db.scanNodeRegionKeys(ctx, codes, 0, wm, false, set); err != nil { + return nil, err + } + log.Printf("[nodes-region] %s: full build, %d nodes in %v (obs id <= %d)", + key, len(set), time.Since(start).Round(time.Millisecond), wm) + return newNodeRegionEntry(set, wm, start) +} + +// extendNodeRegion adds nodes heard since prev's watermark. +func (db *DB) extendNodeRegion( + ctx context.Context, key string, codes []string, prev *nodeRegionEntry, +) (*nodeRegionEntry, error) { + wm, err := db.maxObservationID(ctx) + if err != nil { + return nil, err + } + set := make(map[string]struct{}, len(prev.keys)) + for _, k := range prev.keys { + set[k] = struct{}{} + } + start := time.Now() + added := 0 + if wm > prev.watermark { + added, err = db.scanNodeRegionKeys(ctx, codes, prev.watermark, wm, true, set) + if err != nil { + return nil, err + } + } + if added == 0 { + // Same membership: keep the encoded keys, advance the watermark. + e := *prev + e.watermark = max(wm, prev.watermark) + e.refreshed = time.Now() + return &e, nil + } + log.Printf("[nodes-region] %s: +%d nodes from obs %d..%d in %v", + key, added, prev.watermark+1, wm, time.Since(start).Round(time.Millisecond)) + return newNodeRegionEntry(set, wm, prev.built) +} + +// maxObservationID pins the upper bound of a scan. The ingestor is the single +// writer and observations.id is AUTOINCREMENT, so every id <= the returned +// value is committed and visible to a later read; newer rows are left for the +// next delta. +func (db *DB) maxObservationID(ctx context.Context) (int64, error) { + const q = "SELECT COALESCE(MAX(id), 0) FROM observations" + // Issue exactly one query: an unscanned *sql.Row keeps its connection + // checked out, and the pool is 4 (OpenDB). + var row *sql.Row + if db.stmtMaxObsID != nil { + row = db.stmtMaxObsID.QueryRowContext(ctx) + } else { + row = db.conn.QueryRowContext(ctx, q) + } + var wm int64 + if err := row.Scan(&wm); err != nil { + return 0, fmt.Errorf("max observation id: %w", err) + } + return wm, nil +} + +// scanNodeRegionKeys adds to set every ADVERT from_pubkey observed by an +// observer in codes with lo < observations.id <= hi, returning how many were +// new. For a delta scan the join order is forced (CROSS JOIN) to drive from +// the observations rowid range; left to the planner, sqlite_stat1 (#2058) +// makes it start from the region's observers and walk their whole history. +func (db *DB) scanNodeRegionKeys( + ctx context.Context, codes []string, lo, hi int64, delta bool, set map[string]struct{}, +) (int, error) { + placeholders := make([]string, len(codes)) + args := []any{payloadTypeAdvert} + for i, c := range codes { + placeholders[i] = "?" + args = append(args, c) + } + args = append(args, lo, hi) + + joinCond := "obs.rowid = o.observer_idx" + if !db.isV3 { + joinCond = "obs.id = o.observer_id" + } + join := "JOIN" + if delta { + join = "CROSS JOIN" + } + // Only fixed fragments and "?" placeholders are formatted in; every value + // is bound. #1143: from_pubkey is a dedicated, indexed column populated at + // ingest (and backfilled) for ADVERT rows, so no JSON_EXTRACT is needed. + q := fmt.Sprintf(`SELECT DISTINCT t.from_pubkey + FROM observations o + %[1]s transmissions t ON t.id = o.transmission_id + %[1]s observers obs ON %[2]s + WHERE t.payload_type = ? + AND t.from_pubkey IS NOT NULL + AND UPPER(TRIM(obs.iata)) IN (%[3]s) + AND o.id > ? AND o.id <= ?`, join, joinCond, strings.Join(placeholders, ",")) + + rows, err := db.conn.QueryContext(ctx, q, args...) + if err != nil { + return 0, fmt.Errorf("scan region keys: %w", err) + } + defer rows.Close() + added := 0 + for rows.Next() { + var pk string + if err := rows.Scan(&pk); err != nil { + return added, fmt.Errorf("scan region keys: %w", err) + } + if _, ok := set[pk]; !ok { + set[pk] = struct{}{} + added++ + } + } + if err := rows.Err(); err != nil { + return added, fmt.Errorf("scan region keys: %w", err) + } + return added, nil +} diff --git a/cmd/server/nodes_region_cache_test.go b/cmd/server/nodes_region_cache_test.go new file mode 100644 index 000000000..30a376899 --- /dev/null +++ b/cmd/server/nodes_region_cache_test.go @@ -0,0 +1,387 @@ +package main + +import ( + "fmt" + "sort" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +const ( + regionSJC = "SJC" // seedTestData: obs1 + regionSFO = "SFO" // seedTestData: obs2 +) + +// legacyRegionPubkeys is the pre-cache GetNodes region subquery, kept here as +// the oracle the cached membership must agree with. +func legacyRegionPubkeys(t *testing.T, db *DB, region string) []string { + t.Helper() + codes := normalizeRegionCodes(region) + if len(codes) == 0 { + return nil + } + ph := make([]string, len(codes)) + args := make([]any, len(codes)) + for i, c := range codes { + ph[i] = "?" + args[i] = c + } + joinCond := "obs.rowid = o.observer_idx" + if !db.isV3 { + joinCond = "obs.id = o.observer_id" + } + q := fmt.Sprintf(`SELECT public_key FROM nodes WHERE public_key IN ( + SELECT DISTINCT t.from_pubkey + FROM transmissions t + JOIN observations o ON o.transmission_id = t.id + JOIN observers obs ON %s + WHERE t.payload_type = 4 + AND UPPER(TRIM(obs.iata)) IN (%s) + )`, joinCond, strings.Join(ph, ",")) + rows, err := db.conn.Query(q, args...) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var out []string + for rows.Next() { + var pk string + if err := rows.Scan(&pk); err != nil { + t.Fatal(err) + } + out = append(out, pk) + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + sort.Strings(out) + return out +} + +func regionNodePubkeys(t *testing.T, db *DB, region string) []string { + t.Helper() + nodes, total, _, err := db.GetNodes(500, 0, "", "", "", "", "", region) + if err != nil { + t.Fatal(err) + } + if total != len(nodes) { + t.Fatalf("region %q: total %d != page length %d", region, total, len(nodes)) + } + out := make([]string, 0, len(nodes)) + for _, n := range nodes { + pk, ok := n["public_key"].(string) + if !ok { + t.Fatalf("region %q: public_key %v is not a string", region, n["public_key"]) + } + out = append(out, pk) + } + sort.Strings(out) + return out +} + +// seedRegionAdvert adds a node heard via ADVERT by observer observerIdx (v3). +func seedRegionAdvert(t *testing.T, db *DB, pubkey, hash string, observerIdx int) { + t.Helper() + now := time.Now().UTC() + mustExec(t, db, `INSERT OR IGNORE INTO nodes (public_key, name, role, last_seen, first_seen) + VALUES (?, ?, 'repeater', ?, ?)`, pubkey, "n-"+pubkey, now.Format(time.RFC3339), now.Format(time.RFC3339)) + res, err := db.conn.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json, from_pubkey) + VALUES ('AA', ?, ?, 1, 4, '{}', ?)`, hash, now.Format(time.RFC3339), pubkey) + if err != nil { + t.Fatal(err) + } + txID, err := res.LastInsertId() + if err != nil { + t.Fatal(err) + } + mustExec(t, db, `INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp) + VALUES (?, ?, 5, -100, '[]', ?)`, txID, observerIdx, now.Unix()) +} + +// ageNodeRegionEntries makes every cached entry stale, and optionally due a +// full rebuild, without sleeping through the real intervals. Entries are +// immutable in production, so each is replaced rather than edited. +func ageNodeRegionEntries(db *DB, rebuild bool) { + db.nodeRegionCacheMu.Lock() + defer db.nodeRegionCacheMu.Unlock() + for k, e := range db.nodeRegionCache { + aged := *e + aged.refreshed = aged.refreshed.Add(-2 * nodeRegionFreshTTL) + if rebuild { + aged.built = aged.built.Add(-2 * nodeRegionRebuildInterval) + } + db.nodeRegionCache[k] = &aged + } +} + +// waitForRegionNodes polls until region's node list equals want: after a +// stale hit the refresh runs in the background. +func waitForRegionNodes(t *testing.T, db *DB, region string, want []string) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + var got []string + for time.Now().Before(deadline) { + got = regionNodePubkeys(t, db, region) + if strings.Join(got, ",") == strings.Join(want, ",") { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatalf("region %q: got %v, want %v", region, got, want) +} + +func assertMatchesLegacy(t *testing.T, db *DB, region string, wantNonEmpty bool) { + t.Helper() + want := legacyRegionPubkeys(t, db, region) + if wantNonEmpty && len(want) == 0 { + t.Fatalf("region %q: fixture produced no legacy matches, the comparison would be vacuous", region) + } + if got := regionNodePubkeys(t, db, region); strings.Join(got, ",") != strings.Join(want, ",") { + t.Errorf("region %q: cached %v, legacy %v", region, got, want) + } +} + +func TestNodeRegionCacheMatchesLegacyQuery(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + seedTestData(t, db) + seedRegionAdvert(t, db, "cafe000000000001", "h-sfo-only", 2) + + for _, region := range []string{regionSJC, regionSFO, "SJC,SFO", "sfo , sjc", "SJC,SJC", "SJC,AMS"} { + assertMatchesLegacy(t, db, region, true) + } + assertMatchesLegacy(t, db, "AMS", false) +} + +func TestNodeRegionCacheMatchesLegacyQueryV2(t *testing.T) { + db := setupTestDBV2(t) + defer db.Close() + now := time.Now().UTC().Format(time.RFC3339) + mustExec(t, db, `INSERT INTO observers (id, name, iata, last_seen, first_seen, packet_count) + VALUES ('obs-v2-1', 'V2 Observer', ' lax ', ?, '2026-01-01T00:00:00Z', 10)`, now) + mustExec(t, db, `INSERT INTO nodes (public_key, name, role, last_seen, first_seen) + VALUES ('v2pubkey11223344', 'V2Node', 'repeater', ?, '2026-01-01T00:00:00Z')`, now) + mustExec(t, db, `INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json, from_pubkey) + VALUES ('AABB', 'v2hash0001', ?, 1, 4, '{}', 'v2pubkey11223344')`, now) + mustExec(t, db, `INSERT INTO observations (transmission_id, observer_id, observer_name, snr, rssi, path_json, timestamp) + VALUES (1, 'obs-v2-1', 'V2 Observer', 10.0, -90, '[]', ?)`, time.Now().Unix()) + + assertMatchesLegacy(t, db, "LAX", true) + assertMatchesLegacy(t, db, "lax", true) + assertMatchesLegacy(t, db, "JFK", false) +} + +// Equivalent spellings of one region set share one cache entry, and the +// COUNT + page of a request, and later pages, do not rescan. +func TestNodeRegionCacheKeyAndReuse(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + seedTestData(t, db) + var scans atomic.Int32 + db.nodeRegionQueryHook = func() { scans.Add(1) } + + for _, region := range []string{"SJC,SFO", "sfo, sjc", "SFO,SJC,SJC"} { + for offset := range 3 { + if _, _, _, err := db.GetNodes(1, offset, "", "", "", "", "", region); err != nil { + t.Fatal(err) + } + } + } + if n := scans.Load(); n != 1 { + t.Errorf("expected 1 membership scan for one region set, got %d", n) + } +} + +// Concurrent cold requests for the same region share one scan (#1910 class). +func TestNodeRegionCacheCoalescesColdMisses(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + seedTestData(t, db) + var scans atomic.Int32 + release := make(chan struct{}) + db.nodeRegionQueryHook = func() { + scans.Add(1) + <-release + } + + const callers = 8 + var wg sync.WaitGroup + errs := make(chan error, callers) + for range callers { + wg.Add(1) + go func() { + defer wg.Done() + _, _, _, err := db.GetNodes(50, 0, "", "", "", "", "", regionSJC) + errs <- err + }() + } + time.Sleep(100 * time.Millisecond) // let every caller reach the flight + close(release) + wg.Wait() + close(errs) + for err := range errs { + if err != nil { + t.Fatal(err) + } + } + if n := scans.Load(); n != 1 { + t.Errorf("expected 1 coalesced scan for %d cold callers, got %d", callers, n) + } +} + +// A node first heard after the cache was built appears after the next +// (delta) refresh; until the entry goes stale the cached set is served. +func TestNodeRegionCacheDeltaRefresh(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + seedTestData(t, db) + + before := regionNodePubkeys(t, db, regionSJC) + seedRegionAdvert(t, db, "cafe000000000002", "h-new-sjc", 1) + if got := regionNodePubkeys(t, db, regionSJC); strings.Join(got, ",") != strings.Join(before, ",") { + t.Fatalf("fresh entry should be served unchanged, got %v want %v", got, before) + } + + builtBefore := db.getNodeRegionEntry(regionSJC).built + ageNodeRegionEntries(db, false) + waitForRegionNodes(t, db, regionSJC, legacyRegionPubkeys(t, db, regionSJC)) + + e := db.getNodeRegionEntry(regionSJC) + if !e.built.Equal(builtBefore) { + t.Errorf("expected a delta refresh (built unchanged), built moved %v -> %v", builtBefore, e.built) + } + if !contains(e.keys, "cafe000000000002") { + t.Errorf("new node missing from refreshed keys %v", e.keys) + } +} + +// A delta that finds nothing new keeps the encoded keys and advances the +// watermark past observations that did not change membership. +func TestNodeRegionCacheDeltaWithoutNewNodes(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + seedTestData(t, db) + regionNodePubkeys(t, db, regionSJC) + prev := db.getNodeRegionEntry(regionSJC) + + // A new observation of a node already in the region. + mustExec(t, db, `INSERT INTO observations (transmission_id, observer_idx, snr, rssi, path_json, timestamp) + VALUES (1, 1, 5, -100, '[]', ?)`, time.Now().Unix()) + ageNodeRegionEntries(db, false) + e, err := db.refreshNodeRegion(regionSJC, []string{regionSJC}) + if err != nil { + t.Fatal(err) + } + if e.keysJSON != prev.keysJSON { + t.Errorf("membership unchanged, keysJSON should be reused: %s vs %s", e.keysJSON, prev.keysJSON) + } + if e.watermark <= prev.watermark { + t.Errorf("watermark should advance past the new observation: %d -> %d", prev.watermark, e.watermark) + } +} + +// Deltas only add; the periodic full rebuild drops nodes whose region +// observations have since been pruned. +func TestNodeRegionCacheFullRebuildDropsPruned(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + seedTestData(t, db) + seedRegionAdvert(t, db, "cafe000000000003", "h-prune-sjc", 1) + if got := regionNodePubkeys(t, db, regionSJC); !contains(got, "cafe000000000003") { + t.Fatalf("seeded node missing: %v", got) + } + + mustExec(t, db, `DELETE FROM observations WHERE transmission_id IN + (SELECT id FROM transmissions WHERE hash = 'h-prune-sjc')`) + ageNodeRegionEntries(db, true) + waitForRegionNodes(t, db, regionSJC, legacyRegionPubkeys(t, db, regionSJC)) +} + +// A failed refresh reports the error and leaves the previous entry serving. +func TestNodeRegionCacheRefreshFailureKeepsPrevious(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + seedTestData(t, db) + want := regionNodePubkeys(t, db, regionSJC) + prev := db.getNodeRegionEntry(regionSJC) + + mustExec(t, db, `ALTER TABLE observations RENAME TO observations_gone`) + ageNodeRegionEntries(db, false) + if _, err := db.refreshNodeRegion(regionSJC, []string{regionSJC}); err == nil { + t.Fatal("expected refresh to fail without an observations table") + } + if e := db.getNodeRegionEntry(regionSJC); e.keysJSON != prev.keysJSON { + t.Errorf("failed refresh replaced the entry: %s -> %s", prev.keysJSON, e.keysJSON) + } + // Stale hits keep serving the previous set while refreshes fail. + if got := regionNodePubkeys(t, db, regionSJC); strings.Join(got, ",") != strings.Join(want, ",") { + t.Errorf("stale entry should still be served, got %v want %v", got, want) + } +} + +// The cache is bounded: distinct region sets beyond nodeRegionMaxEntries +// reset it instead of growing it. +func TestNodeRegionCacheBounded(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + seedTestData(t, db) + for i := range nodeRegionMaxEntries + 5 { + if _, _, _, err := db.GetNodes(1, 0, "", "", "", "", "", fmt.Sprintf("SJC,X%02d", i)); err != nil { + t.Fatal(err) + } + } + db.nodeRegionCacheMu.Lock() + n := len(db.nodeRegionCache) + db.nodeRegionCacheMu.Unlock() + if n > nodeRegionMaxEntries { + t.Errorf("cache holds %d entries, cap is %d", n, nodeRegionMaxEntries) + } +} + +// Refreshes must hand every connection back. The production pool is 4 +// (OpenDB); a leaked *sql.Row per refresh exhausted it after four stale +// hits and hung every DB-backed endpoint. The prepared MAX(id) statement is +// set up here because production always has it. +func TestNodeRegionCacheReleasesConnections(t *testing.T) { + db := setupTestDB(t) + seedTestData(t, db) + stmt, err := db.conn.Prepare("SELECT COALESCE(MAX(id), 0) FROM observations") + if err != nil { + t.Fatal(err) + } + db.stmtMaxObsID = stmt + + // The test pool holds one connection, so a leak deadlocks rather than + // failing; refresh off the test goroutine and fail on a deadline instead. + // Closing is left to the success path: with a connection leaked, Close + // would block on it too. + for i := range 10 { + seedRegionAdvert(t, db, fmt.Sprintf("cafe0000000001%02d", i), fmt.Sprintf("h-leak-%d", i), 1) + ageNodeRegionEntries(db, i%3 == 0) + done := make(chan error, 1) + go func() { + _, err := db.refreshNodeRegion(regionSJC, []string{regionSJC}) + done <- err + }() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(10 * time.Second): + t.Fatalf("refresh %d did not complete: a pooled connection was not returned", i) + } + } + if inUse := db.conn.Stats().InUse; inUse != 0 { + t.Errorf("expected all connections returned after refreshes, %d still in use", inUse) + } + if err := stmt.Close(); err != nil { + t.Error(err) + } + if err := db.Close(); err != nil { + t.Error(err) + } +}