port(upstream#1882): nodes region filter reads the indexed from_pubkey column

cmd/ingestor/from_pubkey_test.go

@@ -92,3 +92,57 @@ func TestBuildPacketData_PopulatesFromPubkey(t *testing.T) {
 		t.Fatalf("BuildPacketData FromPubkey for non-ADVERT = %q, want empty", pd2.FromPubkey)
 	}
 }
+
+// The backfill is the second path that writes from_pubkey, for ADVERT rows
+// ingested before #1143. The server's /api/nodes region filter matches on
+// from_pubkey alone (PR #38), so a backfilled row must carry exactly the
+// pubKey its decoded_json holds, or that node silently drops out of every
+// region it was heard in. Rows without a pubkey get the "" sentinel, and
+// non-ADVERTs are left alone.
+func TestBackfillFromPubkey_CopiesDecodedPubKey_PR38(t *testing.T) {
+	s, err := OpenStore(tempDBPath(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Close()
+
+	const pk1 = "a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1"
+	const pk2 = "b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2b2"
+	rows := []struct {
+		hash        string
+		payloadType int
+		decoded     string
+	}{
+		{"legacy-adv-1", 4, `{"type":"ADVERT","pubKey":"` + pk1 + `","name":"one"}`},
+		{"legacy-adv-2", 4, `{"type":"ADVERT","pubKey":"` + pk2 + `","name":"two"}`},
+		{"legacy-adv-nokey", 4, `{"type":"ADVERT","name":"no key"}`},
+		{"legacy-adv-corrupt", 4, `NOT JSON`},
+		{"legacy-txt", 2, `{"type":"TXT_MSG","pubKey":"` + pk1 + `"}`},
+	}
+	for _, r := range rows {
+		if _, err := s.db.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, payload_version, decoded_json)
+			VALUES ('00', ?, '2026-01-01T00:00:00Z', 1, ?, 0, ?)`, r.hash, r.payloadType, r.decoded); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	// A chunk of 2 makes the 4 ADVERTs span several batches.
+	s.BackfillFromPubkey(2, 0, nil)
+
+	want := map[string]sql.NullString{
+		"legacy-adv-1":       {String: pk1, Valid: true},
+		"legacy-adv-2":       {String: pk2, Valid: true},
+		"legacy-adv-nokey":   {String: "", Valid: true},
+		"legacy-adv-corrupt": {String: "", Valid: true},
+		"legacy-txt":         {},
+	}
+	for hash, w := range want {
+		var got sql.NullString
+		if err := s.db.QueryRow(`SELECT from_pubkey FROM transmissions WHERE hash = ?`, hash).Scan(&got); err != nil {
+			t.Fatal(err)
+		}
+		if got != w {
+			t.Errorf("%s: from_pubkey = %q (valid=%v), want %q (valid=%v)", hash, got.String, got.Valid, w.String, w.Valid)
+		}
+	}
+}

cmd/server/db.go

@@ -1169,8 +1169,40 @@ func (db *DB) GetNodes(limit, offset int, role, search, before, lastHeard, sortB
 			if !db.isV3() {
 				joinCond = "obs.id = o.observer_id"
 			}
+			// #1143: from_pubkey is a dedicated column the ingestor fills at
+			// write time for ADVERT rows, so this lookup does not have to
+			// JSON_EXTRACT and parse decoded_json for every candidate row.
+			// Measured against a live 4.9GB database, 9 interleaved rounds so
+			// concurrent ingest cannot bias one variant: 934ms -> 835ms median
+			// (1.12x) for a single-region filter, identical result set both
+			// ways. The win is purely the avoided JSON parse — EXPLAIN QUERY
+			// PLAN is byte-identical before and after, both driving off
+			// idx_transmissions_payload_type. Despite the column being
+			// indexed, idx_transmissions_from_pubkey does not participate in
+			// this plan at all.
+			//
+			// THE TRAP, for whoever touches this next: from_pubkey is only
+			// written for ADVERTs that actually carry a pubkey — see the
+			// guard in cmd/ingestor/db.go. A NULL here drops out of IN (...)
+			// silently, so a write path that fills decoded_json but forgets
+			// from_pubkey would cost nodes from this filter with no error to
+			// notice. That is safe today only because it is measured to be:
+			// on live staging 25 of 62104 ADVERTs have NULL from_pubkey, and
+			// all 25 have no pubKey in decoded_json either — nothing is lost,
+			// because there was never a pubkey to record. A COALESCE fallback
+			// to JSON_EXTRACT was measured and rejected: it rescued 0 rows,
+			// and it parses decoded_json again for every NULL row, so one
+			// corrupt ADVERT that the backfill has not reached yet fails the
+			// whole query with "malformed JSON" (a 500 on /api/nodes?region=)
+			// exactly as the old expression did. Its speed is not the reason:
+			// on a 4.3GB synthetic DB it times within noise of from_pubkey.
+			// If you add a new write path for transmissions, populate
+			// from_pubkey there rather than reintroducing a per-row JSON parse
+			// here. nodes_region_from_pubkey_test.go locks the result set
+			// against the JSON_EXTRACT expression; TestBackfillFromPubkey_*
+			// in cmd/ingestor locks the backfill's half of the contract.
 			subq := fmt.Sprintf(`public_key IN (
-				SELECT DISTINCT JSON_EXTRACT(t.decoded_json, '$.pubKey')
+				SELECT DISTINCT t.from_pubkey
 				FROM transmissions t
 				JOIN observations o ON o.transmission_id = t.id
 				JOIN observers obs ON %s

cmd/server/nodes_region_from_pubkey_test.go

@@ -0,0 +1,267 @@
+package main
+
+// PR #38 (port of upstream #1882): the /api/nodes region filter reads
+// transmissions.from_pubkey instead of JSON_EXTRACT(decoded_json, '$.pubKey').
+// These tests lock the claim that justifies the swap — the region filter
+// returns exactly the node set the JSON_EXTRACT expression returned — and the
+// one behaviour that changed on purpose: a corrupt decoded_json no longer
+// fails the whole query.
+//
+// The helpers in db_test.go install trigger test_from_pubkey_advert, which
+// copies decoded_json.pubKey into from_pubkey on every ADVERT insert. That
+// makes the two expressions equal by construction, so a test built on them
+// cannot tell the variants apart. The tests here drop the trigger and write
+// from_pubkey the way each real write path does.
+
+import (
+	"database/sql"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+	"testing"
+)
+
+// regionNodesByJSONExtract is the oracle: the node set the region filter
+// returned before the swap, built with the pre-#38 expression. It is written
+// out here on purpose — it is the reference the real GetNodes is compared to,
+// not a copy of the code under test.
+func regionNodesByJSONExtract(t *testing.T, db *DB, region string) []string {
+	t.Helper()
+	codes := normalizeRegionCodes(region)
+	if len(codes) == 0 {
+		t.Fatalf("region %q normalizes to no codes", region)
+	}
+	joinCond := "obs.rowid = o.observer_idx"
+	if !db.isV3() {
+		joinCond = "obs.id = o.observer_id"
+	}
+	args := make([]interface{}, len(codes))
+	for i, c := range codes {
+		args[i] = c
+	}
+	q := `SELECT public_key FROM nodes WHERE public_key IN (
+		SELECT DISTINCT JSON_EXTRACT(t.decoded_json, '$.pubKey')
+		FROM transmissions t
+		JOIN observations o ON o.transmission_id = t.id
+		JOIN observers obs ON ` + joinCond + `
+		WHERE t.payload_type = 4
+		AND UPPER(TRIM(obs.iata)) IN (?` + strings.Repeat(",?", len(codes)-1) + `)
+	)`
+	rows, err := db.conn.Query(q, args...)
+	if err != nil {
+		t.Fatalf("oracle query for %q: %v", region, 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
+}
+
+// regionNodesByGetNodes runs the real region filter and returns the sorted
+// public keys. The limit is far above any test's node count, and total must
+// agree with the page, so nothing is hidden by paging.
+func regionNodesByGetNodes(t *testing.T, db *DB, region string) []string {
+	t.Helper()
+	nodes, total, _, err := db.GetNodes(100000, 0, "", "", "", "", "", region)
+	if err != nil {
+		t.Fatalf("GetNodes(region=%q): %v", region, err)
+	}
+	if total != len(nodes) {
+		t.Fatalf("GetNodes(region=%q): total=%d but %d rows returned", region, total, len(nodes))
+	}
+	out := make([]string, 0, len(nodes))
+	for _, n := range nodes {
+		pk, _ := n["public_key"].(string)
+		out = append(out, pk)
+	}
+	sort.Strings(out)
+	return out
+}
+
+// assertSameNodeSet reports only the difference, so a failure on the fixture
+// names the lost nodes instead of printing every key twice.
+func assertSameNodeSet(t *testing.T, region string, got, want []string) {
+	t.Helper()
+	inGot := make(map[string]bool, len(got))
+	for _, pk := range got {
+		inGot[pk] = true
+	}
+	var missing []string
+	for _, pk := range want {
+		if !inGot[pk] {
+			missing = append(missing, pk)
+		}
+		delete(inGot, pk)
+	}
+	var extra []string
+	for pk := range inGot {
+		extra = append(extra, pk)
+	}
+	sort.Strings(extra)
+	if len(missing) > 0 || len(extra) > 0 {
+		t.Errorf("region %q: GetNodes %d nodes, JSON_EXTRACT oracle %d; missing %v, extra %v",
+			region, len(got), len(want), missing, extra)
+	}
+}
+
+// On the committed e2e fixture (real staging data) every region, alone and
+// combined, returns the same node set as the JSON_EXTRACT expression did.
+func TestGetNodesRegionFilter_FixtureMatchesJSONExtract_PR38(t *testing.T) {
+	src := filepath.Join("..", "..", "test-fixtures", "e2e-fixture.db")
+	raw, err := os.ReadFile(src)
+	if err != nil {
+		t.Skipf("fixture not available: %v", err)
+	}
+	dst := filepath.Join(t.TempDir(), "fixture.db")
+	if err := os.WriteFile(dst, raw, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	db, err := OpenDB(dst)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+
+	rows, err := db.conn.Query(`SELECT DISTINCT UPPER(TRIM(iata)) FROM observers WHERE iata IS NOT NULL AND TRIM(iata) <> '' ORDER BY 1`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var regions []string
+	for rows.Next() {
+		var r string
+		if err := rows.Scan(&r); err != nil {
+			t.Fatal(err)
+		}
+		regions = append(regions, r)
+	}
+	rows.Close()
+	if len(regions) < 2 {
+		t.Fatalf("fixture has %d regions, want at least 2 to exercise the filter", len(regions))
+	}
+	// Multi-region, lower-case with padding (normalizeRegionCodes), and a
+	// code no observer carries.
+	cases := append([]string{}, regions...)
+	cases = append(cases, strings.Join(regions, ","), " "+strings.ToLower(regions[0])+" , "+regions[1], "ZZZ")
+
+	nonEmpty := 0
+	for _, region := range cases {
+		got := regionNodesByGetNodes(t, db, region)
+		assertSameNodeSet(t, region, got, regionNodesByJSONExtract(t, db, region))
+		if len(got) > 0 {
+			nonEmpty++
+		}
+	}
+	if nonEmpty < 2 {
+		t.Fatalf("only %d region cases matched any node; the comparison is vacuous", nonEmpty)
+	}
+}
+
+// setupRegionFromPubkeyDB is the v3 test schema without the auto-populate
+// trigger, so from_pubkey holds exactly what the test writes.
+func setupRegionFromPubkeyDB(t *testing.T) *DB {
+	t.Helper()
+	db := setupTestDB(t)
+	if _, err := db.conn.Exec(`DROP TRIGGER test_from_pubkey_advert`); err != nil {
+		t.Fatal(err)
+	}
+	mustExec := func(q string, args ...interface{}) {
+		t.Helper()
+		if _, err := db.conn.Exec(q, args...); err != nil {
+			t.Fatalf("%s: %v", q, err)
+		}
+	}
+	mustExec(`INSERT INTO observers (rowid, id, name, iata) VALUES (1, 'obs-sjc', 'SJC obs', 'SJC'), (2, 'obs-sfo', 'SFO obs', ' sfo ')`)
+	for _, pk := range []string{"aa00000000000001", "bb00000000000002", "cc00000000000003", "dd00000000000004"} {
+		mustExec(`INSERT INTO nodes (public_key, name, role, last_seen, first_seen) VALUES (?, ?, 'repeater', '2026-10-01T00:00:00Z', '2026-01-01T00:00:00Z')`, pk, "n-"+pk[:2])
+	}
+	return db
+}
+
+// regionTx inserts one transmission and its observations, one per observer rowid.
+func regionTx(t *testing.T, db *DB, hash string, payloadType int, decoded string, fromPubkey sql.NullString, observerIdx ...int) {
+	t.Helper()
+	res, err := db.conn.Exec(`INSERT INTO transmissions (raw_hex, hash, first_seen, route_type, payload_type, decoded_json, from_pubkey)
+		VALUES ('00', ?, '2026-10-01T00:00:00Z', 1, ?, ?, ?)`, hash, payloadType, decoded, fromPubkey)
+	if err != nil {
+		t.Fatal(err)
+	}
+	id, _ := res.LastInsertId()
+	for i, idx := range observerIdx {
+		if _, err := db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, timestamp) VALUES (?, ?, ?)`, id, idx, 1790000000+i); err != nil {
+			t.Fatal(err)
+		}
+	}
+}
+
+func advertJSON(pk string) string {
+	return `{"type":"ADVERT","pubKey":"` + pk + `","name":"n"}`
+}
+
+// The row shapes production writes, side by side: ADVERTs the ingestor
+// stamped with their pubkey, an ADVERT that carried no pubkey (left NULL by
+// the guard in cmd/ingestor/db.go — the 25 such rows on staging), a legacy
+// pubkey-less ADVERT the backfill marked with the "" sentinel, and a
+// non-ADVERT. The region filter must return the JSON_EXTRACT node set for
+// every region, and the explicit expected set besides.
+func TestGetNodesRegionFilter_WritePathShapesMatchJSONExtract_PR38(t *testing.T) {
+	db := setupRegionFromPubkeyDB(t)
+	defer db.Close()
+	pk := func(s string) sql.NullString { return sql.NullString{String: s, Valid: true} }
+	null := sql.NullString{}
+
+	regionTx(t, db, "adv-a-sjc", 4, advertJSON("aa00000000000001"), pk("aa00000000000001"), 1)
+	regionTx(t, db, "adv-b-sfo", 4, advertJSON("bb00000000000002"), pk("bb00000000000002"), 2)
+	regionTx(t, db, "adv-c-both", 4, advertJSON("cc00000000000003"), pk("cc00000000000003"), 1, 2)
+	regionTx(t, db, "adv-nopk-null", 4, `{"type":"ADVERT","name":"no key"}`, null, 1, 2)
+	regionTx(t, db, "adv-nopk-backfilled", 4, `{"type":"ADVERT"}`, pk(""), 1, 2)
+	// dd is heard in SJC only through a non-ADVERT, which never places a
+	// node in a region either way.
+	regionTx(t, db, "txt-d-sjc", 2, `{"type":"TXT_MSG","pubKey":"dd00000000000004"}`, null, 1)
+
+	for _, tc := range []struct {
+		region string
+		want   []string
+	}{
+		{"SJC", []string{"aa00000000000001", "cc00000000000003"}},
+		{"SFO", []string{"bb00000000000002", "cc00000000000003"}},
+		{"sjc, SFO", []string{"aa00000000000001", "bb00000000000002", "cc00000000000003"}},
+		{"LAX", nil},
+	} {
+		got := regionNodesByGetNodes(t, db, tc.region)
+		assertSameNodeSet(t, tc.region, got, regionNodesByJSONExtract(t, db, tc.region))
+		if strings.Join(got, ",") != strings.Join(tc.want, ",") {
+			t.Errorf("region %q: got %v, want %v", tc.region, got, tc.want)
+		}
+	}
+}
+
+// One ADVERT whose decoded_json is not valid JSON made the JSON_EXTRACT
+// expression raise "malformed JSON", failing every region query that joined
+// it — /api/nodes?region= then answered 500. The row is left with a NULL
+// from_pubkey, the state the backfill has not reached yet.
+func TestGetNodesRegionFilter_MalformedAdvertJSONDoesNotFailQuery_PR38(t *testing.T) {
+	db := setupRegionFromPubkeyDB(t)
+	defer db.Close()
+
+	regionTx(t, db, "adv-a-sjc", 4, advertJSON("aa00000000000001"), sql.NullString{String: "aa00000000000001", Valid: true}, 1)
+	regionTx(t, db, "adv-corrupt-sjc", 4, `NOT JSON`, sql.NullString{}, 1)
+
+	nodes, total, _, err := db.GetNodes(100, 0, "", "", "", "", "", "SJC")
+	if err != nil {
+		t.Fatalf("GetNodes(region=SJC) with one corrupt ADVERT in the region: %v", err)
+	}
+	if total != 1 || len(nodes) != 1 || nodes[0]["public_key"] != "aa00000000000001" {
+		t.Fatalf("GetNodes(region=SJC) = total %d, %d nodes %v; want only aa00000000000001", total, len(nodes), nodes)
+	}
+}