From 34ed4247e07e2fcab465caa82a2ed79d56537a57 Mon Sep 17 00:00:00 2001 From: R0ck Date: Fri, 2 Oct 2026 22:29:27 +0100 Subject: [PATCH] perf(nodes): serve /api/nodes?region= from an ingestor-built table (#2101) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `/api/nodes?region=X` decided membership with an inline `public_key IN (SELECT DISTINCT from_pubkey FROM transmissions ⋈ observations ⋈ observers ...)` subquery: uncached, unbounded in time, and evaluated twice per request (COUNT and page). fetchAllNodes() pages at 500, so each Nodes or Map view ran it several times. On a 1.4 GB database (1.47M observations, 2 vCPU) one evaluation took 3-20 s depending on page cache, and one open tab kept the reader pool and both cores busy. The answer is small (7,333 node/observer pairs on that database) and changes slowly, so it is now derived once and kept, the way neighbor_edges is (#1287, #1339): - dbschema creates node_advert_observers(public_key, observer_idx, last_seen, last_obs_id), one row per observer that has heard a node's ADVERTs, and AssertReady requires it. - The ingestor is its only writer. Observations are folded in by observations.id in bounded steps (50k ids, ~0.3 s each): the first full build runs as an async migration, then a 30 s ticker folds new rows. PruneOldPackets drops pairs at the same cutoff that removes their newest advert. NULL from_pubkey rows use the pubkey BackfillFromPubkey will write. - The server reads the table once that migration is `done` (re-checked at most every 30 s, then latched); until then, on an ingestor that predates it, or on a v2 schema it keeps the legacy query unchanged. The region tag is joined from observers at query time, as before. Results match the legacy query; tests use it as the oracle on both modules. Measured on a snapshot of that database: backfill 3.6 s, every region and the all-regions set identical to the legacy query before and after a 3-day prune, table query 0.3-3 ms against 38-677 ms for the legacy query on the same (warm, faster) machine. Co-Authored-By: Claude Opus 5.5 --- cmd/ingestor/db.go | 3 + cmd/ingestor/main.go | 6 + cmd/ingestor/maintenance.go | 9 + cmd/ingestor/node_advert_observers.go | 325 +++++++++++ cmd/ingestor/node_advert_observers_test.go | 617 +++++++++++++++++++++ cmd/server/db.go | 42 +- cmd/server/node_advert_observers.go | 70 +++ cmd/server/node_advert_observers_test.go | 299 ++++++++++ internal/dbschema/dbschema.go | 40 ++ internal/dbschema/dbschema_test.go | 41 ++ 10 files changed, 1438 insertions(+), 14 deletions(-) create mode 100644 cmd/ingestor/node_advert_observers.go create mode 100644 cmd/ingestor/node_advert_observers_test.go create mode 100644 cmd/server/node_advert_observers.go create mode 100644 cmd/server/node_advert_observers_test.go diff --git a/cmd/ingestor/db.go b/cmd/ingestor/db.go index ff80b2827..45b633fa1 100644 --- a/cmd/ingestor/db.go +++ b/cmd/ingestor/db.go @@ -102,6 +102,9 @@ type Store struct { // pubkey -> rxTime of the last last_seen write (#1598). Guarded by // writerMu, which InsertTransmission holds for its whole body. relayTouched map[string]time.Time + + // advertObs is the node_advert_observers builder's state (#2101). + advertObs advertObserversState } // relayTouchDebounce is the minimum interval between two last_seen writes diff --git a/cmd/ingestor/main.go b/cmd/ingestor/main.go index ce1ad6a08..792a7232e 100644 --- a/cmd/ingestor/main.go +++ b/cmd/ingestor/main.go @@ -604,6 +604,12 @@ func main() { defer stopNeighborBuilder() log.Printf("[neighbor-build] enabled (interval=%s)", NeighborEdgesBuilderInterval) + // node_advert_observers builder (#2101): which observers heard each + // node's adverts, read by the server's /api/nodes region filter. + stopAdvertObservers := store.StartNodeAdvertObserversBuilder(NodeAdvertObserversInterval) + defer stopAdvertObservers() + log.Printf("[node-advert-observers] enabled (interval=%s)", NodeAdvertObserversInterval) + // #1212: per-source stall watchdog. Detects "silently dead" sources // where the client reports connected but no messages have flowed. Logs // a WARN line every minute for any source silent for >5m. Scan every diff --git a/cmd/ingestor/maintenance.go b/cmd/ingestor/maintenance.go index 6ee0fe77d..ee02adead 100644 --- a/cmd/ingestor/maintenance.go +++ b/cmd/ingestor/maintenance.go @@ -101,6 +101,15 @@ func (s *Store) PruneOldPackets(days int) (int64, error) { if total > 0 { log.Printf("[prune] deleted %d transmissions older than %d days", total, days) } + // Keep node_advert_observers in step: a pair whose newest advert was just + // pruned no longer matches the region filter (#2101). + n, err := s.pruneAdvertObservers(cutoff) + if err != nil { + return total, err + } + if n > 0 { + log.Printf("[prune] deleted %d node_advert_observers pairs older than %d days", n, days) + } return total, nil } diff --git a/cmd/ingestor/node_advert_observers.go b/cmd/ingestor/node_advert_observers.go new file mode 100644 index 000000000..45c74ddc6 --- /dev/null +++ b/cmd/ingestor/node_advert_observers.go @@ -0,0 +1,325 @@ +package main + +import ( + "context" + "database/sql" + "errors" + "fmt" + "log" + "sync" + "sync/atomic" + "time" + + "github.com/meshcore-analyzer/dbschema" +) + +// node_advert_observers (#2101) records which observers have heard each +// node's ADVERTs: one row per (public_key, observer_idx). The server's +// /api/nodes?region= filter reads it instead of re-deriving the same set +// from transmissions ⋈ observations ⋈ observers on every request, which on +// a 1.4 GB database took ~10 s per evaluation, twice per page. +// +// The table is derived, and this file is its only writer. Observations are +// folded in by observations.id, in bounded steps: +// +// - the first build is an async migration (dbschema.NodeAdvertObserversBackfill) +// that walks the whole history; the server keeps its legacy query until +// that migration is `done`; +// - after it, a ticker folds in whatever has arrived since the last step; +// - PruneOldPackets drops pairs whose newest advert it has just pruned. +// +// A pair matches the legacy subquery's semantics exactly: it exists while +// at least one ADVERT transmission with a non-empty from_pubkey, still in +// the database, was observed by that observer. The region code is not +// stored; the server joins observers at query time, so an observer moving +// region is reflected immediately, as before. + +const ( + // NodeAdvertObserversInterval is how often new observations are folded + // in after the backfill. It bounds how long a node newly heard in a + // region takes to appear under that region's filter. + NodeAdvertObserversInterval = 30 * time.Second + + // nodeAdvertObserversBatch caps the observation ids one step reads. + // The ingestor has a single connection (db.go SetMaxOpenConns(1)), so a + // step holds it for its read and its upsert; on a 2.4M-observation + // database the slowest 50k step read in 0.34 s (#1339 uses the same cap). + nodeAdvertObserversBatch = 50000 + + // nodeAdvertObserversYield is the pause between steps of a multi-step + // catch-up, so live ingest is never starved by a backfill. + nodeAdvertObserversYield = 50 * time.Millisecond + + // nodeAdvertObserversSlowStep is logged loudly when exceeded. + nodeAdvertObserversSlowStep = 2 * time.Second + + // nodeAdvertObserversStopWait bounds how long stop waits for an + // in-flight step, matching the neighbor-edges builder. + nodeAdvertObserversStopWait = 5 * time.Second +) + +// advertObserversState is the builder's in-process state. mu serialises +// steps and the prune of stale pairs, so a step can never write back a pair +// whose transmissions a concurrent prune has just removed. +type advertObserversState struct { + mu sync.Mutex + watermark int64 // highest observations.id folded in; guarded by mu + loaded bool // watermark read from the table; guarded by mu + ready atomic.Bool +} + +type advertObserverKey struct { + publicKey string + observerIdx int64 +} + +type advertObserverPair struct { + lastSeen string + lastObsID int64 +} + +// StartNodeAdvertObserversBuilder schedules the backfill (once per database) +// and the periodic fold, and returns a stop function. +func (s *Store) StartNodeAdvertObserversBuilder(interval time.Duration) func() { + if interval <= 0 { + interval = NodeAdvertObserversInterval + } + ctx, cancel := context.WithCancel(context.Background()) + s.scheduleAdvertObserversBackfill(ctx) + + done := make(chan struct{}) + go func() { + defer close(done) + s.runAdvertObserversTicker(ctx, interval) + }() + + var once sync.Once + return func() { + once.Do(cancel) + select { + case <-done: + case <-time.After(nodeAdvertObserversStopWait): + } + } +} + +// scheduleAdvertObserversBackfill runs the first full build as an async +// migration, unless a previous run already completed it. +func (s *Store) scheduleAdvertObserversBackfill(ctx context.Context) { + status, err := s.AsyncMigrationStatus(dbschema.NodeAdvertObserversBackfill) + if err == nil && status == "done" { + s.advertObs.ready.Store(true) + return + } + if err != nil && !errors.Is(err, sql.ErrNoRows) { + // Not scheduled: the builder stays not ready and the server keeps + // its legacy query, which is correct, just slow. + log.Printf("[node-advert-observers] reading backfill status failed: %v", err) + return + } + err = s.RunAsyncMigration(ctx, dbschema.NodeAdvertObserversBackfill, func(ctx context.Context, _ *sql.DB) error { + start := time.Now() + n, err := s.catchUpAdvertObservers(ctx, nodeAdvertObserversYield) + if err != nil { + return err + } + log.Printf("[node-advert-observers] backfill: %d pair updates in %s", + n, time.Since(start).Round(time.Millisecond)) + s.advertObs.ready.Store(true) + return nil + }) + if err != nil { + log.Printf("[node-advert-observers] scheduling backfill failed: %v", err) + } +} + +// runAdvertObserversTicker folds new observations every interval until ctx +// ends. Until the backfill is done it owns the catch-up, so ticks are skipped +// rather than contending for the same steps. +func (s *Store) runAdvertObserversTicker(ctx context.Context, interval time.Duration) { + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-ctx.Done(): + return + case <-t.C: + } + if !s.advertObs.ready.Load() { + continue + } + start := time.Now() + n, err := s.catchUpAdvertObservers(ctx, nodeAdvertObserversYield) + if err != nil { + log.Printf("[node-advert-observers] tick error: %v", err) + } else if n > 0 { + log.Printf("[node-advert-observers] tick: %d pair updates in %s", + n, time.Since(start).Round(time.Millisecond)) + } + } +} + +// catchUpAdvertObservers folds every observation up to the current +// MAX(observations.id) into node_advert_observers, one bounded step at a +// time, and returns the number of pair upserts. +func (s *Store) catchUpAdvertObservers(ctx context.Context, yield time.Duration) (int, error) { + return s.catchUpAdvertObserversBatched(ctx, yield, nodeAdvertObserversBatch) +} + +func (s *Store) catchUpAdvertObserversBatched(ctx context.Context, yield time.Duration, batch int64) (int, error) { + var target int64 + if err := s.db.QueryRowContext(ctx, `SELECT COALESCE(MAX(id), 0) FROM observations`).Scan(&target); err != nil { + return 0, fmt.Errorf("read observations high-water mark: %w", err) + } + total := 0 + for { + n, more, err := s.stepAdvertObservers(ctx, target, batch) + total += n + if err != nil || !more { + return total, err + } + if yield > 0 { + select { + case <-ctx.Done(): + return total, ctx.Err() + case <-time.After(yield): + } + } + } +} + +// stepAdvertObservers folds the next batch of observations at or below +// target and reports whether more remain. The watermark advances by the +// whole batch even when it held no adverts. +func (s *Store) stepAdvertObservers(ctx context.Context, target, batch int64) (int, bool, error) { + st := &s.advertObs + st.mu.Lock() + defer st.mu.Unlock() + + if !st.loaded { + if err := s.db.QueryRowContext(ctx, + `SELECT COALESCE(MAX(last_obs_id), 0) FROM node_advert_observers`).Scan(&st.watermark); err != nil { + return 0, false, fmt.Errorf("read node_advert_observers watermark: %w", err) + } + st.loaded = true + } + if st.watermark >= target { + return 0, false, nil + } + hi := min(st.watermark+batch, target) + + start := time.Now() + n, err := s.foldAdvertObservers(ctx, st.watermark, hi) + if err != nil { + return 0, false, err + } + if d := time.Since(start); d > nodeAdvertObserversSlowStep { + log.Printf("[node-advert-observers] SLOW step: observations %d..%d took %s", st.watermark+1, hi, d.Round(time.Millisecond)) + } + st.watermark = hi + return n, hi < target, nil +} + +// foldAdvertObservers upserts the (node, observer) pairs found in ADVERT +// observations with lo < observations.id <= hi. Caller holds advertObs.mu. +func (s *Store) foldAdvertObservers(ctx context.Context, lo, hi int64) (int, error) { + // The read returns, and so releases the single connection, before + // WriterTx asks for it. + pairs, err := s.readAdvertObserverPairs(ctx, lo, hi) + if err != nil || len(pairs) == 0 { + return 0, err + } + return len(pairs), s.upsertAdvertObserverPairs(ctx, pairs) +} + +// readAdvertObserverPairs collects the newest first_seen and observation id +// per (node, observer) among ADVERT observations in (lo, hi]. +func (s *Store) readAdvertObserverPairs( + ctx context.Context, lo, hi int64, +) (map[advertObserverKey]advertObserverPair, error) { + // CROSS JOIN pins the join order to the observations rowid range, so the + // step's cost is bounded by the batch whatever the planner statistics + // say (#2058 changed the plan of the region query this table replaces). + // decoded_json is read only where from_pubkey is still NULL: those are + // legacy rows BackfillFromPubkey has not reached yet, and it will set + // from_pubkey to exactly extractPubkeyFromAdvertJSON(decoded_json). + rows, err := s.db.QueryContext(ctx, ` + SELECT o.id, o.observer_idx, t.first_seen, t.from_pubkey, + CASE WHEN t.from_pubkey IS NULL THEN t.decoded_json END + FROM observations o + CROSS JOIN transmissions t ON t.id = o.transmission_id + WHERE o.id > ? AND o.id <= ? + AND t.payload_type = ? + AND o.observer_idx IS NOT NULL`, lo, hi, payloadADVERT) + if err != nil { + return nil, fmt.Errorf("scan advert observations: %w", err) + } + defer rows.Close() + pairs := make(map[advertObserverKey]advertObserverPair) + for rows.Next() { + var obsID, observerIdx int64 + var firstSeen string + var fromPubkey, decodedJSON sql.NullString + if err := rows.Scan(&obsID, &observerIdx, &firstSeen, &fromPubkey, &decodedJSON); err != nil { + return nil, fmt.Errorf("scan advert observation: %w", err) + } + pk := fromPubkey.String + if !fromPubkey.Valid { + pk = extractPubkeyFromAdvertJSON(decodedJSON.String) + } + if pk == "" || firstSeen == "" { + continue + } + k := advertObserverKey{publicKey: pk, observerIdx: observerIdx} + p := pairs[k] + p.lastSeen = max(p.lastSeen, firstSeen) + p.lastObsID = max(p.lastObsID, obsID) + pairs[k] = p + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("scan advert observations: %w", err) + } + return pairs, nil +} + +// upsertAdvertObserverPairs writes pairs in one writer transaction, keeping +// each row's newest last_seen and last_obs_id. +func (s *Store) upsertAdvertObserverPairs(ctx context.Context, pairs map[advertObserverKey]advertObserverPair) error { + return s.WriterTx("node_advert_observers", func(tx *sql.Tx) error { + stmt, err := tx.PrepareContext(ctx, ` + INSERT INTO node_advert_observers (public_key, observer_idx, last_seen, last_obs_id) + VALUES (?, ?, ?, ?) + ON CONFLICT(public_key, observer_idx) DO UPDATE SET + last_seen = MAX(last_seen, excluded.last_seen), + last_obs_id = MAX(last_obs_id, excluded.last_obs_id)`) + if err != nil { + return fmt.Errorf("prepare upsert: %w", err) + } + defer stmt.Close() + for k, p := range pairs { + if _, err := stmt.ExecContext(ctx, k.publicKey, k.observerIdx, p.lastSeen, p.lastObsID); err != nil { + return fmt.Errorf("upsert node_advert_observers: %w", err) + } + } + return nil + }) +} + +// pruneAdvertObservers drops pairs whose newest ADVERT is older than cutoff, +// the same transmissions.first_seen cutoff PruneOldPackets has just applied. +// It holds advertObs.mu so a builder step cannot interleave and write a pair +// back from transmissions that are already gone. +func (s *Store) pruneAdvertObservers(cutoff string) (int64, error) { + s.advertObs.mu.Lock() + defer s.advertObs.mu.Unlock() + var n int64 + err := s.WriterTx("prune_node_advert_observers", func(tx *sql.Tx) error { + res, err := tx.Exec(`DELETE FROM node_advert_observers WHERE last_seen < ?`, cutoff) + if err != nil { + return fmt.Errorf("prune node_advert_observers: %w", err) + } + n, _ = res.RowsAffected() + return nil + }) + return n, err +} diff --git a/cmd/ingestor/node_advert_observers_test.go b/cmd/ingestor/node_advert_observers_test.go new file mode 100644 index 000000000..7abdd1234 --- /dev/null +++ b/cmd/ingestor/node_advert_observers_test.go @@ -0,0 +1,617 @@ +package main + +import ( + "context" + "fmt" + "sort" + "strings" + "sync" + "testing" + "time" + + "github.com/meshcore-analyzer/dbschema" +) + +// advertFixture seeds observers and ADVERT traffic into a test store and +// records the rowids it created. +type advertFixture struct { + t *testing.T + s *Store + observers map[string]int64 // observer id -> rowid + nextHash int + nextPath int +} + +func newAdvertFixture(t *testing.T, s *Store) *advertFixture { + t.Helper() + return &advertFixture{t: t, s: s, observers: map[string]int64{}} +} + +func (f *advertFixture) exec(q string, args ...any) int64 { + f.t.Helper() + res, err := f.s.db.Exec(q, args...) + if err != nil { + f.t.Fatalf("%v\n%s", err, q) + } + id, err := res.LastInsertId() + if err != nil { + f.t.Fatal(err) + } + return id +} + +// observer adds an observer with the given IATA tag (stored verbatim, so +// padding and case exercise the region filter's normalisation). +func (f *advertFixture) observer(id string, iata any) int64 { + f.t.Helper() + f.exec(`INSERT INTO observers (id, name, iata) VALUES (?, ?, ?)`, id, id, iata) + var rowid int64 + if err := f.s.db.QueryRow(`SELECT rowid FROM observers WHERE id = ?`, id).Scan(&rowid); err != nil { + f.t.Fatal(err) + } + f.observers[id] = rowid + return rowid +} + +// node adds a nodes row so the legacy query, which filters nodes, sees it. +func (f *advertFixture) node(pk string) { + f.t.Helper() + f.exec(`INSERT OR IGNORE INTO nodes (public_key, name) VALUES (?, ?)`, pk, "n-"+pk) +} + +// tx adds a transmission. fromPubkey nil stores NULL, as on a legacy row +// BackfillFromPubkey has not reached; decoded carries the pubKey it will use. +func (f *advertFixture) tx(payloadType int, fromPubkey any, decoded, firstSeen string) int64 { + f.t.Helper() + f.nextHash++ + return f.exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json, from_pubkey) + VALUES ('', ?, ?, 1, ?, ?, ?)`, fmt.Sprintf("h%04d", f.nextHash), firstSeen, payloadType, decoded, fromPubkey) +} + +// obs adds an observation of txID by observer rowid (nil stores NULL). Each +// gets its own path, as a re-heard copy does, so the dedup index +// (transmission, observer, path) accepts repeats of the same pair. +func (f *advertFixture) obs(txID int64, observerIdx any) int64 { + f.t.Helper() + f.nextPath++ + return f.exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp) VALUES (?, ?, ?, ?)`, + txID, observerIdx, fmt.Sprintf(`["%04x"]`, f.nextPath), time.Now().Unix()) +} + +// advert adds an ADVERT by pk heard by the given observers. +func (f *advertFixture) advert(pk, firstSeen string, observerIDs ...string) int64 { + f.t.Helper() + f.node(pk) + txID := f.tx(int(payloadADVERT), pk, `{"pubKey":"`+pk+`"}`, firstSeen) + for _, o := range observerIDs { + f.obs(txID, f.observers[o]) + } + return txID +} + +func (f *advertFixture) query(q string, args ...any) []string { + f.t.Helper() + rows, err := f.s.db.Query(q, args...) + if err != nil { + f.t.Fatal(err) + } + defer rows.Close() + var out []string + for rows.Next() { + var pk string + if err := rows.Scan(&pk); err != nil { + f.t.Fatal(err) + } + out = append(out, pk) + } + if err := rows.Err(); err != nil { + f.t.Fatal(err) + } + sort.Strings(out) + return out +} + +func regionPlaceholders(codes []string) (string, []any) { + ph := make([]string, len(codes)) + args := make([]any, len(codes)) + for i, c := range codes { + ph[i] = "?" + args[i] = c + } + return strings.Join(ph, ","), args +} + +// legacy is the server's pre-#2101 region subquery, verbatim: the oracle. +func (f *advertFixture) legacy(codes ...string) []string { + f.t.Helper() + ph, args := regionPlaceholders(codes) + return f.query(`SELECT public_key FROM nodes WHERE public_key IN ( + SELECT DISTINCT t.from_pubkey + FROM transmissions t + JOIN observations o ON o.transmission_id = t.id + JOIN observers obs ON obs.rowid = o.observer_idx + WHERE t.payload_type = 4 + AND UPPER(TRIM(obs.iata)) IN (`+ph+`) + )`, args...) +} + +// fromTable is the server's #2101 region subquery over node_advert_observers. +func (f *advertFixture) fromTable(codes ...string) []string { + f.t.Helper() + ph, args := regionPlaceholders(codes) + return f.query(`SELECT public_key FROM nodes WHERE public_key IN ( + SELECT nao.public_key + FROM node_advert_observers nao + JOIN observers obs ON obs.rowid = nao.observer_idx + WHERE UPPER(TRIM(obs.iata)) IN (`+ph+`) + )`, args...) +} + +// assertParity compares table and oracle for each region set. wantNonEmpty +// guards against a vacuous pass on a fixture that matched nothing. +func (f *advertFixture) assertParity(wantNonEmpty bool, regionSets ...[]string) { + f.t.Helper() + for _, codes := range regionSets { + want := f.legacy(codes...) + if wantNonEmpty && len(want) == 0 { + f.t.Fatalf("region %v: legacy query matched nothing, the comparison would be vacuous", codes) + } + if got := f.fromTable(codes...); strings.Join(got, ",") != strings.Join(want, ",") { + f.t.Errorf("region %v: table %v, legacy %v", codes, got, want) + } + } +} + +func (f *advertFixture) pair(pk, observerID string) (lastSeen string, lastObsID int64, ok bool) { + f.t.Helper() + err := f.s.db.QueryRow(`SELECT last_seen, last_obs_id FROM node_advert_observers + WHERE public_key = ? AND observer_idx = ?`, pk, f.observers[observerID]).Scan(&lastSeen, &lastObsID) + if err != nil { + return "", 0, false + } + return lastSeen, lastObsID, true +} + +func catchUp(t *testing.T, s *Store) int { + t.Helper() + n, err := s.catchUpAdvertObservers(context.Background(), 0) + if err != nil { + t.Fatal(err) + } + return n +} + +const recent = "2026-09-30T12:00:00Z" + +// The table answers every region filter exactly as the legacy subquery did: +// padded and lower-case IATA tags, multi-region sets, nodes heard in two +// regions, and the rows the legacy query never matched (non-ADVERTs, empty +// from_pubkey, observations without an observer, untagged observers). +func TestNodeAdvertObserversMatchesLegacyQuery(t *testing.T) { + s := newTestStore(t) + f := newAdvertFixture(t, s) + f.observer("o-edi", "EDI") + f.observer("o-edi2", " edi ") + f.observer("o-gla", "GLA") + f.observer("o-none", nil) + + f.advert("aa01", recent, "o-edi") + f.advert("aa02", recent, "o-gla") + f.advert("aa03", recent, "o-edi2", "o-gla") + f.advert("aa04", recent, "o-none") + // Not adverts: never region members. + f.node("bb01") + f.obs(f.tx(5, "bb01", `{}`, recent), f.observers["o-edi"]) + // Advert whose pubkey could not be decoded: BackfillFromPubkey's "" sentinel. + f.node("cc01") + f.obs(f.tx(int(payloadADVERT), "", `{}`, recent), f.observers["o-edi"]) + // Advert observation with no observer. + f.node("dd01") + f.obs(f.tx(int(payloadADVERT), "dd01", `{"pubKey":"dd01"}`, recent), nil) + + catchUp(t, s) + // Codes arrive upper-cased (server normalizeRegionCodes); the stored + // tags' case and padding are what " edi " exercises. + f.assertParity(true, []string{"EDI"}, []string{"GLA"}, []string{"EDI", "GLA"}) + f.assertParity(false, []string{"AMS"}) +} + +// A legacy ADVERT row whose from_pubkey is still NULL is recorded under the +// pubkey BackfillFromPubkey will write, so once that backfill has run the +// table and the legacy query agree without the builder revisiting the row. +func TestNodeAdvertObserversNullFromPubkeyMatchesBackfill(t *testing.T) { + s := newTestStore(t) + f := newAdvertFixture(t, s) + f.observer("o-edi", "EDI") + f.node("ee01") + f.obs(f.tx(int(payloadADVERT), nil, `{"pubKey":"ee01"}`, recent), f.observers["o-edi"]) + + catchUp(t, s) + if got := f.fromTable("EDI"); strings.Join(got, ",") != "ee01" { + t.Fatalf("NULL from_pubkey advert should be recorded via decoded_json, got %v", got) + } + s.BackfillFromPubkey(100, 0, nil) + f.assertParity(true, []string{"EDI"}) +} + +// Steps advance through batches that contain no adverts at all; a watermark +// that only moved on a match would never leave the first such batch. +func TestNodeAdvertObserversBatchesAdvanceThroughGaps(t *testing.T) { + s := newTestStore(t) + f := newAdvertFixture(t, s) + f.observer("o-edi", "EDI") + for i := range 6 { + f.obs(f.tx(5, nil, `{}`, recent), f.observers["o-edi"]) // non-advert padding + f.advert(fmt.Sprintf("ff%02d", i), recent, "o-edi") + } + for range 5 { + f.obs(f.tx(5, nil, `{}`, recent), f.observers["o-edi"]) + } + + done := make(chan error, 1) + go func() { + _, err := s.catchUpAdvertObserversBatched(context.Background(), 0, 1) + done <- err + }() + select { + case err := <-done: + if err != nil { + t.Fatal(err) + } + case <-time.After(10 * time.Second): + t.Fatal("catch-up with batch 1 did not finish: the watermark is not advancing") + } + f.assertParity(true, []string{"EDI"}) + var maxObs int64 + if err := s.db.QueryRow(`SELECT MAX(id) FROM observations`).Scan(&maxObs); err != nil { + t.Fatal(err) + } + s.advertObs.mu.Lock() + wm := s.advertObs.watermark + s.advertObs.mu.Unlock() + if wm != maxObs { + t.Errorf("watermark %d, want MAX(observations.id) %d", wm, maxObs) + } +} + +// New observations are folded in on the next catch-up, and a pair keeps the +// newest transmission first_seen and observation id it has seen, whatever +// order the observations arrive in. +func TestNodeAdvertObserversDeltaKeepsNewest(t *testing.T) { + s := newTestStore(t) + f := newAdvertFixture(t, s) + f.observer("o-edi", "EDI") + f.observer("o-gla", "GLA") + newer := f.advert("ab01", "2026-09-30T12:00:00Z", "o-edi") + catchUp(t, s) + + // A late copy of an OLDER advert, folded in its own step: the upsert, not + // the in-step aggregation, must keep the newer last_seen, or prune would + // drop the pair while its newer advert is still retained. + older := f.advert("ab01", "2026-09-01T12:00:00Z") + lateOld := f.obs(older, f.observers["o-edi"]) + if n := catchUp(t, s); n == 0 { + t.Fatal("delta folded nothing") + } + lastSeen, lastObs, ok := f.pair("ab01", "o-edi") + if !ok { + t.Fatal("pair ab01/o-edi missing") + } + if lastSeen != "2026-09-30T12:00:00Z" { + t.Errorf("last_seen = %s after an older advert, want the newest advert's first_seen", lastSeen) + } + if lastObs != lateOld { + t.Errorf("last_obs_id = %d, want %d", lastObs, lateOld) + } + + lateID := f.obs(newer, f.observers["o-edi"]) + f.advert("ab02", recent, "o-gla") + catchUp(t, s) + if _, lastObs, _ = f.pair("ab01", "o-edi"); lastObs != lateID { + t.Errorf("last_obs_id = %d, want %d", lastObs, lateID) + } + f.assertParity(true, []string{"EDI"}, []string{"GLA"}) + if n := catchUp(t, s); n != 0 { + t.Errorf("caught-up builder folded %d pairs, want 0", n) + } +} + +// After a restart the builder resumes from the table rather than rescanning +// the whole history. +func TestNodeAdvertObserversResumesFromTable(t *testing.T) { + s := newTestStore(t) + f := newAdvertFixture(t, s) + f.observer("o-edi", "EDI") + f.advert("ac01", recent, "o-edi") + catchUp(t, s) + _, lastObs, _ := f.pair("ac01", "o-edi") + + s.advertObs.mu.Lock() + s.advertObs.loaded, s.advertObs.watermark = false, 0 // as on a fresh process + s.advertObs.mu.Unlock() + if _, _, err := s.stepAdvertObservers(context.Background(), lastObs, nodeAdvertObserversBatch); err != nil { + t.Fatal(err) + } + s.advertObs.mu.Lock() + wm := s.advertObs.watermark + s.advertObs.mu.Unlock() + if wm != lastObs { + t.Errorf("resumed watermark %d, want MAX(last_obs_id) %d", wm, lastObs) + } +} + +// PruneOldPackets drops a pair once every advert behind it is pruned, and +// keeps one with any advert left, so the table tracks the legacy query +// across retention. +func TestNodeAdvertObserversPrunedWithPackets(t *testing.T) { + s := newTestStore(t) + f := newAdvertFixture(t, s) + f.observer("o-edi", "EDI") + old := time.Now().UTC().AddDate(0, 0, -30).Format(time.RFC3339) + fresh := time.Now().UTC().Format(time.RFC3339) + f.advert("ad01", old, "o-edi") + f.advert("ad02", old, "o-edi") + f.advert("ad02", fresh, "o-edi") + f.advert("ad03", fresh, "o-edi") + catchUp(t, s) + + if _, err := s.PruneOldPackets(7); err != nil { + t.Fatal(err) + } + if _, _, ok := f.pair("ad01", "o-edi"); ok { + t.Error("ad01's only advert was pruned, its pair should be gone") + } + if _, _, ok := f.pair("ad02", "o-edi"); !ok { + t.Error("ad02 still has a fresh advert, its pair should stay") + } + f.assertParity(true, []string{"EDI"}) +} + +// The builder's backfill runs as an async migration: it is recorded done, +// the table is complete, and a later start on the same database does not +// run it again but resumes folding on the ticker. +func TestNodeAdvertObserversBuilderBackfillAndTick(t *testing.T) { + s := newTestStore(t) + f := newAdvertFixture(t, s) + f.observer("o-edi", "EDI") + f.advert("ae01", recent, "o-edi") + + stop := s.StartNodeAdvertObserversBuilder(10 * time.Millisecond) + t.Cleanup(stop) + s.WaitForAsyncMigrations() + if st, err := s.AsyncMigrationStatus(dbschema.NodeAdvertObserversBackfill); err != nil || st != "done" { + t.Fatalf("backfill status %q (%v), want done", st, err) + } + if !s.advertObs.ready.Load() { + t.Fatal("builder not ready after its backfill") + } + f.assertParity(true, []string{"EDI"}) + + f.advert("ae02", recent, "o-edi") + waitUntil(t, func() bool { _, _, ok := f.pair("ae02", "o-edi"); return ok }) + stop() + + // A second start (a restart) finds the migration done and is ready at once. + s.advertObs.ready.Store(false) + stop2 := s.StartNodeAdvertObserversBuilder(10 * time.Millisecond) + t.Cleanup(stop2) + if !s.advertObs.ready.Load() { + t.Error("restart with a done backfill should be ready immediately") + } +} + +// A backfill that fails is recorded failed and leaves the builder not ready, +// so the server keeps its legacy query rather than reading a partial table. +func TestNodeAdvertObserversBackfillFailureNotReady(t *testing.T) { + s := newTestStore(t) + if _, err := s.db.Exec(`DROP TABLE node_advert_observers`); err != nil { + t.Fatal(err) + } + stop := s.StartNodeAdvertObserversBuilder(time.Hour) + t.Cleanup(stop) + s.WaitForAsyncMigrations() + if st, _ := s.AsyncMigrationStatus(dbschema.NodeAdvertObserversBackfill); st != "failed" { + t.Errorf("backfill status %q, want failed", st) + } + if s.advertObs.ready.Load() { + t.Error("builder must not be ready after a failed backfill") + } +} + +// Builder steps, packet pruning and live inserts run together without a +// race (go test -race) or a deadlock on the single connection, and the +// table still agrees with the oracle afterwards. +func TestNodeAdvertObserversConcurrentWithPruneAndIngest(t *testing.T) { + s := newTestStore(t) + f := newAdvertFixture(t, s) + f.observer("o-edi", "EDI") + old := time.Now().UTC().AddDate(0, 0, -30).Format(time.RFC3339) + for i := range 20 { + f.advert(fmt.Sprintf("af%02d", i), old, "o-edi") + } + + var wg sync.WaitGroup + errs := make(chan error, 3) + wg.Add(3) + go func() { + defer wg.Done() + for range 20 { + if _, err := s.catchUpAdvertObserversBatched(context.Background(), 0, 3); err != nil { + errs <- err + return + } + } + }() + go func() { + defer wg.Done() + for range 5 { + if _, err := s.PruneOldPackets(7); err != nil { + errs <- err + return + } + } + }() + go func() { + defer wg.Done() + for i := range 20 { + if _, err := s.db.Exec(`INSERT INTO nodes (public_key, name) VALUES (?, 'n')`, fmt.Sprintf("ag%02d", i)); err != nil { + errs <- err + return + } + res, err := s.db.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json, from_pubkey) + VALUES ('', ?, ?, 1, 4, '{}', ?)`, fmt.Sprintf("hag%02d", i), time.Now().UTC().Format(time.RFC3339), fmt.Sprintf("ag%02d", i)) + if err != nil { + errs <- err + return + } + txID, _ := res.LastInsertId() + if _, err := s.db.Exec(`INSERT INTO observations (transmission_id, observer_idx, timestamp) VALUES (?, ?, ?)`, + txID, f.observers["o-edi"], time.Now().Unix()); err != nil { + errs <- err + return + } + } + }() + finished := make(chan struct{}) + go func() { wg.Wait(); close(finished) }() + select { + case <-finished: + case <-time.After(30 * time.Second): + t.Fatal("builder, prune and ingest deadlocked") + } + close(errs) + for err := range errs { + t.Fatal(err) + } + + // A final prune and catch-up settle the table; it must match the oracle. + if _, err := s.PruneOldPackets(7); err != nil { + t.Fatal(err) + } + catchUp(t, s) + f.assertParity(true, []string{"EDI"}) +} + +func waitUntil(t *testing.T, cond func() bool) { + t.Helper() + deadline := time.Now().Add(10 * time.Second) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(10 * time.Millisecond) + } + t.Fatal("condition not met within 10s") +} + +func advertWatermark(s *Store) int64 { + s.advertObs.mu.Lock() + defer s.advertObs.mu.Unlock() + return s.advertObs.watermark +} + +// A step whose write fails must not advance the watermark, or the +// observations it covered would never be folded in. +func TestNodeAdvertObserversFailedWriteKeepsWatermark(t *testing.T) { + s := newTestStore(t) + f := newAdvertFixture(t, s) + f.observer("o-edi", "EDI") + f.advert("ba01", recent, "o-edi") + catchUp(t, s) + before := advertWatermark(s) + + f.advert("ba02", recent, "o-edi") + f.exec(`ALTER TABLE node_advert_observers RENAME TO node_advert_observers_gone`) + _, err := s.catchUpAdvertObservers(context.Background(), 0) + if err == nil || !strings.Contains(err.Error(), "prepare upsert") { + t.Fatalf("expected the upsert to fail, got %v", err) + } + if got := advertWatermark(s); got != before { + t.Fatalf("watermark moved %d -> %d past observations that were not written", before, got) + } + + f.exec(`ALTER TABLE node_advert_observers_gone RENAME TO node_advert_observers`) + catchUp(t, s) + if _, _, ok := f.pair("ba02", "o-edi"); !ok { + t.Error("observations behind a failed step were not folded in on the retry") + } +} + +// Cancelling a multi-step catch-up stops between steps; the steps already +// written stand and the watermark covers exactly them. +func TestNodeAdvertObserversCatchUpCancelled(t *testing.T) { + s := newTestStore(t) + f := newAdvertFixture(t, s) + f.observer("o-edi", "EDI") + for i := range 5 { + f.advert(fmt.Sprintf("bb%02d", i), recent, "o-edi") + } + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + _, err := s.catchUpAdvertObserversBatched(ctx, time.Hour, 1) // yields between steps + done <- err + }() + waitUntil(t, func() bool { return advertWatermark(s) >= 1 }) + cancel() + select { + case err := <-done: + if err == nil { + t.Fatal("cancelled catch-up returned no error") + } + case <-time.After(10 * time.Second): + t.Fatal("cancelled catch-up did not return") + } + var lastObs int64 + if err := s.db.QueryRow(`SELECT COALESCE(MAX(last_obs_id), 0) FROM node_advert_observers`).Scan(&lastObs); err != nil { + t.Fatal(err) + } + if wm := advertWatermark(s); wm != 1 || lastObs != 1 { + t.Errorf("after one step: watermark %d, MAX(last_obs_id) %d; want both 1", wm, lastObs) + } +} + +// Ticks are skipped until the backfill is done, and a failing tick is +// logged and survived: the next tick folds normally. +func TestNodeAdvertObserversTickSkipsAndSurvivesErrors(t *testing.T) { + s := newTestStore(t) + f := newAdvertFixture(t, s) + f.observer("o-edi", "EDI") + f.advert("bc01", recent, "o-edi") + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { defer close(done); s.runAdvertObserversTicker(ctx, 5*time.Millisecond) }() + t.Cleanup(func() { cancel(); <-done }) + + time.Sleep(50 * time.Millisecond) // several ticks while not ready + if _, _, ok := f.pair("bc01", "o-edi"); ok { + t.Fatal("ticker folded before the backfill was done") + } + + f.exec(`ALTER TABLE node_advert_observers RENAME TO node_advert_observers_gone`) + s.advertObs.ready.Store(true) + time.Sleep(50 * time.Millisecond) // several failing ticks + f.exec(`ALTER TABLE node_advert_observers_gone RENAME TO node_advert_observers`) + waitUntil(t, func() bool { _, _, ok := f.pair("bc01", "o-edi"); return ok }) +} + +// PruneOldPackets reports a failure to prune the pairs table. +func TestNodeAdvertObserversPruneError(t *testing.T) { + s := newTestStore(t) + f := newAdvertFixture(t, s) + f.exec(`DROP TABLE node_advert_observers`) + if _, err := s.PruneOldPackets(7); err == nil || !strings.Contains(err.Error(), "node_advert_observers") { + t.Fatalf("expected the pairs prune to fail, got %v", err) + } +} + +// A non-positive interval falls back to the default, as the neighbor +// builder does. +func TestNodeAdvertObserversDefaultInterval(t *testing.T) { + s := newTestStore(t) + stop := s.StartNodeAdvertObserversBuilder(0) + s.WaitForAsyncMigrations() + stop() + stop() // idempotent +} diff --git a/cmd/server/db.go b/cmd/server/db.go index e6f1a3dd3..1c2693b5c 100644 --- a/cmd/server/db.go +++ b/cmd/server/db.go @@ -55,6 +55,13 @@ type DB struct { // startup probe can lose the race. See declaredRegionsTablePresent. declaredRegionsTableLate atomic.Bool + // node_advert_observers readiness for the /api/nodes region filter + // (#2101): latched once the ingestor's backfill is done, re-checked at + // most every nodeAdvertObserversRecheck until then. See + // node_advert_observers.go. + nodeAdvertObserversLatched atomic.Bool + nodeAdvertObserversCheckedAt atomic.Int64 // unix nanos of the last check + // Channel list caches, keyed by region param — avoids repeated GROUP BY // scans (#762). Keyed per-region (not a single slot) so mixed-region // traffic doesn't evict and re-run the query on every request. @@ -1139,21 +1146,28 @@ func (db *DB) GetNodes(limit, offset int, role, search, before, lastHeard, sortB placeholders[i] = "?" regionArgs[i] = c } - joinCond := "obs.rowid = o.observer_idx" - if !db.isV3 { - joinCond = "obs.id = o.observer_id" + var subq string + if db.nodeAdvertObserversReady() { + // #2101: membership from the ingestor-maintained pairs table, + // a lookup instead of a scan of the whole advert history. + subq = nodeAdvertObserversRegionFilter(placeholders) + } else { + joinCond := "obs.rowid = o.observer_idx" + if !db.isV3 { + joinCond = "obs.id = o.observer_id" + } + // #1143: from_pubkey is a dedicated, indexed column populated at + // ingest (and backfilled) for ADVERT rows specifically so pubkey + // lookups don't need to JSON_EXTRACT + parse decoded_json per row. + subq = fmt.Sprintf(`public_key IN ( + SELECT DISTINCT t.from_pubkey + FROM transmissions t + JOIN observations o ON o.transmission_id = t.id + JOIN observers obs ON %s + WHERE t.payload_type = 4 + AND UPPER(TRIM(obs.iata)) IN (%s) + )`, joinCond, strings.Join(placeholders, ",")) } - // #1143: from_pubkey is a dedicated, indexed column populated at - // ingest (and backfilled) for ADVERT rows specifically so pubkey - // lookups don't need to JSON_EXTRACT + parse decoded_json per row. - subq := fmt.Sprintf(`public_key IN ( - SELECT DISTINCT t.from_pubkey - FROM transmissions t - JOIN observations o ON o.transmission_id = t.id - JOIN observers obs ON %s - WHERE t.payload_type = 4 - AND UPPER(TRIM(obs.iata)) IN (%s) - )`, joinCond, strings.Join(placeholders, ",")) where = append(where, subq) args = append(args, regionArgs...) } diff --git a/cmd/server/node_advert_observers.go b/cmd/server/node_advert_observers.go new file mode 100644 index 000000000..6277a9289 --- /dev/null +++ b/cmd/server/node_advert_observers.go @@ -0,0 +1,70 @@ +package main + +import ( + "fmt" + "log" + "strings" + "time" + + "github.com/meshcore-analyzer/dbschema" +) + +// /api/nodes?region= reads node membership from node_advert_observers +// (#2101), the (node, observer) pairs the ingestor derives from ADVERT +// observations (cmd/ingestor/node_advert_observers.go). The legacy filter +// re-derived the same set with a transmissions ⋈ observations ⋈ observers +// scan over the whole advert history, uncached and evaluated twice per +// request (COUNT and page): ~10 s per evaluation on a 1.4 GB database. +// +// The table is only trusted once the ingestor's backfill has completed, so a +// partly built table is never read. Until then, and on a v2 schema the +// ingestor no longer writes, GetNodes keeps the legacy query. + +// nodeAdvertObserversRecheck bounds how often a server that started before +// the backfill finished looks again, so it switches over without a restart +// and without a query per request. +const nodeAdvertObserversRecheck = 30 * time.Second + +// nodeAdvertObserversReady reports whether the region filter may read +// node_advert_observers. Once true it stays true for the process. +func (db *DB) nodeAdvertObserversReady() bool { + if !db.isV3 { + return false + } + if db.nodeAdvertObserversLatched.Load() { + return true + } + now := time.Now().UnixNano() + last := db.nodeAdvertObserversCheckedAt.Load() + if last != 0 && now-last < int64(nodeAdvertObserversRecheck) { + return false + } + // One caller per interval checks; the rest keep the legacy query. + if !db.nodeAdvertObserversCheckedAt.CompareAndSwap(last, now) { + return false + } + var status string + err := db.conn.QueryRow(`SELECT status FROM _async_migrations WHERE name = ?`, + dbschema.NodeAdvertObserversBackfill).Scan(&status) + if err != nil || status != "done" { + // No row, no table (an ingestor that predates #2101) or not done yet. + return false + } + if db.nodeAdvertObserversLatched.CompareAndSwap(false, true) { + log.Printf("[nodes] region filter now reads node_advert_observers (#2101)") + } + return true +} + +// nodeAdvertObserversRegionFilter is the GetNodes WHERE clause for the region +// codes bound to placeholders: nodes with an ADVERT heard by an observer +// whose IATA tag is one of them. The tag is joined at query time, as in the +// legacy filter, so it follows an observer that changes region. +func nodeAdvertObserversRegionFilter(placeholders []string) string { + return fmt.Sprintf(`public_key IN ( + SELECT nao.public_key + FROM node_advert_observers nao + JOIN observers obs ON obs.rowid = nao.observer_idx + WHERE UPPER(TRIM(obs.iata)) IN (%s) + )`, strings.Join(placeholders, ",")) +} diff --git a/cmd/server/node_advert_observers_test.go b/cmd/server/node_advert_observers_test.go new file mode 100644 index 000000000..45364edec --- /dev/null +++ b/cmd/server/node_advert_observers_test.go @@ -0,0 +1,299 @@ +package main + +import ( + "fmt" + "sort" + "strings" + "sync" + "testing" + + "github.com/meshcore-analyzer/dbschema" +) + +const ( + regionSJC = "SJC" // seedTestData: obs1 + regionSFO = "SFO" // seedTestData: obs2 +) + +// legacyRegionPubkeys runs the pre-#2101 region subquery directly: the +// oracle the table-backed filter must agree with. +func legacyRegionPubkeys(t *testing.T, db *DB, region string) []string { + t.Helper() + codes := normalizeRegionCodes(region) + if len(codes) == 0 { + return nil + } + ph := make([]string, len(codes)) + args := make([]any, len(codes)) + for i, c := range codes { + ph[i] = "?" + args[i] = c + } + q := fmt.Sprintf(`SELECT public_key FROM nodes WHERE public_key IN ( + SELECT DISTINCT t.from_pubkey + FROM transmissions t + JOIN observations o ON o.transmission_id = t.id + JOIN observers obs ON obs.rowid = o.observer_idx + WHERE t.payload_type = 4 + AND UPPER(TRIM(obs.iata)) IN (%s) + )`, strings.Join(ph, ",")) + return queryPubkeys(t, db, q, args...) +} + +func queryPubkeys(t *testing.T, db *DB, q string, args ...any) []string { + t.Helper() + rows, err := db.conn.Query(q, args...) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var out []string + for rows.Next() { + var pk string + if err := rows.Scan(&pk); err != nil { + t.Fatal(err) + } + out = append(out, pk) + } + if err := rows.Err(); err != nil { + t.Fatal(err) + } + sort.Strings(out) + return out +} + +func regionNodePubkeys(t *testing.T, db *DB, region string) []string { + t.Helper() + nodes, total, _, err := db.GetNodes(500, 0, "", "", "", "", "", region) + if err != nil { + t.Fatal(err) + } + if total != len(nodes) { + t.Fatalf("region %q: total %d != page length %d", region, total, len(nodes)) + } + out := make([]string, 0, len(nodes)) + for _, n := range nodes { + pk, ok := n["public_key"].(string) + if !ok { + t.Fatalf("region %q: public_key %v is not a string", region, n["public_key"]) + } + out = append(out, pk) + } + sort.Strings(out) + return out +} + +// regionTestDB is seedTestData under the real schema, with +// node_advert_observers filled from the advert history the way the ingestor +// builds it (its builder is checked against the same oracle in +// cmd/ingestor/node_advert_observers_test.go). +func regionTestDB(t *testing.T) *DB { + t.Helper() + db := setupTestDB(t) + t.Cleanup(func() { db.Close() }) + seedTestData(t, db) + if err := dbschema.Apply(db.conn, nil); err != nil { + t.Fatal(err) + } + mustExec(t, db, `INSERT INTO node_advert_observers (public_key, observer_idx, last_seen, last_obs_id) + SELECT t.from_pubkey, o.observer_idx, MAX(t.first_seen), MAX(o.id) + FROM observations o JOIN transmissions t ON t.id = o.transmission_id + WHERE t.payload_type = 4 AND t.from_pubkey IS NOT NULL AND t.from_pubkey != '' + AND o.observer_idx IS NOT NULL + GROUP BY t.from_pubkey, o.observer_idx`) + mustExec(t, db, `CREATE TABLE IF NOT EXISTS _async_migrations ( + name TEXT PRIMARY KEY, status TEXT NOT NULL, started_at TEXT, ended_at TEXT, error TEXT)`) + return db +} + +func setBackfillStatus(t *testing.T, db *DB, status string) { + t.Helper() + mustExec(t, db, `INSERT INTO _async_migrations (name, status) VALUES (?, ?) + ON CONFLICT(name) DO UPDATE SET status = excluded.status`, dbschema.NodeAdvertObserversBackfill, status) + db.nodeAdvertObserversCheckedAt.Store(0) // don't wait out the re-check interval +} + +// addPhantomPair records a pair the advert history does not contain, so a +// result that includes it proves the filter read the table. +func addPhantomPair(t *testing.T, db *DB) string { + t.Helper() + const pk = "1122334455667788" // seedTestData's TestRoom: no ADVERT in SFO + mustExec(t, db, `INSERT INTO node_advert_observers (public_key, observer_idx, last_seen, last_obs_id) + VALUES (?, 2, '2026-01-01T00:00:00Z', 1)`, pk) + return pk +} + +func TestRegionFilterMatchesLegacyFromTable(t *testing.T) { + db := regionTestDB(t) + setBackfillStatus(t, db, "done") + + for _, region := range []string{regionSJC, regionSFO, "SJC,SFO", "sfo , sjc", "SJC,SJC"} { + want := legacyRegionPubkeys(t, db, region) + if len(want) == 0 { + t.Fatalf("region %q: fixture matched nothing, the comparison would be vacuous", region) + } + if got := regionNodePubkeys(t, db, region); strings.Join(got, ",") != strings.Join(want, ",") { + t.Errorf("region %q: table %v, legacy %v", region, got, want) + } + } + if got := regionNodePubkeys(t, db, "AMS"); len(got) != 0 { + t.Errorf("unknown region should match nothing, got %v", got) + } + if !db.nodeAdvertObserversLatched.Load() { + t.Fatal("expected the table path to be latched") + } + phantom := addPhantomPair(t, db) + if got := regionNodePubkeys(t, db, regionSFO); !contains(got, phantom) { + t.Errorf("filter did not read node_advert_observers: %v lacks %s", got, phantom) + } +} + +// Until the backfill is done the table may be partial, so it is not read. +func TestRegionFilterLegacyUntilBackfillDone(t *testing.T) { + for _, status := range []string{"", "pending_async", "failed"} { + t.Run("status="+status, func(t *testing.T) { + db := regionTestDB(t) + if status != "" { + setBackfillStatus(t, db, status) + } + phantom := addPhantomPair(t, db) + got := regionNodePubkeys(t, db, regionSFO) + if contains(got, phantom) { + t.Errorf("table read before its backfill was done: %v", got) + } + if want := legacyRegionPubkeys(t, db, regionSFO); strings.Join(got, ",") != strings.Join(want, ",") { + t.Errorf("legacy path: got %v, want %v", got, want) + } + }) + } +} + +// An ingestor that predates #2101 has no _async_migrations row and may have +// no table at all; the server keeps the legacy query. +func TestRegionFilterLegacyWithoutAsyncMigrationsTable(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + seedTestData(t, db) + if db.nodeAdvertObserversReady() { + t.Fatal("ready without an _async_migrations table") + } + if got, want := regionNodePubkeys(t, db, regionSJC), legacyRegionPubkeys(t, db, regionSJC); strings.Join(got, ",") != strings.Join(want, ",") { + t.Errorf("got %v, want %v", got, want) + } +} + +// A server that started before the backfill finished switches over within +// the re-check interval, without a restart, and then stays switched. +func TestRegionFilterSwitchesWithoutRestart(t *testing.T) { + db := regionTestDB(t) + setBackfillStatus(t, db, "pending_async") + phantom := addPhantomPair(t, db) + if contains(regionNodePubkeys(t, db, regionSFO), phantom) { + t.Fatal("table read while pending") + } + + // Done, but inside the re-check interval: still the legacy query. + mustExec(t, db, `UPDATE _async_migrations SET status = 'done' WHERE name = ?`, dbschema.NodeAdvertObserversBackfill) + if contains(regionNodePubkeys(t, db, regionSFO), phantom) { + t.Fatal("re-checked inside the interval") + } + + db.nodeAdvertObserversCheckedAt.Store(1) // the interval has passed + if !contains(regionNodePubkeys(t, db, regionSFO), phantom) { + t.Fatal("did not switch to the table once the backfill was done") + } + + // Latched: a later status change does not switch back. + mustExec(t, db, `UPDATE _async_migrations SET status = 'failed' WHERE name = ?`, dbschema.NodeAdvertObserversBackfill) + db.nodeAdvertObserversCheckedAt.Store(1) + if !contains(regionNodePubkeys(t, db, regionSFO), phantom) { + t.Error("switched back after latching") + } +} + +// The v2 schema keys observations by observer_id, which the table does not +// model; the ingestor no longer writes it, so the legacy query stays. +func TestRegionFilterV2SchemaStaysLegacy(t *testing.T) { + db := setupTestDBV2(t) + defer db.Close() + mustExec(t, db, `CREATE TABLE _async_migrations (name TEXT PRIMARY KEY, status TEXT NOT NULL)`) + mustExec(t, db, `INSERT INTO _async_migrations (name, status) VALUES (?, 'done')`, dbschema.NodeAdvertObserversBackfill) + if db.nodeAdvertObserversReady() { + t.Error("v2 schema must keep the legacy region query") + } +} + +// Readiness is looked up at most once per interval while not ready, from any +// number of concurrent requests (go test -race), and a concurrent switch is +// safe. +func TestRegionFilterReadinessCheckBounded(t *testing.T) { + db := regionTestDB(t) + setBackfillStatus(t, db, "pending_async") + if db.nodeAdvertObserversReady() { + t.Fatal("ready while pending") + } + checked := db.nodeAdvertObserversCheckedAt.Load() + for range 50 { + db.nodeAdvertObserversReady() + } + if db.nodeAdvertObserversCheckedAt.Load() != checked { + t.Error("re-checked inside the interval") + } + + var wg sync.WaitGroup + errs := make(chan error, 16) + for i := range 16 { + wg.Add(1) + go func() { + defer wg.Done() + if i == 8 { + // Not setBackfillStatus: t.Fatal must not run off the test goroutine. + if _, err := db.conn.Exec(`UPDATE _async_migrations SET status = 'done' WHERE name = ?`, + dbschema.NodeAdvertObserversBackfill); err != nil { + errs <- err + return + } + db.nodeAdvertObserversCheckedAt.Store(0) + } + if _, _, _, err := db.GetNodes(50, 0, "", "", "", "", "", regionSJC); err != nil { + errs <- err + } + }() + } + wg.Wait() + close(errs) + for err := range errs { + t.Fatal(err) + } +} + +// The table-backed filter never touches observations or transmissions: its +// cost is bounded by the pairs table, not the packet history (rule 0 proof). +func TestRegionFilterPlanAvoidsHistory(t *testing.T) { + db := regionTestDB(t) + q := `EXPLAIN QUERY PLAN SELECT public_key FROM nodes WHERE ` + + nodeAdvertObserversRegionFilter([]string{"?", "?"}) + rows, err := db.conn.Query(q, regionSJC, regionSFO) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var plan []string + for rows.Next() { + var id, parent, unused int + var detail string + if err := rows.Scan(&id, &parent, &unused, &detail); err != nil { + t.Fatal(err) + } + plan = append(plan, detail) + } + joined := strings.Join(plan, " | ") + if !strings.Contains(joined, "node_advert_observers") { + t.Errorf("plan does not read node_advert_observers: %s", joined) + } + for _, history := range []string{"observations", "transmissions"} { + if strings.Contains(joined, history) { + t.Errorf("plan touches %s: %s", history, joined) + } + } +} diff --git a/internal/dbschema/dbschema.go b/internal/dbschema/dbschema.go index 433ee8de0..5762aecbb 100644 --- a/internal/dbschema/dbschema.go +++ b/internal/dbschema/dbschema.go @@ -52,6 +52,10 @@ func Apply(rw *sql.DB, logf Logger) error { if err := ensureInactiveNodesTable(rw); err != nil { return fmt.Errorf("ensure inactive_nodes: %w", err) } + // PREFLIGHT: async=true reason="new, empty table + index: O(1); backfill is async in the ingestor" + if err := ensureNodeAdvertObserversTable(rw); err != nil { + return fmt.Errorf("ensure node_advert_observers: %w", err) + } if err := ensureResolvedPathColumn(rw, logf); err != nil { return fmt.Errorf("ensure resolved_path: %w", err) } @@ -139,6 +143,10 @@ func AssertReady(ro *sql.DB) error { } mustTable("neighbor_edges") + // #2101: which observers have heard each node's adverts. Ingestor + // builds it (cmd/ingestor/node_advert_observers.go); server reads it + // for the /api/nodes region filter once its backfill is done. + mustTable("node_advert_observers") mustCol("observations", "resolved_path") mustCol("observers", "inactive") mustCol("observers", "last_packet_at") @@ -321,6 +329,38 @@ func ensureNeighborEdgesTable(rw *sql.DB) error { return err } +// NodeAdvertObserversBackfill names the ingestor's one-time build of +// node_advert_observers in _async_migrations. The server reads that row and +// uses the table only once its status is `done`, so both sides share this +// name. +const NodeAdvertObserversBackfill = "node_advert_observers_backfill_v1" + +// ensureNodeAdvertObserversTable creates the derived (node, observer) table +// behind the /api/nodes region filter (#2101): one row per observer that +// has heard at least one of a node's ADVERTs. It replaces a +// transmissions ⋈ observations ⋈ observers scan over the whole history +// with a lookup in a table the size of nodes × observers. +// +// The ingestor is the only writer (cmd/ingestor/node_advert_observers.go). +// last_seen is the newest transmissions.first_seen among those adverts, so +// PruneOldPackets can drop a pair at the same cutoff that removes its last +// advert. last_obs_id is the newest observations.id folded in; its MAX is +// the builder's resume point. +func ensureNodeAdvertObserversTable(rw *sql.DB) error { + if _, err := rw.Exec(`CREATE TABLE IF NOT EXISTS node_advert_observers ( + public_key TEXT NOT NULL, + observer_idx INTEGER NOT NULL, + last_seen TEXT NOT NULL, + last_obs_id INTEGER NOT NULL, + PRIMARY KEY (public_key, observer_idx) + ) WITHOUT ROWID`); err != nil { + return err + } + _, err := rw.Exec(`CREATE INDEX IF NOT EXISTS idx_node_advert_observers_observer + ON node_advert_observers(observer_idx)`) + return err +} + // ensureInactiveNodesTable creates the inactive_nodes table if missing. // The ingestor's applySchema also creates this table — duplicating it // here makes dbschema.Apply self-sufficient when called against a diff --git a/internal/dbschema/dbschema_test.go b/internal/dbschema/dbschema_test.go index 5145777b5..4d635e85b 100644 --- a/internal/dbschema/dbschema_test.go +++ b/internal/dbschema/dbschema_test.go @@ -253,3 +253,44 @@ func TestPartialIdxTxLastSeenZero_FullIndexDropped(t *testing.T) { t.Fatalf("legacy idx_tx_last_seen must be dropped after partial index is in place (#1740); still present as %q", legacyName) } } + +// TestNodeAdvertObserversTable_AppliedAndAsserted pins #2101's derived table: +// Apply creates it (with its observer index, which the region filter's join +// uses), and AssertReady refuses a DB without it, so a server can never start +// against a schema whose region filter would silently fall back for ever. +func TestNodeAdvertObserversTable_AppliedAndAsserted(t *testing.T) { + db := minimalDB(t) + defer db.Close() + + if err := Apply(db, nil); err != nil { + t.Fatalf("Apply: %v", err) + } + for _, c := range []string{"public_key", "observer_idx", "last_seen", "last_obs_id"} { + has, err := TableHasColumn(db, "node_advert_observers", c) + if err != nil { + t.Fatalf("probe node_advert_observers.%s: %v", c, err) + } + if !has { + t.Errorf("after Apply: node_advert_observers.%s missing", c) + } + } + var idx int + if err := db.QueryRow(`SELECT COUNT(*) FROM sqlite_master WHERE type = 'index' + AND name = 'idx_node_advert_observers_observer'`).Scan(&idx); err != nil { + t.Fatal(err) + } + if idx != 1 { + t.Error("after Apply: idx_node_advert_observers_observer missing") + } + if err := AssertReady(db); err != nil { + t.Fatalf("AssertReady after Apply: %v", err) + } + + if _, err := db.Exec(`DROP TABLE node_advert_observers`); err != nil { + t.Fatal(err) + } + err := AssertReady(db) + if err == nil || !strings.Contains(err.Error(), "table:node_advert_observers") { + t.Fatalf("AssertReady should name the missing node_advert_observers table, got: %v", err) + } +}