diff --git a/cmd/server/area_sql_fallback_569_test.go b/cmd/server/area_sql_fallback_569_test.go new file mode 100644 index 000000000..92697b851 --- /dev/null +++ b/cmd/server/area_sql_fallback_569_test.go @@ -0,0 +1,265 @@ +package main + +import ( + "encoding/json" + "fmt" + "net/http/httptest" + "reflect" + "sort" + "strings" + "testing" + "time" +) + +// #569: the GPS `area` filter (originator membership) must survive the SQL +// path — the store's history fallback (since/until older than oldestLoaded) +// and the direct-DB handler path — with exactly the in-memory semantics: +// - originator = decoded ADVERT pubkey; packets without one are excluded; +// - an unknown area key, or an area no positioned node falls inside, +// applies no filter at all (resolveAreaNodes returns nil); +// - nodes without GPS (NULL or 0,0) are never members; +// - area and region are cumulative. + +var area569Config = &Config{Areas: map[string]AreaEntry{ + "BEL": {Label: "Belgium", Polygon: [][2]float64{{50.0, 2.5}, {51.5, 2.5}, {51.5, 6.4}, {50.0, 6.4}}}, + "BOX": {Label: "Box", LatMin: f64p569(50.9), LatMax: f64p569(51.5), LonMin: f64p569(4.9), LonMax: f64p569(6.4)}, + "SEA": {Label: "Empty", LatMin: f64p569(10), LatMax: f64p569(11), LonMin: f64p569(10), LonMax: f64p569(11)}, +}} + +func f64p569(v float64) *float64 { return &v } + +func pk569(c string) string { return strings.Repeat(c, 64) } + +// seedArea569 writes nodes, observers and packets. Every packet has a unique +// first_seen and observation timestamp so ordinary (id) and grouped (latest) +// orderings are total and identical across the two paths. +func seedArea569(t *testing.T, db *DB) { + t.Helper() + nodes := []struct { + pk string + lat, lon interface{} + }{ + {pk569("a"), 50.85, 4.35}, // BEL only + {pk569("b"), 51.0, 5.0}, // BEL + BOX + {pk569("c"), 48.0, 4.35}, // outside + {pk569("d"), nil, nil}, // no GPS + {pk569("e"), 0.0, 0.0}, // (0,0) = no GPS + } + for _, n := range nodes { + if _, err := db.conn.Exec(`INSERT INTO nodes (public_key, name, lat, lon) VALUES (?, ?, ?, ?)`, n.pk, "n-"+n.pk[:1], n.lat, n.lon); err != nil { + t.Fatal(err) + } + } + mustExecDB(t, db, `INSERT INTO observers (id, name, iata) VALUES ('obsA', 'A', 'AAA'), ('obsB', 'B', 'BBB')`) + + // originator ("" = no pubkey), payload type, observer rowid. + packets := []struct { + pk string + typ int + obs int + }{ + {pk569("a"), 4, 1}, {pk569("b"), 4, 2}, {pk569("c"), 4, 1}, {pk569("a"), 4, 2}, + {pk569("d"), 4, 1}, {pk569("e"), 4, 2}, {"", 4, 1}, {"", 5, 1}, + {pk569("b"), 4, 1}, {pk569("a"), 4, 1}, {pk569("c"), 4, 2}, {pk569("b"), 4, 2}, + } + base := time.Date(2026, 1, 1, 0, 0, 0, 0, time.UTC) + for i, p := range packets { + ts := base.Add(time.Duration(i) * time.Minute) + decoded := `{"type":"GRP_TXT","channelHashHex":"AA"}` + if p.typ == 4 { + decoded = `{"type":"ADVERT"}` + if p.pk != "" { + decoded = fmt.Sprintf(`{"type":"ADVERT","pubKey":%q}`, p.pk) + } + } + res, err := db.conn.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) VALUES ('AA', ?, ?, 1, ?, ?)`, + fmt.Sprintf("%016x", i+1), ts.Format(time.RFC3339), p.typ, decoded) + if err != nil { + t.Fatal(err) + } + id, _ := res.LastInsertId() + if _, err := db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp) VALUES (?, ?, '[]', ?)`, id, p.obs, ts.Unix()); err != nil { + t.Fatal(err) + } + } +} + +type area569Page struct { + Total int + Hashes []string +} + +func newArea569Server(t *testing.T, mode string) *Server { + t.Helper() + db := setupTestDB(t) + t.Cleanup(func() { db.Close() }) + seedArea569(t, db) + srv := NewServer(db, area569Config, NewHub()) + if mode == "database" { + return srv + } + srv.store = NewPacketStore(db, nil) + srv.store.config = area569Config + if err := srv.store.Load(); err != nil { + t.Fatal(err) + } + if !srv.store.WaitIndexesReady(5 * time.Second) { + t.Fatal("indexes not ready") + } + if mode == "memory" { + srv.store.oldestLoaded = "" // never fall back + } else { + srv.store.oldestLoaded = "2026-02-01T00:00:00Z" // newer than every window bound below + } + return srv +} + +func getArea569(t *testing.T, srv *Server, query string) area569Page { + t.Helper() + w := httptest.NewRecorder() + srv.handlePackets(w, httptest.NewRequest("GET", "/api/packets?"+query, nil)) + if w.Code != 200 { + t.Fatalf("%s: %d %s", query, w.Code, w.Body.String()) + } + var got PacketResult + if err := json.Unmarshal(w.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + page := area569Page{Total: got.Total, Hashes: []string{}} + for _, p := range got.Packets { + page.Hashes = append(page.Hashes, fmt.Sprint(p["hash"])) + } + return page +} + +// TestAreaFilterSQLMatchesMemory569 is the equivalence matrix: every SQL +// route (store fallback on since, on until, and the store-less direct DB +// path) must return the same rows, order and total as the in-memory path. +func TestAreaFilterSQLMatchesMemory569(t *testing.T) { + memory := newArea569Server(t, "memory") + sqlModes := map[string]*Server{"fallback": newArea569Server(t, "fallback"), "database": newArea569Server(t, "database")} + windows := []string{"since=2025-12-01T00:00:00Z", "until=2026-01-15T00:00:00Z", "since=2026-01-01T00:02:30Z&until=2026-01-15T00:00:00Z"} + + for _, window := range windows { + for _, grouped := range []bool{false, true} { + for _, area := range []string{"", "BEL", "BOX", "SEA", "NOPE"} { + for _, extra := range []string{"", "®ion=AAA", "®ion=BBB", "&type=4", "&observer=obsB"} { + for _, page := range []string{"limit=50", "limit=2", "limit=2&offset=2", "limit=2&offset=40"} { + query := fmt.Sprintf("%s&%s&groupByHash=%t%s", page, window, grouped, extra) + if area != "" { + query += "&area=" + area + } + want := getArea569(t, memory, query) + for name, srv := range sqlModes { + if got := getArea569(t, srv, query); !reflect.DeepEqual(got, want) { + t.Errorf("%s %s:\n got %+v\n want %+v", name, query, got, want) + } + } + } + } + } + } + } +} + +// TestAreaFilterSQLFallbackContract569 pins the absolute contract on the +// fallback, so the matrix above cannot pass by both paths drifting together. +func TestAreaFilterSQLFallbackContract569(t *testing.T) { + for _, mode := range []string{"memory", "fallback", "database"} { + srv := newArea569Server(t, mode) + for _, grouped := range []bool{false, true} { + for _, window := range []string{"since=2025-12-01T00:00:00Z", "until=2026-01-15T00:00:00Z"} { + base := fmt.Sprintf("limit=50&%s&groupByHash=%t", window, grouped) + cases := []struct { + query string + hashes []string // set, sorted + }{ + // a×3 + b×3 inside the polygon; c (outside), d/e (no GPS), + // the pubkey-less ADVERT and the GRP_TXT are excluded. + {"&area=BEL", []string{"0000000000000001", "0000000000000002", "0000000000000004", "0000000000000009", "000000000000000a", "000000000000000c"}}, + {"&area=BOX", []string{"0000000000000002", "0000000000000009", "000000000000000c"}}, + {"&area=BEL®ion=AAA", []string{"0000000000000001", "0000000000000009", "000000000000000a"}}, + // Unknown key / empty area: no filter (in-memory nil set). + {"&area=NOPE", all569()}, + {"&area=SEA", all569()}, + {"&area=BOX&type=5", nil}, + } + for _, c := range cases { + got := getArea569(t, srv, base+c.query) + sorted := append([]string{}, got.Hashes...) + sort.Strings(sorted) + if got.Total != len(c.hashes) || !reflect.DeepEqual(sorted, nonNil569(c.hashes)) { + t.Errorf("%s %s: total=%d hashes=%v want %v", mode, base+c.query, got.Total, sorted, c.hashes) + } + } + } + } + } +} + +func all569() []string { + out := make([]string, 12) + for i := range out { + out[i] = fmt.Sprintf("%016x", i+1) + } + return out +} + +func nonNil569(s []string) []string { + if s == nil { + return []string{} + } + return s +} + +// TestAreaFilterAcrossCacheBoundary569: history older than the loaded window +// exists only in SQLite. A since/until window reaching past oldestLoaded +// must return the old in-area packets and none of the old outside ones, +// with a total that counts both sides of the boundary. +func TestAreaFilterAcrossCacheBoundary569(t *testing.T) { + srv := newArea569Server(t, "memory") + // Old rows, SQL-only: inside (a), outside (c), inside (b). + for i, pk := range []string{pk569("a"), pk569("c"), pk569("b")} { + ts := time.Date(2025, 12, 1, 0, i, 0, 0, time.UTC) + res, err := srv.db.conn.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json) VALUES ('AA', ?, ?, 1, 4, ?)`, + fmt.Sprintf("old%013x", i+1), ts.Format(time.RFC3339), fmt.Sprintf(`{"type":"ADVERT","pubKey":%q}`, pk)) + if err != nil { + t.Fatal(err) + } + id, _ := res.LastInsertId() + if _, err := srv.db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp) VALUES (?, 1, '[]', ?)`, id, ts.Unix()); err != nil { + t.Fatal(err) + } + } + srv.store.oldestLoaded = "2026-01-01T00:00:00Z" // the real loaded boundary + + for _, grouped := range []bool{false, true} { + g := fmt.Sprintf("&groupByHash=%t", grouped) + // In-memory control: the old rows are invisible, area applies. + if got := getArea569(t, srv, "limit=50&area=BEL"+g); got.Total != 6 { + t.Fatalf("memory control total=%d want 6", got.Total) + } + for _, window := range []string{"since=2025-11-30T00:00:00Z", "since=2025-11-30T00:00:00Z&until=2026-01-15T00:00:00Z", "until=2025-12-31T00:00:00Z"} { + wantTotal := 8 // 6 in memory + old a, old b + if strings.HasPrefix(window, "until=") { + wantTotal = 2 // only the SQL-only side + } + seen := map[string]bool{} + for offset := 0; offset < wantTotal+2; offset += 3 { + got := getArea569(t, srv, fmt.Sprintf("limit=3&offset=%d&%s&area=BEL%s", offset, window, g)) + if got.Total != wantTotal { + t.Fatalf("%s grouped=%t offset=%d total=%d want %d", window, grouped, offset, got.Total, wantTotal) + } + for _, h := range got.Hashes { + if h == "old0000000000002" || seen[h] { + t.Fatalf("%s grouped=%t: outside or duplicate row %s", window, grouped, h) + } + seen[h] = true + } + } + if len(seen) != wantTotal || !seen["old0000000000001"] || !seen["old0000000000003"] { + t.Fatalf("%s grouped=%t: pages covered %v, want %d rows incl. both old inside rows", window, grouped, seen, wantTotal) + } + } + } +} diff --git a/cmd/server/area_sql_predicate_569_test.go b/cmd/server/area_sql_predicate_569_test.go new file mode 100644 index 000000000..85fa98af1 --- /dev/null +++ b/cmd/server/area_sql_predicate_569_test.go @@ -0,0 +1,109 @@ +package main + +import ( + "encoding/json" + "fmt" + "strings" + "testing" +) + +// TestAreaFilterSQLPredicateBounded569: the area predicate binds ONE +// parameter whatever the member count, and drives off +// idx_transmissions_from_pubkey (no per-row JSON parse, no table scan). +func TestAreaFilterSQLPredicateBounded569(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + members := make(map[string]bool, 5000) + for i := 0; i < 5000; i++ { + members[fmt.Sprintf("%064x", i)] = true + } + where, args := db.buildTransmissionWhere(PacketQuery{Area: "BEL", AreaNodes: members}) + if len(where) != 1 || len(args) != 1 { + t.Fatalf("want one predicate with one bound arg, got %d/%d: %v", len(where), len(args), where) + } + var keys []string + if err := json.Unmarshal([]byte(args[0].(string)), &keys); err != nil || len(keys) != 5000 { + t.Fatalf("arg must be a JSON array of all members: %d %v", len(keys), err) + } + if strings.Contains(where[0], "decoded_json") { + t.Fatalf("predicate must not parse decoded_json per row: %s", where[0]) + } + rows, err := db.conn.Query("EXPLAIN QUERY PLAN SELECT COUNT(*) FROM transmissions t WHERE "+where[0], args...) + if err != nil { + t.Fatal(err) + } + defer rows.Close() + var plan []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) + } + plan = append(plan, detail) + } + joined := strings.Join(plan, "\n") + if !strings.Contains(joined, "idx_transmissions_from_pubkey") || strings.Contains(joined, "SCAN t") { + t.Fatalf("area predicate must search idx_transmissions_from_pubkey:\n%s", joined) + } + + // The empty-pubkey backfill sentinel ("") is never a member, matching + // the in-memory `pk == ""` exclusion. + _, args = db.buildTransmissionWhere(PacketQuery{Area: "BEL", AreaNodes: map[string]bool{"": true, "x": true}}) + if len(args) != 1 || args[0] != `["x"]` { + t.Fatalf("empty key must be dropped, got %v", args) + } + // Unresolved / empty membership applies no filter (in-memory nil set). + if where, _ := db.buildTransmissionWhere(PacketQuery{Area: "SEA"}); len(where) != 0 { + t.Fatalf("nil membership must not filter: %v", where) + } +} + +// BenchmarkAreaSQL569 times the full SQL packet query (COUNT + page) over +// 30K ADVERTs from 2K nodes: no filter, a 1K-member area, and a since window +// (the shape of every history fallback) without and with that area. +func BenchmarkAreaSQL569(b *testing.B) { + db := setupTestDB(b) + defer db.Close() + tx, _ := db.conn.Begin() + for i := 0; i < 30000; i++ { + ts := int64(1767225600 + i) + res, err := tx.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, payload_type, decoded_json) VALUES ('AA', ?, ?, 4, ?)`, + fmt.Sprintf("%016x", i+1), fmt.Sprintf("2026-01-01T%08d", i), fmt.Sprintf(`{"pubKey":"%064x"}`, i%2000)) + if err != nil { + b.Fatal(err) + } + id, _ := res.LastInsertId() + tx.Exec(`INSERT INTO observations (transmission_id, observer_idx, timestamp) VALUES (?, 1, ?)`, id, ts) + } + tx.Commit() + members := make(map[string]bool, 1000) + for i := 0; i < 1000; i++ { + members[fmt.Sprintf("%064x", i*2)] = true + } + for _, c := range []struct { + name string + q PacketQuery + }{ + {"none", PacketQuery{Limit: 50}}, + {"area1k", PacketQuery{Limit: 50, Area: "X", AreaNodes: members}}, + {"since", PacketQuery{Limit: 50, Since: "2026-01-01T08:00:00Z"}}, + {"area1k+since", PacketQuery{Limit: 50, Area: "X", AreaNodes: members, Since: "2026-01-01T08:00:00Z"}}, + } { + for _, grouped := range []bool{false, true} { + b.Run(fmt.Sprintf("%s/grouped=%t", c.name, grouped), func(b *testing.B) { + for i := 0; i < b.N; i++ { + var err error + if grouped { + _, err = db.QueryGroupedPackets(c.q) + } else { + _, err = db.QueryPackets(c.q) + } + if err != nil { + b.Fatal(err) + } + } + }) + } + } +} diff --git a/cmd/server/db.go b/cmd/server/db.go index 43092e403..37d8c433b 100644 --- a/cmd/server/db.go +++ b/cmd/server/db.go @@ -736,6 +736,12 @@ type PacketQuery struct { Channel string // channel_hash filter (#812). Plain names like "#test"/"public" or "enc_" for encrypted Order string // ASC or DESC ExpandObservations bool // when true, include observation sub-maps in txToMap output + + // AreaNodes is Area resolved to its member pubkeys (resolveAreaNodes / + // areaMemberSet) — the DB layer has no config.Areas. nil applies no area + // filter, exactly like the in-memory path's nil set for an unknown or + // empty area (#569). + AreaNodes map[string]bool } // PacketResult wraps paginated packet list. @@ -1069,9 +1075,33 @@ func (db *DB) buildTransmissionWhere(q PacketQuery) ([]string, []interface{}) { } args = append(args, q.Region) } + if len(q.AreaNodes) > 0 { + // #569: same originator contract as filterPacketsPrepared — the + // decoded ADVERT pubkey, which the ingestor copies into the indexed + // from_pubkey column (#1143). One bound JSON array regardless of the + // member count; the plan probes idx_transmissions_from_pubkey once + // per member instead of parsing decoded_json per row. + where = append(where, "t.from_pubkey IN (SELECT value FROM json_each(?))") + args = append(args, areaMembersJSON(q.AreaNodes)) + } return where, args } +// areaMembersJSON encodes an area member set as a sorted JSON array for +// json_each. "" is dropped: the in-memory path never matches an empty +// originator, and "" is the from_pubkey backfill's no-pubkey sentinel. +func areaMembersJSON(members map[string]bool) string { + keys := make([]string, 0, len(members)) + for pk, ok := range members { + if ok && pk != "" { + keys = append(keys, pk) + } + } + sort.Strings(keys) + b, _ := json.Marshal(keys) + return string(b) +} + func (db *DB) resolveNodePubkey(nodeIDOrName string) string { var pk string err := db.conn.QueryRow("SELECT public_key FROM nodes WHERE public_key = ? OR name = ? LIMIT 1", nodeIDOrName, nodeIDOrName).Scan(&pk) diff --git a/cmd/server/routes.go b/cmd/server/routes.go index 68d8661ce..da3134839 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -1398,6 +1398,14 @@ func (s *Server) handlePackets(w http.ResponseWriter, r *http.Request) { q.Route = &t } + if s.store == nil && q.Area != "" { + // #569: no store to resolve the area — resolve it here so the + // direct DB query filters exactly as the store would. + if entry, ok := s.cfg.Areas[q.Area]; ok { + q.AreaNodes = areaMemberSet(s.db, entry) + } + } + if r.URL.Query().Get("groupByHash") == "true" { var result *PacketResult var err error diff --git a/cmd/server/store.go b/cmd/server/store.go index 4cd4f2844..1963b76aa 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -2494,6 +2494,11 @@ func (s *PacketStore) QueryPackets(q PacketQuery) *PacketResult { needsSQL := (q.Since != "" && q.Since < oldest) || (q.Until != "" && q.Until < oldest) if needsSQL { + // #569: carry the same cached area membership the in-memory + // filter uses, so SQL history applies it before COUNT/LIMIT. + if q.Area != "" { + q.AreaNodes = s.resolveAreaNodes(q.Area) + } if result, err := s.db.QueryPackets(q); err == nil { return result } else { @@ -2570,6 +2575,11 @@ func (s *PacketStore) QueryGroupedPackets(q PacketQuery) *PacketResult { needsSQL := (q.Since != "" && q.Since < oldest) || (q.Until != "" && q.Until < oldest) if needsSQL { + // #569: carry the same cached area membership the in-memory + // filter uses, so SQL history applies it before COUNT/LIMIT. + if q.Area != "" { + q.AreaNodes = s.resolveAreaNodes(q.Area) + } if result, err := s.db.QueryGroupedPackets(q); err == nil { return result } else { @@ -4555,14 +4565,7 @@ func (s *PacketStore) resolveAreaNodes(areaKey string) map[string]bool { s.areaNodeMu.RUnlock() // Slow path: query the DB outside any lock, then write back under Lock. - pks, err := s.db.GetNodePubkeysInArea(entry) - var m map[string]bool - if err == nil && len(pks) > 0 { - m = make(map[string]bool, len(pks)) - for _, pk := range pks { - m[pk] = true - } - } + m := areaMemberSet(s.db, entry) s.areaNodeMu.Lock() // Re-check in case another goroutine already refreshed while we queried. @@ -4576,6 +4579,22 @@ func (s *PacketStore) resolveAreaNodes(areaKey string) map[string]bool { return m } +// areaMemberSet returns the pubkeys of positioned nodes inside entry, or nil +// when there are none (or the lookup fails) — nil means "no area filter". +// Shared by the store's cached resolveAreaNodes and the store-less handler +// path so both SQL and in-memory queries see one membership contract. +func areaMemberSet(db *DB, entry AreaEntry) map[string]bool { + pks, err := db.GetNodePubkeysInArea(entry) + if err != nil || len(pks) == 0 { + return nil + } + m := make(map[string]bool, len(pks)) + for _, pk := range pks { + m[pk] = true + } + return m +} + // iataMatchesRegion returns true if iata matches any of the comma-separated // region codes in regionParam. Comparison is case-insensitive and trim-tolerant. // Empty iata never matches; empty regionParam never matches.