mirror of
https://github.com/Kpa-clawbot/meshcore-analyzer.git
synced 2026-10-15 03:39:53 +00:00
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 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5.5
parent
f50312767f
commit
3e21b179ec
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user