perf(store): in-place resolved-path pubkey scanning

cmd/server/resolved_index.go

@@ -718,24 +718,22 @@ func (s *PacketStore) resolvedPubkeysForEvictionBatch(txIDs []int) map[int][]str
 			if !rpJSON.Valid || rpJSON.String == "" {
 				continue
 			}
-			rp := unmarshalResolvedPath(rpJSON.String)
-			for _, p := range rp {
-				if p != nil && *p != "" {
-					result[txID] = append(result[txID], *p)
-				}
-			}
+			result[txID] = appendPersistedRelayPubkeys(result[txID], rpJSON.String)
 		}
 		rows.Close()
 	}
 
-	// Deduplicate per-txID
+	// Deduplicate per-txID. The pubkeys point into the scanned rows
+	// (appendPersistedRelayPubkeys); keep a copy of each distinct one,
+	// because evictStale re-assigns s.byNode[pk], which stores pk as the
+	// map key and would otherwise pin the whole row (#533).
 	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)
+				deduped = append(deduped, strings.Clone(pk))
 			}
 		}
 		result[txID] = deduped

cmd/server/resolved_path_scan.go

@@ -0,0 +1,91 @@
+package main
+
+import "strings"
+
+// In-place reading of persisted resolved_path values (#533).
+//
+// The ingestor writes observations.resolved_path with json.Marshal of a
+// []*string: a compact array of full pubkeys and nulls, e.g.
+// ["a1…",null,"b2…"]. Load, the chunk merge, live ingest and the eviction
+// prefetch read one per observation, 2.5M at startup on the large instance,
+// and decoding them with encoding/json was 17% of Load's allocations and 59%
+// of the eviction prefetch's (pprof, see the PR description).
+//
+// scanResolvedPathPubkeys reads that compact form directly. It accepts only
+// what json.Unmarshal would return unchanged: strings of printable ASCII
+// without escapes, and null. Anything else -- whitespace, an escape, a byte
+// outside 0x20..0x7e (which the decoder would reject or rewrite to U+FFFD), a
+// number, a nested value, trailing data -- makes it report false, and the
+// caller decodes with encoding/json as before. So the result never differs
+// from extractResolvedPubkeys(unmarshalResolvedPath(s)): same strings
+// verbatim (no case folding or validation, as before), same order, duplicates
+// kept, null and "" dropped (FuzzPersistedRelayPath_533).
+//
+// The pubkeys are substrings of s. Callers that retain them must copy them
+// (decodePersistedRelayPath does), or one map key keeps the whole column
+// value alive.
+//
+// No string is shared through the #489 pool: every pubkey the store retains
+// from here is a byNode/nodeHashes/byPathHop map key, which already holds one
+// copy per distinct pubkey, so a pool would add a lock and a lookup per hop
+// and save no retained bytes.
+
+// scanResolvedPathPubkeys appends the non-empty pubkeys of the compact
+// resolved_path s to dst. It returns dst unchanged and false when s is not
+// in the compact form.
+func scanResolvedPathPubkeys(dst []string, s string) ([]string, bool) {
+	n := len(s)
+	if n < 2 || s[0] != '[' || s[n-1] != ']' {
+		return dst, false
+	}
+	start := len(dst)
+	if n == 2 {
+		return dst, true
+	}
+	i := 1
+	for {
+		switch {
+		case s[i] == '"':
+			j := i + 1
+			for j < n && s[j] != '"' {
+				if c := s[j]; c < 0x20 || c > 0x7e || c == '\\' {
+					return dst[:start], false
+				}
+				j++
+			}
+			if j == n {
+				return dst[:start], false
+			}
+			if j > i+1 {
+				dst = append(dst, s[i+1:j])
+			}
+			i = j + 1
+		case strings.HasPrefix(s[i:], "null"):
+			i += 4
+		default:
+			return dst[:start], false
+		}
+		if i == n-1 {
+			return dst, true
+		}
+		if s[i] != ',' || i+1 >= n-1 {
+			return dst[:start], false
+		}
+		i++
+	}
+}
+
+// appendPersistedRelayPubkeys appends the non-empty pubkeys of the
+// resolved_path column value s to dst, scanning in place when it can. The
+// appended strings may point into s.
+func appendPersistedRelayPubkeys(dst []string, s string) []string {
+	if out, ok := scanResolvedPathPubkeys(dst, s); ok {
+		return out
+	}
+	for _, p := range unmarshalResolvedPath(s) {
+		if p != nil && *p != "" {
+			dst = append(dst, *p)
+		}
+	}
+	return dst
+}

cmd/server/resolved_path_scan_533_bench_test.go

@@ -0,0 +1,131 @@
+package main
+
+import (
+	"os"
+	"runtime"
+	"strconv"
+	"testing"
+	"time"
+)
+
+// Measurement harness for #533 (in-place scanning of persisted resolved_path
+// pubkeys). It compiles against master too, so the same benchmarks give the
+// before and after numbers.
+//
+// Inputs come from obs489Data, the large-instance shape used for #489: 1 to
+// 3 hops per row, each a 64-hex pubkey.
+
+// obs533Values returns the resolved_path column values obs489Data writes for
+// about n observations, marshalled exactly as the ingestor persists them.
+func obs533Values(n int) []string {
+	txs := obs489Data(n, 533)
+	out := make([]string, 0, n)
+	for _, tx := range txs {
+		for _, r := range tx.rows {
+			out = append(out, marshalResolvedPath(stringPtrs(r.resolved)))
+		}
+	}
+	return out
+}
+
+func stringPtrs(ss []string) []*string {
+	out := make([]*string, len(ss))
+	for i := range ss {
+		out[i] = &ss[i]
+	}
+	return out
+}
+
+// BenchmarkDecodePersistedRelayPath_533 is the per-observation decode the
+// bulk-load and live-ingest paths run (decodePersistedRelayPath).
+//
+//	go test -run '^$' -bench DecodePersistedRelayPath_533 -count 6 .
+func BenchmarkDecodePersistedRelayPath_533(b *testing.B) {
+	vals := obs533Values(100000)
+	b.ReportAllocs()
+	b.ResetTimer()
+	var n int
+	for i := 0; i < b.N; i++ {
+		n += len(decodePersistedRelayPath(vals[i%len(vals)]).pubkeys)
+	}
+	runtime.KeepAlive(n)
+}
+
+// BenchmarkEvictionBatch_533 is the SQL prefetch RunEviction does before it
+// takes s.mu (resolvedPubkeysForEvictionBatch), over 10K transmissions of a
+// 250K-observation database: one eviction tick's worth on the large instance.
+//
+//	go test -run '^$' -bench EvictionBatch_533 -count 6 .
+func BenchmarkEvictionBatch_533(b *testing.B) {
+	txs := obs489Data(obs489Bench, 533)
+	db, err := OpenDB(obs489DB(b, txs))
+	if err != nil {
+		b.Fatal(err)
+	}
+	defer db.conn.Close()
+	s := NewPacketStore(db, &PacketStoreConfig{})
+	ids := make([]int, 0, 10000)
+	for _, tx := range txs[:10000] {
+		ids = append(ids, tx.id)
+	}
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		if got := s.resolvedPubkeysForEvictionBatch(ids); len(got) != len(ids) {
+			b.Fatalf("batch covers %d txs, want %d", len(got), len(ids))
+		}
+	}
+}
+
+// TestResolvedPathScan533_Measure loads a CORESCOPE_533_MEASURE-observation
+// synthetic store (2500000 is the large instance: about 170K transmissions)
+// and reports Load wall time, allocations, heap after load and peak RSS
+// proxy (Sys), then times one eviction prefetch over every transmission.
+// Run it with -cpuprofile/-memprofile to attribute the cost.
+//
+//	CORESCOPE_533_MEASURE=2500000 go test -run ResolvedPathScan533_Measure -v -timeout 60m
+func TestResolvedPathScan533_Measure(t *testing.T) {
+	n, _ := strconv.Atoi(os.Getenv("CORESCOPE_533_MEASURE"))
+	if n <= 0 {
+		t.Skip("set CORESCOPE_533_MEASURE=2500000 to run")
+	}
+	txs := obs489Data(n, 533)
+	dbPath := obs489DB(t, txs)
+	db, err := OpenDB(dbPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.conn.Close()
+
+	runtime.GC()
+	var m0, m1 runtime.MemStats
+	runtime.ReadMemStats(&m0)
+	t0 := time.Now()
+	s := NewPacketStore(db, &PacketStoreConfig{})
+	if err := s.Load(); err != nil {
+		t.Fatal(err)
+	}
+	load := time.Since(t0)
+	s.WaitIndexesReady(10 * time.Minute)
+	runtime.ReadMemStats(&m1)
+	allocs, bytes := m1.Mallocs-m0.Mallocs, m1.TotalAlloc-m0.TotalAlloc
+	runtime.GC()
+	runtime.ReadMemStats(&m1)
+	t.Logf("load: tx=%d obs=%d wall=%v allocs=%d allocMiB=%.1f heapMiB=%.1f sysMiB=%.0f",
+		len(s.packets), s.totalObs, load, allocs, float64(bytes)/1048576,
+		float64(int64(m1.HeapAlloc)-int64(m0.HeapAlloc))/1048576, float64(m1.Sys)/1048576)
+
+	ids := make([]int, 0, len(txs))
+	for _, tx := range txs {
+		ids = append(ids, tx.id)
+	}
+	runtime.GC()
+	runtime.ReadMemStats(&m0)
+	t0 = time.Now()
+	batch := s.resolvedPubkeysForEvictionBatch(ids)
+	evict := time.Since(t0)
+	runtime.ReadMemStats(&m1)
+	t.Logf("eviction prefetch: tx=%d wall=%v allocs=%d allocMiB=%.1f",
+		len(batch), evict, m1.Mallocs-m0.Mallocs, float64(m1.TotalAlloc-m0.TotalAlloc)/1048576)
+	runtime.KeepAlive(s)
+}