diff --git a/cmd/server/db.go b/cmd/server/db.go index e5ab31a1f..bdf5f9a31 100644 --- a/cmd/server/db.go +++ b/cmd/server/db.go @@ -136,26 +136,25 @@ type DB struct { nodeRegionTick int64 nodeRegionSF singleflight.Group nodeRegionFullMu sync.Mutex - // nodeRegionQueryHook (test seam, nil in production) runs once per real - // membership scan, before the query; a non-nil error fails that refresh - // the way a failing scan does. + // nodeRegionColdMu serialises the per-set builds cold requests fall + // back to while the region index build fails (coldNodeRegion). + nodeRegionColdMu sync.Mutex + // nodeRegionQueryHook (test seam, nil in production) runs once per + // per-set refresh, before it reads the region index; a non-nil error + // fails that refresh the way a failing scan does. nodeRegionQueryHook func() error - - // Direct-evidence region placement (#488 M2, region_placement.go). - // regionPlacement is set once at startup, before the first request. - // regionAgg is the global evidence aggregate, owned by the refresh - // that holds regionAggMu; readers only ever see regionSnap, an - // immutable snapshot swapped in after each refresh. + // nodeRegionPerSetHook (test seam, nil in production) runs at the start + // of every per-set build (buildNodeRegionCold). + nodeRegionPerSetHook func() + + // Global region indexes the per-set entries are cut from (#618, + // region_index.go): one per region rule, the heard rule's and the + // direct-evidence placement's (#488 M2, region_placement.go). + // regionPlacement is set once at startup, before the first request, + // and selects the one served. regionPlacement regionPlacementSettings - regionAggMu sync.Mutex - regionAgg *regionEvidenceAgg - regionAggBuilt time.Time - regionSnap atomic.Pointer[regionPlacementSnapshot] - // regionPlacementScanHook (test seam, nil in production) runs at the - // end of every aggregate scan, full or delta, after its rows were - // folded in; a non-nil error fails that scan the way an error part-way - // through the rows does. - regionPlacementScanHook func(full bool) error + regionHeard regionIndex + regionPlaced regionIndex // identityNamesQueryHook (#410, test seam, nil in production) runs once // per real hidden-name lookup query, right before it executes, with the // lower-case keys it looks up. @@ -1282,10 +1281,10 @@ func (db *DB) GetNodes(limit, offset int, role, search, before, lastHeard, sortB } // The region filter is served from the bounded membership cache in - // nodes_region_cache.go: the IN (...) observation join is evaluated there - // once per region set, not twice per request (COUNT(*) and the page). - // #38's from_pubkey contract — and the trap that comes with it — lives - // with the query, in scanNodeRegionKeys. + // nodes_region_cache.go, cut from the per-IATA region index + // (region_index.go) rather than evaluated twice per request (COUNT(*) + // and the page). #38's from_pubkey contract — and the trap that comes + // with it — lives with the query, in scanHeardRegions. if codes := normalizeRegionCodes(region); len(codes) > 0 { keysJSON, err := db.nodeRegionKeysJSON(codes) if err != nil { diff --git a/cmd/server/main.go b/cmd/server/main.go index 6dc8cdf2c..9f211898b 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -413,6 +413,12 @@ func main() { stopEviction := store.StartEvictionTicker() defer stopEviction() + // #618: build the per-IATA region index for /api/nodes?region= in the + // background and keep it refreshed, so no request pays the full history + // scan. It reads SQLite only, never the PacketStore or its lock. + stopRegionIndex := database.StartRegionIndexWarmer() + defer stopRegionIndex() + // Steady-state analytics recomputers (issue #1240). Replaces the // on-request compute-then-cache pattern for the default (region="", // zero-window) analytics queries with a background refresh loop so @@ -594,6 +600,9 @@ func main() { if stopAnalyticsRecomp != nil { stopAnalyticsRecomp() } + // 1d. Stop the region index warmer (#618) for the same reason: it + // cancels a scan in flight and returns once the goroutine has. + stopRegionIndex() // 2. Gracefully drain HTTP connections (up to shutdownDrainTimeout) ctx, cancel := context.WithTimeout(context.Background(), shutdownDrainTimeout) diff --git a/cmd/server/nodes_region_cache.go b/cmd/server/nodes_region_cache.go index 13085254a..c61e26be0 100644 --- a/cmd/server/nodes_region_cache.go +++ b/cmd/server/nodes_region_cache.go @@ -23,18 +23,27 @@ import ( // 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: +// Since #618 an entry is cut from the global heard index (region_index.go): +// one shared scan of the observation history grouped by observer IATA, so a +// region set is the union of its codes' member lists, not a full scan of its +// own. Entries keep their ages and the stale-while-revalidate contract: // - 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. +// - stale: served as-is while ONE background refresh re-cuts it from the +// index, refreshing the index first — a delta scan of observations with +// id > watermark (a rowid range, milliseconds); +// - every nodeRegionRebuildInterval the index 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). +// An entry carries the ages of the snapshot it was cut from, so those are +// the ages of the data. A request that has to wait for an entry (a new +// region set, or one past nodeRegionMaxStale) is cut from any snapshot +// younger than nodeRegionMaxStale at once. Only a cold start waits — for +// the index's first build, which the warmer already has in flight, or the +// set's own per-set build if that index build fails — and concurrent +// requests share it (the #1910 singleflight pattern, as channelsSF). +// StartRegionIndexWarmer runs the index build right after startup and keeps +// the index refreshed. // // Two bounds make that safe to expose to a query parameter (AGENTS.md rule 0, // "no unbounded data structures"): @@ -45,6 +54,10 @@ import ( // - an entry older than nodeRegionMaxStale is not served blind: the caller // waits for its refresh, so the ages promised above are the ages // actually served even when refreshes keep failing. +// +// The index itself is bounded by the distinct (node, region) pairs in the +// retained history: a few thousand nodes times the regions each was heard +// in, never by the number of hearings. const ( nodeRegionFreshTTL = 30 * time.Second @@ -63,11 +76,11 @@ const ( // nodeRegionMaxStale bounds the age of a served entry. Below it a stale // entry is served while one refresh runs in the background — the fast // path every page uses. At or above it the caller waits for the refresh, - // which is a full rebuild at this age, so the documented bounds hold: - // without the wait a refresh that keeps failing serves membership of - // unbounded age with only a log line to show it. The cost is that the - // first region request after nodeRegionMaxStale with no region traffic - // pays one full scan, exactly as a cold start does. + // so the documented bounds hold: without the wait a refresh that keeps + // failing serves membership of unbounded age with only a log line to + // show it. The wait is a re-cut from the index, which the warmer keeps + // younger than this; only an index that is itself this old (its + // refreshes failing) makes the caller wait for a scan. nodeRegionMaxStale = nodeRegionRebuildInterval ) @@ -78,23 +91,19 @@ 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 + keys []string // sorted from_pubkeys that are members of 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 + built time.Time // the index's 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) { +func newNodeRegionEntry(set map[string]struct{}, watermark int64, refreshed, built time.Time) (*nodeRegionEntry, error) { keys := make([]string, 0, len(set)) for k := range set { keys = append(keys, k) @@ -108,7 +117,7 @@ func newNodeRegionEntry(set map[string]struct{}, watermark int64, built time.Tim keys: keys, keysJSON: string(buf), watermark: watermark, - refreshed: time.Now(), + refreshed: refreshed, built: built, }, nil } @@ -210,16 +219,16 @@ func (db *DB) nodeRegionKeysJSON(codes []string) (string, error) { // 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 db.refreshNodeRegion(key, codes, true) }) return e.keysJSON, nil } // Past nodeRegionMaxStale the entry is too old to serve blind, so - // wait for the refresh (a full rebuild at this age). A refresh that - // fails still falls back to the stale entry: an outage of the - // observation scan must not turn /api/nodes?region= into a 500. + // wait for a re-cut from the index. A refresh that fails still falls + // back to the stale entry: an outage of the observation scan must + // not turn /api/nodes?region= into a 500. v, err, _ := db.nodeRegionSF.Do(key, func() (any, error) { - return db.refreshNodeRegion(key, codes) + return db.refreshNodeRegion(key, codes, false) }) if err != nil { log.Printf("[nodes-region] %s: serving entry %v old, refresh failed: %v", @@ -233,7 +242,7 @@ func (db *DB) nodeRegionKeysJSON(codes []string) (string, error) { } v, err, _ := db.nodeRegionSF.Do(key, func() (any, error) { - return db.refreshNodeRegion(key, codes) + return db.refreshNodeRegion(key, codes, false) }) if err != nil { return "", err @@ -245,12 +254,17 @@ func (db *DB) nodeRegionKeysJSON(codes []string) (string, error) { 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) { +// refreshNodeRegion re-cuts the cache entry for key from the active region +// index and stores it. With waitIndex (a background refresh, which nobody +// waits on) a stale index is refreshed first; without it (a caller is +// waiting) any snapshot younger than nodeRegionMaxStale is cut at once and +// the index refresh runs in the background (currentRegionIndex), and before +// the first snapshot exists it waits for the index build, falling back to +// the set's own build only if that fails (coldNodeRegion). 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, waitIndex bool) (*nodeRegionEntry, error) { prev := db.getNodeRegionEntry(key) if prev != nil && prev.fresh() { return prev, nil // a flight that finished just before this one got here @@ -266,17 +280,16 @@ func (db *DB) refreshNodeRegion(key string, codes []string) (*nodeRegionEntry, e ctx, cancel := context.WithTimeout(context.Background(), nodeRegionRebuildInterval) defer cancel() + ix, spec := db.activeRegionIndex() var e *nodeRegionEntry var err error - switch { - case db.regionPlacement.Mode == regionModePlaced: - // #488 M2: cut from the global placement aggregate, which keeps its - // own watermark and rebuild clock (region_placement.go). - e, err = db.placedNodeRegion(ctx, codes) - case prev == nil || prev.rebuildDue(): - e, err = db.buildNodeRegion(ctx, key, codes) - default: - e, err = db.extendNodeRegion(ctx, key, codes, prev) + if !waitIndex && spec.perSet != nil && ix.snap.Load() == nil { + e, err = db.coldNodeRegion(ctx, ix, spec, codes) + } else { + var s *regionMembershipSnapshot + if s, err = db.currentRegionIndex(ctx, ix, spec, waitIndex); err == nil { + e, err = s.entry(codes) + } } if err != nil { err = fmt.Errorf("nodes-region %s: %w", key, err) @@ -287,60 +300,6 @@ func (db *DB) refreshNodeRegion(key string, codes []string) (*nodeRegionEntry, 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 @@ -357,91 +316,70 @@ func (db *DB) maxObservationID(ctx context.Context) (int64, error) { return wm, nil } +// buildNodeRegionCold is the per-set build a request falls back to when the +// region index has no snapshot and its build failed (coldNodeRegion, which +// serialises these builds under nodeRegionColdMu): the full scan every cold +// region set ran before #618. +func (db *DB) buildNodeRegionCold(ctx context.Context, codes []string) (*nodeRegionEntry, error) { + if db.nodeRegionPerSetHook != nil { + db.nodeRegionPerSetHook() + } + 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, wm, set); err != nil { + return nil, err + } + log.Printf("[nodes-region] %s: per-set build, the index build failed, %d nodes in %v (obs id <= %d)", + nodeRegionKey(codes), len(set), time.Since(start).Round(time.Millisecond), wm) + return newNodeRegionEntry(set, wm, time.Now(), start) +} + // 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) { +// observer in codes with 0 < observations.id <= hi: the heard rule for one +// region set, as master built it, left to the planner, which can drive it +// from the set's own observers. Keep both id bounds: without the lower one +// the planner walks observations by rowid instead (2.5x slower on the #176 +// perf database). Only fixed fragments and "?" placeholders are formatted +// in; every value is bound. scanHeardRegions carries the from_pubkey +// contract. +func (db *DB) scanNodeRegionKeys(ctx context.Context, codes []string, hi int64, set map[string]struct{}) 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) - + args = append(args, int64(0), 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 / PR #38: from_pubkey is a dedicated column the ingestor fills at - // write time for ADVERT rows, so this lookup does not have to - // JSON_EXTRACT and parse decoded_json for every candidate row. Measured - // against a live 4.9GB database, 9 interleaved rounds so concurrent - // ingest cannot bias one variant: 934ms -> 835ms median (1.12x) for a - // single-region filter, identical result set both ways. The win is purely - // the avoided JSON parse — EXPLAIN QUERY PLAN is byte-identical before - // and after, both driving off idx_transmissions_payload_type. Despite the - // column being indexed, idx_transmissions_from_pubkey does not - // participate in this plan at all. - // - // THE TRAP, for whoever touches this next: from_pubkey is only written - // for ADVERTs that actually carry a pubkey — see the guard in - // cmd/ingestor/db.go. A NULL here drops out of IN (...) silently, so a - // write path that fills decoded_json but forgets from_pubkey would cost - // nodes from this filter with no error to notice. That is safe today only - // because it is measured to be: on live staging 25 of 62104 ADVERTs have - // NULL from_pubkey, and all 25 have no pubKey in decoded_json either — - // nothing is lost, because there was never a pubkey to record. A COALESCE - // fallback to JSON_EXTRACT was measured and rejected: it rescued 0 rows, - // and it parses decoded_json again for every NULL row, so one corrupt - // ADVERT that the backfill has not reached yet fails the whole query with - // "malformed JSON" (a 500 on /api/nodes?region=). Behind this cache that - // is worse, not better: the failed refresh leaves the cached entry stale - // for every later request too. Its speed is not the reason: on a 4.3GB - // synthetic DB it times within noise of from_pubkey. If you add a new - // write path for transmissions, populate from_pubkey there rather than - // reintroducing a per-row JSON parse here. nodes_region_from_pubkey_test.go - // locks the result set against the JSON_EXTRACT expression; - // TestBackfillFromPubkey_* in cmd/ingestor locks the backfill's half of - // the contract. 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 + JOIN transmissions t ON t.id = o.transmission_id + JOIN observers obs ON %[1]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, ",")) - + AND UPPER(TRIM(obs.iata)) IN (%[2]s) + AND o.id > ? AND o.id <= ?`, joinCond, strings.Join(placeholders, ",")) rows, err := db.conn.QueryContext(ctx, q, args...) if err != nil { - return 0, fmt.Errorf("scan region keys: %w", err) + return 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++ + return fmt.Errorf("scan region keys: %w", err) } + set[pk] = struct{}{} } if err := rows.Err(); err != nil { - return added, fmt.Errorf("scan region keys: %w", err) + return fmt.Errorf("scan region keys: %w", err) } - return added, nil + return nil } diff --git a/cmd/server/nodes_region_cache_176_test.go b/cmd/server/nodes_region_cache_176_test.go index 494ecf381..23082e54e 100644 --- a/cmd/server/nodes_region_cache_176_test.go +++ b/cmd/server/nodes_region_cache_176_test.go @@ -24,7 +24,8 @@ import ( // ageNodeRegionEntry rewrites the cached entry for key with its refresh and // build times pushed back, so a test can stand where a 30-second-old or // 30-minute-old entry stands without waiting. The entry is immutable once -// stored, so this stores a copy. +// stored, so this stores a copy. Since #618 an entry is a cut of the active +// region index, whose ages are the data's, so the index is aged with it. func ageNodeRegionEntry(t *testing.T, db *DB, key string, refreshedAgo, builtAgo time.Duration) *nodeRegionEntry { t.Helper() prev := db.getNodeRegionEntry(key) @@ -37,9 +38,28 @@ func ageNodeRegionEntry(t *testing.T, db *DB, key string, refreshedAgo, builtAgo aged.built = time.Now().Add(-builtAgo) } db.setNodeRegionEntry(key, &aged) + ix, _ := db.activeRegionIndex() + ageRegionIndex(ix, refreshedAgo, builtAgo) return &aged } +// ageRegionIndex pushes ix's snapshot refresh back by refreshedAgo and, when +// builtAgo > 0, its last full build back by builtAgo (making a rebuild due +// once that reaches nodeRegionRebuildInterval). The snapshot is immutable, +// so this stores a copy. +func ageRegionIndex(ix *regionIndex, refreshedAgo, builtAgo time.Duration) { + if s := ix.snap.Load(); s != nil { + c := *s + c.refreshed = time.Now().Add(-refreshedAgo) + ix.snap.Store(&c) + } + if builtAgo > 0 { + ix.mu.Lock() + ix.built = time.Now().Add(-builtAgo) + ix.mu.Unlock() + } +} + // waitForNodeRegionRefresh waits for the background refresh kicked by a // stale read to replace the entry, and fails if it never does. func waitForNodeRegionRefresh(t *testing.T, db *DB, key string, was time.Time) { diff --git a/cmd/server/nodes_region_cache_port_test.go b/cmd/server/nodes_region_cache_port_test.go index 13fdb2f4d..b1c1f5504 100644 --- a/cmd/server/nodes_region_cache_port_test.go +++ b/cmd/server/nodes_region_cache_port_test.go @@ -90,7 +90,7 @@ func TestNodeRegionCacheDeltaFollowsFromPubkeyContract_176(t *testing.T) { } ageNodeRegionEntry(t, db, "SJC", 2*nodeRegionFreshTTL, 0) - if _, err := db.refreshNodeRegion("SJC", []string{"SJC"}); err != nil { + if _, err := db.refreshNodeRegion("SJC", []string{"SJC"}, true); err != nil { t.Fatalf("delta refresh over a corrupt ADVERT: %v", err) } keys := db.getNodeRegionEntry("SJC").keysJSON @@ -117,7 +117,8 @@ func TestNodeRegionCachePortFullRebuildDropsPrunedRegion(t *testing.T) { aged.refreshed = time.Now().Add(-2 * nodeRegionFreshTTL) aged.built = time.Now().Add(-2 * nodeRegionRebuildInterval) db.setNodeRegionEntry("SJC", &aged) - if _, err := db.refreshNodeRegion("SJC", []string{"SJC"}); err != nil { + ageRegionIndex(&db.regionHeard, 2*nodeRegionFreshTTL, 2*nodeRegionRebuildInterval) + if _, err := db.refreshNodeRegion("SJC", []string{"SJC"}, true); err != nil { t.Fatal(err) } nodes, total, _, err := db.GetNodes(10, 0, "", "", "", "", "", "SJC") diff --git a/cmd/server/nodes_region_index_618_test.go b/cmd/server/nodes_region_index_618_test.go new file mode 100644 index 000000000..92a473a06 --- /dev/null +++ b/cmd/server/nodes_region_index_618_test.go @@ -0,0 +1,1165 @@ +package main + +// #618: /api/nodes?region= is served from ONE per-IATA index built by one +// shared scan (region_index.go), so a region set is a union of per-code key +// sets instead of a full history scan of its own. +// +// - equivalence: for every region set the union equals the per-set build +// it replaces (idx618PerSetKeys, the query master ran per set), on a +// generated database with awkward observer regions, on the committed +// fixture, and (env-gated) on a large generated database; +// - one full scan serves every set, cold or warm; the delta and the +// periodic rebuild keep their #176 semantics; +// - the background warmer builds the index before the first request, +// keeps it from ageing past nodeRegionMaxStale, and needs no PacketStore; +// - a cold request (no snapshot yet) waits for the index build in +// flight and runs its own per-set build only if that build fails; +// identical sets share one refresh, per-set fallbacks are serialised +// and give way to an index published meanwhile, and a request never +// queues behind a running rebuild when a usable snapshot exists; +// - an idle refresh re-stamps the snapshot without scanning; +// - memory grows with distinct (node, region) pairs only. + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "reflect" + "slices" + "sort" + "strings" + "sync" + "sync/atomic" + "syscall" + "testing" + "time" +) + +// idx618PerSetKeys is the oracle: the full per-set build master ran for every +// region set (scanNodeRegionKeys with lo = 0, hi = MAX(observations.id)), +// kept here verbatim so the index is compared with the code it replaces. +func idx618PerSetKeys(t testing.TB, db *DB, codes []string) []string { + t.Helper() + wm, err := db.maxObservationID(context.Background()) + if err != nil { + t.Fatal(err) + } + placeholders := make([]string, len(codes)) + args := []any{payloadTypeAdvert} + for i, c := range codes { + placeholders[i] = "?" + args = append(args, c) + } + args = append(args, int64(0), wm) + joinCond := "obs.rowid = o.observer_idx" + if !db.isV3() { + joinCond = "obs.id = o.observer_id" + } + q := fmt.Sprintf(`SELECT DISTINCT t.from_pubkey + FROM observations o + JOIN transmissions t ON t.id = o.transmission_id + JOIN observers obs ON %[1]s + WHERE t.payload_type = ? + AND t.from_pubkey IS NOT NULL + AND UPPER(TRIM(obs.iata)) IN (%[2]s) + AND o.id > ? AND o.id <= ?`, joinCond, strings.Join(placeholders, ",")) + rows, err := db.conn.Query(q, args...) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + 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 +} + +// idx618Keys is what GetNodes binds for a raw ?region= value. +func idx618Keys(t testing.TB, db *DB, region string) []string { + t.Helper() + js, err := db.nodeRegionKeysJSON(normalizeRegionCodes(region)) + if err != nil { + t.Fatalf("region %q: %v", region, err) + } + keys := []string{} + if err := json.Unmarshal([]byte(js), &keys); err != nil { + t.Fatal(err) + } + if !sort.StringsAreSorted(keys) { + t.Fatalf("region %q: keys not sorted: %v", region, keys) + } + return keys +} + +// idx618Subsets returns every non-empty subset of codes as a ?region= value +// when there are few codes, else every single code, every pair and all of +// them: singles, multi-code sets and quick-pick groups alike. +func idx618Subsets(codes []string) []string { + var out []string + if len(codes) <= 6 { + for mask := 1; mask < 1< 0 { + nonEmpty++ + } + } + return nonEmpty +} + +// idx618CountScans counts the heard index's full and delta scans. +func idx618CountScans(db *DB) (full, delta *atomic.Int32) { + full, delta = &atomic.Int32{}, &atomic.Int32{} + db.regionHeard.scanHook = func(isFull bool) error { + if isFull { + full.Add(1) + } else { + delta.Add(1) + } + return nil + } + return full, delta +} + +// idx618AgeIndex pushes the heard index's snapshot refresh (and, when +// builtAgo > 0, its last full build) back, as ageNodeRegionEntry does for an +// entry. +func idx618AgeIndex(db *DB, refreshedAgo, builtAgo time.Duration) { + ageRegionIndex(&db.regionHeard, refreshedAgo, builtAgo) +} + +// idx618SeedAwkward adds what a real observers table holds besides clean +// codes: a lower-case padded code that must fold into AAA, a code only it +// carries, NULL and blank regions that belong to no set, plus an ADVERT +// without from_pubkey and a non-ADVERT, which no set may contain. +func idx618SeedAwkward(t *testing.T, db *DB, now int64) { + t.Helper() + exec := func(q string, args ...any) { + t.Helper() + if _, err := db.conn.Exec(q, args...); err != nil { + t.Fatalf("%s: %v", q, err) + } + } + for _, o := range []struct { + rowid int + iata any + }{{100, " aaa "}, {101, "eee"}, {102, nil}, {103, " "}} { + exec(`INSERT INTO observers (rowid, id, name, iata, last_seen, first_seen) + VALUES (?, ?, ?, ?, '2026-10-01T00:00:00Z', '2026-01-01T00:00:00Z')`, + o.rowid, fmt.Sprintf("obs-x%d", o.rowid), fmt.Sprintf("Obs x%d", o.rowid), o.iata) + } + s := &rpm2Seeder{t: t, exec: db.conn, seq: 1 << 20} + for i, obs := range []int{100, 101, 102, 103} { + pk := fmt.Sprintf("ee%02x000000000000", i) + s.directAdvert(pk, now-int64(i)*60, obs) + } + // An ADVERT heard in AAA whose from_pubkey was never stamped (the test + // trigger needs a pubKey to stamp it, and there is none). + exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) + VALUES ('1200', 'idx618-nopk', '2026-10-01T00:00:00Z', 2, 4, '{}')`) + exec(`INSERT INTO observations (transmission_id, observer_idx, timestamp) + VALUES ((SELECT id FROM transmissions WHERE hash = 'idx618-nopk'), 1, ?)`, now) + // A GRP_TXT from a pubkey-shaped sender, heard in BBB. + exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json, from_pubkey) + VALUES ('1500', 'idx618-grp', '2026-10-01T00:00:00Z', 1, 5, '{}', 'ff00000000000000')`) + exec(`INSERT INTO observations (transmission_id, observer_idx, timestamp) + VALUES ((SELECT id FROM transmissions WHERE hash = 'idx618-grp'), 4, ?)`, now) +} + +var idx618Codes = []string{"AAA", "BBB", "CCC", "DDD", "EEE", "ZZZ"} + +// idx618RefreshAll takes every set's entry (the ones the LRU still holds) +// and the index past the fresh TTL and runs each set's refresh +// synchronously, as its background refresh would. +func idx618RefreshAll(t *testing.T, db *DB, regions []string) { + t.Helper() + for _, region := range regions { + codes := normalizeRegionCodes(region) + key := nodeRegionKey(codes) + if e := db.getNodeRegionEntry(key); e != nil { + c := *e + c.refreshed = time.Now().Add(-2 * nodeRegionFreshTTL) + db.setNodeRegionEntry(key, &c) + } + idx618AgeIndex(db, 2*nodeRegionFreshTTL, 0) + if _, err := db.refreshNodeRegion(key, codes, true); err != nil { + t.Fatal(err) + } + } +} + +// TestNodeRegionIndex618_UnionEqualsPerSetBuild: property test over every +// subset of the codes (63 sets, including one no observer carries), cold, +// after new adverts (delta) and after a rebuild that has to drop a region. +// All 63 sets are served from ONE full scan. +func TestNodeRegionIndex618_UnionEqualsPerSetBuild(t *testing.T) { + db := rpm2EmptyTestDB(t) + rpm2SeedScenario(t, db.conn, 2, rpm2T0) + idx618SeedAwkward(t, db, rpm2T0) + full, delta := idx618CountScans(db) + regions := idx618Subsets(idx618Codes) + // Spellings the normaliser folds into sets above. + regions = append(regions, " aaa , Bbb ", "ccc,CCC,ccc", "eee,all", "zzz") + + if n := idx618AssertEquivalent(t, db, "cold", regions); n < len(regions)/2 { + t.Fatalf("cold: only %d of %d sets are non-empty; the comparison is too weak", n, len(regions)) + } + // The first sets may have been served by their own per-set build while + // the index built; wait for it, then every set is a cut of it. + waitForHeardIndex(t, db, rpm2MaxID(t, db)) + idx618AssertEquivalent(t, db, "warm", regions) + if full.Load() != 1 || delta.Load() != 0 { + t.Fatalf("cold: %d sets cost %d full and %d delta scans, want 1 full", len(regions), full.Load(), delta.Load()) + } + + // New adverts arrive: a new node in DDD and an existing CCC node first + // heard in EEE. Every entry and the index go stale; each set's refresh + // (run synchronously, as its background refresh would) extends. + s := &rpm2Seeder{t: t, exec: db.conn, seq: 1 << 21} + s.directAdvert("dd99000000000001", rpm2T0, s.obs(3, 0)) + s.directAdvert("0201000000000000", rpm2T0, 101) + idx618AgeIndex(db, 2*nodeRegionFreshTTL, 0) + idx618RefreshAll(t, db, regions) + idx618AssertEquivalent(t, db, "after delta", regions) + if full.Load() != 1 || delta.Load() != 1 { + t.Fatalf("delta: %d full and %d delta scans, want 1 and 1", full.Load(), delta.Load()) + } + + // Observer edits below the watermark: only the rebuild sees them. + if _, err := db.conn.Exec(`UPDATE observers SET iata = 'ccc' WHERE rowid = 100`); err != nil { + t.Fatal(err) + } + idx618AgeIndex(db, 2*nodeRegionFreshTTL, 2*nodeRegionRebuildInterval) + idx618RefreshAll(t, db, regions) + idx618AssertEquivalent(t, db, "after rebuild", regions) + if full.Load() != 2 { + t.Fatalf("rebuild: %d full scans, want 2", full.Load()) + } +} + +// TestNodeRegionIndex618_OneScanServesEverySet_Fixture runs on the committed +// e2e fixture through GetNodes only, and counts the "full build" log lines, +// so it states the issue's acceptance in terms master can be measured by: +// every subset of the fixture's regions equals the per-set build, and all of +// them together cost one full scan (master: one per set). +func TestNodeRegionIndex618_OneScanServesEverySet_Fixture(t *testing.T) { + path := rpm2CopyFile(t, filepath.Join("..", "..", "test-fixtures", "e2e-fixture.db")) + db, err := OpenDB(path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + var codes []string + rows, err := db.conn.Query(`SELECT DISTINCT UPPER(TRIM(iata)) FROM observers + WHERE iata IS NOT NULL AND TRIM(iata) != '' ORDER BY 1`) + if err != nil { + t.Fatal(err) + } + for rows.Next() { + var c string + if err := rows.Scan(&c); err != nil { + t.Fatal(err) + } + codes = append(codes, c) + } + rows.Close() + if len(codes) < 2 { + t.Fatalf("fixture has %d regions; need at least 2", len(codes)) + } + regions := idx618Subsets(append(codes, "ZZZ")) + + capture := startLogCapture() + defer capture.restoreWriter() + nonEmpty := 0 + for _, region := range regions { + want := idx618PerSetKeys(t, db, normalizeRegionCodes(region)) + nodes, total, _, err := db.GetNodes(10000, 0, "", "", "", "", "", region) + if err != nil { + t.Fatal(err) + } + inNodes := map[string]bool{} + for _, n := range nodes { + pk, _ := n["public_key"].(string) + inNodes[pk] = true + } + got := idx618Keys(t, db, region) + if !slices.Equal(got, want) { + t.Errorf("region %q: union %v, per-set build %v", region, got, want) + } + // GetNodes lists exactly the keys that are in the nodes table. + listed := 0 + for _, pk := range want { + if inNodes[pk] { + listed++ + } + } + if total != listed || len(nodes) != listed { + t.Errorf("region %q: GetNodes total %d, %d rows; %d of the set's keys are nodes", region, total, len(nodes), listed) + } + if len(want) > 0 { + nonEmpty++ + } + } + waitForHeardIndex(t, db, rpm2MaxID(t, db)) + capture.restoreWriter() + builds := 0 + for _, line := range strings.Split(capture.read(), "\n") { + if strings.Contains(line, "[nodes-region]") && strings.Contains(line, "full build") { + builds++ + } + } + t.Logf("fixture: %d regions, %d sets (%d non-empty), %d full builds", len(codes), len(regions), nonEmpty, builds) + if nonEmpty < len(regions)/2 { + t.Fatalf("only %d of %d sets are non-empty; the comparison is too weak", nonEmpty, len(regions)) + } + if builds != 1 { + t.Fatalf("%d region sets cost %d full builds, want 1 shared scan", len(regions), builds) + } +} + +// TestNodeRegionIndex618_DeltaEqualsFullBuild: the heard scan folded in two +// parts at any split point equals one full scan, pair for pair. +func TestNodeRegionIndex618_DeltaEqualsFullBuild(t *testing.T) { + db := rpm2EmptyTestDB(t) + rpm2SeedScenario(t, db.conn, 1, rpm2T0) + idx618SeedAwkward(t, db, rpm2T0) + ctx := context.Background() + wm := rpm2MaxID(t, db) + full := newRegionEvidenceAgg() + if _, err := db.scanHeardRegions(ctx, 0, wm, false, full); err != nil { + t.Fatal(err) + } + if full.pairs == 0 || full.watermark != wm { + t.Fatalf("full scan: %d pairs, watermark %d/%d", full.pairs, full.watermark, wm) + } + for _, mid := range []int64{1, wm / 3, wm / 2, wm - 1} { + inc := newRegionEvidenceAgg() + if _, err := db.scanHeardRegions(ctx, 0, mid, false, inc); err != nil { + t.Fatal(err) + } + d := newRegionEvidenceAgg() + if _, err := db.scanHeardRegions(ctx, mid, wm, true, d); err != nil { + t.Fatal(err) + } + inc.merge(d) + if !reflect.DeepEqual(inc.nodes, full.nodes) || inc.pairs != full.pairs || inc.watermark != wm { + t.Fatalf("split at %d: %d pairs (watermark %d), full build %d pairs", mid, inc.pairs, inc.watermark, full.pairs) + } + } +} + +// waitForHeardIndex waits until the heard index has published a snapshot +// with watermark >= wm. +func waitForHeardIndex(t *testing.T, db *DB, wm int64) *regionMembershipSnapshot { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + if s := db.regionHeard.snap.Load(); s != nil && s.watermark >= wm { + return s + } + time.Sleep(2 * time.Millisecond) + } + t.Fatalf("heard index never reached watermark %d", wm) + return nil +} + +// TestNodeRegionIndex618_WarmupServesFirstRequests: after the warmer's build +// the first requests for any set — single, multi, a quick-pick group — scan +// nothing. The DB has no PacketStore at all, so the warmup cannot depend on +// it (or on its lock). +func TestNodeRegionIndex618_WarmupServesFirstRequests(t *testing.T) { + db := rpm2EmptyTestDB(t) + rpm2SeedScenario(t, db.conn, 1, rpm2T0) + full, delta := idx618CountScans(db) + stop := db.startRegionIndexWarmer(time.Hour) + defer stop() + waitForHeardIndex(t, db, rpm2MaxID(t, db)) + if full.Load() != 1 { + t.Fatalf("warmup: %d full scans, want 1", full.Load()) + } + for _, region := range []string{"AAA", "BBB,CCC", "AAA,BBB,CCC,DDD", "ddd"} { + start := time.Now() + got := idx618Keys(t, db, region) + took := time.Since(start) + if want := idx618PerSetKeys(t, db, normalizeRegionCodes(region)); !slices.Equal(got, want) { + t.Errorf("region %q after warmup: %v, want %v", region, got, want) + } + t.Logf("first request for %q after warmup: %v", region, took) + } + if full.Load() != 1 || delta.Load() != 0 { + t.Fatalf("first requests after warmup scanned: %d full, %d delta", full.Load(), delta.Load()) + } + stop() + stop() // idempotent +} + +// TestNodeRegionIndex618_WarmerKeepsIndexCurrent: with no requests at all +// the warmer keeps refreshing — new adverts land in the snapshot and a due +// rebuild runs in the background — so the index never ages past +// nodeRegionMaxStale and no request ever pays the periodic full scan. +func TestNodeRegionIndex618_WarmerKeepsIndexCurrent(t *testing.T) { + db := rpm2EmptyTestDB(t) + rpm2SeedScenario(t, db.conn, 1, rpm2T0) + full, delta := idx618CountScans(db) + stop := db.startRegionIndexWarmer(5 * time.Millisecond) + defer stop() + waitForHeardIndex(t, db, rpm2MaxID(t, db)) + + s := &rpm2Seeder{t: t, exec: db.conn, seq: 1 << 21} + s.directAdvert("dd99000000000001", rpm2T0, s.obs(3, 0)) + snap := waitForHeardIndex(t, db, rpm2MaxID(t, db)) + if !slices.Contains(snap.byRegion["DDD"], "dd99000000000001") { + t.Fatalf("warmer refresh missed a new advert: DDD = %v", snap.byRegion["DDD"]) + } + if delta.Load() == 0 { + t.Fatal("the warmer picked up a new advert without a delta scan") + } + + before := full.Load() + idx618AgeIndex(db, 0, 2*nodeRegionRebuildInterval) + deadline := time.Now().Add(5 * time.Second) + for full.Load() == before && time.Now().Before(deadline) { + time.Sleep(2 * time.Millisecond) + } + if full.Load() == before { + t.Fatal("a due rebuild never ran in the warmer") + } +} + +// TestNodeRegionIndex618_WarmerStopDoesNotWaitForForeignFlight: a warmer +// that joined an index refresh some request started (whose context it +// cannot cancel) stops at once instead of waiting for that whole scan — +// on a large database a minute, longer than the shutdown is given. +func TestNodeRegionIndex618_WarmerStopDoesNotWaitForForeignFlight(t *testing.T) { + db := rpm2EmptyTestDB(t) + rpm2SeedScenario(t, db.conn, 1, rpm2T0) + entered, release := make(chan struct{}), make(chan struct{}) + defer close(release) + var once sync.Once + db.regionHeard.scanHook = func(bool) error { + once.Do(func() { close(entered) }) + <-release + return nil + } + go db.refreshNodeRegion("AAA", []string{"AAA"}, true) // the request's flight + <-entered + stop := db.startRegionIndexWarmer(time.Hour) + time.Sleep(20 * time.Millisecond) // let the warmer join the flight + stopped := make(chan struct{}) + go func() { stop(); close(stopped) }() + select { + case <-stopped: + case <-time.After(2 * time.Second): + t.Fatal("warmer stop waited for a refresh it did not start") + } +} + +// idx618CountPerSet counts the per-set builds cold requests run. +func idx618CountPerSet(db *DB) *atomic.Int32 { + n := &atomic.Int32{} + db.nodeRegionPerSetHook = func() { n.Add(1) } + return n +} + +var errIdx618Scan = errors.New("idx618: injected scan failure") + +// TestNodeRegionIndex618_ColdRequestsDuringWarmup: requests that arrive +// while the warmup's full scan is still running (here: held) wait for that +// build and are served by it as soon as it lands — at most the rest of the +// warmup. None runs a per-set build of its own (it would compete with the +// one build that serves every set) or starts an index scan of its own, and +// identical sets share one refresh. +func TestNodeRegionIndex618_ColdRequestsDuringWarmup(t *testing.T) { + db := rpm2EmptyTestDB(t) + rpm2SeedScenario(t, db.conn, 1, rpm2T0) + entered, release := make(chan struct{}), make(chan struct{}) + var releaseOnce sync.Once + releaseAll := func() { releaseOnce.Do(func() { close(release) }) } + defer releaseAll() + var full, delta atomic.Int32 + var once sync.Once + db.regionHeard.scanHook = func(isFull bool) error { + if isFull { + full.Add(1) + } else { + delta.Add(1) + } + once.Do(func() { close(entered) }) + <-release + return nil + } + var refreshes atomic.Int32 + db.nodeRegionQueryHook = func() error { refreshes.Add(1); return nil } + perSet := idx618CountPerSet(db) + stop := db.startRegionIndexWarmer(time.Hour) + defer stop() + <-entered + + sets := []string{"AAA", "AAA", "aaa", "BBB,CCC", "CCC,BBB", "DDD"} + got := make([][]string, len(sets)) + var served atomic.Int32 + var wg sync.WaitGroup + for i, region := range sets { + wg.Add(1) + go func() { + defer wg.Done() + got[i] = idx618Keys(t, db, region) + served.Add(1) + }() + } + time.Sleep(100 * time.Millisecond) + if n := served.Load(); n != 0 { + t.Errorf("%d cold requests were served before the warmup's build landed (%d per-set builds)", n, perSet.Load()) + } + releaseAll() + done := make(chan struct{}) + go func() { wg.Wait(); close(done) }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("cold requests were not served once the warmup's build landed") + } + for i, region := range sets { + if want := idx618PerSetKeys(t, db, normalizeRegionCodes(region)); !slices.Equal(got[i], want) { + t.Errorf("%q during warmup: %v, want %v", region, got[i], want) + } + } + if n := refreshes.Load(); n != 3 { + t.Errorf("%d refreshes for 3 distinct sets; identical sets must share one", n) + } + if n := perSet.Load(); n != 0 { + t.Errorf("%d per-set builds during warmup, want 0: cold requests wait for the index build", n) + } + if full.Load() != 1 || delta.Load() != 0 { + t.Fatalf("during warmup: %d full and %d delta index scans, want only the warmup's", full.Load(), delta.Load()) + } + if got, want := idx618Keys(t, db, "BBB,DDD"), idx618PerSetKeys(t, db, []string{"BBB", "DDD"}); !slices.Equal(got, want) { + t.Errorf("BBB,DDD after warmup: %v, want %v", got, want) + } + if full.Load() != 1 || delta.Load() != 0 || perSet.Load() != 0 { + t.Fatalf("a new set after warmup scanned: %d full, %d delta, %d per-set", full.Load(), delta.Load(), perSet.Load()) + } +} + +// TestNodeRegionIndex618_ColdRequestsShareOneIndexBuild: with no warmer, +// concurrent cold requests for different sets start one index build between +// them and are all cut from it; none needs a per-set build (held here, so +// one that did would never return). +func TestNodeRegionIndex618_ColdRequestsShareOneIndexBuild(t *testing.T) { + db := rpm2EmptyTestDB(t) + rpm2SeedScenario(t, db.conn, 1, rpm2T0) + full, delta := idx618CountScans(db) + db.nodeRegionColdMu.Lock() // no per-set build can run + var unlockOnce sync.Once + unlock := func() { unlockOnce.Do(db.nodeRegionColdMu.Unlock) } + defer unlock() + sets := []string{"AAA", "BBB,CCC", "DDD", "aaa,ddd"} + got := make([][]string, len(sets)) + var wg sync.WaitGroup + for i, region := range sets { + wg.Add(1) + go func() { + defer wg.Done() + got[i] = idx618Keys(t, db, region) + }() + } + served := make(chan struct{}) + go func() { wg.Wait(); close(served) }() + select { + case <-served: + case <-time.After(5 * time.Second): + unlock() + t.Fatal("cold requests waited for per-set builds although the index was built") + } + unlock() + for i, region := range sets { + if want := idx618PerSetKeys(t, db, normalizeRegionCodes(region)); !slices.Equal(got[i], want) { + t.Errorf("%q: %v, want %v", region, got[i], want) + } + } + if full.Load() != 1 || delta.Load() != 0 { + t.Fatalf("%d concurrent cold sets: %d full and %d delta index scans, want 1 full", len(sets), full.Load(), delta.Load()) + } +} + +// TestNodeRegionIndex618_ColdRequestFallsBackWhenIndexFails: while the index +// build fails, a cold request is served by its own per-set build — the scan +// every cold request ran before #618 — instead of a 500. Once the index +// builds, new sets are cut from it again. +func TestNodeRegionIndex618_ColdRequestFallsBackWhenIndexFails(t *testing.T) { + db := rpm2EmptyTestDB(t) + rpm2SeedScenario(t, db.conn, 1, rpm2T0) + perSet := idx618CountPerSet(db) + var failing atomic.Bool + failing.Store(true) + db.regionHeard.scanHook = func(bool) error { + if failing.Load() { + return errIdx618Scan + } + return nil + } + for _, region := range []string{"AAA", "BBB,CCC"} { + want := idx618PerSetKeys(t, db, normalizeRegionCodes(region)) + if got := idx618Keys(t, db, region); !slices.Equal(got, want) { + t.Errorf("%q with the index failing: %v, want %v", region, got, want) + } + } + if n := perSet.Load(); n != 2 { + t.Errorf("%d per-set builds for 2 cold sets with the index failing, want 2", n) + } + if db.regionHeard.snap.Load() != nil { + t.Fatal("a failing index build published a snapshot") + } + failing.Store(false) + if got, want := idx618Keys(t, db, "DDD"), idx618PerSetKeys(t, db, []string{"DDD"}); !slices.Equal(got, want) { + t.Errorf("DDD after the index recovered: %v, want %v", got, want) + } + if n := perSet.Load(); n != 2 { + t.Errorf("a cold set after the index recovered ran a per-set build (%d in all)", n) + } +} + +// TestNodeRegionIndex618_QueuedFallbackTakesRecoveredIndex: per-set +// fallbacks are serialised, so a burst of cold sets after an index failure +// queues. A fallback still queued when a retry (the warmer's next tick) +// has published the index is cut from it instead of running its full scan. +func TestNodeRegionIndex618_QueuedFallbackTakesRecoveredIndex(t *testing.T) { + db := rpm2EmptyTestDB(t) + rpm2SeedScenario(t, db.conn, 1, rpm2T0) + perSet := idx618CountPerSet(db) + var failing atomic.Bool + failing.Store(true) + var failed atomic.Int32 + db.regionHeard.scanHook = func(bool) error { + if failing.Load() { + failed.Add(1) + return errIdx618Scan + } + return nil + } + db.nodeRegionColdMu.Lock() // a fallback is already running + var unlockOnce sync.Once + unlock := func() { unlockOnce.Do(db.nodeRegionColdMu.Unlock) } + defer unlock() + sets := []string{"AAA", "BBB,CCC"} + got := make([][]string, len(sets)) + var wg sync.WaitGroup + for i, region := range sets { + wg.Add(1) + go func() { + defer wg.Done() + got[i] = idx618Keys(t, db, region) + }() + } + deadline := time.Now().Add(5 * time.Second) + for failed.Load() == 0 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if failed.Load() == 0 { + t.Fatal("the cold requests never tried the index build") + } + time.Sleep(50 * time.Millisecond) // let them queue for the fallback + failing.Store(false) + if _, err := db.regionIndexFlight(context.Background(), &db.regionHeard, db.heardRegionSpec()); err != nil { + t.Fatal(err) + } + unlock() + done := make(chan struct{}) + go func() { wg.Wait(); close(done) }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("queued cold requests were not served") + } + for i, region := range sets { + if want := idx618PerSetKeys(t, db, normalizeRegionCodes(region)); !slices.Equal(got[i], want) { + t.Errorf("%q: %v, want %v", region, got[i], want) + } + } + if n := perSet.Load(); n != 0 { + t.Errorf("%d queued fallbacks ran a per-set build after the index was published, want 0", n) + } +} + +// TestNodeRegionIndex618_ColdPerSetBuildsAreSerialised: the per-set +// fallbacks cold requests run while the index build fails take +// nodeRegionColdMu, as master's per-set builds took nodeRegionFullMu, so a +// burst of cold sets holds one pooled connection for a full scan (the pool +// is 4). +func TestNodeRegionIndex618_ColdPerSetBuildsAreSerialised(t *testing.T) { + db := rpm2EmptyTestDB(t) + rpm2SeedScenario(t, db.conn, 1, rpm2T0) + db.regionHeard.scanHook = func(bool) error { return errIdx618Scan } + db.nodeRegionColdMu.Lock() + var unlockOnce sync.Once + unlock := func() { unlockOnce.Do(db.nodeRegionColdMu.Unlock) } + defer unlock() + var done atomic.Int32 + var wg sync.WaitGroup + for _, region := range []string{"AAA", "BBB"} { + wg.Add(1) + go func() { + defer wg.Done() + idx618Keys(t, db, region) + done.Add(1) + }() + } + time.Sleep(100 * time.Millisecond) + if n := done.Load(); n != 0 { + t.Fatalf("%d cold requests were served by a per-set build that skipped nodeRegionColdMu", n) + } + unlock() + wg.Wait() +} + +// TestNodeRegionIndex618_IdleRefreshReStampsIndex: on a database with no new +// observations a refresh (one warmer tick) finds nothing past the watermark +// and scans nothing, but re-stamps the snapshot, so an idle index never ages +// past nodeRegionMaxStale and requests never start waiting on refreshes. +func TestNodeRegionIndex618_IdleRefreshReStampsIndex(t *testing.T) { + db := rpm2EmptyTestDB(t) + rpm2SeedScenario(t, db.conn, 1, rpm2T0) + full, delta := idx618CountScans(db) + stop := db.startRegionIndexWarmer(time.Hour) + defer stop() + wm := rpm2MaxID(t, db) + waitForHeardIndex(t, db, wm) + idx618AgeIndex(db, nodeRegionMaxStale, 0) + s, err := db.regionIndexFlight(context.Background(), &db.regionHeard, db.heardRegionSpec()) + if err != nil { + t.Fatal(err) + } + for what, snap := range map[string]*regionMembershipSnapshot{"returned": s, "published": db.regionHeard.snap.Load()} { + if age := time.Since(snap.refreshed); age >= nodeRegionFreshTTL { + t.Errorf("idle refresh: %s snapshot is %v old, want re-stamped", what, age.Round(time.Second)) + } + if snap.watermark != wm { + t.Errorf("idle refresh: %s snapshot watermark %d, want %d", what, snap.watermark, wm) + } + } + if full.Load() != 1 || delta.Load() != 0 { + t.Fatalf("idle refresh scanned: %d full (want the warmup's 1), %d delta", full.Load(), delta.Load()) + } +} + +// TestNodeRegionIndex618_RequestsDoNotQueueBehindRebuild: while a full +// rebuild runs, a new region set and an over-stale entry are cut from the +// existing (stale but usable) snapshot at once instead of waiting for it — +// that wait is the minute-long blank Nodes page the issue reports. +func TestNodeRegionIndex618_RequestsDoNotQueueBehindRebuild(t *testing.T) { + db := rpm2EmptyTestDB(t) + rpm2SeedScenario(t, db.conn, 1, rpm2T0) + wantAAA := idx618Keys(t, db, "AAA") + if len(wantAAA) == 0 { + t.Fatal("AAA is empty") + } + waitForHeardIndex(t, db, rpm2MaxID(t, db)) + entered, release := make(chan struct{}), make(chan struct{}) + var once sync.Once + db.regionHeard.scanHook = func(isFull bool) error { + if isFull { + once.Do(func() { close(entered) }) + <-release + } + return nil + } + defer func() { + select { + case <-release: + default: + close(release) + } + }() + // AAA's entry is past nodeRegionMaxStale; the index is stale, still + // usable, and due a rebuild (aged second: the entry helper ages it too). + ageNodeRegionEntry(t, db, "AAA", nodeRegionMaxStale, 0) + idx618AgeIndex(db, 2*nodeRegionFreshTTL, 2*nodeRegionRebuildInterval) + go db.refreshNodeRegion("BBB", []string{"BBB"}, true) // starts the rebuild + <-entered + + for _, region := range []string{"AAA", "CCC,DDD"} { + ch := make(chan []string, 1) + go func() { ch <- idx618Keys(t, db, region) }() + select { + case got := <-ch: + if region == "AAA" && !slices.Equal(got, wantAAA) { + t.Errorf("AAA during the rebuild: %v, want %v", got, wantAAA) + } + case <-time.After(2 * time.Second): + t.Fatalf("request for %q waited for the running rebuild", region) + } + // Cut from a stale snapshot, the entry is stale too: the next + // request re-cuts it instead of serving it for a full TTL. + if e := db.getNodeRegionEntry(nodeRegionKey(normalizeRegionCodes(region))); e == nil || e.fresh() { + t.Errorf("entry for %q cut from a stale snapshot: %+v, want a stale entry", region, e) + } + } + close(release) +} + +// TestNodeRegionIndex618_ServerWarmsAtStartup: the real server main() builds +// the heard index on its own after startup — no region request is made — +// and its shutdown stops the warmer before closing the database. +func TestNodeRegionIndex618_ServerWarmsAtStartup(t *testing.T) { + p := startServerProc(t, fullSchemaDB(t), time.Minute) + p.waitReady(t) + p.waitLog(t, "[nodes-region] heard: full build", 30*time.Second) + p.cmd.Process.Signal(syscall.SIGTERM) + p.waitExit(t, 30*time.Second) + if out := p.out.String(); !strings.Contains(out, "[server] shutdown complete") { + t.Fatalf("server did not shut down cleanly:\n%s", out) + } +} + +// TestNodeRegionIndex618_BoundedMemory: the index grows with distinct (node, +// region) pairs, never with hearings or refreshes; the snapshot holds one +// key per pair; the per-set cache stays capped however many sets are asked. +func TestNodeRegionIndex618_BoundedMemory(t *testing.T) { + db := rpm2OpenTestDB(t) + seq := 0 + hearAll := func(ts int64) { + for n := range 5 { + pk := fmt.Sprintf("%04x000000000000", n+1) + for obs := 1; obs <= 3; obs++ { + seq++ + rpm2Insert(t, db, pk, seq, obs, ts, rpm2Str("1200aabbccdd"), rpm2Str("[]")) + } + } + } + hearAll(rpm2T0) + idx618Keys(t, db, "AAA") + waitForHeardIndex(t, db, rpm2MaxID(t, db)) + for round := 1; round <= 20; round++ { + hearAll(rpm2T0 + int64(round)*3600) + idx618AgeIndex(db, 2*nodeRegionFreshTTL, 0) + ageNodeRegionEntry(t, db, "AAA", 2*nodeRegionFreshTTL, 0) + if _, err := db.refreshNodeRegion("AAA", []string{"AAA"}, true); err != nil { + t.Fatal(err) + } + } + db.regionHeard.mu.Lock() + agg := db.regionHeard.agg + db.regionHeard.mu.Unlock() + if agg.pairs != 15 || len(agg.nodes) != 5 || agg.watermark != rpm2MaxID(t, db) { + t.Fatalf("index grew with hearings: %d pairs, %d nodes, watermark %d; want 15, 5, %d", + agg.pairs, len(agg.nodes), agg.watermark, rpm2MaxID(t, db)) + } + members := 0 + for _, keys := range db.regionHeard.snap.Load().byRegion { + members += len(keys) + } + if members != agg.pairs { + t.Fatalf("snapshot holds %d keys for %d pairs", members, agg.pairs) + } + for i := range 2 * nodeRegionMaxEntries { + idx618Keys(t, db, fmt.Sprintf("Z%02d,AAA", i)) + } + db.nodeRegionCacheMu.Lock() + n := len(db.nodeRegionCache) + db.nodeRegionCacheMu.Unlock() + if n > nodeRegionMaxEntries { + t.Fatalf("cached %d region sets, cap %d", n, nodeRegionMaxEntries) + } +} + +// TestNodeRegionIndex618_V2Schema: the heard scan also runs on the legacy +// schema (observations.observer_id). +func TestNodeRegionIndex618_V2Schema(t *testing.T) { + db := setupTestDBv2(t) + t.Cleanup(func() { db.Close() }) + if db.isV3() { + t.Fatal("setupTestDBv2 produced a v3 schema") + } + for _, o := range []string{"AAA", "BBB"} { + if _, err := db.conn.Exec(`INSERT INTO observers (id, name, iata, last_seen, first_seen) VALUES (?, ?, ?, '2026-10-01T00:00:00Z', '2026-01-01T00:00:00Z')`, + "v2-"+o, "V2 "+o, o); err != nil { + t.Fatal(err) + } + } + if _, err := db.conn.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json, from_pubkey) + VALUES ('1100aabb', 'h-v2', '2026-10-01T00:00:00Z', 1, 4, '{}', ?)`, rpm2PK); err != nil { + t.Fatal(err) + } + if _, err := db.conn.Exec(`INSERT INTO observations (transmission_id, observer_id, observer_name, path_json, timestamp) + VALUES (1, 'v2-AAA', 'v2-AAA', '[]', '2026-10-01T00:00:00Z')`); err != nil { + t.Fatal(err) + } + for _, region := range []string{"AAA", "BBB", "AAA,BBB"} { + want := idx618PerSetKeys(t, db, normalizeRegionCodes(region)) + if got := idx618Keys(t, db, region); !slices.Equal(got, want) { + t.Errorf("v2 %q: %v, want %v", region, got, want) + } + } + if got := idx618Keys(t, db, "AAA"); !slices.Equal(got, []string{rpm2PK}) { + t.Errorf("v2 AAA: %v", got) + } +} + +// TestNodeRegionIndex618_SourceGuards: the region index reads SQLite only — +// never the PacketStore or its lock — and adds no untyped maps. +func TestNodeRegionIndex618_SourceGuards(t *testing.T) { + for _, f := range []string{"region_index.go", "nodes_region_cache.go", "region_placement.go"} { + src, err := os.ReadFile(f) + if err != nil { + t.Fatal(err) + } + for i, line := range strings.Split(string(src), "\n") { + if strings.HasPrefix(strings.TrimSpace(line), "//") { + continue + } + for _, banned := range []string{"PacketStore", "store.", "map[string]interface{}", "map[string]any"} { + if strings.Contains(line, banned) { + t.Errorf("%s:%d uses %q: %s", f, i+1, banned, strings.TrimSpace(line)) + } + } + } + } +} + +// ---- large-database equivalence and cost (env-gated) ---- + +// TestNodeRegionIndex618_UnionEqualsPerSetBuild_LargeDB runs the property +// on a generated database (TestPerf_RegionPlacementM2Generate_488 or +// TestPerf_GenerateRegionDB_176): +// +// CORESCOPE_PERF_618_DB=/path/big.db go test -run 'LargeDB$' -v -timeout 2h . +func TestNodeRegionIndex618_UnionEqualsPerSetBuild_LargeDB(t *testing.T) { + path := os.Getenv("CORESCOPE_PERF_618_DB") + if path == "" { + t.Skip("set CORESCOPE_PERF_618_DB= to run the property on a large database") + } + db, err := OpenDB(path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + full, _ := idx618CountScans(db) + codes := idx618Regions(t, db) + regions := idx618Subsets(codes) + n := idx618AssertEquivalent(t, db, filepath.Base(path), regions) + t.Logf("%s: %d regions, %d sets (%d non-empty) equal the per-set build; %d full scans", + filepath.Base(path), len(codes), len(regions), n, full.Load()) + if full.Load() != 1 { + t.Errorf("%d full scans, want 1", full.Load()) + } +} + +func idx618Regions(t testing.TB, db *DB) []string { + t.Helper() + rows, err := db.conn.Query(`SELECT DISTINCT UPPER(TRIM(iata)) FROM observers + WHERE iata IS NOT NULL AND TRIM(iata) != '' ORDER BY 1`) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var out []string + for rows.Next() { + var c string + if err := rows.Scan(&c); err != nil { + t.Fatal(err) + } + out = append(out, c) + } + return out +} + +// TestPerf_NodeRegionIndex618 times, on one database (medians of 5): +// +// - per-set: master's full build for every single code and every +// quick-pick-sized group, summed — what a cold instance paid; +// +// - index: one cold full build of every code; +// +// - warm: cutting a new set from the warm index (the first request for a +// set after warmup) and a cached hit; +// +// - heap of the index and its snapshot. +// +// CORESCOPE_PERF_618=/path/big.db go test -run '^TestPerf_NodeRegionIndex618$' -v -timeout 2h . +func TestPerf_NodeRegionIndex618(t *testing.T) { + path := os.Getenv("CORESCOPE_PERF_618") + if path == "" { + t.Skip("set CORESCOPE_PERF_618= to measure") + } + db, err := OpenDB(path) + if err != nil { + t.Fatal(err) + } + defer db.Close() + codes := idx618Regions(t, db) + sets := make([][]string, 0, len(codes)+3) + for _, c := range codes { + sets = append(sets, []string{c}) + } + // Quick-pick-sized groups: halves and all. + sets = append(sets, codes[:len(codes)/2], codes[len(codes)/2:], codes) + const runs = 5 + median := func(f func()) time.Duration { + d := make([]time.Duration, runs) + for i := range d { + start := time.Now() + f() + d[i] = time.Since(start) + } + slices.Sort(d) + return d[runs/2] + } + var perSet time.Duration + for _, codes := range sets { + perSet += median(func() { idx618PerSetKeys(t, db, codes) }) + } + clear := func() { + db.nodeRegionCacheMu.Lock() + db.nodeRegionCache, db.nodeRegionUsed = nil, nil + db.nodeRegionCacheMu.Unlock() + db.regionHeard.mu.Lock() + db.regionHeard.agg = nil + db.regionHeard.mu.Unlock() + db.regionHeard.snap.Store(nil) + } + cold := median(func() { + clear() + if _, err := db.refreshRegionIndex(context.Background(), &db.regionHeard, db.heardRegionSpec()); err != nil { + t.Fatal(err) + } + }) + snap := db.regionHeard.snap.Load() + var cutMax time.Duration + for _, codes := range sets { + d := median(func() { + if _, err := snap.entry(codes); err != nil { + t.Fatal(err) + } + }) + cutMax = max(cutMax, d) + } + firstAll := median(func() { + db.nodeRegionCacheMu.Lock() + db.nodeRegionCache, db.nodeRegionUsed = nil, nil + db.nodeRegionCacheMu.Unlock() + if _, err := db.nodeRegionKeysJSON(codes); err != nil { + t.Fatal(err) + } + }) + hit := median(func() { + if _, err := db.nodeRegionKeysJSON(codes); err != nil { + t.Fatal(err) + } + }) + // A cold request with no warmer, for the largest single code and for + // every code: it starts the index build and waits for it. + largest, largestN := codes[0], -1 + for _, c := range codes { + if n := len(snap.byRegion[c]); n > largestN { + largest, largestN = c, n + } + } + coldFirst := func(set []string) time.Duration { + d := make([]time.Duration, runs) + for i := range d { + clear() + start := time.Now() + if _, err := db.nodeRegionKeysJSON(set); err != nil { + t.Fatal(err) + } + d[i] = time.Since(start) + } + slices.Sort(d) + return d[runs/2] + } + coldOne, coldAll := coldFirst([]string{largest}), coldFirst(codes) + perSetOne := median(func() { idx618PerSetKeys(t, db, []string{largest}) }) + // A cold request for the largest code that arrives at 0/25/50/75 % of + // the warmer's build: it waits for that build, so its latency is the + // rest of the warmup ("remaining": the build's landing minus the + // arrival), never a second scan. + type duringWarmup struct{ pct, latency, remaining, landed time.Duration } + var warmup []duringWarmup + for _, pct := range []time.Duration{0, 25, 50, 75} { + lat, rem, land := make([]time.Duration, runs), make([]time.Duration, runs), make([]time.Duration, runs) + for i := range lat { + clear() + t0 := time.Now() + stop := db.startRegionIndexWarmer(time.Hour) + time.Sleep(cold * pct / 100) + arrive := time.Now() + if _, err := db.nodeRegionKeysJSON([]string{largest}); err != nil { + t.Fatal(err) + } + lat[i] = time.Since(arrive) + s := db.regionHeard.snap.Load() + stop() + if s == nil { + t.Fatal("request served before the warmup's snapshot") + } + land[i], rem[i] = s.refreshed.Sub(t0), s.refreshed.Sub(arrive) + } + for _, d := range [][]time.Duration{lat, rem, land} { + slices.Sort(d) + } + warmup = append(warmup, duringWarmup{pct, lat[runs/2], rem[runs/2], land[runs/2]}) + } + clear() + h0 := rpm2Heap() + if _, err := db.refreshRegionIndex(context.Background(), &db.regionHeard, db.heardRegionSpec()); err != nil { + t.Fatal(err) + } + heap := int64(rpm2Heap()) - int64(h0) + db.regionHeard.mu.Lock() + pairs, nodes := db.regionHeard.agg.pairs, len(db.regionHeard.agg.nodes) + db.regionHeard.mu.Unlock() + wm := rpm2MaxID(t, db) + t.Logf("%s: %d observations, %d regions, %d nodes, %d (node, region) pairs", filepath.Base(path), wm, len(codes), nodes, pairs) + t.Logf("per-set builds, %d sets (%d singles + 3 groups), summed: %v", len(sets), len(codes), perSet.Round(time.Millisecond)) + t.Logf("index cold build, all codes: %v (%.1fx less than the sum)", cold.Round(time.Millisecond), float64(perSet)/float64(cold)) + t.Logf("cold request, no warmer (waits for the index build it starts): %s %v (its per-set build alone %v) | all codes %v", + largest, coldOne.Round(time.Millisecond), perSetOne.Round(time.Millisecond), coldAll.Round(time.Millisecond)) + for _, w := range warmup { + t.Logf("cold request for %s at %d%% of the warmup: %v (%.2fx its per-set build alone); rest of the warmup %v; warmup landed after %v", + largest, w.pct, w.latency.Round(time.Millisecond), float64(w.latency)/float64(perSetOne), + w.remaining.Round(time.Millisecond), w.landed.Round(time.Millisecond)) + } + t.Logf("warm: worst cut of a set from the snapshot %v | first request for all codes (cut + encode) %v | cached hit %v | index heap %d KiB", + cutMax.Round(time.Microsecond), firstAll.Round(time.Microsecond), hit, heap/1024) +} diff --git a/cmd/server/region_index.go b/cmd/server/region_index.go new file mode 100644 index 000000000..c6f9d391c --- /dev/null +++ b/cmd/server/region_index.go @@ -0,0 +1,429 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log" + "sort" + "sync" + "sync/atomic" + "time" + + "golang.org/x/sync/singleflight" +) + +// Global region → members index for /api/nodes?region= (#488 M2, #618). +// +// A region set is served as the union of per-code member lists cut from ONE +// index per region rule, kept by one shared scan of the ADVERT observation +// history grouped by observer IATA. Region sets used to be built one by one +// in the heard rule, each a full history scan of its own: a minute per set +// on a large database, paid again for every new combination or quick pick. +// +// An index keeps the machinery #488 M2 introduced for placement: a +// regionEvidenceAgg with an observations.id watermark, delta scans past it, +// a full rebuild every nodeRegionRebuildInterval (retention pruning, +// observer region edits and late from_pubkey backfills change rows below +// the watermark), and an immutable snapshot readers load without a lock. +// The two rules differ only in their scan and in how a node's per-region +// evidence becomes member regions (regionIndexSpec). +// +// Refreshes of an index are deduplicated by its singleflight and serialised +// by its mutex. A caller that waits (refreshNodeRegion with waitIndex) shares +// the flight in progress; a caller a stale but usable snapshot can serve +// (younger than nodeRegionMaxStale) is served from it at once and only kicks +// a refresh, so no request queues behind a full rebuild once a snapshot +// exists. StartRegionIndexWarmer builds the index right after startup and +// keeps refreshing it, so the full scans run there and not under a request. +// A request that arrives before the first snapshot waits for the build +// already in flight, so it waits at most the rest of the warmup; it runs a +// per-set build of its own only if that build fails (coldNodeRegion). +// +// Everything is read from SQLite on the server's read-only handle; nothing +// touches the in-memory packet store or its lock. + +// regionIndexKeepInterval is how often the warmer refreshes the index: a +// delta scan past the watermark (milliseconds), or the periodic full +// rebuild once it is due. +const regionIndexKeepInterval = nodeRegionFreshTTL + +// errRegionIndexFlightResult means a regionIndex flight returned something +// other than a *regionMembershipSnapshot, which only a programming error +// can cause. +var errRegionIndexFlightResult = errors.New("unexpected region index flight result") + +// regionIndex is one global index. agg and built are owned by the refresh +// holding mu; readers only ever see snap. +type regionIndex struct { + sf singleflight.Group + mu sync.Mutex + agg *regionEvidenceAgg + built time.Time // start of agg's last full scan + snap atomic.Pointer[regionMembershipSnapshot] + // scanHook (test seam, nil in production) runs at the end of every + // scan, full or delta, after its rows were folded in; a non-nil error + // fails that scan the way an error part-way through the rows does. + scanHook func(full bool) error +} + +// regionIndexSpec is what one region rule scans and how it derives a node's +// member regions from its per-region evidence. +type regionIndexSpec struct { + label string + scan func(ctx context.Context, lo, hi int64, delta bool, agg *regionEvidenceAgg) (int, error) + regions func(perRegion map[string]*regionEvidenceStats) []string + // perSet builds one region set on its own, the fallback for a request + // that arrives before the index's first snapshot when the index build + // fails (coldNodeRegion); nil when the rule has no per-set form + // (placement needs every region's evidence). + perSet func(ctx context.Context, codes []string) (*nodeRegionEntry, error) +} + +// heardRegions is the heard rule: a node is a member of every region that +// heard any of its adverts. +func heardRegions(perRegion map[string]*regionEvidenceStats) []string { + out := make([]string, 0, len(perRegion)) + for r := range perRegion { + out = append(out, r) + } + return out +} + +func (db *DB) heardRegionSpec() regionIndexSpec { + return regionIndexSpec{label: "heard", scan: db.scanHeardRegions, regions: heardRegions, perSet: db.buildNodeRegionCold} +} + +func (db *DB) placedRegionSpec() regionIndexSpec { + p := db.regionPlacement.Params + return regionIndexSpec{ + label: "placement", + scan: db.scanRegionEvidence, + regions: func(perRegion map[string]*regionEvidenceStats) []string { + return placeNode(perRegion, p).Regions + }, + } +} + +// activeRegionIndex is the index the configured mode serves from. +func (db *DB) activeRegionIndex() (*regionIndex, regionIndexSpec) { + if db.regionPlacement.Mode == regionModePlaced { + return &db.regionPlaced, db.placedRegionSpec() + } + return &db.regionHeard, db.heardRegionSpec() +} + +// regionMembershipSnapshot is the immutable result of one index refresh: +// the region → members lists the per-set cache entries are cut from. +type regionMembershipSnapshot struct { + byRegion map[string][]string // sorted pubkeys that are members of each region + watermark int64 + refreshed time.Time + built time.Time // the index's last full build +} + +// newRegionMembershipSnapshot derives every node's member regions in agg. +// It shares the pubkey strings with agg, so the lists cost one slice slot +// per membership (at most agg.pairs). +func newRegionMembershipSnapshot(agg *regionEvidenceAgg, spec regionIndexSpec, built time.Time) *regionMembershipSnapshot { + byRegion := make(map[string][]string) + for pk, per := range agg.nodes { + for _, r := range spec.regions(per) { + byRegion[r] = append(byRegion[r], pk) + } + } + for _, keys := range byRegion { + sort.Strings(keys) + } + return ®ionMembershipSnapshot{byRegion: byRegion, watermark: agg.watermark, refreshed: time.Now(), built: built} +} + +// members is the set of nodes that are members of any of codes. +func (s *regionMembershipSnapshot) members(codes []string) map[string]struct{} { + set := make(map[string]struct{}) + for _, c := range codes { + for _, pk := range s.byRegion[c] { + set[pk] = struct{}{} + } + } + return set +} + +// entry cuts the per-set cache entry for codes. It carries the snapshot's +// own ages, so an entry cut from a stale snapshot is stale too and the +// fresh/stale ages nodeRegionKeysJSON promises are those of the data. +func (s *regionMembershipSnapshot) entry(codes []string) (*nodeRegionEntry, error) { + return newNodeRegionEntry(s.members(codes), s.watermark, s.refreshed, s.built) +} + +// currentRegionIndex returns a snapshot of ix to serve from. A fresh one is +// returned as is. With wait set, a stale one is refreshed first (sharing the +// refresh in flight, if any). Without it, a snapshot younger than +// nodeRegionMaxStale is returned at once and one refresh is kicked in the +// background; only a missing or over-stale snapshot is waited for. +func (db *DB) currentRegionIndex(ctx context.Context, ix *regionIndex, spec regionIndexSpec, wait bool) (*regionMembershipSnapshot, error) { + s := ix.snap.Load() + if s != nil { + age := time.Since(s.refreshed) + if age < nodeRegionFreshTTL { + return s, nil + } + if !wait && age < nodeRegionMaxStale { + // DoChan dedups against a refresh in flight, and its buffered + // result channel may be left unread. + ix.sf.DoChan("", func() (any, error) { + ctx, cancel := context.WithTimeout(context.Background(), nodeRegionRebuildInterval) + defer cancel() + return db.refreshRegionIndexLogged(ctx, ix, spec) + }) + return s, nil + } + } + return db.regionIndexFlight(ctx, ix, spec) +} + +// coldNodeRegion serves a region set that a caller waits for before ix has +// published its first snapshot (the warmup still running, or failing). It +// waits for the index build — the warmer's, or one it starts — so a cold +// request waits at most the rest of the warmup and adds no scan that would +// slow the one build every region set is served from. Only when that build +// fails does it fall back to the set's own per-set build, the scan every +// cold request ran before #618, so a failing index does not turn +// /api/nodes?region= into a 500. Fallbacks are serialised by +// nodeRegionColdMu, so a burst of cold sets holds one pooled connection for +// a full scan, and one that queued while a retry (the warmer's next tick, or +// another request) published the index is cut from it instead. +func (db *DB) coldNodeRegion(ctx context.Context, ix *regionIndex, spec regionIndexSpec, codes []string) (*nodeRegionEntry, error) { + s, err := db.regionIndexFlight(ctx, ix, spec) + if err == nil { + return s.entry(codes) + } + db.nodeRegionColdMu.Lock() + defer db.nodeRegionColdMu.Unlock() + if s := ix.snap.Load(); s != nil { + return s.entry(codes) + } + e, setErr := spec.perSet(ctx, codes) + if setErr != nil { + return nil, errors.Join(err, setErr) + } + return e, nil +} + +// regionIndexFlight runs one refresh of ix, or waits for the one in flight. +// The flight runs under the context of whoever started it, so a caller +// whose own ctx ends stops waiting rather than sit out a scan it cannot +// cancel (the warmer's stop at shutdown). +func (db *DB) regionIndexFlight(ctx context.Context, ix *regionIndex, spec regionIndexSpec) (*regionMembershipSnapshot, error) { + ch := ix.sf.DoChan("", func() (any, error) { + return db.refreshRegionIndexLogged(ctx, ix, spec) + }) + var res singleflight.Result + select { + case res = <-ch: + case <-ctx.Done(): + return nil, ctx.Err() + } + if res.Err != nil { + return nil, res.Err + } + v := res.Val + s, ok := v.(*regionMembershipSnapshot) + if !ok { + return nil, fmt.Errorf("nodes-region %s: %w: %T", spec.label, errRegionIndexFlightResult, v) + } + return s, nil +} + +// refreshRegionIndexLogged is refreshRegionIndex for a flight: a background +// refresh has no caller to report to, so a failure is logged here. +func (db *DB) refreshRegionIndexLogged(ctx context.Context, ix *regionIndex, spec regionIndexSpec) (*regionMembershipSnapshot, error) { + s, err := db.refreshRegionIndex(ctx, ix, spec) + if err != nil { + err = fmt.Errorf("nodes-region %s: %w", spec.label, err) + log.Printf("[nodes-region] index refresh failed: %v", err) + return nil, err + } + return s, nil +} + +// refreshRegionIndex brings ix up to date — a full build when there is none +// or one is due, else a delta past the watermark scanned into a separate +// aggregate and merged only on success — and publishes a new snapshot. A +// failed refresh leaves the aggregate and the published snapshot as they +// were. +func (db *DB) refreshRegionIndex(ctx context.Context, ix *regionIndex, spec regionIndexSpec) (*regionMembershipSnapshot, error) { + ix.mu.Lock() + defer ix.mu.Unlock() + if ix.agg == nil || time.Since(ix.built) >= nodeRegionRebuildInterval { + // Full scans of either index share nodeRegionFullMu, so at most one + // holds a pooled connection for the whole history. + db.nodeRegionFullMu.Lock() + defer db.nodeRegionFullMu.Unlock() + wm, err := db.maxObservationID(ctx) + if err != nil { + return nil, err + } + start := time.Now() + agg := newRegionEvidenceAgg() + if err := ix.scan(ctx, spec, 0, wm, false, agg); err != nil { + return nil, err + } + ix.agg, ix.built = agg, start + s := newRegionMembershipSnapshot(agg, spec, start) + ix.snap.Store(s) + log.Printf("[nodes-region] %s: full build, %d nodes, %d (node, region) pairs, %d regions in %v (obs id <= %d)", + spec.label, len(agg.nodes), agg.pairs, len(s.byRegion), time.Since(start).Round(time.Millisecond), wm) + return s, nil + } + wm, err := db.maxObservationID(ctx) + if err != nil { + return nil, err + } + agg := ix.agg + if prev := ix.snap.Load(); prev != nil && wm <= agg.watermark { + s := *prev + s.refreshed = time.Now() + ix.snap.Store(&s) + return &s, nil + } + if wm > agg.watermark { + delta := newRegionEvidenceAgg() + if err := ix.scan(ctx, spec, agg.watermark, wm, true, delta); err != nil { + return nil, err + } + agg.merge(delta) + } + s := newRegionMembershipSnapshot(agg, spec, ix.built) + ix.snap.Store(s) + return s, nil +} + +// scan runs spec's scan and then the test seam. +func (ix *regionIndex) scan(ctx context.Context, spec regionIndexSpec, lo, hi int64, delta bool, agg *regionEvidenceAgg) error { + if _, err := spec.scan(ctx, lo, hi, delta, agg); err != nil { + return err + } + if ix.scanHook != nil { + return ix.scanHook(!delta) + } + return nil +} + +// scanHeardRegions folds every ADVERT observation with lo < id <= hi, heard +// by an observer with a region, into agg as a (node, region) pair, and +// raises agg's watermark to hi. It is the heard rule's per-set filter +// (UPPER(TRIM(iata)) IN (codes), scanNodeRegionKeys) run once for every +// code: normalised codes are never empty, so the blank regions it skips +// match no set. +// +// The join order is forced (CROSS JOIN) both ways. A delta scan drives from +// the observations rowid range; left to the planner, sqlite_stat1 (#2058) +// makes it start from the observers and walk their whole history. A full +// scan drives from the ADVERT transmissions (idx_transmissions_payload_type, +// then idx_observations_transmission_id, both created by the ingestor's +// schema): with no region filter the planner walks every observation by +// rowid and looks up its transmission, most of them not ADVERTs. Rows are +// not DISTINCT-ed in SQL; agg.pair dedupes them, which measured cheaper +// than SQLite's temp B-tree (#618 PR). +func (db *DB) scanHeardRegions(ctx context.Context, lo, hi int64, delta bool, agg *regionEvidenceAgg) (int, error) { + joinCond := "obs.rowid = o.observer_idx" + if !db.isV3() { + joinCond = "obs.id = o.observer_id" + } + from := "observations o CROSS JOIN transmissions t ON t.id = o.transmission_id" + if !delta { + from = "transmissions t CROSS JOIN observations o ON o.transmission_id = t.id" + } + // Only fixed fragments are formatted in; every value is bound. + // + // #1143 / PR #38: from_pubkey is a dedicated column the ingestor fills at + // write time for ADVERT rows, so this lookup does not have to + // JSON_EXTRACT and parse decoded_json for every candidate row. Measured + // against a live 4.9GB database, 9 interleaved rounds so concurrent + // ingest cannot bias one variant: 934ms -> 835ms median (1.12x) for a + // single-region filter, identical result set both ways. The win is purely + // the avoided JSON parse. + // + // THE TRAP, for whoever touches this next: from_pubkey is only written + // for ADVERTs that actually carry a pubkey — see the guard in + // cmd/ingestor/db.go. A NULL here drops out silently, so a write path + // that fills decoded_json but forgets from_pubkey would cost nodes from + // the region filter with no error to notice. That is safe today only + // because it is measured to be: on live staging 25 of 62104 ADVERTs have + // NULL from_pubkey, and all 25 have no pubKey in decoded_json either — + // nothing is lost, because there was never a pubkey to record. A COALESCE + // fallback to JSON_EXTRACT was measured and rejected: it rescued 0 rows, + // and it parses decoded_json again for every NULL row, so one corrupt + // ADVERT that the backfill has not reached yet fails the whole scan with + // "malformed JSON" (a 500 on /api/nodes?region=). Behind this index that + // is worse, not better: the failed refresh leaves the index stale for + // every region set. If you add a new write path for transmissions, + // populate from_pubkey there rather than reintroducing a per-row JSON + // parse here. nodes_region_from_pubkey_test.go locks the result set + // against the JSON_EXTRACT expression; TestBackfillFromPubkey_* in + // cmd/ingestor locks the backfill's half of the contract. + q := fmt.Sprintf(`SELECT t.from_pubkey, UPPER(TRIM(obs.iata)) + FROM %[1]s + CROSS JOIN observers obs ON %[2]s + WHERE t.payload_type = ? + AND t.from_pubkey IS NOT NULL + AND obs.iata IS NOT NULL AND TRIM(obs.iata) != '' + AND o.id > ? AND o.id <= ?`, from, joinCond) + rows, err := db.conn.QueryContext(ctx, q, payloadTypeAdvert, lo, hi) + if err != nil { + return 0, fmt.Errorf("scan heard regions: %w", err) + } + defer rows.Close() + n := 0 + for rows.Next() { + var pk, region string + if err := rows.Scan(&pk, ®ion); err != nil { + return n, fmt.Errorf("scan heard regions: %w", err) + } + agg.pair(pk, region) + n++ + } + if err := rows.Err(); err != nil { + return n, fmt.Errorf("scan heard regions: %w", err) + } + agg.watermark = max(agg.watermark, hi) + return n, nil +} + +// StartRegionIndexWarmer builds the active region index in the background +// and keeps refreshing it every regionIndexKeepInterval, so the first +// /api/nodes?region= request after startup does not pay for the full scan, +// the index never ages past nodeRegionMaxStale, and its periodic rebuild +// runs here rather than under a request. The returned stop cancels a scan +// in flight and waits for the goroutine; it is safe to call more than once. +func (db *DB) StartRegionIndexWarmer() (stop func()) { + return db.startRegionIndexWarmer(regionIndexKeepInterval) +} + +func (db *DB) startRegionIndexWarmer(every time.Duration) (stop func()) { + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + ticker := time.NewTicker(every) + defer ticker.Stop() + for { + ix, spec := db.activeRegionIndex() + // Failures are logged by the flight; the next tick retries. + _, _ = db.regionIndexFlight(ctx, ix, spec) + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } + }() + var once sync.Once + return func() { + once.Do(func() { + cancel() + <-done + }) + } +} diff --git a/cmd/server/region_placement.go b/cmd/server/region_placement.go index 9874e916d..26bb6347f 100644 --- a/cmd/server/region_placement.go +++ b/cmd/server/region_placement.go @@ -4,7 +4,6 @@ import ( "context" "database/sql" "fmt" - "log" "math" "sort" "time" @@ -36,15 +35,13 @@ import ( // // Placement needs every region's evidence for a node at once (direct // evidence anywhere decides whether the fallback applies here), so it is -// computed from ONE global aggregate rather than per region set. The -// aggregate keeps the heard cache's machinery: an observations.id -// watermark, delta scans past it, a full rebuild every -// nodeRegionRebuildInterval (retention pruning, observer region edits and -// late from_pubkey backfills change rows below the watermark), the per-set -// singleflight and the same fresh/stale ages. Unlike the heard rule, -// placement is not monotone — a new direct hearing removes fallback -// regions — so every refresh re-derives the placements instead of -// extending them. +// computed from ONE global aggregate rather than per region set: the +// region index in region_index.go, which the heard rule shares since #618 +// (watermark, delta scans, periodic full rebuild, singleflight, warmer). +// Placement is not monotone — a new direct hearing removes fallback +// regions — so every refresh re-derives the placements from the aggregate +// instead of extending them; the heard rule's do the same, which costs one +// pass over the pairs. // // Everything is read from SQLite on the server's read-only handle; nothing // touches the PacketStore or its lock. @@ -236,10 +233,10 @@ func (a *regionEvidenceAgg) merge(d *regionEvidenceAgg) { // scanRegionEvidence folds every ADVERT observation with lo < id <= hi, // heard by an observer with a region, into agg, and raises agg's watermark -// to hi. Same scope as scanNodeRegionKeys (so a placed region is always a +// to hi. Same scope as scanHeardRegions (so a placed region is always a // heard one), plus the per-observation frame and path the classifier needs. // A delta scan forces the join order to drive from the rowid range, as -// scanNodeRegionKeys explains. +// scanHeardRegions explains. func (db *DB) scanRegionEvidence(ctx context.Context, lo, hi int64, delta bool, agg *regionEvidenceAgg) (int, error) { joinCond, tsCol := "obs.rowid = o.observer_idx", "o.timestamp" if !db.isV3() { @@ -256,7 +253,7 @@ func (db *DB) scanRegionEvidence(ctx context.Context, lo, hi int64, delta bool, rawCol = "o.raw_hex" } // Only fixed fragments are formatted in; every value is bound. See - // scanNodeRegionKeys for the from_pubkey contract. + // scanHeardRegions for the from_pubkey contract. q := fmt.Sprintf(`SELECT t.from_pubkey, UPPER(TRIM(obs.iata)), %[3]s, o.path_json, %[4]s FROM observations o %[1]s transmissions t ON t.id = o.transmission_id @@ -281,11 +278,6 @@ func (db *DB) scanRegionEvidence(ctx context.Context, lo, hi int64, delta bool, agg.add(pk, region, ts.Int64, packetpath.ClassifyAdvert(raw.String, path.String)) n++ } - if db.regionPlacementScanHook != nil { - if err := db.regionPlacementScanHook(!delta); err != nil { - return n, err - } - } if err := rows.Err(); err != nil { return n, fmt.Errorf("scan region evidence: %w", err) } @@ -352,117 +344,3 @@ func placeNode(perRegion map[string]*regionEvidenceStats, p regionPlacementParam } return unknown } - -// regionPlacementSnapshot is the immutable result of one aggregate refresh: -// the region → members index the per-set cache entries are cut from. -type regionPlacementSnapshot struct { - byRegion map[string][]string // sorted pubkeys placed in each region - watermark int64 - refreshed time.Time - built time.Time // the aggregate's last full build -} - -// newRegionPlacementSnapshot places every node in agg. It shares the -// pubkey strings with agg, so the index costs one slice slot per placed -// membership (at most agg.pairs). -func newRegionPlacementSnapshot(agg *regionEvidenceAgg, p regionPlacementParams, built time.Time) *regionPlacementSnapshot { - byRegion := make(map[string][]string) - for pk, per := range agg.nodes { - for _, r := range placeNode(per, p).Regions { - byRegion[r] = append(byRegion[r], pk) - } - } - for _, keys := range byRegion { - sort.Strings(keys) - } - return ®ionPlacementSnapshot{byRegion: byRegion, watermark: agg.watermark, refreshed: time.Now(), built: built} -} - -// members is the set of nodes placed in any of codes. -func (s *regionPlacementSnapshot) members(codes []string) map[string]struct{} { - set := make(map[string]struct{}) - for _, c := range codes { - for _, pk := range s.byRegion[c] { - set[pk] = struct{}{} - } - } - return set -} - -// currentRegionPlacement returns a snapshot no older than -// nodeRegionFreshTTL, refreshing the aggregate when needed. It is called -// from a per-set refresh (already deduplicated per region set by -// nodeRegionSF), so it never runs on a request a cached entry can serve. -// Refreshes for different region sets coalesce on regionAggMu: the first -// builds, the ones queued behind it find no rows past its watermark and -// only re-stamp its snapshot. -func (db *DB) currentRegionPlacement(ctx context.Context) (*regionPlacementSnapshot, error) { - if s := db.regionSnap.Load(); s != nil && time.Since(s.refreshed) < nodeRegionFreshTTL { - return s, nil - } - return db.refreshRegionPlacement(ctx) -} - -// refreshRegionPlacement brings the aggregate up to date — a full build when -// there is none or one is due, else a delta past the watermark scanned into -// a separate aggregate and merged only on success — and publishes a new -// snapshot. A failed refresh leaves the aggregate and the published -// snapshot as they were. -func (db *DB) refreshRegionPlacement(ctx context.Context) (*regionPlacementSnapshot, error) { - db.regionAggMu.Lock() - defer db.regionAggMu.Unlock() - params := db.regionPlacement.Params - if db.regionAgg == nil || time.Since(db.regionAggBuilt) >= nodeRegionRebuildInterval { - // Full builds share nodeRegionFullMu with the heard cache's, so at - // most one full observation scan holds a pooled connection. - db.nodeRegionFullMu.Lock() - defer db.nodeRegionFullMu.Unlock() - wm, err := db.maxObservationID(ctx) - if err != nil { - return nil, err - } - start := time.Now() - agg := newRegionEvidenceAgg() - if _, err := db.scanRegionEvidence(ctx, 0, wm, false, agg); err != nil { - return nil, err - } - db.regionAgg, db.regionAggBuilt = agg, start - s := newRegionPlacementSnapshot(agg, params, start) - db.regionSnap.Store(s) - log.Printf("[nodes-region] placement: full build, %d nodes, %d (node, region) pairs in %v (obs id <= %d)", - len(agg.nodes), agg.pairs, time.Since(start).Round(time.Millisecond), wm) - return s, nil - } - wm, err := db.maxObservationID(ctx) - if err != nil { - return nil, err - } - agg := db.regionAgg - if wm <= agg.watermark { - prev := db.regionSnap.Load() - s := *prev - s.refreshed = time.Now() - db.regionSnap.Store(&s) - return &s, nil - } - delta := newRegionEvidenceAgg() - if _, err := db.scanRegionEvidence(ctx, agg.watermark, wm, true, delta); err != nil { - return nil, err - } - agg.merge(delta) - s := newRegionPlacementSnapshot(agg, params, db.regionAggBuilt) - db.regionSnap.Store(s) - return s, nil -} - -// placedNodeRegion cuts the entry for one region set from the current -// placement snapshot. The entry's age and stale-while-revalidate handling -// stay those of nodeRegionKeysJSON; its membership is re-derived on every -// refresh, never extended. -func (db *DB) placedNodeRegion(ctx context.Context, codes []string) (*nodeRegionEntry, error) { - s, err := db.currentRegionPlacement(ctx) - if err != nil { - return nil, err - } - return newNodeRegionEntry(s.members(codes), s.watermark, s.built) -} diff --git a/cmd/server/region_placement_488_test.go b/cmd/server/region_placement_488_test.go index 78f0c30ab..b20a44654 100644 --- a/cmd/server/region_placement_488_test.go +++ b/cmd/server/region_placement_488_test.go @@ -278,18 +278,14 @@ func rpm2Members(t testing.TB, db *DB, codes ...string) []string { // path does in the background. func rpm2Refresh(t testing.TB, db *DB, codes ...string) []string { t.Helper() - if s := db.regionSnap.Load(); s != nil { - c := *s - c.refreshed = time.Now().Add(-2 * nodeRegionFreshTTL) - db.regionSnap.Store(&c) - } + ageRegionIndex(&db.regionPlaced, 2*nodeRegionFreshTTL, 0) key := nodeRegionKey(codes) if e := db.getNodeRegionEntry(key); e != nil { c := *e c.refreshed = time.Now().Add(-2 * nodeRegionFreshTTL) db.setNodeRegionEntry(key, &c) } - e, err := db.refreshNodeRegion(key, codes) + e, err := db.refreshNodeRegion(key, codes, true) if err != nil { t.Fatal(err) } @@ -300,15 +296,13 @@ func rpm2Refresh(t testing.TB, db *DB, codes ...string) []string { // rpm2RebuildDue makes the aggregate's next refresh a full rebuild. func rpm2RebuildDue(db *DB) { - db.regionAggMu.Lock() - db.regionAggBuilt = time.Now().Add(-2 * nodeRegionRebuildInterval) - db.regionAggMu.Unlock() + ageRegionIndex(&db.regionPlaced, 0, 2*nodeRegionRebuildInterval) } // rpm2CountScans installs the scan hook and returns the full/delta counters. func rpm2CountScans(db *DB) (full, delta *atomic.Int32) { full, delta = &atomic.Int32{}, &atomic.Int32{} - db.regionPlacementScanHook = func(isFull bool) error { + db.regionPlaced.scanHook = func(isFull bool) error { if isFull { full.Add(1) } else { @@ -357,7 +351,7 @@ func TestRegionPlacement488_DefaultModeIsHeard(t *testing.T) { if got := rpm2Members(t, db, "BBB"); !reflect.DeepEqual(got, []string{rpm2PK}) { t.Fatalf("default mode: BBB members %v, want the relayed-only node (heard rule)", got) } - if db.regionSnap.Load() != nil || full.Load()+delta.Load() != 0 { + if db.regionPlaced.snap.Load() != nil || full.Load()+delta.Load() != 0 { t.Fatalf("default mode built the placement aggregate (%d full, %d delta scans)", full.Load(), delta.Load()) } } @@ -403,7 +397,7 @@ func TestRegionPlacement488_NewObservationCanRemoveRegion(t *testing.T) { if full.Load() != 1 || delta.Load() != 1 { t.Errorf("scans: %d full, %d delta; want 1 full (cold) and 1 delta (the refresh)", full.Load(), delta.Load()) } - if s := db.regionSnap.Load(); s == nil || s.watermark != rpm2MaxID(t, db) { + if s := db.regionPlaced.snap.Load(); s == nil || s.watermark != rpm2MaxID(t, db) { t.Errorf("snapshot watermark not advanced: %+v", s) } } @@ -496,7 +490,7 @@ func TestRegionPlacement488_StartupSingleflight(t *testing.T) { rpm2Insert(t, db, "bbbb000000000002", 2, 2, rpm2T0, rpm2Str("1100aabbccdd"), rpm2Str("[]")) release := make(chan struct{}) var full, delta atomic.Int32 - db.regionPlacementScanHook = func(isFull bool) error { + db.regionPlaced.scanHook = func(isFull bool) error { if isFull { full.Add(1) } else { @@ -546,32 +540,30 @@ func TestRegionPlacement488_FailedDeltaLeavesAggregateIntact(t *testing.T) { rpm2Insert(t, db, rpm2PK, 1, 1, rpm2T0, rpm2Str("1101a1aabbccdd"), rpm2Str(`["A1"]`)) rpm2Members(t, db, "AAA") rpm2Insert(t, db, rpm2PK, 2, 2, rpm2T0+60, rpm2Str("1200aabbccdd"), rpm2Str("[]")) - db.regionPlacementScanHook = func(bool) error { return errors.New("injected scan failure") } + db.regionPlaced.scanHook = func(bool) error { return errors.New("injected scan failure") } key := nodeRegionKey([]string{"AAA"}) if e := db.getNodeRegionEntry(key); e != nil { c := *e c.refreshed = time.Now().Add(-2 * nodeRegionFreshTTL) db.setNodeRegionEntry(key, &c) } - s := *db.regionSnap.Load() - s.refreshed = time.Now().Add(-2 * nodeRegionFreshTTL) - db.regionSnap.Store(&s) - if _, err := db.refreshNodeRegion(key, []string{"AAA"}); err == nil { + ageRegionIndex(&db.regionPlaced, 2*nodeRegionFreshTTL, 0) + if _, err := db.refreshNodeRegion(key, []string{"AAA"}, true); err == nil { t.Fatal("refresh with a failing scan reported success") } - db.regionAggMu.Lock() - wm, pairs := db.regionAgg.watermark, db.regionAgg.pairs - db.regionAggMu.Unlock() + db.regionPlaced.mu.Lock() + wm, pairs := db.regionPlaced.agg.watermark, db.regionPlaced.agg.pairs + db.regionPlaced.mu.Unlock() if wm != 1 || pairs != 1 { t.Fatalf("failed delta changed the aggregate: watermark %d, pairs %d", wm, pairs) } - db.regionPlacementScanHook = nil + db.regionPlaced.scanHook = nil if got := rpm2Refresh(t, db, "AAA"); len(got) != 0 { t.Fatalf("after recovery: AAA members %v, want none", got) } - db.regionAggMu.Lock() - inc := db.regionAgg - db.regionAggMu.Unlock() + db.regionPlaced.mu.Lock() + inc := db.regionPlaced.agg + db.regionPlaced.mu.Unlock() fresh := rpm2Scan(t, db, 0, rpm2MaxID(t, db), nil) if inc.pairs != fresh.pairs || inc.hearings != fresh.hearings || !reflect.DeepEqual(rpm2Placements(inc, defaultRegionPlacementParams), rpm2Placements(fresh, defaultRegionPlacementParams)) { @@ -601,9 +593,9 @@ func TestRegionPlacement488_BoundedMemory(t *testing.T) { hearAll(rpm2T0 + int64(round)*3600) rpm2Refresh(t, db, "AAA") } - db.regionAggMu.Lock() - agg := db.regionAgg - db.regionAggMu.Unlock() + db.regionPlaced.mu.Lock() + agg := db.regionPlaced.agg + db.regionPlaced.mu.Unlock() if agg.pairs != 15 || len(agg.nodes) != 5 { t.Fatalf("aggregate grew with hearings: %d pairs, %d nodes; want 15 and 5", agg.pairs, len(agg.nodes)) } @@ -611,7 +603,7 @@ func TestRegionPlacement488_BoundedMemory(t *testing.T) { t.Fatalf("hearing counters %v, want %d hearings in all", agg.hearings, seq) } members := 0 - for _, keys := range db.regionSnap.Load().byRegion { + for _, keys := range db.regionPlaced.snap.Load().byRegion { members += len(keys) } if members > agg.pairs { @@ -686,9 +678,9 @@ func TestRegionPlacement488_SyntheticAccuracy(t *testing.T) { for _, pk := range placedWrong { t.Errorf("placed %s: got %v, want %v", pk, placed[pk], truth[pk]) } - db.regionAggMu.Lock() - pl := rpm2Placements(db.regionAgg, params) - db.regionAggMu.Unlock() + db.regionPlaced.mu.Lock() + pl := rpm2Placements(db.regionPlaced.agg, params) + db.regionPlaced.mu.Unlock() for pk, want := range transient { if !reflect.DeepEqual(pl[pk].Transient, want) { t.Errorf("placed %s transient %v, want %v", pk, pl[pk].Transient, want) diff --git a/cmd/server/region_placement_perf_488_test.go b/cmd/server/region_placement_perf_488_test.go index 607d9f34b..6881e146f 100644 --- a/cmd/server/region_placement_perf_488_test.go +++ b/cmd/server/region_placement_perf_488_test.go @@ -153,11 +153,12 @@ func TestPerf_RegionPlacementM2Generate_488(t *testing.T) { } // TestPerf_RegionPlacementM2_488 times, on one database, medians of 7: -// - heard: cold build of every region (each set scans separately), warm +// - heard: cold build of every region (one shared index since #618), warm // hit, delta refresh over the last N observations; // - placed: cold global aggregate + placements serving every region, // placement-only recompute, delta scan + merge + recompute, warm hit; -// - heap of the four heard entries vs the aggregate and its snapshot. +// - heap of the heard index and its entries vs the placement aggregate +// and its snapshot. func TestPerf_RegionPlacementM2_488(t *testing.T) { path := os.Getenv("CORESCOPE_PERF_488_DB") if path == "" { @@ -229,13 +230,15 @@ func rpm2PerfReport(t *testing.T, path string, deltaRows int64) { db.nodeRegionCache, db.nodeRegionUsed = nil, nil db.nodeRegionCacheMu.Unlock() } - clearPlaced := func() { + clearIndex := func(ix *regionIndex) { clearEntries() - db.regionAggMu.Lock() - db.regionAgg = nil - db.regionAggMu.Unlock() - db.regionSnap.Store(nil) + ix.mu.Lock() + ix.agg = nil + ix.mu.Unlock() + ix.snap.Store(nil) } + clearHeard := func() { clearIndex(&db.regionHeard) } + clearPlaced := func() { clearIndex(&db.regionPlaced) } serveAll := func() { for _, r := range regions { if _, err := db.nodeRegionKeysJSON([]string{r}); err != nil { @@ -246,40 +249,45 @@ func rpm2PerfReport(t *testing.T, path string, deltaRows int64) { // heard (today) db.regionPlacement = regionPlacementSettings{Mode: regionModeHeard, Params: defaultRegionPlacementParams} - heardCold := rpm2Median(func() { clearEntries(); serveAll() }) + heardCold := rpm2Median(func() { clearHeard(); serveAll() }) heardWarm := rpm2Median(serveAll) + // #618: every region is cut from one heard index, so a delta is one + // scan past the index's watermark for all of them. heardDelta := rpm2Median(func() { + db.regionHeard.mu.Lock() + db.regionHeard.agg.watermark = lo + db.regionHeard.mu.Unlock() + ageRegionIndex(&db.regionHeard, 2*nodeRegionFreshTTL, 0) for _, r := range regions { key := nodeRegionKey([]string{r}) e := *db.getNodeRegionEntry(key) - e.watermark, e.refreshed = lo, time.Now().Add(-2*nodeRegionFreshTTL) + e.refreshed = time.Now().Add(-2 * nodeRegionFreshTTL) db.setNodeRegionEntry(key, &e) - if _, err := db.refreshNodeRegion(key, []string{r}); err != nil { + if _, err := db.refreshNodeRegion(key, []string{r}, true); err != nil { t.Fatal(err) } } }) heardAdvScan := rpm2Median(func() { - for _, r := range regions { - if _, err := db.scanNodeRegionKeys(ctx, []string{r}, advLo, advHi, true, map[string]struct{}{}); err != nil { - t.Fatal(err) - } + if _, err := db.scanHeardRegions(ctx, advLo, advHi, true, newRegionEvidenceAgg()); err != nil { + t.Fatal(err) } }) - clearEntries() + clearHeard() h0 := rpm2Heap() serveAll() heardHeap := int64(rpm2Heap()) - int64(h0) - clearEntries() + clearHeard() // placed rpm2SetPlaced(db, defaultRegionPlacementParams) placedCold := rpm2Median(func() { clearPlaced(); serveAll() }) placedWarm := rpm2Median(serveAll) - db.regionAggMu.Lock() - agg := db.regionAgg - db.regionAggMu.Unlock() - placeOnly := rpm2Median(func() { newRegionPlacementSnapshot(agg, defaultRegionPlacementParams, time.Now()) }) + db.regionPlaced.mu.Lock() + agg := db.regionPlaced.agg + db.regionPlaced.mu.Unlock() + placedSpec := db.placedRegionSpec() + placeOnly := rpm2Median(func() { newRegionMembershipSnapshot(agg, placedSpec, time.Now()) }) // The delta re-folds the same tail rows each run: the stats merge by // min/max, so the placement work is the same as for new rows. placedDeltaOver := func(lo, hi int64) time.Duration { @@ -289,9 +297,9 @@ func rpm2PerfReport(t *testing.T, path string, deltaRows int64) { t.Fatal(err) } agg.merge(d) - snap := newRegionPlacementSnapshot(agg, defaultRegionPlacementParams, time.Now()) + snap := newRegionMembershipSnapshot(agg, placedSpec, time.Now()) for _, r := range regions { - if _, err := newNodeRegionEntry(snap.members([]string{r}), hi, time.Now()); err != nil { + if _, err := snap.entry([]string{r}); err != nil { t.Fatal(err) } } @@ -307,7 +315,7 @@ func rpm2PerfReport(t *testing.T, path string, deltaRows int64) { placedHeap := int64(rpm2Heap()) - int64(h1) t.Logf("%s: %d observations, %d regions, %d nodes, %d (node, region) pairs", filepath.Base(path), wm, len(regions), nodes, pairs) - t.Logf("heard cold %v | warm %v | delta refresh, newest %d obs %v | delta scan, %d advert obs %v | heap %d KiB", + t.Logf("heard cold %v | warm %v | delta refresh, newest %d obs %v | delta scan, %d advert obs %v (all regions, one scan) | heap %d KiB", heardCold.Round(time.Microsecond), heardWarm, deltaRows, heardDelta.Round(time.Microsecond), deltaRows, heardAdvScan.Round(time.Microsecond), heardHeap/1024) t.Logf("placed cold %v | warm %v | delta scan+merge+placement+entries, newest %d obs %v, %d advert obs %v | placement only %v | heap %d KiB",