From 3e21b179ecccae005aa5f6f45bc095353c94b343 Mon Sep 17 00:00:00 2001 From: efiten Date: Fri, 9 Oct 2026 21:20:34 +0200 Subject: [PATCH] fix(server): drive the reach scan from the observations time window (#2155) Refs #2146 (item: `/api/nodes/:pubkey/reach`). ## Situation The reach scan joins `observations` to `transmissions` and filters on `o.timestamp >= ?` plus one `o.path_json LIKE '%"XX"%'` per token. With the planner statistics `ANALYZE` produces on a large database, SQLite runs it as `SCAN t`: every transmission, with an index probe into `observations` (`idx_observations_tx_ts`) per row. The cost is then the size of `transmissions`, whatever node is asked for and whatever the window. ## My change in this PR `scanReachRows` uses `CROSS JOIN` for the transmissions join. In SQLite that fixes the join order: `observations` is the outer loop, read through `idx_observations_timestamp` for the window, and each row looks up its transmission by primary key. Results are the same rows; it is still an inner join. The query text moves into `reachScanSQL(nTokens)`, so a test can inspect the plan. ## Measurements Read-only on a database with 18.3M observations and 1.5M transmissions (Python `sqlite3` 3.45.1 on the host), 7-day window: | node (rows returned) | `JOIN`, first run | `JOIN`, warm | `CROSS JOIN`, both runs | |---|---|---|---| | busiest relay (54,387) | 10.29 s | 3.27 s | 1.95 s | | quartile (3,913) | 3.16 s | 3.15 s | 1.81 s | | median (1,310) | 3.17 s | 3.18 s | 1.82 s | | a repeater reported as timing out in the browser (34,209) | 14.11 s | 3.25 s | 1.93 s | On that instance `/api/nodes/:pubkey/reach` reports p95 188.6 s and max 192.6 s over 34 calls in `/api/perf`; a single uncached call from the host took 3.4 to 3.6 s, so most of that is waiting for the 2 build slots and the read pool. On a staging instance (smaller database), with unique cache keys and `days` 6 or 7, one cycle per build after the ingestor's startup `ANALYZE` had finished: | | without this PR | with this PR | |---|---|---| | sequential, 12 calls, p50 | 3402-3801 ms | 2078-2467 ms | | 2 concurrent clients (= `reachMaxConcurrentBuilds`), p50 / p95 | 3575-4087 / 4049-5291 ms | 2271-2462 / 2814-2928 ms | | calls completed in 120 s | 58-68 | 96-101 | The staging builds also contain #2147. ## Verification - `TestReachScanStartsFromObservationsTimeIndex` creates the two production indexes, loads the `sqlite_stat1` values from the 18.3M-observation database, and checks that the plan starts with `SEARCH o USING INDEX idx_observations_timestamp`. With the old `JOIN` it fails with the production plan (`SCAN t`, then `SEARCH o USING INDEX idx_observations_tx_ts`). - The existing reach tests (`TestScanReachRows_DecodesRows`, `TestScanReachRows_CapTruncates`, attribution tests) pass unchanged. - `cmd/server`: `go test ./...` green. ## Not in this PR - The scan still reads every observation in the window and applies the `LIKE` per row, so its cost grows with the window and with traffic. Removing the scan needs an index on hop tokens (for example a table the ingestor maintains). I have not measured what that would cost in storage. - The build-slot limit (2) and the 5-minute cache are unchanged. Co-authored-by: Claude Opus 5.5 --- cmd/server/node_reach.go | 30 ++++++++++++++------- cmd/server/node_reach_test.go | 50 +++++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 9 deletions(-) diff --git a/cmd/server/node_reach.go b/cmd/server/node_reach.go index 0cfa9789..3ec2567f 100644 --- a/cmd/server/node_reach.go +++ b/cmd/server/node_reach.go @@ -735,6 +735,26 @@ func (s *Server) getDegreeSnapshot(ctx context.Context) *degreeSnapshot { return snap } +// reachScanSQL is the reach scan for nTokens path tokens. Arguments: the +// window start (epoch seconds), one LIKE pattern per token, the row limit. +// +// CROSS JOIN fixes the join order in SQLite: observations is the outer loop, +// read through idx_observations_timestamp for the window. With a plain JOIN +// the planner scanned every transmission and probed observations per row +// (3.2-3.3 s against 1.8-2.0 s on an 18M-observation database, #2146). +func reachScanSQL(nTokens int) string { + likes := make([]string, nTokens) + for i := range likes { + likes[i] = "o.path_json LIKE ?" + } + return `SELECT LOWER(COALESCE(obs.id,'')), LOWER(COALESCE(t.from_pubkey,'')), COALESCE(t.payload_type,0), o.path_json, o.snr + FROM observations o + CROSS JOIN transmissions t ON t.id = o.transmission_id + LEFT JOIN observers obs ON obs.rowid = o.observer_idx + WHERE o.timestamp >= ? AND (` + strings.Join(likes, " OR ") + `) + LIMIT ?` +} + // scanReachRows reads windowed observations whose path contains any reliable // token, with the originator + observer + snr needed for attribution. Observer // id and originator pubkey are lowercased in SQL (not per row), the path slice @@ -749,7 +769,6 @@ func (s *Server) scanReachRows(ctx context.Context, tokens map[string]bool, sinc if len(tokens) == 0 { return nil, nil // defensive: an empty LIKE chain would render `AND ()` (SQL error) } - likes := make([]string, 0, len(tokens)) args := []interface{}{sinceEpoch} // Sort tokens so the generated SQL text is byte-stable across requests // with the same token set — preserves the driver's prepared-statement @@ -760,17 +779,10 @@ func (s *Server) scanReachRows(ctx context.Context, tokens map[string]bool, sinc } sort.Strings(toks) for _, tok := range toks { - likes = append(likes, "o.path_json LIKE ?") args = append(args, "%\""+tok+"\"%") } - q := `SELECT LOWER(COALESCE(obs.id,'')), LOWER(COALESCE(t.from_pubkey,'')), COALESCE(t.payload_type,0), o.path_json, o.snr - FROM observations o - JOIN transmissions t ON t.id = o.transmission_id - LEFT JOIN observers obs ON obs.rowid = o.observer_idx - WHERE o.timestamp >= ? AND (` + strings.Join(likes, " OR ") + `) - LIMIT ?` args = append(args, reachScanRowLimit) - rows, err := s.db.conn.QueryContext(ctx, q, args...) + rows, err := s.db.conn.QueryContext(ctx, reachScanSQL(len(toks)), args...) if err != nil { log.Printf("[reach] scan query failed: %v", err) return nil, err diff --git a/cmd/server/node_reach_test.go b/cmd/server/node_reach_test.go index 55938919..fef55896 100644 --- a/cmd/server/node_reach_test.go +++ b/cmd/server/node_reach_test.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "strconv" + "strings" "testing" _ "github.com/mattn/go-sqlite3" @@ -289,3 +290,52 @@ func TestScanReachRows_DecodesRows(t *testing.T) { t.Fatalf("snr not decoded: valid=%v val=%v", got.snrValid, got.snr) } } + +// #2146: the reach scan must be driven from the observations time window. +// With the default join order SQLite picked SCAN transmissions (every row) +// and probed observations per transmission: 3.2-3.3 s on an 18M-observation +// database against 1.8-2.0 s when it starts from idx_observations_timestamp. +func TestReachScanStartsFromObservationsTimeIndex(t *testing.T) { + db := newReachScanTestDB(t) + // The planner's choice depends on sqlite_stat1. These are the statistics + // ANALYZE produced on that database (18.3M observations, 1.5M + // transmissions), loaded so the test plans like production does. + for _, s := range []string{ + `CREATE INDEX idx_observations_timestamp ON observations(timestamp)`, + `CREATE INDEX idx_observations_tx_ts ON observations(transmission_id, timestamp)`, + `ANALYZE`, + `DELETE FROM sqlite_stat1`, + `INSERT INTO sqlite_stat1 (tbl, idx, stat) VALUES + ('observations', 'idx_observations_timestamp', '18270402 4'), + ('observations', 'idx_observations_tx_ts', '18270402 11 3'), + ('observations', NULL, '18270402'), + ('transmissions', NULL, '1517599'), + ('observers', NULL, '88')`, + `ANALYZE sqlite_master`, + } { + if _, err := db.conn.Exec(s); err != nil { + t.Fatal(err) + } + } + args := []interface{}{int64(0), `%"01FA"%`, `%"01FA00"%`, 10} + rows, err := db.conn.Query("EXPLAIN QUERY PLAN "+reachScanSQL(2), args...) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var steps []string + for rows.Next() { + var id, parent, notused int + var detail string + if err := rows.Scan(&id, &parent, ¬used, &detail); err != nil { + t.Fatal(err) + } + steps = append(steps, detail) + } + if len(steps) == 0 { + t.Fatal("no query plan") + } + if !strings.Contains(steps[0], "o USING INDEX idx_observations_timestamp") { + t.Fatalf("reach scan must start from observations by timestamp; plan: %q", steps) + } +}