perf(store): read resolved_path pubkeys in place, one copy each

cmd/server/chunked_load.go

@@ -629,8 +629,7 @@ func (s *PacketStore) scanAndMergeChunk(rows *sql.Rows, relayPM *prefixMap, cold
 
 			rpStr := nullStrVal(resolvedPathStr)
 			if rpStr != "" {
-				rp := unmarshalResolvedPath(rpStr)
-				pks := extractResolvedPubkeys(rp)
+				pks := resolvedPubkeysFromJSON(rpStr)
 				s.indexResolvedPathHops(tx, pks, hopsSeen)
 			} else if relayPM != nil && obsPJ != "" && obsPJ != "[]" {
 				// resolved_path is NULL on live (since #1287 relay data is

cmd/server/eviction_resolved_pubkeys_test.go

@@ -0,0 +1,54 @@
+package main
+
+import (
+	"fmt"
+	"runtime"
+	"testing"
+)
+
+func evictionResolvedStore(t testing.TB, n int) (*PacketStore, []int) {
+	t.Helper()
+	db := setupTestDB(t)
+	t.Cleanup(func() { db.Close() })
+	pk := func(i int) string { return fmt.Sprintf("%064x", i) }
+	ids := make([]int, 0, n)
+	for i := 1; i <= n; i++ {
+		if _, err := db.conn.Exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen) VALUES (?, '00', ?, '2026-01-01T00:00:00Z')`, i, fmt.Sprintf("h%015d", i)); err != nil {
+			t.Fatal(err)
+		}
+		for o := 0; o < 3; o++ {
+			rp := `["` + pk(i%9) + `",null,"` + pk(o+20) + `","` + pk(i%9) + `"]`
+			if _, err := db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp, resolved_path) VALUES (?, 1, '["aa","bb","cc","dd"]', 0, ?)`, i, rp); err != nil {
+				t.Fatal(err)
+			}
+		}
+		ids = append(ids, i)
+	}
+	return &PacketStore{db: db}, ids
+}
+
+// Eviction looks up the resolved pubkeys of every packet it drops, once a
+// minute in steady state. It decoded each resolved_path into a pointer per
+// hop and built a de-dup map per packet: about 130MB of allocations per
+// five minutes on a busy week.
+func TestResolvedPubkeysForEvictionBatch(t *testing.T) {
+	s, ids := evictionResolvedStore(t, 300)
+	got := s.resolvedPubkeysForEvictionBatch(ids)
+	want := []string{fmt.Sprintf("%064x", 1), fmt.Sprintf("%064x", 20), fmt.Sprintf("%064x", 21), fmt.Sprintf("%064x", 22)}
+	if fmt.Sprint(got[1]) != fmt.Sprint(want) {
+		t.Fatalf("tx 1: %v, want %v (first-seen order, de-duplicated)", got[1], want)
+	}
+	if len(got) != 300 {
+		t.Fatalf("%d packets resolved, want 300", len(got))
+	}
+	var a, b runtime.MemStats
+	runtime.GC()
+	runtime.ReadMemStats(&a)
+	_ = s.resolvedPubkeysForEvictionBatch(ids)
+	runtime.ReadMemStats(&b)
+	per := float64(b.Mallocs-a.Mallocs) / float64(len(ids))
+	t.Logf("%.1f allocations per evicted packet (3 observations, 4 hops each)", per)
+	if per > 20 {
+		t.Fatalf("%.1f allocations per evicted packet, limit 20", per)
+	}
+}

cmd/server/neighbor_persist.go

@@ -221,3 +221,102 @@ func unmarshalResolvedPath(s string) []*string {
 	}
 	return result
 }
+
+// resolvedPubkeysFromJSON returns the non-empty pubkeys of a resolved_path
+// JSON array, exactly extractResolvedPubkeys(unmarshalResolvedPath(s)). The
+// bulk loaders call it for most observations; the usual array of plain
+// strings and nulls is read in place, one string per pubkey, instead of
+// decoding a string and a pointer per hop. Anything else is decoded.
+//
+// Pubkeys go through internStr: relays repeat across nearly every
+// observation, so a pubkey seen before costs no allocation. Interned keys
+// live for the process, evicted nodes' keys included; the table is bounded
+// (internMax), and past it a pubkey is returned uninterned.
+func resolvedPubkeysFromJSON(s string) []string {
+	if s == "" {
+		return nil
+	}
+	if out, ok := scanResolvedPath(s); ok {
+		return out
+	}
+	return extractResolvedPubkeys(unmarshalResolvedPath(s))
+}
+
+// scanResolvedPath reads s as a JSON array of escape-free strings and nulls
+// with JSON whitespace between tokens. ok is false for anything else.
+func scanResolvedPath(s string) (out []string, ok bool) {
+	i := 0
+	ws := func() {
+		for i < len(s) && (s[i] == ' ' || s[i] == '\t' || s[i] == '\n' || s[i] == '\r') {
+			i++
+		}
+	}
+	ws()
+	if i >= len(s) {
+		return nil, false
+	}
+	if strings.HasPrefix(s[i:], "null") {
+		// unmarshalResolvedPath gives nil for a bare null.
+		i += 4
+		ws()
+		return nil, i == len(s)
+	}
+	if s[i] != '[' {
+		return nil, false
+	}
+	i++
+	ws()
+	if i < len(s) && s[i] == ']' {
+		i++
+		ws()
+		return nil, i == len(s)
+	}
+	for {
+		switch {
+		case i < len(s) && s[i] == '"':
+			i++
+			start := i
+			for i < len(s) && s[i] != '"' {
+				if s[i] == '\\' || s[i] < 0x20 || s[i] >= 0x80 {
+					return nil, false
+				}
+				i++
+			}
+			if i >= len(s) {
+				return nil, false
+			}
+			if i > start {
+				if out == nil {
+					out = make([]string, 0, 8)
+				}
+				out = append(out, internStr(s[start:i]))
+			}
+			i++
+		case strings.HasPrefix(s[i:], "null"):
+			i += 4
+		default:
+			return nil, false
+		}
+		ws()
+		if i >= len(s) {
+			return nil, false
+		}
+		switch s[i] {
+		case ',':
+			i++
+			ws()
+		case ']':
+			i++
+			ws()
+			if i != len(s) {
+				return nil, false
+			}
+			if out == nil {
+				out = []string{}
+			}
+			return out, true
+		default:
+			return nil, false
+		}
+	}
+}

cmd/server/resolved_index.go

@@ -516,38 +516,35 @@ func (s *PacketStore) resolvedPubkeysForEvictionBatch(txIDs []int) map[int][]str
 			continue
 		}
 
+		var txID int
+		var rpJSON sql.NullString
 		for rows.Next() {
-			var txID int
-			var rpJSON sql.NullString
 			if err := rows.Scan(&txID, &rpJSON); err != nil {
 				continue
 			}
 			if !rpJSON.Valid || rpJSON.String == "" {
 				continue
 			}
-			rp := unmarshalResolvedPath(rpJSON.String)
-			for _, p := range rp {
-				if p != nil && *p != "" {
-					result[txID] = append(result[txID], *p)
+			// De-duplicate as we go: a packet has a handful of keys, so a
+			// scan of its list is cheaper than a map per packet.
+			pks := result[txID]
+			for _, pk := range resolvedPubkeysFromJSON(rpJSON.String) {
+				dup := false
+				for _, have := range pks {
+					if have == pk {
+						dup = true
+						break
+					}
+				}
+				if !dup {
+					pks = append(pks, pk)
 				}
 			}
+			result[txID] = pks
 		}
 		rows.Close()
 	}
 
-	// Deduplicate per-txID
-	for txID, pks := range result {
-		seen := make(map[string]bool, len(pks))
-		deduped := pks[:0]
-		for _, pk := range pks {
-			if !seen[pk] {
-				seen[pk] = true
-				deduped = append(deduped, pk)
-			}
-		}
-		result[txID] = deduped
-	}
-
 	return result
 }
 

cmd/server/resolved_pubkeys_bench_test.go

@@ -0,0 +1,37 @@
+package main
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+)
+
+// One resolved_path per observation: 5 relays and 3 unresolved hops, drawn
+// from a pool of 2,000 relays, as on a busy instance.
+func BenchmarkResolvedPubkeysFromJSON(b *testing.B) {
+	const pool = 2000
+	paths := make([]string, 256)
+	for i := range paths {
+		hops := make([]string, 0, 8)
+		for h := 0; h < 5; h++ {
+			hops = append(hops, fmt.Sprintf(`"%064x"`, (i*7+h*131)%pool))
+		}
+		hops = append(hops, "null", "null", "null")
+		paths[i] = "[" + strings.Join(hops, ",") + "]"
+	}
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_ = resolvedPubkeysFromJSON(paths[i%len(paths)])
+	}
+}
+
+// An eviction batch of 1,000 packets with 3 observations each.
+func BenchmarkResolvedPubkeysForEvictionBatch(b *testing.B) {
+	s, ids := evictionResolvedStore(b, 1000)
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		_ = s.resolvedPubkeysForEvictionBatch(ids)
+	}
+}

cmd/server/resolved_pubkeys_json_test.go

@@ -0,0 +1,52 @@
+package main
+
+import (
+	"math/rand"
+	"strings"
+	"testing"
+)
+
+// The bulk loaders decode each observation's resolved_path (a JSON array of
+// pubkeys and nulls, ~420 chars on live data, set on most observations)
+// only to collect its non-empty pubkeys. Decoding into []*string allocated
+// a string and a pointer per hop plus two slices: about 210MB over one
+// startup on a busy week. resolvedPubkeysFromJSON reads the usual array in
+// place, and a pubkey it has seen before costs nothing: only the result is
+// allocated.
+func TestResolvedPubkeysFromJSONAllocations(t *testing.T) {
+	pk := strings.Repeat("ab", 32)
+	rp := `["` + pk + `",null,"` + strings.Repeat("cd", 32) + `",null,"` + strings.Repeat("ef", 32) + `"]`
+	_ = resolvedPubkeysFromJSON(rp) // first sight keeps one copy of each
+	if n := testing.AllocsPerRun(100, func() { _ = resolvedPubkeysFromJSON(rp) }); n > 1 {
+		t.Fatalf("resolvedPubkeysFromJSON allocates %.0f times for 3 known pubkeys, want at most 1 (the result)", n)
+	}
+}
+
+// It returns exactly what decoding and extracting would, odd inputs included.
+func TestResolvedPubkeysFromJSONMatchesDecoding(t *testing.T) {
+	cases := []string{
+		"", "[]", "null", "[null]", `["aa"]`, `["aa",null,"bb"]`, ` [ "aa" , null ] `, `[""]`, `["",null,""]`,
+		`["aa",]`, `["aa"`, `["aa"]x`, `[1]`, `["aa",1]`, `[nul]`, `[nulll]`, `["a\"b"]`, `["é"]`,
+		`[["aa"]]`, `{"a":1}`, "[\"a\x01\"]", `["aa" "bb"]`, `[,]`, `[null,]`, "[\"\xff\"]", `[true]`,
+	}
+	rng := rand.New(rand.NewSource(2))
+	alphabet := []string{`[`, `]`, `"`, `,`, ` `, `a`, `1`, `\`, `null`, `nul`, `{`, `:`}
+	for i := 0; i < 20000; i++ {
+		var b strings.Builder
+		for j := rng.Intn(12); j >= 0; j-- {
+			b.WriteString(alphabet[rng.Intn(len(alphabet))])
+		}
+		cases = append(cases, b.String())
+	}
+	for _, s := range cases {
+		got, want := resolvedPubkeysFromJSON(s), extractResolvedPubkeys(unmarshalResolvedPath(s))
+		if len(got) != len(want) {
+			t.Fatalf("%q: got %q, want %q", s, got, want)
+		}
+		for i := range want {
+			if got[i] != want[i] {
+				t.Fatalf("%q: got %q, want %q", s, got, want)
+			}
+		}
+	}
+}

cmd/server/store.go

@@ -1074,8 +1074,7 @@ func (s *PacketStore) Load() error {
 			// Decode-window: extract resolved pubkeys for index, don't store on struct.
 			rpStr := nullStrVal(resolvedPathStr)
 			if rpStr != "" {
-				rp := unmarshalResolvedPath(rpStr)
-				pks := extractResolvedPubkeys(rp)
+				pks := resolvedPubkeysFromJSON(rpStr)
 				// Single point of truth — see indexResolvedPathHops doc + #1558.
 				s.indexResolvedPathHops(tx, pks, hopsSeen)
 			} else if relayPM != nil && obsPJ != "" && obsPJ != "[]" {
@@ -1419,8 +1418,7 @@ func (s *PacketStore) loadChunk(from, to time.Time) error {
 			// they were heard via — Load() does the equivalent inline.
 			rpStr := nullStrVal(resolvedPathStr)
 			if rpStr != "" {
-				rp := unmarshalResolvedPath(rpStr)
-				pks := extractResolvedPubkeys(rp)
+				pks := resolvedPubkeysFromJSON(rpStr)
 				accumulateDedup(localResolvedPKsByTx, localResolvedSeenByTx, txID, pks)
 			} else if relayPM != nil && obsPJ != "" && obsPJ != "[]" {
 				// resolved_path not persisted — reconstruct relay hops from