From 5199a4435ad23173337d6a1cbd71475b58e3b2cb Mon Sep 17 00:00:00 2001 From: dborup Date: Sat, 24 Oct 2026 18:19:54 +0200 Subject: [PATCH] perf(store): scan persisted resolved_path pubkeys in place decodePersistedRelayPath (Load, chunk merge, live ingest) and the eviction prefetch decoded every observations.resolved_path value with encoding/json. The ingestor writes it as compact json.Marshal output of []*string, so read that form directly and keep the JSON decoder as the fallback for anything else (whitespace, escapes, non-ASCII, non-string values, trailing data). The result is unchanged: strings verbatim, same order, duplicates kept, null and "" dropped (differential + fuzz test against extractResolvedPubkeys(unmarshalResolvedPath)). Retained pubkeys are copies, never substrings of the row: the load path clones each kept hop, and the eviction prefetch clones each distinct pubkey per transmission, since evictStale re-assigns s.byNode[pk] and a string-key assignment stores the new key. 2.5M observations / 121K transmissions: Load allocations 140.4M -> 127.9M, allocated 5958 -> 5131 MiB, retained heap unchanged; full eviction prefetch 2.95 s -> 1.67 s, 46.4M -> 24.8M allocations. Relates to #533 Co-Authored-By: Claude Opus 5.5 --- cmd/server/resolved_index.go | 14 +- cmd/server/resolved_path_scan.go | 91 ++++++++ .../resolved_path_scan_533_bench_test.go | 131 +++++++++++ cmd/server/resolved_path_scan_533_test.go | 212 ++++++++++++++++++ cmd/server/store.go | 23 +- 5 files changed, 458 insertions(+), 13 deletions(-) create mode 100644 cmd/server/resolved_path_scan.go create mode 100644 cmd/server/resolved_path_scan_533_bench_test.go create mode 100644 cmd/server/resolved_path_scan_533_test.go diff --git a/cmd/server/resolved_index.go b/cmd/server/resolved_index.go index abe62cb89..a94669a6f 100644 --- a/cmd/server/resolved_index.go +++ b/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 diff --git a/cmd/server/resolved_path_scan.go b/cmd/server/resolved_path_scan.go new file mode 100644 index 000000000..d5a3fc1e0 --- /dev/null +++ b/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 +} diff --git a/cmd/server/resolved_path_scan_533_bench_test.go b/cmd/server/resolved_path_scan_533_bench_test.go new file mode 100644 index 000000000..f83281019 --- /dev/null +++ b/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) +} diff --git a/cmd/server/resolved_path_scan_533_test.go b/cmd/server/resolved_path_scan_533_test.go new file mode 100644 index 000000000..b82b6622b --- /dev/null +++ b/cmd/server/resolved_path_scan_533_test.go @@ -0,0 +1,212 @@ +package main + +import ( + "strings" + "testing" + "unsafe" +) + +// #533: decodePersistedRelayPath and the eviction prefetch read the +// resolved_path column without encoding/json when it is in the compact form +// the ingestor writes (json.Marshal of []*string). Everything else goes +// through the JSON decoder, so the result must always equal what +// extractResolvedPubkeys(unmarshalResolvedPath(s)) gives. + +const ( + pk533A = "a1" + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789ab" + pk533B = "B2" + "FEDCBA9876543210FEDCBA9876543210FEDCBA9876543210FEDCBA987654" +) + +// resolved533Cases are column values on both sides of the compact grammar. +// Each one is checked against the JSON oracle, so a case only documents +// which side it is meant to exercise. +var resolved533Cases = []string{ + // compact form: scanned in place + `[]`, `[null]`, `[null,null]`, `[""]`, `["",""]`, + `["` + pk533A + `"]`, + `["` + pk533A + `",null,"` + pk533B + `"]`, + `["` + pk533A + `","` + pk533A + `"]`, // duplicates are kept + `[null,"` + pk533B + `",""]`, + `["ab","x y","{}","[1]","null"]`, // short, odd but valid strings, kept verbatim + // anything else: JSON fallback + ``, `null`, `[`, `]`, `[,]`, `[null,]`, `[,null]`, `["a",]`, `["a"`, `["a"]]`, + `["a"] `, ` ["a"]`, `[ "a"]`, `["a" ,"b"]`, `["a",` + "\n" + `"b"]`, "[\"a\"]\t", + `[1]`, `[1,"` + pk533A + `"]`, `[true]`, `[{}]`, `[[]]`, `{}`, `"a"`, `nul`, `[nul]`, `[nulll]`, `[Null]`, + `["a\"b"]`, `["a\\b"]`, `["A"]`, `["å"]`, `["\ud800"]`, `["a\/b"]`, `["\x"]`, + "[\"a\tb\"]", "[\"a\x00b\"]", "[\"a\x1fb\"]", "[\"a\x7fb\"]", + "[\"\xc3\xa5\"]", "[\"\xff\"]", "[\"\xed\xa0\x80\"]", // valid UTF-8, invalid byte, surrogate + `["a"]["b"]`, `[null]x`, `[null]null`, `{"ab"]`, `xnull]`, ` "` + pk533A + `"]`, +} + +func oracle533(s string) persistedRelayPath { + if s == "" { + return persistedRelayPath{} + } + return persistedRelayPath{pubkeys: extractResolvedPubkeys(unmarshalResolvedPath(s)), present: true} +} + +func samePath533(a, b persistedRelayPath) bool { + if a.present != b.present || len(a.pubkeys) != len(b.pubkeys) { + return false + } + for i := range a.pubkeys { + if a.pubkeys[i] != b.pubkeys[i] { + return false + } + } + return true +} + +func TestDecodePersistedRelayPathMatchesJSON_533(t *testing.T) { + for _, in := range resolved533Cases { + if got, want := decodePersistedRelayPath(in), oracle533(in); !samePath533(got, want) { + t.Errorf("decodePersistedRelayPath(%q) = %q/%v, JSON gives %q/%v", in, got.pubkeys, got.present, want.pubkeys, want.present) + } + var dst []string + if got, want := appendPersistedRelayPubkeys(dst, in), oracle533(in).pubkeys; strings.Join(got, "\x00") != strings.Join(want, "\x00") { + t.Errorf("appendPersistedRelayPubkeys(%q) = %q, JSON gives %q", in, got, want) + } + } +} + +// The scanner accepts exactly json.Marshal's output for []*string of plain +// printable-ASCII strings and rejects everything with an escape, whitespace, +// a non-string value or a byte outside 0x20..0x7e: those are what the JSON +// decoder rewrites (escapes, invalid UTF-8 -> U+FFFD) or rejects. +func TestScanResolvedPathPubkeysGrammar_533(t *testing.T) { + accept := map[string][]string{ + `[]`: {}, + `[null,""]`: {}, + `["` + pk533A + `",null,"ab"]`: {pk533A, "ab"}, + `["` + pk533B + `","` + pk533B + `"]`: {pk533B, pk533B}, + } + for in, want := range accept { + got, ok := scanResolvedPathPubkeys(nil, in) + if !ok || strings.Join(got, ",") != strings.Join(want, ",") { + t.Errorf("scan(%q) = %q, %v; want %q, true", in, got, ok, want) + } + } + for _, in := range []string{``, `null`, `[ ]`, `["a\"b"]`, `["a\\b"]`, "[\"\xc3\xa5\"]", "[\"a\x7fb\"]", "[\"a\x1fb\"]", + `[1]`, `[null,]`, `["a"] `, `[nulL]`, `["a"`, `[null]x`, `{"ab"]`, `xnull]`} { + pre := []string{"keep"} + if got, ok := scanResolvedPathPubkeys(pre, in); ok || len(got) != 1 || got[0] != "keep" { + t.Errorf("scan(%q) = %q, %v; want the fallback (dst untouched, false)", in, got, ok) + } + } +} + +// The pubkeys decodePersistedRelayPath returns are retained by the store as +// byNode/nodeHashes/byPathHop keys, so none may point into the column value: +// a 64-byte key would otherwise keep the whole row string alive. +func TestDecodePersistedRelayPathDoesNotPinInput_533(t *testing.T) { + in := strings.Clone(`["` + pk533A + `",null,"` + pk533B + `"]`) + lo := uintptr(unsafe.Pointer(unsafe.StringData(in))) + hi := lo + uintptr(len(in)) + got := decodePersistedRelayPath(in) + if len(got.pubkeys) != 2 { + t.Fatalf("decoded %q", got.pubkeys) + } + for _, pk := range got.pubkeys { + if p := uintptr(unsafe.Pointer(unsafe.StringData(pk))); p >= lo && p < hi { + t.Errorf("pubkey %q points into the column value", pk) + } + } +} + +// Allocation ceilings, the point of #533. The JSON decoder costs 7 +// allocations for a 2-hop path (BenchmarkDecodePersistedRelayPath_533); +// the scan costs the result slice and one copy per kept hop, and the +// eviction form appends substrings into a caller's slice with no allocation. +func TestPersistedRelayPathAllocs_533(t *testing.T) { + in := `["` + pk533A + `",null,"` + pk533B + `"]` + if a := testing.AllocsPerRun(200, func() { decodePersistedRelayPath(in) }); a > 3 { + t.Errorf("decodePersistedRelayPath: %.0f allocs, want <= 3 (slice + one copy per hop)", a) + } + dst := make([]string, 0, 8) + if a := testing.AllocsPerRun(200, func() { dst = appendPersistedRelayPubkeys(dst[:0], in) }); a != 0 { + t.Errorf("appendPersistedRelayPubkeys: %.0f allocs, want 0", a) + } +} + +// The eviction prefetch must give the pubkeys the JSON decoder gives, per +// transmission and deduplicated in first-seen order, for compact and +// fallback rows alike. +func TestEvictionBatchMatchesJSON_533(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + rows := map[int][]string{ + 1: {`["` + pk533A + `",null,"` + pk533B + `"]`, `["` + pk533B + `","` + pk533A + `"]`}, + 2: {`["ab"]`, `["ab"]`, `[1,"` + pk533A + `"]`}, + 3: {`[null]`, `[]`, `["",""]`}, + 4: {` ["` + pk533B + `"] `, "[\"\xff\"]", `["` + pk533A + `"]`}, + } + for tx, vals := range rows { + if _, err := db.conn.Exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen) VALUES (?, '00', ?, '2026-01-01T00:00:00Z')`, + tx, strings.Repeat("0", 15)+string(rune('0'+tx))); err != nil { + t.Fatal(err) + } + for _, v := range vals { + if _, err := db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp, resolved_path) VALUES (?, 1, '[]', 1, ?)`, tx, v); err != nil { + t.Fatal(err) + } + } + } + s := NewPacketStore(db, &PacketStoreConfig{}) + got := s.resolvedPubkeysForEvictionBatch([]int{1, 2, 3, 4}) + for tx, vals := range rows { + var want []string + seen := map[string]bool{} + for _, v := range vals { + for _, pk := range oracle533(v).pubkeys { + if !seen[pk] { + seen[pk] = true + want = append(want, pk) + } + } + } + if strings.Join(got[tx], ",") != strings.Join(want, ",") { + t.Errorf("tx %d: batch %q, JSON gives %q", tx, got[tx], want) + } + } +} + +func FuzzPersistedRelayPath_533(f *testing.F) { + for _, c := range resolved533Cases { + f.Add(c) + } + f.Fuzz(func(t *testing.T, in string) { + want := oracle533(in) + if got := decodePersistedRelayPath(in); !samePath533(got, want) { + t.Fatalf("decodePersistedRelayPath(%q) = %q/%v, JSON gives %q/%v", in, got.pubkeys, got.present, want.pubkeys, want.present) + } + if got := appendPersistedRelayPubkeys(nil, in); strings.Join(got, "\x00") != strings.Join(want.pubkeys, "\x00") || len(got) != len(want.pubkeys) { + t.Fatalf("appendPersistedRelayPubkeys(%q) = %q, JSON gives %q", in, got, want.pubkeys) + } + }) +} + +// The eviction prefetch's pubkeys reach a retained map key: evictStale +// re-assigns s.byNode[pk] for every affected node, and assigning to an +// existing string key stores the new key header. So they must be copies, +// not substrings of the row: two hops of one row must not sit 3 bytes ('","') +// apart in the same allocation. +func TestEvictionBatchDoesNotPinRows_533(t *testing.T) { + db := setupTestDB(t) + defer db.Close() + if _, err := db.conn.Exec(`INSERT INTO transmissions (id, raw_hex, hash, first_seen) VALUES (1, '00', '0000000000000001', '2026-01-01T00:00:00Z')`); err != nil { + t.Fatal(err) + } + if _, err := db.conn.Exec(`INSERT INTO observations (transmission_id, observer_idx, path_json, timestamp, resolved_path) VALUES (1, 1, '[]', 1, ?)`, + `["`+pk533A+`","`+pk533B+`"]`); err != nil { + t.Fatal(err) + } + got := NewPacketStore(db, &PacketStoreConfig{}).resolvedPubkeysForEvictionBatch([]int{1})[1] + if len(got) != 2 { + t.Fatalf("batch %q", got) + } + a := uintptr(unsafe.Pointer(unsafe.StringData(got[0]))) + b := uintptr(unsafe.Pointer(unsafe.StringData(got[1]))) + if b == a+uintptr(len(got[0]))+3 { + t.Errorf("pubkeys are substrings of the row value (data at %#x and %#x)", a, b) + } +} diff --git a/cmd/server/store.go b/cmd/server/store.go index c8888590e..36a46f52d 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -2218,15 +2218,28 @@ type persistedRelayPath struct { // before s.mu is taken, so the JSON work does not extend the critical // section (#158 follow-up). // -// The result equals extractResolvedPubkeys(unmarshalResolvedPath(s)). It -// decodes straight into []string (a JSON null becomes "", which is what -// extractResolvedPubkeys drops) and compacts in place, saving the *string -// allocated per hop and the second slice on a path that runs once per -// observation. +// The result equals extractResolvedPubkeys(unmarshalResolvedPath(s)). The +// compact form the ingestor writes is scanned in place (#533, +// scanResolvedPathPubkeys) and each pubkey copied out of the column value, +// since the store keeps them as index keys. Anything else is decoded +// straight into []string (a JSON null becomes "", which is what +// extractResolvedPubkeys drops) and compacted in place, saving the *string +// allocated per hop and the second slice. func decodePersistedRelayPath(resolvedPath string) persistedRelayPath { if resolvedPath == "" { return persistedRelayPath{} } + var buf [8]string + if scanned, ok := scanResolvedPathPubkeys(buf[:0], resolvedPath); ok { + if len(scanned) == 0 { + return persistedRelayPath{present: true} + } + pks := make([]string, len(scanned)) + for i, pk := range scanned { + pks[i] = strings.Clone(pk) + } + return persistedRelayPath{pubkeys: pks, present: true} + } var hops []string if json.Unmarshal([]byte(resolvedPath), &hops) != nil { return persistedRelayPath{present: true} // as unmarshalResolvedPath: corrupt means no hops